AI Engineer: construye un agente de IA real, módulo a módulo

¿Qué es un LLM y cómo genera texto?

Un Large Language Model es una red neuronal entrenada para predecir el siguiente token dado un contexto. No "entiende" como lo hacemos nosotros — aprendió patrones estadísticos en billones de tokens de texto. Cada vez que genera una palabra, está calculando una distribución de probabilidad sobre el vocabulario y muestreando de ella.

Arquitectura Transformer — lo que necesitas saber

No necesitas implementar un Transformer, pero sí entender sus implicaciones prácticas:

Conceptos clave y su impacto en producción

• Self-attention: cada token "atiende" a todos los demás del contexto. Implicación: el modelo puede relacionar información que está lejos en el texto.
• Ventana de contexto: límite máximo de tokens activos. GPT-4: 128K, Claude 3.5: 200K, Gemini 1.5 Pro: 1M. Más contexto = más costo y latencia.
• Decoder-only: los modelos modernos (GPT, Claude, Gemini) solo generan — no tienen un encoder separado. Procesan todo el contexto cada vez que generan un token.
• Tokenización (BPE): el texto se divide en sub-palabras. "tokenización" puede ser 3-4 tokens. El costo real de una llamada depende del número de tokens, no de palabras ni caracteres.

Parámetros de inferencia — el dial de control

Cuando haces una llamada al API, estos parámetros determinan cómo el modelo muestrea su respuesta:

temperature

0 = determinista (siempre el token más probable). 1 = más variado. Para agentes en producción: usa 0–0.3. Para generación creativa: 0.7–1.0.

top_p (nucleus sampling)

Solo considera los tokens cuya probabilidad acumulada llega a p. top_p=0.9 ignora el 10% de tokens menos probables. Alternativa a temperature, no se usan juntos.

max_tokens

Límite de tokens en la respuesta. Impacta costo y latencia directamente. Para respuestas estructuradas (JSON), un límite bajo previene respuestas truncadas inesperadamente.

stop_sequences

El modelo para de generar cuando encuentra este string. Útil para delimitar outputs: ["</response>", "###"]. Más confiable que max_tokens para outputs estructurados.

Selección de modelo — el trade-off más importante

Guía de selección para producción 2025

• Claude 3 Haiku / GPT-4o-mini: clasificación, routing, extracción simple. ~$0.25/M tokens. Latencia: <1s. Usa cuando el error tiene bajo impacto.
• Claude 3.5 Sonnet / GPT-4o: razonamiento, generación, tool calling complejo. ~$3/M tokens. Balance ideal para la mayoría de agentes en producción.
• Claude 3 Opus / GPT-4-turbo: análisis profundo, decisiones de alto impacto. ~$15/M tokens. Solo cuando la calidad es crítica y el costo es secundario.
• Mistral / LLaMA 3 (self-hosted): datos sensibles, cumplimiento regulatorio, costo a escala extrema. Requiere infraestructura propia.

LLMClient — wrapper base del sistema

El patrón correcto no es llamar al SDK directamente desde cada agente. Se crea un wrapper centralizado que maneja: retry automático, logging estructurado, conteo de tokens y selección 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)

El decorador @retry de tenacity maneja automáticamente los rate limits (429) y errores temporales del API con backoff exponencial. Sin esto, cualquier fallo transitorio rompe el flujo del agente.

Recursos

anthropic-sdk docs, tenacity, structlog, tiktoken

El system prompt es la constitución del agente

El system prompt no es una "instrucción inicial" — es el documento que define completamente quién es el agente, qué puede hacer, cómo debe comportarse y cuándo debe pedir ayuda. Un agente sin system prompt bien estructurado es un agente impredecible.

Estructura en 6 secciones (todas obligatorias)

• IDENTITY: nombre, propósito, personalidad. Define el "quién" del agente.
• CAPABILITIES: lista explícita de qué puede y NO puede hacer. Los "no puede" son tan importantes como los "puede".
• CONTEXT: variables dinámicas del entorno: usuario, estado, herramientas disponibles.
• BEHAVIOR RULES: cómo actuar ante situaciones específicas — edge cases explícitos.
• OUTPUT FORMAT: estructura exacta, longitud y canal de la respuesta.
• ESCALATION: criterios exactos para transferir a humano.

Chain-of-Thought controlado — separar razonamiento de respuesta

El CoT (Chain-of-Thought) mejora la calidad del razonamiento, pero en producción no queremos mostrar el proceso interno al usuario. El patrón correcto es separar los dos:

CoT sin control

• El usuario ve el razonamiento interno
• Expone lógica que puede ser manipulada
• Aumenta tokens de output innecesariamente
• Dificulta el parsing de la respuesta final

CoT con <thinking> separado

• El razonamiento queda en logs internos
• Permite debugging sin exposición al usuario
• El output final es limpio y parseable
• Puedes monitorear la calidad del razonamiento

Few-shot con ejemplos negativos

Los ejemplos positivos enseñan el comportamiento esperado. Los ejemplos negativos son igual de críticos — le muestran al modelo exactamente qué evitar. Sin ellos, el modelo puede caer en respuestas "razonables pero incorrectas".

Prompts como código — versionado y testing

Un prompt que cambia sin control es una regresión silenciosa. La misma disciplina que aplicamos al código aplica a los prompts:

Pipeline de deployment de prompts

• PR en git con el cambio de prompt + justificación en descripción
• Evaluación automática en CI contra el test set base
• Deploy a staging → 10% de tráfico → 48h de monitoreo
• Si métricas OK → promover a 100%. Si degradan → rollback automático

System prompt completo con 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 — gestión centralizada de versiones

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

¿Qué es un agente LLM?

Un agente es un sistema donde el LLM no solo genera texto — también toma decisiones sobre qué acciones ejecutar, observa los resultados de esas acciones, y decide qué hacer a continuación. La diferencia con un simple chatbot es que el agente tiene agencia: puede actuar sobre el mundo.

Patrón ReAct — Reason + Act

ReAct es el patrón más usado en producción. En cada turno, el agente: (1) razona sobre el estado actual, (2) decide una acción, (3) observa el resultado, y repite hasta tener suficiente información para responder al usuario.

Ciclo ReAct paso a paso

• Thought: "El usuario pregunta por su pedido. Necesito llamar a get_order_status con su ID."
• Action:get_order_status(order_id="ORD-123")
• Observation:{"status": "en camino", "eta": "mañana 14:00"}
• Thought: "Tengo la información necesaria. Puedo responder al usuario."
• FINISH: "Tu pedido está en camino y llegará mañana antes de las 14:00."

Critic loop — el agente evalúa su propio output

El Critic es un segundo agente (o una segunda llamada al LLM) que evalúa la respuesta del agente principal antes de enviarla al usuario. Responde PASS/FAIL + razón. Duplica el costo pero aumenta significativamente la calidad en 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: el patrón más robusto para producción

Un orquestador central recibe todos los mensajes, los clasifica con un agente ligero (barato y rápido), y delega al agente especializado correcto con el contexto completo.

El clasificador es la pieza más crítica del sistema

• Usa el modelo más barato: Haiku con un prompt de 5 líneas clasifica mejor que Sonnet con un prompt ambiguo
• Categorías exhaustivas: toda consulta debe caer en alguna categoría — incluye "GENERAL/OTHER"
• Output tipado: el classifier nunca retorna texto libre — retorna un enum con la categoría
• Fallback seguro: si el classifier falla, el sistema enruta al agente general — nunca rompe

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

El problema de la memoria en LLMs

Por defecto, un LLM no recuerda nada entre sesiones. Cada llamada al API es stateless. Para un agente de soporte, esto es un problema: el usuario no debería tener que repetir su problema en cada interacción.

Los 4 tipos de memoria y cuándo usar cada uno

• Short-term (ventana activa): historial del turno actual en el contexto del LLM. Gratis en costo, se pierde al cerrar sesión.
• Long-term (vector store): knowledge base del dominio. Búsqueda semántica por similitud de embeddings. Para documentación, FAQs, políticas.
• Episodic (historial de interacciones): qué dijo el usuario en sesiones anteriores. Base de datos estructurada con timestamp.
• Semantic (entidades del usuario): datos persistentes: plan, preferencias, historial de tickets. Structured DB.

Pipeline RAG — cómo funciona en producción

Los 4 pasos del pipeline

• Ingestion: documento → chunking (512 tokens, 10% overlap) → embedding → vector store + metadata
• Retrieval: query → embed → ANN search (top-10) → filtro por metadata → reranking → top-3
• Augmentation: chunks recuperados → inyectar en el contexto del LLM
• Evaluation: ¿la respuesta usa los chunks? ¿los chunks eran 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 no es un edge case — es diseño

El error más común es tratar la escalación como algo excepcional. En producción, entre el 10-30% de las interacciones terminarán en un humano. El sistema debe estar diseñado para esto desde el principio, no añadirlo después.

5 tipos de criterios de escalamiento

• Umbral de negocio: ticket_priority="critical", account_type="enterprise"
• Confianza baja: confidence_score < 0.70 en la acción a tomar
• Fuera de scope: el agente no puede resolver la solicitud
• Solicitud explícita: el usuario pide hablar con una persona
• Señal emocional: crisis, urgencia extrema, frustración acumulada

Fallback en cascada — el sistema nunca muere

Un sistema de producción debe responder siempre, incluso cuando todo falla. El patrón de fallback en cascada define una cadena de degradación gradual:

Cadena de fallback

• Nivel 1: SupportAgent con Sonnet (normal)
• Nivel 2: SupportAgent con Haiku (más rápido y barato si hay latencia)
• Nivel 3: Respuesta genérica hardcodeada + escalación automática a humano
• Nivel 4: Mensaje de error amigable con número de ticket creado automáticamente

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

El agente nunca toca la infraestructura directamente

La regla más importante del tool layer: el agente llama a contratos (schemas tipados), no a implementaciones. Esto permite cambiar la implementación subyacente sin tocar el agente, y testear el agente con mocks sin infraestructura real.

Anatomía de una herramienta bien diseñada

• Schema tipado: Pydantic model con validación, descriptions y constraints
• Dry-run mode: validar sin ejecutar efectos — permite verificar antes de actuar
• Timeout por herramienta: cada tool tiene su propio SLA — no el global del workflow
• Idempotencia: ejecutar la misma herramienta 2 veces con los mismos params = mismo resultado
• Audit log: toda ejecución queda registrada — exitosa y fallida

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)

Los 4 trade-offs que todo senior debe dominar

Latencia vs Calidad

Haiku responde en <500ms. Sonnet tarda 1-3s. Opus puede tardar 5-10s. La pregunta no es "cuál es mejor" sino "cuál necesita el usuario en este contexto".

Costo vs Profundidad

Sonnet cuesta 12x más que Haiku. Para clasificación (simple), Haiku es suficiente. Para razonamiento complejo con herramientas, Sonnet vale cada centavo.

Autonomía vs Control

Más autonomía = mejor experiencia de usuario. Más control = menos riesgo de errores costosos. La respuesta depende de la reversibilidad de la acción.

Agente vs Pipeline

Si el flujo siempre sigue los mismos pasos, un DAG determinista es más rápido, barato y predecible. El agente añade valor cuando el input es ambiguo.

Model Routing — la optimización de mayor impacto

El principio: usar el modelo más barato que resuelve correctamente el caso. Un clasificador ligero (Haiku) decide qué modelo necesita cada consulta. El 70-80% de las consultas de soporte son simples y pueden resolverse con Haiku.

Estrategias de reducción de costo

• Prompt caching: la parte estática del system prompt se cachea. Anthropic ofrece 90% de descuento en tokens cacheados. Pon la parte estática siempre primero.
• Context compression: resumir el historial largo en vez de enviarlo completo. Ahorro de 40-60% en conversaciones de muchos turnos.
• Batch API: 50% de descuento para tareas no urgentes (evaluaciones, generación offline).
• El 80/20 del costo: el 80% del gasto viene del 20% de las solicitudes más largas. Optimiza la cola, no el promedio.

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 qué el debugging en sistemas probabilísticos es diferente

En sistemas deterministas, el mismo input produce el mismo output — siempre. En sistemas con LLMs, el mismo input puede producir outputs ligeramente diferentes en cada llamada. Esto cambia completamente la estrategia de debugging.

Los 3 tipos de fallo más frecuentes

• Alucinaciones: el modelo genera información incorrecta con confianza alta. Causa: contexto insuficiente o constraints débiles en el prompt. Mitigación: RAG + constraints explícitos + grounding checks.
• Loops: el agente repite la misma acción indefinidamente. Causa: la herramienta falla pero el modelo no lo reconoce como error. Mitigación: MAX_ITERATIONS + loop detector + circuit breaker.
• Degradación silenciosa: la calidad cae gradualmente sin alerta visible. Causa: model drift del proveedor o prompt drift por ediciones acumuladas. Mitigación: evaluación continua + alertas en dashboard.

Framework de debugging en 5 pasos

El proceso correcto

• 1. Reproducir: guardar el input completo (prompt, historial, tool results, modelo, versión). Sin reproducibilidad, el debugging es imposible.
• 2. Aislar: ¿falla en planning, execution o evaluation? Testear cada componente por separado con inputs sintéticos.
• 3. Trazar: revisar el <thinking> del agente. ¿El razonamiento fue correcto? ¿Los datos estaban bien?
• 4. Cuantificar: ¿es un caso aislado o sistémico? Ejecuta 20+ veces antes de concluir.
• 5. Iterar: cambiar UNA variable a la vez. Sin A/B test no hay conclusiones 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

No puedes mejorar lo que no mides

El módulo de LLMOps es el que cierra el ciclo. Sin observabilidad, el sistema es una caja negra que funciona (o no) sin que nadie sepa por qué. Con observabilidad, cada decisión de mejora está respaldada por datos.

Las métricas que importan — y en qué orden

• Task completion rate: la métrica más importante. ¿Qué % de conversaciones terminaron con el problema del usuario resuelto?
• Escalation rate: % escalado a humano. Si sube → el agente está empeorando. Si baja mucho → puede estar dejando pasar casos que debería escalar.
• Cost per successful interaction: (tokens × precio) / interacciones exitosas. La métrica de eficiencia del sistema.
• Latencia p95: el percentil 95 de latencia — lo que el 95% de los usuarios experimenta. El promedio miente.
• Tool error rate: % de llamadas a herramientas que fallan. Señala problemas de APIs externas.

LLM-as-judge — evaluación automática escalable

Evaluar manualmente la calidad de 1000 respuestas por semana es inviable. El patrón LLM-as-judge usa un segundo LLM para evaluar el output del primero. El evaluador recibe: la consulta original, la respuesta generada y los criterios de evaluación.

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)