Construye un agente
de IA real,
módulo a módulo.

• ✓El retry se activa ante errores 429 y 500 (testeable con mocks)
• ✓Cada llamada loguea trace_id, modelo, tokens y costo
• ✓El cálculo de costo es correcto para los 3 tiers de modelo
• ✓El wrapper funciona con cualquier lista de mensajes (no acoplado a un agente)

• ✓Cada system prompt tiene las 6 secciones completas
• ✓Al menos 3 BEHAVIOR RULES con ejemplos negativos incluidos
• ✓El output JSON es parseable en 10 llamadas consecutivas sin fallos
• ✓CHANGELOG.md documenta el razonamiento de cada cambio

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

• ✓El agente resuelve correctamente 4 de 5 casos del test set
• ✓MAX_ITERATIONS dispara escalación (nunca excepción)
• ✓El Critic rechaza al menos 1 caso de prueba inválido
• ✓Cada llamada a herramienta pasa validación antes de ejecutar

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)

• ✓Classifier enruta correctamente 4+ tipos de intent
• ✓Handoff preserva contexto completo entre agentes
• ✓Timeout global retorna respuesta degradada, nunca excepción
• ✓Mensajes entre agentes usan dataclasses tipados

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

• ✓Pipeline procesa 50 docs sin errores
• ✓Retrieval retorna chunks relevantes en >80% del test set
• ✓Context compression activa cuando el historial supera 3000 tokens
• ✓El agente cita la fuente de su respuesta correctamente

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

• ✓5 criterios de escalamiento con tests
• ✓Fallback nunca lanza excepción — siempre responde
• ✓Audit log registra toda escalación con razón
• ✓Circuit breaker activa y desactiva correctamente

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

• ✓4 herramientas con schema Pydantic completo
• ✓Cada tool tiene test de timeout y error handling
• ✓Audit log captura toda ejecución (exitosa y fallida)
• ✓Estado de sesión persiste entre llamadas al agente

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

• ✓Model routing reduce costo ≥40% sin bajar task completion
• ✓Context compression funcional con test de verificación
• ✓ADR documenta 3 trade-offs con datos medidos
• ✓Benchmark incluye latencia p50/p95 antes y después

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

• ✓FailureStore guarda y recupera el contexto completo de un fallo
• ✓LoopDetector se activa correctamente en el escenario de prueba
• ✓Análisis escrito de 2 fallos reales con root cause identificado
• ✓Test set creció con casos derivados de los fallos encontrados

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

• ✓Cada request genera trace_id y spans completos
• ✓Dashboard con 5 métricas clave en tiempo real
• ✓Eval pipeline ejecuta en cada PR de prompts y falla si hay regresión
• ✓Al menos 1 guardrail activo con test de verificación