Evaluar agentes de LangGraph con datasets de Langfuse: casos dorados por agente, evaluadores en código, un juez calibrado y una puerta de regresión en CI

TL;DR

Un caso dorado es una entrada, una salida esperada y metadatos, y la salida esperada casi nunca es un texto: es qué herramienta tenía que llamarse, con qué argumentos, si había que escalar, si había que parar a pedir aprobación, qué veredicto tenía que dar el evaluador. Comparar textos generados con textos esperados es lo que menos dice y lo que más cuesta.

Los casos salen de tres sitios que ya existen: las trazas de producción, marcadas desde la aplicación con un pulgar; las decisiones de los encargados del caso 2, que son etiquetas gratis; y la muestra semanal de fichas que alguien de catálogo ya lee. create_dataset_item con source_trace_id convierte una traza en un caso con un clic y una línea.

El SDK de Langfuse 4.15 trae run_experiment: una función de tarea que ejecuta el agente por elemento, evaluadores por elemento que reciben entrada, salida y salida esperada y devuelven Evaluation, evaluadores de ejecución que reciben todos los resultados, y un ExperimentResult con las puntuaciones. Verificado: funciona sobre un dataset del servidor y sobre una lista local, y produce el resultado aunque el servidor no responda.

Los evaluadores son código para todo lo que se puede decidir con código: herramienta correcta, argumentos, escalado, cita de sección, idioma, promesa de plazo. Un modelo juez solo para lo que exige leer, y solo después de medir su acuerdo con las personas sobre la misma muestra. Un juez sin calibrar es un segundo prompt opinando sobre el primero.

run_experiment tiene max_concurrency=50 por defecto. Contra un vLLM con una clave de gateway limitada a 24, son 26 peticiones en 429 desde el primer segundo. Se fija al valor de la clave, como en el caso 3.

La puerta en CI es una línea: si la métrica que importa baja del umbral, RegressionError. La acción oficial es de GitHub; con Forgejo se hace lo mismo con un script de veinte líneas que lee el ExperimentResult y sale con código distinto de cero.

Estás aquí: lo que el guion no prueba

El artículo anterior probó los cuatro agentes de la serie de retail con un modelo guionizado. Esos tests dicen que el grafo hace lo que el código dice cuando el modelo decide lo que el guion decide. No dicen nada de lo que el modelo decide solo.

Eso es lo que aquí se mide. Y se mide en Langfuse porque es donde ya están las trazas, donde ya se agrupan por caso y thread_id, y donde el par operativo con LiteLLM ya deja el coste por generación.

La analogía: el cliente misterioso

Una cadena de tiendas no sabe si sus dependientes atienden bien por lo que dicen los dependientes. Manda a un cliente misterioso con un guion (pregunta esto, pide una devolución sin ticket, insiste) y una hoja de comprobación: ¿miró el stock?, ¿citó el procedimiento?, ¿llamó al encargado cuando tocaba?, ¿prometió algo que no podía? La hoja es objetiva en casi todo. Solo una casilla, “¿fue amable?”, la rellena una persona, y a esa persona se la entrena antes con casos en los que ya se sabe la respuesta.

El dataset es el guion del cliente misterioso. Los evaluadores en código son la hoja. El juez con modelo es la casilla de “amable”, y la calibración es el entrenamiento de quien la rellena.

Qué es un caso dorado en cada agente

La columna del medio es lo que no es texto libre. En el caso 1 no se espera “En T001 no queda. T002 tiene 14”, se espera {"tool": "stock_sku", "args": {"sku": "SKU-0912"}, "escala": false}. La redacción cambia con cada versión del modelo y con la temperatura; la decisión no debería.

Cuando sí hay texto esperado, como en el caso 4 con el veredicto, se compara lo estructurado: aprobada y el conjunto de criterios rechazados, no el detalle.

De dónde salen los casos

Trazas con pulgar

La aplicación de sala tiene dos botones debajo de cada respuesta. Cada pulsación es una puntuación sobre la traza:

langfuse.create_score(
    trace_id=trace_id,
    name="pulgar",
    value=1.0 if arriba else 0.0,
    data_type="BOOLEAN",
    comment=comentario_opcional,
)

Una vez a la semana, alguien filtra en Langfuse las trazas del caso tienda con pulgar = 0, lee la conversación, decide cuál era la respuesta correcta y crea el caso:

langfuse.create_dataset_item(
    dataset_name="tienda-dorado",
    input={"pregunta": "...", "tienda": "T001", "rol": "dependiente"},
    expected_output={"tool": "procedimiento", "seccion": "PROC-4.3", "escala": False},
    metadata={"origen": "pulgar_abajo", "semana": "2026-W38"},
    source_trace_id=trace_id,
)

source_trace_id enlaza el caso con la traza de la que salió. En la interfaz, desde el caso se abre la conversación original, y desde cada ejecución del experimento se ve la nueva. Es lo que hace que el dataset no sea una hoja de cálculo.

Los casos con pulgar arriba también entran, en menor número, para que el dataset no sea solo fallos.

Las decisiones de los encargados

El caso 2 tiene etiquetas gratis. Cada fila de decisiones_encargado es un caso: la entrada es la conversación hasta la interrupción, la salida esperada es la decisión. Si el encargado aprobó tal cual, el agente propuso bien. Si editó el importe, el agente propuso mal y la etiqueta dice cuánto. Si rechazó con motivo, el agente no debió proponer.

Y una etiqueta sobre la propia política: si when no interrumpió y el cliente reclamó después, ese caso entra con debia_interrumpir: true. Esa es la evaluación de la función when, que no tiene modelo y se ajusta con estos datos.

La muestra de catálogo

El caso 4 ya tiene una muestra semanal que lee una persona. Cada ficha leída, con el veredicto de la persona y sus criterios, es un caso del dataset fichas-dorado. Es el único dataset de los cuatro donde el juez con modelo tiene sentido, y también el único donde hay con qué calibrarlo.

Ejecutar un experimento

from langfuse import Langfuse
from langfuse.experiment import Evaluation

langfuse = Langfuse()
dataset = langfuse.get_dataset("tienda-dorado")

def tarea(*, item, **kwargs):
    entrada = item.input
    ctx = Contexto(tienda=entrada["tienda"], empleado="EVAL", rol=entrada["rol"])
    thread_id = f"eval:{item.id}:{uuid.uuid4()}"
    llamadas, escalo = [], False
    for update in agente.stream(
        {"messages": [{"role": "user", "content": entrada["pregunta"]}]},
        context=ctx,
        config={"configurable": {"thread_id": thread_id}, "max_concurrency": 4},
        stream_mode="updates",
    ):
        for nodo, cambio in update.items():
            if nodo == "tools":
                llamadas += [json.loads(m.content) | {"tool": m.name} for m in cambio["messages"]]
    respuesta = agente.get_state({"configurable": {"thread_id": thread_id}}).values["messages"][-1].content
    return {"llamadas": llamadas, "respuesta": respuesta}

resultado = dataset.run_experiment(
    name="tienda-prompt-v7",
    run_name=f"{modelo_nombre}-{fecha}",
    task=tarea,
    evaluators=[herramienta_correcta, escalado_correcto, cita_seccion, castellano, sin_plazos],
    run_evaluators=[tasa_escalado],
    max_concurrency=16,
    metadata={"modelo": modelo_nombre, "prompt": "v7"},
)
print(resultado.format())

Lo que importa de la tarea: cada elemento corre en un hilo nuevo con un identificador que no colisiona con producción, con el contexto del caso y no de un usuario real, y devuelve lo estructurado además del texto. El checkpointer del agente de evaluación es un InMemorySaver; no se quiere el dataset en la tabla checkpoints de producción.

run_experiment acepta data= con una lista local cuando no hay dataset en el servidor, y es lo que se usa en desarrollo: los mismos evaluadores sobre diez casos escritos a mano en un fichero. Verificado en el SDK 4.15.4 que el ExperimentResult se construye igual, con item_results, run_evaluations y las medias, y que si el servidor no responde el resultado se produce igual y solo fallan las exportaciones de trazas.

Con el dataset del servidor, cada elemento deja una traza enlazada a la ejecución, y la ejecución aparece en la pestaña de datasets con sus puntuaciones. Dos ejecuciones con run_name distinto sobre el mismo dataset son comparables lado a lado en la interfaz; es el uso principal.

Cuando el agente interrumpe

La tarea del caso 2 tiene que reanudar. El caso dorado lleva la decisión esperada, y la tarea la aplica:

def tarea_cliente(*, item, **kwargs):
    cfg = {"configurable": {"thread_id": f"eval:{item.id}:{uuid.uuid4()}"}}
    r = agente.invoke({"messages": [{"role": "user", "content": item.input["mensaje"]}]}, context=ctx_de(item), config=cfg)
    interrumpio = "__interrupt__" in r
    propuesta = r["__interrupt__"][0].value["action_requests"] if interrumpio else []
    if interrumpio:
        r = agente.invoke(Command(resume={"decisions": [{"type": "approve"}] * len(propuesta)}), context=ctx_de(item), config=cfg)
    return {"interrumpio": interrumpio, "propuesta": propuesta, "respuesta": r["messages"][-1].content}

Se reanuda siempre con approve para que la respuesta final exista y se pueda evaluar el texto. Lo que se compara con la etiqueta del encargado es propuesta, no la decisión de reanudación.

Los evaluadores

El contrato es una función con argumentos nominales input, output, expected_output y metadata, que devuelve un Evaluation o una lista. Puede ser síncrona o asíncrona.

def herramienta_correcta(*, input, output, expected_output, metadata, **kwargs):
    esperada = expected_output.get("tool")
    llamadas = [c["tool"] for c in output["llamadas"]]
    if esperada is None:
        ok = llamadas == []
    else:
        ok = esperada in llamadas
    return Evaluation(name="herramienta_correcta", value=float(ok),
                      comment=f"esperada={esperada} llamadas={llamadas}")

def escalado_correcto(*, output, expected_output, **kwargs):
    escalo = any(c.get("error") == "permiso_denegado" for c in output["llamadas"]) or "encargado" in output["respuesta"].lower()
    return Evaluation(name="escalado_correcto", value=float(escalo == expected_output["escala"]))

def cita_seccion(*, output, expected_output, **kwargs):
    if not expected_output.get("seccion"):
        return Evaluation(name="cita_seccion", value=1.0, comment="no aplica")
    return Evaluation(name="cita_seccion", value=float(expected_output["seccion"] in output["respuesta"]))

def sin_plazos(*, output, **kwargs):
    promete = re.search(r"\b(\d+\s*(días|horas)|mañana|hoy mismo)\b", output["respuesta"], re.I)
    return Evaluation(name="sin_plazos", value=float(promete is None))

def tasa_escalado(*, item_results, **kwargs):
    escalos = [r.output["llamadas"] and any(c.get("error") == "permiso_denegado" for c in r.output["llamadas"]) for r in item_results]
    return Evaluation(name="tasa_escalado", value=sum(map(bool, escalos)) / len(item_results))

Cinco evaluadores, cero modelos. herramienta_correcta y escalado_correcto son las dos métricas del caso 1 que un responsable de tienda entiende. cita_seccion comprueba que el identificador de sección está en la respuesta, no que la respuesta sea buena. sin_plazos es una expresión regular, y es suficiente para detectar que una versión del prompt ha empezado a prometer entregas. tasa_escalado es de ejecución: no compara con nada, describe la distribución, y sirve para ver que una versión del prompt escala el doble sin que ningún caso individual falle.

Un evaluador de idioma, castellano, usa un detector de idioma sobre la respuesta. Existe porque el caso 2 devuelve el rechazo del encargado en inglés al modelo y se ha visto al modelo contestar en inglés.

El juez con modelo, y su calibración

Para C04 (afirmaciones no respaldadas) y C06 (tono de la casa) del caso 4 no hay expresión regular. Hace falta leer. El juez es el propio evaluador del caso 4, con su prompt y ProviderStrategy(Veredicto), ejecutado como evaluador del experimento:

async def juez_c04_c06(*, output, expected_output, **kwargs):
    v = await evaluador_fichas.ainvoke({"messages": [{"role": "user", "content": mensaje_evaluador(output["ficha"], output["comprobaciones"], output["origen"])}]})
    veredicto = v["structured_response"]
    rechazados = {p.criterio for p in veredicto.problemas}
    esperados = set(expected_output["criterios_rechazados"])
    return [
        Evaluation(name="juez_aprobada_coincide", value=float(veredicto.aprobada == expected_output["aprobada"])),
        Evaluation(name="juez_c04_coincide", value=float(("C04" in rechazados) == ("C04" in esperados))),
        Evaluation(name="juez_c06_coincide", value=float(("C06" in rechazados) == ("C06" in esperados))),
    ]

Aquí el experimento no evalúa al generador. Evalúa al evaluador: cuánto coincide con la persona de catálogo sobre las fichas que la persona ya leyó. Es la calibración. Mientras juez_aprobada_coincide esté por debajo del 90 por ciento sobre la muestra, el juez no se usa para nada más que para medirse a sí mismo, y lo que se toca es su prompt y sus criterios.

Cuando pasa del umbral, se usa como evaluador del generador: el experimento del generador lleva juez_c04_c06 en sus evaluadores, y su puntuación se lee sabiendo que el juez acierta nueve de cada diez veces frente a una persona. Un juez que no se ha medido así es un segundo prompt opinando sobre el primero, y sus puntuaciones son números sin unidad.

Comparar dos versiones

Lo que un experimento sirve para decidir es si la versión nueva es mejor que la vieja en las métricas que importan y no peor en las demás. Dos ejecuciones sobre el mismo dataset:

Las cifras de esta tabla son el formato, no una medida: dependen del modelo, del dataset y del prompt de cada despliegue. Lo que la tabla enseña es cómo se lee. La v7 acierta más y cita más, y ha empezado a prometer plazos en el 2 por ciento de los casos. Eso es un caso concreto en item_results con sin_plazos = 0 y un comment con la frase, y se mira antes de decidir. Y cuesta un 15 por ciento más de tokens, que es lo que hay que poner en la balanza con el 6 por ciento de mejora en escalado.

El mismo cuadro sirve para cambiar de modelo con el mismo prompt, que es la comparación que más se hace en un cluster propio cuando sale una versión nueva del modelo abierto. El artículo de versionado de prompts cubre cómo se etiqueta la versión que se despliega; aquí solo importa que run_name y metadata lleven prompt y modelo, para que dentro de tres meses la tabla se entienda.

La puerta en CI

from langfuse.experiment import RegressionError

resultado = dataset.run_experiment(...)
medias = {e.name: e.value for e in resultado.run_evaluations}
media_escalado = sum(
    e.value for r in resultado.item_results for e in r.evaluations if e.name == "escalado_correcto"
) / len(resultado.item_results)

if media_escalado < 0.90:
    raise RegressionError(result=resultado, metric="escalado_correcto", value=media_escalado, threshold=0.90)

RegressionError está pensada para la acción langfuse/experiment-action de GitHub, que la captura, falla el flujo y escribe el comentario en el PR. Con Forgejo no hay acción, y hace lo mismo un paso del flujo que ejecuta el script y deja que la excepción salga: código distinto de cero, flujo rojo. El ExperimentResult se guarda como artefacto del flujo con format(include_item_results=True) para leer los casos que fallaron sin abrir Langfuse.

Qué métricas son puerta y cuáles son informe es una decisión por agente:

Las puertas con = 1,00 son las que no admiten un caso: una promesa de plazo o una operación con dinero sin interrupción no es una regresión del 2 por ciento, es un incidente.

Un experimento contra vLLM con un dataset de doscientos casos y max_concurrency=16 tarda unos minutos y cuesta lo que doscientas conversaciones. Corre en cada cambio de prompt o de modelo, no en cada commit; los tests del artículo anterior sí corren en cada commit y son gratis.

Concurrencia y coste del experimento

run_experiment tiene max_concurrency=50 por defecto. Es el número de tareas en vuelo, y cada tarea es una conversación entera con sus llamadas al modelo. Contra un vLLM detrás de una clave de gateway con max_parallel_requests a 24, el valor por defecto son 26 tareas recibiendo 429 y reintentando. El valor correcto es el de la clave del agente de evaluación, que es distinta de la de producción para que un experimento no consuma el presupuesto de la tienda.

Dentro de cada tarea, max_concurrency del grafo aplica al abanico interno, como en el caso 3. Las dos capas se multiplican: 16 tareas por 4 de abanico son 64 peticiones potenciales, y eso es lo que tiene que caber.

El coste se lee en Langfuse por la traza de cada elemento, agrupando por el run_name. Un experimento del caso 1 sobre doscientos casos son unos 400.000 tokens de entrada con el prompt v7 de la tabla, la mayoría en prefijo compartido, y cuestan lo que el gateway diga que cuesta ese modelo, que en un cluster propio es la cifra que el artículo de FinOps explica cómo asignar.

Checklist

• Cada agente tiene un dataset con salida esperada estructurada, no texto.
• La aplicación escribe pulgar con create_score y alguien convierte los pulgares abajo en casos cada semana.
• Las decisiones de encargados y las ediciones de compras alimentan sus datasets sin trabajo manual.
• La tarea corre en hilos eval: con InMemorySaver y contexto del caso.
• Los evaluadores en código cubren herramienta, argumentos, escalado, cita, idioma y plazos.
• El juez con modelo se calibra contra la muestra humana antes de puntuar al generador.
• max_concurrency del experimento es el de la clave de evaluación del gateway.
• run_name y metadata llevan prompt y modelo.
• Hay una tabla de qué métricas son puerta y cuáles informe, con umbrales.
• La puerta lanza RegressionError y el flujo de CI la deja salir.
• El experimento corre en cambios de prompt o modelo; los tests, en cada commit.

Trampas y cosas que no son lo que parecen

Comparar texto con texto. Es la evaluación más fácil de escribir y la que menos dice. Se compara la decisión: herramienta, argumentos, escalado, veredicto.

Un juez sin calibrar. Es un prompt puntuando a otro prompt. Primero se mide contra personas sobre la misma muestra; después se usa, y se usa sabiendo su acuerdo.

max_concurrency=50 por defecto. Contra una clave de gateway de 24, son 429 desde el primer segundo. Se fija al valor de la clave.

El dataset en la tabla de checkpoints de producción. La tarea usa su propio checkpointer en memoria. Doscientos hilos eval: por experimento en Postgres es lo que la retención acabaría borrando, pero mientras tanto están ahí.

El experimento con la clave de producción. Consume el presupuesto de la tienda y sus métricas de coste. Clave propia para evaluación.

Un experimento que pasa en la media y falla en un caso de dinero. Las puertas de = 1,00 existen para eso. Una media del 98 por ciento en sin_plazos son cuatro promesas en doscientos casos.

El resultado se produce aunque Langfuse esté caído. Verificado: run_experiment devuelve el ExperimentResult con las puntuaciones y solo fallan las exportaciones. Un flujo de CI que no compruebe que las trazas llegaron puede estar evaluando sin dejar rastro.

El dataset que no crece. Un dataset de cuarenta casos escritos el primer día mide el primer día. Los pulgares, las decisiones y la muestra semanal son lo que lo mantiene vivo.

Cierre

La evaluación de un agente es un dataset que crece solo, evaluadores que en su mayoría son código, un juez con modelo que se ha medido antes contra personas, y una puerta con umbrales por agente. Nada de eso es específico de LangGraph; lo específico es qué se guarda como salida de la tarea, y es lo que el grafo ya expone: las herramientas que llamó, con qué argumentos, si interrumpió, qué veredicto dio. Con los tests del artículo anterior fijando el grafo y esto midiendo al modelo, un cambio de prompt o de modelo deja de ser una opinión y pasa a ser una tabla con dos filas.

Ver también

• Probar agentes de LangGraph sin modelo: tests de grafo con pytest
• Agentes con LangGraph para retail: el marco de la serie
• Agente de datos de producto con LangGraph: generador, evaluador y bucle acotado
• LiteLLM y Langfuse como par operativo
• Versionado de prompts con Langfuse y MLflow
• FinOps y multi-tenancy de GPU con LiteLLM
• Arquitectura y ajuste de Langfuse self-hosted

Fuentes

• Langfuse, datasets y experimentos: https://langfuse.com/docs/evaluation/experiments/experiments-via-sdk
• Langfuse, puntuaciones (create_score): https://langfuse.com/docs/evaluation/evaluation-methods/custom-scores
• Código del SDK langfuse 4.15.4: Langfuse.run_experiment, Langfuse.create_dataset_item, Langfuse.create_score, langfuse.experiment (Evaluation, TaskFunction, EvaluatorFunction, RunEvaluatorFunction, ExperimentResult, RegressionError)
• Acción langfuse/experiment-action para GitHub: https://github.com/langfuse/experiment-action