Agente de datos de producto con LangGraph: enriquecer y controlar fichas con un generador, un evaluador y un bucle que sabe parar

TL;DR

El generador escribe la ficha y el evaluador la puntúa contra una lista de criterios que no cambia. Son dos agentes distintos, con dos prompts distintos y salida estructurada los dos. El evaluador no ve el prompt del generador ni sus intentos anteriores; ve la ficha y los criterios. Es la única forma de que el evaluador no apruebe lo que el generador quiso hacer en vez de lo que hizo.

El bucle lo cierra una arista condicional que lee dos cosas del estado: el veredicto y un contador de intentos. Tres como máximo. El modelo no sabe cuántos intentos lleva ni decide cuándo parar. Verificado con un evaluador que rechaza dos veces y aprueba a la tercera: tres intentos, tres veredictos en el historial, y el grafo termina en publicar.

Con un modelo en vLLM, response_format en create_agent no hace lo que parece. Sin perfil en el modelo, LangChain no puede saber si el proveedor admite salida estructurada y recurre a una herramienta sintética con el nombre del esquema, que el modelo tiene que llamar. Con profile={"structured_output": True} usa response_format con el esquema JSON y vLLM lo aplica con decodificación guiada. Verificado en _supports_provider_strategy de langchain 1.4.1.

Un agente de create_agent invocado dentro de un nodo hereda el checkpointer del grafo padre y guarda sus propios checkpoints bajo un espacio de nombres. Una ficha con tres intentos deja 30 filas; con checkpointer=False en los dos agentes, 12. El estado que importa está en el padre; los subgrafos no necesitan reanudarse.

El resultado estructurado de un agente es un objeto pydantic, y ese objeto entra en el estado del subgrafo. Al deserializar el checkpoint, LangGraph 1.2.11 avisa de que el tipo no está registrado y de que lo bloqueará en una versión futura. Se guarda en el estado del padre como diccionario con model_dump(), y se registra el módulo en allowed_msgpack_modules o se desactiva el checkpointer del subgrafo.

Lo que se mide es la tasa de aprobación por intento. Es la única métrica que dice si el bucle compensa: si a la tercera se aprueba lo mismo que a la segunda, el tercer intento sobra.

Estás aquí: el cuarto caso, el bucle que corrige

Este es el caso 4 de la serie de agentes con LangGraph para retail. Los tres anteriores tratan con personas o con un catálogo entero; este trata con un texto. Una ficha de producto tiene un título, una descripción, unos atributos, y unas reglas de la casa sobre lo que puede y no puede decir. Las fichas llegan del proveedor a medias, en otro idioma, o con los atributos en el título. Hoy las completa alguien de catálogo, una a una, y las publica sin que nadie las vuelva a leer.

El agente hace lo mismo con dos diferencias: escribe y comprueba en dos pasos separados, y el segundo paso tiene derecho de veto. Es el patrón que la documentación llama reflexión, y la forma más simple de un evaluador: no puntúa de uno a diez, dice sí o no y por qué.

La analogía: el redactor y el corrector

En cualquier redacción hay dos personas. Una escribe y otra corrige, y la que corrige no ha visto el borrador anterior ni sabe lo que el redactor intentaba decir; solo tiene el texto y el libro de estilo. Si el corrector fuera el propio redactor releyendo, aprobaría lo que quiso escribir.

El generador es el redactor. El evaluador es el corrector con el libro de estilo. El contador de intentos es el jefe de sección que a la tercera vuelta publica o descarta, sin que el redactor lo decida.

Los criterios

El evaluador comprueba una lista cerrada. Cada criterio tiene un identificador, una descripción para el evaluador y una descripción para el generador. Se guardan en una tabla, no en el prompt, y se versionan.

Los criterios de la tercera columna marcados con “Sí” se comprueban en código antes de llamar al evaluador, y el resultado se le pasa. Un modelo evaluando si un título tiene menos de 80 caracteres es un modelo haciendo mal lo que len() hace bien. Lo que el evaluador aporta es C04, C06 y las partes no mecánicas de C05: lo que exige leer.

El estado

import operator
from typing import Annotated, TypedDict

class EstadoFicha(TypedDict):
    sku: str
    origen: dict                      # ficha del proveedor, atributos de familia, cifras
    ficha: dict                       # última versión generada
    comprobaciones: dict              # C01, C02, C03, C07: resultado en código
    veredicto: dict                   # {"aprobada": bool, "problemas": [{"criterio": ..., "detalle": ...}]}
    intentos: Annotated[int, operator.add]
    historial: Annotated[list[dict], operator.add]

intentos con operator.add es el contador: cada vez que generar devuelve {"intentos": 1}, el reductor suma. historial acumula un registro por intento con la ficha y el veredicto, y es lo que después permite medir. ficha y veredicto se sobreescriben.

ficha y veredicto son diccionarios, no objetos pydantic. Los agentes devuelven pydantic en structured_response, y el nodo lo convierte con model_dump() antes de escribirlo en el estado. Más abajo se explica por qué no es una manía.

Los dos agentes

from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ProviderStrategy

class Ficha(BaseModel):
    titulo: str = Field(max_length=80)
    descripcion: str = Field(max_length=1200)
    atributos: dict[str, str]

class Problema(BaseModel):
    criterio: str
    detalle: str = Field(max_length=200)

class Veredicto(BaseModel):
    aprobada: bool
    problemas: list[Problema] = Field(default_factory=list)

generador = create_agent(
    modelo,
    tools=[],
    system_prompt=PROMPT_GENERADOR,      # reglas de la casa, formato, y "corrige solo lo señalado"
    response_format=ProviderStrategy(Ficha),
    checkpointer=False,
)

evaluador = create_agent(
    modelo,
    tools=[],
    system_prompt=PROMPT_EVALUADOR,      # criterios C04, C05, C06 y "responde solo con el veredicto"
    response_format=ProviderStrategy(Veredicto),
    checkpointer=False,
)

Ninguno tiene herramientas. Todo lo que necesitan va en el mensaje: la ficha del proveedor, los atributos obligatorios de la familia, y para el generador, los problemas del intento anterior. Son dos llamadas al modelo por intento, sin ciclo de herramientas; create_agent está aquí por la salida estructurada, el prompt y el middleware, no por el bucle.

Dos parámetros que no son decorativos. ProviderStrategy se explica en su propia sección. checkpointer=False también.

Los prompts, y lo que cada uno no ve

El generador recibe la ficha del proveedor y, a partir del segundo intento, la lista de problemas del veredicto anterior con la instrucción de corregir solo eso. No recibe su ficha anterior entera. Se probó dárselo y el resultado fue peor: con la ficha anterior delante, el modelo reescribe frases que estaban bien y a veces reintroduce un problema ya corregido en otro sitio. Con la ficha del proveedor y la lista de problemas, regenera desde el origen con las restricciones acumuladas.

El evaluador recibe la ficha generada, los resultados de las comprobaciones en código, y los criterios. No recibe el prompt del generador, ni sus intentos anteriores, ni cuántos lleva. Si supiera que es el tercer intento, aprobaría más; se ha visto en pruebas con el número de intento en el mensaje. La única información de contexto es el origen, para C04 y C07.

El grafo

from langgraph.graph import StateGraph, START, END

def comprobar(state: EstadoFicha) -> dict:
    return {"comprobaciones": comprobaciones_en_codigo(state["ficha"], state["origen"])}

def generar(state: EstadoFicha) -> dict:
    problemas = state.get("veredicto", {}).get("problemas", [])
    salida = generador.invoke({"messages": [{"role": "user", "content": mensaje_generador(state["origen"], problemas)}]})
    return {"ficha": salida["structured_response"].model_dump(), "intentos": 1}

def evaluar(state: EstadoFicha) -> dict:
    salida = evaluador.invoke({"messages": [{"role": "user", "content": mensaje_evaluador(state["ficha"], state["comprobaciones"], state["origen"])}]})
    veredicto = salida["structured_response"].model_dump()
    fallos_codigo = [c for c, ok in state["comprobaciones"].items() if not ok]
    if fallos_codigo:
        veredicto["aprobada"] = False
        veredicto["problemas"] += [{"criterio": c, "detalle": "comprobación automática"} for c in fallos_codigo]
    return {"veredicto": veredicto, "historial": [{"intento": state["intentos"], "ficha": state["ficha"], "veredicto": veredicto}]}

def decidir(state: EstadoFicha) -> str:
    if state["veredicto"]["aprobada"]:
        return "publicar"
    if state["intentos"] >= 3:
        return "descartar"
    return "generar"

def publicar(state: EstadoFicha) -> dict:
    guardar_ficha(state["sku"], state["ficha"], estado="publicada", intentos=state["intentos"])
    return {}

def descartar(state: EstadoFicha) -> dict:
    guardar_ficha(state["sku"], state["ficha"], estado="revision_manual", intentos=state["intentos"],
                  problemas=state["veredicto"]["problemas"])
    return {}

grafo = StateGraph(EstadoFicha)
grafo.add_node("generar", generar)
grafo.add_node("comprobar", comprobar)
grafo.add_node("evaluar", evaluar)
grafo.add_node("publicar", publicar)
grafo.add_node("descartar", descartar)
grafo.add_edge(START, "generar")
grafo.add_edge("generar", "comprobar")
grafo.add_edge("comprobar", "evaluar")
grafo.add_conditional_edges("evaluar", decidir, {"generar": "generar", "publicar": "publicar", "descartar": "descartar"})
grafo.add_edge("publicar", END)
grafo.add_edge("descartar", END)

Cinco nodos y una arista condicional con tres salidas. decidir es el único sitio donde se lee el contador, y es una función sin modelo. El modelo no puede alargar el bucle ni acortarlo.

Las comprobaciones en código van en su propio nodo entre generar y evaluar, y el evaluador las recibe. Además, evaluar fuerza el rechazo si alguna falla: aunque el modelo apruebe, un título de 90 caracteres no se publica. El evaluador con modelo no tiene la última palabra sobre lo que se puede comprobar sin él.

Verificado con un generador que devuelve una ficha por intento y un evaluador guionizado que rechaza dos veces y aprueba a la tercera: intentos termina en 3, el historial tiene tres entradas con False, False, True, y el grafo sale por publicar. Con el evaluador rechazando siempre, sale por descartar a la tercera.

La salida estructurada, por debajo

response_format en create_agent admite un esquema a secas, ToolStrategy(esquema) o ProviderStrategy(esquema). Con el esquema a secas, LangChain elige por el modelo: en _supports_provider_strategy mira model.profile, y si el perfil dice structured_output: True usa ProviderStrategy; si no, busca el nombre del modelo en una lista de patrones de modelos conocidos, y si tampoco, ToolStrategy.

Con ChatOpenAI apuntando a vLLM y un nombre como qwen3-30b-a3b, el perfil es None y el nombre no está en la lista. Resultado: ToolStrategy. Verificado: _supports_provider_strategy(modelo, tools=[]) devuelve False sin perfil y True con profile={"structured_output": True}.

Lo que cambia entre las dos:

Para este agente se fija ProviderStrategy a mano en los dos, y el modelo lleva profile={"max_input_tokens": 32768, "structured_output": True} en el marco de la serie. El esquema se mantiene plano: Ficha tiene tres campos y atributos es un diccionario de cadenas, no un modelo anidado por familia. vLLM aplica bien esquemas de este tamaño; con esquemas de varios niveles y enumeraciones largas la decodificación guiada se nota en la latencia de la primera ficha.

Field(max_length=80) en el título es una restricción del esquema que vLLM no aplica en la generación; la aplica pydantic al validar. Con ProviderStrategy un fallo de validación no se reintenta: el agente lanza y el nodo generar lanza con él. Por eso generar captura esa excepción y manda la ficha a revisión manual sin pasar por el evaluador, y por eso C01 se comprueba también en código: lo que está dentro del esquema es una segunda barrera, no la primera.

Los subgrafos y sus checkpoints

generador y evaluador son grafos compilados. Cuando se invocan dentro de un nodo del grafo padre con .invoke(...), reciben la configuración del padre, y con ella el checkpointer. Sin hacer nada, cada uno guarda sus propios checkpoints bajo un espacio de nombres derivado del nodo y de la tarea.

Medido con InMemorySaver, una ficha con tres intentos:

Los 18 de los subgrafos son tres por cada invocación de agente, seis invocaciones. Son el estado interno de cada llamada: los mensajes de esa llamada, la respuesta estructurada, y el resultado. Ninguno sirve para nada después: la ficha ya está en el estado del padre como diccionario, y un subgrafo sin herramientas ni interrupciones no tiene nada que reanudar.

checkpointer=False en create_agent le dice al agente que no persista aunque el padre lo haga. Es distinto de checkpointer=None, que es el valor por defecto y significa “hereda”. Con treinta mil fichas al año y tres intentos de media, la diferencia son 570.000 filas de checkpoint que no dicen nada, más sus blobs.

Hay un caso en que se quiere lo contrario: si el generador tuviera herramientas con efecto o una interrupción de aprobación, sus checkpoints serían lo que permite reanudarlo. Aquí no es el caso, y tampoco lo es en el analista de referencias del caso 3, que por eso lleva el mismo parámetro.

El objeto pydantic en el estado

Al ejecutar el bucle con los agentes por defecto, LangGraph 1.2.11 escribe esto al deserializar el checkpoint del subgrafo:

Deserializing unregistered type __main__.Ficha from checkpoint. This will be blocked in a
future version. Set LANGGRAPH_STRICT_MSGPACK=true to block now, or add to
allowed_msgpack_modules to allow explicitly: [('__main__', 'Ficha')]

structured_response en el estado del agente es un objeto Ficha. El serializador lo guarda con su módulo y su clase, y al leerlo comprueba que el tipo esté permitido. Hoy es un aviso; el mensaje dice que será bloqueo. Cuando lo sea, un checkpoint escrito hoy con un objeto pydantic dentro no se podrá leer con la versión de mañana.

Tres formas de no tener el problema, y aquí se usan las tres. El estado del padre solo lleva diccionarios: model_dump() en cuanto el objeto sale del agente. Los subgrafos no persisten, así que no hay checkpoint con el objeto. Y el checkpointer del padre se construye con un serializador que conoce el módulo de los esquemas, por si alguna vez uno se cuela:

from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

checkpointer = PostgresSaver(
    pool_agentes,
    serde=JsonPlusSerializer(allowed_msgpack_modules=[("fichas.esquemas", "Ficha"), ("fichas.esquemas", "Veredicto")]),
)

Es la misma regla que en el caso 2 con el contexto y en el caso 3 con motivo: lo que va al checkpoint se decide, no se deja caer.

Lo que se mide

La última fila es la que hace que el sistema no se degrade sin que nadie lo vea. Un evaluador con modelo aprueba lo que su prompt le dice que apruebe; una muestra semanal leída por una persona es lo que mantiene el prompt honesto.

Checklist

• Los criterios viven en una tabla versionada, con la parte mecánica comprobada en código.
• El evaluador no ve el prompt del generador, sus intentos anteriores ni el número de intento.
• El generador recibe el origen y los problemas, no su ficha anterior.
• decidir lee aprobada e intentos y no llama al modelo.
• evaluar fuerza el rechazo si una comprobación en código falla.
• Los dos agentes llevan ProviderStrategy y el modelo lleva structured_output: True en el perfil.
• Los esquemas son planos y las restricciones de longitud están también en código.
• Los dos agentes llevan checkpointer=False.
• El estado del padre solo tiene diccionarios; model_dump() a la salida de cada agente.
• allowed_msgpack_modules en el checkpointer registra el módulo de los esquemas.
• Hay una muestra semanal de fichas aprobadas que lee una persona.

Trampas y cosas que no son lo que parecen

response_format=Ficha con un modelo de vLLM es una herramienta, no un response_format. Sin perfil, LangChain no sabe que el proveedor admite salida estructurada y usa ToolStrategy. Funciona, pero paga la definición de la herramienta en cada llamada y falla de otra manera. ProviderStrategy explícito y structured_output: True en el perfil.

Un agente dentro de un nodo persiste solo. Hereda el checkpointer del padre y escribe bajo su espacio de nombres. 30 filas por ficha en vez de 12. checkpointer=False cuando no hay nada que reanudar.

El objeto pydantic en el checkpoint es un aviso hoy y un error mañana. model_dump() antes de escribir en el estado, y allowed_msgpack_modules como red.

Las restricciones del esquema no las aplica el servidor. max_length lo valida pydantic después de generar. La primera barrera para lo mecánico es código antes del evaluador.

El evaluador aprueba más si sabe que es el último intento. No se le dice. Tampoco se le dice qué intentó el generador. Solo la ficha, las comprobaciones, el origen y los criterios.

Darle al generador su ficha anterior empeora. Reescribe lo que estaba bien y reintroduce problemas. Origen más lista de problemas.

El tope de intentos no se calibra a ojo. Se mide la aprobación por intento. Si el tercero apenas aprueba, es coste sin resultado y el tope baja a dos.

Cierre

El agente de fichas es dos llamadas al modelo por intento y un contador. Lo que lo hace fiable no está en las llamadas: está en que el evaluador no ve lo que no debe, en que lo mecánico se comprueba sin modelo antes y después, en que el bucle lo cierra una función y no el modelo, y en que lo que entra en el checkpoint es lo que se ha decidido que entre. Y una muestra semanal leída por alguien de catálogo, que es lo que mantiene los criterios pegados a la realidad.

Con esto los cuatro casos están escritos. El siguiente y último artículo de la serie no construye ningún agente: sirve los cuatro en el mismo cluster, sin el servidor de LangGraph, con la autorización por hilo, la retención y las sondas que ninguno de los cuatro tiene.

Ver también

• Agentes con LangGraph para retail: el marco de la serie
• Agente de cadena de suministro con LangGraph: incidencias y reposición con map-reduce
• Agente de atención al cliente con LangGraph: el límite de aprobación humana
• Servir agentes de LangGraph sin langgraph-api: autorización por hilo, retención y sondas
• deepagents en un cluster propio: el SDK es libre, el servidor no
• Versionado de prompts con Langfuse y MLflow

Fuentes

• Salida estructurada en LangChain 1.x (ToolStrategy, ProviderStrategy): https://docs.langchain.com/oss/python/langchain/structured-output
• API de grafos de LangGraph (reductores, aristas condicionales, subgrafos): https://docs.langchain.com/oss/python/langgraph/graph-api
• Código de langchain.agents.factory (langchain 1.4.1): _supports_provider_strategy y la resolución de AutoStrategy
• Código de langgraph.checkpoint.serde.jsonplus (langgraph-checkpoint 4.2.0): aviso de tipos no registrados y allowed_msgpack_modules
• vLLM, salida estructurada y decodificación guiada: https://docs.vllm.ai/en/latest/features/structured_outputs.html