LangGraph en Python: Estado, nodos paralelos, puntos de control y revisión humana

Cualquiera que haya creado algo más que un juguete basado en un LLM conoce tres problemas recurrentes. Las llamadas realizadas una tras otra acumulan latencia. Un flujo de trabajo largo puede colapsar y perder todo su progreso. Además, en el peor momento posible, el modelo puede devolver una respuesta segura que resulte completamente incorrecta.

LangGraph aborda los tres problemas. Se trata de un marco de orquestación basado en grafos construido a partir de elementos simples: el estado es un TypedDict, los nodos son funciones ordinarias y las aristas representan la lógica de enrutamiento. Estos componentes básicos se combinan para crear flujos de trabajo que ejecutan nodos en paralelo, guardan el estado en puntos de control para poder recuperarse de un colapso y permiten hacer una pausa para que una persona revise el proceso cuando las consecuencias son graves.

Las secciones siguientes explican el proceso paso a paso. Primero aparece un nodo aislado, luego la ejecución en paralelo, el enrutamiento condicional y la memoria persistente, seguidos de las interrupciones con intervención humana. Al final, un flujo de trabajo completo de triaje de correos electrónicos combina todo esto. Cada ejemplo está diseñado para ejecutarse sin cambios. El SDK de TypeScript expone casi las mismas API, por lo que se aplica el mismo modelo mental.

Configuración

Primero se necesitan dos paquetes:

pip install langgraph langchain-openai

Luego, las variables de entorno para OpenAI y, opcionalmente, el seguimiento de LangSmith:

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

Los ejemplos se probaron con langgraph 1.1.3 (marzo de 2026) y langchain-openai 1.1.10, y requieren Python 3.10 o superior. Las guías antiguas a veces utilizan LANGCHAIN_TRACING_V2 y LANGCHAIN_API_KEY; ambos siguen funcionando, pero LANGSMITH_TRACING y LANGSMITH_API_KEY son ahora los nombres preferidos.

Todo en LangGraph parte de dos conceptos básicos: el estado almacena los datos, y los nodos son funciones que los transforman. Una vez que se dominan estos conceptos, el resto del framework encaja por sí solo.

Definir el estado con TypedDict

El estado en LangGraph es un TypedDict estándar de Python, sin clases base especiales ni decoradores propios del framework, simplemente diccionarios tipados:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

La clase define la estructura de los datos que fluyen a través del grafo, y cada nodo lee y escribe en esta estructura compartida.

¿Por qué usar TypedDict en lugar de un diccionario normal o una dataclass? Porque permite la verificación estática de tipos, ya que tanto mypy como Pyright lo comprenden, además de incluir indicaciones de tipo, y evita la sobrecarga en tiempo de ejecución que implican las dataclasses o los modelos de Pydantic. LangGraph también puede inspeccionar las anotaciones para conocer el esquema, lo cual es importante para la generación y validación de datos estructurados.

La línea que hay que entender es Annotated[list[str], operator.add], que declara un reductor. Es el concepto más importante en el estado de LangGraph, y casi siempre confunde a los usuarios la primera vez que lo ven.

Sin un reductor, un nodo que devuelve {"summary": "new value"} simplemente reemplaza el valor almacenado. Para campos escalares como summary eso es exactamente lo que se desea. Los problemas surgen cuando dos nodos en paralelo escriben en messages: solo se mantiene el que finaliza por último. Al declarar operator.add en su lugar, se indica a LangGraph que añada los valores, de modo que cada contribución a messages se acumula de forma segura.

Una lógica de fusión más compleja requiere un reductor personalizado:

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

Este reductor une la lista existente con los nuevos mensajes y luego la reduce a los diez más recientes, lo cual ayuda en conversaciones largas donde, de lo contrario, el recuento de tokens aumentaría exponencialmente. operator.add se utiliza para un simple apilamiento. Un reductor personalizado es la opción adecuada para eliminar duplicados, limitar el tamaño de la lista o realizar cualquier tipo de fusión que vaya más allá del simple apilamiento.

El error más común es olvidar incluir un reductor en un campo al que escriben nodos paralelos. Sin uno, las ramas sobrescriben silenciosamente una a otra, ganando siempre la última en procesarse, y nunca aparece ningún error.

TypedDict versus Pydantic

Ambos aparecen en LangGraph, y la separación entre ellos es intencionada. TypedDict se utiliza para el estado del grafo, el contenedor que se mueve a través del mismo. LangGraph fue diseñado en torno a él; sus campos soportan la sintaxis del reductor Annotated, se serializan de manera limpia para la creación de puntos de control y no añaden carga adicional en tiempo de ejecución.

Pydantic se emplea para los esquemas que interactúan con el LLM. Al llamar a .with_structured_output(SomeModel) se proporciona al LLM un esquema JSON al que debe ajustarse, lo que permite validar que la respuesta cumpla con los requisitos. Una respuesta que no coincida genera un error de inmediato y evita que los datos mal formados circulen por el grafo.

El flujo de trabajo final sigue exactamente esta división: EmailState es un TypedDict que funciona como contenedor del grafo, mientras que EmailClassification es un modelo de Pydantic que describe la salida del LLM.

Escribiendo el primer nodo

Cada nodo toma el estado completo como entrada y devuelve un diccionario con la actualización parcial. LangGraph integra esa actualización en el estado y aplica los reductores que se hayan declarado:

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Análisis del código:

• StateGraph(ChatState) crea un grafo que utiliza ChatState como su esquema, y LangGraph lee las anotaciones para conocer los reductores y los tipos de campos.
• graph.add_node("chatbot", chatbot) registra la función con el nombre "chatbot", que es al que se refieren las definiciones de enlaces.
• graph.add_edge(START, "chatbot") hace que "chatbot" sea el primer nodo en ejecutarse. START es un sentinela integrado.
• graph.add_edge("chatbot", END) finaliza la ejecución una vez que chatbot ha terminado. END es otro sentinela integrado.
• graph.compile() valida la estructura, verificando la existencia de enlaces sueltos y nodos inalcanzables, y devuelve un CompiledGraph listo para ejecutarse.
• app.invoke({"messages": [...]}) ejecuta el grafo de forma síncrona y devuelve el estado final.

El nodo envuelve su cadena en una lista ya que messages utiliza operator.add como reductor. Devolver {"messages": [response.content]} agrega una lista con un único elemento a la lista existente, mientras que devolver la cadena sin más causaría un error en tiempo de ejecución porque operator.add no puede añadir una cadena a una lista.

La regla consiste en que lo que devuelve un nodo debe coincidir con el tipo del campo, incluido el reductor. Un campo list[str] con operator.add acepta un list[str], y un campo str simple sin reductor acepta un str.

Esa es una aplicación LangGraph completa y funcional en aproximadamente 20 líneas. La siguiente pregunta es qué sucede dentro de invoke().

Pasos superiores: cómo funciona la ejecución

LangGraph ejecuta los nodos en pasos superiores. Cada nodo programado para el mismo paso se ejecuta de forma concurrente, y todos deben finalizar antes de que comience el paso siguiente. El ejemplo de un solo nodo tiene un paso superior que contiene un único nodo. Al agregar aristas paralelas, varios nodos se incluyen en el mismo paso superior.

Esto es importante por dos razones. La ejecución paralela está integrada en el modelo desde el principio y no se añade posteriormente. Además, con un checkpointer, LangGraph guarda una captura de estado completa después de que finalice cada paso superior. Eso es lo que hace posible la recuperación en caso de fallo: si el proceso se detiene a mitad de camino, la ejecución se reanuda desde el último paso superior completado.

Una vez comprendido cómo fluye el estado a través de un único nodo, la siguiente pregunta es cómo controlar el orden o ejecutar nodos al mismo tiempo. Esa es la función de las aristas.

Aristas: seriales, paralelas y condicionales

Los bordes controlan cómo se mueve la ejecución entre nodos. LangGraph ofrece tres tipos, y entre todos ellos abarcan cada patrón de enrutamiento que pueda surgir. La forma más simple ya apareció en el ejemplo del chatbot.

Bordes en serie

El caso más simple es uno nodo seguido del siguiente.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft se ejecuta primero. Una vez que termina y LangGraph fusiona el campo draft en el estado, se ejecuta review_draft y puede leer el borrador desde state["draft"]. El orden secuencial está garantizado, ya que el enlace impone la dependencia, y no se necesita código de secuenciación. Los enlaces secuenciales son adecuados siempre que el nodo B dependa de la salida del nodo A: generar primero y luego revisar, obtener primero y luego procesar, clasificar primero y luego enrutar.

Los ciclos accidentales son una trampa común. Por defecto, LangGraph espera un grafo dirigido acíclico, por lo que agregar un enlace desde review de vuelta a generate genera un error en tiempo de compilación, a menos que el grafo esté configurado explícitamente para ser cíclico. Los bucles iterativos suelen construirse con enlaces condicionales que pueden enrutar de vuelta a un nodo anterior.

Enlaces paralelos

Cuando dos aristas se extienden desde un mismo nodo, sus destinos se procesan simultáneamente en el mismo paso superior. Los reducidores se vuelven esenciales en este caso:

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Dos aristas desde START programan research y outline en el mismo paso superior, y LangGraph los ejecuta de forma concurrente. Ambos también tienen una arista hacia combine, y LangGraph no programará combine hasta que todos los nodos que apuntan a él hayan finalizado, lo cual funciona como una barrera de sincronización automática.

findings cuenta con un reductor aunque aquí solo research escribe en él, porque un sistema real podría tener cuatro agentes de investigación en paralelo que añadan elementos a la misma lista, y sin el reductor solo sobreviviría un resultado.

La matemática de la latencia es sencilla. Si research tarda 2 segundos y outline tarda 3, la versión secuencial necesita 5 segundos mientras que la versión paralela necesita 3, que es la duración de la tarea más larga. En el caso de llamadas independientes a LLM, la ejecución paralela permite ahorros de latencia prácticamente nulos.

Los candidatos para la paralelización son todas las parejas de nodos que no dependen de la salida del otro. Incluso mover dos llamadas de un segundo cada una al mismo paso superior suma efectos en miles de solicitudes. Preste atención a los nodos que parecen independientes pero comparten estado de maneras menos obvias. Si dos nodos en paralelo escriben en un mismo campo sin un reductor, se produce una condición de carrera, donde el último en terminar decide el valor, y LangGraph no emite ninguna advertencia.

Este patrón de difusión y recepción de datos es la base del proyecto final relacionado con correos electrónicos, el cual realiza una búsqueda en la base de conocimientos y una consulta en el sistema de seguimiento de errores en paralelo antes de redactar una respuesta.

Aristas condicionales

Cuando el siguiente nodo depende de lo que produjo el actual, se aplica add_conditional_edges:

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

La función route_by_sentiment analiza el estado y devuelve una clave de tipo cadena, y el diccionario de mapeo proporcionado a add_conditional_edges convierte esa clave en un nombre de nodo. Una función de enrutamiento que permanece pura (estado de entrada, cadena de salida, sin efectos secundarios) es fácil de probar, ya que se puede probar de forma unitaria sin necesidad de utilizar un LLM.

add_conditional_edges acepta tres argumentos: el nodo de origen, la función de enrutamiento y un diccionario que relaciona las claves de tipo cadena con los nombres de los nodos de destino. Cada clave que pueda devolver la función de enrutamiento debe estar presente en ese diccionario; de lo contrario, LangGraph genera un error en tiempo de ejecución.

La alternativa de Command

Cuando un nodo necesita actualizar su estado y enrutarse en el mismo paso, puede devolver directamente un Command y omitir add_conditional_edges:

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

Un valor de retorno de Command contiene tanto una actualización de estado (update=) como una elección de enrutamiento (goto=), de modo que el nodo escribe datos y elige el siguiente destino en un solo paso.

add_conditional_edges es la opción adecuada cuando la lógica de enrutamiento solo lee el estado y no escribe nada nuevo, cuando el enrutamiento debe mantenerse separado de la lógica del nodo, o cuando varios nodos de origen comparten la misma función de enrutamiento.

Command es la opción adecuada cuando el nodo genera nuevos datos que influyen en su propio enrutamiento, cuando un caso aislado no justifica una función de enrutamiento separada, o cuando el destino depende de algo calculado dentro del nodo que aún no forma parte del estado.

Para la etapa de revisión humana, el mecanismo principal de correo electrónico depende de add_conditional_edges, ya que el enrutamiento en este caso es simple: solo verifica la prioridad y la confianza ya almacenadas en el estado.

Memoria y puntos de control

Con el ordenamiento y las ramificaciones bajo control, el siguiente problema es la persistencia: ¿qué sucede cuando el proceso se reinicia a mitad de una conversación?

Hasta ahora, cada llamada comenzaba desde cero. Las conversaciones y los flujos de trabajo de múltiples pasos necesitan un estado que sobreviva entre las llamadas. LangGraph maneja esto mediante puntos de control, y la API es lo suficientemente sencilla como para que el cambio para la persistencia en entorno de producción sea de una sola línea.

MemorySaver para el desarrollo

MemorySaver mantiene el estado en memoria, lo cual es adecuado para el desarrollo y las pruebas:

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

Con checkpointer=memory, el grafo compilado almacena una captura completa después de cada paso superior. Una nueva llamada a invoke() utilizando el mismo thread_id hace que LangGraph cargue primero el punto de control más reciente para ese hilo. El thread_id funciona como un ID de sesión: úsalo nuevamente para continuar o cámbialo para comenzar desde cero.

La segunda llamada puede omitir los mensajes anteriores porque el checkpointer ya los contiene. Si Alice llama con {"messages": ["What should I learn first?"]}, LangGraph carga el punto de control "user-alice", encuentra la lista almacenada y agrega la nueva entrada mediante el reductor operator.add. Una vez que se ejecuta chatbot, state["messages"] contiene toda la historia.

Reutilizar un thread_id para conversaciones no relacionadas es un error. Si "user-alice" se reutiliza para un tema diferente, el chatbot recibe todo el historial anterior de conversaciones de Alice como contexto, por lo que cada nueva sesión debe contar con un thread_id nuevo. Durante el desarrollo, app.get_state_history(config) muestra el historial completo de puntos de control de un hilo, y el mismo mecanismo se utiliza para la depuración con viajes en el tiempo en LangSmith.

Puntos de control duraderos para producción

MemorySaver permanece en la memoria del proceso, por lo que cuando este termina, el estado también se pierde. La producción requiere un backend duradero:

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() crea las tablas de base de datos necesarias la primera vez que se ejecuta y no realiza ninguna acción posteriormente, por lo que solo es necesario llamarlo una vez durante la inicialización. La API no cambia entre diferentes backends, ya que los checkpointers operan detrás de una abstracción compartida. MemorySaver, PostgresSaver y RedisSaver siguen el mismo protocolo, por lo que el código del grafo no sabe con cuál de ellos está comunicándose.

Elegir entre ellos:

• MemorySaver es adecuado para el desarrollo, las pruebas y las demostraciones, donde no es importante que el sistema sobreviva a un reinicio.
• PostgresSaver es adecuado para flujos de trabajo en producción que requieren durabilidad y la posibilidad de consultar el historial de puntos de control.
• RedisSaver, de langgraph-redis, es adecuado para casos de alto rendimiento que necesitan lecturas rápidas de puntos de control y pueden aceptar las limitaciones en términos de persistencia de Redis.

Qué se guarda

Se almacena una captura completa en cada paso superior. Esto permite el viaje en el tiempo: se puede inspeccionar el estado en cualquier punto de la historia, reproducirlo a partir de un punto de control elegido o depurarlo para determinar con exactitud dónde ocurrieron los errores. El mismo mecanismo sustenta las interrupciones. Cuando un nodo llama a interrupt(), LangGraph necesita un lugar donde guardar el estado pausado para poder reanudarlo más tarde. Sin un punto de control, nada puede retenerlo, y el patrón de interrupciones se desintegra.

El ser humano en el proceso con interrupciones

Algunas decisiones son demasiado importantes como para automatizarlas. Antes de que un agente envíe un correo electrónico al cliente, apruebe un reembolso o implemente código, una persona debe revisarlo y aprobarlo. LangGraph gestiona esto mediante la función interrupt().

La función interrupt()

Al colocar interrupt() dentro de un nodo se detiene la ejecución, se escribe el estado en el checkpointer y se devuelve una carga útil al llamante:

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

En orden, la llamada realiza cuatro acciones:

1. Detiene la ejecución en la línea exacta donde se realiza la llamada.
2. Guarda el estado completo en el checkpointer, por lo que este es necesario.
3. Devuelve el payload que se pasó, en este caso el texto preliminar, a quien llamó a invoke().
4. Espera, dejando el grafo en estado suspendido.

La variable decision no recibe nada al principio, ya que interrupt() no devuelve valor hasta que el grafo se reanuda con Command(resume=value). En ese momento, decision recibe lo que se pasó como parámetro en resume=.

El payload indica al revisor humano qué está viendo, por lo que puede contener todo lo necesario: el borrador, un resumen o un diccionario estructurado con contexto. Quien realiza la llamada lo recibe como parte de la notificación de interrupción.

Reanudar con Command

Continuar después de una interrupción implica invocar el gráfico nuevamente con Command(resume=value):

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

En el primer paso, LangGraph ejecuta generate, almacena el borrador en el estado y llega a interrupt() en el nodo review. Realiza un punto de control del estado y devuelve el resultado; el result de esta primera llamada a invoke() contiene la información de interrupción, es decir, el borrador para el revisor. En el segundo paso, Command(resume="approve") hace que LangGraph cargue el punto de control correspondiente a "email-review-1", continúe desde la llamada a interrupt() y devuelva "approve" como valor de retorno. La decision del nodo review pasa a ser "approve"; se establece approved=True y la ejecución continúa hacia send.

Se debe utilizar el mismo config en ambas llamadas, ya que LangGraph encuentra el punto de control adecuado a través de thread_id. Un thread_id diferente en la segunda llamada apuntaría a un punto de control que no existe, y la llamada fallaría.

Un checkpointer es obligatorio para las interrupciones. Sin él, el estado pausado no tiene a dónde ir, por lo que app.compile(checkpointer=memory) no es opcional en este caso.

Desde LangGraph 1.1.0, invoke() acepta version="v2" y devuelve un GraphOutput con los atributos .value y .interrupts, lo que permite un manejo de interrupciones más seguro desde el punto de vista tipológico, algo que vale la pena probar una vez que se conozcan los conceptos básicos. La carga útil de una interrupción puede ser cualquier valor serializable, ya sea una cadena, un diccionario o una lista. Las configuraciones en entornos de producción suelen pasar un diccionario estructurado que contiene todo lo que necesita un revisor, como el borrador, la clasificación, el remitente y la URL de revisión, información que la capa de aplicación puede mostrar en una interfaz de usuario o enviarla a través de Slack o correo electrónico.

Hay otro problema. Llamar nuevamente a invoke() con el estado inicial completo en lugar de Command(resume=value) es un error: app.invoke({"messages": [...]}, config=config) en el mismo thread_id reinicia el grafo de ese hilo y descarta el estado en pausa. Un grafo interrumpido solo debe continuarse utilizando Command(resume=value).

Proyecto final: un flujo de trabajo de triaje de correos electrónicos

Con todas las piezas en su lugar, se pueden combinar para crear un sistema de triaje de correos electrónicos en formato de producción. Este sistema lee un correo electrónico entrante, clasifica su intención mediante una salida estructurada, busca simultáneamente en una base de conocimientos y en un sistema de seguimiento de errores, redacta una respuesta y puede detenerse para revisión humana antes de enviarla.

Cada concepto anterior se corresponde con una parte de este flujo de trabajo:

• El estado de TypedDict con reducidos se muestra como EmailState, con operator.add en search_results.
• Los bordes en serie conectan read_email con classify_intent.
• Los bordes paralelos permiten que classify_intent se divida en search_docs y bug_tracking.
• La sincronización por entrada en abanico significa que ambos nodos paralelos alimentan write_response.
• El enrutamiento condicional es route_by_priority, que elige entre revisión humana y envío automático.
• La salida estructurada con Pydantic es EmailClassification, la cual restringe la respuesta del LLM.
• La interrupción es human_review, que pausa el proceso a la espera de aprobación.
• El checkpointer es MemorySaver, el cual habilita el ciclo de interrupción y reanudación.

Estado y esquemas

Primero vienen el estado y el esquema Pydantic para la clasificación estructurada:

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification es un modelo de Pydantic y no forma parte de EmailState, que representa la separación entre TypedDict y Pydantic en acción. Es la estructura que se entrega al LLM como objetivo de esquema JSON, y Pydantic valida la respuesta y lanza un error de inmediato si no coincide. EmailState es el contenedor del grafo, un TypedDict porque LangGraph necesita la sintaxis de reductor y una serialización eficiente.

classification se almacena como un dict en EmailState porque, después de que Pydantic valide la respuesta, .model_dump() la convierte en un dicta simple antes de que sea guardada en el estado. Esto permite que el estado sea serializable por cualquier checkpointer, sin necesidad de depender de Pydantic al restaurarlo.

El campo más importante es search_results: Annotated[list[str], operator.add]. Dado que search_docs y bug_tracking se ejecutan en paralelo y ambos pueden escribir resultados, el reductor acumula sus salidas juntas; sin él, una rama sobrescribiría silenciosamente a la otra.

Construcción de los nodos

Los nodos se mantienen pequeños y con un único propósito. Cada uno obtiene solo los valores que necesita del estado, realiza su tarea y devuelve una actualización parcial:

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

A primera vista, read_email parece inútil, ya que solo devuelve campos que ya están presentes en el estado, pero funciona como un paso explícito de análisis. Una versión para producción tomaría un payload bruto de webhook y extraería los campos, por lo que la función aquí sirve como sustituto de esa extracción.

classify_intent es donde Pydantic resulta útil. En structured_llm = llm.with_structured_output(EmailClassification), el LLM se envuelve de tal manera que las respuestas deben cumplir con el esquema. Recibe un esquema JSON derivado de EmailClassification y debe devolver JSON válido que se ajuste a él; LangChain vuelve a intentarlo automáticamente antes de generar un error si no lo hace. Luego, .model_dump() convierte el objeto validado en un diccionario simple para el estado.

Los dos nodos paralelos escriben en campos separados, search_results y bug_info, de modo que sus llamadas al LLM se ejecutan al mismo tiempo. Para search_results, search_docs devuelve una lista ([f"[Docs] {response.content}"]), ya que operator.add necesita entradas en forma de lista. bug_tracking devuelve una cadena simple para bug_info, el cual no cuenta con un reductor, por lo que basta con una asignación normal.

El nodo de entrada es write_response. Cuando se activa, ambos nodos paralelos ya han finalizado y sus resultados se encuentran en el estado, por lo que este construye el prompt a partir de todo ello y redacta la respuesta.

human_review pasa un diccionario estructurado a interrupt() y no una cadena simple, lo cual constituye una buena práctica en entornos de producción. El diccionario contiene el borrador, la clasificación y un mensaje para el revisor, de modo que cualquier sistema que se encuentre al otro lado, ya sea una interfaz web, un bot de Slack o una notificación por correo electrónico, recibe todo el contexto.

send_reply utiliza state.get("approved", True) como valor por defecto defensivo; así, si la clave approved falta de alguna manera en el estado, envía la respuesta de todos modos. Este valor por defecto puede cambiarse a False cuando el perfil de riesgo requiere mayor cautela.

En producción, search_docs y bug_tracking llamarían a sistemas reales como una base de datos vectorial, Jira o GitHub Issues, y el decorador @tool de LangChain facilita el encapsulado de esas llamadas. La estructura de los nodos se mantiene igual, con una llamada a una herramienta que reemplaza a la llamada al LLM. Otra trampa: write_response lee state["search_results"]; por lo tanto, si esa lista sigue estando vacía, el acceso puede fallar, lo cual se evita inicializando search_results: [] en el payload de invocación inicial para que el reductor comience a añadir elementos a partir de allí.

Conexión de los enlaces

Los enlaces en secuencia, en paralelo y condicionales conectan entonces los nodos:

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority envía los correos electrónicos de alta prioridad a revisión humana independientemente del nivel de confianza. Las clasificaciones con un nivel de confianza inferior a 0.8 también pasan a revisión, ya que si el modelo no está seguro de qué trata un correo, debe ser una persona quien lo decida. Todo lo demás se envía automáticamente, lo cual constituye un valor por defecto razonable en entornos de producción que se debe ajustar según las tasas reales de error.

Este enrutamiento se realiza a través de add_conditional_edges y no de Command, porque route_by_priority es una función pura: solo inspecciona el estado y devuelve una clave, sin crear nuevos datos, que es exactamente para lo que está diseñado add_conditional_edges.

A continuación se muestran todas las conexiones en el grafo:

• Un proceso en secuencia parte de START, pasa por read_email y llega a classify_intent.
• La salida desde classify_intent inicia simultáneamente tanto search_docs como bug_tracking.
• La entrada envía los resultados de esos dos nodos a write_response.
• Después de write_response, un enlace condicional elige entre human_review o send_reply.
• Una vez finalizada la revisión, human_review pasa la información a send_reply, que termina en END.

Ejecutar el flujo de trabajo de principio a fin

Primero se ingresa un correo electrónico de ejemplo, y la primera ejecución continúa hasta que se produce la interrupción:

config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

Para entonces, el grafo ya había realizado mucha trabajo:

1. read_email analizó el mensaje.
2. classify_intent lo clasificó como un informe de error de alta prioridad utilizando una salida estructurada de Pydantic.
3. La búsqueda en la documentación y la consulta en el sistema de seguimiento de errores se ejecutaron en paralelo dentro de un mismo paso.
4. write_response redactó una respuesta que utilizó todo ese contexto.
5. El nodo human_review llamó a interrupt() y puso en pausa el proceso.

Toda la cadena de procesamiento, incluidas las llamadas paralelas al LLM, finalizó antes de la pausa. Cuando el grafo se detiene, el estado se guarda por completo, de modo que el revisor puede tardar minutos u horas en responder sin perder nada.

Luego el revisor aprueba:

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph cargó el punto de control para "email-001", reanudó human_review a partir de su llamada a interrupt(), devolvió "approve" como valor de retorno, estableció approved=True en el estado y continuó con send_reply, el cual devolvió el estado de confirmación.

Toda idea anterior aparece en la solución final: estado TypedDict respaldado por un reductor, aristas seriales y paralelas, enrutamiento condicional, salida estructurada con Pydantic, revisión humana mediante una interrupción y memoria conservada por un punto de control.

Para un correo de baja prioridad con alta confianza, route_by_priority devuelve "auto_send" y se omite human_review. El proceso se ejecuta de forma continua, leyendo, clasificando, buscando en paralelo, redactando y enviando el correo, y la interrupción solo existe en el camino que la necesita. Ese es el beneficio del enrutamiento condicional.

Una advertencia final concierne al estado inicial. TypedDict no establece valores por defecto para los campos faltantes, a diferencia de Pydantic. Si write_response intenta acceder a state["search_results"] y la clave no estaba presente en el estado inicial, se produce un KeyError. Cada campo debe formar parte del diccionario de estado inicial, incluso si comienza como una cadena vacía, una lista vacía o False.

¿A dónde ir a continuación?

La ruta que se sigue aquí comienza con un único nodo y un TypedDict, y termina con un flujo de trabajo de tipo producción que permite ejecutar llamadas a modelos de lenguaje en paralelo, validar la salida estructurada, tomar decisiones condicionales, superar fallos mediante puntos de control y esperar a un revisor. Desde allí hay varias direcciones posibles:

• LangGraph Platform se encarga del despliegue, del control de puntos de control y de la escalabilidad horizontal, lo que elimina la carga de gestionar la infraestructura cuando el objetivo es manejar tráfico real.
• LangSmith rastrea y depura cada ejecución de grafo y muestra el estado exacto en cada paso cuando un nodo no funciona correctamente. Esto se vuelve importante cuando los grafos cuentan con cinco o más nodos y rutas condicionales.
• Los subgrafos combinan grafos más pequeños en sistemas multiagente más grandes. Cada uno gestiona su propio estado y comunica con el grafo padre, lo cual representa un paso natural después de los flujos de trabajo basados en un único grafo.
• La herramienta que se llama a través de LangChain @tool y ToolNode permite a los agentes utilizar APIs, bases de datos y servicios externos desde dentro del grafo, y los nodos clave search_docs y bug_tracking son lugares obvios para probarlo.
• La persistencia en producción implica instalar langgraph-checkpoint-postgres para mantener un estado duradero que sobreviva a los reinicios, y el intercambio de API se realiza con una sola línea.

Resumen rápido

• El estado es un TypedDict. Cualquier campo tipado como Annotated[list, operator.add] recopila valores a lo largo de las escrituras, mientras que un campo sin reductor se reemplaza en cada escritura.
• Un nodo es simplemente una función: recibe el estado completo y devuelve un diccionario de actualización parcial, siendo LangGraph quien realiza la fusión.
• Los bordes determinan el flujo. Utilice bordes en serie cuando el orden sea importante, bordes paralelos para tareas independientes que se ejecutan simultáneamente, y bordes condicionales cuando los datos deban elegir la rama correspondiente.
• Use TypedDict para el estado del grafo, y emplee Pydantic con .with_structured_output() para definir con precisión lo que devuelve el LLM.
• Los mecanismos de punto de control proporcionan persistencia y permiten las interrupciones: MemorySaver durante el desarrollo, PostgresSaver en producción. Sin punto de control, no hay interrupciones.
• Una interrupción no es más que un estado guardado y pausado. interrupt(payload) detiene y guarda el estado en un punto de control, y Command(resume=value) reinicia desde ese mismo punto exacto.
• El paralelismo está integrado por defecto: dos bordes que salen de un nodo ejecutan sus destinos al mismo tiempo, y cada campo que escribe ese nodo requiere un reductor.

Preguntas para reflexionar

1. MemorySaver es adecuado para el desarrollo y PostgresSaver para producción. ¿Qué otros backends de almacenamiento merecen ser considerados para un sistema de alto rendimiento, y qué compromisos determinarían la elección?
2. route_by_priority envía los correos de alta prioridad para revisión y envía automáticamente el resto. ¿Cómo se podría ajustar ese umbral en un sistema real, y qué métricas indicarían si está configurado correctamente?
3. La función final dirige el flujo después de write_response mediante add_conditional_edges. ¿Existe algún escenario en el flujo de trabajo de correos en el que el patrón Command sería más adecuado que una función de enrutamiento separada?

Leer más

• Notas prácticas: Construir un agente de IA autocorregible con Self-RAG y LangGraph — Guía paso a paso de las Notas prácticas: Construir un agente de IA autocorregible con Self-RAG y LangGraph: contratos, verificaciones y espacios para código listo para usar para los equipos que implementan este patrón.
• Construir un agente de investigación bursátil con ReAct y LangGraph — Un agente LangGraph ReAct toma un símbolo bursátil, consulta datos fundamentales, noticias, indicadores técnicos y herramientas de analistas en su propio orden, y redacta un resumen breve, con pausas para la intervención humana.

Un proceso de trabajo local de agente LLM creado a partir de carpetas y archivos Markdown — Construya una tubería de agente local con múltiples etapas y control de acceso, donde cada etapa es una carpeta, los prompts están en formato Markdown editable, y un humano revisa la salida antes del paso costoso.

Notas prácticas: Agente texto a SQL en Python: Tutorial de llamadas a herramientas LLM — Guía paso a paso de las Notas prácticas: Agente texto a SQL en Python: Tutorial de llamadas a herramientas LLM: contratos, verificaciones y espacios de código listos para usar para los equipos que implementan este patrón.

Notas prácticas: ¿Por qué nadie habla de LangChain, LangGraph o AutoAgent? — Guía paso a paso de las Notas prácticas: ¿Por qué nadie habla de LangChain, LangGraph o AutoAgent?: contratos, verificaciones y espacios de código listos para usar para los equipos que implementan este patrón.

Equipo multiagente autoreparable en LangGraph: Arquitecto, Tester, Depurador — Crea una máquina de estados en LangGraph con Python en la que un Arquitecto, un Desarrollador, un entorno de pruebas pytest, un Depurador y un Revisor colaboran en un ciclo hasta que las pruebas se aprueben.

Notas prácticas: Creación de un bucle de revisión de código con dos agentes utilizando A2A, LangGraph y — Guía paso a paso de las Notas prácticas: Creación de un bucle de revisión de código con dos agentes utilizando A2A, LangGraph y: contratos, verificaciones y espacios para código listos para uso para los equipos que implementan este patrón.