AI Engineer: build a real AI agent, module by module

What is an LLM and how does it generate text?

A Large Language Model is a neural network trained to predict the next token given a context. It doesn't "understand" the way we do: it learned statistical patterns from trillions of tokens of text. Every time it generates a word, it computes a probability distribution over the vocabulary and samples from it.

The Transformer architecture: what you need to know

You don't need to implement a Transformer, but you do need to understand its practical implications:

Key concepts and their impact in production

• Self-attention: each token "attends" to every other token in the context. Implication: the model can relate information that is far apart in the text.
• Context window: the maximum number of active tokens. GPT-4: 128K, Claude 3.5: 200K, Gemini 1.5 Pro: 1M. More context = more cost and latency.
• Decoder-only: modern models (GPT, Claude, Gemini) only generate; they don't have a separate encoder. They process the whole context every time they generate a token.
• Tokenization (BPE): text is split into sub-words. "tokenization" can be 3-4 tokens. The real cost of a call depends on the number of tokens, not words or characters.

Inference parameters: the control dial

When you call the API, these parameters determine how the model samples its response:

temperature

0 = deterministic (always the most likely token). 1 = more varied. For agents in production: use 0–0.3. For creative generation: 0.7–1.0.

top_p (nucleus sampling)

Only considers the tokens whose cumulative probability reaches p. top_p=0.9 ignores the least likely 10% of tokens. An alternative to temperature; they aren't used together.

max_tokens

The token limit for the response. It directly affects cost and latency. For structured responses (JSON), a low limit prevents unexpectedly truncated responses.

stop_sequences

The model stops generating when it hits this string. Useful for delimiting outputs: ["</response>", "###"]. More reliable than max_tokens for structured outputs.

Model selection: the most important trade-off

Selection guide for production, 2025

• Claude 3 Haiku / GPT-4o-mini: classification, routing, simple extraction. ~$0.25/M tokens. Latency: <1s. Use it when errors have low impact.
• Claude 3.5 Sonnet / GPT-4o: reasoning, generation, complex tool calling. ~$3/M tokens. The ideal balance for most agents in production.
• Claude 3 Opus / GPT-4-turbo: deep analysis, high-impact decisions. ~$15/M tokens. Only when quality is critical and cost is secondary.
• Mistral / LLaMA 3 (self-hosted): sensitive data, regulatory compliance, cost at extreme scale. Requires your own infrastructure.

LLMClient: the system's base wrapper

The right pattern is not to call the SDK directly from each agent. You create a centralized wrapper that handles automatic retries, structured logging, token counting and model selection.

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)

The @retry decorator from tenacity automatically handles rate limits (429) and transient API errors with exponential backoff. Without it, any transient failure breaks the agent's flow.

Resources

anthropic-sdk docs, tenacity, structlog, tiktoken

The system prompt is the agent's constitution

The system prompt is not an "initial instruction". It is the document that fully defines who the agent is, what it can do, how it should behave and when it should ask for help. An agent without a well-structured system prompt is an unpredictable agent.

6-section structure (all required)

• IDENTITY: name, purpose, personality. Defines the "who" of the agent.
• CAPABILITIES: an explicit list of what it can and CANNOT do. The "cannots" matter as much as the "cans".
• CONTEXT: dynamic variables from the environment: user, state, available tools.
• BEHAVIOR RULES: how to act in specific situations, with explicit edge cases.
• OUTPUT FORMAT: the exact structure, length and channel of the response.
• ESCALATION: exact criteria for handing off to a human.

Controlled Chain-of-Thought: separating reasoning from the answer

CoT (Chain-of-Thought) improves reasoning quality, but in production we don't want to show the internal process to the user. The right pattern is to separate the two:

Uncontrolled CoT

• The user sees the internal reasoning
• It exposes logic that can be manipulated
• It needlessly increases output tokens
• It makes the final answer harder to parse

CoT with a separate <thinking>

• The reasoning stays in internal logs
• It allows debugging without exposing it to the user
• The final output is clean and parseable
• You can monitor the quality of the reasoning

Few-shot with negative examples

Positive examples teach the expected behavior. Negative examples are just as critical: they show the model exactly what to avoid. Without them, the model can fall into "reasonable but wrong" answers.

Prompts as code: versioning and testing

A prompt that changes without control is a silent regression. The same discipline we apply to code applies to prompts:

Prompt deployment pipeline

• A git PR with the prompt change + the rationale in the description
• Automatic evaluation in CI against the base test set
• Deploy to staging → 10% of traffic → 48h of monitoring
• If metrics are OK → promote to 100%. If they degrade → automatic rollback

Complete system prompt with controlled CoT

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: centralized version management

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]",
})

Resources

jinja2, jsonlines, Anthropic prompting guide, Learn Prompting

What is an LLM agent?

An agent is a system where the LLM doesn't just generate text: it also decides which actions to execute, observes the results of those actions, and decides what to do next. The difference from a simple chatbot is that the agent has agency: it can act on the world.

The ReAct pattern: Reason + Act

ReAct is the most widely used pattern in production. On each turn, the agent (1) reasons about the current state, (2) decides on an action, (3) observes the result, and repeats until it has enough information to answer the user.

The ReAct cycle, step by step

• Thought: "The user is asking about their order. I need to call get_order_status with their ID."
• Action:get_order_status(order_id="ORD-123")
• Observation:{"status": "in transit", "eta": "tomorrow 14:00"}
• Thought: "I have the information I need. I can answer the user."
• FINISH: "Your order is on its way and will arrive tomorrow before 2:00 p.m."

Critic loop: the agent evaluates its own output

The Critic is a second agent (or a second LLM call) that evaluates the main agent's response before sending it to the user. It answers PASS/FAIL + a reason. It doubles the cost but significantly raises quality in high-impact cases.

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

Resources

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

Hub-and-spoke: the most robust pattern for production

A central orchestrator receives every message, classifies it with a lightweight agent (cheap and fast), and delegates to the right specialized agent with the full context.

The classifier is the most critical piece of the system

• Use the cheapest model: Haiku with a 5-line prompt classifies better than Sonnet with an ambiguous prompt
• Exhaustive categories: every query has to land in some category, so include "GENERAL/OTHER"
• Typed output: the classifier never returns free text; it returns an enum with the category
• Safe fallback: if the classifier fails, the system routes to the general agent and never breaks

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)

Resources

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

The memory problem in LLMs

By default, an LLM remembers nothing between sessions. Every API call is stateless. For a support agent, that's a problem: users shouldn't have to repeat their issue in every interaction.

The 4 types of memory and when to use each

• Short-term (active window): the current turn's history in the LLM's context. Free in terms of cost, lost when the session closes.
• Long-term (vector store): the domain knowledge base. Semantic search by embedding similarity. For documentation, FAQs, policies.
• Episodic (interaction history): what the user said in previous sessions. A structured database with timestamps.
• Semantic (user entities): persistent data: plan, preferences, ticket history. Structured DB.

The RAG pipeline: how it works in production

The 4 steps of the pipeline

• Ingestion: document → chunking (512 tokens, 10% overlap) → embedding → vector store + metadata
• Retrieval: query → embed → ANN search (top-10) → metadata filter → reranking → top-3
• Augmentation: retrieved chunks → inject into the LLM's context
• Evaluation: does the answer use the chunks? Were the chunks relevant?

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

Resources

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

Human-in-the-loop is not an edge case: it's design

The most common mistake is treating escalation as something exceptional. In production, 10-30% of interactions will end up with a human. The system has to be designed for this from the start, not have it bolted on later.

5 types of escalation criteria

• Business threshold: ticket_priority="critical", account_type="enterprise"
• Low confidence: confidence_score < 0.70 on the action to take
• Out of scope: the agent can't resolve the request
• Explicit request: the user asks to talk to a person
• Emotional signal: crisis, extreme urgency, accumulated frustration

Cascading fallback: the system never dies

A production system must always respond, even when everything fails. The cascading fallback pattern defines a chain of gradual degradation:

Fallback chain

• Level 1: SupportAgent with Sonnet (normal)
• Level 2: SupportAgent with Haiku (faster and cheaper if there is latency)
• Level 3: hardcoded generic response + automatic escalation to a human
• Level 4: friendly error message with an automatically created ticket number

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"
        )

Resources

pybreaker, structlog, pydantic

The agent never touches infrastructure directly

The most important rule of the tool layer: the agent calls contracts (typed schemas), not implementations. This lets you change the underlying implementation without touching the agent, and test the agent with mocks without real infrastructure.

Anatomy of a well-designed tool

• Typed schema: a Pydantic model with validation, descriptions and constraints
• Dry-run mode: validate without executing side effects, so you can check before acting
• Per-tool timeout: each tool has its own SLA, not the workflow's global one
• Idempotency: running the same tool twice with the same params = the same result
• Audit log: every execution is recorded, successful and failed

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()]

Resources

pydantic v2, redis-py, httpx (async)

The 4 trade-offs every senior engineer must master

Latency vs Quality

Haiku responds in <500ms. Sonnet takes 1-3s. Opus can take 5-10s. The question isn't "which one is better" but "which one does the user need in this context".

Cost vs Depth

Sonnet costs 12x more than Haiku. For classification (simple), Haiku is enough. For complex reasoning with tools, Sonnet is worth every cent.

Autonomy vs Control

More autonomy = a better user experience. More control = less risk of costly mistakes. The answer depends on how reversible the action is.

Agent vs Pipeline

If the flow always follows the same steps, a deterministic DAG is faster, cheaper and more predictable. The agent adds value when the input is ambiguous.

Model Routing: the highest-impact optimization

The principle: use the cheapest model that solves the case correctly. A lightweight classifier (Haiku) decides which model each query needs. 70-80% of support queries are simple and can be resolved with Haiku.

Cost reduction strategies

• Prompt caching: the static part of the system prompt is cached. Anthropic offers a 90% discount on cached tokens. Always put the static part first.
• Context compression: summarize long history instead of sending it in full. Saves 40-60% in long multi-turn conversations.
• Batch API: a 50% discount for non-urgent tasks (evaluations, offline generation).
• The 80/20 of cost: 80% of spend comes from the 20% longest requests. Optimize the tail, not the average.

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

Resources

litellm, time.perf_counter, Anthropic prompt caching

Why debugging probabilistic systems is different

In deterministic systems, the same input produces the same output, always. In systems with LLMs, the same input can produce slightly different outputs on each call. That completely changes the debugging strategy.

The 3 most frequent types of failure

• Hallucinations: the model generates incorrect information with high confidence. Cause: insufficient context or weak constraints in the prompt. Mitigation: RAG + explicit constraints + grounding checks.
• Loops: the agent repeats the same action indefinitely. Cause: the tool fails but the model doesn't recognize it as an error. Mitigation: MAX_ITERATIONS + loop detector + circuit breaker.
• Silent degradation: quality drops gradually with no visible alert. Cause: model drift on the provider side or prompt drift from accumulated edits. Mitigation: continuous evaluation + dashboard alerts.

A 5-step debugging framework

The right process

• 1. Reproduce: store the full input (prompt, history, tool results, model, version). Without reproducibility, debugging is impossible.
• 2. Isolate: does it fail in planning, execution or evaluation? Test each component separately with synthetic inputs.
• 3. Trace: review the agent's <thinking>. Was the reasoning correct? Was the data right?
• 4. Quantify: is it an isolated case or systemic? Run it 20+ times before drawing conclusions.
• 5. Iterate: change ONE variable at a time. Without an A/B test there are no valid conclusions.

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

Resources

sqlite3, pytest fixtures, hashlib

You can't improve what you don't measure

The LLMOps module is the one that closes the loop. Without observability, the system is a black box that works (or doesn't) without anyone knowing why. With observability, every improvement decision is backed by data.

The metrics that matter, in order

• Task completion rate: the most important metric. What % of conversations ended with the user's problem solved?
• Escalation rate: % escalated to a human. If it rises → the agent is getting worse. If it drops a lot → it may be letting through cases it should escalate.
• Cost per successful interaction: (tokens × price) / successful interactions. The system's efficiency metric.
• p95 latency: the 95th percentile of latency, what 95% of users experience. The average lies.
• Tool error rate: % of tool calls that fail. Points to problems with external APIs.

LLM-as-judge: scalable automatic evaluation

Manually evaluating the quality of 1000 responses a week isn't feasible. The LLM-as-judge pattern uses a second LLM to evaluate the output of the first. The evaluator receives the original query, the generated response and the evaluation criteria.

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())

Resources

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