LangGraph

Cómo funciona

Los nodos leen y actualizan un estado compartido; las conexiones deciden el siguiente paso. Los checkpoints conservan el progreso para reanudar o introducir una revisión humana. Puede coordinar un agente o varios.

Ejemplo: un grafo mínimo

Con langgraph instalado:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class Estado(TypedDict):
    consulta: str
    resultado: str

def preparar(estado: Estado):
    return {"resultado": estado["consulta"].strip()}

grafo = StateGraph(Estado)
grafo.add_node("preparar", preparar)
grafo.add_edge(START, "preparar")
grafo.add_edge("preparar", END)
app = grafo.compile()

Este ejemplo solo muestra la estructura; persistencia, modelos y herramientas se añaden de forma explícita.

Estado con reducers

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class Estado(TypedDict):
    messages: Annotated[list, add_messages]   # los mensajes se acumulan, no se sustituyen
    categoria: str | None
    aprobado: bool

Un reducer define cómo se combina la actualización de un nodo con el valor anterior. Sin reducer, el valor se reemplaza.

Rutas condicionales e intervención humana

from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def clasificar(estado: Estado):
    categoria = modelo_clasificador.invoke(estado["messages"])
    return {"categoria": categoria}

def ruta(estado: Estado) -> str:
    return "revision" if estado["categoria"] == "condonacion" else "responder"

def revision(estado: Estado):
    decision = interrupt({"pregunta": "¿Aprobar la condonación propuesta?", "caso": estado["messages"][-1]})
    return {"aprobado": decision == "si"}

def responder(estado: Estado):
    return {"messages": [modelo.invoke(estado["messages"])]}

g = StateGraph(Estado)
g.add_node("clasificar", clasificar)
g.add_node("revision", revision)
g.add_node("responder", responder)
g.add_edge(START, "clasificar")
g.add_conditional_edges("clasificar", ruta, ["revision", "responder"])
g.add_edge("revision", "responder")
g.add_edge("responder", END)

app = g.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "cliente-42"}}

app.invoke({"messages": [("user", "Quiero que me condonen los intereses")]}, config)
# … la ejecución queda pausada en `revision`; un supervisor responde más tarde:
app.invoke(Command(resume="si"), config)

En producción, el checkpointer en memoria se sustituye por uno persistente (PostgreSQL, Redis) para que la pausa sobreviva a reinicios y pueda durar horas o días.

Capacidades clave

Capacidad	Uso
Checkpointers	Persistencia del estado por hilo; reanudar, depurar o volver atrás (time travel)
interrupt	Aprobación humana antes de acciones sensibles
Streaming	stream_mode="updates", "values" o "messages" (tokens)
Send	Paralelizar (map-reduce) sobre una lista de elementos
Subgrafos	Componer agentes especializados
Agentes prediseñados	Agente ReAct con herramientas en una línea, como punto de partida
Memoria a largo plazo	Store para datos entre hilos (preferencias del usuario)

Cuándo usar un grafo explícito

Situación	Lectura
Flujo con pasos y decisiones conocidos, auditables	Grafo con nodos deterministas y LLM puntuales
Tarea abierta en la que el modelo decide los pasos	Agente ReAct (bucle modelo-herramientas)
Proceso de negocio largo con plazos y compensaciones	Combinar con un motor de workflows o una saga

Traza las ejecuciones con LangSmith o Langfuse y conecta la interfaz con AG-UI. Para un enfoque más ligero y tipado, compara con Pydantic AI.

Criterios de implementación

Haz idempotentes los pasos con efectos externos y delimita reintentos. Para recuperar conversaciones, configura un checkpointer y un identificador de hilo. Controlar el grafo no convierte la respuesta del modelo en determinista.