Realizar el seguimiento de un pipeline de blog de LangGraph con Langfuse: segmentos, puntuaciones y advertencias.
Configurar un pipeline multiagente de LangGraph con Langfuse: un rastro por artículo, segmentos anidados, advertencias, puntuaciones y una rutina de depuración para borradores deficientes.
Fuentes: Building Multi-Agent System From Scratch — Part 6: Observability and Debugging
Las medidas de control (verificaciones de fuentes, verificaciones de citas, límites en los intentos de reintentar y escalado a un humano) solo son útiles si se puede reconstruir una ejecución fallida posteriormente. Un pipeline de blog con múltiples agentes plantea muchas preguntas en ese momento: ¿Eran débiles las fuentes de investigación? ¿Ignoró el autor las advertencias sobre citas? ¿Tuvo razón el revisor al rechazarlo? ¿Se agotó el tiempo de búsqueda? ¿Cuánto costó la ejecución en llamadas al modelo y tokens?
Sin capacidad de observabilidad, el sistema es una caja negra rodeada de otras cajas negras. El rastreo de múltiples agentes es una parte extremadamente difícil del trabajo. Esta configuración añade Langfuse a un pipeline de blog de LangGraph, generando un registro por cada ejecución de artículo que incluye las transferencias de control, las llamadas a modelos y herramientas, los tiempos, los costos, los errores y los resultados de evaluación.
Decidir qué es un registro
Antes de cualquier código, elija la unidad que desea inspeccionar. En esta cadena de procesamiento, una traza equivale a un intento de crear un artículo. Comienza cuando se presenta un tema y termina cuando la cadena de procesamiento finaliza, hace una pausa a la espera de una intervención humana o falla. Dentro de ella se incluyen observaciones sobre cada tarea realizada:
blog-pipeline
├── research-agent
│ ├── search-web
│ └── research-model
├── writer-agent
│ └── writer-model
├── citation-check
│ └── citation-review-model
└── reviewer-agent
└── reviewer-model
Langfuse funciona con tres sustantivos. Un rastro corresponde a una ejecución completa desde el inicio hasta el final. Una observación es un paso individual dentro de ella, que puede ser un intervalo (lógica de la aplicación), una generación (una llamada a un LLM) o un evento (un marcador muy pequeño). Una sesión une varios rastros, como todos los intentos relacionados con el mismo tema del blog.
La jerarquía refleja cómo se realiza la depuración: se parte de la ejecución completa y luego se profundiza en el agente, herramienta o llamada al modelo específico que causó el problema.
Configuración de Langfuse
Cree un proyecto de Langfuse y proporcione sus credenciales a través de variables de entorno. La clave secreta nunca debe escribirse en los archivos fuente.
pip install -U langfuse
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_TRACING_ENVIRONMENT="development"
El paquete langfuse es el SDK de Python. Las dos claves autentican la aplicación frente al proyecto, y LANGFUSE_BASE_URL selecciona el entorno de despliegue. El ejemplo está orientado a la región cloud de la UE, así que cámbielo por otra región o una instancia autohospedada según sea necesario.
Una etiqueta de entorno permite separar los registros de desarrollo de los de producción. Nombres cortos y estables como development, staging y production funcionan bien. Sin ellos, los experimentos y las ejecuciones reales de artículos terminan en la misma vista, lo que complica el depuración.
Conexión del manejador de callbacks
Langfuse incluye un manejador de callbacks de LangChain. LangGraph se basa en la infraestructura de ejecución y callbacks de LangChain, por lo que el mismo manejador captura eventos relacionados con grafos, cadenas, herramientas y modelos:
from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler
langfuse = get_client()
langfuse_handler = CallbackHandler()
get_client() crea un cliente a partir de las variables de entorno. El manejador debe crearse una sola vez al iniciar el programa y reutilizarse en cada ejecución. Escucha los eventos de LangChain y los envía como observaciones, lo que elimina la necesidad de rastrear manualmente cada llamada al modelo.
A continuación interviene propagate_attributes. Aplica las mismas etiquetas a todas las observaciones dentro de una misma ejecución, lo que permite filtrar una llamada de búsqueda, una generación del escritor y una generación del revisor como partes de un mismo flujo de trabajo para artículos.
La invocación del grafo está envuelta en un span raíz, pasando el manejador a través de la configuración de LangGraph:
def run_blog_pipeline(topic: str, blog_id: str, user_id: str):
initial_state = {
"topic": topic,
"audience": "developers new to agent systems",
"research_brief": "",
"sources": [],
"open_questions": [],
"article_draft": "",
"review_feedback": "",
"citation_issues": [],
"approved": False,
"revision_count": 0,
"status": "researching",
}
with langfuse.start_as_current_observation(
as_type="span",
name="blog-pipeline",
input={"topic": topic, "audience": initial_state["audience"]},
) as pipeline_span:
with propagate_attributes(
trace_name="blog-pipeline",
session_id=blog_id,
user_id=user_id,
tags=["blog-pipeline", "langgraph"],
version="1.0.0",
metadata={"workflow": "research-write-review"},
):
trace_id = langfuse.get_current_trace_id()
result = graph.invoke(
initial_state,
config={"callbacks": [langfuse_handler]},
)
pipeline_span.update(
output={
"status": result["status"],
"approved": result["approved"],
"revision_count": result["revision_count"],
}
)
return result, trace_id
Varios detalles de ese contenedor son importantes. La llamada externa a start_as_current_observation() crea la observación raíz, y sus parámetros de entrada se mantienen deliberadamente compactos, solo el tema y el público objetivo. Eso es lo que se muestra de inmediato en la lista de trazas, por lo que un desglose del estado interno no debería estar allí.
propagate_attributes() asigna etiquetas a cada observación anidada. El blog_id funciona como ID de sesión, agrupando los intentos repetidos sobre un mismo artículo. Un valor version identifica el flujo de trabajo y la versión del mensaje detrás de un resultado, de modo que cuando cambia el comportamiento, las trazas pueden filtrarse por versión en lugar de hacerse suposiciones.
El manejador recorre la configuración regular de LangGraph y, a partir de allí, recopila automáticamente los detalles a nivel del framework: cada llamada a nodo, generación por LLM, llamada a herramienta, entrada, salida, latencia, cantidad de tokens y costo.
Al final, el span raíz se actualiza con tres resultados a nivel de pipeline: si la ejecución se completó, si el borrador fue aprobado y cuántas revisiones se utilizaron. El trace_id devuelto debe almacenarse en los registros de la aplicación o en la interfaz de usuario, lo que permite acceder directamente al seguimiento de Langfuse desde la ejecución de un artículo.
Spans a nivel empresarial dentro de los agentes
El seguimiento automático demuestra que se llamó a un modelo. No muestra qué pasó cada especialista al siguiente. Para ello, agregue un span ligero dentro de un nodo de agente y envíe el argumento config a las llamadas anidadas de LangChain o LangGraph para que el manejador mantenga el mismo contexto de seguimiento:
from langchain_core.runnables import RunnableConfig
def research_node(state: BlogState, config: RunnableConfig) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="research-agent",
input={"topic": state["topic"], "audience": state["audience"]},
) as span:
result = research_agent.invoke(
{"topic": state["topic"], "audience": state["audience"]},
config=config,
)
span.update(
output={
"source_count": len(result["sources"]),
"open_question_count": len(result["open_questions"]),
"brief": result["research_brief"],
}
)
return {
"research_brief": result["research_brief"],
"sources": result["sources"],
"open_questions": result["open_questions"],
"status": "writing",
}
El span complementa al manejador de llamadas de retorno y no lo reemplaza. Añade el significado que el framework no puede adivinar, es decir, qué significa “la investigación tuvo éxito” en este flujo de trabajo. Aquí se registra el conteo de fuentes, la cantidad de preguntas sin resolver y las instrucciones entregadas al escritor.
Al pasar config: RunnableConfig a research_agent.invoke(), también se transmiten los manejadores de llamadas de retorno y el contexto de seguimiento. Olvidarlo es un error clásico de observabilidad: se sigue el grafo de nivel superior, pero la llamada anidada aparece como un seguimiento no relacionado o desaparece.
Mantenga constantes los nombres de las observaciones: research-agent, writer-agent, citation-check. Un tema, el nombre de un modelo o un ID de usuario no debería incluirse en los nombres, ya que esa información pertenece a las entradas, atributos del modelo o metadatos. Los nombres constantes garantizan que los paneles de control y filtros sigan funcionando incluso con cambios en la implementación.
Avisos y errores
No todo resultado inesperado constituye una caída del sistema. Una fuente dudosa, un presupuesto de revisión agotado o una cita sin soporte también son importantes y deben aparecer como avisos.
def record_source_assessment(assessment: SourceAssessment) -> None:
if assessment.suspicious_content:
langfuse.update_current_span(
level="WARNING",
status_message="Untrusted source contained agent-directed instructions.",
)
def record_pipeline_failure(error: Exception) -> None:
langfuse.update_current_span(
level="ERROR",
status_message=f"Pipeline failed: {type(error).__name__}",
)
Langfuse admite niveles de registro para las observaciones. ADVERTENCIA se utiliza para ejecuciones que finalizaron pero que requieren revisión, mientras que ERROR está destinado a fallos reales. Los mensajes de estado deben describir el problema operativo sin incluir texto sensible o secretos en la telemetría.
El registrador de evaluación de fuentes debe formar parte del proceso de verificación de fuentes, y el registrador de fallos en un manejador de errores que pueda recuperarse o informar adecuadamente. Registrar cada excepción únicamente con el fin de anotarla es una mala práctica, ya que los errores inesperados deben llegar igualmente al sistema de monitoreo de errores y ser detectados claramente.
Evaluación del resultado
Una traza indica qué ocurrió, y una puntuación indica si el resultado fue satisfactorio. Una vez que finaliza una ejecución, se debe asignar una primera puntuación determinística basada en el estado del pipeline:
result, trace_id = run_blog_pipeline(
topic="How AI agents use tools",
blog_id="blog-ai-tools-001",
user_id="philip",
)
if trace_id:
langfuse.create_score(
trace_id=trace_id,
name="review_approved",
value=1 if result["approved"] else 0,
data_type="BOOLEAN",
comment=result["status"],
)
langfuse.create_score(
trace_id=trace_id,
name="revision_count",
value=float(result["revision_count"]),
data_type="NUMERIC",
)
La primera puntuación refleja el veredicto estructurado de aprobación o rechazo del revisor, mientras que la segunda mide cuántas revisiones requirió un borrador. Al acumularse con el tiempo, revelan algo más que simples anécdotas. ¿Qué modelo, versión de prompt, categoría temática o enfoque de obtención de información genera menos revisiones y más aprobaciones?
Ninguna de ellas constituye la verdad absoluta, ya que un revisor puede equivocarse. La puntuación humana, añadida después de leer el borrador final, puede realizarse más tarde. Lo importante es hacer que la calidad sea medible y compararla entre diferentes casos, en lugar de confiar en unos pocos ejemplos memorables.
Una rutina de depuración
Cuando un artículo resulte deficiente, abra su historial y haga estas preguntas en orden:
- ¿Siguió el gráfico la ruta esperada? Busque las etapas de Investigación, Redactor, Verificación de citas y Revisor, además de cualquier bucle de revisión.
Por eso es importante la estructura del seguimiento. Un árbol bien nombrado convierte una afirmación genérica como “el blog estaba equivocado” en un hallazgo preciso, como “el autor agregó una afirmación que faltaba en el resumen después de la segunda revisión”.
Evitar la salida de datos sensibles
Tracing puede registrar prompts, resultados de salida, resultados de herramientas y metadatos, lo que también significa que puede capturar información que no debería almacenarse. Las claves API, los encabezados de autenticación y la configuración secreta completa nunca deben llegar a Langfuse. Si los datos del usuario entran en la pipeline más tarde, revise qué se captura y configure el enmascaramiento antes de exportar los registros. Langfuse ofrece funciones de enmascaramiento que ocultan valores sensibles antes de que salgan de la aplicación.
La regla general: registre lo suficiente como para explicar una decisión, pero no más de lo que se pueda conservar de forma segura.
Qué obtiene la pipeline
Cada ejecución ahora deja un registro operativo:
- Cada intento de generar un artículo corresponde a un registro individual.
- Los agentes, modelos y herramientas aparecen como observaciones anidadas.
- Los nombres constantes, junto con los metadatos, la versión, la sesión y el entorno, facilitan el filtrado.
La depuración deja de ser una cuestión de desplazarse por los registros del terminal e adivinar, y se convierte en una inspección del momento exacto en que ocurrió el problema.