Construcción de Sistemas Multi-Agente en Producción con LangGraph: Gestión de Estado y Flujos Cíclicos

Los flujos lineales simples resultan insuficientes para tareas complejas en la empresa. Se requieren sistemas multi-agente donde especialistas (investigadores, programadores, revisores) colaboran.

LangGraph introduce una máquina de estados basada en grafos que permite bucles de reintento (loops), ramificación condicional y persistencia con Checkpointing.

Puntos Clave

• StateGraph Centralizado: Todos los nodos leen y actualizan un esquema de estado común TypedDict.
• Flujos Cíclicos Nativos: add_conditional_edges permite bucles de corrección entre agentes.
• Human-in-the-loop: Interrupciones para revisión humana antes de ejecutar acciones críticas.

Ejemplo en Python

from typing import Annotated, TypedDict, Literal
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_step: str

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

def researcher_node(state: AgentState):
    response = llm.invoke([HumanMessage(content="Investiga el tema.")] + state["messages"])
    return {"messages": [response], "next_step": "writer"}

def writer_node(state: AgentState):
    response = llm.invoke([HumanMessage(content="Redacta el informe.")] + state["messages"])
    return {"messages": [response], "next_step": "end"}

def router(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
    step = state.get("next_step", "researcher")
    if step == "writer": return "writer"
    elif step == "end": return END
    return "researcher"

workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("researcher", router)
workflow.add_conditional_edges("writer", router)

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

Conclusión

LangGraph proporciona la estructura necesaria para transformar prototipos de agentes en sistemas empresariales de alto rendimiento.