Trazado de Llamadas LLM y Pasos de Agentes

Intermedio⏱45 min📖5 min de lecturaLección 2 de 5

Trazado de Llamadas LLM y Pasos de Agentes

Aprende a crear traces y spans detallados para llamadas LLM, pasos de agentes e integraciones con frameworks usando LangFuse.

El trazado es el núcleo de la observabilidad de LangFuse. Cada llamada LLM, paso de recuperación, invocación de herramienta o decisión de agente puede capturarse como un span estructurado dentro de un trace. Esta lección muestra cómo construir árboles de trace ricos y anidados y cómo instrumentar pipelines de LangChain y LlamaIndex.

Creando Spans y Traces

Cada trace comienza con langfuse.trace(). Dentro de él, creas spans para cada paso lógico:

python

from langfuse import Langfuse

langfuse = Langfuse()

trace = langfuse.trace(
    name="qa-agent",
    input={"question": "¿Cuál es la capital de Francia?"},
    user_id="user_42",
    session_id="sess_001"
)

Los spans pueden anidarse arbitrariamente:

python

# Span raíz (ej.: "recuperar documentos")
recuperacion = trace.span(name="recuperar")

# Span hijo (ej.: "incrustar consulta")
incrustacion = recuperacion.span(name="incrustar-consulta")
incrustacion.end(
    input={"query": "capital de Francia"},
    output={"embedding_dim": 1536}
)

# Otro span hijo (ej.: "búsqueda vectorial")
busqueda = recuperacion.span(name="busqueda-vectorial")
busqueda.end(
    input={"top_k": 5},
    output={"results": ["doc1", "doc3", "doc7"]}
)

recuperacion.end()

ℹ️Nota

Un trace representa una solicitud completa de extremo a extremo (una consulta de usuario). Un span representa una sola operación dentro de esa solicitud. Múltiples spans forman una jerarquía de árbol. El span raíz de un trace es su primer span; todos los demás son hijos de algún span padre.

⚠️Advertencia

Siempre llama a .end() en un span cuando la operación termine. Los spans huérfanos (sin .end()) permanecerán "abiertos" en el panel de LangFuse y sesgarán las métricas de latencia. Usa el patrón context manager (with trace.span() as s:) para garantizar el cierre.

Jerarquía de Trace (Diagrama ASCII)

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Cada span puede contener sus propios input, output, metadata, usage y level (DEBUG, WARNING, ERROR).

Jerarquía de Span Anidado (Secuencia)

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Ciclo de Vida del Trace

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Agregando Metadatos y Puntuaciones

python

# Agregando metadatos a un span
span = trace.span(
    name="llm-call",
    metadata={
        "model": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 500
    }
)

# Agregando una puntuación después de que el span termine
trace.score(
    name="utilidad",
    value=0.85,
    comment="Buena respuesta, pero podría ser más corta"
)

# Tipos de puntuación: NUMERIC, BOOLEAN, CATEGORICAL
trace.score(name="toxicidad", value=False, data_type="BOOLEAN")
trace.score(name="dificultad", value="medio", data_type="CATEGORICAL")

⚠️Advertencia

Las puntuaciones se adjuntan a un trace o span después del hecho. No bloquean la ejecución. Asegúrate de tener una referencia al objeto trace/span (o su ID) para puntuarlo posteriormente.

💡Consejo

Usa metadatos para etiquetar spans con contexto de negocio: environment, region, model_version, prompt_template_name, user_tier. Estos campos se convierten en dimensiones filtrables en los paneles. El etiquetado consistente en todos los spans permite potentes filtros cruzados.

Tipos de Datos de Puntuación

Tipo de Dato	Ejemplo Python	Visualización en Panel	Caso de Uso
NUMERIC	`value=0.85`	Histograma, avg/min/max	Corrección, utilidad, relevancia
BOOLEAN	`value=True`	Tasa de aprobación/rechazo, gráfico circular	Toxicidad, verificaciones de seguridad, guardrails
CATEGORICAL	`value="medio"`	Gráfico de barras, distribución	Dificultad, prioridad, clase de intención

Trazando Ejecuciones de LangChain

LangFuse proporciona un callback handler para LangChain que auto-instrumenta chains:

python

from langfuse.callback import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Crear el handler (uno por proyecto)
langfuse_handler = CallbackHandler()

prompt = ChatPromptTemplate.from_template("Cuenta un chiste corto sobre {tema}")
model = ChatOpenAI(model="gpt-4")
chain = prompt | model

# El handler se conecta automáticamente a cada paso
resultado = chain.invoke({"tema": "programación"}, config={"callbacks": [langfuse_handler]})

Cada paso de LangChain (template de prompt, llamada LLM, parser, recuperador) se convierte en un span separado dentro de un solo trace.

Avanzado: Agente LangChain con Herramientas

python

from langfuse.callback import CallbackHandler
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import tool
from langchain_openai import ChatOpenAI

langfuse_handler = CallbackHandler()

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 22°C in {city}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

llm = ChatOpenAI(model="gpt-4")
agent = create_openai_functions_agent(llm, [get_weather, calculate])
executor = AgentExecutor(agent=agent, tools=[get_weather, calculate])

result = executor.invoke(
    {"input": "What is the weather in Paris plus 5?"},
    config={"callbacks": [langfuse_handler]}
)

Cada invocación de herramienta aparece como un span hijo separado, y el bucle de razonamiento del agente crea un árbol de trace que muestra la ruta completa de decisión.

Trazando Pipelines de LlamaIndex

python

from langfuse.llama_index import LlamaIndexCallbackHandler
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# Inicializar handler
handler = LlamaIndexCallbackHandler()

documentos = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documentos)

query_engine = index.as_query_engine()
respuesta = query_engine.query("¿Qué es LangFuse?")

# Descargar traces
handler.flush()

Instrumentación Personalizada con Decoradores

Para máximo control, usa el decorador @observe():

python

from langfuse.decorators import observe

@observe()
def obtener_clima(ciudad: str) -> str:
    """Esta función se traza automáticamente."""
    respuesta = call_weather_api(ciudad)
    return respuesta

@observe(as_type="generation")
def llamar_llm(prompt: str, model: str = "gpt-4") -> str:
    """Marca este span como una 'generation' (llamada LLM)."""
    ...

⚠️Advertencia

El decorador @observe funciona con cualquier función de Python, no solo llamadas LLM. Usa el parámetro as_type para distinguir generaciones (llamadas LLM) de spans regulares.

Avanzado: Instrumentación Personalizada con Agrupación de Traces

Agrupa traces relacionados bajo una sola sesión para conversaciones completas de múltiples turnos:

python

# trace_grouping.py
from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe()
def process_message(session_id: str, message: str, turn_number: int) -> str:
    """Process a single message in a multi-turn conversation."""
    context = retrieve_context(message)
    response = generate_response(message, context)

    trace = langfuse.current_trace()
    if trace:
        trace.score(name="coherence", value=0.9, data_type="NUMERIC")
        trace.update(session_id=session_id)

    return response

@observe(as_type="generation")
def generate_response(message: str, context: str) -> str:
    """Call the LLM with context."""
    return "París es la capital de Francia."

session_1 = "sess_conversation_001"
for i, msg in enumerate(["¡Hola!", "¿Cuál es la capital de Francia?"]):
    process_message(session_1, msg, i + 1)

langfuse.flush()

💡Consejo

Al trazar bucles de agentes, establece session_id en cada trace para que el panel de LangFuse agrupe todos los turnos de una conversación. Luego puedes filtrar por sesión para reproducir toda la trayectoria del agente.

Comparación: Enfoques de Instrumentación

Enfoque	Esfuerzo	Granularidad	Alcance automático	Mejor para
Spans manuales	Alto	Control total	Manual	Pipelines personalizados, investigación
CallbackHandler LangChain	Bajo	Por paso de chain	Automático	Aplicaciones LangChain
CallbackHandler LlamaIndex	Bajo	Por paso de índice	Automático	Aplicaciones LlamaIndex
Decorador `@observe`	Medio	Por función	Envuelve la función	Cualquier código Python

Resumen de Tipos de Span

Tipo de Span	Convención de `name`	Metadatos Recomendados	Seguimiento de Uso
Generación LLM	`llm-call`, `openai-completion`, `anthropic-generate`	`model`, `temperature`, `max_tokens`, `provider`	`prompt_tokens`, `completion_tokens`, `total`
Recuperación	`vector-search`, `embed-query`, `bm25-search`	`top_k`, `index_name`, `embedding_model`	Generalmente ninguno
Ejecución de Herramienta	`get_weather`, `calculate`, `search_web`	`tool_name`, `tool_input`	Generalmente ninguno
Lógica / Enrutamiento	`classify-intent`, `guardrail-check`, `format-response`	`decision`, `confidence`	Generalmente ninguno
Manejador de Error	`error-handling`, `fallback`	`error_type`, `retry_count`	Generalmente ninguno