Seis primitivas de LangGraph y el modo de fallo oculto en cada una

Un agente LangGraph inicial suele funcionar rápidamente: responde a una pregunta, perfecciona la respuesta y se detiene. Luego alguien agrega un bucle de intentos, y de repente el grafo deja de terminar, consumiendo créditos de la API hasta que se interrumpe el proceso. La solución suele ser un solo enlace faltante, pero el verdadero problema es la falta de un modelo mental que explique por qué el grafo se comporta de esa manera.

Esta guía construye dicho modelo a partir de las seis primitivas que componen LangGraph: estado, nodos, aristas directas, aristas condicionales, puntos de control y intervención humana. Para cada una se mostrará un ejemplo mínimo, el error más frecuente que cometen los equipos al trabajar con ella y la versión adecuada para su implementación. Si desea un recorrido más amplio sobre los patrones de agentes basados en estas componentes, LangGraph en la práctica: estado, nodos, aristas y cinco patrones de agentes aborda ese tema; aquí el enfoque está en los modos de fallo.

¿Por qué un grafo en lugar de una cadena?

La sintaxis de tuberías de LangChain, prompt | llm | parser, resulta práctica para un único proceso a través del modelo. Deja de ser adecuada en cuanto el agente debe tomar una decisión: buscar o responder directamente, intentarlo de nuevo o rendirse, preguntarle a una persona o continuar. Una cadena no tiene concepto de “depende de”, por lo que los desarrolladores envuelven las llamadas a la cadena en instrucciones if, y pronto terminan creando una máquina de estados no documentada y más difícil de depurar.

LangGraph hace que esa máquina de estados sea explícita. Se obtienen nodos, aristas y un objeto de estado compartido que se puede inspeccionar en cualquier momento. No hay nada mágico en ello, y ese es precisamente el vantaje: cada decisión que toma el agente corresponde a algo que se puede leer en la definición del grafo.

1. Estado: un objeto compartido y reductores importantes

El estado es el único objeto del cual cada nodo lee y al que escribe. Sin él, el contexto tiende a ser pasado como argumentos de función, y resulta difícil determinar qué sabía realmente cada paso. La definición a continuación es un TypedDict que contiene una pregunta, una respuesta y una lista de mensajes cuyas actualizaciones se fusionan mediante el reductor add_messages.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add no es un reductor de mensajes

Muchos tutoriales anotan el campo de mensajes con operator.add en su lugar. Parece correcto: add agrega al final de la lista en lugar de sobrescribirla, lo cual es necesario para una conversación que va creciendo. El problema es que realiza la concatenación de forma automática y sin control. Tan pronto como se necesita actualizar o eliminar un mensaje existente, por ejemplo al reducir el historial o reemplazar el resultado de una llamada a una herramienta, en su lugar se agrega un duplicado, y el historial de la conversación se llena de entradas obsoletas sin que aparezca ningún error.

add_messages está diseñado específicamente para esto. Compara los mensajes por su ID y reemplaza un mensaje existente cuando ese ID ya está presente, añadiendo únicamente aquellos que son realmente nuevos. La regla es sencilla: utilice add_messages para campos que contengan objetos HumanMessage y AIMessage, y mantenga operator.add para listas simples de acumulación, como un registro continuo de las herramientas que se han utilizado.

Mantenga la estructura del estado al mínimo

El segundo error común es diseñar el estado como un esquema de base de datos, con un campo para cada necesidad que alguien pueda tener en el futuro. Añada un campo solo cuando un nodo realmente lo lea o lo escriba. El costo de ignorar esto es concreto: considere un grafo de procesamiento de documentos que almacena en el estado las respuestas completas y sin procesar del LLM, incluyendo metadatos de uso de tokens. Procesar 50 documentos en un bucle hizo que cada punto de control alcanzara unos 180 KB, y las operaciones de escritura en Postgres superaron los 400 ms, lo suficientemente lento como para que los usuarios que esperaban una respuesta se dieran cuenta. La solución no fue nada elegante: reducir el estado a los tres campos que realmente utilizaban los nodos siguientes. Recuerde que, con un punto de control asociado, todo lo que está en el estado se serializa y se guarda en cada paso.

2. Nodos: devuelvan solo lo que ha cambiado

Un nodo es una función ordinaria de Python. Recibe el estado, realiza su trabajo y devuelve un diccionario que contiene únicamente los campos que modificó. Ese es todo el contrato. El primer ejemplo llama a un modelo de chat de OpenAI con la pregunta y escribe la respuesta en answer.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Mientras iteras sobre la estructura del grafo, que corresponde a la mayor parte del trabajo inicial, es posible que no quieras que cada versión preliminar acceda a una API de pago. Un modelo local servido por Ollama implementa la misma interfaz, por lo que el cuerpo del nodo se mantiene idéntico y depurar no cuesta nada:

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Esta versión requiere que Ollama esté ejecutándose localmente con el modelo descargado (ollama pull llama3.1) y que esté instalado el paquete de integración (pip install langchain-ollama). Establecer temperature=0 también hace que las ejecuciones sean más reproducibles, lo cual es útil al probar la lógica de enrutamiento.

Devolver todo el estado sobrescribe otras actualizaciones

Un error frecuente es devolver todo el diccionario de estado desde un nodo en lugar de solo las claves que han cambiado. En un grafo lineal pequeño esto parece funcionar, ya que nada más afecta a esos campos. Sin embargo, cuando dos nodos actualizan campos coincidentes, la devolución completa de uno de los nodos sobrescribe los cambios del otro con valores antiguos. El síntoma se asemeja a un problema de enrutamiento, por lo que los desarrolladores tienden a buscar en la lógica de las aristas, cuando la causa real es que un nodo devuelve demasiada información. Devolver una actualización mínima también permite que los reducers cumplan su función: un campo sin reducer simplemente se reemplaza por lo que devuelva el nodo.

3. Aristas directas: siempre conecte la salida

Los bordes determinan qué se ejecuta a continuación. Un borde directo es incondicional: cuando el nodo A termina, se ejecuta el nodo B. El gráfico a continuación registra dos nodos, conecta answer con refine, conecta refine con END, establece el punto de entrada y compila.

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

El borde END es esa parte que la gente olvida, y constituye la causa clásica de un grafo que parece ejecutarse para siempre. La costumbre totalmente fiable es hacer que cada camino a través del grafo termine explícitamente en END, de modo que se pueda determinar el final al leer la definición. Esto es especialmente importante cuando aparecen ciclos: un bucle de intentos sin camino hacia END, o con una condición que nunca se vuelve verdadera, continúa en ciclos hasta que el límite de recursión de LangGraph lo detiene con un GraphRecursionError. Ese límite es una red de seguridad, no parte del diseño; cada una de esas iteraciones sigue costando tokens. Cuando un grafo parezca atascado, revise primero su definición.

4. Bordes condicionales: donde el agente toma realmente decisiones

Los bordes condicionales son lo que convierte al grafo en un agente en lugar de una secuencia fija. Una función de enrutamiento inspecciona el estado y devuelve una etiqueta; un mapeo traduce cada etiqueta al nodo siguiente. En este ejemplo, una respuesta corta (menos de 50 caracteres) se envía a refine, y todo lo demás va a END.

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

Tenga en cuenta que este borde condicional reemplaza el borde directo answer a refine del fragmento anterior. Si registra ambos, se seguirán ambos caminos, lo cual rara vez es lo que se desea.

Las etiquetas de ruta incompatibles generan errores evidentes pero poco claros

El error recurrente aquí es una función de enrutamiento que devuelve una cadena que no está presente en el diccionario de mapeo. El error resultante es un error de clave bastante genérico, oculto a varios niveles profundamente en el rastro de pila, y algo tan sencillo como un espacio al final puede hacer que se pierda una cantidad sorprendente de tiempo. Un hábito fiable: escribir primero el mapeo y luego el enrutador copiando las claves exactas de él. Aún mejor, definir las etiquetas una vez como constantes o anotar el tipo de retorno del enrutador con Literal["refine", "done"] para que los verificadores de tipo y los lectores vean inmediatamente los valores permitidos.

5. Puntos de control: memoria que persiste entre llamadas

Un checkpointer convierte una llamada a función sin estado en una conversación con la memoria. Sin él, cada app.invoke() comienza desde cero. Con él, el estado se guarda por hilo, y cualquier llamada que incluya el mismo thread_id en su configuración continúa desde donde se detuvo la anterior. En el ejemplo, la segunda invocación en el hilo user-session-42 recuerda la primera pregunta.

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver es adecuado únicamente para el desarrollo local. Vive en la memoria del proceso, por lo que un reinicio del servidor borra todas las conversaciones. Cualquier sistema del que dependan los usuarios reales necesita un backend persistente: SQLite para un único servidor, o Postgres cuando varias instancias deben compartir estado.

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

Cada backend se entrega como su propio paquete, como indican los comentarios de instalación. En las versiones actuales, estos salvadores suelen crearse a partir de una cadena de conexión (por ejemplo, mediante from_conn_string), y Postgres necesita una llamada única a setup() para crear sus tablas; por lo tanto, consulte la documentación de checkpointer para conocer la inicialización exacta en su versión.

El escenario de fallo aquí es utilizar el salvador en memoria en producción y darse cuenta de ello cuando un reinicio en entorno de pruebas borra una demostración en tiempo real. La buena noticia es que el cambio es sencillo si el grafo está bien estructurado: checkpointer es un parámetro en tiempo de compilación, no una rediseño, y pasar a SqliteSaver puede realizarse en menos de una hora.

6. Intervención humana: puntos de interrupción estáticos versus interrupciones dinámicas

El patrón que muestran la mayoría de los tutoriales es interrupt_before, una lista de nombres de nodos donde el grafo compilado se detiene antes de ejecutarse:

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

Funciona y es fácil de explicar, pero es estático. El punto de pausa está fijado por el nombre del nodo; no se puede hacer condicional ni se puede adjuntar una carga útil que describa qué debe observar el revisor. Los requisitos reales lo superan rápidamente, ya que “pausar antes de este nodo” y “pausar solo cuando el reembolso supere los $500” son reglas diferentes, y solo la primera se puede expresar de esta manera.

Pausa desde dentro del nodo con interrupt()

El patrón más flexible consiste en llamar a interrupt() desde dentro del nodo. El nodo mostrado a continuación verifica el monto del reembolso: si es superior a 500 dólares, se detiene y muestra el borrador junto con el monto a una persona. La primera llamada a invoke se ejecuta hasta ese momento de pausa. La segunda llamada envía Command(resume="approve") en el mismo hilo, y el valor proporcionado a resume se convierte en el valor de retorno de interrupt(); por lo tanto, el nodo procede a enviar el mensaje o devuelve un estado de cancelación. Se necesita un puntero de verificación, ya que el estado de pausa debe almacenarse en algún lugar mientras espera.

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

El estado del ejemplo utiliza campos como refund_amount, draft y task que no forman parte del anterior AgentState; en un grafo real, estos campos deberían declararse allí.

Reanudar ejecuta todo el nodo

El comportamiento que sorprende a las personas: en el flujo de ejecución, LangGraph no continúa desde la línea interrupt(). Vuelve a ejecutar todo el nodo desde el principio, y esta vez interrupt() devuelve el valor del flujo en lugar de pausar. Cualquier código anterior a la llamada se ejecuta nuevamente. Un nodo que incrementa un contador antes de interrumpir lo incrementará dos veces por cada aprobación. Mantenga todo lo que está antes de interrupt() como idempotente, o mueva los efectos secundarios a un nodo anterior. El mismo razonamiento se aplica a las llamadas a API o las escrituras en la base de datos que se realicen antes de la pausa.

Un modelo que se aprueba a sí mismo no cuenta con intervención humana

Cualquiera que sea el mecanismo que elija, preguntarle al modelo “¿Debo continuar?” y confiar en su respuesta no constituye supervisión humana, independientemente de cómo se denomine. Se trata del agente que aprueba su propia decisión. Un verdadero paso de aprobación transfiere el control a una persona fuera del grafo y espera su respuesta.

Un vistazo a las seis primitivas

El resumen a continuación relaciona cada concepto con su función y el error típico asociado a él.

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

Un orden de construcción sensato

Para el primer grafo real, haga que el bucle completo funcione de principio a fin con InMemorySaver y sin interrupciones. Mantenga el estado reducido y limitado a lo que necesitan los nodos, y asegúrese de que cada arista condicional devuelva exactamente las etiquetas que espera su mapeo. Solo una vez que funcione sin problemas debería introducir un checkpointer persistente y agregar una interrupción en el paso que realmente requiera intervención humana, generalmente cualquier acción que mueva dinero, envíe un correo electrónico externo o elimine datos.

Las funciones más avanzadas, como reductores personalizados además de add_messages, subgrafos que dividen un grafo grande en partes probables, y transmisión a nivel de token, se basan en el mismo esqueleto. Son mucho más fáciles de adoptar una vez que haya creado, roto y arreglado un grafo utilizando únicamente estas seis ideas.

Puntos clave

• Use add_messages para el historial de chat y operator.add únicamente para listas simples, y mantenga el estado minimalista ya que se persiste en cada paso.
• Devuelva solo los campos modificados de los nodos; las devoluciones con el estado completo sobrescribirán silenciosamente actualizaciones paralelas o anteriores.
• Asigne a cada ruta un destino explícito hacia END, y utilice bucles de intento restringidos en lugar de depender del límite de recursión.
• Obtenga las etiquetas de enrutamiento a partir del mapeo para que no se desvíen entre sí.
• Trate InMemorySaver como algo exclusivo para el desarrollo; el intercambio de checkpointer es sencillo, así que hágalo antes de que los usuarios dependan del grafo.
• Prefiera usar interrupt() dinámicamente para aprobaciones condicionales, y mantenga el código antes de que sea idempotente, ya que reanudarlo ejecutará nuevamente el nodo.

Documentación de referencia: la documentación de la API Graph para LangGraph, la guía sobre interrupciones y la referencia de la API interrupt().