Tracing y Observabilidad

Descripción de la cápsula

Tracing registra cada operación que tu agente realiza: llamadas al modelo, ejecuciones de tools, cambios de estado, decisiones de routing. En LangSmith, todo esto aparece en una línea temporal visual que puedes explorar paso a paso.

Pero configurar tracing es una línea de código. La habilidad real es leer traces. Saber identificar cuál fue la operación más lenta, ver exactamente qué prompt se envió al modelo, detectar tool calls redundantes que desperdician tokens, y encontrar el punto exacto donde el agente tomó una decisión incorrecta. Eso es lo que esta cápsula te enseña.

Setup: una línea, tracing completo

Configurar LangSmith tracing requiere exactamente una variable de entorno:

from dotenv import load_dotenv
load_dotenv()

# En tu .env:
# LANGSMITH_TRACING=true
# LANGSMITH_API_KEY=lsv2_pt_...
# LANGSMITH_PROJECT=research-assistant

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke("¿Qué es tracing en AI?")
print(response.content[:100])
# Output esperado:
# Tracing en AI es el proceso de registrar y monitorear cada operación...

Eso es todo. Con LANGSMITH_TRACING=true, cada operación de LangChain y LangGraph se envía automáticamente a LangSmith. No necesitas decoradores, wrappers, ni cambios en tu código. Tus agentes existentes — todos, desde el Módulo 1 hasta el Módulo 11 — se trazan automáticamente.

Verificar que el tracing está activo

from dotenv import load_dotenv
load_dotenv()

import os

tracing = os.getenv("LANGSMITH_TRACING", "false")
api_key = os.getenv("LANGSMITH_API_KEY", "")
project = os.getenv("LANGSMITH_PROJECT", "default")

print(f"Tracing activo: {tracing}")
print(f"API key configurada: {'✅' if api_key else '❌'}")
print(f"Proyecto: {project}")
# Output esperado:
# Tracing activo: true
# API key configurada: ✅
# Proyecto: research-assistant

Qué contiene un trace

Un trace es el registro completo de una ejecución. Cada trace contiene una jerarquía de runs (operaciones individuales):

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

@tool
def search_web(query: str) -> str:
    """Busca información en la web."""
    return f"Resultados para '{query}': AI safety es un campo activo de investigación con 500+ papers en 2025."

@tool
def calculate(expression: str) -> str:
    """Evalúa una expresión matemática."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

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

result = agent.invoke({"messages": [{"role": "user", "content": "¿Cuántos papers de AI safety se publicaron en 2025? Multiplica eso por 12."}]})
print(result["messages"][-1].content)
# Output esperado:
# Según la búsqueda, se publicaron más de 500 papers de AI safety en 2025. 500 × 12 = 6000.

En LangSmith, este trace muestra:

Trace: "¿Cuántos papers de AI safety..."
│
├─ [LLM] gpt-4.1-mini — decide usar search_web
│   Input: system prompt + user message
│   Output: tool_call(search_web, "AI safety papers 2025")
│   Tokens: 180 input, 45 output
│   Latencia: 0.8s
│
├─ [Tool] search_web("AI safety papers 2025")
│   Output: "Resultados para 'AI safety papers 2025': ..."
│   Latencia: 0.01s
│
├─ [LLM] gpt-4.1-mini — decide usar calculate
│   Input: historial + tool result
│   Output: tool_call(calculate, "500 * 12")
│   Tokens: 290 input, 30 output
│   Latencia: 0.6s
│
├─ [Tool] calculate("500 * 12")
│   Output: "6000"
│   Latencia: 0.001s
│
└─ [LLM] gpt-4.1-mini — genera respuesta final
    Input: historial completo
    Output: "Según la búsqueda... 500 × 12 = 6000."
    Tokens: 350 input, 40 output
    Latencia: 0.5s

Total: 3 LLM calls, 2 tool calls, 820 input tokens, 115 output tokens, 1.9s

Cada run tiene: input exacto, output exacto, tokens usados, latencia, y su posición en la jerarquía.

Leer traces: la habilidad real

Configurar tracing toma 30 segundos. Leer traces es la habilidad que separa a un developer de un AI engineer de producción.

Habilidad 1: Identificar el bottleneck

La operación más lenta domina la latencia total. En un trace con 5 operaciones, si una toma 3 segundos y las demás 0.5 cada una, optimizar las cuatro rápidas no tiene impacto. Tienes que encontrar y optimizar la lenta.

from dotenv import load_dotenv
load_dotenv()

import time
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    search_results: str
    analysis: str
    report: str

def search(state: State) -> dict:
    time.sleep(2.0)
    return {"search_results": f"5 fuentes encontradas para '{state['query']}'"}

def analyze(state: State) -> dict:
    time.sleep(0.3)
    return {"analysis": f"Análisis de: {state['search_results'][:40]}"}

def write_report(state: State) -> dict:
    time.sleep(0.5)
    return {"report": f"Reporte: {state['analysis']}"}

builder = StateGraph(State)
builder.add_node("search", search)
builder.add_node("analyze", analyze)
builder.add_node("report", write_report)
builder.add_edge(START, "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "report")
builder.add_edge("report", END)

graph = builder.compile()

start = time.time()
result = graph.invoke({"query": "AI in healthcare", "search_results": "", "analysis": "", "report": ""})
elapsed = time.time() - start

print(f"Resultado: {result['report']}")
print(f"Tiempo total: {elapsed:.1f}s")
print(f"\nEn LangSmith verás:")
print(f"  search:  2.0s (71% del total) ← BOTTLENECK")
print(f"  analyze: 0.3s (11%)")
print(f"  report:  0.5s (18%)")
print(f"\n→ Optimizar 'search' tiene 4x más impacto que optimizar 'analyze'.")
# Output esperado:
# Resultado: Reporte: Análisis de: 5 fuentes encontradas para 'AI in hea
# Tiempo total: 2.8s
#
# En LangSmith verás:
#   search:  2.0s (71% del total) ← BOTTLENECK
#   analyze: 0.3s (11%)
#   report:  0.5s (18%)
#
# → Optimizar 'search' tiene 4x más impacto que optimizar 'analyze'.

En el dashboard de LangSmith, las operaciones se muestran como barras en una línea temporal. La barra más larga es tu bottleneck. No adivinas — lo ves.

Habilidad 2: Ver el prompt exacto

Cuando el modelo produce una respuesta inesperada, lo primero que necesitas saber es: ¿qué prompt recibió exactamente? No el prompt que tú escribiste — el prompt completo que se envió al modelo, incluyendo system prompt, historial de mensajes, tool results, y metadata.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

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

messages = [
    SystemMessage(content="Eres un analista financiero. Responde SOLO con datos numéricos y porcentajes. No uses texto narrativo."),
    HumanMessage(content="¿Cuál es el impacto de AI en el sector bancario?"),
]

response = model.invoke(messages)
print(f"Respuesta: {response.content[:200]}")
print(f"\nEn LangSmith verás el input EXACTO:")
print(f"  system: 'Eres un analista financiero. Responde SOLO con datos...'")
print(f"  human: '¿Cuál es el impacto de AI en el sector bancario?'")
print(f"\n→ Si la respuesta es narrativa, el prompt es claro pero el modelo lo ignora.")
print(f"→ Si la respuesta es correcta, el prompt está funcionando.")
# Output esperado:
# Respuesta: • Reducción de costos operativos: 20-30%...
#
# En LangSmith verás el input EXACTO:
#   system: 'Eres un analista financiero. Responde SOLO con datos...'
#   human: '¿Cuál es el impacto de AI en el sector bancario?'

En LangSmith, haces click en el run del modelo y ves: input (mensajes exactos), output (respuesta exacta), model parameters (temperature, model name), y token counts. El prompt debugging se vuelve trivial.

Habilidad 3: Detectar tool calls redundantes

Un agente puede llamar a la misma tool dos veces con el mismo query, o llamar a una tool que no necesita. Cada tool call redundante cuesta tiempo y tokens.

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

call_log = []

@tool
def search_database(query: str) -> str:
    """Busca en la base de datos de investigación."""
    call_log.append(query)
    return f"3 resultados encontrados para '{query}'"

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

result = agent.invoke({
    "messages": [{"role": "user", "content": "Busca papers sobre transformers y también busca papers sobre attention mechanisms"}]
})
print(f"Respuesta: {result['messages'][-1].content[:150]}")
print(f"\nTool calls realizados: {len(call_log)}")
for i, q in enumerate(call_log):
    print(f"  Call {i+1}: search_database('{q}')")
print(f"\n→ En LangSmith, verás cada tool call en la línea temporal.")
print(f"→ Si 'transformers' y 'attention mechanisms' producen los mismos resultados,")
print(f"   una de las dos llamadas es redundante.")
# Output esperado:
# Respuesta: Encontré resultados sobre ambos temas...
#
# Tool calls realizados: 2
#   Call 1: search_database('transformers')
#   Call 2: search_database('attention mechanisms')
#
# → En LangSmith, verás cada tool call en la línea temporal.
# → Si 'transformers' y 'attention mechanisms' producen los mismos resultados,
#    una de las dos llamadas es redundante.

Habilidad 4: Encontrar decisiones incorrectas

En un agente con routing condicional, el modelo decide qué camino tomar. Si toma el camino incorrecto, el trace te muestra exactamente qué input recibió y qué decisión tomó:

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

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

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

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

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

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

builder = StateGraph(State)
builder.add_node("router", router)
builder.add_node("technical", technical_handler)
builder.add_node("business", business_handler)

builder.add_edge(START, "router")
builder.add_conditional_edges("router", route_decision, {
    "technical": "technical",
    "business": "business",
})
builder.add_edge("technical", END)
builder.add_edge("business", END)

graph = builder.compile()

result = graph.invoke({"query": "¿Cuánto cuesta implementar RAG en producción?", "route": "", "result": ""})
print(f"Ruta elegida: {result['route']}")
print(f"Resultado: {result['result']}")
print(f"\nEn LangSmith verás:")
print(f"  1. El prompt exacto que recibió el router")
print(f"  2. La respuesta del modelo ('{result['route']}')")
print(f"  3. Qué nodo se ejecutó como consecuencia")
print(f"\n→ Si la ruta es incorrecta, ves exactamente por qué.")
# Output esperado:
# Ruta elegida: business
# Resultado: [BUSINESS] Análisis de negocio de: ¿Cuánto cuesta implementar RAG en producción?
#
# En LangSmith verás:
#   1. El prompt exacto que recibió el router
#   2. La respuesta del modelo ('business')
#   3. Qué nodo se ejecutó como consecuencia

Jerarquía de traces: run → child runs

Un trace tiene estructura jerárquica. El run principal (la invocación del grafo) contiene child runs (nodos), que a su vez contienen child runs (llamadas al modelo, tool calls):

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

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima de una ciudad."""
    return f"En {city}: 22°C, parcialmente nublado"

@tool
def get_population(city: str) -> str:
    """Obtiene la población de una ciudad."""
    return f"{city}: 8.3 millones de habitantes"

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

result = agent.invoke({
    "messages": [{"role": "user", "content": "Dame el clima y la población de Ciudad de México"}]
})
print(result["messages"][-1].content[:200])
print(f"\nJerarquía del trace en LangSmith:")
print(f"  Run principal: agent.invoke()")
print(f"    ├─ LLM call: decide usar tools")
print(f"    ├─ Tool: get_weather('Ciudad de México')")
print(f"    ├─ Tool: get_population('Ciudad de México')")
print(f"    └─ LLM call: genera respuesta final con ambos resultados")
# Output esperado:
# En Ciudad de México el clima es de 22°C, parcialmente nublado. La población es de 8.3 millones...
#
# Jerarquía del trace en LangSmith:
#   Run principal: agent.invoke()
#     ├─ LLM call: decide usar tools
#     ├─ Tool: get_weather('Ciudad de México')
#     ├─ Tool: get_population('Ciudad de México')
#     └─ LLM call: genera respuesta final con ambos resultados

En el dashboard, puedes expandir/contraer niveles de la jerarquía. Para debugging rápido, miras el nivel superior. Para debugging profundo, expandas hasta ver el input/output de cada operación individual.

Custom metadata: filtrar y organizar traces

En producción, generas miles de traces. Sin metadata, encontrar un trace específico es buscar una aguja en un pajar. LangSmith permite agregar metadata custom para filtrar:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

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

config = RunnableConfig(
    tags=["research", "production", "v7"],
    metadata={
        "user_id": "user_abc123",
        "session_id": "session_789",
        "agent_version": "7.0.1",
        "environment": "production",
    },
    run_name="Research Query - AI Safety",
)

response = model.invoke(
    "¿Cuáles son los riesgos principales de AI en producción?",
    config=config,
)
print(f"Respuesta: {response.content[:120]}...")
print(f"\nEn LangSmith puedes filtrar por:")
print(f"  - Tags: 'research', 'production', 'v7'")
print(f"  - Metadata: user_id='user_abc123'")
print(f"  - Run name: 'Research Query - AI Safety'")
# Output esperado:
# Respuesta: Los riesgos principales de AI en producción incluyen: hallucinations, costos impredecibles...
#
# En LangSmith puedes filtrar por:
#   - Tags: 'research', 'production', 'v7'
#   - Metadata: user_id='user_abc123'
#   - Run name: 'Research Query - AI Safety'

Metadata útil para producción

Tracing en LangGraph: visualizar la estructura del grafo

Cuando trazas un LangGraph, LangSmith muestra no solo las operaciones individuales sino la estructura del grafo: qué nodos se ejecutaron, en qué orden, y qué edges se recorrieron:

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

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: 3 artículos sobre {state['topic']}"]}

def search_papers(state: ResearchState) -> dict:
    return {"sources": [f"Papers: 2 papers sobre {state['topic']}"]}

def analyze(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analiza brevemente (1 oración): {', '.join(state['sources'])}"
    )
    return {"analysis": response.content}

def write_report(state: ResearchState) -> dict:
    return {"report": f"Reporte sobre '{state['topic']}': {state['analysis']}"}

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

result = graph.invoke({"topic": "LLM observability", "sources": [], "analysis": "", "report": ""})
print(f"Reporte: {result['report'][:150]}")
print(f"Fuentes: {len(result['sources'])}")
print(f"\nEn LangSmith verás:")
print(f"  START → [search_web + search_papers] (paralelo) → analyze → report → END")
print(f"  Los nodos paralelos se muestran como barras simultáneas en la timeline")
# Output esperado:
# Reporte: Reporte sobre 'LLM observability': Las fuentes indican que...
# Fuentes: 2
#
# En LangSmith verás:
#   START → [search_web + search_papers] (paralelo) → analyze → report → END
#   Los nodos paralelos se muestran como barras simultáneas en la timeline

El trace de un LangGraph muestra la ejecución como un árbol jerárquico donde el run principal contiene child runs por cada nodo. Si hay nodos paralelos (como search_web y search_papers), LangSmith los muestra simultáneamente en la línea temporal — puedes ver que se ejecutaron al mismo tiempo.

Filtrar traces en el dashboard

En producción, generas cientos de traces por día. LangSmith ofrece filtros para encontrar lo que necesitas:

Por proyecto

Cada proyecto en LangSmith agrupa traces. Organiza por entorno o aplicación:

# .env para desarrollo
LANGSMITH_PROJECT=research-assistant-dev

# .env para producción
LANGSMITH_PROJECT=research-assistant-prod

Por estado (éxito/error)

Filtra traces que terminaron con error para debugging:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

runs = list(client.list_runs(
    project_name="research-assistant-prod",
    is_root=True,
    filter='eq(status, "error")',
    limit=5,
))

print(f"Últimos {len(runs)} traces con error:")
for run in runs:
    print(f"  - {run.name}: {run.error[:80] if run.error else 'Sin mensaje'}")
    print(f"    ID: {run.id}")
    print(f"    Fecha: {run.start_time}")
# Output esperado:
# Últimos 3 traces con error:
#   - Research Query: RateLimitError: Rate limit exceeded...
#     ID: abc123-def456-...
#     Fecha: 2025-12-15 14:32:00
#   ...

Por tags

Filtra por tags que agregaste en la metadata:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client

client = Client()

runs = list(client.list_runs(
    project_name="research-assistant-prod",
    is_root=True,
    filter='has(tags, "research")',
    limit=5,
))

print(f"Últimos {len(runs)} traces con tag 'research':")
for run in runs:
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    print(f"  - {run.name}: {latency:.1f}s, {run.total_tokens or 0} tokens")
# Output esperado:
# Últimos 5 traces con tag 'research':
#   - Research Query - AI Safety: 3.2s, 1450 tokens
#   - Research Query - RAG: 2.8s, 1200 tokens
#   ...

Por fecha

Filtra por rango de fechas para investigar incidentes:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

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

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

print(f"Traces en las últimas 24 horas: {len(runs)}")
if runs:
    latencies = [(r.end_time - r.start_time).total_seconds() for r in runs if r.end_time]
    if latencies:
        print(f"  Latencia promedio: {sum(latencies)/len(latencies):.1f}s")
        print(f"  Latencia máxima: {max(latencies):.1f}s")
        print(f"  Latencia mínima: {min(latencies):.1f}s")
# Output esperado:
# Traces en las últimas 24 horas: 10
#   Latencia promedio: 3.4s
#   Latencia máxima: 8.1s
#   Latencia mínima: 1.2s

Naming conventions para traces

Un buen nombre de trace facilita el debugging. Un mal nombre lo hace imposible:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig

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

queries = [
    ("¿Qué es RAG?", "simple-qa"),
    ("Investiga AI en finanzas, analiza tendencias, genera reporte ejecutivo", "full-research"),
    ("Resume este artículo en 3 puntos", "summarization"),
]

for query, query_type in queries:
    config = RunnableConfig(
        run_name=f"[{query_type}] {query[:50]}",
        metadata={"query_type": query_type},
    )
    response = model.invoke(query, config=config)
    print(f"[{query_type}] → {response.content[:60]}...")
# Output esperado:
# [simple-qa] → RAG (Retrieval-Augmented Generation) es un patrón que...
# [full-research] → El impacto de la inteligencia artificial en el sector...
# [summarization] → Sin artículo proporcionado, no puedo resumir...

En el dashboard, estos traces aparecen como:

[simple-qa] ¿Qué es RAG?
[full-research] Investiga AI en finanzas, analiza tendencias...
[summarization] Resume este artículo en 3 puntos

Mucho más útil que tres traces llamados "ChatOpenAI".

Tracing programático: acceder a runs desde código

Puedes acceder a los traces programáticamente usando el SDK de LangSmith:

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
import time

client = Client()
model = init_chat_model("openai:gpt-4.1-mini")

config = RunnableConfig(
    run_name="Trace Programático Demo",
    metadata={"demo": "true"},
    tags=["demo"],
)

response = model.invoke("¿Qué es LangSmith? Responde en una oración.", config=config)
print(f"Respuesta: {response.content}")

time.sleep(3)

runs = list(client.list_runs(
    project_name="research-assistant",
    filter='has(tags, "demo")',
    limit=1,
))

if runs:
    run = runs[0]
    print(f"\nRun encontrado:")
    print(f"  Nombre: {run.name}")
    print(f"  Status: {run.status}")
    print(f"  Tokens input: {run.prompt_tokens}")
    print(f"  Tokens output: {run.completion_tokens}")
    print(f"  Total tokens: {run.total_tokens}")
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    print(f"  Latencia: {latency:.2f}s")
# Output esperado:
# Respuesta: LangSmith es una plataforma de observabilidad para...
#
# Run encontrado:
#   Nombre: Trace Programático Demo
#   Status: success
#   Tokens input: 28
#   Tokens output: 25
#   Total tokens: 53
#   Latencia: 0.85s

El acceso programático es útil para: scripts de monitoreo, reportes de costos automatizados, alertas custom, y integración con dashboards externos.

Troubleshooting

Problema 1: "Los traces no aparecen en LangSmith"

Síntoma: Ejecutas código con LangChain pero no ves traces en el dashboard.

Causa: LANGSMITH_TRACING no está configurado, o la API key es incorrecta.

Solución:

import os
print(f"LANGSMITH_TRACING: {os.getenv('LANGSMITH_TRACING')}")
print(f"LANGSMITH_API_KEY: {'set' if os.getenv('LANGSMITH_API_KEY') else 'NOT SET'}")
print(f"LANGSMITH_PROJECT: {os.getenv('LANGSMITH_PROJECT', 'default')}")

Verifica que LANGSMITH_TRACING=true (con minúsculas), que la API key es válida, y que load_dotenv() se ejecutó antes de importar LangChain.

Problema 2: "Los traces tardan en aparecer"

Síntoma: El trace no aparece inmediatamente en el dashboard.

Causa: LangSmith procesa traces de forma asíncrona. Hay un delay de 1-5 segundos.

Solución: Espera unos segundos y recarga el dashboard. Para acceso programático, agrega time.sleep(3) antes de buscar el trace.

Problema 3: "Los traces no tienen token counts"

Síntoma: El trace aparece pero total_tokens es None.

Causa: El proveedor no retorna usage info, o el modelo no soporta token counting.

Solución: Verifica que usas un modelo que reporta tokens (OpenAI, Anthropic). Algunos modelos locales o wrappers custom no incluyen esta información.

Problema 4: "Tengo demasiados traces y no encuentro el que busco"

Síntoma: El dashboard muestra miles de traces.

Causa: No estás usando metadata ni tags para organizar.

Solución: Agrega tags y metadata en RunnableConfig y usa los filtros del dashboard. Organiza por proyecto (LANGSMITH_PROJECT) separando entornos.

Problema 5: "El tracing hace mi agente más lento"

Síntoma: Notas latencia adicional cuando el tracing está activo.

Causa: El envío de traces a LangSmith agrega overhead mínimo (el envío es asíncrono), pero en entornos con red lenta puede ser perceptible.

Solución: El envío es asíncrono y no debería afectar la latencia percibida. Si lo hace, verifica tu conexión de red. En tests de performance, puedes desactivar el tracing temporalmente con LANGSMITH_TRACING=false.

Ejercicios

Ejercicio 1: Primer trace con metadata (Fácil)

Crea un script que invoque un modelo con RunnableConfig que incluya: un run_name descriptivo, un tag "ejercicio", y metadata con tu nombre y la fecha. Ejecuta y verifica que el trace aparece en LangSmith con la metadata correcta.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from datetime import date

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

config = RunnableConfig(
    run_name="Ejercicio 1 - Primer Trace",
    tags=["ejercicio", "modulo-12"],
    metadata={
        "author": "estudiante",
        "date": str(date.today()),
        "exercise": "01-primer-trace",
    },
)

response = model.invoke(
    "¿Qué es observabilidad en sistemas de AI? Responde en 2 oraciones.",
    config=config,
)
print(f"Respuesta: {response.content}")
print(f"\n→ Abre LangSmith, filtra por tag 'ejercicio', y verifica:")
print(f"  - Run name: 'Ejercicio 1 - Primer Trace'")
print(f"  - Metadata: author='estudiante', date='{date.today()}'")
# Output esperado:
# Respuesta: La observabilidad en sistemas de AI es la capacidad de monitorear...
#
# → Abre LangSmith, filtra por tag 'ejercicio', y verifica:
#   - Run name: 'Ejercicio 1 - Primer Trace'
#   - Metadata: author='estudiante', date='2025-12-15'

Ejercicio 2: Trazar un grafo multi-nodo (Fácil)

Crea un StateGraph con 3 nodos secuenciales (fetch → process → format). Ejecútalo con tracing activo. En LangSmith, identifica: cuántos child runs tiene el trace, cuál nodo fue el más lento, y cuántos tokens totales se usaron.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    input: str
    fetched: str
    processed: str
    formatted: str

def fetch(state: State) -> dict:
    time.sleep(0.5)
    return {"fetched": f"Datos sobre '{state['input']}'"}

def process(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Analiza en 1 oración: {state['fetched']}")
    return {"processed": response.content}

def format_output(state: State) -> dict:
    return {"formatted": f"📊 RESULTADO: {state['processed']}"}

builder = StateGraph(State)
builder.add_node("fetch", fetch)
builder.add_node("process", process)
builder.add_node("format", format_output)
builder.add_edge(START, "fetch")
builder.add_edge("fetch", "process")
builder.add_edge("process", "format")
builder.add_edge("format", END)

graph = builder.compile()

start = time.time()
result = graph.invoke({"input": "AI observability", "fetched": "", "processed": "", "formatted": ""})
elapsed = time.time() - start

print(f"Resultado: {result['formatted']}")
print(f"Tiempo total: {elapsed:.1f}s")
print(f"\n→ En LangSmith verás 3 child runs (fetch, process, format)")
print(f"→ 'process' debería ser el más lento (incluye LLM call)")
print(f"→ Solo 'process' consume tokens (fetch y format son locales)")
# Output esperado:
# Resultado: 📊 RESULTADO: AI observability involves monitoring and...
# Tiempo total: 1.5s

Ejercicio 3: Identificar bottleneck con tiempos (Medio)

Crea un grafo con 4 nodos donde cada uno simula una latencia diferente (0.1s, 2.0s, 0.3s, 0.5s). Ejecútalo, luego usa el SDK de LangSmith para obtener el trace y calcular programáticamente qué nodo consumió el mayor porcentaje del tiempo total.

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_core.runnables import RunnableConfig

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

def fast_node(state: State) -> dict:
    time.sleep(0.1)
    return {"data": "fast", "log": ["fast:0.1s"]}

def slow_node(state: State) -> dict:
    time.sleep(2.0)
    return {"data": "slow", "log": ["slow:2.0s"]}

def medium_node(state: State) -> dict:
    time.sleep(0.3)
    return {"data": "medium", "log": ["medium:0.3s"]}

def final_node(state: State) -> dict:
    time.sleep(0.5)
    return {"data": "final", "log": ["final:0.5s"]}

builder = StateGraph(State)
builder.add_node("fast", fast_node)
builder.add_node("slow", slow_node)
builder.add_node("medium", medium_node)
builder.add_node("final", final_node)

builder.add_edge(START, "fast")
builder.add_edge("fast", "slow")
builder.add_edge("slow", "medium")
builder.add_edge("medium", "final")
builder.add_edge("final", END)

graph = builder.compile()

config = RunnableConfig(
    run_name="Bottleneck Analysis",
    tags=["bottleneck-exercise"],
)

start = time.time()
result = graph.invoke({"data": "", "log": []}, config=config)
total_elapsed = time.time() - start

print(f"Log: {result['log']}")
print(f"Tiempo total: {total_elapsed:.1f}s")

expected_times = {"fast": 0.1, "slow": 2.0, "medium": 0.3, "final": 0.5}
total_expected = sum(expected_times.values())

print(f"\nAnálisis de bottleneck:")
for name, t in sorted(expected_times.items(), key=lambda x: -x[1]):
    pct = (t / total_expected) * 100
    bar = "█" * int(pct / 2)
    print(f"  {name:>8}: {t:.1f}s ({pct:.0f}%) {bar}")

print(f"\n→ 'slow' consume {expected_times['slow']/total_expected*100:.0f}% del tiempo total")
print(f"→ Optimizar 'slow' tiene el mayor impacto")
# Output esperado:
# Log: ['fast:0.1s', 'slow:2.0s', 'medium:0.3s', 'final:0.5s']
# Tiempo total: 2.9s
#
# Análisis de bottleneck:
#      slow: 2.0s (69%) ██████████████████████████████████
#     final: 0.5s (17%) ████████
#    medium: 0.3s (10%) █████
#      fast: 0.1s (3%) █
#
# → 'slow' consume 69% del tiempo total
# → Optimizar 'slow' tiene el mayor impacto

Ejercicio 4: Detectar tool calls redundantes (Medio)

Crea un agente con una tool search que loggea cada llamada. Dale un prompt que podría causar llamadas redundantes. Después de ejecutar, analiza el log para identificar si hubo queries duplicados o muy similares. Imprime un reporte.

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

call_log = []

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    call_log.append(query)
    return f"Resultados para '{query}': información relevante encontrada."

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

call_log = []
result = agent.invoke({
    "messages": [{"role": "user", "content": (
        "Necesito información sobre machine learning. "
        "Busca sobre machine learning applications. "
        "También busca sobre ML use cases."
    )}]
})

print(f"Respuesta: {result['messages'][-1].content[:120]}...")
print(f"\n=== REPORTE DE TOOL CALLS ===")
print(f"Total calls: {len(call_log)}")
for i, q in enumerate(call_log):
    print(f"  {i+1}. search('{q}')")

if len(call_log) > 1:
    print(f"\n=== ANÁLISIS DE REDUNDANCIA ===")
    from difflib import SequenceMatcher
    for i in range(len(call_log)):
        for j in range(i+1, len(call_log)):
            similarity = SequenceMatcher(None, call_log[i].lower(), call_log[j].lower()).ratio()
            status = "⚠️ POSIBLE REDUNDANCIA" if similarity > 0.5 else "✅ Diferentes"
            print(f"  '{call_log[i]}' vs '{call_log[j]}': {similarity:.0%} similar → {status}")
# Output esperado:
# Respuesta: Aquí está la información sobre machine learning...
#
# === REPORTE DE TOOL CALLS ===
# Total calls: 3
#   1. search('machine learning')
#   2. search('machine learning applications')
#   3. search('ML use cases')
#
# === ANÁLISIS DE REDUNDANCIA ===
#   'machine learning' vs 'machine learning applications': 72% similar → ⚠️ POSIBLE REDUNDANCIA
#   'machine learning' vs 'ML use cases': 25% similar → ✅ Diferentes
#   'machine learning applications' vs 'ML use cases': 28% similar → ✅ Diferentes

Ejercicio 5: Monitoreo de latencia con el SDK (Medio)

Ejecuta 5 invocaciones de un modelo con diferentes longitudes de input. Luego usa el SDK de LangSmith para obtener los traces y calcular: latencia promedio, latencia máxima, y correlación entre longitud de input y latencia.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langsmith import Client

model = init_chat_model("openai:gpt-4.1-mini")
client = Client()

inputs = [
    "¿Qué es AI?",
    "Explica el concepto de machine learning y sus principales tipos en 2 oraciones.",
    "Describe en detalle cómo funciona el mecanismo de attention en transformers, incluyendo query, key, value, y multi-head attention. Sé conciso.",
    "Compara y contrasta 5 arquitecturas de modelos de lenguaje: GPT, BERT, T5, LLaMA, y Mixtral. Para cada uno, menciona: año de lanzamiento, tamaño, y caso de uso principal.",
    "Escribe un análisis completo sobre el estado del arte en Retrieval-Augmented Generation (RAG) en 2025, cubriendo: chunking strategies, embedding models, vector databases, hybrid search, reranking, y evaluation metrics. Incluye ventajas y desventajas de cada approach.",
]

latencies = []
for i, input_text in enumerate(inputs):
    config = RunnableConfig(
        run_name=f"Latency Test {i+1}",
        tags=["latency-test"],
        metadata={"input_length": len(input_text), "test_index": i},
    )
    start = time.time()
    model.invoke(input_text, config=config)
    elapsed = time.time() - start
    latencies.append({"index": i+1, "input_len": len(input_text), "latency": elapsed})
    print(f"Test {i+1}: {len(input_text):>3} chars → {elapsed:.2f}s")

print(f"\n=== ANÁLISIS DE LATENCIA ===")
lats = [l["latency"] for l in latencies]
print(f"  Promedio: {sum(lats)/len(lats):.2f}s")
print(f"  Máxima:   {max(lats):.2f}s")
print(f"  Mínima:   {min(lats):.2f}s")

print(f"\n→ En LangSmith, filtra por tag 'latency-test' para ver los 5 traces")
print(f"→ Compara los token counts de cada uno con su latencia")
# Output esperado:
# Test 1:  12 chars → 0.65s
# Test 2:  80 chars → 0.82s
# Test 3: 170 chars → 1.10s
# Test 4: 220 chars → 1.45s
# Test 5: 350 chars → 2.30s
#
# === ANÁLISIS DE LATENCIA ===
#   Promedio: 1.26s
#   Máxima:   2.30s
#   Mínima:   0.65s

Ejercicio 6: Dashboard de traces con el SDK (Avanzado)

Escribe un script que use el SDK de LangSmith para generar un reporte textual de los últimos 10 traces de tu proyecto. Para cada trace, muestra: nombre, status, latencia, tokens totales, y si tiene errores. Al final, calcula estadísticas agregadas.

from dotenv import load_dotenv
load_dotenv()

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

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

print(f"{'='*70}")
print(f" REPORTE DE TRACES — {project_name}")
print(f" Generado: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f"{'='*70}\n")

total_tokens = 0
total_latency = 0
errors = 0

for i, run in enumerate(runs):
    latency = (run.end_time - run.start_time).total_seconds() if run.end_time else 0
    tokens = run.total_tokens or 0
    status = "✅" if run.status == "success" else "❌"

    print(f"{i+1}. {status} {run.name or 'Sin nombre'}")
    print(f"   Status: {run.status} | Latencia: {latency:.1f}s | Tokens: {tokens}")
    if run.error:
        print(f"   Error: {run.error[:80]}")
        errors += 1

    total_tokens += tokens
    total_latency += latency

print(f"\n{'='*70}")
print(f" ESTADÍSTICAS AGREGADAS")
print(f"{'='*70}")
if runs:
    print(f"  Total traces: {len(runs)}")
    print(f"  Exitosos: {len(runs) - errors} | Errores: {errors}")
    print(f"  Latencia promedio: {total_latency/len(runs):.1f}s")
    print(f"  Tokens totales: {total_tokens:,}")
    print(f"  Tokens promedio: {total_tokens//len(runs) if runs else 0:,}")
    print(f"  Tasa de error: {errors/len(runs)*100:.0f}%")
# Output esperado:
# ======================================================================
#  REPORTE DE TRACES — research-assistant
#  Generado: 2025-12-15 15:30
# ======================================================================
#
# 1. ✅ Research Query - AI Safety
#    Status: success | Latencia: 3.2s | Tokens: 1450
# 2. ✅ Ejercicio 1 - Primer Trace
#    Status: success | Latencia: 0.9s | Tokens: 53
# ...

Resumen

En esta cápsula aprendiste:

• Configurar tracing es una línea: LANGSMITH_TRACING=true. Todas las operaciones de LangChain/LangGraph se trazan automáticamente sin cambiar código
• La habilidad real es leer traces, no configurarlos. Cuatro habilidades: identificar bottlenecks (la operación más lenta domina la latencia), ver el prompt exacto (debugging de prompt), detectar tool calls redundantes (optimización de costos), y encontrar decisiones incorrectas (debugging de routing)
• Un trace tiene jerarquía: run principal → child runs → nested operations. En LangGraph, los nodos del grafo aparecen como child runs con sus propios sub-runs (model calls, tool calls)
• Custom metadata es esencial en producción: tags, metadata, y run_name en RunnableConfig te permiten filtrar y organizar miles de traces. Sin metadata, encontrar un trace específico es imposible
• Tracing en LangGraph muestra la estructura del grafo: qué nodos se ejecutaron, en qué orden, cuáles fueron paralelos, y qué edges se recorrieron
• El SDK de LangSmith permite acceso programático a los traces: listar runs, filtrar por status/tags/fecha, y calcular estadísticas — base para scripts de monitoreo y alertas

Próxima cápsula: Debugging Visual de Agentes — cómo usar LangSmith para pasar de "agregar print y rezar" a "hacer click en el trace y VER qué pasó", con un workflow sistemático de debugging.

Recursos adicionales

1. LangSmith — Tracing Concepts — Conceptos fundamentales: runs, traces, projects
2. LangSmith — Annotate traces with metadata — Cómo agregar tags, metadata, y run names
3. LangSmith — Filter traces in the application — Filtros avanzados en el dashboard
4. LangSmith SDK Reference — Referencia del SDK de Python para acceso programático
5. LangGraph Tracing — Cómo funciona el tracing específicamente con LangGraph
6. OpenTelemetry for LLMs — LangChain Blog — Contexto sobre estándares de tracing en aplicaciones LLM

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