Loop de Chamada de Agentes de IA Personalizados Pós-Demo e Estratégias de Controle de Custo de Tokens

Computing/Software19 min read

TuBrief Editorial Written with AI assistance from the source video. The video is the authority. Source video: A decisão entre "Construir ou Comprar" está se inclinando para "construir" · Maximilian Schwarzmüller

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.


Três Obstáculos Enfrentados no Ambiente de Produção

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.


Sistema de Observabilidade para Interromper Loops Infinitos e Bombas de Custo

É 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:

  1. Preparação do backend Redis Vector Search: Inicie uma instância do Redis para converter e armazenar o prompt de entrada como um vetor de embedding.
  2. Avaliando similaridade e retornando o cache: Se a similaridade de cosseno entre o vetor do novo prompt e os vetores armazenados for igual ou superior a 0,92, a resposta armazenada é retornada imediatamente sem chamar a LLM externa.
  3. Armazenamento no Cache Miss: Somente quando a similaridade for inferior a 0,92, a API externa é chamada, e a resposta recebida é atualizada no Redis.

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.


Separação de Versões de Prompt e Automação de CI/CD

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:

  • role: "system"
    content: |
    당신은 사내 데이터베이스 분석 에이전트입니다.
    지정된 JSON 스키마에 맞춰 답변해야 하며, 정보가 부족할 경우 추가 질문을 생성하십시오.
  • role: "user"
    content: "{{ user_input }}"
    `

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

1. Definição dos critérios de avaliação

correctness_metric = GEval(
name="정확성 및 스키마 준수",
criteria="LLM 응답이 질문에 정확히 답변하고, 요청된 JSON 형식을 완벽히 준수하는가?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)

2. Teste de regressão baseado em Pytest

@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:

  1. Criação do conjunto de dados de avaliação: Crie um conjunto de consultas do domínio principal e respostas corretas, e defina a pontuação de corte (Threshold).
  2. Integração com fluxo de trabalho do GitHub Actions: Configure as actions para que o comando deepeval test run seja executado automaticamente ao abrir um PR após a modificação do código.
  3. Aplicação de Shadow Testing: Antes de integrar à branch principal, duplique e envie solicitações do tráfego real de forma assíncrona para comparar a qualidade da resposta e o tempo de latência com a versão anterior antes da implantação final.

`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
`


Aplicação de Mascaramento e Controle de Acesso

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

1. Registro do Analyzer Engine e regras de ID de funcionário

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)

2. Inicialização do Anonymizer Engine

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.


Ciclo de Vida da Infraestrutura e Garantia de Consistência por Etapas

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 │ └─────────────────────────────────────────────────────────────────────────┘


Sequência de Estabilização Operacional

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.

Related Video

A decisão entre "Construir ou Comprar" está se inclinando para "construir"

Maximilian Schwarzmüller10:07

More from the community

Comments

Log in to leave a comment.

No comments yet.