Colaboración multiagente con LangGraph en la práctica: patrón Supervisor y distribución de tareas

El mes pasado ayudé a un equipo a refactorizar un sistema de Agents. Al principio, un solo Agent cargaba con 12 herramientas: búsqueda, ejecución de código, generación de documentos, envío de correos… ¿El resultado? El LLM se debatía a menudo entre herramientas y usaba el ejecutor de código cuando debía buscar. Al depurar, mirando los logs, era imposible saber qué llamada a herramienta había fallado: todo el sistema era una caja negra.

Luego dividimos la arquitectura en patrón Supervisor + Workers: un Agent de control central encargado del enrutamiento y tres Agents especialistas, cada uno con su rol. La tasa de error en la selección de herramientas bajó a un tercio y la depuración quedó mucho más clara: se puede hacer trace capa por capa.

En pocas palabras: cuando tu Agent tiene más de 10 herramientas, la arquitectura de un solo Agent empieza a fallar. Este artículo te lleva desde los principios del patrón Supervisor hasta el uso completo de la API create_supervisor, y un caso práctico de un equipo Research + Writing. El código se puede ejecutar directamente; el enlace al repositorio de GitHub está al final.

1. ¿Por qué hace falta un sistema multiagente?

Tres puntos débiles del Agent único

Errores que yo mismo he cometido; quizá tú también. Un sistema de un solo Agent parece simple, pero tiene tres problemas graves:

Primero: demasiadas herramientas, parálisis al elegir.

No es exageración. Cuando un solo Agent tiene más de 10 herramientas, la tasa de error del LLM al elegir sube de forma notable. Puedes decir que los modelos son cada vez más listos; sí, pero las descripciones de herramientas se amontonan en el prompt y el modelo debe acertar entre más de 10 opciones. Esa carga cognitiva no es pequeña.

En mi sistema anterior, las descripciones de búsqueda y de ejecución de código se solapaban un poco (ambas podían «buscar información»), y el modelo saltaba de una a otra varias rondas, desperdiciando diálogo.

Segundo: acumulación de contexto, ventana saturada.

Todo el historial de subtareas se comprime en una sola ventana de contexto. Tras varias rondas, la ventana queda inundada de tool calls históricos y las instrucciones centrales se diluyen. El modelo empieza a «olvidar» e incluso pierde la petición original del usuario.

Lo viví en carne propia. Depurando un Agent que corrió 20 rondas, el contexto estaba lleno de registros de llamadas a funciones; al final la salida ya no tenía mucho que ver con lo que el usuario pidió al principio.

Tercero: imposible depurar, difícil localizar fallos.

Cuando algo falla, no sabes qué llamada a herramienta es la culpable. Un Agent único es una caja negra: los logs son un flujo de tool calls; localizar el problema implica revisar línea por línea.

El patrón Supervisor resuelve esto: un «Agent de control central» coordina varios «Agents especialistas», con responsabilidades separadas y roles claros.

Railway - Plataforma de despliegue moderna, sin ops complejas, en producción en minutos

Empezar ahora →Anuncio

Idea central del patrón Supervisor

En el fondo, es división del trabajo.

Imagina un equipo: un jefe de proyecto coordina; un investigador se centra en la investigación; un ingeniero en la implementación; un redactor en el informe. Cada uno tiene su especialidad; el jefe no tiene que saberlo todo, solo a quién asignar cada tarea.

El patrón Supervisor sigue esa lógica:

• Supervisor (Agent de control central): no hace el trabajo concreto; solo enruta, coordina e integra resultados
• Worker Agents (Agents especialistas): cada uno en un dominio, con un conjunto de herramientas reducido y responsabilidades claras

¿Qué ventajas trae?

Las herramientas se reparten entre Workers; cada Agent elige solo dentro de su conjunto. El contexto también se reparte: cada Agent mantiene solo su parte del historial. Al depurar puedes hacer trace por capas: a qué Worker asignó el Supervisor y qué ejecutó el Worker, de un vistazo.

2. Principios de arquitectura del patrón Supervisor

Primero, un diagrama de arquitectura:

                    ┌─────────────────┐
                    │ Petición usuario │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │ (Agent control) │
                    │                 │
                    │ Enrutamiento +  │
                    │ coordinación +  │
                    │ integración     │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
              ▼              ▼              ▼
       ┌──────────┐   ┌──────────┐   ┌──────────┐
       │ Research │   │   Math   │   │ Writing  │
       │  Agent   │   │  Agent   │   │  Agent   │
       │          │   │          │   │          │
       │ Búsqueda │   │ Cálculo  │   │ Generac. │
       └────┬─────┘   └────┬─────┘   └────┬─────┘
            │              │              │
            │   Worker     │   Worker     │   Worker
            │   resultado  │   resultado  │   resultado
            │              │              │
            └──────────────┴──────────────┘
                           │
                           ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │ Integra resultado│
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │ Respuesta final │
                    └─────────────────┘

Responsabilidades de los componentes clave

El Supervisor hace tres cosas:

1. Enrutamiento: analiza la petición del usuario y decide a qué Worker enviarla
2. Coordinación: gestiona el flujo de tareas entre Workers
3. Integración: resume los resultados de los Workers y devuelve la respuesta final

Cada Worker Agent con su rol:

Cada Worker solo tiene su conjunto de herramientas. Por ejemplo, el Research Agent puede tener solo búsqueda y scraping web; el Math Agent solo calculadora. Menos herramientas, mayor precisión al elegir.

Mecanismo de paso de mensajes

Punto clave: el estado global del grafo (Global Graph State).

Todos los Agents comparten el mismo objeto de estado. Tras ejecutar una tarea, el Worker añade el resultado al campo messages del estado. El Supervisor ve el nuevo mensaje y decide el siguiente paso.

Es un mecanismo append-only: los mensajes solo crecen, lo que preserva la integridad del historial.

Fan-out / Fan-in

Tareas complejas pueden requerir varios Workers en paralelo. Si el usuario pide «comparar datos de mercado de los productos A y B», el Supervisor puede lanzar dos tareas Research (una para A, otra para B): eso es fan-out.

Cuando ambos Workers devuelven resultados, el Supervisor los integra: fan-in.

LangGraph soporta este modo paralelo; en esta introducción no lo desarrollamos; lo veremos en técnicas avanzadas.

"LangGraph ofrece una forma de construir sistemas multiagente en los que cada Agent tiene su propio conjunto de herramientas y ámbito de responsabilidad, coordinados y distribuidos por un Supervisor."

3. API create_supervisor en detalle

Teoría hecha; ahora código.

Instalación e importación

pip install langgraph-supervisor langchain-openai

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

Definición de herramientas

Primero, herramientas para cada Worker:

from typing import Annotated

# Herramienta de cálculo matemático
def add(
    a: Annotated[float, "Primer número"],
    b: Annotated[float, "Segundo número"]
) -> float:
    """Add two numbers together."""
    return a + b

def multiply(
    a: Annotated[float, "Primer número"],
    b: Annotated[float, "Segundo número"]
) -> float:
    """Multiply two numbers."""
    return a * b

# Herramienta de búsqueda (simulada)
def web_search(query: str) -> str:
    """Search the web for information."""
    # En un proyecto real puedes integrar Tavily, Serper, etc.
    if "population" in query.lower():
        return "Pekín tiene unos 21,89 millones de habitantes (datos 2023)"
    elif "weather" in query.lower():
        return "Hoy en Pekín: soleado, 15-25 °C"
    else:
        return f"Resultados de búsqueda: {query}"

Aquí usamos Annotated de Python para que el modelo entienda mejor cada parámetro. El docstring de cada función también importa: el modelo lo usa para entender la herramienta.

Alibaba Cloud - Opción preferida para desarrolladores, 15 % de descuento exclusivo en Lighthouse

Obtener 15% dto. →Anuncio

Creación de Worker Agents

model = ChatOpenAI(model="gpt-4o")

# Agent experto en matemáticas
math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="Eres un experto en matemáticas, enfocado en cálculos numéricos. Cuando el usuario necesite operaciones matemáticas, usa tus herramientas para completar la tarea."
)

# Agent experto en investigación
research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="Eres un investigador senior, experto en buscar y organizar información. Cuando el usuario necesite consultar material, usa la herramienta de búsqueda para obtener respuestas."
)

Varios puntos a tener en cuenta:

1. El campo name importa: el Supervisor identifica e invoca Workers por nombre
2. El prompt define el rol: indica la especialidad del Agent
3. Conjunto de herramientas mínimo: cada Agent solo las herramientas necesarias, ni más ni menos

Creación del Supervisor

# Crear el sistema Supervisor
supervisor = create_supervisor(
    agents=[math_agent, research_agent],
    model=model,
    prompt="""Eres el líder del equipo y coordinas a los Agents especialistas.

Según la petición del usuario, decide a quién asignar la tarea:
- Cálculo matemático → math_expert
- Búsqueda de información → research_expert
- Tarea completada → responde directamente al usuario

Si varios especialistas deben colaborar, invócalos en un orden razonable."""
)

# Compilar en una aplicación ejecutable
app = supervisor.compile()

create_supervisor recibe tres parámetros clave:

• agents: lista de Worker Agents
• model: el LLM que usa el Supervisor
• prompt: cómo debe el Supervisor asignar tareas

Ejemplo de ejecución

from langchain_core.messages import HumanMessage

# Probar pregunta matemática
result = app.invoke({
    "messages": [HumanMessage(content="Calcula cuánto es 123 más 456")]
})
print(result["messages"][-1].content)
# Salida: 123 más 456 es 579

# Probar pregunta de búsqueda
result = app.invoke({
    "messages": [HumanMessage(content="¿Cuál es la población de Pekín?")]
})
print(result["messages"][-1].content)
# Salida: Según la búsqueda, Pekín tiene unos 21,89 millones de habitantes

El Supervisor infiere el tipo de petición y enruta al Worker correcto. Para el usuario es transparente: no necesita saber que hay varios Agents detrás.

4. Caso práctico: equipo Research + Writing

Lo anterior es el ejemplo más básico. Ahora un sistema más completo: un equipo que investiga y genera artículos técnicos automáticamente.

Descripción del escenario

El usuario introduce un tema técnico y el sistema:

1. Investiga material relacionado
2. Genera el esquema del artículo
3. Redacta el contenido completo
4. Revisa y corrige

Hacen falta tres Agents especializados trabajando en cadena.

Conjunto completo de herramientas

from typing import TypedDict, List
import json

# Herramienta de búsqueda simulada
def tech_search(query: str) -> str:
    """Buscar documentación y material técnico."""
    # En un proyecto real integra Tavily o Serper
    database = {
        "langgraph": "LangGraph es el framework de Agents de LangChain, con gestión de estado y grafos cíclicos.",
        "supervisor": "El patrón Supervisor es la arquitectura central de sistemas multiagente: un Agent de control coordina varios Agents especialistas.",
        "multi-agent": "Los sistemas multiagente resuelven el exceso de herramientas y la explosión de contexto del Agent único mediante distribución y colaboración."
    }

    results = []
    for key, value in database.items():
        if key in query.lower():
            results.append(value)

    return json.dumps(results) if results else "No se encontró material relevante; amplía el alcance de búsqueda"

# Herramienta de generación de esquema
def generate_outline(topic: str) -> str:
    """Generar esquema del artículo según el tema."""
    return json.dumps({
        "title": f"Guía completa de {topic}",
        "sections": [
            "1. Panorama y contexto",
            "2. Conceptos clave",
            "3. Caso práctico",
            "4. Buenas prácticas",
            "5. Conclusión"
        ]
    }, ensure_ascii=False)

# Herramienta de generación de contenido
def write_section(section_title: str, context: str) -> str:
    """Generar párrafos según título y contexto."""
    # En un proyecto real aquí puedes llamar al LLM
    return f"## {section_title}\n\nSegún el material investigado, los puntos clave de {section_title} son los siguientes...\n\n"

# Herramienta de revisión
def review_content(content: str) -> str:
    """Revisar precisión y legibilidad del contenido."""
    issues = []
    if len(content) < 100:
        issues.append("Contenido demasiado corto; conviene ampliarlo")
    if "TODO" in content:
        issues.append("Quedan marcadores TODO sin completar")

    return json.dumps({
        "passed": len(issues) == 0,
        "issues": issues,
        "suggestion": "Calidad aceptable, listo para publicar" if not issues else "Corrige según los problemas y vuelve a enviar"
    }, ensure_ascii=False)

Creación de Worker Agents

# Agent investigador
researcher = create_react_agent(
    model=model,
    tools=[tech_search],
    name="researcher",
    prompt="""Eres un investigador técnico senior, experto en investigar y entender tecnologías nuevas rápidamente.

Responsabilidades:
1. Recibir el tema de investigación
2. Usar la herramienta de búsqueda para encontrar material
3. Organizar un informe de investigación estructurado

Nota: solo investigas, no escribes. Entrega los resultados al writer."""
)

# Agent autor
writer = create_react_agent(
    model=model,
    tools=[generate_outline, write_section],
    name="writer",
    prompt="""Eres un experto en redacción técnica, capaz de convertir conceptos complejos en artículos claros.

Responsabilidades:
1. Recibir el informe de investigación
2. Generar el esquema del artículo
3. Redactar cada sección

Nota: tras el borrador inicial, pásalo al reviewer para revisión."""
)

# Agent revisor
reviewer = create_react_agent(
    model=model,
    tools=[review_content],
    name="reviewer",
    prompt="""Eres un revisor estricto que garantiza calidad y precisión del artículo.

Responsabilidades:
1. Comprobar integridad y precisión del contenido
2. Evaluar legibilidad y coherencia lógica
3. Proponer cambios o confirmar publicación

Nota: si hay problemas, devuelve el borrador al writer."""
)

Lógica del Supervisor

# Construir Supervisor personalizado con StateGraph
from langgraph.graph import StateGraph, END
from typing import Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

# Definir estado
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next_agent: str

# Lógica de decisión del Supervisor
def supervisor_node(state: AgentState) -> dict:
    """Decidir qué Agent ejecutar según el progreso actual."""
    messages = state["messages"]

    # Invocar el LLM para decidir
    decision = model.invoke([
        {"role": "system", "content": """Eres el líder del equipo; según el historial de conversación decide el siguiente paso:

- Si aún no hay material de investigación → devuelve 'researcher'
- Si hay investigación pero aún no hay artículo → devuelve 'writer'
- Si hay artículo pero sin revisar → devuelve 'reviewer'
- Si la revisión pasó → devuelve 'FINISH'

Devuelve solo el nombre del Agent, sin más texto."""},
        *messages
    ])

    next_agent = decision.content.strip()

    # Mapear al nombre correcto del Agent
    agent_map = {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "FINISH": END
    }

    return {"next_agent": agent_map.get(next_agent, "researcher")}

# Construir el grafo
workflow = StateGraph(AgentState)

# Añadir nodos
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

# Añadir aristas condicionales (lógica de enrutamiento del Supervisor)
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_agent"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        END: END
    }
)

# Tras completar, cada Agent vuelve al Supervisor
for agent in ["researcher", "writer", "reviewer"]:
    workflow.add_edge(agent, "supervisor")

# Punto de entrada
workflow.set_entry_point("supervisor")

# Compilar
app = workflow.compile()

Esta arquitectura tiene un bucle: cada Worker, al terminar, vuelve al Supervisor, que decide si asignar otro Worker o finalizar.

Flujo de ejecución

result = app.invoke({
    "messages": [HumanMessage(content="Escribe un artículo técnico sobre el patrón Supervisor de LangGraph")]
})

# Ver resultado final
print(result["messages"][-1].content)

# Ver traza de ejecución
for i, msg in enumerate(result["messages"]):
    print(f"{i+1}. {msg.__class__.__name__}: {msg.content[:100]}...")

El flujo aproximado es:

Petición usuario → Supervisor analiza → asigna a Researcher →
Researcher investiga → vuelve al Supervisor → asigna a Writer →
Writer redacta → vuelve al Supervisor → asigna a Reviewer →
Reviewer revisa → vuelve al Supervisor → confirma fin → salida

Cada capa es trazable; depurar es mucho más claro.

5. Técnicas avanzadas

Optimización de reenvío de mensajes: create_forward_message_tool

Quizá ya lo notaste: cuando un Worker termina, el Supervisor recibe el mensaje y a veces lo reformula. Eso gasta tokens y puede diluir la información.

LangGraph ofrece create_forward_message_tool para esto:

from langgraph_supervisor.handoff import create_forward_message_tool

# Crear herramienta de reenvío
forward_tool = create_forward_message_tool("supervisor")

# Pasarla al crear el Supervisor
supervisor = create_supervisor(
    agents=[researcher, writer, reviewer],
    model=model,
    tools=[forward_tool]  # Añadir herramienta de reenvío
)

Esta herramienta permite al Supervisor reenviar la respuesta del Worker al usuario sin resumirla de nuevo. Ahorra tokens y mejora la eficiencia.

Arquitectura de equipos jerárquicos

En proyectos más complejos puedes tener varios niveles de Supervisor:

                    ┌──────────────┐
                    │ Top Supervisor│
                    └──────┬───────┘
                           │
           ┌───────────────┼───────────────┐
           │               │               │
           ▼               ▼               ▼
    ┌────────────┐  ┌────────────┐  ┌────────────┐
    │Research    │  │ Writing    │  │ QA         │
    │Team        │  │ Team       │  │ Team       │
    │Supervisor  │  │ Supervisor │  │ Supervisor │
    └─────┬──────┘  └─────┬──────┘  └─────┬──────┘
          │               │               │
     ┌────┼────┐    ┌────┼────┐    ┌────┼────┐
     │    │    │    │    │    │    │    │    │
   Web  Doc  API   Out  Cont Rev   Test Code Audit
 Search Scrp Parse line ent  iew

Cada subequipo tiene su Supervisor; arriba un Supervisor global coordina. Sirve para proyectos grandes con roles más finos.

Manejo de errores

¿Qué pasa si un Worker falla?

from langgraph.pregel import RetryPolicy

# Configurar política de reintento
retry_policy = RetryPolicy(
    max_attempts=3,
    initial_interval=1.0,
    backoff_factor=2.0
)

app = workflow.compile(retry_policy=retry_policy)

También puedes añadir lógica de errores en el prompt del Supervisor:

KiloClaw - OpenClaw gestionado para empresas: orquestación de agentes IA y enrutamiento inteligente

Empezar →Anuncio

Si un Agent falla al ejecutar:
1. Registrar el error
2. Intentar invocar un Agent de respaldo
3. Si falla varias veces, informar al usuario del problema

Persistencia de estado

Las conversaciones multironda requieren guardar estado. LangGraph ofrece Checkpointer:

from langgraph.checkpoint.memory import MemorySaver

# Almacenamiento en memoria (entorno de desarrollo)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# Especificar thread_id al ejecutar
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [HumanMessage(content="...")]}, config=config)

# Las conversaciones posteriores conservan el contexto
result2 = app.invoke({"messages": [HumanMessage(content="Continúa la tarea anterior")]}, config=config)

En producción puedes usar Redis o PostgreSQL como Checkpointer.

"Hierarchical Agent Teams muestra cómo construir arquitecturas Supervisor multinivel para colaboración multiagente más compleja."

6. Recomendaciones para despliegue en producción

Monitorización y depuración

LangSmith es la plataforma oficial de monitorización de LangChain; rastrea cada paso de ejecución:

import os

os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "multi-agent-project"

Tras configurarlo, cada ejecución deja una traza completa en LangSmith, incluyendo:

• Entrada y salida de cada Agent
• Parámetros y valores de retorno de tool calls
• Estadísticas de consumo de tokens
• Análisis de tiempo de ejecución

Muy útil al depurar; ya no hace falta revisar logs línea por línea.

Control del costo en tokens

El patrón Supervisor ayuda, pero varios Agents aumentan el consumo de tokens. Algunas optimizaciones:

1. Prompt del Supervisor conciso: solo la lógica de enrutamiento necesaria
2. Usar forward_message_tool: evitar resúmenes duplicados
3. Reparto razonable de herramientas: cada Worker solo lo imprescindible
4. Limitar rondas de diálogo: tope máximo de iteraciones

# Establecer máximo de pasos
app = workflow.compile(
    checkpointer=memory,
    interrupt_after=20  # Máximo 20 pasos de ejecución
)

Integración con AWS Bedrock

Si tu proyecto está en AWS, puedes usar modelos de Bedrock:

from langchain_aws import ChatBedrock

model = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    region_name="us-east-1"
)

# El resto del código igual; solo sustituye model
supervisor = create_supervisor(
    agents=[math_agent, research_agent],
    model=model
)

Resumen de buenas prácticas

En resumen, varias lecciones de la práctica:

1. Empieza pequeño: 2-3 Agents al principio; amplía gradualmente
2. Roles claros: dominio definido por Worker, sin solapamientos
3. Monitorización con LangSmith: conéctalo en desarrollo para depurar
4. Cuidado con el costo en tokens: varios Agents amplifican el consumo
5. Aprovecha forward_message_tool: ahorra bastantes tokens
6. Repositorio oficial: langgraph-supervisor-py tiene ejemplos completos

Conclusión

El patrón Supervisor es, en el fondo, trabajo en equipo: dividir tareas grandes y dejar que cada Agent haga lo que mejor sabe.

Desde los tres puntos débiles del Agent único (parálisis al elegir herramientas, explosión de contexto, imposibilidad de depurar) hasta la solución elegante del Supervisor, este artículo recorre el camino completo. La API create_supervisor no es complicada de usar; lo importante es entender la arquitectura detrás.

Te recomiendo practicar con un proyecto pequeño: un sistema de dos Agents (por ejemplo búsqueda + resumen), y ampliar cuando funcione. Configura monitorización en LangSmith; al depurar lo agradecerás.

Ejemplos de código completos en GitHub: langgraph-supervisor-py. También merece la pena el tutorial oficial: Hierarchical Agent Teams.

Si tienes dudas, déjalas en los comentarios; las reviso.

Referencias

• LangGraph Supervisor Reference
• Hierarchical Agent Teams Tutorial
• LangGraph Multi-Agent Workflows Blog
• GitHub - langgraph-supervisor-py
• Build Multi-Agent Systems with AWS Bedrock