Agentes de IA con estado en LangGraph

Visión general

LangGraph extiende LangChain con grafos con estado y cíclicos para construir workflows agénticos. En lugar de cadenas lineales, defines una máquina de estados: los nodos son funciones que leen y actualizan un estado compartido, las aristas son transiciones, y las aristas condicionales deciden el siguiente paso en tiempo de ejecución. Eso permite que un agente itere (planifica, llama una herramienta, mira el resultado, planifica de nuevo) en lugar de ejecutarse una vez y esperar lo mejor.

Esta receta construye un agente de investigación con tool calling, enrutamiento condicional, checkpointing y una puerta de aprobación humana. El mismo esqueleto sirve para asistentes de código, agentes de calidad de datos y cualquier workflow donde el modelo necesite varios turnos para converger en una respuesta.

Las piezas que sostienen el diseño:

• AgentState es el contrato que todo nodo lee y escribe. El reducer add_messages agrega mensajes nuevos en lugar de reemplazar la lista, así que el historial de conversación sobrevive entre nodos.
• should_continue es una función normal que inspecciona el estado y devuelve a dónde ir después. La topología del grafo queda como datos; el enrutamiento queda como código.
• El checkpointer (MemorySaver en desarrollo, SQLite/Postgres en producción) guarda una foto del estado tras cada nodo bajo un thread_id. Es lo que hace posible reanudar conversaciones y el human-in-the-loop.

Cuándo usarla

• Agentes multi-paso que necesitan iterar: buscar, evaluar, refinar, repetir
• Workflows con branching condicional (si está seguro, responde; si no, busca más)
• Flujos human-in-the-loop donde una persona aprueba las llamadas a herramientas antes de ejecutarlas
• Conversaciones con estado que deben persistir entre sesiones o reinicios
• Configuraciones multi-agente donde varios nodos especializados comparten un mismo estado

Cuándo no usarla

Usa algo más simple cuando:

• Una sola llamada al LLM con bind_tools() resuelve la petición, así que no hace falta un grafo para una function call de un solo paso.
• El pipeline es lineal: para flujos prompt → modelo → parser, las cadenas de LangChain con LCEL son más cortas y fáciles de razonar.
• Ejecutas trabajos por lotes (clasificar 10.000 textos): un bucle simple sobre model.invoke() gana a la orquestación de agentes, y las respuestas streaming cubren el lado UX si hace falta.
• El agente nunca necesita estado: los checkpointers escriben una foto por nodo, así que no pagues por persistencia que no vas a leer.

Requisitos previos

• Python 3.10+
• Paquetes langgraph, langchain-openai y langchain-core
• Una OPENAI_API_KEY para las llamadas al modelo en los ejemplos

Solución

1. Instalar dependencias

pip install langgraph langchain-openai langchain-core

2. Definir el estado del agente

import re
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    search_results: str
    attempts: int
    max_attempts: int

3. Definir herramientas

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Búsqueda simulada — reemplaza con Tavily, SerpAPI, etc.
    results = {
        "redis caching": "Redis cache-aside pattern: check cache, if miss, load from DB, set cache. TTL recommended.",
        "docker compose": "Docker Compose defines multi-container apps in YAML. Use profiles for environment-specific services.",
    }
    return results.get(query.lower(), f"No results found for: {query}")

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression (digits and operators only)."""
    if not re.fullmatch(r"[\d\s+\-*/().]+", expression):
        return "Error: expression must contain only numbers and operators"
    try:
        return f"Result: {eval(expression)}"
    except Exception as e:
        return f"Error: {e}"

tools = [search_web, calculate]

La puerta del regex importa: eval() sobre la salida cruda del modelo es un agujero de inyección de código. Poner en lista blanca dígitos y operadores deja la herramienta lo bastante segura para un ejemplo. Un parser más estricto (ast.literal_eval o un pequeño evaluador de expresiones) es lo correcto en producción.

4. Definir nodos del grafo

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    """Main agent node — decides what to do next."""
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    """Execute tool calls from the agent's last message."""
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_id = tool_call["id"]

        # Encontrar y ejecutar la herramienta
        tool_map = {t.name: t for t in tools}
        if tool_name in tool_map:
            try:
                result = tool_map[tool_name].invoke(tool_args)
            except Exception as e:
                result = f"Tool {tool_name} failed: {e}"
            tool_messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_id,
            ))

    return {"messages": tool_messages, "attempts": state.get("attempts", 0) + 1}

def should_continue(state: AgentState) -> str:
    """Conditional edge — decide whether to use tools or end."""
    last_message = state["messages"][-1]
    attempts = state.get("attempts", 0)
    max_attempts = state.get("max_attempts", 5)

    if last_message.tool_calls and attempts < max_attempts:
        return "tools"
    return END

El try/except dentro de tool_node es deliberado: si una herramienta lanza y el error nunca llega al modelo como ToolMessage, el agente no puede recuperarse; solo ve silencio y o reintenta para siempre o abandona sin contexto.

5. Construir el grafo

def build_agent_graph(checkpointer=None):
    """Build and compile the LangGraph agent."""
    workflow = StateGraph(AgentState)

    # Agregar nodos
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tool_node)

    # Establecer punto de entrada
    workflow.set_entry_point("agent")

    # Agregar arista condicional desde el agente
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {
            "tools": "tools",
            END: END,
        },
    )

    # Después de tools, volver al agente
    workflow.add_edge("tools", "agent")

    return workflow.compile(checkpointer=checkpointer)

app = build_agent_graph(checkpointer=MemorySaver())

should_continue devuelve un string y el dict de mapeo lo resuelve a un nodo. Devolver END (la constante __end__) termina la ejecución; puedes mapearlo a un nodo de limpieza o resumidor si el agente necesita un epílogo.

6. Ejecutar el agente

config = {"configurable": {"thread_id": "session-1"}}

result = app.invoke(
    {
        "messages": [HumanMessage(content="What is the Redis cache-aside pattern?")],
        "max_attempts": 5,
    },
    config=config,
)

# Imprimir la respuesta final
for msg in result["messages"]:
    if isinstance(msg, AIMessage):
        print(f"AI: {msg.content}")
    elif isinstance(msg, ToolMessage):
        print(f"Tool: {msg.content}")

7. Stream de pasos del agente

for event in app.stream(
    {
        "messages": [HumanMessage(content="Calculate 15 * 23 and then search for Docker Compose")],
        "max_attempts": 5,
    },
    config=config,
):
    for node_name, node_output in event.items():
        print(f"\n--- {node_name} ---")
        if "messages" in node_output:
            for msg in node_output["messages"]:
                if isinstance(msg, AIMessage):
                    print(f"AI: {msg.content[:100]}")
                elif isinstance(msg, ToolMessage):
                    print(f"Tool result: {msg.content[:100]}")

Cómo funciona

El diagrama muestra la topología compilada: el agente itera por tools hasta que should_continue devuelve END.

1. StateGraph compila un grafo dirigido donde cada nodo es una función que recibe el AgentState compartido y devuelve una actualización parcial. Los nodos nunca mutan el estado en sitio; devuelven dicts que LangGraph fusiona usando el reducer de cada campo.
2. Los reducers controlan la fusión. add_messages agrega; un campo simple como attempts sobrescribe. Elige los reducers a propósito. Un campo lista sin add_messages se reemplaza en cada actualización de nodo y pierde el historial en silencio.
3. Las aristas condicionales enrutan en tiempo de ejecución. should_continue ve el estado después de que corriera el nodo agente: si el último mensaje pide herramientas y el presupuesto de intentos no se agotó, enruta a tools; si no, la ejecución termina.
4. El ciclo es lo que convierte esto en un agente y no en una cadena. Los resultados de herramientas vuelven como entradas ToolMessage, el agente los lee y decide de nuevo. El bucle solo para cuando el modelo responde sin llamadas a herramientas o se agota el presupuesto.
5. El checkpointing guarda una foto del estado completo tras cada nodo bajo el thread_id del config. El mismo thread_id hace que el siguiente invoke retome con los mensajes previos, y es también por lo que cada sesión de usuario necesita su propio ID.

Variantes

Aprobación human-in-the-loop

from langgraph.types import interrupt, Command

def tool_node_with_approval(state: AgentState) -> dict:
    """Execute tools with human approval for sensitive operations."""
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        if tool_call["name"] == "calculate":
            # Interrumpir y esperar aprobación humana
            approval = interrupt({
                "question": f"Approve calculation: {tool_call['args']['expression']}?",
                "tool_call": tool_call,
            })

            if not approval:
                return {"messages": [ToolMessage(
                    content="Calculation rejected by user.",
                    tool_call_id=tool_call["id"],
                )]}

    # Proceder con la ejecución normal de herramientas
    return tool_node(state)

# La primera llamada pausa dentro del interrupt:
result = app.invoke(
    {"messages": [HumanMessage(content="Calculate 100 / 0")]},
    config=config,
)
# Reanudar con la respuesta del humano:
result = app.invoke(Command(resume="approved"), config=config)

Dos detalles que conviene saber. interrupt y Command viven en langgraph.types, no en langgraph.graph; la ruta de import antigua no existe en las versiones actuales. Y reanudar vuelve a ejecutar todo el nodo desde el principio, así que mantén el código del nodo idempotente (o pon el interrupt al inicio del nodo).

Colaboración multi-agente

def researcher_node(state: AgentState) -> dict:
    """Researcher agent — searches for information."""
    response = model_with_tools.invoke([
        {"role": "system", "content": "You are a research assistant. Search for information."},
        *state["messages"],
    ])
    return {"messages": [response]}

def writer_node(state: AgentState) -> dict:
    """Writer agent — synthesizes findings into a report."""
    response = model.invoke([
        {"role": "system", "content": "You are a technical writer. Synthesize the research into a clear report."},
        *state["messages"],
    ])
    return {"messages": [response]}

def build_multi_agent_graph(checkpointer=None):
    workflow = StateGraph(AgentState)
    workflow.add_node("researcher", researcher_node)
    workflow.add_node("tools", tool_node)
    workflow.add_node("writer", writer_node)

    workflow.set_entry_point("researcher")
    workflow.add_conditional_edges("researcher", should_continue, {"tools": "tools", END: "writer"})
    workflow.add_edge("tools", "researcher")
    workflow.add_edge("writer", END)

    return workflow.compile(checkpointer=checkpointer)

Aquí END de should_continue mapea a writer en lugar del final real: el valor devuelto por la arista condicional pasa por el mapeo, así que “la ejecución terminó” puede significar “entregar al siguiente especialista” en lugar de “parar”.

Estado persistente con SQLite

from langgraph.checkpoint.sqlite import SqliteSaver

# Persistir el estado a SQLite para continuidad entre sesiones
app = build_agent_graph(checkpointer=SqliteSaver.from_conn_string("agent_state.db"))

# El estado persiste entre sesiones con el mismo thread_id

MemorySaver muere con el proceso: vale para notebooks, inútil para un servicio. SqliteSaver sobrevive reinicios en una sola máquina; PostgresSaver es la opción de producción cuando varios workers necesitan los mismos checkpoints.

Solución de problemas

• GraphInterrupt propaga como excepción: estás usando interrupt() sin checkpointer. Los interrupts persisten el estado para reanudar después, así que compile(checkpointer=...) es obligatorio: sin checkpointer no hay pausa.
• El agente itera para siempre: should_continue devuelve "tools" cada ronda. Revisa el presupuesto max_attempts, y luego el freno duro: recursion_limit en el config (por defecto 25) mata grafos descontrolados.
• El estado no persiste entre invokes: thread_id distinto en cada llamada, o no hay checkpointer. Tanto el thread ID como el checkpointer deben ser estables.
• Errores de validación de ToolMessage: un ToolMessage necesita un tool_call_id que coincida con la llamada que responde. Si construyes los resultados a mano y los IDs no cuadran, la validación falla. O usa el ToolNode preconstruido y olvida el cableado manual.
• ImportError en interrupt o Command: viven en langgraph.types, no en langgraph.graph. Actualiza el import en lugar de bajar de versión.

Mejores prácticas

• Establece max_attempts: previene bucles infinitos donde el agente sigue llamando herramientas sin converger. Trátalo como un presupuesto, no como un parche.
• Usa MemorySaver en desarrollo y SqliteSaver o PostgresSaver en producción.
• Define prompts de sistema claros: el agente necesita instrucciones explícitas sobre cuándo usar herramientas frente a responder directamente.
• Stream para UX en tiempo real: app.stream() muestra cada paso a medida que ocurre; las respuestas streaming del LLM cubren la variante a nivel de token.
• Pon interrupt al inicio del nodo: reanudar re-ejecuta el nodo desde el principio, así que todo lo que esté antes del interrupt corre dos veces.
• Nombra los nodos semánticamente: agent, tools, writer aparecen en cada evento del stream y en cada checkpoint, así que depurar es más rápido cuando los nombres significan algo.

Errores comunes

• Olvidar la anotación add_messages: sin ella, cada actualización de nodo reemplaza la lista de mensajes en lugar de agregar, y el historial de conversación desaparece a mitad de ejecución.
• Dejar escapar las excepciones de herramientas: si una herramienta lanza, el agente se atasca o reintenta a ciegas. Captura los errores dentro del nodo de herramientas y devuélvelos como ToolMessage para que el modelo vea el fallo.
• Sin límite de max_attempts: el agente puede iterar indefinidamente llamando herramientas sin producir una respuesta final.
• Reutilizar un thread_id entre usuarios: el estado se filtra entre sesiones. Un thread ID por conversación — siempre.
• Importar interrupt desde langgraph.graph: vive en langgraph.types en las versiones actuales; la ruta antigua lanza ImportError.

Ver también

• Documentación de LangGraph
• LangGraph en GitHub
• Guía completa de LangChain en producción
• Patrón Human-in-the-Loop