AI Engineer: construa um agente de IA real, módulo a módulo

O que é um LLM e como ele gera texto?

Um Large Language Model é uma rede neural treinada para prever o próximo token dado um contexto. Ele não "entende" como nós entendemos — aprendeu padrões estatísticos em trilhões de tokens de texto. Cada vez que gera uma palavra, está calculando uma distribuição de probabilidade sobre o vocabulário e amostrando dela.

Arquitetura Transformer — o que você precisa saber

Você não precisa implementar um Transformer, mas precisa entender suas implicações práticas:

Conceitos-chave e seu impacto em produção

• Self-attention: cada token "presta atenção" a todos os outros do contexto. Implicação: o modelo consegue relacionar informações que estão distantes no texto.
• Janela de contexto: limite máximo de tokens ativos. GPT-4: 128K, Claude 3.5: 200K, Gemini 1.5 Pro: 1M. Mais contexto = mais custo e latência.
• Decoder-only: os modelos modernos (GPT, Claude, Gemini) só geram — não têm um encoder separado. Processam todo o contexto cada vez que geram um token.
• Tokenização (BPE): o texto é dividido em subpalavras. "tokenização" pode virar 3-4 tokens. O custo real de uma chamada depende do número de tokens, não de palavras nem de caracteres.

Parâmetros de inferência — o painel de controle

Quando você faz uma chamada à API, estes parâmetros determinam como o modelo amostra a resposta:

temperature

0 = determinístico (sempre o token mais provável). 1 = mais variado. Para agentes em produção: use 0–0.3. Para geração criativa: 0.7–1.0.

top_p (nucleus sampling)

Considera apenas os tokens cuja probabilidade acumulada chega a p. top_p=0.9 ignora os 10% de tokens menos prováveis. É uma alternativa à temperature; não se usam juntos.

max_tokens

Limite de tokens na resposta. Impacta diretamente custo e latência. Para respostas estruturadas (JSON), um limite baixo evita respostas truncadas de forma inesperada.

stop_sequences

O modelo para de gerar quando encontra essa string. Útil para delimitar outputs: ["</response>", "###"]. Mais confiável que max_tokens para outputs estruturados.

Seleção de modelo — o trade-off mais importante

Guia de seleção para produção 2025

• Claude 3 Haiku / GPT-4o-mini: classificação, routing, extração simples. ~$0.25/M tokens. Latência: <1s. Use quando o erro tem baixo impacto.
• Claude 3.5 Sonnet / GPT-4o: raciocínio, geração, tool calling complexo. ~$3/M tokens. Equilíbrio ideal para a maioria dos agentes em produção.
• Claude 3 Opus / GPT-4-turbo: análise profunda, decisões de alto impacto. ~$15/M tokens. Só quando a qualidade é crítica e o custo é secundário.
• Mistral / LLaMA 3 (self-hosted): dados sensíveis, conformidade regulatória, custo em escala extrema. Exige infraestrutura própria.

LLMClient — wrapper base do sistema

O padrão correto não é chamar o SDK diretamente de cada agente. Cria-se um wrapper centralizado que cuida de: retry automático, logging estruturado, contagem de tokens e seleção de modelo.

from anthropic import Anthropic
from tenacity import retry, stop_after_attempt, wait_exponential
from enum import Enum
import structlog, time

log = structlog.get_logger()

class ModelTier(Enum):
    FAST     = "claude-3-haiku-20240307"      # barato y rápido
    STANDARD = "claude-3-5-sonnet-20241022"  # balance ideal
    POWERFUL = "claude-3-opus-20240229"      # máxima calidad

class LLMResponse:
    text: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: float

class LLMClient:
    def __init__(self):
        self.client = Anthropic()
        self.cost_per_token = {
            ModelTier.FAST:     (0.00025, 0.00125),   # (input, output) por 1K tokens
            ModelTier.STANDARD: (0.003,   0.015),
            ModelTier.POWERFUL: (0.015,   0.075),
        }

    @retry(stop=stop_after_attempt(3),
           wait=wait_exponential(multiplier=1, min=1, max=10))
    def call(
        self,
        messages: list[dict],
        model: ModelTier = ModelTier.STANDARD,
        temperature: float = 0.3,
        max_tokens: int = 1024,
        trace_id: str = None,
    ) -> LLMResponse:
        start = time.time()

        response = self.client.messages.create(
            model=model.value,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )

        latency = (time.time() - start) * 1000
        cost = self._calculate_cost(model, response.usage)

        # Log estructurado para observabilidad
        log.info("llm_call",
            trace_id=trace_id,
            model=model.value,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            cost_usd=round(cost, 6),
            latency_ms=round(latency, 1),
        )

        return LLMResponse(
            text=response.content[0].text,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            cost_usd=cost,
            latency_ms=latency,
        )

    def _calculate_cost(self, model, usage) -> float:
        inp, out = self.cost_per_token[model]
        return (usage.input_tokens/1000*inp) + (usage.output_tokens/1000*out)

O decorator @retry do tenacity lida automaticamente com rate limits (429) e erros temporários da API com backoff exponencial. Sem isso, qualquer falha transitória quebra o fluxo do agente.

Recursos

anthropic-sdk docs, tenacity, structlog, tiktoken

O system prompt é a constituição do agente

O system prompt não é uma "instrução inicial" — é o documento que define por completo quem é o agente, o que ele pode fazer, como deve se comportar e quando deve pedir ajuda. Um agente sem um system prompt bem estruturado é um agente imprevisível.

Estrutura em 6 seções (todas obrigatórias)

• IDENTITY: nome, propósito, personalidade. Define o "quem" do agente.
• CAPABILITIES: lista explícita do que ele pode e NÃO pode fazer. Os "não pode" são tão importantes quanto os "pode".
• CONTEXT: variáveis dinâmicas do ambiente: usuário, estado, ferramentas disponíveis.
• BEHAVIOR RULES: como agir em situações específicas — edge cases explícitos.
• OUTPUT FORMAT: estrutura exata, tamanho e canal da resposta.
• ESCALATION: critérios exatos para transferir para um humano.

Chain-of-Thought controlado — separar raciocínio de resposta

O CoT (Chain-of-Thought) melhora a qualidade do raciocínio, mas em produção não queremos mostrar o processo interno ao usuário. O padrão correto é separar os dois:

CoT sem controle

• O usuário vê o raciocínio interno
• Expõe lógica que pode ser manipulada
• Aumenta tokens de output sem necessidade
• Dificulta o parsing da resposta final

CoT com <thinking> separado

• O raciocínio fica em logs internos
• Permite debugging sem exposição ao usuário
• O output final é limpo e parseável
• Você pode monitorar a qualidade do raciocínio

Few-shot com exemplos negativos

Os exemplos positivos ensinam o comportamento esperado. Os exemplos negativos são igualmente críticos — mostram ao modelo exatamente o que evitar. Sem eles, o modelo pode cair em respostas "razoáveis, mas incorretas".

Prompts como código — versionamento e testes

Um prompt que muda sem controle é uma regressão silenciosa. A mesma disciplina que aplicamos ao código vale para os prompts:

Pipeline de deployment de prompts

• PR no git com a mudança de prompt + justificativa na descrição
• Avaliação automática em CI contra o test set base
• Deploy em staging → 10% do tráfego → 48h de monitoramento
• Se as métricas estiverem OK → promover para 100%. Se piorarem → rollback automático

System prompt completo com CoT controlado

IDENTITY:
Eres SupportBot, asistente de atención al cliente de Nexus.
Objetivo: resolver consultas de soporte con empatía y precisión.
Tono: cercano, claro, sin jerga técnica innecesaria.

CAPABILITIES:
✓ Puedes: get_order_status, create_ticket, send_notification, schedule_callback
✗ NO puedes: modificar precios, eliminar cuentas, acceder a datos de pago

CONTEXT:
Usuario: {{user_name}} | Plan: {{plan_name}} | Estado: {{account_status}}
Canal: {{channel}} | Herramientas: {{available_tools}}

BEHAVIOR RULES:
- Lenguaje agresivo → desescalar sin confrontar: "Entiendo tu frustración,
  mi objetivo es encontrar una solución que funcione para ti."
- Solicitud fuera de alcance → explicar límite + ofrecer alternativa real
- Input ambiguo → preguntar UNA cosa antes de actuar
- Señal de crisis o urgencia alta → escalar a humano inmediatamente

ESCALATION:
Transferir SIEMPRE cuando: ticket_priority="critical" OR usuario solicita
hablar con persona OR confidence_score < 0.70

REASONING FORMAT:
Antes de responder, razona en <thinking>:
1. ¿Qué pide exactamente el usuario?
2. ¿Qué información tengo vs qué me falta?
3. ¿Qué regla de comportamiento aplica?
4. ¿Debo escalar o puedo resolver?
El contenido de <thinking> NO se muestra al usuario.

OUTPUT FORMAT:
- Máx 3 oraciones por turno (canal: chat/WhatsApp)
- Cuando uses herramienta: responde SOLO JSON válido sin texto adicional:
  {"action": "<tool>", "params": {...}, "reason": "<1 oración>", "confidence": 0.0-1.0}

PromptLoader — gestão centralizada de versões

import os
from pathlib import Path
from jinja2 import Template

PROMPTS_DIR = Path("prompts")

class PromptLoader:
    def get(self, agent: str, version: str, context: dict) -> str:
        """Carga un prompt versionado e inyecta variables de contexto."""
        path = PROMPTS_DIR / agent / f"v{version}_system.txt"
        template_str = path.read_text(encoding="utf-8")
        return Template(template_str).render(**context)

    def latest(self, agent: str) -> str:
        """Lee la versión actual desde el archivo VERSION del agente."""
        version_file = PROMPTS_DIR / agent / "VERSION"
        return version_file.read_text().strip()

    def load_latest(self, agent: str, context: dict) -> str:
        """Atajo: carga siempre la versión más reciente."""
        return self.get(agent, self.latest(agent), context)

# Uso en un agente:
loader = PromptLoader()
system_prompt = loader.load_latest("support_agent", {
    "user_name": "Ana López",
    "plan_name": "Pro",
    "account_status": "active",
    "channel": "whatsapp",
    "available_tools": "[get_order_status, create_ticket]",
})

Recursos

jinja2, jsonlines, Anthropic prompting guide, Learn Prompting

O que é um agente LLM?

Um agente é um sistema em que o LLM não só gera texto — ele também toma decisões sobre quais ações executar, observa os resultados dessas ações e decide o que fazer em seguida. A diferença para um simples chatbot é que o agente tem agência: pode agir sobre o mundo.

Padrão ReAct — Reason + Act

ReAct é o padrão mais usado em produção. A cada turno, o agente: (1) raciocina sobre o estado atual, (2) decide uma ação, (3) observa o resultado e repete até ter informação suficiente para responder ao usuário.

Ciclo ReAct passo a passo

• Thought: "O usuário pergunta pelo pedido. Preciso chamar get_order_status com o ID dele."
• Action:get_order_status(order_id="ORD-123")
• Observation:{"status": "en camino", "eta": "mañana 14:00"}
• Thought: "Tenho a informação necessária. Posso responder ao usuário."
• FINISH: "Seu pedido está a caminho e vai chegar amanhã antes das 14:00."

Critic loop — o agente avalia o próprio output

O Critic é um segundo agente (ou uma segunda chamada ao LLM) que avalia a resposta do agente principal antes de enviá-la ao usuário. Responde PASS/FAIL + motivo. Dobra o custo, mas aumenta significativamente a qualidade em casos de alto impacto.

from dataclasses import dataclass, field
from enum import Enum

class AgentAction(Enum):
    FINISH   = "FINISH"
    ESCALATE = "ESCALATE"
    TOOL     = "TOOL"

@dataclass
class AgentThought:
    reasoning: str          # contenido del <thinking>
    action: AgentAction
    tool_name: str | None = None
    tool_params: dict      = field(default_factory=dict)
    final_answer: str | None = None
    confidence: float       = 1.0

class SupportAgent:
    max_iterations = 8

    def __init__(self, llm_client, tool_registry, prompt_loader):
        self.llm    = llm_client
        self.tools  = tool_registry
        self.loader = prompt_loader

    def run(self, user_message: str, context: dict) -> AgentResult:
        system = self.loader.load_latest("support_agent", context)
        history = []

        for i in range(self.max_iterations):
            # Paso 1: REASON — el agente piensa qué hacer
            messages = self._build_messages(system, user_message, history)
            response = self.llm.call(messages, trace_id=context["trace_id"])
            thought  = self._parse_thought(response.text)

            # Paso 2: verificar stopping criteria
            if thought.action == AgentAction.FINISH:
                return AgentResult(answer=thought.final_answer, iterations=i+1)

            if thought.action == AgentAction.ESCALATE:
                return AgentResult(escalate=True, reason=thought.reasoning, iterations=i+1)

            # Paso 3: ACT — ejecutar la herramienta
            observation = self.tools.execute(thought.tool_name, thought.tool_params)

            # Paso 4: OBSERVE — agregar al historial
            history.append({"thought": thought, "observation": observation})

        # MAX_ITERATIONS alcanzado → siempre escalar, nunca lanzar excepción
        return AgentResult(escalate=True, reason="max_iterations_reached", iterations=self.max_iterations)


class CriticAgent:
    def evaluate(self, agent_result: AgentResult, original_query: str) -> CriticVerdict:
        """Evalúa si la respuesta del agente es correcta antes de enviarla."""
        prompt = f"""
Evalúa esta respuesta de soporte:
Consulta original: {original_query}
Respuesta del agente: {agent_result.answer}

Responde SOLO con JSON:
{{"status": "PASS" o "FAIL", "reason": "...", "suggestion": "..."}}
"""
        response = self.llm.call([{"role": "user", "content": prompt}],
                                  model=ModelTier.FAST)  # Haiku para el critic = más barato
        return CriticVerdict(**json.loads(response.text))

Recursos

ReAct paper (Yao 2022), Anthropic tool use, pydantic v2

Hub-and-spoke: o padrão mais robusto para produção

Um orquestrador central recebe todas as mensagens, classifica-as com um agente leve (barato e rápido) e delega ao agente especializado correto com o contexto completo.

O classificador é a peça mais crítica do sistema

• Use o modelo mais barato: Haiku com um prompt de 5 linhas classifica melhor que Sonnet com um prompt ambíguo
• Categorias exaustivas: toda consulta precisa cair em alguma categoria — inclua "GENERAL/OTHER"
• Output tipado: o classifier nunca retorna texto livre — retorna um enum com a categoria
• Fallback seguro: se o classifier falhar, o sistema roteia para o agente geral — nunca quebra

class Intent(Enum):
    ORDER_STATUS  = "order_status"
    CREATE_TICKET = "create_ticket"
    ESCALATE      = "escalate"
    GENERAL       = "general"

@dataclass
class AgentHandoff:
    """Contexto completo que pasa entre agentes en un handoff."""
    user_id: str
    user_message: str
    intent: Intent
    conversation_history: list[dict]
    previous_actions: list[str]   # qué ya intentó el agente anterior
    context: dict                  # datos del usuario (plan, status, etc.)
    trace_id: str

class Orchestrator:
    global_timeout = 30  # segundos — nunca un workflow dura más

    def handle(self, user_id: str, message: str) -> OrchestratorResponse:
        context = self.context_builder.build(user_id)
        trace_id = self._new_trace_id()

        # 1. Clasificar intención con modelo barato (Haiku)
        intent = self.classifier.classify(message, context)

        # 2. Construir handoff con contexto completo
        handoff = AgentHandoff(
            user_id=user_id, user_message=message, intent=intent,
            conversation_history=self.session.get_history(user_id),
            previous_actions=[], context=context, trace_id=trace_id
        )

        # 3. Routing al agente correcto con timeout global
        with timeout(self.global_timeout):
            agent = self.router[intent]
            result = agent.run(handoff)

        # 4. Evaluar si escalar antes de responder al usuario
        verdict = self.escalation_router.evaluate(result, context)
        if verdict.should_escalate:
            return self._escalate(handoff, verdict.reason)

        return OrchestratorResponse(message=result.answer, trace_id=trace_id)

Recursos

signal (timeout), claude-3-haiku, pydantic

O problema da memória em LLMs

Por padrão, um LLM não lembra de nada entre sessões. Cada chamada à API é stateless. Para um agente de suporte, isso é um problema: o usuário não deveria ter que repetir o problema dele a cada interação.

Os 4 tipos de memória e quando usar cada um

• Short-term (janela ativa): histórico do turno atual no contexto do LLM. Sem custo adicional, se perde ao encerrar a sessão.
• Long-term (vector store): knowledge base do domínio. Busca semântica por similaridade de embeddings. Para documentação, FAQs, políticas.
• Episodic (histórico de interações): o que o usuário disse em sessões anteriores. Banco de dados estruturado com timestamp.
• Semantic (entidades do usuário): dados persistentes: plano, preferências, histórico de tickets. Structured DB.

Pipeline RAG — como funciona em produção

Os 4 passos do pipeline

• Ingestion: documento → chunking (512 tokens, 10% de overlap) → embedding → vector store + metadata
• Retrieval: query → embed → ANN search (top-10) → filtro por metadata → reranking → top-3
• Augmentation: chunks recuperados → injetar no contexto do LLM
• Evaluation: a resposta usa os chunks? Os chunks eram relevantes?

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.vector_stores.postgres import PGVectorStore
import redis

class MemoryManager:
    def __init__(self, pg_conn_str: str, redis_url: str):
        self.vector_store = PGVectorStore.from_params(pg_conn_str, embed_dim=1536)
        self.session      = redis.from_url(redis_url)
        self.index        = VectorStoreIndex.from_vector_store(self.vector_store)

    def get_context(self, query: str, user_id: str) -> MemoryContext:
        return MemoryContext(
            # Short-term: historial de la sesión actual
            short_term = self._get_session(user_id),

            # Long-term: knowledge base del dominio (RAG)
            long_term  = self._retrieve_relevant(query, k=3),

            # Episodic: últimas 3 interacciones del usuario
            episodic   = self._get_recent_episodes(user_id, n=3),
        )

    def _retrieve_relevant(self, query: str, k: int) -> list[str]:
        retriever = self.index.as_retriever(similarity_top_k=k*3)  # más para re-rankear
        nodes = retriever.retrieve(query)
        # Reranking: ordenar por relevancia real, no solo similitud vectorial
        reranked = sorted(nodes, key=lambda n: n.score, reverse=True)[:k]
        return [n.text for n in reranked]

    def _get_session(self, user_id: str) -> list[dict]:
        raw = self.session.get(f"session:{user_id}")
        return json.loads(raw) if raw else []

    def save_turn(self, user_id: str, user_msg: str, agent_response: str):
        history = self._get_session(user_id)
        history.append({"user": user_msg, "agent": agent_response})
        self.session.setex(f"session:{user_id}", 3600, json.dumps(history))  # TTL 1h

Recursos

llama-index, pgvector, redis-py, Cohere Rerank, RAGAS (eval)

Human-in-the-loop não é edge case — é design

O erro mais comum é tratar o escalonamento como algo excepcional. Em produção, entre 10-30% das interações vão terminar em um humano. O sistema deve ser projetado para isso desde o início, não receber isso depois.

5 tipos de critérios de escalonamento

• Limite de negócio: ticket_priority="critical", account_type="enterprise"
• Confiança baixa: confidence_score < 0.70 na ação a ser tomada
• Fora do escopo: o agente não consegue resolver a solicitação
• Pedido explícito: o usuário pede para falar com uma pessoa
• Sinal emocional: crise, urgência extrema, frustração acumulada

Fallback em cascata — o sistema nunca morre

Um sistema de produção deve responder sempre, mesmo quando tudo falha. O padrão de fallback em cascata define uma cadeia de degradação gradual:

Cadeia de fallback

• Nível 1: SupportAgent com Sonnet (normal)
• Nível 2: SupportAgent com Haiku (mais rápido e barato se houver latência)
• Nível 3: Resposta genérica hardcoded + escalonamento automático para humano
• Nível 4: Mensagem de erro amigável com número de ticket criado automaticamente

from pybreaker import CircuitBreaker, CircuitBreakerError

# Circuit breaker por herramienta — evita cascada de fallos
ticket_breaker = CircuitBreaker(fail_max=5, reset_timeout=60)

class EscalationRule:
    name: str
    check: callable  # función que recibe (result, context) → bool
    reason: str

class EscalationRouter:
    rules: list[EscalationRule] = [
        EscalationRule("critical_ticket",
            lambda r, ctx: ctx.get("ticket_priority") == "critical", "ticket_critico"),
        EscalationRule("low_confidence",
            lambda r, ctx: r.confidence < 0.70, "confianza_baja"),
        EscalationRule("user_requested",
            lambda r, ctx: ctx.get("user_requested_human", False), "usuario_solicito"),
    ]

    def evaluate(self, result, context) -> Verdict:
        for rule in self.rules:
            if rule.check(result, context):
                audit_log.record("escalation", rule=rule.name)
                return Verdict(should_escalate=True, reason=rule.reason)
        return Verdict(should_escalate=False)

class FallbackChain:
    def run(self, handoff: AgentHandoff) -> AgentResult:
        try:
            return self.support_agent.run(handoff)         # Nivel 1: normal
        except (TimeoutError, RateLimitError):
            try:
                return self.support_agent_fast.run(handoff)  # Nivel 2: modelo barato
            except Exception:
                return self._static_fallback(handoff)        # Nivel 3: respuesta fija

    def _static_fallback(self, handoff) -> AgentResult:
        ticket_id = self._create_fallback_ticket(handoff)
        return AgentResult(
            answer=f"Estamos experimentando problemas técnicos. Creamos el ticket #{ticket_id} y un agente te contactará pronto.",
            escalate=True, reason="system_fallback"
        )

Recursos

pybreaker, structlog, pydantic

O agente nunca toca a infraestrutura diretamente

A regra mais importante do tool layer: o agente chama contratos (schemas tipados), não implementações. Isso permite trocar a implementação subjacente sem mexer no agente, e testar o agente com mocks sem infraestrutura real.

Anatomia de uma ferramenta bem projetada

• Schema tipado: model Pydantic com validação, descriptions e constraints
• Modo dry-run: validar sem executar efeitos — permite verificar antes de agir
• Timeout por ferramenta: cada tool tem seu próprio SLA — não o global do workflow
• Idempotência: executar a mesma ferramenta 2 vezes com os mesmos params = mesmo resultado
• Audit log: toda execução fica registrada — bem-sucedida ou com falha

from pydantic import BaseModel, Field
from typing import Literal

# 1. Schema tipado — lo que el LLM ve y debe rellenar
class CreateTicketParams(BaseModel):
    user_id:  str            = Field(description="ID único del usuario")
    subject:  str            = Field(min_length=5, description="Asunto del ticket")
    priority: Literal["low","medium","high","critical"]
    category: str            = Field(description="Categoría: billing, technical, general")
    notes:    str | None     = None

# 2. Implementación con todas las capas de seguridad
class CreateTicketTool:
    name    = "create_ticket"
    timeout = 5  # segundos

    def execute(self, raw_params: dict) -> dict:
        # Validación — lanza ValidationError si algo está mal
        params = CreateTicketParams(**raw_params)

        # Dry-run check — ¿hay conflicto con un ticket abierto?
        existing = self.ticket_service.get_open(params.user_id)
        if existing and existing.subject.lower() == params.subject.lower():
            return {"warning": "duplicate_ticket", "existing_id": existing.id}

        # Ejecución con timeout
        with timeout(self.timeout):
            result = self.ticket_service.create(params)

        # Audit log — inmutable
        audit_log.record(tool=self.name, params=params.dict(),
                         result={"ticket_id": result.id}, user_id=params.user_id)

        return {"ticket_id": result.id, "status": "created"}

# 3. Registry — el agente solo conoce el registry, no las implementaciones
class ToolRegistry:
    def __init__(self):
        self._tools = {
            "create_ticket":     CreateTicketTool(),
            "get_order_status":  GetOrderStatusTool(),
            "send_notification": SendNotificationTool(),
            "schedule_callback": ScheduleCallbackTool(),
        }

    def execute(self, name: str, params: dict) -> dict:
        if name not in self._tools:
            raise ValueError(f"Herramienta desconocida: {name}")
        return self._tools[name].execute(params)

    def get_schemas(self) -> list[dict]:
        # Genera los schemas para el API de Anthropic automáticamente
        return [t.get_anthropic_schema() for t in self._tools.values()]

Recursos

pydantic v2, redis-py, httpx (async)

Os 4 trade-offs que todo sênior precisa dominar

Latência vs Qualidade

Haiku responde em <500ms. Sonnet leva 1-3s. Opus pode levar 5-10s. A pergunta não é "qual é melhor", e sim "de qual o usuário precisa neste contexto".

Custo vs Profundidade

Sonnet custa 12x mais que Haiku. Para classificação (simples), Haiku basta. Para raciocínio complexo com ferramentas, Sonnet vale cada centavo.

Autonomia vs Controle

Mais autonomia = melhor experiência do usuário. Mais controle = menos risco de erros caros. A resposta depende da reversibilidade da ação.

Agente vs Pipeline

Se o fluxo sempre segue os mesmos passos, um DAG determinístico é mais rápido, barato e previsível. O agente agrega valor quando o input é ambíguo.

Model Routing — a otimização de maior impacto

O princípio: usar o modelo mais barato que resolve o caso corretamente. Um classificador leve (Haiku) decide de qual modelo cada consulta precisa. 70-80% das consultas de suporte são simples e podem ser resolvidas com Haiku.

Estratégias de redução de custo

• Prompt caching: a parte estática do system prompt fica em cache. A Anthropic oferece 90% de desconto em tokens em cache. Coloque sempre a parte estática primeiro.
• Context compression: resumir o histórico longo em vez de enviá-lo completo. Economia de 40-60% em conversas com muitos turnos.
• Batch API: 50% de desconto para tarefas não urgentes (avaliações, geração offline).
• O 80/20 do custo: 80% do gasto vem dos 20% de requisições mais longas. Otimize a cauda, não a média.

class ModelRouter:
    def select(self, query: str, context: dict) -> ModelTier:
        # Regla 1: casos críticos siempre al modelo estándar
        if context.get("ticket_priority") == "critical":
            return ModelTier.STANDARD

        # Regla 2: clasificar complejidad con el modelo más barato posible
        complexity_prompt = f"""Clasifica esta consulta: '{query}'
Responde SOLO con: SIMPLE o COMPLEX
SIMPLE: saludos, estado de pedido, preguntas de FAQ
COMPLEX: problemas técnicos, disputas, múltiples pasos"""

        response = self.llm.call(
            [{"role": "user", "content": complexity_prompt}],
            model=ModelTier.FAST,  # Haiku para clasificar
            max_tokens=5
        )

        if "SIMPLE" in response.text:
            return ModelTier.FAST      # Haiku: 10x más barato
        return ModelTier.STANDARD       # Sonnet: balance ideal


class ContextCompressor:
    max_history_tokens = 3000

    def compress(self, history: list[dict]) -> list[dict]:
        if self._count_tokens(history) <= self.max_history_tokens:
            return history  # No necesita compresión

        # Mantener los últimos 3 turnos intactos (más relevantes)
        recent = history[-3:]
        older  = history[:-3]

        # Resumir los turnos más antiguos
        summary_prompt = f"Resume en 2 oraciones los puntos clave de esta conversación: {older}"
        summary = self.llm.call([{"role": "user", "content": summary_prompt}],
                                 model=ModelTier.FAST)

        return [{"role": "system",
                  "content": f"Contexto previo (resumido): {summary.text}"}] + recent

Recursos

litellm, time.perf_counter, Anthropic prompt caching

Por que o debugging em sistemas probabilísticos é diferente

Em sistemas determinísticos, o mesmo input produz o mesmo output — sempre. Em sistemas com LLMs, o mesmo input pode produzir outputs ligeiramente diferentes a cada chamada. Isso muda completamente a estratégia de debugging.

Os 3 tipos de falha mais frequentes

• Alucinações: o modelo gera informação incorreta com alta confiança. Causa: contexto insuficiente ou constraints fracas no prompt. Mitigação: RAG + constraints explícitas + grounding checks.
• Loops: o agente repete a mesma ação indefinidamente. Causa: a ferramenta falha, mas o modelo não reconhece isso como erro. Mitigação: MAX_ITERATIONS + loop detector + circuit breaker.
• Degradação silenciosa: a qualidade cai gradualmente sem alerta visível. Causa: model drift do provedor ou prompt drift por edições acumuladas. Mitigação: avaliação contínua + alertas no dashboard.

Framework de debugging em 5 passos

O processo correto

• 1. Reproduzir: guardar o input completo (prompt, histórico, tool results, modelo, versão). Sem reprodutibilidade, o debugging é impossível.
• 2. Isolar: falha no planning, na execution ou na evaluation? Teste cada componente separadamente com inputs sintéticos.
• 3. Rastrear: revisar o <thinking> do agente. O raciocínio estava correto? Os dados estavam certos?
• 4. Quantificar: é um caso isolado ou sistêmico? Execute 20+ vezes antes de concluir.
• 5. Iterar: mudar UMA variável por vez. Sem teste A/B não há conclusões válidas.

import hashlib, json
from datetime import datetime

class LoopDetector:
    def __init__(self, window: int = 3):
        self.window = window  # comparar los últimos N estados

    def check(self, history: list) -> bool:
        if len(history) < self.window:
            return False
        # Si los últimos N thoughts son iguales → loop detectado
        last_n = history[-self.window:]
        hashes = [hashlib.md5(json.dumps(h["thought"].tool_name).encode()).hexdigest()
                  for h in last_n]
        return len(set(hashes)) == 1  # todos iguales = loop

class FailureStore:
    def capture(self, context: dict, error: Exception, agent_history: list) -> str:
        failure_id = f"fail-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
        record = {
            "id":             failure_id,
            "timestamp":      datetime.utcnow().isoformat(),
            "error_type":     type(error).__name__,
            "error_message":  str(error),
            "prompt_version": context.get("prompt_version"),
            "model":          context.get("model"),
            "full_context":   context,   # TODO: redactar PII antes de guardar
            "agent_history":  agent_history,
        }
        self.db.save(failure_id, json.dumps(record))
        return failure_id

    def replay(self, failure_id: str) -> dict:
        """Recupera el contexto completo para reproducir el fallo exactamente."""
        return json.loads(self.db.get(failure_id))

Recursos

sqlite3, pytest fixtures, hashlib

Você não consegue melhorar o que não mede

O módulo de LLMOps é o que fecha o ciclo. Sem observabilidade, o sistema é uma caixa-preta que funciona (ou não) sem que ninguém saiba por quê. Com observabilidade, cada decisão de melhoria é respaldada por dados.

As métricas que importam — e em que ordem

• Task completion rate: a métrica mais importante. Que % das conversas terminou com o problema do usuário resolvido?
• Escalation rate: % escalado para humano. Se sobe → o agente está piorando. Se cai demais → pode estar deixando passar casos que deveria escalar.
• Cost per successful interaction: (tokens × preço) / interações bem-sucedidas. A métrica de eficiência do sistema.
• Latência p95: o percentil 95 de latência — o que 95% dos usuários experimentam. A média mente.
• Tool error rate: % de chamadas a ferramentas que falham. Sinaliza problemas em APIs externas.

LLM-as-judge — avaliação automática escalável

Avaliar manualmente a qualidade de 1000 respostas por semana é inviável. O padrão LLM-as-judge usa um segundo LLM para avaliar o output do primeiro. O avaliador recebe: a consulta original, a resposta gerada e os critérios de avaliação.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

tracer = trace.get_tracer("nexus-support-agent")

class AgentTracer:
    def trace_request(self, trace_id: str, user_id: str):
        return tracer.start_as_current_span("agent_request",
            attributes={"trace_id": trace_id, "user_id": user_id})


class MetricsCollector:
    def record_interaction(self, result: AgentResult, context: dict):
        TASK_COMPLETION.inc(1 if result.success else 0)
        ESCALATION_RATE.inc(1 if result.escalated else 0)
        COST_COUNTER.inc(result.cost_usd)
        LATENCY_HISTOGRAM.observe(result.latency_ms)
        TOKEN_COUNTER.inc(result.total_tokens)


class LLMJudge:
    judge_prompt = """Evalúa esta respuesta de soporte.
Query: {query}
Respuesta: {response}

Puntúa 1-5 en cada criterio y responde SOLO JSON:
{{"relevance": 1-5, "accuracy": 1-5, "tone": 1-5, "completeness": 1-5,
  "overall": 1-5, "reasoning": "explicación breve"}}"""

    def evaluate(self, query: str, response: str) -> dict:
        result = self.llm.call(
            [{"role": "user", "content": self.judge_prompt.format(
                query=query, response=response)}],
            model=ModelTier.STANDARD  # el judge necesita buen criterio
        )
        return json.loads(result.text)


class EvalPipeline:
    def run(self, prompt_version: str) -> EvalReport:
        results = []
        for case in self.load_test_set():
            output = self.agent.run(case["input"], case["context"])
            score  = self.judge.evaluate(case["input"], output.answer)
            results.append({"case_id": case["id"], "score": score, "passed": score["overall"] >= 3})

        pass_rate = sum(1 for r in results if r["passed"]) / len(results)
        return EvalReport(results=results, pass_rate=pass_rate,
                           version=prompt_version, baseline=self.get_baseline())

Recursos

opentelemetry, langsmith, prometheus, grafana, presidio (PII)