Loop de Chamada de Agentes de IA Personalizados Pós-Demo e Estratégias de Controle de Custo de Tokens
makedream28 Juli 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
Pegar um framework e criar um agente de IA com dados internos da empresa flui perfeitamente até a fase de protótipo. O verdadeiro problema começa no momento em que colocamos esse código em produção. Um agente que funcionava perfeitamente na tela fica preso por causa de uma única entrada inesperada, entra em um loop infinito de chamadas e, durante o fim de semana, gera uma fatura de API de milhares de dólares.
A maioria das falhas que ocorrem em campo acontece não por causa das limitações do modelo de LLM em si, mas pela falha no gerenciamento de estado e pela ausência de controle externo. É preciso acabar com a situação em que, após criar uma ferramenta personalizada por ordem da diretoria, você passa o dia inteiro depurando alucinações de agentes que surgem do nada e lidando com faturas astronômicas de tokens, sem conseguir focar no seu trabalho principal.
Agentes que saem do ambiente de demonstração esbarram em três barreiras: operação em caixa-preta, loops não determinísticos e risco de vazamento de dados.
Se as entradas e saídas de cada etapa — como o processo de raciocínio (Chain-of-Thought), chamadas de ferramentas externas (Tool Call) e buscas em bancos de dados vetoriais — não forem registradas em logs, será impossível encontrar a causa quando ocorrer um problema. Leva-se um dia inteiro apenas para rastrear onde o prompt se perdeu.
Mais grave ainda é o loop infinito. Agentes com arquitetura ReAct (Reasoning + Acting) reenviam repetidamente a mesma solicitação quando o resultado da execução de uma ferramenta é ambíguo.
┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘
No momento em que o sistema não consegue sair da condição de término e continua bombardeando a API com os mesmos argumentos, o orçamento mensal de tokens se esgota em questão de minutos. Falhas em cadeia, em que o agente principal e os subagentes chamam uns aos outros e tentam novamente, representam mais de 30% de todas as falhas do sistema.
Também surge o problema em que o histórico de conversa continua acumulando, preenchendo a janela de contexto e reduzindo o próprio desempenho do modelo. Além disso, prompts codificados como strings diretamente no código Python exigem que todo o sistema seja reconstruído e reimplantado até mesmo para alterar um simples trecho de texto.
É necessário separar as soluções de depuração local das ferramentas de observabilidade de produção para definir os pontos de coleta. Na fase de desenvolvimento, utilizamos o Arize Phoenix baseado em OpenTelemetry para verificar a qualidade das incorporações (embeddings) de RAG e as chamadas de ferramentas. No ambiente de produção, conectamos o Langfuse usando o ClickHouse como backend para monitorar logs de rastreamento e o consumo de tokens em tempo real.
┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘
Os custos de API gerados pela repetição de consultas idênticas são bloqueados por uma camada de Caching Semântico (Semantic Caching).
De acordo com dados de análise de 60.000 consultas publicados pela AWS, a implementação de um cache semântico baseado no critério de similaridade de cosseno reduziu os custos de inferência de LLM em até 86% e melhorou a latência de resposta em 88%. O framework RouteLLM da LMSYS também dividiu as solicitações entre modelos de alto e baixo custo de acordo com a dificuldade da pergunta, economizando 85% dos custos.
O caching semântico é construído na seguinte ordem:
Para bloquear fundamentalmente os loops infinitos, é necessário inserir um padrão de Circuit Breaker para validar o estado do agente diretamente no fluxo de controle do pipeline. Depender das opções padrão de tentativa (retry) fornecidas pelo framework fará com que o serviço inteiro caia quando ocorrer uma exceção. É mais seguro gerar um hash SHA-256 do nome da ferramenta e dos valores de argumento, adicioná-los a uma lista e, se o mesmo hash se acumular 3 vezes consecutivas, redirecionar forçadamente para outro nó.
`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]
def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()
def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. Se o número total de execuções exceder 5, ir para o nó de rota de fuga definido
if state["steps"] > 5:
return "fallback_graceful_node"
# 2. Bloquear imediatamente se a mesma Tool e Argument forem chamados 3 vezes consecutivas
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
return "fallback_graceful_node"
# 3. Verificar palavra-chave de término normal
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
return "end"
return "continue_tools"
`
Independentemente do estado de saída não determinístico da LLM, este roteador valida as estruturas condicionais no ambiente de execução Python, impedindo fisicamente a entrada em loops causados por alucinações.
Deixar os prompts abandonados dentro do código impede a captura de falhas de regressão (Regression Failures), nas quais funcionalidades que antes funcionavam bem param de operar após a modificação do modelo. Os prompts devem ser extraídos do código Python e gerenciados em arquivos YAML independentes.
`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:
Os prompts separados dessa forma são testados automaticamente no pipeline de CI/CD combinando frameworks de avaliação de código aberto como DeepEval e Pytest. Definimos as métricas do G-Eval como ponto de referência e validamos os valores de resposta do modelo antes do implantação.
`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams
correctness_metric = GEval(
name="정확성 및 스키마 준수",
criteria="LLM 응답이 질문에 정확히 답변하고, 요청된 JSON 형식을 완벽히 준수하는가?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)
@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024년 1분기 매출 데이터를 요약해줘.", "1분기 총 매출은 50억 원입니다."),
("퇴직금 계산 규정을 알려줘.", "퇴직금은 근속연수 1년에 대해 30일분 이상의 평균임금입니다.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)
test_case = LLMTestCase(
input=user_input,
actual_output=actual_output,
expected_output=expected_output
)
# Interromper o build se a pontuação for inferior ao critério definido
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])
`
O sistema de avaliação é integrado em três etapas:
deepeval test run seja executado automaticamente ao abrir um PR após a modificação do código.`yaml
name: AI Agent Evaluation Gate
on:
pull_request:
branches: [ main ]
jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`
Para evitar o vazamento de dados internos da empresa, são necessárias barreiras de proteção (guardrails) que desidentifiquem informações de identificação pessoal (PII) antes que a solicitação de API saia para o exterior. Integrar o motor Microsoft Presidio ao gateway permite mascarar automaticamente números de registro de residente, e-mails, números de telefone e IDs de funcionários.
`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)
anonymizer = AnonymizerEngine()
def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)
anonymized_result = anonymizer.anonymize(
text=raw_prompt,
analyzer_results=results,
operators={
"DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
"EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
}
)
return anonymized_result.text
`
Mesmo ao conectar a busca de RAG, devem ser tomadas medidas para trazer apenas os documentos correspondentes ao nível de permissão do usuário. Bancos de dados vetoriais como Qdrant ou Pinecone lidam com a separação de permissões por meio de Filtragem de Metadados (Metadata Filtering) ao realizar buscas de similaridade.
`python
from qdrant_client import QdrantClient
from qdrant_client.http import models
client = QdrantClient(host="localhost", port=6333)
def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`
A defesa contra Prompt Injection baseia-se em separar completamente o canal de prompt do sistema do canal de entrada do usuário. Tarefas perigosas que modificam bancos de dados ou chamam APIs externas devem passar por um processo de aprovação humana direta (Human-in-the-Loop) ou ter suas permissões de execução restritas a uma sandbox isolada para garantir a segurança.
Para substituir os módulos SaaS e continuar operando com sistemas internos, é necessária uma rotina de verificação periódica.
| Ciclo | Item de Verificação Operacional | Tarefas Detalhadas |
|---|---|---|
| Diário | Taxa de erro e uso de tokens | Verificar a taxa de erro HTTP 5xx e o consumo de tokens por departamento no painel do Langfuse |
| Diário | Histórico de bloqueio do Circuit Breaker | Coletar e corrigir padrões de nomes de ferramentas e argumentos de sessões interrompidas por detecção de loop |
| Semanal | Extração de consultas com falha na busca de RAG | Selecionar solicitações com pontuação de similaridade inferior a 0,6 e adicioná-las ao conjunto de respostas de teste |
| Semanal | Validação de falsos positivos no mascaramento de PII | Amostrar logs de processamento do Presidio para verificar se há omissões no mascaramento |
| Mensal | Atualização dos critérios de avaliação do CI/CD | Modificar casos de teste automáticos para refletir alterações nos negócios |
Se você está em dúvida entre o serviço do próprio modelo (baseado em vLLM) e a assinatura de APIs externas, faça o cálculo com base no tráfego diário.
O custo mensal de utilização de uma instância AWS EC2 g5.2xlarge com 1 GPU NVIDIA A10G é de cerca de $880. Por outro lado, o custo unitário da API do GPT-4o é da ordem de $2.50 por 1 milhão de tokens de entrada e $10.00 por 1 milhão de tokens de saída. Se o tráfego diário for inferior a 50 milhões de tokens, a assinatura de API é mais vantajosa, considerando o esforço de gerenciamento do engenheiro de DevOps e os custos fixos da GPU. Quando o tráfego diário ultrapassar 100 milhões de tokens ou o isolamento perfeito da rede interna for essencial, torna-se adequado migrar para o serviço próprio baseado em vLLM.
Para se preparar contra falhas no modelo principal, crie uma estrutura de roteamento de backup em 4 níveis.
┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘
O trabalho de aprimoramento do sistema de agentes personalizados deve ser realizado em blocos de 90 dias.
| Período | Objetivo de Aplicação | Conteúdo Específico do Trabalho |
|---|---|---|
| Dias 1 a 30 | Visibilidade e proteção de dados | Instalação do motor de mascaramento Presidio e integração do SDK de logging do Langfuse |
| Dias 31 a 60 | Bloqueio de loops e redução de custos | Aplicação de caching semântico Redis e conexão do Circuit Breaker baseado em Python |
| Dias 61 a 90 | Automação de validação e controle de acesso | Gerenciamento de prompts via Git, integração CI/CD do DeepEval e aplicação de metadados RBAC no Qdrant |
No primeiro mês, previna o vazamento de informações pessoais e estabeleça uma solução de observabilidade para que todas as solicitações fiquem registradas em logs. No segundo mês, adicione o caching de prompts repetidos e o roteador de bloqueio de loops para cortar despesas inesperadas. No último mês, ao concluir o gerenciamento de versões de prompts e o sistema de avaliação automatizada de testes, será possível garantir uma operação estável sem a preocupação de maus funcionamentos dos agentes a cada implantação.