Sistemas Multiagente LangGraph: Supervisores, Subgrafos y Envío

Requisitos previos: los ejemplos suponen que ya puedes crear un grafo LangGraph de un solo agente con herramientas y memoria, y que sabes cómo pausar y reanudar una ejecución con interrupt(). Todo lo que sigue se basa en eso, pasando de la pregunta de por qué se querrían varios agentes a un asistente de investigación multiagente completo.

¿Por qué dividir un agente en muchos?

Supongamos que un asistente de un solo agente funciona bien. Entonces los usuarios comienzan a pedirle que revise correos electrónicos, investigue temas, escriba informes y gestione tareas, todo en la misma conversación. La respuesta habitual es añadir más herramientas e incluir más instrucciones en el prompt del sistema. Pronto hay 15 herramientas y un prompt de 2,000 palabras, y la calidad disminuye silenciosamente. El modelo comienza a mezclar cosas, utilizando la herramienta de correos electrónicos durante una tarea de investigación y la herramienta de investigación durante una tarea de correos.

Eso es sobrecarga cognitiva, y es la razón principal por la que existen los diseños multiagente. La solución es en la que los ingenieros de software han confiado durante décadas: dividir las responsabilidades. En LangGraph, hacerlo aborda tres problemas separados:

• Sobrecarga cognitiva. Un único modelo debe manejar demasiadas tareas no relacionadas. Los agentes especializados, cada uno bueno en una cosa, reducen esa presión.
• Cuellos de botella secuenciales. Las tareas que podrían realizarse simultáneamente se ejecutan una tras otra. Los agentes paralelos disminuyen la latencia.
• Complejidad. Nadie puede razonar sobre un grafo de 40 nodos. Los subgrafos más pequeños pueden entenderse y probarse uno por uno.

Cada patrón que se muestra a continuación aborda uno de esos tres problemas.

Cinco niveles de arquitectura

Imaginen esta progresión como una escalera donde cada peldaño añade una idea única.

Nivel 1: supervisor. Un agente supervisor coordina a los trabajadores dedicados. Sigue siendo un único tomador de decisiones, pero delega tareas en lugar de hacerlo todo él mismo, lo que ya lo sitúa por encima de un simple chatbot que utiliza herramientas.

Nivel 2: paralelización. Las tareas independientes se dividen y se ejecutan al mismo tiempo para luego fusionarse. La idea consiste en ramificaciones y conexiones: varias ramas se ejecutan de forma concurrente y suministran información a un paso posterior. Es adecuado para trabajos repetitivos en múltiples fuentes o subtareas.

Nivel 3: subgrafos. En lugar de un único flujo de trabajo enorme, cada rama se convierte en un grafo más pequeño con pasos internos privados y su propio esquema de salida. Las partes complejas se vuelven más fáciles de crear, probar y comprender.

Nivel 4: paralelismo dinámico. Con el patrón Send, que sigue un estilo map-reduce, la cantidad de ramas no se conoce de antemano. Los trabajadores se crean en tiempo de ejecución, uno por elemento, tema o analista. Los agentes no están predefinidos; todos trabajan simultáneamente y cada uno se encarga de una tarea específica.

Nivel 5: todo combinado. Un sistema realista utiliza todos los métodos anteriores al mismo tiempo, como en el asistente de investigación mencionado al final.

Una forma breve de recordarlo: el Nivel 1 hace que las rutas pasen por un supervisor único, el Nivel 2 ejecuta un conjunto fijo de trabajadores juntos, el Nivel 3 permite que cada rama sea un grafo completo y el Nivel 4 crea trabajadores según sea necesario.

Nivel 1: el patrón del supervisor

Este es el patrón multiagente más simple y el punto de partida adecuado. Un supervisor recibe la solicitud del usuario y elige qué especialista debe encargarse de ella. El especialista realiza el trabajo y devuelve los resultados, mientras que el supervisor decide qué sucederá a continuación.

Imagínese un bufete de abogados. El socio principal habla con el cliente, comprende sus necesidades y asigna tareas a los asociados junior: uno para contratos, otro para litigios y otro para cuestiones de cumplimiento. El asociado se ocupa del trabajo detallado, y el socio principal lo revisa y responde al cliente.

El flujo es el siguiente:

User Input
    ↓
[Supervisor Node]  ← decides who handles what
    ↓         ↓
[Specialist A] [Specialist B]   ← do the actual work
    ↓         ↓
[Supervisor Node]  ← reviews results, responds or routes again
    ↓
Response

Estado

La decisión de diseño clave es el campo next_agent. El supervisor escribe en él y el enrutador lo lee.

# ── MODULE 2: STATE ─────────────────────────────────────────
from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState

class SupervisorState(MessagesState):
    # The supervisor writes this field to indicate who should act next.
    # "researcher", "writer", "FINISH" are the possible values.
    next_agent: str

Herramientas especializadas

Cada especialista cuenta con sus propias herramientas específicas, sin solapamientos entre ellas.

# ── MODULE 3: TOOLS ─────────────────────────────────────────
from langchain_core.tools import tool

# Researcher's tools
@tool
def search_web(query: str) -> str:
    """Search the web for factual information, news, or research."""
    return f"[Web results for: {query}]"
@tool
def search_academic(query: str) -> str:
    """Search academic papers and journals for scholarly sources."""
    return f"[Academic results for: {query}]"
# Writer's tools
@tool
def format_as_report(content: str, title: str) -> str:
    """Format raw content into a structured report with headings."""
    return f"# {title}\n\n{content}"
@tool
def check_grammar(text: str) -> str:
    """Check and correct grammar in the provided text."""
    return f"[Grammar-checked]: {text}"
# Each specialist gets only its own tools
researcher_tools = [search_web, search_academic]
writer_tools = [format_as_report, check_grammar]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
researcher_llm = llm.bind_tools(researcher_tools)
writer_llm     = llm.bind_tools(writer_tools)

Nodos

A continuación vienen el nodo supervisor y los nodos especializados. La respuesta en texto plano del supervisor es, en sí misma, la decisión de enrutamiento.

# ── MODULE 4: NODES ─────────────────────────────────────────
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import ToolNode

# ── Supervisor ────────────────────────────────────────────
SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of specialists.
Given the conversation, decide who should act next.
Your team:
- researcher: finds information, searches web and academic sources
- writer: formats content, writes reports, checks grammar
- FINISH: use this when the task is complete and you have a final answer
Respond with ONLY the name of the next agent: researcher, writer, or FINISH."""
def supervisor_node(state: SupervisorState) -> dict:
    """Reads the conversation history and routes to the right specialist."""
    messages = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    response = llm.invoke(messages)
    # The supervisor's plain text response IS the routing decision
    return {"next_agent": response.content.strip()}

# ── Researcher Specialist ──────────────────────────────────
def researcher_node(state: SupervisorState) -> dict:
    """Specialist that handles all research and information-gathering tasks."""
    messages = [
        SystemMessage(content="You are a research specialist. Use your search tools to find accurate, thorough information.")
    ] + state["messages"]
    response = researcher_llm.invoke(messages)
    return {"messages": [response]}

# ── Writer Specialist ──────────────────────────────────────
def writer_node(state: SupervisorState) -> dict:
    """Specialist that handles all writing, formatting, and editing tasks."""
    messages = [
        SystemMessage(content="You are a writing specialist. Use your tools to produce well-structured, polished output.")
    ] + state["messages"]
    response = writer_llm.invoke(messages)
    return {"messages": [response]}

# Tool execution nodes for each specialist
researcher_tool_node = ToolNode(researcher_tools)
writer_tool_node     = ToolNode(writer_tools)

Enrutamiento

# ── MODULE 5: ROUTING ────────────────────────────────────────
def route_after_supervisor(state: SupervisorState) -> Literal["researcher", "writer", "__end__"]:
    """Read the supervisor's decision and route accordingly."""
    decision = state["next_agent"]
    if decision == "FINISH":
        return "__end__"
    return decision  # "researcher" or "writer"

def route_researcher(state: SupervisorState) -> Literal["researcher_tools", "supervisor"]:
    """After researcher acts: did it call a tool, or produce a final answer?"""
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "researcher_tools"
    return "supervisor"  # Research done - report back to supervisor
def route_writer(state: SupervisorState) -> Literal["writer_tools", "supervisor"]:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "writer_tools"
    return "supervisor"

Ensamblaje de grafos

# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

graph_builder = StateGraph(SupervisorState)
# Register all nodes
graph_builder.add_node("supervisor",        supervisor_node)
graph_builder.add_node("researcher",        researcher_node)
graph_builder.add_node("researcher_tools",  researcher_tool_node)
graph_builder.add_node("writer",            writer_node)
graph_builder.add_node("writer_tools",      writer_tool_node)
# Entry: always start at supervisor
graph_builder.add_edge(START, "supervisor")
# Supervisor routes to specialist or ends
graph_builder.add_conditional_edges(
    "supervisor",
    route_after_supervisor,
    {"researcher": "researcher", "writer": "writer", "__end__": END}
)
# Researcher loop: tools → researcher → supervisor
graph_builder.add_conditional_edges(
    "researcher", route_researcher,
    {"researcher_tools": "researcher_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("researcher_tools", "researcher")
# Writer loop: tools → writer → supervisor
graph_builder.add_conditional_edges(
    "writer", route_writer,
    {"writer_tools": "writer_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("writer_tools", "writer")
graph = graph_builder.compile(checkpointer=MemorySaver())

Lo importante en el ensamblaje es el flujo que vuelve al supervisor. Cuando un especialista termina, informa al supervisor. Este revisa toda la conversación y luego elige una de tres opciones: transferir la tarea a otro especialista, enviarla nuevamente al mismo o finalizarla. En esencia, eso es lo que significa la orquestación.

Nivel 2: paralelización con expansión y convergencia

El patrón del supervisor es secuencial, con un solo especialista activo a la vez. Muchas tareas reales contienen subtareas independientes que pueden ejecutarse simultáneamente, y el paralelismo resuelve el problema de la latencia.

La cocina es una buena analogía. Nadie termina la ensalada antes de empezar con la pasta y luego con la salsa. Preparas la ensalada, pones el agua a hervir y empiezas a preparar la salsa al mismo tiempo, de modo que la cena está lista más rápido.

En LangGraph, un nodo con varias aristas salientes genera la división paralela. LangGraph detecta esta expansión en abanico y ejecuta todos los destinos de forma concurrente en lo que denomina superpaso. Los resultados se recopilan antes de que comience cualquier nodo posterior, lo cual constituye la entrada en abanico.

Existe una regla estricta: cuando varios nodos en el mismo superpaso escriben en el mismo campo de estado, LangGraph debe saber cómo fusionar esas escrituras. Sin un reductor, se genera un error. Con operator.add, los resultados se concatenan de forma segura.

# ── MODULE 2: STATE FOR PARALLEL PATTERN ────────────────────
import operator
from typing import Annotated

class ResearchState(TypedDict):
    query: str
    # Without Annotated + operator.add, parallel writes to 'context' would CRASH.
    # With it, LangGraph concatenates results from all parallel nodes.
    context: Annotated[list[str], operator.add]
    final_answer: str

Un ejemplo de investigación paralela utiliza dos fuentes:

# ── MODULE 4: PARALLEL WORKER NODES ─────────────────────────

def web_search_node(state: ResearchState) -> dict:
    """Runs in parallel with academic_search_node."""
    result = f"[WEB] Results for: {state['query']}"
    return {"context": [result]}   # Returns a LIST - operator.add will concatenate
def academic_search_node(state: ResearchState) -> dict:
    """Runs in parallel with web_search_node."""
    result = f"[ACADEMIC] Papers about: {state['query']}"
    return {"context": [result]}   # Also a LIST
def synthesize_node(state: ResearchState) -> dict:
    """Runs AFTER both parallel nodes complete (fan-in point).
    state['context'] already contains results from BOTH parallel nodes."""
    combined = "\n".join(state["context"])
    prompt = [
        SystemMessage(content="Synthesize the following research into a clear answer."),
        HumanMessage(content=f"Research:\n{combined}\n\nQuery: {state['query']}")
    ]
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# ── MODULE 6: PARALLEL GRAPH ASSEMBLY ───────────────────────
from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(ResearchState)
graph_builder.add_node("web_search",      web_search_node)
graph_builder.add_node("academic_search", academic_search_node)
graph_builder.add_node("synthesize",      synthesize_node)
# Fan-out: START branches to BOTH search nodes simultaneously
graph_builder.add_edge(START,       "web_search")
graph_builder.add_edge(START,       "academic_search")
# Fan-in: BOTH search nodes must complete before synthesize runs
graph_builder.add_edge("web_search",      "synthesize")
graph_builder.add_edge("academic_search", "synthesize")
graph_builder.add_edge("synthesize", END)
parallel_graph = graph_builder.compile()

La expansión consiste simplemente en dos aristas que parten del mismo origen, START. LangGraph detecta esto y ejecuta simultáneamente ambas destinos. La inserción también ocurre por sí sola, ya que synthesize solo se ejecuta una vez que se cumplen ambas aristas de entrada, es decir, una vez que ambos nodos paralelos han finalizado.

Nivel 3: subgrafos para la encapsulación

Los nodos paralelos simples manejan casos sencillos. Cuando cada ramificación es compleja, con varios pasos, sus propias herramientas y su propia lógica, los subgrafos son la herramienta más adecuada. Un subgrafo es un StateGraph completamente compilado que funciona como un único nodo dentro de un grafo padre.

Imagínese los departamentos de una empresa. El CEO, que representa el grafo padre, delega tareas a Ingeniería y Marketing. Cada departamento tiene procesos y reuniones internas, pero el CEO solo transfiere trabajos y recibe los resultados finales.

El padre y el subgráfico se comunican a través de claves de estado compartidas. Cualquier clave presente tanto en el estado del padre como en el del subgráfico se transmite al iniciar el subgráfico y se devuelve al finalizarlo.

Parent State: {topic, cleaned_data, report_A, report_B, processed_ids}
                         ↓
Subgraph A State: {cleaned_data, internal_step_1, report_A, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_A' and 'processed_ids' back to parent

Subgraph B State: {cleaned_data, internal_step_2, report_B, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_B' and 'processed_ids' back to parent

Las claves que existen únicamente en el subgrafo, como internal_step_1 arriba, permanecen privadas, y el padre nunca las ve.

A continuación se muestra una implementación completa:

# ── MODULE 2: STATE SCHEMAS (Three Levels) ──────────────────

# Parent state - what the orchestrating graph tracks
class ParentState(TypedDict):
    raw_input: str
    cleaned_data: str                               # passed INTO both subgraphs
    failure_summary: str                            # written back by subgraph A
    performance_report: str                         # written back by subgraph B
    processed_ids: Annotated[list[int], operator.add]  # BOTH subgraphs write this

# Subgraph A's internal state - includes its private fields + shared keys
class FailureAnalysisState(TypedDict):
    cleaned_data: str        # shared input from parent
    failures: list[str]      # private intermediate state - parent never sees this
    failure_summary: str     # shared output to parent
    processed_ids: Annotated[list[int], operator.add]  # shared output (accumulated)

# Subgraph A's output schema - filters which keys return to parent
# Without this, ALL keys return, which can cause conflicts
class FailureAnalysisOutput(TypedDict):
    failure_summary: str     # only these two keys go back to parent
    processed_ids: Annotated[list[int], operator.add]

# Subgraph B's internal state
class PerformanceAnalysisState(TypedDict):
    cleaned_data: str
    metrics: list[str]          # private
    performance_report: str     # shared output
    processed_ids: Annotated[list[int], operator.add]

class PerformanceAnalysisOutput(TypedDict):
    performance_report: str
    processed_ids: Annotated[list[int], operator.add]

# ── SUBGRAPH A: Failure Analysis ─────────────────────────────
def extract_failures(state: FailureAnalysisState) -> dict:
    failures = [f"Error in {state['cleaned_data']}: timeout"]
    return {"failures": failures, "processed_ids": [1]}
def summarize_failures(state: FailureAnalysisState) -> dict:
    summary = f"Found {len(state['failures'])} failure(s): {state['failures']}"
    return {"failure_summary": summary}
failure_builder = StateGraph(FailureAnalysisState, output=FailureAnalysisOutput)
failure_builder.add_node("extract_failures",  extract_failures)
failure_builder.add_node("summarize_failures", summarize_failures)
failure_builder.add_edge(START, "extract_failures")
failure_builder.add_edge("extract_failures", "summarize_failures")
failure_builder.add_edge("summarize_failures", END)
failure_subgraph = failure_builder.compile()  # compiled → can be used as a node

# ── SUBGRAPH B: Performance Analysis ────────────────────────
def collect_metrics(state: PerformanceAnalysisState) -> dict:
    metrics = [f"Latency: 120ms", "Uptime: 99.8%"]
    return {"metrics": metrics, "processed_ids": [2]}
def write_report(state: PerformanceAnalysisState) -> dict:
    report = f"Performance Report:\n" + "\n".join(state["metrics"])
    return {"performance_report": report}
performance_builder = StateGraph(PerformanceAnalysisState, output=PerformanceAnalysisOutput)
performance_builder.add_node("collect_metrics", collect_metrics)
performance_builder.add_node("write_report",    write_report)
performance_builder.add_edge(START, "collect_metrics")
performance_builder.add_edge("collect_metrics", "write_report")
performance_builder.add_edge("write_report", END)
performance_subgraph = performance_builder.compile()

# ── PARENT GRAPH ─────────────────────────────────────────────
def clean_data_node(state: ParentState) -> dict:
    return {"cleaned_data": state["raw_input"].strip().lower()}
def combine_reports_node(state: ParentState) -> dict:
    combined = f"{state['failure_summary']}\n\n{state['performance_report']}"
    print(f"Final combined report:\n{combined}")
    return {}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("clean_data",           clean_data_node)
parent_builder.add_node("failure_analysis",     failure_subgraph)    # subgraph as node
parent_builder.add_node("performance_analysis", performance_subgraph) # subgraph as node
parent_builder.add_node("combine_reports",      combine_reports_node)
parent_builder.add_edge(START, "clean_data")
# Fan-out to both subgraphs (parallel execution)
parent_builder.add_edge("clean_data",           "failure_analysis")
parent_builder.add_edge("clean_data",           "performance_analysis")
# Fan-in: combine only after both subgraphs complete
parent_builder.add_edge("failure_analysis",     "combine_reports")
parent_builder.add_edge("performance_analysis", "combine_reports")
parent_builder.add_edge("combine_reports", END)
parent_graph = parent_builder.compile()

El esquema de salida, FailureAnalysisOutput y PerformanceAnalysisOutput en este ejemplo, es la parte que muchos tutoriales omiten, y por eso la gente se encuentra con conflictos en los estados. Este esquema filtra con precisión qué campos expone el subgráfico a su padre, y todo lo que falte se considera estado interno privado.

Nivel 4: la API Send y map-reduce

Los ramas paralelas y los subgráficos abarcan casos en los que el número de ramas se conoce en el momento del diseño. Pero, ¿qué pasa si un usuario quiere que se investiguen 5 temas hoy y 20 mañana? Codificar 20 ramas de forma fija no es una opción. Para eso existe el patrón map-reduce, impulsado por Send.

Imagínese a un editor que asigna capítulos. Un editor recibe el manuscrito, lo divide en capítulos y, al mismo tiempo, se los entrega a un editor de texto diferente. Ya sea que haya dos capítulos o veinte, ese es el número de editores de texto que se contratan. Cuando todos terminan, los resultados se reúnen en el libro final. La cantidad de trabajadores en paralelo se decide en tiempo de ejecución, no en el momento del diseño.

Tres palabras clave son importantes:

• Send(node_name, state), importado de langgraph.constants, crea una instancia nueva de un nodo con su propio contenido de estado. Una función de enrutamiento que devuelve una lista de ellos hace que LangGraph inicie cada uno al mismo tiempo, cada uno con estado independiente.
• operator.add en el campo de recolección corresponde al paso de reducción. A medida que los trabajadores terminan, sus resultados se acumulan en una lista que crece dentro del estado padre.

Estado del trabajador puede ser un diccionario personalizado que no necesita coincidir con el estado del padre. Un trabajador solo lleva los datos de su propia porción de trabajo.

He aquí la implementación completa de map-reduce:

# ── MODULE 2: MAP-REDUCE STATE ──────────────────────────────
from langgraph.constants import Send

class OverallState(TypedDict):
    topic: str
    subjects: list[str]                         # populated by the map step
    jokes: Annotated[list[str], operator.add]   # accumulated by the reduce step
    best_joke: str
class JokeState(TypedDict):
    """The private state each parallel worker gets.
    Notice: this does NOT need to match OverallState."""
    subject: str

# ── MODULE 4: MAP-REDUCE NODES ──────────────────────────────
def generate_subjects(state: OverallState) -> dict:
    """MAP step 1: Expands the topic into a list of subjects to process."""
    prompt = [
        SystemMessage(content="Generate a list of 3 sub-topics related to the given topic. Return as comma-separated values only."),
        HumanMessage(content=state["topic"])
    ]
    response = llm.invoke(prompt)
    subjects = [s.strip() for s in response.content.split(",")]
    return {"subjects": subjects}

def generate_joke(state: JokeState) -> dict:
    """MAP step 2: Worker node. Each instance handles exactly ONE subject.
    Receives its own isolated JokeState - not the full OverallState."""
    prompt = [HumanMessage(content=f"Write a short, funny joke about: {state['subject']}")]
    response = llm.invoke(prompt)
    # Returns a LIST - operator.add in OverallState will accumulate these
    return {"jokes": [response.content]}

def pick_best_joke(state: OverallState) -> dict:
    """REDUCE step: All jokes are now in state['jokes']. Pick the winner."""
    jokes_text = "\n".join([f"{i}. {j}" for i, j in enumerate(state["jokes"])])
    prompt = [
        SystemMessage(content="You are a comedy judge. Pick the funniest joke from the list below. Reply with ONLY the number."),
        HumanMessage(content=jokes_text)
    ]
    response = llm.invoke(prompt)
    winning_index = int(response.content.strip())
    return {"best_joke": state["jokes"][winning_index]}

# ── MODULE 5: THE SEND ROUTER ────────────────────────────────
def continue_to_jokes(state: OverallState):
    """The routing function that LAUNCHES parallel workers.

    Returning a LIST of Send objects (not a string) is what triggers map-reduce.
    Each Send creates an independent, parallel execution of 'generate_joke'
    with its own state payload. The number of parallel workers is determined
    at runtime by how many subjects exist in state.
    """
    return [
        Send("generate_joke", {"subject": subject})
        for subject in state["subjects"]
    ]

# ── MODULE 6: MAP-REDUCE GRAPH ASSEMBLY ─────────────────────
graph_builder = StateGraph(OverallState)
graph_builder.add_node("generate_subjects", generate_subjects)
graph_builder.add_node("generate_joke",     generate_joke)
graph_builder.add_node("pick_best_joke",    pick_best_joke)
graph_builder.add_edge(START, "generate_subjects")
# This conditional edge uses the Send router - it fans out dynamically
graph_builder.add_conditional_edges(
    "generate_subjects",
    continue_to_jokes,   # returns a LIST of Send objects
    ["generate_joke"]    # list of possible destination nodes (for graph validation)
)
# Fan-in: pick_best_joke runs after ALL parallel generate_joke instances complete
graph_builder.add_edge("generate_joke", "pick_best_joke")
graph_builder.add_edge("pick_best_joke", END)
map_reduce_graph = graph_builder.compile()

La magia reside en continue_to_jokes. Devuelve una lista de objetos Send en lugar de una cadena, y cuando LangGraph recibe dicha lista de una función de enrutamiento, ejecuta inmediatamente cada uno de ellos. Tres elementos en state["subjects"] significan tres nodos generate_joke en paralelo, y veinte elementos significan veinte. El grafo escala según lo que genere el tiempo de ejecución.

Nivel 5: un asistente de investigación multiagente

Ahora todo se combina: orquestación al estilo supervisor, aprobación humana, subgrafos en paralelo y la API Send, en un único sistema de tipo producción. Es el proyecto final del curso de LangGraph Academy, aquí comentado.

El funcionamiento: un usuario indica un tema de investigación. El sistema genera un equipo de personajes de analistas de IA, y una persona puede aprobarlos o perfeccionarlos. Cada analista actúa entonces como un subagente de entrevista en paralelo, haciendo preguntas a un “experto” de IA mediante búsquedas en la web y generando una sección del informe. Finalmente, un nodo de redactor compila las secciones para formar el documento final.

User provides topic
        ↓
[create_analysts] → LLM generates N analyst personas
        ↓
[human_feedback]  ← INTERRUPT: human approves or sends feedback
        ↓         ↑
[should_continue] → if feedback, loop back to create_analysts
        ↓
[initiate_research] → Send API: spawns N parallel interview subgraphs
        ↓↓↓ (N parallel workers running simultaneously)
[interview_subgraph × N] → each analyst interviews an "expert" via web search
        ↓
[write_report]    → LLM compiles all sections into a final report

Estado en tres niveles

from pydantic import BaseModel, Field

# ── Analyst persona - a Pydantic model for structured LLM output ──
class Analyst(BaseModel):
    name: str = Field(description="The analyst's full name")
    role: str = Field(description="Their professional role (e.g. 'Financial Analyst')")
    focus: str = Field(description="Their specific analytical focus area")

    @property
    def persona(self) -> str:
        return f"Name: {self.name}\nRole: {self.role}\nFocus: {self.focus}"
class Perspectives(BaseModel):
    analysts: list[Analyst]

# ── Outer graph state ─────────────────────────────────────────
class ResearchGraphState(TypedDict):
    topic: str
    max_analysts: int
    human_analyst_feedback: str      # human writes here during HITL pause
    analysts: list[Analyst]
    sections: Annotated[list, operator.add]  # accumulated from all parallel interviews
    final_report: str

# ── Interview subgraph state ──────────────────────────────────
class InterviewState(MessagesState):
    # inherits: messages: Annotated[list[BaseMessage], add_messages]
    max_num_turns: int
    context: Annotated[list, operator.add]  # search results accumulate here
    analyst: Analyst
    interview: str       # formatted transcript
    sections: list       # completed section (sent back to outer graph)

Generación de analistas, con verificación humana

# ── ANALYST GENERATION NODE ───────────────────────────────────
ANALYST_PROMPT = """You are generating a team of {max_analysts} research analysts
for the topic: {topic}.
{feedback_section}
Each analyst should have a unique perspective and area of focus.
Generate analysts with diverse viewpoints - financial, technical, social, etc."""
def create_analysts(state: ResearchGraphState) -> dict:
    """Generates analyst personas using structured LLM output."""
    feedback = state.get("human_analyst_feedback", "")
    feedback_section = (
        f"User feedback to incorporate:\n{feedback}" if feedback
        else "No feedback yet - generate your best initial set."
    )

    prompt = ANALYST_PROMPT.format(
        max_analysts=state["max_analysts"],
        topic=state["topic"],
        feedback_section=feedback_section
    )

    # with_structured_output: forces LLM to return a Pydantic model
    # instead of free-form text - ensures clean, usable data
    structured_llm = llm.with_structured_output(Perspectives)
    result = structured_llm.invoke([SystemMessage(content=prompt)])
    return {"analysts": result.analysts}

def human_feedback_node(state: ResearchGraphState) -> dict:
    """No-op node. Its only purpose is to be a pause point.
    The graph is compiled with interrupt_before=['human_feedback'].
    See Part 3 for the full HITL pattern."""
    pass  # Does nothing - the pause happens BEFORE this node via interrupt_before

def should_continue_to_research(state: ResearchGraphState) -> Literal["create_analysts", "initiate_research"]:
    """After human feedback: loop back to refine, or proceed to interviews."""
    if state.get("human_analyst_feedback"):
        return "create_analysts"
    return "initiate_research"

El subgrafo de la entrevista

Cada analista funciona como su propio subgrafo, un pequeño agente ReAct que entrevista a un “experto” (otro modelo LLM que desempeña ese papel) y busca en la web.

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def web_search_tool(query: str) -> str:
    """Search the web for information on a topic."""
    return f"[Search results for '{query}']"  # Replace with real search API
interview_tools = [web_search_tool]
interview_llm = llm.bind_tools(interview_tools)

def generate_question(state: InterviewState) -> dict:
    """The analyst node: generates the next interview question."""
    analyst = state["analyst"]
    messages = [
        SystemMessage(content=(
            f"You are {analyst.name}, a {analyst.role} focused on {analyst.focus}. "
            f"You are interviewing an expert about the research topic. "
            f"Ask insightful questions that align with your analytical focus. "
            f"When you have enough information, say 'Thank you, that is all I needed.'"
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def generate_answer(state: InterviewState) -> dict:
    """The expert node: answers the analyst's question, using web search."""
    messages = [
        SystemMessage(content=(
            "You are an expert being interviewed. Answer thoroughly and factually. "
            "Use the web search tool when you need current data."
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def save_interview(state: InterviewState) -> dict:
    """Formats the full Q&A exchange into a clean transcript string."""
    transcript = []
    for msg in state["messages"]:
        if hasattr(msg, "content"):
            role = "Analyst" if type(msg).__name__ == "HumanMessage" else "Expert"
            transcript.append(f"{role}: {msg.content}")
    return {"interview": "\n\n".join(transcript)}

def write_section(state: InterviewState) -> dict:
    """The final node: uses the interview transcript to write a report section."""
    analyst = state["analyst"]
    prompt = [
        SystemMessage(content=(
            f"Based on the following interview conducted by {analyst.name} ({analyst.role}), "
            f"write a structured report section focused on {analyst.focus}. "
            f"Be concise, factual, and cite specific points from the interview."
        )),
        HumanMessage(content=state["interview"])
    ]
    response = llm.invoke(prompt)
    # sections is the key that Send passes back to the outer graph
    return {"sections": [response.content]}

def route_interview(state: InterviewState) -> Literal["generate_answer", "save_interview"]:
    """Continue interviewing OR wrap up when analyst is satisfied or max turns hit."""
    last_message = state["messages"][-1]
    if (
        "thank you, that is all" in last_message.content.lower()
        or len(state["messages"]) >= state.get("max_num_turns", 6) * 2
    ):
        return "save_interview"
    # Check if analyst made a tool call (web search)
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "generate_answer"
    return "generate_answer"

# ── INTERVIEW SUBGRAPH ASSEMBLY ───────────────────────────────
interview_builder = StateGraph(InterviewState)
interview_tool_node = ToolNode(interview_tools)
interview_builder.add_node("generate_question", generate_question)
interview_builder.add_node("generate_answer",   generate_answer)
interview_builder.add_node("interview_tools",   interview_tool_node)
interview_builder.add_node("save_interview",    save_interview)
interview_builder.add_node("write_section",     write_section)
interview_builder.add_edge(START, "generate_question")
interview_builder.add_conditional_edges(
    "generate_question",
    route_interview,
    {"generate_answer": "generate_answer", "save_interview": "save_interview"}
)
interview_builder.add_conditional_edges(
    "generate_answer",
    lambda s: "interview_tools" if (hasattr(s["messages"][-1], "tool_calls") and s["messages"][-1].tool_calls) else "generate_question",
    {"interview_tools": "interview_tools", "generate_question": "generate_question"}
)
interview_builder.add_edge("interview_tools", "generate_answer")
interview_builder.add_edge("save_interview",  "write_section")
interview_builder.add_edge("write_section",   END)
interview_graph = interview_builder.compile()

El gráfico de orquestación externa

# ── SEND ROUTER: Spawn one interview per analyst ─────────────
def initiate_research(state: ResearchGraphState):
    """Launches N parallel interview subgraphs via Send — one per analyst.
    This is the map-reduce Send pattern from Level 4, applied to subgraphs."""
    return [
        Send("interview", {
            "analyst": analyst,
            "messages": [HumanMessage(content=f"Research topic: {state['topic']}")],
            "context": [],
            "max_num_turns": 3,
            "interview": "",
            "sections": []
        })
        for analyst in state["analysts"]
    ]
# ── REPORT COMPILATION NODE ───────────────────────────────────
def write_report(state: ResearchGraphState) -> dict:
    """Fan-in point: all sections have arrived from parallel interviews.
    The LLM compiles them into a final report."""
    sections_text = "\n\n---\n\n".join(state["sections"])
    prompt = [
        SystemMessage(content=(
            "You are a senior editor. Compile the following report sections written by "
            "different analysts into one cohesive, well-structured final report. "
            "Add a proper introduction and conclusion. Do not repeat content."
        )),
        HumanMessage(content=f"Topic: {state['topic']}\n\nSections:\n{sections_text}")
    ]
    response = llm.invoke(prompt)
    return {"final_report": response.content}

# ── OUTER GRAPH ASSEMBLY ─────────────────────────────────────
outer_builder = StateGraph(ResearchGraphState)
outer_builder.add_node("create_analysts",   create_analysts)
outer_builder.add_node("human_feedback",    human_feedback_node)
outer_builder.add_node("interview",         interview_graph)   # subgraph as node
outer_builder.add_node("write_report",      write_report)
outer_builder.add_edge(START, "create_analysts")
outer_builder.add_edge("create_analysts", "human_feedback")
outer_builder.add_conditional_edges(
    "human_feedback",
    should_continue_to_research,
    {"create_analysts": "create_analysts", "initiate_research": "initiate_research"}
)
# initiate_research is a ROUTING FUNCTION (returns Send objects), not a node
outer_builder.add_conditional_edges(
    "human_feedback",
    initiate_research,   # returns list of Send objects
    ["interview"]        # validation: lists possible destinations
)
outer_builder.add_edge("interview",    "write_report")
outer_builder.add_edge("write_report", END)
research_graph = outer_builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_feedback"]  # HITL pause for analyst approval
)

Ejecutar todo el proceso

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "research-001"}}

    # Step 1: Generate analysts
    result = research_graph.invoke(
        {
            "topic": "The impact of AI on the future of software engineering",
            "max_analysts": 3,
            "human_analyst_feedback": "",
            "analysts": [],
            "sections": [],
            "final_report": ""
        },
        config=config
    )

    # Step 2: HITL — graph paused at human_feedback node
    snapshot = research_graph.get_state(config)
    analysts = snapshot.values["analysts"]

    print("Generated analysts:")
    for a in analysts:
        print(f"  - {a.name} ({a.role}): {a.focus}")

    feedback = input("\nProvide feedback to refine analysts (or press Enter to approve): ")

    if feedback.strip():
        # Human has feedback — inject it and let supervisor regenerate
        research_graph.update_state(
            config,
            {"human_analyst_feedback": feedback},
            as_node="human_feedback"
        )
    else:
        # Approved — clear feedback and proceed to interviews
        research_graph.update_state(
            config,
            {"human_analyst_feedback": None},
            as_node="human_feedback"
        )

    # Step 3: Resume — N parallel interviews launch via Send, then report is compiled
    print("\nRunning parallel interviews...")
    final_result = research_graph.invoke(None, config=config)

    print("\n" + "="*60)
    print("FINAL RESEARCH REPORT")
    print("="*60)
    print(final_result["final_report"])

Elegir un patrón

• Un único agente que se confunde entre tareas requiere el patrón de supervisor. El supervisor dirige las tareas y los especialistas las ejecutan.
• Tareas independientes y lentas necesitan los patrones de fan-out y fan-in. Ejúzcalas simultáneamente y añada reducidores operator.add para los campos compartidos.
• Ramas paralelas que son complejas en sí mismas requieren subgrafos. Compile cada rama como un gráfico independiente, incorpórela como un nodo y utilice esquemas de salida para controlar qué se devuelve al nodo padre.
• Un número de trabajadores decidido en tiempo de ejecución necesita el patrón Send. Un enrutador genera una lista de objetos Send, y cada uno crea un trabajador independiente con su propio estado.
• Un sistema de producción real utiliza todo lo anterior. El asistente de nivel 5 combina todos los patrones porque una tarea real los necesita a todos.

Referencia de palabras clave

Estructura de supervisor. supervisor_node es el orquestador. Lee el estado, escribe next_agent y nunca realiza tareas especializadas. specialist_node se encarga de una tarea específica con sus propias herramientas y comando, y luego envía la información de vuelta al supervisor. next_agent: str es el campo de estado convencional para la redirección: el supervisor escribe un nombre y el router lo lee.

Paralelismo. El “fan-out” se refiere a varias llamadas a add_edge desde un mismo nodo de origen, las cuales LangGraph ejecuta de forma concurrente. El “fan-in” significa varios bordes que llegan a un mismo nodo de destino, el cual espera a que lleguen todos los datos de los nodos de origen. Annotated[list, operator.add] es obligatorio en cualquier campo al que escriban varios nodos en paralelo, ya que sin él las escrituras simultáneas causan errores; además, operator.add es el reductor habitual, encargado de concatenar las listas provenientes de los nodos concurrentes.

Subgrafos. StateGraph(InternalState, output=OutputSchema) es la forma de dos argumentos en la que el segundo argumento filtra qué claves se devuelven al padre. Un esquema de salida es un TypedDict que solo contiene las claves que se exponen, mientras que todo lo demás permanece privado. Las claves coincidentes actúan como canal entre el padre y el hijo, ya que cualquier clave presente en ambos esquemas se comparte automáticamente. subgraph.compile() convierte el subgrafo en una función llamable, la cual luego se pasa a parent.add_node("name", compiled_subgraph). Al llamar a graph.get_graph(xray=1).draw_mermaid() se muestra la estructura interna del subgrafo en el diagrama.

Map-reduce. Send(node_name, state_dict) representa una llamada a un trabajador en paralelo, y su carga no necesita coincidir con el estado padre. Al devolver [Send(...), Send(...)] desde un enrutador, se ejecutan todos al mismo tiempo. Al utilizar Send, el tercer argumento de add_conditional_edges es una lista, y no un diccionario, con nombres de destino válidos, utilizada únicamente para la validación del grafo. Finalmente, with_structured_output(PydanticModel) se usa después de una llamada al modelo para forzar que el resultado sea JSON que cumpla con un esquema Pydantic. Es la forma habitual de obtener decisiones fiables del supervisor y listas de analistas.

La composición es la habilidad clave

Pasar de un agente a muchos no es realmente un salto cualitativo. Se trata de los mismos componentes básicos aplicados repetidamente y conectados entre sí. Cada agente en un sistema multiagente es un grafo, un nodo o un subgrafo, y las primitivas nunca cambian. Lo importante es saber cómo componerlos.

El camino seguido fue intencional: un supervisor, seguido de nodos paralelos simples, subgrafos paralelos para ramas complejas, trabajadores dinámicos controlados por Send, y finalmente todo ello combinado. Si algún paso resultó sencillo, eso significa que el diseño cumplió su función, ya que cada nivel se basa de manera clara en el anterior.

El asistente de investigación de nivel 5 se acerca a lo que contiene una base de código multiagente para producción: bucles de aprobación humana, especialistas en paralelo, salida de modelos estructurada, generación dinámica de trabajadores y un paso de síntesis final. Junto con una estructura de proyecto canónica, gestión de memoria y seguridad con intervención humana, estos patrones abarcan la arquitectura detrás de la mayoría de las aplicaciones reales de LangGraph.