476f4f954d
- CustomToolRouter como callback (roda em todas as chamadas) - ToolRouterGuardrail para ativação por request - Modos: jaccard (offline) e embedding (API) - Dockerfile + docker-compose prontos - README com docs completas - Teste offline validando ~74% redução de tokens
603 lines
23 KiB
Python
603 lines
23 KiB
Python
"""
|
|
Tool Router - LiteLLM Custom Logger / Guardrail
|
|
===============================================
|
|
Intercepta as chamadas do proxy do LiteLLM e faz retrieval das tools,
|
|
reduzindo drasticamente o consumo de tokens.
|
|
|
|
Como funciona:
|
|
1. Indexa cada tool (name + description)
|
|
2. No pre-call hook, calcula similaridade com a query do usuário
|
|
3. Seleciona top-K tools (+ always_include)
|
|
4. Só envia essas tools para o LLM
|
|
|
|
Uso como CALLBACK (roda em todas as chamadas):
|
|
litellm_settings:
|
|
callbacks:
|
|
- tool_router_logger.tool_router_instance
|
|
|
|
Uso como GUARDRAIL (ativa por request):
|
|
litellm_settings:
|
|
guardrails:
|
|
- guardrail_name: tool_router
|
|
litellm_params:
|
|
guardrail: tool_router_logger.ToolRouterGuardrail
|
|
mode: pre_call
|
|
default_on: false
|
|
|
|
Ou programaticamente:
|
|
litellm.callbacks = [CustomToolRouter(...)]
|
|
"""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Any, Dict, List, Optional, Tuple, Union
|
|
|
|
import numpy as np
|
|
from sklearn.metrics.pairwise import cosine_similarity
|
|
|
|
from litellm.integrations.custom_logger import CustomLogger
|
|
from litellm.integrations.custom_guardrail import CustomGuardrail
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
from litellm.caching import DualCache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Embedding local leve (sem dependência externa pesada)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class FastEmbedder:
|
|
"""
|
|
Usa o modelo de embedding do LiteLLM (qualquer provider via API)
|
|
pra evitar depender de sentence-transformers local.
|
|
|
|
Alternativa: Cache as embeddings no Valkey para reuse entre chamadas.
|
|
"""
|
|
|
|
def __init__(self, embedding_model: str = "text-embedding-ada-002", cache: Optional[DualCache] = None):
|
|
self.embedding_model = embedding_model
|
|
self.cache = cache
|
|
self._local_cache: Dict[str, List[float]] = {}
|
|
|
|
async def embed(self, texts: List[str], litellm_kwargs: Optional[Dict] = None) -> List[List[float]]:
|
|
"""Gera embeddings em lote usando o LiteLLM."""
|
|
import litellm
|
|
|
|
to_embed = []
|
|
indices = []
|
|
results = [None] * len(texts)
|
|
|
|
# Verifica cache primeiro (se disponível)
|
|
for i, text in enumerate(texts):
|
|
text_hash = hashlib.md5(text.encode()).hexdigest()
|
|
cache_key = f"tool_embed:{text_hash}"
|
|
|
|
if cache_key in self._local_cache:
|
|
results[i] = self._local_cache[cache_key]
|
|
elif self.cache:
|
|
cached = await self.cache.async_get_cache(cache_key)
|
|
if cached is not None:
|
|
val = cached
|
|
if isinstance(val, str):
|
|
val = json.loads(val)
|
|
self._local_cache[cache_key] = val
|
|
results[i] = val
|
|
else:
|
|
to_embed.append(text)
|
|
indices.append(i)
|
|
else:
|
|
to_embed.append(text)
|
|
indices.append(i)
|
|
|
|
if not to_embed:
|
|
return results # type: ignore
|
|
|
|
# Faz chamada de embedding
|
|
kwargs = {"model": self.embedding_model, "input": to_embed}
|
|
if litellm_kwargs:
|
|
kwargs.update(litellm_kwargs)
|
|
|
|
try:
|
|
response = await litellm.aembedding(**kwargs)
|
|
for idx, emb_data in zip(indices, response.data):
|
|
embedding = emb_data["embedding"] if isinstance(emb_data, dict) else emb_data.embedding
|
|
results[idx] = embedding
|
|
|
|
# Salva no cache
|
|
text = to_embed[indices.index(idx)]
|
|
text_hash = hashlib.md5(text.encode()).hexdigest()
|
|
cache_key = f"tool_embed:{text_hash}"
|
|
self._local_cache[cache_key] = embedding
|
|
if self.cache:
|
|
await self.cache.async_set_cache(cache_key, embedding)
|
|
except Exception as e:
|
|
logger.warning(f"Embedding failed, falling back to empty: {e}")
|
|
for idx in indices:
|
|
results[idx] = [0.0] * 1536 # fallback
|
|
|
|
return results # type: ignore
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Similaridade textual inteligente (fallback sem embedding externo)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import string
|
|
from collections import Counter
|
|
|
|
# Mapa de sinônimos e lematização para português + inglês (tools geralmente
|
|
# têm descrições em inglês, queries em português)
|
|
_SYNONYM_MAP = {
|
|
# verbos para substantivos (português)
|
|
"pesquise": "pesquisa", "pesquisar": "pesquisa", "buscar": "pesquisa", "busca": "pesquisa",
|
|
"procure": "pesquisa", "procurar": "pesquisa", "encontre": "pesquisa",
|
|
"leia": "leitura", "ler": "leitura", "abrir": "leitura", "abra": "leitura",
|
|
"escreva": "escrita", "escrever": "escrita", "salvar": "escrita",
|
|
"execute": "execução", "executar": "execução", "rodar": "execução", "rode": "execução",
|
|
"liste": "listagem", "listar": "listagem", "mostre": "listagem",
|
|
"envie": "envio", "enviar": "envio", "mande": "envio",
|
|
"calcule": "cálculo", "calcular": "cálculo", "calcular": "cálculo",
|
|
# português → inglês (tools descriptions são em inglês)
|
|
"hora": "time", "horas": "time", "horário": "time", "horario": "time",
|
|
"relógio": "time", "relogio": "time",
|
|
"tempo": "weather time", # ambos
|
|
"clima": "weather", "previsão": "weather", "previsao": "weather",
|
|
"chuva": "weather", "temperatura": "weather", "temperatures": "weather",
|
|
"cidade": "city",
|
|
"arquivo": "file", "arquivos": "file", "documento": "file",
|
|
"pasta": "directory", "diretório": "directory", "diretorio": "directory",
|
|
"email": "email", "e-mail": "email", "mensagem": "email",
|
|
"web": "web", "internet": "web", "site": "web", "pesquisar": "search",
|
|
"bash": "bash", "terminal": "bash", "comando": "bash command", "shell": "bash",
|
|
"linha": "line",
|
|
"navegador": "browser",
|
|
"sistema": "system", "arquivos": "files",
|
|
}
|
|
|
|
|
|
# Stopwords comuns em português e inglês (palavras muito curtas/frequentes)
|
|
_STOPWORDS = {
|
|
"a", "e", "o", "em", "de", "da", "do", "na", "no", "um", "uma",
|
|
"é", "ser", "tem", "com", "para", "por", "pra", "os", "as",
|
|
"ao", "aos", "das", "dos", "nas", "nos", "que", "se", "mas",
|
|
"mais", "muito", "como", "vai", "vou", "me", "te", "se",
|
|
"the", "a", "an", "in", "on", "at", "to", "for", "of",
|
|
"is", "it", "be", "with", "and", "or", "from", "by",
|
|
"um", "uns", "uma", "umas",
|
|
}
|
|
|
|
|
|
def _tokenize(text: str) -> list:
|
|
"""Tokeniza e normaliza, retornando lista (não set) pra preservar frequência."""
|
|
text = text.lower().translate(str.maketrans('', '', string.punctuation))
|
|
tokens = re.findall(r'[a-záéíóúãõâêîôûç]+', text)
|
|
# Remove stopwords
|
|
tokens = [t for t in tokens if t not in _STOPWORDS and len(t) > 1]
|
|
# Aplica sinônimos
|
|
expanded = []
|
|
for t in tokens:
|
|
expanded.append(t)
|
|
if t in _SYNONYM_MAP:
|
|
expanded.extend(_SYNONYM_MAP[t].split())
|
|
return expanded
|
|
|
|
|
|
def _jaccard_similarity(a: set, b: set) -> float:
|
|
if not a or not b:
|
|
return 0.0
|
|
return len(a & b) / len(a | b)
|
|
|
|
|
|
def _tf_keyword_score(query_tokens: list, tool_text: str) -> float:
|
|
"""
|
|
Score baseado em quantas palavras-chave da query aparecem no texto da tool.
|
|
Penaliza tools que não contêm NENHUMA palavra relevante.
|
|
"""
|
|
tool_lower = tool_text.lower()
|
|
matches = sum(1 for q in query_tokens if q in tool_lower)
|
|
if matches == 0:
|
|
return 0.0
|
|
# Normaliza pelo tamanho da query pra favorecer matches parciais
|
|
return matches / max(len(set(query_tokens)), 1)
|
|
|
|
|
|
def _combined_text_similarity(query: str, tool_text: str) -> float:
|
|
"""
|
|
Combina Jaccard (set overlap) + TF-keyword score pra dar
|
|
similaridade mais robusta sem precisar de embeddings externos.
|
|
"""
|
|
query_tokens = _tokenize(query)
|
|
tool_tokens = _tokenize(tool_text)
|
|
|
|
query_set = set(query_tokens)
|
|
tool_set = set(tool_tokens)
|
|
|
|
# Jaccard entre sets
|
|
jac = _jaccard_similarity(query_set, tool_set)
|
|
|
|
# TF-score: quantas palavras da query estão no texto da tool
|
|
tf = _tf_keyword_score(query_tokens, tool_text)
|
|
|
|
# Identifica match exato no nome da tool (peso máximo)
|
|
query_words = set(w for w in query_tokens if len(w) > 2)
|
|
tool_name = tool_text.split()[0] if tool_text else ""
|
|
name_match = 1.0 if tool_name in query_words else 0.0
|
|
|
|
# Se o nome da tool apareceu na query, é quase certeza
|
|
if name_match > 0:
|
|
return 1.0
|
|
|
|
# Peso maior pro TF quando Jaccard é baixo (captura relações indiretas)
|
|
if jac == 0 and tf > 0:
|
|
return tf * 0.4 # meio caminho, mais conservador
|
|
|
|
# Só retorna algo relevante se passou de um mínimo
|
|
raw = max(jac, tf * 0.6)
|
|
return raw
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CustomLogger principal
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DEFAULT_TOOL_NAMES = {"get_time", "help", "ping", "status"}
|
|
|
|
class CustomToolRouter(CustomLogger):
|
|
"""
|
|
Router semântico de tools para LiteLLM Proxy.
|
|
|
|
No pre-call hook, faz retrieval das tools mais relevantes
|
|
para a query do usuário, reduzindo o prompt de tools enviado ao LLM.
|
|
|
|
Args:
|
|
embedding_model: Modelo de embedding (ex: 'text-embedding-ada-002')
|
|
top_k: Número de tools a incluir (default: 3)
|
|
min_similarity: Similaridade mínima para incluir tool (default: 0.15)
|
|
always_include_tools: Lista de nomes de tools sempre enviadas
|
|
similarity_mode: "embedding" (preferido) ou "jaccard" (fallback leve)
|
|
cache_host: Host do Valkey/Redis (opcional)
|
|
cache_port: Porta do Valkey/Redis (opcional)
|
|
cache_password: Senha (opcional)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
embedding_model: str = "text-embedding-ada-002",
|
|
top_k: int = 3,
|
|
min_similarity: float = 0.15,
|
|
always_include_tools: Optional[list] = None,
|
|
similarity_mode: str = "embedding",
|
|
cache_host: Optional[str] = None,
|
|
cache_port: Optional[str] = None,
|
|
cache_password: Optional[str] = None,
|
|
):
|
|
super().__init__()
|
|
self.top_k = top_k
|
|
self.min_similarity = min_similarity
|
|
self.always_include_tools = set(always_include_tools or DEFAULT_TOOL_NAMES)
|
|
self.similarity_mode = similarity_mode
|
|
self.embedding_model = embedding_model
|
|
|
|
# Cache compartilhado (inicializado no async_pre_call_hook)
|
|
self._dual_cache: Optional[DualCache] = None
|
|
self._embedder: Optional[FastEmbedder] = None
|
|
|
|
# Cache local de embeddings de tools
|
|
self._tool_embeddings: Dict[str, List[float]] = {}
|
|
self._tool_texts: Dict[str, str] = {}
|
|
|
|
logger.info(
|
|
f"CustomToolRouter initialized: "
|
|
f"top_k={top_k}, min_similarity={min_similarity}, "
|
|
f"mode={similarity_mode}, model={embedding_model}, "
|
|
f"always_include={self.always_include_tools}"
|
|
)
|
|
|
|
def _extract_user_message(self, messages: List[Dict]) -> str:
|
|
"""Extrai a última mensagem do usuário."""
|
|
for msg in reversed(messages):
|
|
role = msg.get("role", "")
|
|
content = msg.get("content", "")
|
|
if role == "user":
|
|
if isinstance(content, list):
|
|
# Pode ser multimodal - extrai texto
|
|
texts = []
|
|
for part in content:
|
|
if isinstance(part, dict) and part.get("type") == "text":
|
|
texts.append(part.get("text", ""))
|
|
return " ".join(texts)
|
|
return str(content)
|
|
return ""
|
|
|
|
def _build_tool_text(self, tool: Dict) -> str:
|
|
"""Cria um texto combinado para a tool (nome + descrição + parâmetros)."""
|
|
func = tool.get("function", tool)
|
|
name = func.get("name", "")
|
|
description = func.get("description", "")
|
|
parameters = func.get("parameters", {})
|
|
|
|
# Concatena informações relevantes
|
|
parts = [name, description]
|
|
|
|
# Adiciona nomes dos parâmetros
|
|
if isinstance(parameters, dict):
|
|
props = parameters.get("properties", {})
|
|
if isinstance(props, dict):
|
|
param_names = " ".join(props.keys())
|
|
if param_names:
|
|
parts.append(param_names)
|
|
|
|
return " ".join(parts).lower()
|
|
|
|
async def _build_tool_index(self, tools: List[Dict]):
|
|
"""Indexa as tools: gera/cacheia embeddings."""
|
|
texts = {}
|
|
for tool in tools:
|
|
func = tool.get("function", tool)
|
|
name = func.get("name", "")
|
|
text = self._build_tool_text(tool)
|
|
texts[name] = text
|
|
self._tool_texts[name] = text
|
|
|
|
# Gera embeddings para tools ainda não indexadas
|
|
if self.similarity_mode == "embedding" and self._embedder:
|
|
new_tools = [name for name in texts if name not in self._tool_embeddings]
|
|
if new_tools:
|
|
tool_texts_list = [texts[name] for name in new_tools]
|
|
try:
|
|
embeddings = await self._embedder.embed(tool_texts_list)
|
|
for name, emb in zip(new_tools, embeddings):
|
|
self._tool_embeddings[name] = emb
|
|
except Exception as e:
|
|
logger.warning(f"Tool indexing failed, using jaccard: {e}")
|
|
self.similarity_mode = "jaccard"
|
|
|
|
async def _compute_similarities(
|
|
self, query: str, tool_names: List[str]
|
|
) -> List[Tuple[str, float]]:
|
|
"""Calcula similaridade entre query e cada tool."""
|
|
if not query.strip():
|
|
return [(name, 0.0) for name in tool_names]
|
|
|
|
if self.similarity_mode == "embedding" and self._embedder:
|
|
try:
|
|
query_emb = (await self._embedder.embed([query]))[0]
|
|
similarities = []
|
|
for name in tool_names:
|
|
tool_emb = self._tool_embeddings.get(name)
|
|
if tool_emb:
|
|
sim = cosine_similarity([query_emb], [tool_emb])[0][0]
|
|
similarities.append((name, float(sim)))
|
|
else:
|
|
similarities.append((name, 0.0))
|
|
return similarities
|
|
except Exception as e:
|
|
logger.warning(f"Embedding similarity failed: {e}")
|
|
# fallback pra jaccard
|
|
|
|
# Similaridade combinada (fallback inteligente)
|
|
similarities = []
|
|
for name in tool_names:
|
|
tool_text = self._tool_texts.get(name, name)
|
|
sim = _combined_text_similarity(query, tool_text)
|
|
similarities.append((name, sim))
|
|
|
|
return similarities
|
|
|
|
async def async_pre_call_hook(
|
|
self,
|
|
user_api_key_dict: UserAPIKeyAuth,
|
|
cache: DualCache,
|
|
data: dict,
|
|
call_type: str,
|
|
) -> Optional[Union[Exception, str, dict]]:
|
|
"""
|
|
Hook principal: intercepta a requisição e filtra tools.
|
|
|
|
Retorna:
|
|
- dict modificado → segue com as tools filtradas
|
|
- None → continua sem modificação
|
|
- Exception → rejeita a request
|
|
"""
|
|
# Só processa completion
|
|
if call_type not in ("completion", "acompletion"):
|
|
return None
|
|
|
|
messages = data.get("messages", [])
|
|
tools = data.get("tools", [])
|
|
|
|
# Se não tem tools ou só 1, não precisa fazer nada
|
|
if not tools or len(tools) <= 1:
|
|
return None
|
|
|
|
# Inicializa cache e embedder
|
|
self._dual_cache = cache
|
|
if self._embedder is None:
|
|
self._embedder = FastEmbedder(
|
|
embedding_model=self.embedding_model,
|
|
cache=cache,
|
|
)
|
|
|
|
# Extrai query do usuário
|
|
user_query = self._extract_user_message(messages)
|
|
|
|
# Indexa as tools (se necessário)
|
|
await self._build_tool_index(tools)
|
|
|
|
# Nomes de todas as tools
|
|
all_tool_names = []
|
|
for tool in tools:
|
|
func = tool.get("function", tool)
|
|
all_tool_names.append(func.get("name", ""))
|
|
|
|
# Calcula similaridades
|
|
similarities = await self._compute_similarities(user_query, all_tool_names)
|
|
|
|
# Filtra: sempre inclui + top-k por similaridade
|
|
selected_names = set()
|
|
|
|
# 1. Tools always-include
|
|
for name in self.always_include_tools:
|
|
if name in all_tool_names:
|
|
selected_names.add(name)
|
|
|
|
# 2. Top-k por similaridade (acima do threshold)
|
|
sorted_tools = sorted(similarities, key=lambda x: x[1], reverse=True)
|
|
for name, sim in sorted_tools:
|
|
if len(selected_names) >= self.top_k:
|
|
break
|
|
if sim >= self.min_similarity:
|
|
selected_names.add(name)
|
|
|
|
# Se não selecionou nada, pelo menos a top-1
|
|
if not selected_names and sorted_tools:
|
|
selected_names.add(sorted_tools[0][0])
|
|
|
|
# Monta novo tools array
|
|
new_tools = [t for t in tools if (
|
|
t.get("function", t).get("name", "") in selected_names
|
|
)]
|
|
|
|
# Log do que aconteceu
|
|
logger.info(
|
|
f"ToolRouter: tools {len(tools)}→{len(new_tools)} "
|
|
f"(query: '{user_query[:60]}...' ) "
|
|
f"selected: {sorted(selected_names)} "
|
|
f"similarities: {dict(sorted_tools)}"
|
|
)
|
|
|
|
# Retorna dados modificados
|
|
if len(new_tools) < len(tools):
|
|
data["tools"] = new_tools
|
|
# Se reduziu drasticamente, também podemos avisar o modelo
|
|
if len(new_tools) <= 1 and messages:
|
|
system_msg = (
|
|
"Você tem acesso apenas a tools selecionadas para esta conversa. "
|
|
"Se precisar de uma tool específica que não está disponível, "
|
|
"explique qual precisa e o usuário poderá liberá-la."
|
|
)
|
|
# Adiciona system message sutil, sem poluir
|
|
existing_system = False
|
|
for msg in messages:
|
|
if msg.get("role") == "system":
|
|
existing_system = True
|
|
if system_msg not in msg.get("content", ""):
|
|
msg["content"] = msg.get("content", "") + "\n\n" + system_msg
|
|
break
|
|
if not existing_system:
|
|
messages.insert(0, {"role": "system", "content": system_msg})
|
|
data["messages"] = messages
|
|
|
|
return data
|
|
|
|
async def async_log_success_event(self, *args, **kwargs):
|
|
"""Apenas para logging/monitoramento."""
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Factory para criação do logger via string path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# =============================================================================
|
|
# Instância pronta + factory + wrapper de guardrail
|
|
# =============================================================================
|
|
#
|
|
# Uso como CALLBACK (recomendado — roda em todas as chamadas):
|
|
# litellm_settings:
|
|
# callbacks: ["tool_router_logger.tool_router_instance"]
|
|
#
|
|
# Uso como GUARDRAIL (ativa por request com `guardrails: ["tool_router"]`):
|
|
# litellm_settings:
|
|
# guardrails:
|
|
# - guardrail_name: tool_router
|
|
# litellm_params:
|
|
# guardrail: tool_router_logger.ToolRouterGuardrail
|
|
# mode: pre_call
|
|
# default_on: false
|
|
# top_k: 3
|
|
# min_similarity: 0.15
|
|
# similarity_mode: jaccard
|
|
#
|
|
# Parâmetros configuráveis via env vars:
|
|
# TOOL_ROUTER_EMBEDDING_MODEL, TOOL_ROUTER_TOP_K, TOOL_ROUTER_MIN_SIMILARITY,
|
|
# TOOL_ROUTER_ALWAYS_INCLUDE (csv), TOOL_ROUTER_SIMILARITY_MODE
|
|
# =============================================================================
|
|
|
|
def _router_params_from_env() -> dict:
|
|
"""Lê os parâmetros do router a partir de env vars (com defaults sãos)."""
|
|
return {
|
|
"embedding_model": os.environ.get("TOOL_ROUTER_EMBEDDING_MODEL", "text-embedding-3-small"),
|
|
"top_k": int(os.environ.get("TOOL_ROUTER_TOP_K", "3")),
|
|
"min_similarity": float(os.environ.get("TOOL_ROUTER_MIN_SIMILARITY", "0.15")),
|
|
"always_include_tools": set(
|
|
filter(None, os.environ.get("TOOL_ROUTER_ALWAYS_INCLUDE", "get_time,help,list_tools").split(","))
|
|
),
|
|
"similarity_mode": os.environ.get("TOOL_ROUTER_SIMILARITY_MODE", "jaccard"),
|
|
}
|
|
|
|
|
|
# Instância global reutilizável — é o que o proxy carrega via:
|
|
# callbacks: ["tool_router_logger.tool_router_instance"]
|
|
tool_router_instance: CustomToolRouter = CustomToolRouter(**_router_params_from_env())
|
|
|
|
|
|
def get_custom_tool_router(**kwargs) -> CustomToolRouter:
|
|
"""
|
|
Factory function para uso programático.
|
|
|
|
Se nenhum kwarg for passado, retorna a instância global
|
|
(equivalente a `tool_router_instance`). Se kwargs forem passados,
|
|
cria uma instância dedicada com aqueles parâmetros.
|
|
"""
|
|
global tool_router_instance
|
|
if kwargs:
|
|
return CustomToolRouter(**kwargs)
|
|
return tool_router_instance
|
|
|
|
|
|
class ToolRouterGuardrail(CustomGuardrail):
|
|
"""
|
|
Wrapper para usar o ToolRouter como guardrail do LiteLLM Proxy.
|
|
|
|
Guardrails permitem ativar o roteamento de tools por request:
|
|
curl ... -d '{"model": "gpt-4o-mini", "guardrails": ["tool_router"], ...}'
|
|
|
|
Ou globalmente com `default_on: true` no config.yaml.
|
|
"""
|
|
|
|
_ROUTER_PARAMS = (
|
|
"embedding_model", "top_k", "min_similarity",
|
|
"always_include_tools", "similarity_mode",
|
|
"cache_host", "cache_port", "cache_password",
|
|
)
|
|
|
|
def __init__(self, guardrail_name: str = "tool_router", **kwargs):
|
|
# `guardrail_name`, `event_hook`, `default_on` e demais campos do
|
|
# guardrail são repassados à base; os parâmetros do router são
|
|
# extraídos para criar a instância interna.
|
|
super().__init__(guardrail_name=guardrail_name, **kwargs)
|
|
router_params = {k: v for k, v in kwargs.items() if k in self._ROUTER_PARAMS}
|
|
self._router = CustomToolRouter(**router_params)
|
|
|
|
async def async_pre_call_hook(
|
|
self,
|
|
user_api_key_dict: UserAPIKeyAuth,
|
|
cache: DualCache,
|
|
data: dict,
|
|
call_type: str,
|
|
) -> Optional[Union[Exception, str, dict]]:
|
|
"""Delega o filtro de tools ao CustomToolRouter."""
|
|
return await self._router.async_pre_call_hook(
|
|
user_api_key_dict=user_api_key_dict,
|
|
cache=cache,
|
|
data=data,
|
|
call_type=call_type,
|
|
)
|
|
|
|
async def async_log_success_event(self, *args, **kwargs):
|
|
pass |