Módulo 12: LangSmith y Producción

Production Checklist y Deployment

Descripción de la cápsula

Este es el pre-flight checklist antes de deployar tu agente de AI a producción. No es una checklist genérica de software — es específica para sistemas powered by LLMs, con sus desafíos únicos: los modelos pueden generar respuestas incorrectas, los costos escalan con el uso, las APIs de proveedores pueden caer sin aviso, y las respuestas pueden contener información sensible o dañina.

En las cápsulas anteriores de este módulo, aprendiste cada pieza individual: tracing con LangSmith (02), debugging visual (03), evaluation automatizada (04), token tracking (05), y rate limiting (06). Ahora vas a integrar todo en un checklist que te diga "sí, tu sistema está listo para producción" o "no, te falta X antes de deployar."

La diferencia entre un agente que "funciona" y un agente "production-ready" es confianza. Con tracing, sabes qué hizo. Con evaluation, sabes si lo hizo bien. Con cost control, sabes cuánto costó. Con esta checklist, sabes que cubriste todos los ángulos. Eso es lo que te permite deployar y dormir tranquilo.


La checklist: 6 categorías

La production checklist tiene 6 categorías. Cada categoría es un área que debes cubrir antes de deployar. No son opcionales — cada una existe porque un incidente real la motivó.

Production Readiness Score:

 ✅ 1. Configuration Management  [  /4 ]
 ✅ 2. Error Monitoring           [  /4 ]
 ✅ 3. Safety & Compliance        [  /4 ]
 ✅ 4. Cost Control               [  /4 ]
 ✅ 5. Observability              [  /4 ]
 ✅ 6. Resilience                 [  /4 ]

                          Total: [  /24 ]

1. Configuration Management

Las APIs de LLMs son servicios externos. Las credenciales, versiones de modelo, y configuración de prompts deben gestionarse con cuidado.

1.1 API keys en variables de entorno

Nunca en código. Nunca en commits. Nunca en logs.

from dotenv import load_dotenv
load_dotenv()

import os

REQUIRED_ENV_VARS = [
    "OPENAI_API_KEY",
    "LANGSMITH_API_KEY",
    "LANGSMITH_TRACING",
]

def verify_env_vars():
    """Verifica que todas las variables de entorno requeridas estén configuradas."""
    missing = []
    masked = []

    for var in REQUIRED_ENV_VARS:
        value = os.getenv(var)
        if not value:
            missing.append(var)
        else:
            if "KEY" in var or "SECRET" in var:
                masked.append(f"  {var}: {'*' * 8}...{value[-4:]}")
            else:
                masked.append(f"  {var}: {value}")

    if missing:
        print(f"MISSING environment variables:")
        for var in missing:
            print(f"  ❌ {var}")
        return False

    print("Environment variables OK:")
    for line in masked:
        print(f"  ✅ {line}")
    return True

result = verify_env_vars()
print(f"\nConfiguration check: {'PASS' if result else 'FAIL'}")
# Output esperado:
# Environment variables OK:
#   ✅   OPENAI_API_KEY: ********...a1b2
#   ✅   LANGSMITH_API_KEY: ********...c3d4
#   ✅   LANGSMITH_TRACING: true
#
# Configuration check: PASS

1.2 Versiones de modelo pinned

Nunca uses "latest". Si OpenAI actualiza GPT-4.1 y cambia el comportamiento, tu agente se rompe sin que hayas cambiado nada.

from dotenv import load_dotenv
load_dotenv()

MODEL_CONFIG = {
    "researcher": {
        "model": "openai:gpt-4.1-mini",
        "temperature": 0.2,
        "max_tokens": 1000,
    },
    "analyst": {
        "model": "openai:gpt-4.1",
        "temperature": 0.1,
        "max_tokens": 2000,
    },
    "writer": {
        "model": "openai:gpt-4.1-mini",
        "temperature": 0.3,
        "max_tokens": 1500,
    },
}

def verify_model_config():
    """Verifica que los modelos están configurados correctamente."""
    issues = []
    for agent_name, config in MODEL_CONFIG.items():
        model_id = config["model"]

        if "latest" in model_id.lower():
            issues.append(f"  ❌ {agent_name}: uses 'latest' — pin to specific version")

        if config["temperature"] > 0.5:
            issues.append(f"  ⚠️ {agent_name}: high temperature ({config['temperature']}) — may produce inconsistent output")

        if config.get("max_tokens", 0) > 4000:
            issues.append(f"  ⚠️ {agent_name}: high max_tokens ({config['max_tokens']}) — increases cost")

        if not issues or not any(agent_name in i for i in issues):
            print(f"  ✅ {agent_name}: {model_id} (temp={config['temperature']})")

    if issues:
        print("\nIssues found:")
        for issue in issues:
            print(issue)
    else:
        print("\nAll models properly configured.")

print("Model Configuration:")
verify_model_config()
# Output esperado:
# Model Configuration:
#   ✅ researcher: openai:gpt-4.1-mini (temp=0.2)
#   ✅ analyst: openai:gpt-4.1 (temp=0.1)
#   ✅ writer: openai:gpt-4.1-mini (temp=0.3)
#
# All models properly configured.

1.3 Prompt versions tracked

Los prompts son código. Deben versionarse como código. Un cambio de prompt puede cambiar completamente el comportamiento del agente.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from datetime import datetime

@dataclass
class PromptVersion:
    name: str
    version: str
    content: str
    created_at: str
    description: str

PROMPT_REGISTRY = {
    "researcher_system": PromptVersion(
        name="researcher_system",
        version="1.2.0",
        content="Eres un investigador especializado. Tu única tarea es buscar información relevante. NO analices ni escribas reportes.",
        created_at="2026-03-01",
        description="Added explicit constraint to NOT analyze",
    ),
    "analyst_system": PromptVersion(
        name="analyst_system",
        version="2.0.1",
        content="Eres un analista de investigación. Identificas patrones, contradicciones y tendencias. Responde con hallazgos numerados.",
        created_at="2026-03-05",
        description="Changed output format to numbered findings",
    ),
    "writer_system": PromptVersion(
        name="writer_system",
        version="1.1.0",
        content="Eres un escritor técnico. Redactas reportes claros y concisos. Usa bullet points y resúmenes ejecutivos.",
        created_at="2026-02-28",
        description="Added executive summary requirement",
    ),
}

def verify_prompts():
    """Verifica que todos los prompts están versionados."""
    print("Prompt Registry:")
    for name, prompt in PROMPT_REGISTRY.items():
        print(f"  ✅ {name} v{prompt.version} ({prompt.created_at})")
        print(f"     {prompt.description}")

verify_prompts()
# Output esperado:
# Prompt Registry:
#   ✅ researcher_system v1.2.0 (2026-03-01)
#      Added explicit constraint to NOT analyze
#   ✅ analyst_system v2.0.1 (2026-03-05)
#      Changed output format to numbered findings
#   ✅ writer_system v1.1.0 (2026-02-28)
#      Added executive summary requirement

1.4 Secrets no expuestos en logs o traces

LangSmith captura los prompts y respuestas en traces. Si tu prompt incluye datos sensibles, esos datos quedan en LangSmith.

from dotenv import load_dotenv
load_dotenv()

import os
import re

def check_for_secrets_in_text(text: str) -> list[str]:
    """Detecta posibles secretos en texto que podría ir a logs o traces."""
    patterns = [
        (r'sk-[a-zA-Z0-9]{20,}', "OpenAI API key"),
        (r'lsv2_[a-zA-Z0-9]{20,}', "LangSmith API key"),
        (r'password\s*[=:]\s*["\'][^"\']+["\']', "Hardcoded password"),
        (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', "Email address"),
        (r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', "Phone number"),
    ]

    found = []
    for pattern, description in patterns:
        if re.search(pattern, text):
            found.append(description)
    return found

test_prompts = [
    "Analiza el impacto de AI en educación.",
    "Mi API key es sk-abc123def456ghi789jklmnopqrst. Úsala para buscar.",
    "Contacta a usuario@ejemplo.com para más información.",
]

print("Secret Detection Check:")
for prompt in test_prompts:
    secrets = check_for_secrets_in_text(prompt)
    if secrets:
        print(f"  ❌ '{prompt[:50]}...'")
        for s in secrets:
            print(f"     Found: {s}")
    else:
        print(f"  ✅ '{prompt[:50]}...' — clean")
# Output esperado:
# Secret Detection Check:
#   ✅ 'Analiza el impacto de AI en educación....' — clean
#   ❌ 'Mi API key es sk-abc123def456ghi789jklmnopqrst...'
#      Found: OpenAI API key
#   ❌ 'Contacta a usuario@ejemplo.com para más informa...'
#      Found: Email address

2. Error Monitoring

Los LLMs fallan de formas que las APIs tradicionales no fallan: timeouts largos, respuestas parciales, JSON inválido, alucinaciones, y rate limits del proveedor.

2.1 Fallback providers

Si OpenAI cae, ¿tu agente deja de funcionar? Con fallback, switches automáticamente a Anthropic u otro proveedor.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

def create_model_with_fallback(
    primary: str = "openai:gpt-4.1-mini",
    fallback: str = "openai:gpt-4.1-nano",
) -> object:
    """Crea un modelo con fallback automático."""
    primary_model = init_chat_model(primary)
    fallback_model = init_chat_model(fallback)
    return primary_model.with_fallbacks([fallback_model])

model = create_model_with_fallback()

response = model.invoke("¿Qué es producción en AI? 1 oración.")
print(f"Respuesta: {response.content}")
# Output esperado:
# Respuesta: Producción en AI es el proceso de deployar modelos de inteligencia artificial en sistemas reales...

2.2 Timeout por modelo

Los LLMs pueden tardar mucho. Un timeout de 30 segundos es razonable para la mayoría de llamadas. Sin timeout, un request colgado bloquea tu sistema.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model_with_timeout = init_chat_model(
    "openai:gpt-4.1-mini",
    timeout=30,
    max_retries=2,
)

response = model_with_timeout.invoke("¿Qué es un timeout? 1 oración.")
print(f"Respuesta: {response.content}")
print(f"Configuración: timeout=30s, max_retries=2")
# Output esperado:
# Respuesta: Un timeout es un límite de tiempo configurado para una operación que, al excederse, cancela la operación.
# Configuración: timeout=30s, max_retries=2

2.3 Retry con backoff

Los errores transitorios (rate limits, timeouts de red) se resuelven con retry. El backoff exponencial evita saturar al proveedor.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "openai:gpt-4.1-mini",
    max_retries=3,
)

response = model.invoke("¿Qué es retry con backoff? 1 oración.")
print(f"Respuesta: {response.content}")
print("Configurado: 3 reintentos automáticos con backoff exponencial")
# Output esperado:
# Respuesta: Retry con backoff es una estrategia que reintenta operaciones fallidas con intervalos crecientes entre intentos.
# Configurado: 3 reintentos automáticos con backoff exponencial

2.4 Logging estructurado de errores

Cuando algo falla, necesitas saber qué, cuándo, y en qué contexto. Logging estructurado te da esa información.

from dotenv import load_dotenv
load_dotenv()

import logging
import json
from datetime import datetime

logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger("ai_agent")

def log_agent_error(
    agent_name: str,
    operation: str,
    error: Exception,
    context: dict = None,
):
    """Registra un error de agente de forma estructurada."""
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "level": "ERROR",
        "agent": agent_name,
        "operation": operation,
        "error_type": type(error).__name__,
        "error_message": str(error),
        "context": context or {},
    }
    logger.error(json.dumps(log_entry, ensure_ascii=False))

try:
    result = 1 / 0
except Exception as e:
    log_agent_error(
        agent_name="analyst",
        operation="synthesize_findings",
        error=e,
        context={"topic": "AI en educación", "num_sources": 8},
    )
# Output esperado:
# {"timestamp": "2026-03-08T...", "level": "ERROR", "agent": "analyst", "operation": "synthesize_findings", "error_type": "ZeroDivisionError", "error_message": "division by zero", "context": {"topic": "AI en educación", "num_sources": 8}}

3. Safety & Compliance

Los LLMs pueden generar contenido dañino, exponer PII, o tomar decisiones que necesitan auditoría. En producción, necesitas guardrails.

3.1 PII detection en outputs

Tu agente puede recibir o generar información personal identificable (PII). Debes detectarla y filtrarla antes de almacenarla o mostrarla.

from dotenv import load_dotenv
load_dotenv()

import re

PII_PATTERNS = {
    "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
    "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
    "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
    "credit_card": r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
}

def detect_pii(text: str) -> list[dict]:
    """Detecta PII en texto."""
    findings = []
    for pii_type, pattern in PII_PATTERNS.items():
        matches = re.findall(pattern, text)
        for match in matches:
            findings.append({"type": pii_type, "value": match})
    return findings

def redact_pii(text: str) -> str:
    """Redacta PII del texto."""
    redacted = text
    for pii_type, pattern in PII_PATTERNS.items():
        redacted = re.sub(pattern, f"[REDACTED-{pii_type.upper()}]", redacted)
    return redacted

test_output = """
El usuario Juan Pérez (correo: juan.perez@email.com, teléfono: 555-123-4567)
reportó que su tarjeta 4532-1234-5678-9012 fue comprometida.
"""

pii_found = detect_pii(test_output)
print(f"PII detectado: {len(pii_found)} items")
for item in pii_found:
    print(f"  ❌ {item['type']}: {item['value']}")

redacted = redact_pii(test_output)
print(f"\nTexto redactado:")
print(redacted)
# Output esperado:
# PII detectado: 3 items
#   ❌ email: juan.perez@email.com
#   ❌ phone: 555-123-4567
#   ❌ credit_card: 4532-1234-5678-9012
#
# Texto redactado:
#
# El usuario Juan Pérez (correo: [REDACTED-EMAIL], teléfono: [REDACTED-PHONE])
# reportó que su tarjeta [REDACTED-CREDIT_CARD] fue comprometida.

3.2 Content filtering

Detecta contenido potencialmente dañino en las respuestas del agente antes de mostrarlo al usuario.

from dotenv import load_dotenv
load_dotenv()

BLOCKED_TOPICS = [
    "cómo hackear", "crear malware", "fabricar explosivos",
    "how to hack", "create malware", "make a bomb",
]

def content_filter(text: str) -> tuple[bool, str]:
    """Filtra contenido potencialmente dañino."""
    text_lower = text.lower()
    for topic in BLOCKED_TOPICS:
        if topic in text_lower:
            return False, f"Blocked topic detected: '{topic}'"
    return True, "OK"

test_responses = [
    "Python es un lenguaje de programación versátil.",
    "Para hackear un sistema, primero necesitas...",
    "La AI tiene muchas aplicaciones beneficiosas.",
]

print("Content Filter Check:")
for response in test_responses:
    is_safe, reason = content_filter(response)
    status = "✅ PASS" if is_safe else "❌ BLOCKED"
    print(f"  {status}: '{response[:50]}...' — {reason}")
# Output esperado:
# Content Filter Check:
#   ✅ PASS: 'Python es un lenguaje de programación versátil....' — OK
#   ❌ BLOCKED: 'Para hackear un sistema, primero necesitas......' — Blocked topic detected: 'hackear'
#   ✅ PASS: 'La AI tiene muchas aplicaciones beneficiosas....' — OK

3.3 Audit logging

Cada decisión del agente debe ser auditable: quién pidió qué, qué decidió el agente, qué herramientas usó, y qué respondió.

from dotenv import load_dotenv
load_dotenv()

import json
from datetime import datetime
from dataclasses import dataclass, asdict

@dataclass
class AuditEntry:
    timestamp: str
    user_id: str
    action: str
    agent: str
    input_summary: str
    output_summary: str
    tools_used: list[str]
    cost_usd: float
    model: str

class AuditLogger:
    def __init__(self):
        self.entries: list[AuditEntry] = []

    def log(self, **kwargs):
        entry = AuditEntry(
            timestamp=datetime.now().isoformat(),
            **kwargs,
        )
        self.entries.append(entry)
        return entry

    def query(self, user_id: str = None, agent: str = None) -> list[AuditEntry]:
        results = self.entries
        if user_id:
            results = [e for e in results if e.user_id == user_id]
        if agent:
            results = [e for e in results if e.agent == agent]
        return results


audit = AuditLogger()

audit.log(
    user_id="user-001",
    action="research",
    agent="supervisor",
    input_summary="Investiga AI en educación",
    output_summary="Reporte de 5 hallazgos generado",
    tools_used=["web_search", "arxiv_search", "format_report"],
    cost_usd=0.035,
    model="gpt-4.1-mini",
)

audit.log(
    user_id="user-001",
    action="feedback",
    agent="supervisor",
    input_summary="Agrega sección sobre costos",
    output_summary="Reporte actualizado con sección de costos",
    tools_used=["format_report"],
    cost_usd=0.012,
    model="gpt-4.1-mini",
)

entries = audit.query(user_id="user-001")
print(f"Audit log para user-001: {len(entries)} entries")
for entry in entries:
    print(f"  [{entry.timestamp[:19]}] {entry.action} via {entry.agent}")
    print(f"    Input:  {entry.input_summary}")
    print(f"    Output: {entry.output_summary}")
    print(f"    Tools:  {', '.join(entry.tools_used)}")
    print(f"    Cost:   ${entry.cost_usd:.3f}")
# Output esperado:
# Audit log para user-001: 2 entries
#   [2026-03-08T...] research via supervisor
#     Input:  Investiga AI en educación
#     Output: Reporte de 5 hallazgos generado
#     Tools:  web_search, arxiv_search, format_report
#     Cost:   $0.035
#   [2026-03-08T...] feedback via supervisor
#     Input:  Agrega sección sobre costos
#     Output: Reporte actualizado con sección de costos
#     Tools:  format_report
#     Cost:   $0.012

4. Cost Control

Ya lo cubriste en detalle en las cápsulas 05 y 06. La checklist resume los items que debes tener activos.

from dotenv import load_dotenv
load_dotenv()

COST_CHECKLIST = {
    "token_tracking_enabled": True,
    "cost_breakdown_by_operation": True,
    "per_user_budgets_configured": True,
    "rate_limiting_active": True,
    "cost_alerts_configured": True,
    "auto_degradation_enabled": True,
    "monthly_projection_reviewed": True,
}

def verify_cost_control():
    """Verifica que todos los controles de costo están activos."""
    print("Cost Control Checklist:")
    all_ok = True
    for item, status in COST_CHECKLIST.items():
        icon = "✅" if status else "❌"
        print(f"  {icon} {item.replace('_', ' ').title()}")
        if not status:
            all_ok = False
    return all_ok

result = verify_cost_control()
print(f"\nCost control: {'PASS' if result else 'FAIL'}")
# Output esperado:
# Cost Control Checklist:
#   ✅ Token Tracking Enabled
#   ✅ Cost Breakdown By Operation
#   ✅ Per User Budgets Configured
#   ✅ Rate Limiting Active
#   ✅ Cost Alerts Configured
#   ✅ Auto Degradation Enabled
#   ✅ Monthly Projection Reviewed
#
# Cost control: PASS

5. Observability

Tracing y evaluation deben estar activos antes de deployar. Sin ellos, estás operando a ciegas.

5.1 LangSmith tracing habilitado

from dotenv import load_dotenv
load_dotenv()

import os

def verify_langsmith_config():
    """Verifica que LangSmith está configurado para producción."""
    checks = {
        "LANGSMITH_TRACING": os.getenv("LANGSMITH_TRACING") == "true",
        "LANGSMITH_API_KEY": bool(os.getenv("LANGSMITH_API_KEY")),
        "LANGSMITH_PROJECT": bool(os.getenv("LANGSMITH_PROJECT", "default")),
    }

    print("LangSmith Configuration:")
    all_ok = True
    for check, passed in checks.items():
        icon = "✅" if passed else "❌"
        print(f"  {icon} {check}")
        if not passed:
            all_ok = False

    if all_ok:
        project = os.getenv("LANGSMITH_PROJECT", "default")
        print(f"\n  Tracing to project: '{project}'")
        print(f"  Dashboard: https://smith.langchain.com/")

    return all_ok

verify_langsmith_config()
# Output esperado:
# LangSmith Configuration:
#   ✅ LANGSMITH_TRACING
#   ✅ LANGSMITH_API_KEY
#   ✅ LANGSMITH_PROJECT
#
#   Tracing to project: 'research-assistant-prod'
#   Dashboard: https://smith.langchain.com/

5.2 Evaluation dataset listo

from dotenv import load_dotenv
load_dotenv()

EVAL_DATASET = [
    {
        "input": "Investiga el impacto de AI en educación",
        "criteria": ["relevancia", "completitud", "accuracy"],
        "min_score": 0.7,
    },
    {
        "input": "Analiza tendencias en AI generativa 2026",
        "criteria": ["relevancia", "completitud", "actualidad"],
        "min_score": 0.7,
    },
    {
        "input": "Compara LangChain vs CrewAI vs AutoGen",
        "criteria": ["relevancia", "accuracy", "balance"],
        "min_score": 0.7,
    },
]

def verify_eval_dataset():
    """Verifica que el dataset de evaluación está preparado."""
    print(f"Evaluation Dataset: {len(EVAL_DATASET)} test cases")
    for i, case in enumerate(EVAL_DATASET, 1):
        print(f"  ✅ Case {i}: '{case['input'][:50]}...'")
        print(f"     Criteria: {', '.join(case['criteria'])}")
        print(f"     Min score: {case['min_score']}")
    return len(EVAL_DATASET) >= 3

result = verify_eval_dataset()
print(f"\nEvaluation dataset: {'PASS' if result else 'FAIL (need >= 3 cases)'}")
# Output esperado:
# Evaluation Dataset: 3 test cases
#   ✅ Case 1: 'Investiga el impacto de AI en educación...'
#      Criteria: relevancia, completitud, accuracy
#      Min score: 0.7
#   ✅ Case 2: 'Analiza tendencias en AI generativa 2026...'
#      Criteria: relevancia, completitud, actualidad
#      Min score: 0.7
#   ✅ Case 3: 'Compara LangChain vs CrewAI vs AutoGen...'
#      Criteria: relevancia, accuracy, balance
#      Min score: 0.7
#
# Evaluation dataset: PASS (need >= 3 cases)

6. Resilience

Los sistemas de AI dependen de servicios externos (APIs de proveedores, bases de datos, servicios de búsqueda). Cada uno puede fallar. Tu sistema debe sobrevivir esas fallas.

6.1 Graceful degradation

Cuando un servicio falla, tu sistema no debe crashear — debe ofrecer una experiencia reducida pero funcional.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

def create_resilient_model():
    """Crea un modelo con múltiples niveles de fallback."""
    primary = init_chat_model("openai:gpt-4.1-mini")
    fallback_1 = init_chat_model("openai:gpt-4.1-nano")

    return primary.with_fallbacks([fallback_1])

model = create_resilient_model()
response = model.invoke("¿Qué es graceful degradation? 1 oración.")
print(f"Respuesta: {response.content}")
# Output esperado:
# Respuesta: Graceful degradation es la capacidad de un sistema de seguir funcionando con capacidades reducidas cuando un componente falla.

6.2 Checkpoint persistence para tareas largas

Si el Research Assistant falla a mitad de una investigación, los checkpoints permiten resumir desde donde falló.

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

print("Checkpoint Configuration:")
print(f"  ✅ Checkpointer: {type(checkpointer).__name__}")
print(f"  ✅ Purpose: Resume failed executions from last checkpoint")
print(f"  ⚠️ Note: For production, use PostgresSaver or similar persistent backend")
# Output esperado:
# Checkpoint Configuration:
#   ✅ Checkpointer: MemorySaver
#   ✅ Purpose: Resume failed executions from last checkpoint
#   ⚠️ Note: For production, use PostgresSaver or similar persistent backend

Production Readiness Check: todo junto

Combina todas las verificaciones en un solo script que ejecutas antes de cada deployment.

from dotenv import load_dotenv
load_dotenv()

import os

def run_production_checklist() -> dict:
    """Ejecuta el production readiness check completo."""
    results = {}

    print("╔══════════════════════════════════════════════════╗")
    print("║       PRODUCTION READINESS CHECK                ║")
    print("╚══════════════════════════════════════════════════╝\n")

    # 1. Configuration Management
    print("1. CONFIGURATION MANAGEMENT")
    config_checks = {
        "API keys in env vars": bool(os.getenv("OPENAI_API_KEY")),
        "Model versions pinned": True,
        "Prompt versions tracked": True,
        "Secrets not in code": True,
    }
    config_score = sum(config_checks.values())
    for check, passed in config_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["configuration"] = config_score
    print(f"   Score: {config_score}/4\n")

    # 2. Error Monitoring
    print("2. ERROR MONITORING")
    error_checks = {
        "Fallback providers configured": True,
        "Timeout per model call": True,
        "Retry with backoff": True,
        "Structured error logging": True,
    }
    error_score = sum(error_checks.values())
    for check, passed in error_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["error_monitoring"] = error_score
    print(f"   Score: {error_score}/4\n")

    # 3. Safety & Compliance
    print("3. SAFETY & COMPLIANCE")
    safety_checks = {
        "PII detection in outputs": True,
        "Content filtering active": True,
        "Audit logging enabled": True,
        "Input validation": True,
    }
    safety_score = sum(safety_checks.values())
    for check, passed in safety_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["safety"] = safety_score
    print(f"   Score: {safety_score}/4\n")

    # 4. Cost Control
    print("4. COST CONTROL")
    cost_checks = {
        "Token tracking enabled": True,
        "Per-user budgets configured": True,
        "Rate limiting active": True,
        "Cost alerts configured": True,
    }
    cost_score = sum(cost_checks.values())
    for check, passed in cost_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["cost_control"] = cost_score
    print(f"   Score: {cost_score}/4\n")

    # 5. Observability
    print("5. OBSERVABILITY")
    obs_checks = {
        "LangSmith tracing enabled": os.getenv("LANGSMITH_TRACING") == "true",
        "Evaluation dataset ready": True,
        "Key metrics dashboarded": True,
        "Alerting on anomalies": True,
    }
    obs_score = sum(obs_checks.values())
    for check, passed in obs_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["observability"] = obs_score
    print(f"   Score: {obs_score}/4\n")

    # 6. Resilience
    print("6. RESILIENCE")
    res_checks = {
        "Graceful degradation": True,
        "Checkpoint persistence": True,
        "Retry on transient errors": True,
        "Health check endpoint": True,
    }
    res_score = sum(res_checks.values())
    for check, passed in res_checks.items():
        icon = "✅" if passed else "❌"
        print(f"   {icon} {check}")
    results["resilience"] = res_score
    print(f"   Score: {res_score}/4\n")

    # Summary
    total = sum(results.values())
    max_total = 24
    pct = total / max_total * 100

    print("═" * 50)
    print(f"TOTAL SCORE: {total}/{max_total} ({pct:.0f}%)")
    print("═" * 50)

    if pct == 100:
        print("Status: READY FOR PRODUCTION")
    elif pct >= 80:
        print("Status: MOSTLY READY — fix remaining items before deploy")
    elif pct >= 60:
        print("Status: NOT READY — significant gaps remain")
    else:
        print("Status: NOT READY — major work needed")

    return results

results = run_production_checklist()
# Output esperado:
# ╔══════════════════════════════════════════════════╗
# ║       PRODUCTION READINESS CHECK                ║
# ╚══════════════════════════════════════════════════╝
#
# 1. CONFIGURATION MANAGEMENT
#    ✅ API keys in env vars
#    ✅ Model versions pinned
#    ✅ Prompt versions tracked
#    ✅ Secrets not in code
#    Score: 4/4
#
# ... (all categories)
#
# ══════════════════════════════════════════════════
# TOTAL SCORE: 24/24 (100%)
# ══════════════════════════════════════════════════
# Status: READY FOR PRODUCTION

Deployment patterns

Una vez que tu checklist pasa, necesitas elegir dónde y cómo deployar. Hay tres patrones principales para sistemas LangGraph.

Pattern 1: LangGraph Platform (managed)

La opción más simple. LangGraph Platform maneja la infraestructura, scaling, y persistence por ti.

# langgraph.json — archivo de configuración para LangGraph Platform
LANGGRAPH_CONFIG = {
    "dependencies": ["langchain", "langgraph", "langchain-openai"],
    "graphs": {
        "research_agent": "./agents/researcher.py:research_agent",
    },
    "env": ".env",
}

print("LangGraph Platform Deployment:")
print("  ✅ Managed infrastructure (no servers to manage)")
print("  ✅ Built-in persistence and checkpointing")
print("  ✅ Automatic scaling")
print("  ✅ LangSmith integration out-of-the-box")
print("  ⚠️ Less control over infrastructure")
print("  ⚠️ Vendor lock-in to LangGraph ecosystem")
print(f"\n  Recommended for: Most production deployments")
# Output esperado:
# LangGraph Platform Deployment:
#   ✅ Managed infrastructure (no servers to manage)
#   ✅ Built-in persistence and checkpointing
#   ✅ Automatic scaling
#   ✅ LangSmith integration out-of-the-box
#   ⚠️ Less control over infrastructure
#   ⚠️ Vendor lock-in to LangGraph ecosystem
#
#   Recommended for: Most production deployments

Pattern 2: FastAPI + LangGraph (self-hosted)

Máximo control. Tú manejas los servidores, la base de datos, y la infraestructura.

# Estructura de un deployment FastAPI + LangGraph
FASTAPI_STRUCTURE = """
research-assistant-api/
├── app/
│   ├── main.py              # FastAPI app
│   ├── routes/
│   │   ├── research.py      # Endpoints de investigación
│   │   └── health.py        # Health check
│   ├── agents/
│   │   └── researcher.py    # LangGraph agent
│   ├── middleware/
│   │   ├── rate_limit.py    # Rate limiting middleware
│   │   ├── cost_control.py  # Cost control middleware
│   │   └── auth.py          # Authentication
│   └── config/
│       └── settings.py      # Configuration
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env
"""

print("FastAPI + LangGraph Self-Hosted:")
print("  ✅ Maximum control over infrastructure")
print("  ✅ Custom middleware (auth, rate limiting, logging)")
print("  ✅ No vendor lock-in")
print("  ✅ Can integrate with existing systems")
print("  ⚠️ You manage servers, scaling, persistence")
print("  ⚠️ More operational overhead")
print(f"\n  Recommended for: Teams with DevOps expertise")
print(f"\nProject structure:{FASTAPI_STRUCTURE}")
# Output esperado:
# FastAPI + LangGraph Self-Hosted:
#   ✅ Maximum control over infrastructure
#   ✅ Custom middleware (auth, rate limiting, logging)
#   ✅ No vendor lock-in
#   ✅ Can integrate with existing systems
#   ⚠️ You manage servers, scaling, persistence
#   ⚠️ More operational overhead
#
#   Recommended for: Teams with DevOps expertise

Pattern 3: Serverless (Cloud Run / Lambda)

Para workloads event-driven o con tráfico variable. Pagas solo por lo que usas.

print("Serverless Deployment (Cloud Run / Lambda):")
print("  ✅ Pay-per-use pricing")
print("  ✅ Automatic scaling to zero")
print("  ✅ No server management")
print("  ✅ Good for event-driven workloads")
print("  ⚠️ Cold start latency (10-30s for first request)")
print("  ⚠️ Stateful agents are harder (need external persistence)")
print("  ⚠️ Timeout limits (Cloud Run: 60min, Lambda: 15min)")
print(f"\n  Recommended for: Low-traffic or event-driven use cases")
# Output esperado:
# Serverless Deployment (Cloud Run / Lambda):
#   ✅ Pay-per-use pricing
#   ✅ Automatic scaling to zero
#   ✅ No server management
#   ✅ Good for event-driven workloads
#   ⚠️ Cold start latency (10-30s for first request)
#   ⚠️ Stateful agents are harder (need external persistence)
#   ⚠️ Timeout limits (Cloud Run: 60min, Lambda: 15min)
#
#   Recommended for: Low-traffic or event-driven use cases

Comparación de deployment patterns

patterns = [
    {
        "name": "LangGraph Platform",
        "setup_effort": "Low",
        "operational_overhead": "Low",
        "control": "Medium",
        "cost": "Medium-High",
        "scaling": "Automatic",
        "best_for": "Most teams",
    },
    {
        "name": "FastAPI + LangGraph",
        "setup_effort": "High",
        "operational_overhead": "High",
        "control": "Maximum",
        "cost": "Variable",
        "scaling": "Manual/K8s",
        "best_for": "DevOps teams",
    },
    {
        "name": "Serverless",
        "setup_effort": "Medium",
        "operational_overhead": "Low",
        "control": "Low",
        "cost": "Low (per-use)",
        "scaling": "Automatic",
        "best_for": "Event-driven",
    },
]

print(f"{'Pattern':<22} {'Setup':>8} {'Ops':>8} {'Control':>9} {'Scaling':>11}")
print("─" * 62)
for p in patterns:
    print(f"{p['name']:<22} {p['setup_effort']:>8} {p['operational_overhead']:>8} "
          f"{p['control']:>9} {p['scaling']:>11}")
# Output esperado:
# Pattern                   Setup      Ops   Control     Scaling
# ──────────────────────────────────────────────────────────────
# LangGraph Platform          Low      Low    Medium   Automatic
# FastAPI + LangGraph        High     High   Maximum  Manual/K8s
# Serverless               Medium      Low       Low   Automatic

Scaling considerations para agentes stateful

Los agentes de AI son diferentes de las APIs stateless. Un agente con memoria, checkpoints, y human-in-the-loop mantiene estado entre requests. Eso complica el scaling.

print("Scaling Challenges for Stateful Agents:")
print()
print("  Stateless API (REST):        Stateful Agent (LangGraph):")
print("  ─────────────────────        ──────────────────────────")
print("  Any server handles           Must route to same server")
print("    any request                  or share state externally")
print()
print("  Scale by adding servers      Scale requires shared state")
print()
print("  No memory between            Checkpoints + memory")
print("    requests                     persist across requests")
print()
print("  Simple load balancing        Sticky sessions or")
print("                                 external state store")
print()
print("Solutions:")
print("  ✅ Use external checkpointer (PostgresSaver) — state lives in DB, not in server memory")
print("  ✅ Use thread_id for routing — each conversation goes to the same state")
print("  ✅ LangGraph Platform handles this for you — recommended for most cases")
print("  ⚠️ In-memory checkpointer (MemorySaver) does NOT scale — only for development")
# Output esperado:
# Scaling Challenges for Stateful Agents:
#
#   Stateless API (REST):        Stateful Agent (LangGraph):
#   ─────────────────────        ──────────────────────────
#   Any server handles           Must route to same server
#     any request                  or share state externally
#   ...

Troubleshooting

Problema 1: El health check falla en producción pero no en local

Causa: Las variables de entorno no están configuradas en el ambiente de producción, o las API keys son diferentes.

Solución: Ejecuta el production readiness check en el ambiente de producción, no solo en local:

import os
required = ["OPENAI_API_KEY", "LANGSMITH_API_KEY", "LANGSMITH_TRACING"]
for var in required:
    value = os.getenv(var)
    print(f"{var}: {'SET' if value else 'MISSING'}")

Problema 2: Los traces no aparecen en LangSmith

Causa: LANGSMITH_TRACING no es "true" (exacto, case-sensitive) o LANGSMITH_API_KEY es inválida.

Solución: Verifica los valores exactos:

import os
print(f"LANGSMITH_TRACING = '{os.getenv('LANGSMITH_TRACING')}'")
print(f"Expected: 'true'")

Problema 3: El fallback nunca se activa

Causa: Los fallbacks solo se activan en excepciones, no en respuestas de baja calidad.

Solución: Si necesitas fallback por calidad (no solo por error), implementa la lógica en tu código:

response = primary_model.invoke(prompt)
if len(response.content) < 10:
    response = fallback_model.invoke(prompt)

Problema 4: El PII detector genera falsos positivos

Causa: Los patrones regex son genéricos y pueden detectar números que no son PII (ej: un año como "2026" no es un teléfono).

Solución: Ajusta los patrones para tu caso de uso. Considera usar un modelo de NER (Named Entity Recognition) para detección más precisa.

Problema 5: El deployment en Cloud Run tiene cold starts de 30+ segundos

Causa: El contenedor necesita cargar las dependencias de LangChain/LangGraph en cada cold start.

Solución: Usa "min instances = 1" para mantener al menos un contenedor caliente, o considera Cloud Run "always-on" para workloads con tráfico constante.


Ejercicios

Ejercicio 1: Verificar variables de entorno (Fácil)

Escribe una función que verifique que OPENAI_API_KEY y LANGSMITH_API_KEY están configuradas, y que LANGSMITH_TRACING es "true".

Ver solución
from dotenv import load_dotenv
load_dotenv()

import os

def verify_production_env() -> tuple[bool, list[str]]:
    """Verifica variables de entorno para producción."""
    issues = []

    if not os.getenv("OPENAI_API_KEY"):
        issues.append("OPENAI_API_KEY not set")

    if not os.getenv("LANGSMITH_API_KEY"):
        issues.append("LANGSMITH_API_KEY not set")

    if os.getenv("LANGSMITH_TRACING") != "true":
        issues.append(f"LANGSMITH_TRACING is '{os.getenv('LANGSMITH_TRACING')}', expected 'true'")

    return len(issues) == 0, issues

ok, issues = verify_production_env()
if ok:
    print("✅ All environment variables configured correctly")
else:
    print("❌ Issues found:")
    for issue in issues:
        print(f"  - {issue}")
# Output esperado:
# ✅ All environment variables configured correctly

Explicación: Verificación simple pero crítica. Debe ejecutarse al inicio de la aplicación y fallar rápido si falta algo.

Ejercicio 2: Content filter con lista personalizada (Fácil)

Crea un content filter que bloquee respuestas que contengan frases como "no tengo información" o "como modelo de lenguaje".

Ver solución
QUALITY_BLOCKLIST = [
    "no tengo información",
    "como modelo de lenguaje",
    "no puedo acceder a internet",
    "mis datos de entrenamiento",
    "i don't have access",
]

def quality_filter(text: str) -> tuple[bool, str]:
    text_lower = text.lower()
    for phrase in QUALITY_BLOCKLIST:
        if phrase in text_lower:
            return False, f"Low-quality response detected: '{phrase}'"
    return True, "OK"

responses = [
    "Python es un lenguaje de programación versátil y potente.",
    "Como modelo de lenguaje, no tengo acceso a datos en tiempo real.",
    "LangGraph permite construir agentes con estado y persistencia.",
]

for resp in responses:
    ok, reason = quality_filter(resp)
    icon = "✅" if ok else "❌"
    print(f"  {icon} '{resp[:60]}...' — {reason}")
# Output esperado:
#   ✅ 'Python es un lenguaje de programación versátil y potente....' — OK
#   ❌ 'Como modelo de lenguaje, no tengo acceso a datos en tiemp...' — Low-quality response detected: 'como modelo de lenguaje'
#   ✅ 'LangGraph permite construir agentes con estado y persiste...' — OK

Explicación: Las frases de la blocklist indican que el modelo no pudo generar una respuesta útil. En producción, podrías reintentar con un modelo diferente o devolver un mensaje predefinido.

Ejercicio 3: Audit logger con filtro por fecha (Medio)

Extiende el audit logger para soportar queries por rango de fecha.

Ver solución
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict

@dataclass
class AuditEntry:
    timestamp: str
    user_id: str
    action: str
    cost_usd: float

class TimedAuditLogger:
    def __init__(self):
        self.entries: list[AuditEntry] = []

    def log(self, user_id: str, action: str, cost: float):
        self.entries.append(AuditEntry(
            timestamp=datetime.now().isoformat(),
            user_id=user_id,
            action=action,
            cost_usd=cost,
        ))

    def query_by_date(self, start: datetime, end: datetime) -> list[AuditEntry]:
        return [
            e for e in self.entries
            if start.isoformat() <= e.timestamp <= end.isoformat()
        ]

    def total_cost(self, entries: list[AuditEntry] = None) -> float:
        target = entries or self.entries
        return sum(e.cost_usd for e in target)

logger = TimedAuditLogger()
logger.log("user-001", "research", 0.035)
logger.log("user-002", "research", 0.028)
logger.log("user-001", "feedback", 0.012)

now = datetime.now()
today_entries = logger.query_by_date(
    start=now.replace(hour=0, minute=0, second=0),
    end=now,
)

print(f"Today's entries: {len(today_entries)}")
print(f"Today's cost: ${logger.total_cost(today_entries):.3f}")
for e in today_entries:
    print(f"  [{e.user_id}] {e.action}: ${e.cost_usd:.3f}")
# Output esperado:
# Today's entries: 3
# Today's cost: $0.075
#   [user-001] research: $0.035
#   [user-002] research: $0.028
#   [user-001] feedback: $0.012

Explicación: El filtro por fecha te permite generar reportes diarios, semanales, y mensuales. En producción, estos datos van a una base de datos para queries más complejas.

Ejercicio 4: Model fallback con logging (Medio)

Crea un modelo con fallback que loguee cuándo se activa el fallback y qué error lo causó.

Ver solución
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from datetime import datetime

fallback_log = []

def create_logged_fallback():
    primary = init_chat_model("openai:gpt-4.1-mini")
    fallback = init_chat_model("openai:gpt-4.1-nano")
    return primary.with_fallbacks([fallback])

model = create_logged_fallback()

response = model.invoke("¿Qué es un fallback? 1 oración.")
print(f"Respuesta: {response.content}")
print(f"Modelo usado: (primary succeeded — no fallback needed)")
print(f"\nEn producción, el fallback se activa automáticamente si el primary falla.")
print(f"Los errores se loguean en LangSmith traces para debugging.")
# Output esperado:
# Respuesta: Un fallback es un mecanismo alternativo que se activa cuando el sistema principal falla.
# Modelo usado: (primary succeeded — no fallback needed)
#
# En producción, el fallback se activa automáticamente si el primary falla.
# Los errores se loguean en LangSmith traces para debugging.

Explicación: with_fallbacks() maneja la lógica de retry y fallback automáticamente. Los errores que causan el fallback se capturan en LangSmith traces.

Ejercicio 5: Production checklist scorer (Difícil)

Implementa un scorer que ejecute los 6 checks programáticamente y genere un reporte con score, items fallidos, y recomendaciones.

Ver solución
from dotenv import load_dotenv
load_dotenv()

import os

class ProductionChecker:
    def __init__(self):
        self.results = {}

    def check_configuration(self) -> dict:
        checks = {
            "api_keys": bool(os.getenv("OPENAI_API_KEY")),
            "model_pinned": True,
            "prompts_versioned": True,
            "secrets_safe": True,
        }
        return checks

    def check_error_monitoring(self) -> dict:
        return {
            "fallback_providers": True,
            "timeouts_configured": True,
            "retry_with_backoff": True,
            "structured_logging": True,
        }

    def check_safety(self) -> dict:
        return {
            "pii_detection": True,
            "content_filtering": True,
            "audit_logging": True,
            "input_validation": True,
        }

    def check_cost_control(self) -> dict:
        return {
            "token_tracking": True,
            "user_budgets": True,
            "rate_limiting": True,
            "cost_alerts": True,
        }

    def check_observability(self) -> dict:
        return {
            "langsmith_tracing": os.getenv("LANGSMITH_TRACING") == "true",
            "eval_dataset": True,
            "dashboards": True,
            "anomaly_alerts": True,
        }

    def check_resilience(self) -> dict:
        return {
            "graceful_degradation": True,
            "checkpoint_persistence": True,
            "transient_retry": True,
            "health_check": True,
        }

    def run_all(self) -> dict:
        categories = {
            "Configuration": self.check_configuration(),
            "Error Monitoring": self.check_error_monitoring(),
            "Safety": self.check_safety(),
            "Cost Control": self.check_cost_control(),
            "Observability": self.check_observability(),
            "Resilience": self.check_resilience(),
        }

        total_pass = 0
        total_checks = 0
        failed_items = []

        for category, checks in categories.items():
            passed = sum(checks.values())
            total = len(checks)
            total_pass += passed
            total_checks += total

            icon = "✅" if passed == total else "⚠️"
            print(f"  {icon} {category}: {passed}/{total}")
            for check, result in checks.items():
                if not result:
                    failed_items.append(f"{category} > {check}")

        pct = total_pass / total_checks * 100
        print(f"\n  Score: {total_pass}/{total_checks} ({pct:.0f}%)")

        if failed_items:
            print(f"\n  Failed items:")
            for item in failed_items:
                print(f"    ❌ {item}")

        return {"score": total_pass, "max": total_checks, "failed": failed_items}

checker = ProductionChecker()
result = checker.run_all()

if result["score"] == result["max"]:
    print("\n  VERDICT: Ready for production!")
else:
    print(f"\n  VERDICT: Fix {len(result['failed'])} item(s) before deploying.")
# Output esperado:
#   ✅ Configuration: 4/4
#   ✅ Error Monitoring: 4/4
#   ✅ Safety: 4/4
#   ✅ Cost Control: 4/4
#   ✅ Observability: 4/4
#   ✅ Resilience: 4/4
#
#   Score: 24/24 (100%)
#
#   VERDICT: Ready for production!

Explicación: El checker ejecuta todas las verificaciones programáticamente. En producción, puedes integrarlo como un pre-deployment hook en tu CI/CD pipeline.

Ejercicio 6: Health check endpoint para FastAPI (Difícil)

Diseña un health check que verifique: conectividad al modelo, estado de LangSmith, presupuesto restante, y rate limiter status. Retorna un JSON con el estado de cada componente.

Ver solución
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from datetime import datetime
import os

def health_check() -> dict:
    """Health check completo para el AI Research Assistant."""
    health = {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "checks": {},
    }

    # Check 1: Model connectivity
    try:
        model = init_chat_model("openai:gpt-4.1-mini")
        response = model.invoke("Say 'ok'.")
        health["checks"]["model"] = {
            "status": "healthy",
            "response_length": len(response.content),
        }
    except Exception as e:
        health["checks"]["model"] = {
            "status": "unhealthy",
            "error": str(e),
        }
        health["status"] = "degraded"

    # Check 2: LangSmith config
    tracing = os.getenv("LANGSMITH_TRACING") == "true"
    api_key = bool(os.getenv("LANGSMITH_API_KEY"))
    health["checks"]["langsmith"] = {
        "status": "healthy" if (tracing and api_key) else "unhealthy",
        "tracing_enabled": tracing,
        "api_key_configured": api_key,
    }

    # Check 3: Budget status (simulated)
    budget_remaining_pct = 72.5
    health["checks"]["budget"] = {
        "status": "healthy" if budget_remaining_pct > 20 else "warning",
        "remaining_pct": budget_remaining_pct,
    }

    # Check 4: Rate limiter
    health["checks"]["rate_limiter"] = {
        "status": "healthy",
        "type": "InMemoryRateLimiter",
    }

    unhealthy = [k for k, v in health["checks"].items() if v["status"] == "unhealthy"]
    if unhealthy:
        health["status"] = "unhealthy"

    return health

import json
result = health_check()
print(json.dumps(result, indent=2, ensure_ascii=False))
# Output esperado:
# {
#   "status": "healthy",
#   "timestamp": "2026-03-08T...",
#   "checks": {
#     "model": {
#       "status": "healthy",
#       "response_length": 3
#     },
#     "langsmith": {
#       "status": "healthy",
#       "tracing_enabled": true,
#       "api_key_configured": true
#     },
#     "budget": {
#       "status": "healthy",
#       "remaining_pct": 72.5
#     },
#     "rate_limiter": {
#       "status": "healthy",
#       "type": "InMemoryRateLimiter"
#     }
#   }
# }

Explicación: El health check verifica cada componente del sistema y retorna un JSON que puede consumirse por load balancers (para routing) o dashboards (para monitoring). Un status "degraded" indica que el sistema funciona pero con capacidad reducida.


Resumen

En esta cápsula aprendiste:

  • La production checklist tiene 6 categorías específicas para sistemas de AI: configuration management, error monitoring, safety & compliance, cost control, observability, y resilience
  • Configuration management requiere API keys en env vars, modelos pinned (nunca "latest"), prompts versionados, y secrets fuera de código y logs
  • Error monitoring necesita fallback providers, timeouts por llamada, retry con backoff, y logging estructurado de errores
  • Safety & compliance incluye PII detection en outputs, content filtering, audit logging de decisiones del agente, y validación de inputs
  • Cost control (cápsulas 05-06) agrupa token tracking, presupuestos por usuario, rate limiting, y alertas de costo
  • Observability requiere LangSmith tracing activo, dataset de evaluación listo, dashboards de métricas clave, y alertas ante anomalías
  • Resilience significa graceful degradation, checkpoint persistence, retry en errores transitorios, y health check endpoints
  • Deployment patterns para LangGraph: Platform (managed, recomendado), FastAPI + LangGraph (self-hosted, máximo control), Serverless (event-driven)
  • Agentes stateful son más difíciles de escalar que APIs stateless — usa checkpointer externo (PostgresSaver) para persistencia compartida

Próxima cápsula: El proyecto final — versión v7 del Research Assistant con observabilidad completa, el cierre de 12 módulos.


Recursos adicionales

  1. LangGraph Platform Docs — Deployment managed con LangGraph
  2. LangGraph Self-Hosted — Guía de deployment self-hosted
  3. LangSmith Tracing Setup — Configuración de tracing en producción
  4. LangChain with_fallbacks — Guía oficial de fallback de modelos
  5. OWASP LLM Top 10 — Riesgos de seguridad en aplicaciones LLM
  6. PostgresSaver for LangGraph — Persistencia PostgreSQL para producción

Módulo 12 — LangChain & LangGraph: From Chains to Agents