Debugging Visual de Agentes

Descripción de la cápsula

LangSmith transforma el debugging de agentes de "agregar print statements y rezar" a "hacer click en el trace y VER qué pasó." Cuando tu agente produce un resultado inesperado, no necesitas re-ejecutar, no necesitas adivinar, no necesitas gastar dinero en API calls extra. Abres el trace, navegas la línea temporal, y encuentras el paso exacto donde algo salió mal.

Esto no es un nice-to-have. Es la diferencia entre resolver un bug en 2 minutos y pasar una hora adivinando. Y en producción, donde los bugs cuestan dinero real y afectan usuarios reales, esa diferencia es crítica.

En la cápsula anterior aprendiste a configurar tracing y leer traces individuales. En esta, aprendes el workflow completo de debugging: desde "el output está mal" hasta "encontré la causa raíz y la resolví."

El workflow de debugging con LangSmith

Cada vez que tu agente produce un resultado inesperado, sigue estos 7 pasos:

1. El agente produce output inesperado
     ↓
2. Abre LangSmith → encuentra el trace de esa ejecución
     ↓
3. Mira la línea temporal: ¿cuántos pasos? ¿Hubo errores?
     ↓
4. Navega paso a paso: click en cada run para ver input/output
     ↓
5. Identifica el paso donde las cosas se desviaron
     ↓
6. Examina el input y output de ese paso en detalle
     ↓
7. Causa raíz → fix → verificar

Veamos cada paso con un ejemplo concreto.

Escenario de debugging: el reporte mezclado

Tu Research Assistant investigó "AI en educación" pero el reporte mezcla datos de educación con datos de finanzas. ¿Qué pasó?

Paso 1: Reproducir y trazar

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class ResearchState(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    analysis: str
    report: str

def search_web(state: ResearchState) -> dict:
    return {"sources": [
        f"Web: AI está transformando la educación con tutores personalizados",
        f"Web: Los bancos usan AI para detección de fraude",
    ]}

def search_papers(state: ResearchState) -> dict:
    return {"sources": [
        f"Paper: 'Adaptive Learning with LLMs' (2025)",
    ]}

def analyze(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analiza estas fuentes sobre '{state['topic']}'. "
        f"Identifica las 2 tendencias principales.\n\n"
        f"Fuentes:\n" + "\n".join(f"- {s}" for s in state["sources"])
    )
    return {"analysis": response.content}

def write_report(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Escribe un reporte breve (3 oraciones) sobre '{state['topic']}' "
        f"basado en este análisis:\n\n{state['analysis']}"
    )
    return {"report": response.content}

builder = StateGraph(ResearchState)
builder.add_node("search_web", search_web)
builder.add_node("search_papers", search_papers)
builder.add_node("analyze", analyze)
builder.add_node("report", write_report)

builder.add_edge(START, "search_web")
builder.add_edge(START, "search_papers")
builder.add_edge("search_web", "analyze")
builder.add_edge("search_papers", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Debug: Reporte Mezclado",
    tags=["debug-exercise"],
    metadata={"scenario": "mixed-report"},
)

result = graph.invoke(
    {"topic": "AI en educación", "sources": [], "analysis": "", "report": ""},
    config=config,
)

print(f"REPORTE FINAL:\n{result['report']}")
print(f"\n⚠️  ¿El reporte menciona finanzas/bancos? Si sí, hay un bug.")
print(f"\nPara debuggear, sigue los pasos en LangSmith:")
print(f"  1. Abre el trace 'Debug: Reporte Mezclado'")
print(f"  2. Mira el child run 'search_web' → el output incluye datos de finanzas")
print(f"  3. Causa raíz: search_web retornó una fuente irrelevante sobre bancos")
print(f"  4. Fix: mejorar la búsqueda para filtrar por relevancia al topic")
# Output esperado:
# REPORTE FINAL:
# AI está transformando la educación con tutores personalizados...
# (posiblemente menciona detección de fraude bancario)
#
# ⚠️  ¿El reporte menciona finanzas/bancos? Si sí, hay un bug.

Paso 2-4: Lo que ves en LangSmith

En el dashboard de LangSmith, el trace "Debug: Reporte Mezclado" muestra:

Trace: "Debug: Reporte Mezclado"          Total: 2.3s
│
├─ search_web (0.01s)
│   Output: ["Web: AI está transformando la educación...",
│            "Web: Los bancos usan AI para detección de fraude"]
│                                          ← ⚠️ DATO IRRELEVANTE
│
├─ search_papers (0.01s)
│   Output: ["Paper: 'Adaptive Learning with LLMs' (2025)"]
│
├─ analyze (1.2s) — gpt-4.1-mini
│   Input: "Analiza estas fuentes sobre 'AI en educación'...
│           - Web: AI está transformando la educación...
│           - Web: Los bancos usan AI para detección de fraude  ← AQUÍ ESTÁ
│           - Paper: 'Adaptive Learning with LLMs'"
│   Output: "Tendencia 1: Tutores personalizados...
│            Tendencia 2: Detección de fraude..."  ← BUG PROPAGADO
│
└─ report (1.1s) — gpt-4.1-mini
    Input: análisis contaminado con datos de finanzas
    Output: reporte que mezcla educación y finanzas

Paso 5-7: Causa raíz identificada

El bug está en search_web: retorna una fuente sobre bancos que no es relevante para "AI en educación." El análisis la incluye porque recibió las fuentes sin filtrar. El reporte la refleja porque se basó en el análisis contaminado.

La causa raíz no es el modelo — es la búsqueda. El fix es agregar filtrado de relevancia después de la búsqueda.

Visualizar la ejecución del grafo

LangSmith muestra la estructura del grafo en el trace. Para cada ejecución, puedes ver exactamente qué nodos se visitaron y en qué orden:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    query: str
    category: str
    result: str
    path: str

def classify(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica esta consulta como 'factual' o 'opinion'. "
        f"Responde SOLO con una palabra.\n\n{state['query']}"
    )
    category = response.content.strip().lower()
    return {"category": category}

def handle_factual(state: State) -> dict:
    return {
        "result": f"[FACTUAL] Respuesta basada en datos para: {state['query']}",
        "path": "classify → factual → end",
    }

def handle_opinion(state: State) -> dict:
    return {
        "result": f"[OPINION] Perspectiva analítica sobre: {state['query']}",
        "path": "classify → opinion → end",
    }

def route(state: State) -> str:
    return "factual" if "factual" in state["category"] else "opinion"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("factual", handle_factual)
builder.add_node("opinion", handle_opinion)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {
    "factual": "factual",
    "opinion": "opinion",
})
builder.add_edge("factual", END)
builder.add_edge("opinion", END)

graph = builder.compile()

queries = [
    ("¿Cuántos parámetros tiene GPT-4?", "factual-query"),
    ("¿Es mejor usar RAG o fine-tuning?", "opinion-query"),
]

for query, tag in queries:
    config = RunnableConfig(
        run_name=f"Routing: {query[:40]}",
        tags=[tag, "routing-debug"],
    )
    result = graph.invoke({"query": query, "category": "", "result": "", "path": ""}, config=config)
    print(f"Query: {query}")
    print(f"  Categoría: {result['category']}")
    print(f"  Path: {result['path']}")
    print(f"  Resultado: {result['result'][:60]}...")
    print()
# Output esperado:
# Query: ¿Cuántos parámetros tiene GPT-4?
#   Categoría: factual
#   Path: classify → factual → end
#   Resultado: [FACTUAL] Respuesta basada en datos para: ¿Cuántos paráme...
#
# Query: ¿Es mejor usar RAG o fine-tuning?
#   Categoría: opinion
#   Path: classify → opinion → end
#   Resultado: [OPINION] Perspectiva analítica sobre: ¿Es mejor usar RA...

En LangSmith, los dos traces muestran caminos diferentes en el grafo. Si la clasificación es incorrecta (la query factual se routeó a "opinion"), ves exactamente: qué prompt recibió el clasificador, qué respondió, y a qué nodo se envió la consulta.

Identificar bottlenecks: el nodo más lento

Cuando la latencia de tu agente es inaceptable, necesitas saber dónde se va el tiempo. LangSmith muestra la duración de cada operación como barras en una línea temporal — el bottleneck es visual e inmediato:

from dotenv import load_dotenv
load_dotenv()

import time
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    topic: str
    web_results: str
    paper_results: str
    analysis: str
    report: str
    timings: Annotated[list[str], operator.add]

def search_web(state: State) -> dict:
    start = time.time()
    time.sleep(0.3)
    elapsed = time.time() - start
    return {
        "web_results": f"3 artículos sobre {state['topic']}",
        "timings": [f"search_web: {elapsed:.2f}s"],
    }

def search_papers(state: State) -> dict:
    start = time.time()
    time.sleep(0.2)
    elapsed = time.time() - start
    return {
        "paper_results": f"2 papers sobre {state['topic']}",
        "timings": [f"search_papers: {elapsed:.2f}s"],
    }

def analyze(state: State) -> dict:
    start = time.time()
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analiza brevemente (1 oración): {state['web_results']}, {state['paper_results']}"
    )
    elapsed = time.time() - start
    return {
        "analysis": response.content,
        "timings": [f"analyze: {elapsed:.2f}s"],
    }

def write_report(state: State) -> dict:
    start = time.time()
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Escribe un resumen de 1 oración: {state['analysis']}"
    )
    elapsed = time.time() - start
    return {
        "report": response.content,
        "timings": [f"write_report: {elapsed:.2f}s"],
    }

builder = StateGraph(State)
builder.add_node("search_web", search_web)
builder.add_node("search_papers", search_papers)
builder.add_node("analyze", analyze)
builder.add_node("report", write_report)

builder.add_edge(START, "search_web")
builder.add_edge(START, "search_papers")
builder.add_edge("search_web", "analyze")
builder.add_edge("search_papers", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Bottleneck Analysis",
    tags=["performance"],
)

total_start = time.time()
result = graph.invoke(
    {"topic": "AI observability", "web_results": "", "paper_results": "",
     "analysis": "", "report": "", "timings": []},
    config=config,
)
total_elapsed = time.time() - total_start

print(f"Reporte: {result['report'][:80]}...")
print(f"\n=== ANÁLISIS DE PERFORMANCE ===")
print(f"Total: {total_elapsed:.2f}s\n")

for timing in result["timings"]:
    name, duration_str = timing.split(": ")
    duration = float(duration_str.replace("s", ""))
    pct = (duration / total_elapsed) * 100
    bar = "█" * int(pct / 2)
    print(f"  {name:<15} {duration_str:>6} ({pct:4.0f}%) {bar}")

print(f"\n→ En LangSmith, las barras de la timeline son proporcionales al tiempo")
print(f"→ El nodo más largo es tu bottleneck — enfoca la optimización ahí")
# Output esperado:
# Reporte: Las fuentes analizadas muestran que AI observability es...
#
# === ANÁLISIS DE PERFORMANCE ===
# Total: 2.50s
#
#   search_web       0.30s ( 12%) ██████
#   search_papers    0.20s (  8%) ████
#   analyze          1.20s ( 48%) ████████████████████████
#   write_report     0.80s ( 32%) ████████████████

Debugging de tool calls: argumentos y respuestas

Cuando un agente usa tools, cada tool call aparece en el trace con los argumentos exactos que envió y la respuesta que recibió:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_core.runnables import RunnableConfig

@tool
def search_database(query: str, limit: int = 5) -> str:
    """Busca en la base de datos de investigación."""
    if "error" in query.lower():
        raise ValueError(f"Query inválido: '{query}'")
    return f"[DB] {limit} resultados para '{query}': datos relevantes encontrados."

@tool
def analyze_data(data: str, method: str = "statistical") -> str:
    """Analiza datos con un método específico."""
    return f"[ANALYSIS] Método {method}: {data[:50]}... → 3 insights encontrados."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_react_agent(model, [search_database, analyze_data])

config = RunnableConfig(
    run_name="Debug Tool Calls",
    tags=["tool-debug"],
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Busca papers sobre RAG y analízalos estadísticamente"}]},
    config=config,
)

print(f"Respuesta: {result['messages'][-1].content[:150]}...")
print(f"\nEn LangSmith verás cada tool call:")
print(f"  1. search_database(query='RAG', limit=5) → output")
print(f"  2. analyze_data(data=..., method='statistical') → output")
print(f"\n→ Si el agente pasó argumentos incorrectos, lo ves inmediatamente")
print(f"→ Si la tool retornó datos inesperados, lo ves inmediatamente")
# Output esperado:
# Respuesta: Encontré 5 resultados sobre RAG en la base de datos...
#
# En LangSmith verás cada tool call:
#   1. search_database(query='RAG', limit=5) → output
#   2. analyze_data(data=..., method='statistical') → output

Comparar traces: ejecución buena vs mala

Uno de los usos más poderosos de LangSmith es comparar dos ejecuciones: una que produjo buen resultado y otra que falló. La diferencia entre ambas te muestra exactamente qué cambió:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    query: str
    context: str
    response: str

def get_context(state: State) -> dict:
    contexts = {
        "buena": "RAG combina retrieval con generación. Se implementa con vector stores. "
                 "Las métricas clave son precision@k y recall@k.",
        "mala": "",
    }
    variant = "buena" if "RAG" in state["query"] else "mala"
    return {"context": contexts.get(variant, "")}

def generate(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    prompt = f"Responde basándote SOLO en este contexto:\n\n"
    prompt += f"Contexto: {state['context'] or 'Sin contexto disponible.'}\n\n"
    prompt += f"Pregunta: {state['query']}"
    response = model.invoke(prompt)
    return {"response": response.content}

builder = StateGraph(State)
builder.add_node("context", get_context)
builder.add_node("generate", generate)
builder.add_edge(START, "context")
builder.add_edge("context", "generate")
builder.add_edge("generate", END)

graph = builder.compile()

config_good = RunnableConfig(
    run_name="[GOOD] RAG query",
    tags=["comparison", "good"],
    metadata={"expected_quality": "high"},
)
good = graph.invoke(
    {"query": "¿Cómo funciona RAG?", "context": "", "response": ""},
    config=config_good,
)

config_bad = RunnableConfig(
    run_name="[BAD] Empty context query",
    tags=["comparison", "bad"],
    metadata={"expected_quality": "low"},
)
bad = graph.invoke(
    {"query": "¿Cómo funciona quantum computing?", "context": "", "response": ""},
    config=config_bad,
)

print(f"BUENA ejecución:")
print(f"  Context: {good.get('context', '')[:60]}...")
print(f"  Response: {good['response'][:80]}...")
print(f"\nMALA ejecución:")
print(f"  Context: '{bad.get('context', '')[:60] or 'VACÍO'}'")
print(f"  Response: {bad['response'][:80]}...")
print(f"\n=== COMPARACIÓN ===")
print(f"Diferencia: la ejecución buena tiene contexto relevante.")
print(f"La mala tiene contexto vacío → el modelo improvisa → respuesta de baja calidad.")
print(f"\nEn LangSmith, filtra por tag 'comparison' y compara ambos traces:")
print(f"  [GOOD] → context tiene datos → generate produce respuesta informada")
print(f"  [BAD]  → context está vacío → generate produce respuesta genérica")
# Output esperado:
# BUENA ejecución:
#   Context: RAG combina retrieval con generación. Se implementa con vec...
#   Response: RAG funciona combinando un sistema de retrieval con generación...
#
# MALA ejecución:
#   Context: 'VACÍO'
#   Response: Sin contexto específico disponible, no puedo proporcionar...

En LangSmith, abres ambos traces lado a lado. En el trace "GOOD", el nodo context produce datos útiles. En el trace "BAD", produce un string vacío. La causa raíz es clara: el nodo de contexto no encontró datos para "quantum computing."

Error traces: cuando algo explota

Cuando un nodo lanza una excepción, LangSmith captura el error completo incluyendo el traceback, el estado del agente en ese momento, y qué nodos ya se habían ejecutado:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    data: str
    processed: str
    log: Annotated[list[str], operator.add]

def fetch_data(state: State) -> dict:
    return {"data": "raw data from API", "log": ["fetch: OK"]}

def process_data(state: State) -> dict:
    if not state["data"]:
        raise ValueError("No hay datos para procesar")
    result = state["data"].upper()
    return {"processed": result, "log": ["process: OK"]}

def validate(state: State) -> dict:
    if len(state["processed"]) < 5:
        raise ValueError(f"Datos procesados demasiado cortos: '{state['processed']}'")
    return {"log": ["validate: OK"]}

builder = StateGraph(State)
builder.add_node("fetch", fetch_data)
builder.add_node("process", process_data)
builder.add_node("validate", validate)
builder.add_edge(START, "fetch")
builder.add_edge("fetch", "process")
builder.add_edge("process", "validate")
builder.add_edge("validate", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Error Trace Demo",
    tags=["error-debug"],
)

try:
    result = graph.invoke({"data": "", "processed": "", "log": []}, config=config)
    print(f"Resultado: {result}")
except Exception as e:
    print(f"Error capturado: {type(e).__name__}: {e}")
    print(f"\nEn LangSmith verás:")
    print(f"  ✅ fetch: completado (retornó 'raw data from API')")
    print(f"  ✅ process: completado (retornó 'RAW DATA FROM API')")
    print(f"  ✅ validate: completado (datos > 5 caracteres)")
    print(f"\nSi el fetch retornara datos vacíos, verías:")
    print(f"  ✅ fetch: completado (retornó '')")
    print(f"  ❌ process: ERROR — 'No hay datos para procesar'")
    print(f"     Estado en el momento del error: data=''")
    print(f"  ⏭️  validate: no ejecutado")
# Output esperado:
# Resultado: {'data': 'raw data from API', 'processed': 'RAW DATA FROM API', 'log': ['fetch: OK', 'process: OK', 'validate: OK']}

En el trace de LangSmith, un error aparece en rojo con el traceback completo. Puedes ver exactamente qué estado tenía el agente cuando falló, qué nodos ya se habían ejecutado correctamente, y qué nodos nunca se ejecutaron.

Conexión con time-travel debugging (M8)

En el Módulo 8 aprendiste time-travel debugging con checkpoints: navegar el historial de estados de tu agente localmente. LangSmith complementa esto con una capa visual y remota:

Usa time-travel cuando necesitas manipular el estado (fork, replay, update_state). Usa LangSmith cuando necesitas ver el panorama completo de una ejecución y compartir el debugging con tu equipo.

Patrones de debugging en producción

Patrón 1: Debugging reactivo — "un usuario reportó un problema"

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

user_id = "user_abc123"
runs = list(client.list_runs(
    project_name="research-assistant-prod",
    filter=f'has(metadata, {{"user_id": "{user_id}"}})',
    limit=5,
))

print(f"Últimas ejecuciones de {user_id}:")
for run in runs:
    status = "✅" if run.status == "success" else "❌"
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    print(f"  {status} {run.name} — {latency:.1f}s — {run.total_tokens or 0} tokens")
    if run.error:
        print(f"     Error: {run.error[:100]}")
# Output esperado:
# Últimas ejecuciones de user_abc123:
#   ✅ Research Query - AI Safety — 3.2s — 1450 tokens
#   ❌ Research Query - Quantum — 0.5s — 0 tokens
#      Error: RateLimitError: Rate limit exceeded...
#   ✅ Research Query - RAG — 2.8s — 1200 tokens

Patrón 2: Debugging proactivo — "¿hay algo raro hoy?"

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

since = datetime.now() - timedelta(hours=6)

runs = list(client.list_runs(
    project_name="research-assistant-prod",
    is_root=True,
    start_time=since,
    limit=50,
))

if runs:
    errors = [r for r in runs if r.status == "error"]
    latencies = [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    slow_runs = [l for l in latencies if l > 5.0]

    print(f"=== HEALTH CHECK (últimas 6 horas) ===")
    print(f"  Total ejecuciones: {len(runs)}")
    print(f"  Errores: {len(errors)} ({len(errors)/len(runs)*100:.0f}%)")
    if latencies:
        print(f"  Latencia promedio: {sum(latencies)/len(latencies):.1f}s")
        print(f"  Ejecuciones lentas (>5s): {len(slow_runs)}")

    if len(errors) / len(runs) > 0.1:
        print(f"\n  ⚠️  ALERTA: tasa de error > 10%")
    if slow_runs and len(slow_runs) / len(runs) > 0.2:
        print(f"  ⚠️  ALERTA: > 20% de ejecuciones son lentas")
# Output esperado:
# === HEALTH CHECK (últimas 6 horas) ===
#   Total ejecuciones: 45
#   Errores: 3 (7%)
#   Latencia promedio: 3.2s
#   Ejecuciones lentas (>5s): 4

Patrón 3: Debugging comparativo — "¿la nueva versión es mejor?"

Cuando cambias el prompt de un agente, usas tags de versión para comparar:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

def get_version_stats(project: str, version_tag: str) -> dict:
    runs = list(client.list_runs(
        project_name=project,
        is_root=True,
        filter=f'has(tags, "{version_tag}")',
        limit=20,
    ))
    if not runs:
        return {"count": 0, "avg_latency": 0, "avg_tokens": 0, "error_rate": 0}

    latencies = [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    tokens = [r.total_tokens for r in runs if r.total_tokens]
    errors = sum(1 for r in runs if r.status == "error")

    return {
        "count": len(runs),
        "avg_latency": sum(latencies) / len(latencies) if latencies else 0,
        "avg_tokens": sum(tokens) / len(tokens) if tokens else 0,
        "error_rate": errors / len(runs) if runs else 0,
    }

v1 = get_version_stats("research-assistant-prod", "v6")
v2 = get_version_stats("research-assistant-prod", "v7")

print(f"{'Métrica':<20} {'v6':>10} {'v7':>10} {'Cambio':>10}")
print(f"{'-'*50}")
print(f"{'Ejecuciones':<20} {v1['count']:>10} {v2['count']:>10}")
print(f"{'Latencia (s)':<20} {v1['avg_latency']:>10.1f} {v2['avg_latency']:>10.1f}")
print(f"{'Tokens promedio':<20} {v1['avg_tokens']:>10.0f} {v2['avg_tokens']:>10.0f}")
print(f"{'Tasa de error':<20} {v1['error_rate']:>9.0%} {v2['error_rate']:>9.0%}")
# Output esperado:
# Métrica                    v6         v7     Cambio
# --------------------------------------------------
# Ejecuciones                15         20
# Latencia (s)              3.5        2.8
# Tokens promedio          1400       1100
# Tasa de error              7%         5%

Troubleshooting

Problema 1: "No veo la estructura del grafo en el trace"

Síntoma: El trace muestra operaciones planas en vez de la jerarquía del grafo.

Causa: El tracing captura la jerarquía automáticamente, pero si usas model.invoke() fuera de un grafo, no hay estructura de grafo que mostrar.

Solución: Verifica que estás invocando el grafo compilado (graph.invoke()), no nodos individuales. La jerarquía aparece automáticamente cuando LangGraph gestiona la ejecución.

Problema 2: "El trace no muestra los argumentos de la tool call"

Síntoma: Ves que la tool se ejecutó pero no ves qué argumentos recibió.

Causa: La tool no está decorada correctamente o el tracing no capturó los detalles.

Solución: Asegúrate de usar @tool de langchain_core.tools. Las tools nativas de LangChain reportan argumentos automáticamente.

Problema 3: "Quiero comparar dos traces pero no los encuentro"

Síntoma: Necesitas comparar una ejecución buena con una mala.

Causa: Sin naming/tagging consistente, encontrar traces específicos es difícil.

Solución: Usa run_name descriptivos y tags para categorizar. Ejemplo: tags=["research", "v7", "user_abc"] permite filtrar por cualquier combinación.

Problema 4: "El error trace no muestra suficiente contexto"

Síntoma: Ves que un nodo falló pero no entiendes por qué.

Causa: El estado del agente en el momento del error puede no ser obvio solo con el traceback.

Solución: Combina LangSmith con time-travel debugging (M8). El trace te dice dónde falló; get_state_history() te da el estado completo en ese punto.

Ejercicios

Ejercicio 1: Debugging de routing incorrecto (Fácil)

Crea un grafo con un router que clasifica consultas como "técnica" o "negocio". Envía una consulta ambigua ("¿Cuánto cuesta implementar RAG?") y usa el trace en LangSmith para verificar qué clasificación eligió el modelo y por qué.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    query: str
    classification: str
    result: str

def classify(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica como 'technical' o 'business'. Responde SOLO una palabra.\n\n"
        f"Consulta: {state['query']}"
    )
    return {"classification": response.content.strip().lower()}

def technical(state: State) -> dict:
    return {"result": f"[TECH] Análisis técnico: {state['query']}"}

def business(state: State) -> dict:
    return {"result": f"[BIZ] Análisis de negocio: {state['query']}"}

def route(state: State) -> str:
    return "technical" if "technical" in state["classification"] else "business"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("technical", technical)
builder.add_node("business", business)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {"technical": "technical", "business": "business"})
builder.add_edge("technical", END)
builder.add_edge("business", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Route Debug: Ambiguous Query",
    tags=["routing-debug"],
)

result = graph.invoke(
    {"query": "¿Cuánto cuesta implementar RAG?", "classification": "", "result": ""},
    config=config,
)

print(f"Query: {result['query']}")
print(f"Clasificación: {result['classification']}")
print(f"Resultado: {result['result']}")
print(f"\n→ Abre el trace en LangSmith")
print(f"→ Click en el run 'classify' → ve el prompt y la respuesta del modelo")
print(f"→ ¿La clasificación es correcta? La query es ambigua (tiene aspecto técnico Y de negocio)")
# Output esperado:
# Query: ¿Cuánto cuesta implementar RAG?
# Clasificación: business
# Resultado: [BIZ] Análisis de negocio: ¿Cuánto cuesta implementar RAG?

Ejercicio 2: Encontrar el nodo problemático (Medio)

Crea un grafo de 4 nodos donde el tercer nodo tiene un bug intencional (agrega datos basura al resultado). Ejecuta con tracing. Luego usa el SDK de LangSmith para obtener los child runs del trace y encontrar programáticamente cuál nodo introdujo el dato basura.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.runnables import RunnableConfig

class State(TypedDict):
    data: Annotated[list[str], operator.add]

def collect_a(state: State) -> dict:
    return {"data": ["Dato válido A: transformers"]}

def collect_b(state: State) -> dict:
    return {"data": ["Dato válido B: attention"]}

def collect_c(state: State) -> dict:
    return {"data": ["BASURA: compra ahora con descuento!!!"]}

def summarize(state: State) -> dict:
    return {"data": [f"Resumen de {len(state['data'])} items"]}

builder = StateGraph(State)
builder.add_node("a", collect_a)
builder.add_node("b", collect_b)
builder.add_node("c_buggy", collect_c)
builder.add_node("summarize", summarize)

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "c_buggy")
builder.add_edge("c_buggy", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Find the Buggy Node",
    tags=["bug-hunt"],
)

result = graph.invoke({"data": []}, config=config)

print(f"Datos finales: {result['data']}")
print(f"\n=== ANÁLISIS AUTOMÁTICO ===")

suspect_keywords = ["compra", "descuento", "gratis", "click", "spam"]
for i, item in enumerate(result["data"]):
    is_suspect = any(kw in item.lower() for kw in suspect_keywords)
    status = "⚠️ SOSPECHOSO" if is_suspect else "✅ OK"
    print(f"  [{status}] {item}")

print(f"\n→ En LangSmith, click en 'c_buggy' → verás que su output es basura")
print(f"→ Los nodos 'a' y 'b' produjeron datos válidos")
# Output esperado:
# Datos finales: ['Dato válido A: transformers', 'Dato válido B: attention', 'BASURA: compra ahora con descuento!!!', 'Resumen de 3 items']
#
# === ANÁLISIS AUTOMÁTICO ===
#   [✅ OK] Dato válido A: transformers
#   [✅ OK] Dato válido B: attention
#   [⚠️ SOSPECHOSO] BASURA: compra ahora con descuento!!!
#   [✅ OK] Resumen de 3 items

Ejercicio 3: Comparar ejecución buena vs mala (Medio)

Crea un agente que genera resúmenes. Ejecútalo dos veces: una con contexto rico y otra con contexto vacío. Etiqueta ambas con tags "good" y "bad". Luego usa el SDK para obtener ambos traces y comparar: latencia, tokens, y status.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
import time

class State(TypedDict):
    context: str
    summary: str

def summarize(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    prompt = f"Resume en 1 oración:\n\n{state['context'] or 'Sin contexto.'}"
    response = model.invoke(prompt)
    return {"summary": response.content}

builder = StateGraph(State)
builder.add_node("summarize", summarize)
builder.add_edge(START, "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()

good_config = RunnableConfig(
    run_name="[GOOD] Rich Context",
    tags=["comparison-ex", "good"],
)
good_result = graph.invoke(
    {"context": "RAG combina retrieval y generación. Usa vector stores como Chroma o Pinecone. "
                "Las métricas clave son precision@k y recall@k. En 2025, hybrid search supera a dense retrieval.",
     "summary": ""},
    config=good_config,
)

bad_config = RunnableConfig(
    run_name="[BAD] Empty Context",
    tags=["comparison-ex", "bad"],
)
bad_result = graph.invoke(
    {"context": "", "summary": ""},
    config=bad_config,
)

print(f"GOOD: {good_result['summary'][:80]}...")
print(f"BAD:  {bad_result['summary'][:80]}...")

time.sleep(3)

from langsmith import Client
client = Client()

for tag_filter, label in [("good", "GOOD"), ("bad", "BAD")]:
    runs = list(client.list_runs(
        project_name="research-assistant",
        filter=f'and(has(tags, "comparison-ex"), has(tags, "{tag_filter}"))',
        limit=1,
    ))
    if runs:
        run = runs[0]
        latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
        print(f"\n[{label}] {run.name}")
        print(f"  Status: {run.status}")
        print(f"  Latencia: {latency:.2f}s")
        print(f"  Tokens: {run.total_tokens or 'N/A'}")
# Output esperado:
# GOOD: RAG es una técnica que combina retrieval con generación utilizando...
# BAD:  No hay contexto disponible para resumir.
#
# [GOOD] [GOOD] Rich Context
#   Status: success
#   Latencia: 0.95s
#   Tokens: 120
#
# [BAD] [BAD] Empty Context
#   Status: success
#   Latencia: 0.62s
#   Tokens: 35

Ejercicio 4: Health check automatizado (Avanzado)

Escribe un script que funcione como health check: obtiene los últimos 20 traces de producción, calcula tasa de error, latencia p50/p95, tokens promedio, y genera un reporte con alertas si alguna métrica supera un umbral.

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

runs = list(client.list_runs(
    project_name="research-assistant",
    is_root=True,
    limit=20,
))

print(f"{'='*60}")
print(f" HEALTH CHECK — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f" Últimas {len(runs)} ejecuciones")
print(f"{'='*60}\n")

if not runs:
    print("  ⚠️  No hay traces disponibles.")
else:
    errors = [r for r in runs if r.status == "error"]
    latencies = sorted(
        [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    )
    tokens = [r.total_tokens for r in runs if r.total_tokens]

    error_rate = len(errors) / len(runs)
    p50 = latencies[len(latencies) // 2] if latencies else 0
    p95 = latencies[int(len(latencies) * 0.95)] if latencies else 0
    avg_tokens = sum(tokens) / len(tokens) if tokens else 0

    THRESHOLDS = {
        "error_rate": 0.10,
        "p95_latency": 5.0,
        "avg_tokens": 3000,
    }

    print(f"  Métrica             Valor        Umbral       Estado")
    print(f"  {'-'*55}")

    er_status = "❌ ALERTA" if error_rate > THRESHOLDS["error_rate"] else "✅ OK"
    print(f"  Tasa de error       {error_rate:>6.0%}       <{THRESHOLDS['error_rate']:.0%}         {er_status}")

    p95_status = "❌ ALERTA" if p95 > THRESHOLDS["p95_latency"] else "✅ OK"
    print(f"  Latencia p50        {p50:>6.1f}s      —            —")
    print(f"  Latencia p95        {p95:>6.1f}s      <{THRESHOLDS['p95_latency']:.0f}s          {p95_status}")

    tok_status = "❌ ALERTA" if avg_tokens > THRESHOLDS["avg_tokens"] else "✅ OK"
    print(f"  Tokens promedio     {avg_tokens:>6.0f}       <{THRESHOLDS['avg_tokens']}        {tok_status}")

    alerts = []
    if error_rate > THRESHOLDS["error_rate"]:
        alerts.append(f"Tasa de error alta: {error_rate:.0%}")
    if p95 > THRESHOLDS["p95_latency"]:
        alerts.append(f"Latencia p95 alta: {p95:.1f}s")
    if avg_tokens > THRESHOLDS["avg_tokens"]:
        alerts.append(f"Consumo de tokens alto: {avg_tokens:.0f}")

    if alerts:
        print(f"\n  ⚠️  ALERTAS:")
        for alert in alerts:
            print(f"    - {alert}")
    else:
        print(f"\n  ✅ Todo dentro de parámetros normales")
# Output esperado:
# ============================================================
#  HEALTH CHECK — 2025-12-15 15:45
#  Últimas 20 ejecuciones
# ============================================================
#
#   Métrica             Valor        Umbral       Estado
#   -------------------------------------------------------
#   Tasa de error          5%       <10%         ✅ OK
#   Latencia p50          2.3s      —            —
#   Latencia p95          4.8s      <5s          ✅ OK
#   Tokens promedio       1200       <3000        ✅ OK
#
#   ✅ Todo dentro de parámetros normales

Resumen

En esta cápsula aprendiste:

• El workflow de debugging con LangSmith es sistemático: output inesperado → encontrar trace → navegar timeline → identificar paso problemático → examinar input/output → causa raíz → fix. No más "agregar print y re-ejecutar"
• Visualizar la ejecución del grafo muestra qué nodos se visitaron, en qué orden, y qué camino tomó el routing. Si el agente tomó un camino incorrecto, lo ves inmediatamente
• Identificar bottlenecks es visual: las barras de la timeline son proporcionales al tiempo. La barra más larga es tu bottleneck — enfoca la optimización ahí
• Debugging de tool calls muestra argumentos exactos y respuestas. Si el agente envió argumentos incorrectos a una tool, no necesitas adivinar — lo ves en el trace
• Comparar traces (bueno vs malo) revela exactamente qué cambió entre una ejecución exitosa y una fallida. Es el debugging más eficiente que existe
• Error traces capturan el estado completo del agente al momento del fallo: traceback, estado, nodos ejecutados, nodos pendientes
• Tres patrones de debugging en producción: reactivo (un usuario reportó), proactivo (health check periódico), y comparativo (¿la nueva versión es mejor?)

Próxima cápsula: Evaluation: Datasets y Evaluators — cómo pasar de "parece que funciona" a mediciones sistemáticas de calidad con criterios específicos.

Recursos adicionales

1. LangSmith — Tracing FAQ — Preguntas frecuentes sobre tracing y debugging
2. LangSmith — Filter runs — Cómo filtrar y buscar traces en el dashboard
3. LangGraph — Debugging — Debugging de grafos con tracing integrado
4. LangSmith SDK — list_runs — Referencia del SDK para acceso programático a traces
5. LangSmith — Comparison View — Cómo comparar ejecuciones en el dashboard

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