Pattern Supervisor

Descripción de la cápsula

Tu Research Agent v4 hace todo: busca, analiza, y escribe. En la cápsula anterior viste por qué eso es un problema cuando la complejidad crece. Ahora vas a resolver ese problema con el patrón más común de multi-agent: el Supervisor.

El patrón Supervisor funciona como un project manager coordinando especialistas. Un agente central (el supervisor) recibe la consulta del usuario, decide qué especialista debe trabajar, le delega la tarea, recibe el resultado, decide el siguiente paso, y repite hasta tener una respuesta completa. El supervisor NO hace el trabajo — coordina a quienes lo hacen.

En esta cápsula implementarás dos versiones: un supervisor manual con StateGraph (routing basado en reglas, sin LLM para las decisiones de routing) y un supervisor con LLM usando create_supervisor (el modelo decide a quién delegar). Empezarás simple y escalarás solo si es necesario — la misma filosofía que aplicas a todo en ingeniería.

El rol del supervisor: coordinar, no ejecutar

El error más común al diseñar un supervisor es convertirlo en un "super-agente" que entiende todo. No. El supervisor es un router: recibe una tarea, decide a quién delegarla, y recopila resultados. No necesita entender el contenido de la investigación — solo necesita saber a quién enviársela.

Supervisor NO es:                    Supervisor SÍ es:
  Un experto en todo                   Un coordinador
  Un agente con 15 tools               Un router con reglas claras
  Quien hace el trabajo                Quien decide QUIÉN hace el trabajo
  Un genio que entiende todo           Un gerente que sabe a quién preguntar

Piensa en un gerente de proyecto: no necesita saber programar para coordinar un equipo de developers. Necesita saber quién es bueno en qué, cuándo asignar cada tarea, y cuándo el resultado es suficiente.

Arquitectura

  Usuario → Supervisor → Researcher → Supervisor → Analyst → Supervisor → Respuesta
               │                         │                        │
               └── decide quién ─────────┴── decide siguiente ───┘

El flujo: el usuario envía una consulta al supervisor. El supervisor decide qué worker necesita trabajar, delega, recibe el resultado, y decide el siguiente paso. Repite hasta que tiene suficiente información para responder.

Supervisor manual con StateGraph

Empezamos con la versión más simple: el supervisor es una función Python que decide el siguiente paso basándose en reglas. Sin LLM para el routing.

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


class ResearchTeamState(TypedDict):
    query: str
    research: str
    analysis: str
    report: str
    next_worker: str
    iteration: int


def supervisor(state: ResearchTeamState) -> dict:
    iteration = state.get("iteration", 0)
    if iteration == 0:
        return {"next_worker": "researcher", "iteration": 1}
    if iteration == 1:
        return {"next_worker": "analyst", "iteration": 2}
    if iteration == 2:
        return {"next_worker": "reporter", "iteration": 3}
    return {"next_worker": "done", "iteration": iteration + 1}


def researcher(state: ResearchTeamState) -> dict:
    query = state["query"]
    results = (
        f"Investigación sobre '{query}':\n"
        f"  - Fuente 1: Paper de Stanford sobre RAG (2025)\n"
        f"  - Fuente 2: Blog de LangChain sobre hybrid search\n"
        f"  - Fuente 3: Benchmark de MTEB para embeddings"
    )
    return {"research": results}


def analyst(state: ResearchTeamState) -> dict:
    analysis = (
        f"Análisis de hallazgos:\n"
        f"  - Patrón 1: Hybrid search supera dense retrieval en 15%\n"
        f"  - Patrón 2: Re-ranking mejora precisión significativamente\n"
        f"  - Contradicción: Paper de Stanford vs benchmark MTEB en embeddings"
    )
    return {"analysis": analysis}


def reporter(state: ResearchTeamState) -> dict:
    report = (
        f"REPORTE: {state['query']}\n"
        f"{'='*40}\n"
        f"{state['research']}\n\n"
        f"{state['analysis']}\n\n"
        f"Conclusión: hybrid search + re-ranking es el estado del arte."
    )
    return {"report": report}


def route_to_worker(state: ResearchTeamState) -> str:
    return state["next_worker"]


builder = StateGraph(ResearchTeamState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_node("reporter", reporter)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_to_worker, {
    "researcher": "researcher",
    "analyst": "analyst",
    "reporter": "reporter",
    "done": END,
})
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")
builder.add_edge("reporter", "supervisor")

graph = builder.compile()

result = graph.invoke({
    "query": "Estado del arte en RAG 2025",
    "research": "", "analysis": "", "report": "",
    "next_worker": "", "iteration": 0,
})

print(result["report"][:80] + "...")
print(f"Iteraciones del supervisor: {result['iteration']}")
# Output esperado:
# REPORTE: Estado del arte en RAG 2025
# ============================================...
# Iteraciones del supervisor: 4

El supervisor ejecutó 4 veces: delegó a researcher (1), recibió research y delegó a analyst (2), recibió analysis y delegó a reporter (3), recibió report y terminó (4). El patrón hub-and-spoke: todo fluye a través del supervisor.

El parámetro name: identificar agentes

Cuando usas create_agent para workers, el parámetro name es clave. Identifica al agente en el sistema, en los logs, y en los handoffs:

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool


@tool
def web_search(query: str) -> str:
    """Busca información en la web sobre un tema."""
    return f"Resultados web para '{query}': 5 artículos encontrados."


researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[web_search],
    name="researcher",
    prompt="Eres un investigador experto. Tu ÚNICO trabajo es buscar información.",
)

print(f"Nombre del agente: {researcher.name}")
# Output esperado:
# Nombre del agente: researcher

El name aparece en los logs, en los handoff tools (transfer_to_researcher), y en los mensajes etiquetados por agente.

El system prompt del supervisor

El prompt del supervisor define las reglas de delegación. Debe ser claro sobre qué worker maneja qué:

SUPERVISOR_PROMPT = """Eres el supervisor de un equipo de investigación.

Tu equipo:
- researcher: busca información en la web y papers académicos
- analyst: analiza datos, encuentra patrones y contradicciones
- writer: genera reportes ejecutivos

Reglas:
1. Para buscar información → delega al researcher
2. Para analizar hallazgos → delega al analyst
3. Para generar el reporte final → delega al writer
4. Cuando el reporte esté completo → responde directamente al usuario

NO hagas el trabajo tú mismo. SIEMPRE delega al especialista correcto.
"""

Un buen prompt de supervisor:

• ✅ Lista los workers disponibles con sus especialidades
• ✅ Define reglas claras de routing
• ✅ Especifica cuándo terminar
• ❌ No intenta entender el contenido — solo coordina

LLM Supervisor con create_supervisor

La forma idiomática de construir un supervisor en LangGraph: create_supervisor del paquete langgraph-supervisor. Crea handoff tools automáticamente y el LLM decide a quién delegar.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph_supervisor import create_supervisor


@tool
def web_search(query: str) -> str:
    """Busca información en la web sobre un tema."""
    return f"Resultados para '{query}': RAG combina retrieval con generation para mejorar respuestas de LLMs."


@tool
def analyze_text(text: str) -> str:
    """Analiza un texto y extrae patrones clave."""
    return f"Análisis: tendencia principal en '{text[:30]}...' es hybrid search + re-ranking."


researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[web_search],
    name="researcher",
    prompt="Eres un investigador. Busca información relevante y reporta tus hallazgos.",
)

analyst = create_agent(
    init_chat_model("openai:gpt-4.1"),
    tools=[analyze_text],
    name="analyst",
    prompt="Eres un analista. Sintetiza la información y encuentra patrones.",
)

workflow = create_supervisor(
    [researcher, analyst],
    model=init_chat_model("openai:gpt-4.1"),
    prompt=(
        "Eres un supervisor de investigación. "
        "Delega búsqueda de información al researcher. "
        "Delega análisis de patrones al analyst. "
        "Cuando tengas suficiente información, genera la respuesta final."
    ),
)

app = workflow.compile()

result = app.invoke({
    "messages": [("user", "¿Cuál es el estado actual de RAG en 2025?")]
})

print(result["messages"][-1].content)
# Output esperado (varía por LLM):
# Basado en la investigación y análisis, el estado actual de RAG en 2025
# se centra en hybrid search combinado con re-ranking...

create_supervisor hace tres cosas automáticamente:

1. Crea handoff tools (transfer_to_researcher, transfer_to_analyst) que el LLM puede llamar
2. Conecta el supervisor con los workers en un grafo
3. Maneja el flujo de mensajes entre supervisor y workers

Cómo funciona el handoff

El handoff no es mágico — es un tool call. create_supervisor genera tools como transfer_to_researcher y transfer_to_analyst. Cuando el LLM del supervisor decide llamar a uno, el framework transfiere el control al worker correspondiente. El worker ejecuta sus tools, genera su respuesta, y el control regresa al supervisor. El supervisor evalúa: ¿necesita más? → delega a otro. ¿Suficiente? → genera respuesta final.

Stop conditions: cuándo terminar

El supervisor necesita saber cuándo parar. Hay tres estrategias:

1. El supervisor decide (default en create_supervisor)

El LLM del supervisor decide cuándo tiene suficiente información y responde directamente al usuario en lugar de delegar a otro worker.

2. Iteration limit

Agrega un max_iterations al estado y verifica en el supervisor:

def supervisor_with_limit(state):
    if state["iteration"] >= state["max_iterations"]:
        return {"next_worker": "done"}
    # ... routing normal ...

Siempre incluye un iteration limit como safety net, incluso con LLM supervisor.

3. Quality check

El supervisor evalúa si el resultado cumple un criterio de calidad antes de terminar. Útil con LLM supervisor.

Simple vs LLM supervisor: cuándo usar cada uno

La regla: empieza con un supervisor simple. Si el routing por reglas no cubre tus casos, upgradeá a LLM. La mayoría de los sistemas de producción usan un híbrido: reglas para los casos comunes, LLM para los edge cases.

Optimización de modelos por worker

Una de las ventajas más concretas de multi-agent: cada worker usa el modelo óptimo para su tarea.

researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),  # barato: $0.40/M input
    tools=[web_search],
    name="researcher",
    prompt="Busca información relevante.",
)

analyst = create_agent(
    init_chat_model("openai:gpt-4.1"),  # potente: $2.00/M input
    tools=[analyze_data],
    name="analyst",
    prompt="Analiza datos y encuentra patrones complejos.",
)

El researcher no necesita un modelo potente — buscar en la web es simple. El analyst sí necesita razonamiento complejo. Resultado: ~60% de ahorro en la parte de búsqueda sin sacrificar calidad en el análisis.

Recopilando resultados: output_mode

create_supervisor tiene un parámetro output_mode que controla cuánta información del worker llega al supervisor:

• "last_message" (default): solo el último mensaje de cada worker. Menos tokens, más eficiente.
• "full_history": todo el historial del worker. Más contexto, útil si el supervisor necesita ver el razonamiento completo.

Empieza con "last_message". Solo cambia a "full_history" si el supervisor toma malas decisiones por falta de contexto.

Troubleshooting

Problema 1: "El supervisor entra en un loop infinito entre dos workers"

Síntoma: El supervisor delega al researcher, recibe resultado, delega al analyst, recibe resultado, y vuelve al researcher indefinidamente.

Causa: El prompt del supervisor no tiene instrucciones claras sobre cuándo terminar. El LLM no sabe que "suficiente" significa responder directamente.

Solución: Agrega una instrucción explícita al prompt: "Cuando tengas suficiente información para responder la consulta del usuario, responde directamente en lugar de delegar a otro agente." También considera un iteration limit como safety net.

Problema 2: "El supervisor siempre delega al mismo worker"

Síntoma: Independientemente de la consulta, el supervisor siempre llama a transfer_to_researcher.

Causa: El prompt del supervisor no diferencia claramente las especialidades de cada worker. O el nombre de los workers no es descriptivo.

Solución: Usa nombres descriptivos (researcher, analyst, no agent_1, agent_2). Describe la especialidad de cada worker en el prompt del supervisor: "researcher busca información, analyst analiza datos."

Problema 3: "create_supervisor no encuentra los agents"

Síntoma: Error al compilar el workflow con create_supervisor.

Causa: Los agents pasados a create_supervisor no tienen el atributo name o no son objetos Pregel compilados.

Solución: Verifica que cada agent fue creado con create_agent (que retorna un graph compilado) y que tiene name definido. create_agent(model, tools=[...], name="mi_agente") — el name es obligatorio para multi-agent.

Problema 4: "Los workers no ven el contexto de otros workers"

Síntoma: El analyst no puede ver lo que encontró el researcher. Causa: Con output_mode="last_message", solo el último mensaje del worker se propaga. Solución: Cambia a output_mode="full_history" o diseña tus workers para consolidar toda la información en un solo mensaje final.

Ejercicios

Ejercicio 1: Supervisor manual con 2 workers (Fácil)

Crea un supervisor manual que coordina translator (traduce al inglés) y summarizer (resume texto). El supervisor delega en orden: translator → summarizer.

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


class TranslateState(TypedDict):
    original_text: str
    translated: str
    summary: str
    next_worker: str
    step: int


def supervisor(state: TranslateState) -> dict:
    if state["step"] == 0:
        return {"next_worker": "translator", "step": 1}
    elif state["step"] == 1:
        return {"next_worker": "summarizer", "step": 2}
    return {"next_worker": "done", "step": 3}


def translator(state: TranslateState) -> dict:
    return {"translated": f"[EN] Translation of: {state['original_text'][:50]}..."}


def summarizer(state: TranslateState) -> dict:
    return {"summary": f"Summary: {state['translated'][:30]}... (3 key points)"}


def route(state: TranslateState) -> str:
    return state["next_worker"]


builder = StateGraph(TranslateState)
builder.add_node("supervisor", supervisor)
builder.add_node("translator", translator)
builder.add_node("summarizer", summarizer)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "translator": "translator", "summarizer": "summarizer", "done": END,
})
builder.add_edge("translator", "supervisor")
builder.add_edge("summarizer", "supervisor")

graph = builder.compile()
result = graph.invoke({
    "original_text": "La inteligencia artificial está transformando la industria financiera.",
    "translated": "", "summary": "", "next_worker": "", "step": 0,
})

print(f"Traducido: {result['translated']}")
print(f"Resumen: {result['summary']}")
# Output esperado:
# Traducido: [EN] Translation of: La inteligencia artificial está transformando la ...
# Resumen: Summary: [EN] Translation of: La inte... (3 key points)

Ejercicio 2: Agregar iteration limit al supervisor (Fácil)

Modifica el supervisor manual del ejemplo principal para que tenga un max_iterations configurable. Si el supervisor excede el límite, debe terminar con lo que tenga disponible. Prueba con max_iterations=2 (solo research, sin analysis ni report).

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


class LimitedState(TypedDict):
    query: str
    research: str
    analysis: str
    next_worker: str
    iteration: int
    max_iterations: int


def supervisor(state: LimitedState) -> dict:
    iteration = state.get("iteration", 0)

    if iteration >= state["max_iterations"]:
        return {"next_worker": "done", "iteration": iteration + 1}

    if iteration == 0:
        return {"next_worker": "researcher", "iteration": 1}
    elif iteration == 1:
        return {"next_worker": "analyst", "iteration": 2}
    return {"next_worker": "done", "iteration": iteration + 1}


def researcher(state: LimitedState) -> dict:
    return {"research": f"Investigación sobre '{state['query']}': 5 fuentes"}


def analyst(state: LimitedState) -> dict:
    return {"analysis": f"Análisis de: {state['research'][:30]}..."}


def route(state: LimitedState) -> str:
    return state["next_worker"]


builder = StateGraph(LimitedState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "researcher": "researcher",
    "analyst": "analyst",
    "done": END,
})
builder.add_edge("researcher", "supervisor")
builder.add_edge("analyst", "supervisor")

graph = builder.compile()

result = graph.invoke({
    "query": "RAG 2025", "research": "", "analysis": "",
    "next_worker": "", "iteration": 0, "max_iterations": 2,
})

assert result["research"] != "" and result["analysis"] == ""
print(f"Research: {result['research']}")
print(f"Analysis vacío: {result['analysis'] == ''}")
print("✅ Iteration limit: solo research completado, analyst no ejecutó")
# Output esperado:
# Research: Investigación sobre 'RAG 2025': 5 fuentes
# Analysis vacío: True
# ✅ Iteration limit: solo research completado, analyst no ejecutó

Ejercicio 3: Supervisor con collect → validate → format y retry (Medio)

Crea un sistema con 3 workers: data_collector, validator, y formatter. El supervisor delega en orden. Si el validator rechaza los datos (primera vez retorna parciales, segunda vez completos), el supervisor vuelve a data_collector. Máximo 2 retries.

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


class PipelineState(TypedDict):
    source: str
    raw_data: str
    is_valid: bool
    formatted: str
    next_worker: str
    phase: str
    retries: int


def supervisor(state: PipelineState) -> dict:
    phase = state.get("phase", "start")
    retries = state.get("retries", 0)
    if phase == "start":
        return {"next_worker": "data_collector", "phase": "collecting"}
    if phase == "collecting":
        return {"next_worker": "validator", "phase": "validating"}
    if phase == "validating":
        if state.get("is_valid"):
            return {"next_worker": "formatter", "phase": "formatting"}
        if retries < 2:
            return {"next_worker": "data_collector", "phase": "collecting", "retries": retries + 1}
        return {"next_worker": "formatter", "phase": "formatting"}
    return {"next_worker": "done", "phase": "complete"}


def data_collector(state: PipelineState) -> dict:
    if state.get("retries", 0) == 0:
        return {"raw_data": "datos_parciales_incompletos"}
    return {"raw_data": f"datos_completos_de_{state['source']}"}


def validator(state: PipelineState) -> dict:
    return {"is_valid": "completos" in state.get("raw_data", "")}


def formatter(state: PipelineState) -> dict:
    prefix = "✅ VÁLIDOS" if state.get("is_valid") else "⚠️ PARCIALES"
    return {"formatted": f"{prefix}: {state['raw_data']}"}


def route(state: PipelineState) -> str:
    return state["next_worker"]


builder = StateGraph(PipelineState)
builder.add_node("supervisor", supervisor)
builder.add_node("data_collector", data_collector)
builder.add_node("validator", validator)
builder.add_node("formatter", formatter)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "data_collector": "data_collector", "validator": "validator",
    "formatter": "formatter", "done": END,
})
for w in ["data_collector", "validator", "formatter"]:
    builder.add_edge(w, "supervisor")

graph = builder.compile()
result = graph.invoke({
    "source": "API_ventas", "raw_data": "", "is_valid": False,
    "formatted": "", "next_worker": "", "phase": "start", "retries": 0,
})

print(f"Válidos: {result['is_valid']} | Retries: {result['retries']}")
print(f"Resultado: {result['formatted']}")
# Output esperado:
# Válidos: True | Retries: 1
# Resultado: ✅ VÁLIDOS: datos_completos_de_API_ventas

Ejercicio 4: Supervisor con create_supervisor (Medio)

Usa create_supervisor para crear un sistema con 2 workers: un researcher que busca información y un critic que evalúa la calidad. El supervisor debe coordinarlos automáticamente. Necesitas API key.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph_supervisor import create_supervisor


@tool
def search_web(query: str) -> str:
    """Busca información en la web."""
    return (
        f"Resultados para '{query}': "
        "1) RAG mejora precisión en 40%, "
        "2) Hybrid search supera dense retrieval, "
        "3) Fine-tuning complementa RAG en dominios específicos."
    )


@tool
def evaluate_quality(text: str) -> str:
    """Evalúa la calidad y completitud de un texto."""
    word_count = len(text.split())
    if word_count < 10:
        return "CALIDAD: Baja. Necesita más detalle y fuentes."
    return "CALIDAD: Aceptable. Información suficiente para un resumen."


researcher = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[search_web],
    name="researcher",
    prompt="Eres un investigador. Busca información completa sobre el tema solicitado.",
)

critic = create_agent(
    init_chat_model("openai:gpt-4.1-mini"),
    tools=[evaluate_quality],
    name="critic",
    prompt="Eres un crítico de calidad. Evalúa si la información recopilada es suficiente.",
)

workflow = create_supervisor(
    [researcher, critic],
    model=init_chat_model("openai:gpt-4.1"),
    prompt=(
        "Eres un supervisor de investigación. "
        "Primero delega al researcher para buscar información. "
        "Luego delega al critic para evaluar la calidad. "
        "Si la calidad es baja, pide al researcher que busque más. "
        "Si la calidad es aceptable, genera un resumen final."
    ),
)

app = workflow.compile()

result = app.invoke({
    "messages": [("user", "¿Cuál es el impacto de RAG en aplicaciones de AI?")]
})

print(result["messages"][-1].content[:120] + "...")
# Output esperado (varía por LLM):
# Basado en la investigación y evaluación de calidad: RAG mejora la precisión en un 40%...

Ejercicio 5: Supervisor con retry condicional (Avanzado)

Crea un supervisor manual que coordina fetcher y validator. El fetcher retorna datos incompletos las primeras 2 veces (simula con un counter global). El validator verifica si los datos son completos. Si no, el supervisor re-delega al fetcher. Máximo 3 intentos.

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

fetch_call_count = 0


class RetryState(TypedDict):
    data: str
    is_complete: bool
    attempt: int
    max_attempts: int
    next_worker: str
    phase: str


def supervisor(state: RetryState) -> dict:
    phase = state.get("phase", "init")
    attempt = state.get("attempt", 0)

    if phase == "init":
        return {"next_worker": "fetcher", "phase": "fetching", "attempt": 1}
    if phase == "fetching":
        return {"next_worker": "validator", "phase": "validating"}
    if phase == "validating":
        if state["is_complete"]:
            return {"next_worker": "done", "phase": "complete"}
        if attempt < state["max_attempts"]:
            return {"next_worker": "fetcher", "phase": "fetching", "attempt": attempt + 1}
        return {"next_worker": "done", "phase": "complete_partial"}
    return {"next_worker": "done", "phase": "error"}


def fetcher(state: RetryState) -> dict:
    global fetch_call_count
    fetch_call_count += 1
    if fetch_call_count < 3:
        return {"data": "datos_parciales"}
    return {"data": "datos_completos_verificados"}


def validator(state: RetryState) -> dict:
    return {"is_complete": "completos" in state.get("data", "")}


def route(state: RetryState) -> str:
    return state["next_worker"]


builder = StateGraph(RetryState)
builder.add_node("supervisor", supervisor)
builder.add_node("fetcher", fetcher)
builder.add_node("validator", validator)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route, {
    "fetcher": "fetcher", "validator": "validator", "done": END,
})
builder.add_edge("fetcher", "supervisor")
builder.add_edge("validator", "supervisor")

graph = builder.compile()
fetch_call_count = 0

result = graph.invoke({
    "data": "", "is_complete": False, "attempt": 0,
    "max_attempts": 3, "next_worker": "", "phase": "init",
})

print(f"Datos: {result['data']}")
print(f"¿Completos?: {result['is_complete']}")
print(f"Intentos: {result['attempt']}")
print(f"Llamadas a fetcher: {fetch_call_count}")
# Output esperado:
# Datos: datos_completos_verificados
# ¿Completos?: True
# Intentos: 3
# Llamadas a fetcher: 3

Resumen

• El supervisor es un coordinador, no un ejecutor. Su trabajo es decidir qué worker necesita trabajar, no hacer el trabajo. Piensa en un gerente de proyecto: sabe a quién asignar cada tarea, pero no necesita saber programar
• Supervisor manual con StateGraph es el punto de partida: routing basado en reglas (if/else), sin costo adicional de LLM, debugging trivial, flujo predecible. Ideal para pipelines con flujo fijo
• create_supervisor es la forma idiomática para routing con LLM: crea handoff tools automáticamente, el modelo decide a quién delegar basándose en contexto, se adapta a consultas inesperadas. Ideal para consultas variadas
• Empieza simple, escala si es necesario. La mayoría de los sistemas de producción empiezan con routing por reglas y solo agregan LLM routing cuando los casos de uso lo requieren
• Optimización de modelos por worker es una ventaja concreta: modelo barato para búsqueda, modelo potente para análisis. Mismo resultado, menor costo
• Stop conditions previenen loops infinitos: iteration limit como safety net, quality check como criterio de terminación, LLM decision como default en create_supervisor
• El name es obligatorio en multi-agent. Identifica a cada worker en logs, handoffs, y estado compartido

Próxima cápsula: Pattern Handoffs — aprenderás cómo un agente transfiere el control directamente a otro sin supervisor central, cuándo usar handoffs vs supervisor, y cómo implementar cadenas de especialistas.

Recursos adicionales

1. LangGraph — Multi-Agent Supervisor — Documentación oficial del patrón supervisor con ejemplos y variaciones
2. create_supervisor API Reference — Referencia completa de create_supervisor: parámetros, opciones, y ejemplos
3. langgraph-supervisor GitHub — Repositorio oficial con código fuente y ejemplos avanzados
4. How to build a multi-agent supervisor — Guía práctica paso a paso
5. Multi-Agent Architectures — LangChain Blog — Comparación de patrones multi-agent: supervisor, handoffs, hierarchical

Módulo 10 — LangChain & LangGraph: From Chains to Agents