Phase 11 - Lesson 16

LangGraph — Máquinas de Estado para Agentes

Un bucle ReAct escrito a mano es un while True. Un bucle ReAct escrito en LangGraph es un grafo por el cual puedes crear puntos de control, interrumpir, ramificar y viajar en el tiempo. El agente no ha cambiado. El entorno que lo rodea sí.

Tipo: Construcción Lenguajes: Python Prerrequisitos: Fase 11 · 09 (Function Calling), Fase 11 · 14 (Model Context Protocol) Tiempo: ~75 minutos

El Problema

Lanzas a producción un agente de llamada de funciones. Funciona durante tres turnos, luego algo sale mal: el modelo intenta usar una herramienta que devuelve un error 500, el usuario cambia de opinión a mitad de la tarea o el agente decide reembolsar un pedido sin la autorización de un humano. El bucle while True: no tiene ganchos. No puedes pausarlo, no puedes retrocederlo y no puedes ramificarlo para explorar "qué pasaría si el modelo hubiera elegido la otra herramienta". En el momento en que lanzas esto más allá de una demostración, el agente se convierte en una caja negra que funcionó o no funcionó.

El siguiente paso es obvio una vez que lo ves. El agente ya es una máquina de estados: prompt del sistema más historial de mensajes más llamadas a herramientas pendientes más la siguiente acción. Haz que la máquina de estados sea explícita: nodos para "el modelo piensa", "una herramienta se ejecuta", "un humano aprueba" y aristas para las transiciones condicionales entre ellos. Una vez que el grafo es explícito, el entorno obtiene cuatro cosas gratis: checkpointing (guardar el estado entre pasos), interrupts (pausar para un humano), streaming (transmitir tokens y eventos intermedios) y time-travel (retroceder a un estado anterior y probar una rama diferente).

LangGraph es la biblioteca que ofrece esta abstracción. No es un framework de agentes en el sentido de LangChain ("aquí tienes un AgentExecutor, buena suerte"). Es un runtime de grafos con estado de primera clase, persistencia de primera clase e interrupciones de primera clase. El bucle del agente es algo que dibujas, no algo que escribes a mano.

El Concepto

Un StateGraph tiene tres cosas.

1. State (Estado). Un diccionario tipado (TypedDict o modelo Pydantic) que fluye a través del grafo. Cada nodo recibe el estado completo y devuelve una actualización parcial, que LangGraph fusiona usando un reducer por campo: operator.add para listas que deben acumularse, y sobrescritura por defecto.
2. Nodes (Nodos). Funciones de Python state -> partial_state. Cada uno es un paso discreto: "llamar al modelo", "ejecutar herramientas", "resumir".
3. Edges (Aristas). Transiciones entre nodos. Las aristas estáticas van a un solo lugar. Las aristas condicionales toman una función de enrutador state -> next_node_name para que el grafo pueda ramificarse según la salida del modelo.

Compilas el grafo. La compilación vincula la topología, adjunta un checkpointer (opcional pero esencial para producción) y devuelve un ejecutable (runnable). Lo invocas con un estado inicial y un thread_id. Cada paso de la ejecución persiste un checkpoint indexado por (thread_id, checkpoint_id).

Los cuatro superpoderes

Checkpointing (Puntos de control). Cada transición de nodo escribe el nuevo estado en un almacenamiento (en memoria para pruebas, Postgres/Redis/SQLite para producción). Se reanuda llamando al grafo nuevamente con el mismo thread_id. El grafo continúa desde donde se pausó.

Interrupts (Interrupciones). Marca un nodo con interrupt_before=["human_review"] y la ejecución se detendrá antes de que se ejecute ese nodo. El estado persiste. Tu API responde al usuario con "esperando aprobación". Una solicitud posterior al mismo thread_id con Command(resume=...) reanuda la ejecución.

Streaming. graph.stream(state, mode="updates") produce deltas de estado a medida que ocurren. mode="messages" transmite los tokens del LLM dentro de los nodos del modelo. mode="values" produce instantáneas (snapshots) completas. Tú eliges qué mostrar en tu interfaz de usuario.

Time-travel (Viaje en el tiempo). graph.get_state_history(thread_id) devuelve el registro completo de checkpoints. Pasa cualquier checkpoint_id anterior a graph.invoke y realizarás una bifurcación desde ese punto. Excelente para depuración ("¿qué pasaría si el modelo hubiera elegido la herramienta B?") y para pruebas de regresión que reproducen ejecuciones de producción.

El punto clave son los reducers

Cada campo del estado tiene un reducer. La mayoría de los valores por defecto son suficientes: un nuevo valor sobrescribe al anterior. Pero las listas de mensajes necesitan operator.add para que las nuevas mensajes se añadan en lugar de reemplazarse. Las aristas paralelas fusionan sus actualizaciones a través del reducer. Si dos nodos actualizan messages y olvidas Annotated[list, add_messages], el segundo ganará silenciosamente y perderás la mitad del turno. El reducer es el único detalle sutil de la biblioteca; hazlo bien y el resto se compondrá solo.

El grafo ReAct en cuatro nodos

Un agente ReAct en producción consta de cuatro nodos y dos aristas:

1. agent — llama al LLM con el historial de mensajes actual. Devuelve el mensaje del asistente (que puede contener chamadas a herramientas, tool_calls).
2. tools — ejecuta cualquier tool_calls en el último mensaje del asistente y añade los resultados de las herramientas como mensajes de herramienta.
3. Una arista condicional desde agent que enruta a tools si el último mensaje tiene tool_calls, de lo contrario a END.
4. Una arista de retorno estática desde tools a agent.

Eso es todo. Obtienes el bucle ReAct completo (Pensamiento → Acción → Observación → Pensamiento → …) con checkpointing, interrupciones y streaming, en aproximadamente 40 líneas de código.

StateGraph vs Send (fanout)

Send(node_name, state) permite que un nodo despache subgrafos paralelos. Ejemplo: el agente decide consultar tres recuperadores (retrievers) a la vez. Cada Send genera una ejecución paralela del nodo objetivo; sus salidas se fusionan a través del reducer del estado. Así es como LangGraph expresa el patrón orquestador-trabajadores sin primitivas de hilos.

Subgrafos

Un grafo compilado puede ser un nodo en otro grafo. El grafo externo ve un solo nodo; el grafo interno tiene su propio estado y sus propios checkpoints. Así es como los equipos construyen agentes supervisor-trabajador: el grafo del supervisor enruta la intención del usuario a un subgrafo de trabajador por dominio.

Construcción

Paso 1: estado y nodos

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def agent_node(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

tool_node = ToolNode(tools=[search_web, read_file])

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile(checkpointer=MemorySaver())

add_messages es el reducer que hace que la lista de mensajes se acumule en lugar de sobrescribirse. Olvidarlo es el error más común en LangGraph.

Paso 2: ejecutar con un hilo (thread)

config = {"configurable": {"thread_id": "user-42"}}
for event in app.stream(
    {"messages": [HumanMessage("find the Anthropic headquarters address")]},
    config,
    stream_mode="updates",
):
    print(event)

Cada actualización es un diccionario {node_name: state_delta}. Tu frontend puede transmitir estos datos a la interfaz de usuario para que los usuarios vean "el agente está pensando... llamando a search_web... obtuvo el resultado... respondiendo".

Paso 3: agregar una interrupción human-in-the-loop

Marca un nodo para que la ejecución se pause antes de ejecutarse.

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # pause before every tool call
)

state = app.invoke({"messages": [HumanMessage("delete the production database")]}, config)
# state["__interrupt__"] is set. Inspect proposed tool calls.
# If approved:
from langgraph.types import Command
app.invoke(Command(resume=True), config)
# If denied: write a rejection message and resume
app.update_state(config, {"messages": [AIMessage("Blocked by human reviewer.")]})

El estado, el checkpoint y el hilo persisten a lo largo de la interrupción. Nada queda en memoria excepto durante la ejecución.

Paso 4: time-travel para depuración

history = list(app.get_state_history(config))
for snapshot in history:
    print(snapshot.values["messages"][-1].content[:80], snapshot.config)

# Fork from a prior checkpoint
target = history[3].config  # three steps back
for event in app.stream(None, target, stream_mode="values"):
    pass  # replay from that point forward

Pasar None como entrada reproduce la ejecución a partir del checkpoint proporcionado; pasar un valor lo añade como una actualización al estado de ese checkpoint antes de reanudar. Así es como reproduces una ejecución fallida de un agente sin tener que volver a ejecutar toda la conversación.

Paso 5: cambiar el checkpointer para producción

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

SQLite, Redis y Postgres están soportados de forma nativa. MemorySaver es para pruebas. Cualquier cosa que necesite persistir entre reinicios requiere un almacenamiento real.

La Habilidad

Construyes agentes como grafos, no como bucles while True.

Antes de recurrir a LangGraph, realiza un diseño rápido de 60 segundos:

1. Nombra los nodos. Cada decisión discreta o acción con efectos secundarios es un nodo. "El agente piensa", "la herramienta se ejecuta", "el revisor aprueba", "la respuesta se transmite". Si no puedes listarlos, la tarea aún no tiene forma de agente.
2. Declara el estado. Un TypedDict mínimo con un reducer para cada campo de lista. No metas todo dentro de messages; eleva los campos específicos de la tarea (un plan de trabajo plan, un contador de presupuesto budget, una lista de documentos recuperados retrieved_docs) al nivel superior.
3. Dibuja las aristas. Estáticas, a menos que el siguiente paso dependa de la salida del modelo. Cada arista condicional necesita una función de enrutador con ramas nombradas.
4. Elige un checkpointer desde el principio. MemorySaver para pruebas, Postgres/Redis/SQLite para cualquier otra cosa. No envíes a producción sin uno: no tener checkpointer significa no tener reanudación, interrupción ni viaje en el tiempo.
5. Decide las interrupciones antes de que se ejecuten las herramientas, no después. Las aprobaciones van en la arista que entra en un nodo con efectos secundarios para que puedas cancelar antes de causar daño; la validación va en la arista que sale del modelo para rechazar llamadas incorrectas de forma económica.
6. Transmite por streaming por defecto. mode="updates" para la interfaz de usuario, mode="messages" para la transmisión a nivel de tokens dentro de los nodos del modelo, mode="values" para instantáneas completas durante la evaluación.

Niégate a lanzar a producción un agente de LangGraph que no tenga checkpointer. Niégate a lanzar uno que interrumpa después del efecto secundario. Niégate a lanzar un campo messages sin add_messages como su reducer.

Ejercicios

1. Fácil. Implementa el grafo ReAct de cuatro nodos anterior con una herramienta de calculadora y una de búsqueda web. Verifica que list(app.get_state_history(config)) devuelva al menos cuatro checkpoints para una conversación de dos turnos.
2. Medio. Agrega un nodo planner que se ejecute antes de agent y escriba un plan estructurado plan: list[str] en el estado. Haz que agent marque los pasos del plan como completados. Falla la prueba si el plan (plan) se pierde tras reanudar desde un checkpoint (reducer incorrecto).
3. Difícil. Construye un grafo supervisor que enrute entre tres subgrafos (researcher, writer, reviewer) usando Send. Cada subgrafo tiene su propio estado y checkpointer. Agrega un interrupt_before=["writer"] en el grafo externo para que un humano pueda aprobar el informe de investigación. Confirma que el viaje en el tiempo desde un checkpoint anterior ejecute solo la rama bifurcada.

Términos Clave

Lecturas Adicionales

• LangGraph documentation — referencia canónica para StateGraph, reducers, checkpointers e interrupciones.
• LangGraph concepts: state, reducers, checkpointers — el modelo mental que utiliza esta lección, directo de la fuente.
• LangGraph Persistence and Checkpoints — detalles sobre almacenamientos en Postgres/SQLite/Redis, namespaces de checkpoints e IDs de hilos (thread).
• LangGraph Human-in-the-loop — interrupt_before, interrupt_after, Command(resume=...) y el patrón de edición de estado (edit-state).
• Yao et al., "ReAct: Synergizing Reasoning and Acting in Language Models" (ICLR 2023) — el patrón que implementa cada agente de LangGraph; léelo para comprender la lógica detrás de la traza de razonamiento.
• Anthropic — Building effective agents (Dec 2024) — qué formas de grafos (cadena, enrutador, orquestrador-trabajadores, evaluador-optimizado) preferir y cuándo.
• Fase 11 · 09 (Function Calling) — la primitiva de llamada a función que cada nodo de agente de LangGraph reutiliza.
• Fase 11 · 14 (Model Context Protocol) — descubrimiento de herramientas externas que se conecta a un ToolNode de LangGraph a través del adaptador MCP.
• Fase 11 · 17 (Agent framework tradeoffs) — cuándo elegir LangGraph sobre CrewAI, AutoGen o Agno.