Módulo 10: Multi-Agent Systems

Pattern Subagents

Siguiente →

Descripción de la cápsula

En las cápsulas anteriores aprendiste dos formas de coordinar agentes: el supervisor (un agente central que decide) y los handoffs (transferencia directa entre agentes). Ambos comparten una característica: el estado fluye completo entre los agentes. El researcher ve su investigación, el analyst ve la investigación + su análisis, el writer ve todo lo anterior. Con cada paso, el contexto crece.

Los subagents resuelven este problema con aislamiento de contexto. Un agente padre (supervisor) crea una subtarea, la envía a un subagent, y el subagent trabaja en su propio espacio — sin ver la conversación completa del padre. Cuando termina, devuelve solo el resultado. El padre integra ese resultado y continúa. El contexto del padre no crece con cada subtarea.

La diferencia fundamental: en handoffs, el agente destino recibe el contexto acumulado. En subagents, el agente hijo recibe solo su instrucción y trabaja desde cero. Es la diferencia entre "pásame todo el expediente" y "dime qué necesitas que investigue."

El problema del context bloat

Para entender por qué los subagents existen, hay que ver el problema que resuelven:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class SharedState(TypedDict):
    task: str
    agent_a_work: str
    agent_b_work: str
    agent_c_work: str
    context_size: int
    log: Annotated[list[str], operator.add]

def agent_a(state: SharedState) -> Command:
    work = "A" * 200
    size = len(state["task"]) + len(work)
    return Command(
        goto="agent_b",
        update={
            "agent_a_work": work,
            "context_size": size,
            "log": [f"agent_a: generó {len(work)} chars, contexto total: {size}"],
        },
    )

def agent_b(state: SharedState) -> Command:
    work = "B" * 300
    size = state["context_size"] + len(work)
    return Command(
        goto="agent_c",
        update={
            "agent_b_work": work,
            "context_size": size,
            "log": [f"agent_b: generó {len(work)} chars, contexto total: {size}"],
        },
    )

def agent_c(state: SharedState) -> dict:
    work = "C" * 400
    size = state["context_size"] + len(work)
    return {
        "agent_c_work": work,
        "context_size": size,
        "log": [f"agent_c: generó {len(work)} chars, contexto total: {size}"],
    }

builder = StateGraph(SharedState)
builder.add_node("agent_a", agent_a)
builder.add_node("agent_b", agent_b)
builder.add_node("agent_c", agent_c)

builder.add_edge(START, "agent_a")
builder.add_edge("agent_c", END)

graph = builder.compile()

result = graph.invoke({
    "task": "Analizar mercado AI",
    "agent_a_work": "", "agent_b_work": "",
    "agent_c_work": "", "context_size": 0, "log": [],
})

print("=== Context Bloat con estado compartido ===\n")
for entry in result["log"]:
    print(f"  {entry}")
print(f"\nContexto final: {result['context_size']} chars")
print(f"Agent C necesitaba ver todo? Probablemente no.")
# Output esperado:
# === Context Bloat con estado compartido ===
#
#   agent_a: generó 200 chars, contexto total: 218
#   agent_b: generó 300 chars, contexto total: 518
#   agent_c: generó 400 chars, contexto total: 918
#
# Contexto final: 918 chars
# Agent C necesitaba ver todo? Probablemente no.

Con cada handoff, el contexto crece. En este ejemplo simplificado son chars, pero en un sistema real son mensajes de LLM, resultados de tools, razonamientos intermedios. Con 5 agentes y herramientas que retornan documentos largos, fácilmente llegas a miles de tokens que el último agente no necesita.

Con subagents, cada agente trabaja en un espacio limpio. El contexto del padre permanece enfocado.

Cómo funcionan los subagents

El flujo de un subagent tiene 5 pasos:

  1. El padre crea una descripción de la subtarea — "Investiga tendencias AI 2025"
  2. El subagent recibe SOLO esa descripción — no ve la conversación del padre ni el trabajo de otros subagents
  3. El subagent ejecuta con sus propias tools y prompt — trabaja de forma autónoma en su contexto aislado
  4. El subagent retorna un resultado conciso — solo el output final, no todo su razonamiento interno
  5. El padre integra el resultado — lo incorpora a su propio contexto y decide el siguiente paso

La clave: el padre no envía toda su conversación al subagent. Y el subagent no devuelve todo su proceso interno al padre. Ambos mantienen sus contextos separados.

Implementación básica: subagent como tool

El patrón más directo: crear un agente con create_agent y envolverlo como una @tool que el agente padre puede llamar:

from dotenv import load_dotenv
load_dotenv()

from langchain.tools import tool
from langchain.agents import create_agent

research_subagent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    prompt="Eres un investigador especializado. Responde de forma concisa con hallazgos clave.",
)

analysis_subagent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    prompt="Eres un analista de datos. Identifica tendencias y riesgos. Sé conciso.",
)

@tool("research", description="Investiga un tema y retorna hallazgos clave")
def call_researcher(query: str) -> str:
    result = research_subagent.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return result["messages"][-1].content

@tool("analyze", description="Analiza información y retorna tendencias y riesgos")
def call_analyst(data: str) -> str:
    result = analysis_subagent.invoke({
        "messages": [{"role": "user", "content": data}]
    })
    return result["messages"][-1].content

supervisor = create_agent(
    "openai:gpt-4.1-mini",
    tools=[call_researcher, call_analyst],
    prompt=(
        "Eres un supervisor de investigación. "
        "Usa la herramienta 'research' para investigar temas y "
        "'analyze' para analizar los hallazgos. "
        "Combina los resultados en un reporte final conciso."
    ),
)

result = supervisor.invoke({
    "messages": [{"role": "user", "content": "Investiga y analiza el estado de LLMs en producción"}]
})

for msg in result["messages"]:
    msg.pretty_print()
# Output esperado (varía según el modelo):
# ================================ Human Message =================================
# Investiga y analiza el estado de LLMs en producción
# ================================== Ai Message ==================================
# Voy a investigar este tema y luego analizarlo.
# [tool call: research("Estado actual de LLMs en producción empresarial")]
# ================================== Tool Message ================================
# Los LLMs en producción han visto avances significativos: ...
# ================================== Ai Message ==================================
# [tool call: analyze("Los LLMs en producción han visto...")]
# ================================== Tool Message ================================
# Tendencias: adopción creciente. Riesgos: costos y alucinaciones...
# ================================== Ai Message ==================================
# REPORTE: Los LLMs en producción muestran adopción creciente...

El aislamiento de contexto ocurre naturalmente:

  • ✅ El research_subagent solo ve: "Investiga el estado de LLMs en producción" — no ve la conversación completa del supervisor
  • ✅ El analysis_subagent solo ve los hallazgos que el supervisor le pasa — no ve la pregunta original del usuario ni el razonamiento del supervisor
  • ✅ El supervisor solo ve los resultados finales de cada subagent — no ve los razonamientos internos de cada uno

Subagent como subgrafo compilado

Cuando necesitas más control sobre el subagent (estado custom, nodos múltiples, flujos internos), puedes crear un StateGraph compilado e invocarlo desde un nodo del grafo padre:

from dotenv import load_dotenv
load_dotenv()

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

class SubagentState(TypedDict):
    task: str
    result: str

class ParentState(TypedDict):
    objective: str
    research_result: str
    analysis_result: str
    final_report: str
    log: Annotated[list[str], operator.add]

def do_research(state: SubagentState) -> dict:
    return {"result": f"Investigación sobre '{state['task']}': 5 fuentes, 3 hallazgos clave"}

def do_analysis(state: SubagentState) -> dict:
    return {"result": f"Análisis de '{state['task']}': tendencia alcista, 2 riesgos"}

sub_research_builder = StateGraph(SubagentState)
sub_research_builder.add_node("work", do_research)
sub_research_builder.add_edge(START, "work")
sub_research_builder.add_edge("work", END)
research_subgraph = sub_research_builder.compile()

sub_analysis_builder = StateGraph(SubagentState)
sub_analysis_builder.add_node("work", do_analysis)
sub_analysis_builder.add_edge(START, "work")
sub_analysis_builder.add_edge("work", END)
analysis_subgraph = sub_analysis_builder.compile()

def research_node(state: ParentState) -> dict:
    sub_result = research_subgraph.invoke({"task": state["objective"], "result": ""})
    return {
        "research_result": sub_result["result"],
        "log": [f"research subagent retornó: {len(sub_result['result'])} chars"],
    }

def analysis_node(state: ParentState) -> dict:
    sub_result = analysis_subgraph.invoke({"task": state["research_result"], "result": ""})
    return {
        "analysis_result": sub_result["result"],
        "log": [f"analysis subagent retornó: {len(sub_result['result'])} chars"],
    }

def report_node(state: ParentState) -> dict:
    report = (
        f"REPORTE\n"
        f"Objetivo: {state['objective']}\n"
        f"Research: {state['research_result']}\n"
        f"Analysis: {state['analysis_result']}"
    )
    return {"final_report": report, "log": ["report generado"]}

builder = StateGraph(ParentState)
builder.add_node("research", research_node)
builder.add_node("analysis", analysis_node)
builder.add_node("report", report_node)

builder.add_edge(START, "research")
builder.add_edge("research", "analysis")
builder.add_edge("analysis", "report")
builder.add_edge("report", END)

graph = builder.compile()

result = graph.invoke({
    "objective": "Estado de AI agents en 2025",
    "research_result": "", "analysis_result": "",
    "final_report": "", "log": [],
})

print(result["final_report"])
print(f"\nFlujo:")
for entry in result["log"]:
    print(f"  → {entry}")
# Output esperado:
# REPORTE
# Objetivo: Estado de AI agents en 2025
# Research: Investigación sobre 'Estado de AI agents en 2025': 5 fuentes, 3 hallazgos clave
# Analysis: Análisis de 'Investigación sobre 'Estado de AI agents en 2025': 5 fuentes, 3 hallazgos clave': tendencia alcista, 2 riesgos
#
# Flujo:
#   → research subagent retornó: 73 chars
#   → analysis subagent retornó: 113 chars
#   → report generado

El aislamiento es explícito: cada subgrafo tiene su propio SubagentState separado del ParentState. El nodo padre construye el input ({"task": state["objective"]}), invoca el subgrafo, y extrae solo sub_result["result"]. El subgrafo nunca ve ParentState y el padre nunca ve el estado interno del subgrafo.

Aislamiento de contexto: qué ve cada quién

Para que el aislamiento de contexto quede claro, veamos explícitamente qué información tiene disponible cada agente:

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    instruction: str
    result: str

class ParentState(TypedDict):
    user_query: str
    conversation_history: list[str]
    internal_reasoning: str
    sub_result_1: str
    sub_result_2: str
    final_answer: str
    log: Annotated[list[str], operator.add]

def sub_worker(state: SubState) -> dict:
    visible_keys = list(state.keys())
    result = (
        f"Subagent trabajando. "
        f"Ve: {visible_keys}. "
        f"Instrucción: '{state['instruction']}'. "
        f"NO ve: user_query, conversation_history, internal_reasoning"
    )
    return {"result": result}

sub_builder = StateGraph(SubState)
sub_builder.add_node("work", sub_worker)
sub_builder.add_edge(START, "work")
sub_builder.add_edge("work", END)
subagent = sub_builder.compile()

def step_1(state: ParentState) -> dict:
    sub_result = subagent.invoke({
        "instruction": "Busca información sobre Python async",
        "result": "",
    })
    return {
        "sub_result_1": sub_result["result"],
        "conversation_history": state["conversation_history"] + ["step_1 completado"],
        "internal_reasoning": "Necesito datos sobre async para responder al usuario",
        "log": ["step_1: subagent invocado con contexto aislado"],
    }

def step_2(state: ParentState) -> dict:
    sub_result = subagent.invoke({
        "instruction": "Busca benchmarks de async vs sync en Python",
        "result": "",
    })
    return {
        "sub_result_2": sub_result["result"],
        "conversation_history": state["conversation_history"] + ["step_2 completado"],
        "log": ["step_2: segundo subagent invocado, también aislado"],
    }

def synthesize(state: ParentState) -> dict:
    parent_sees = {
        "user_query": state["user_query"][:30],
        "history_length": len(state["conversation_history"]),
        "reasoning": state["internal_reasoning"][:30],
        "sub_result_1": state["sub_result_1"][:40],
        "sub_result_2": state["sub_result_2"][:40],
    }
    return {
        "final_answer": f"Padre ve todo su contexto: {parent_sees}",
        "log": ["synthesize: padre integra resultados"],
    }

builder = StateGraph(ParentState)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("synthesize", synthesize)

builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "synthesize")
builder.add_edge("synthesize", END)

graph = builder.compile()

result = graph.invoke({
    "user_query": "Explícame async/await en Python",
    "conversation_history": [],
    "internal_reasoning": "",
    "sub_result_1": "", "sub_result_2": "",
    "final_answer": "", "log": [],
})

print("=== Qué ve el subagent ===")
print(f"  {result['sub_result_1']}\n")
print("=== Qué ve el padre ===")
print(f"  {result['final_answer']}\n")
print("=== Flujo ===")
for entry in result["log"]:
    print(f"  → {entry}")
# Output esperado:
# === Qué ve el subagent ===
#   Subagent trabajando. Ve: ['instruction', 'result']. Instrucción: 'Busca información sobre Python async'. NO ve: user_query, conversation_history, internal_reasoning
#
# === Qué ve el padre ===
#   Padre ve todo su contexto: {'user_query': 'Explícame async/await en Python', 'history_length': 2, 'reasoning': 'Necesito datos sobre async para', 'sub_result_1': 'Subagent trabajando. Ve: ['instruction', ', 'sub_result_2': 'Subagent trabajando. Ve: ['instruction', '}
#
# === Flujo ===
#   → step_1: subagent invocado con contexto aislado
#   → step_2: segundo subagent invocado, también aislado
#   → synthesize: padre integra resultados

El subagent ve: ['instruction', 'result']. No ve user_query, conversation_history, ni internal_reasoning. El padre ve todo su propio estado y los resultados de los subagents. Cada quién trabaja en su propio espacio.

Subagents en paralelo

Una de las ventajas más prácticas de los subagents: como trabajan en contexto aislado, puedes ejecutar varios al mismo tiempo sin conflictos de estado:

from dotenv import load_dotenv
load_dotenv()

import concurrent.futures
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class SubState(TypedDict):
    source: str
    query: str
    result: str

class ParentState(TypedDict):
    query: str
    web_results: str
    papers_results: str
    docs_results: str
    combined_report: str
    log: Annotated[list[str], operator.add]

def search_source(state: SubState) -> dict:
    return {
        "result": f"[{state['source']}] 3 resultados para '{state['query']}'"
    }

sub_builder = StateGraph(SubState)
sub_builder.add_node("search", search_source)
sub_builder.add_edge(START, "search")
sub_builder.add_edge("search", END)
search_subagent = sub_builder.compile()

def parallel_research(state: ParentState) -> dict:
    sources = [
        {"source": "web", "query": state["query"], "result": ""},
        {"source": "papers", "query": state["query"], "result": ""},
        {"source": "docs", "query": state["query"], "result": ""},
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        futures = {
            executor.submit(search_subagent.invoke, src): src["source"]
            for src in sources
        }
        results = {}
        for future in concurrent.futures.as_completed(futures):
            source_name = futures[future]
            results[source_name] = future.result()["result"]

    return {
        "web_results": results["web"],
        "papers_results": results["papers"],
        "docs_results": results["docs"],
        "log": [f"parallel_research: 3 subagents completados ({', '.join(results.keys())})"],
    }

def combine_results(state: ParentState) -> dict:
    report = (
        f"REPORTE COMBINADO para '{state['query']}':\n"
        f"  Web: {state['web_results']}\n"
        f"  Papers: {state['papers_results']}\n"
        f"  Docs: {state['docs_results']}"
    )
    return {"combined_report": report, "log": ["combine: reporte generado"]}

builder = StateGraph(ParentState)
builder.add_node("parallel_research", parallel_research)
builder.add_node("combine_results", combine_results)

builder.add_edge(START, "parallel_research")
builder.add_edge("parallel_research", "combine_results")
builder.add_edge("combine_results", END)

graph = builder.compile()

result = graph.invoke({
    "query": "RAG patterns en producción",
    "web_results": "", "papers_results": "", "docs_results": "",
    "combined_report": "", "log": [],
})

print(result["combined_report"])
print(f"\nLog:")
for entry in result["log"]:
    print(f"  → {entry}")
# Output esperado:
# REPORTE COMBINADO para 'RAG patterns en producción':
#   Web: [web] 3 resultados para 'RAG patterns en producción'
#   Papers: [papers] 3 resultados para 'RAG patterns en producción'
#   Docs: [docs] 3 resultados para 'RAG patterns en producción'
#
# Log:
#   → parallel_research: 3 subagents completados (web, papers, docs)
#   → combine: reporte generado

Los 3 subagents se ejecutan en paralelo con ThreadPoolExecutor. Cada uno trabaja en su propio estado (SubState), busca en su fuente asignada, y retorna su resultado. El nodo padre recopila todos los resultados y los integra. Sin conflictos de estado porque cada subagent tiene su propio espacio.

Single dispatch tool: un tool para N subagents

Cuando tienes muchos subagents, crear un @tool por cada uno escala mal. El patrón single dispatch usa un solo tool parametrizado que invoca al subagent correcto por nombre:

from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    task: str
    result: str

def researcher_work(state: SubState) -> dict:
    return {"result": f"[RESEARCH] Investigación sobre: {state['task'][:40]}"}

def writer_work(state: SubState) -> dict:
    return {"result": f"[WRITER] Contenido generado para: {state['task'][:40]}"}

def reviewer_work(state: SubState) -> dict:
    return {"result": f"[REVIEWER] Review completado de: {state['task'][:40]}"}

def build_subagent(work_fn) -> object:
    b = StateGraph(SubState)
    b.add_node("work", work_fn)
    b.add_edge(START, "work")
    b.add_edge("work", END)
    return b.compile()

REGISTRY = {
    "researcher": build_subagent(researcher_work),
    "writer": build_subagent(writer_work),
    "reviewer": build_subagent(reviewer_work),
}

def dispatch(agent_name: str, task_description: str) -> str:
    """Invoca un subagent por nombre con aislamiento de contexto."""
    if agent_name not in REGISTRY:
        return f"Error: agente '{agent_name}' no encontrado. Disponibles: {list(REGISTRY.keys())}"
    subagent = REGISTRY[agent_name]
    result = subagent.invoke({"task": task_description, "result": ""})
    return result["result"]


class OrchestratorState(TypedDict):
    objective: str
    steps_completed: list[str]
    final_output: str
    log: Annotated[list[str], operator.add]

def orchestrator(state: OrchestratorState) -> dict:
    research = dispatch("researcher", f"Investiga: {state['objective']}")
    writing = dispatch("writer", f"Escribe basándote en: {research}")
    review = dispatch("reviewer", f"Revisa: {writing}")

    return {
        "steps_completed": [research, writing, review],
        "final_output": review,
        "log": [
            f"dispatch → researcher: {research[:40]}...",
            f"dispatch → writer: {writing[:40]}...",
            f"dispatch → reviewer: {review[:40]}...",
        ],
    }

builder = StateGraph(OrchestratorState)
builder.add_node("orchestrator", orchestrator)
builder.add_edge(START, "orchestrator")
builder.add_edge("orchestrator", END)

graph = builder.compile()

result = graph.invoke({
    "objective": "Mejores prácticas para deploying AI agents",
    "steps_completed": [], "final_output": "", "log": [],
})

print(f"Output final: {result['final_output']}")
print(f"\nPasos completados:")
for step in result["steps_completed"]:
    print(f"  {step}")
print(f"\nLog:")
for entry in result["log"]:
    print(f"  → {entry}")
# Output esperado:
# Output final: [REVIEWER] Review completado de: [WRITER] Contenido generado para: [RES
#
# Pasos completados:
#   [RESEARCH] Investigación sobre: Investiga: Mejores prácticas para deployi
#   [WRITER] Contenido generado para: Escribe basándote en: [RESEARCH] Invest
#   [REVIEWER] Review completado de: Revisa: [WRITER] Contenido generado para
#
# Log:
#   → dispatch → researcher: [RESEARCH] Investigación sobre: Inves...
#   → dispatch → writer: [WRITER] Contenido generado para: Escrib...
#   → dispatch → reviewer: [REVIEWER] Review completado de: Revis...

Un solo dispatch(agent_name, task) invoca cualquier subagent registrado. Para agregar un nuevo subagent, solo lo registras en REGISTRY. El orquestador no necesita cambiar.

Con create_agent, el pattern es idéntico pero los subagents son agentes con LLM:

from langchain.tools import tool
from langchain.agents import create_agent

SUBAGENTS = {
    "researcher": create_agent("openai:gpt-4.1-mini", prompt="Eres un investigador..."),
    "writer": create_agent("openai:gpt-4.1-mini", prompt="Eres un escritor..."),
}

@tool
def task(agent_name: str, description: str) -> str:
    """Lanza un subagent para una tarea específica."""
    agent = SUBAGENTS[agent_name]
    result = agent.invoke({"messages": [{"role": "user", "content": description}]})
    return result["messages"][-1].content

Cuándo usar subagents vs handoffs

CriterioSubagentsHandoffs
ContextoAislado — cada subagent ve solo su tareaCompartido — el contexto fluye entre agentes
Context bloatNo — el padre solo ve resultados finalesSí — crece con cada handoff
Ejecución paralelaNatural — subagents independientesDifícil — el flujo es secuencial
Interacción con usuarioNo directa — pasan por el padreDirecta — cada agente puede interactuar
CoordinaciónCentralizada en el padreDistribuida entre agentes
Caso idealMuchas subtareas independientesFlujo secuencial con contexto compartido
Ejemplo"Busca en 3 fuentes en paralelo""Investiga → analiza → escribe"

Regla práctica:

  • ✅ Usa subagents cuando las subtareas son independientes y no necesitan verse entre sí
  • ✅ Usa handoffs cuando el flujo es secuencial y cada agente necesita el contexto del anterior
  • ✅ Combina ambos: handoffs para el flujo principal, subagents para subtareas internas de cada agente

Manejo de errores y timeouts en subagents

Los subagents pueden fallar: el LLM no responde, una tool da error, o el subagent se queda en un loop infinito. El padre necesita manejar estos casos:

from dotenv import load_dotenv
load_dotenv()

import concurrent.futures
from typing import TypedDict, Annotated
import operator
import time
from langgraph.graph import StateGraph, START, END

class SubState(TypedDict):
    task: str
    result: str

class ParentState(TypedDict):
    query: str
    results: dict
    errors: dict
    final_report: str
    log: Annotated[list[str], operator.add]

def fast_agent(state: SubState) -> dict:
    return {"result": f"Resultado rápido para: {state['task'][:30]}"}

def slow_agent(state: SubState) -> dict:
    time.sleep(0.5)
    return {"result": f"Resultado lento para: {state['task'][:30]}"}

def failing_agent(state: SubState) -> dict:
    raise ValueError(f"Error procesando: {state['task'][:20]}")

fast_sub = StateGraph(SubState)
fast_sub.add_node("work", fast_agent)
fast_sub.add_edge(START, "work")
fast_sub.add_edge("work", END)
fast_subagent = fast_sub.compile()

slow_sub = StateGraph(SubState)
slow_sub.add_node("work", slow_agent)
slow_sub.add_edge(START, "work")
slow_sub.add_edge("work", END)
slow_subagent = slow_sub.compile()

fail_sub = StateGraph(SubState)
fail_sub.add_node("work", failing_agent)
fail_sub.add_edge(START, "work")
fail_sub.add_edge("work", END)
failing_subagent = fail_sub.compile()

AGENTS = {
    "fast": fast_subagent,
    "slow": slow_subagent,
    "failing": failing_subagent,
}

def safe_invoke(agent_name: str, task: str, timeout: float = 2.0) -> dict:
    """Invoca un subagent con timeout y manejo de errores."""
    agent = AGENTS[agent_name]
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(agent.invoke, {"task": task, "result": ""})
            result = future.result(timeout=timeout)
            return {"success": True, "result": result["result"]}
    except concurrent.futures.TimeoutError:
        return {"success": False, "result": f"TIMEOUT: {agent_name} excedió {timeout}s"}
    except Exception as e:
        return {"success": False, "result": f"ERROR en {agent_name}: {str(e)[:50]}"}

def orchestrate(state: ParentState) -> dict:
    agents_to_run = ["fast", "slow", "failing"]
    results = {}
    errors = {}

    for name in agents_to_run:
        outcome = safe_invoke(name, state["query"])
        if outcome["success"]:
            results[name] = outcome["result"]
        else:
            errors[name] = outcome["result"]

    report_parts = [f"Query: {state['query']}"]
    if results:
        report_parts.append(f"Exitosos ({len(results)}): {list(results.keys())}")
    if errors:
        report_parts.append(f"Fallidos ({len(errors)}): {list(errors.keys())}")

    return {
        "results": results,
        "errors": errors,
        "final_report": " | ".join(report_parts),
        "log": [
            f"exitosos: {list(results.keys())}",
            f"errores: {list(errors.keys())}",
        ],
    }

builder = StateGraph(ParentState)
builder.add_node("orchestrate", orchestrate)
builder.add_edge(START, "orchestrate")
builder.add_edge("orchestrate", END)

graph = builder.compile()

result = graph.invoke({
    "query": "Estado de AI agents",
    "results": {}, "errors": {},
    "final_report": "", "log": [],
})

print(f"Reporte: {result['final_report']}")
print(f"\nResultados exitosos:")
for name, res in result["results"].items():
    print(f"  ✅ {name}: {res}")
print(f"\nErrores:")
for name, err in result["errors"].items():
    print(f"  ❌ {name}: {err}")
# Output esperado:
# Reporte: Query: Estado de AI agents | Exitosos (2): ['fast', 'slow'] | Fallidos (1): ['failing']
#
# Resultados exitosos:
#   ✅ fast: Resultado rápido para: Estado de AI agents
#   ✅ slow: Resultado lento para: Estado de AI agents
#
# Errores:
#   ❌ failing: ERROR en failing: Error procesando: Estado de AI agents

La función safe_invoke envuelve cada subagent con:

  • ✅ Timeout: Si el subagent no responde en N segundos, retorna error
  • ✅ Try/except: Captura cualquier excepción del subagent
  • ✅ Resultado estructurado: {"success": bool, "result": str} — el padre sabe qué funcionó y qué no

El padre decide qué hacer con los errores: reintentar, usar resultados parciales, o abortar.

Troubleshooting

Problema 1: "El subagent retorna demasiada información"

Síntoma: El resultado del subagent incluye todo su razonamiento interno, tool calls, y mensajes intermedios. El contexto del padre crece como si fuera un handoff.

Causa: Estás retornando todo el estado del subagent en lugar de solo el resultado final.

Solución: Extrae solo el contenido del último mensaje:

result = subagent.invoke({"messages": [{"role": "user", "content": task}]})
return result["messages"][-1].content

Problema 2: "Subagents en paralelo causan errores de estado"

Síntoma: Al ejecutar subagents en paralelo, algunos fallan o retornan resultados corruptos.

Causa: Los subagents comparten algún recurso mutable (variable global, archivo, conexión).

Solución: Asegúrate de que cada subagent sea verdaderamente independiente. Cada invocación debe crear su propio estado:

result = subagent.invoke({"task": task, "result": ""})

Problema 3: "El subagent no tiene las tools que necesita"

Síntoma: El subagent no puede completar su tarea porque le faltan herramientas.

Causa: Cada subagent tiene su propio conjunto de tools. No hereda las tools del padre.

Solución: Define las tools del subagent al crearlo:

research_subagent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[web_search, paper_search],
    prompt="Eres un investigador con acceso a búsqueda web y papers.",
)

Problema 4: "No puedo debuggear qué hizo el subagent"

Síntoma: El subagent retorna un resultado incorrecto pero no sabes qué pasó internamente.

Causa: Por diseño, el aislamiento de contexto oculta los detalles internos del subagent.

Solución: Agrega logging dentro del subagent, o usa LangSmith para trazar la ejecución completa. También puedes retornar metadata adicional junto con el resultado:

return {
    "result": final_answer,
    "metadata": {"steps_taken": 3, "tools_used": ["web_search"]}
}

Ejercicios

Ejercicio 1: Subagent básico con contexto aislado (Fácil)

Crea un sistema padre-hijo simple. El padre tiene un query y un secret_context (información interna). El subagent solo recibe el query y retorna un resultado. Verifica que el subagent NO puede ver secret_context.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    query: str
    result: str

class ParentState(TypedDict):
    user_query: str
    secret_context: str
    sub_result: str
    log: Annotated[list[str], operator.add]

def subagent_work(state: SubState) -> dict:
    visible = list(state.keys())
    return {"result": f"Subagent ve claves: {visible}. Procesó: '{state['query'][:30]}'"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("work", subagent_work)
sub_builder.add_edge(START, "work")
sub_builder.add_edge("work", END)
subagent = sub_builder.compile()

def parent_node(state: ParentState) -> dict:
    result = subagent.invoke({"query": state["user_query"], "result": ""})
    return {
        "sub_result": result["result"],
        "log": [f"subagent retornó: {result['result'][:50]}..."],
    }

builder = StateGraph(ParentState)
builder.add_node("parent", parent_node)
builder.add_edge(START, "parent")
builder.add_edge("parent", END)

graph = builder.compile()

result = graph.invoke({
    "user_query": "¿Qué es machine learning?",
    "secret_context": "CONFIDENCIAL: presupuesto interno $500k",
    "sub_result": "", "log": [],
})

print(f"Resultado: {result['sub_result']}")
print(f"Secret context intacto: {result['secret_context'][:20]}...")
# Output esperado:
# Resultado: Subagent ve claves: ['query', 'result']. Procesó: '¿Qué es machine learning?'
# Secret context intacto: CONFIDENCIAL: presup...

Ejercicio 2: Tres subagents en paralelo (Medio)

Crea un sistema donde el padre lanza 3 subagents en paralelo: news_agent (busca noticias), social_agent (busca redes sociales), y academic_agent (busca papers). Usa ThreadPoolExecutor para ejecutarlos simultáneamente. Combina los resultados en un resumen.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

import concurrent.futures
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class SubState(TypedDict):
    query: str
    source_type: str
    result: str

class ParentState(TypedDict):
    topic: str
    news: str
    social: str
    academic: str
    summary: str
    log: Annotated[list[str], operator.add]

def search_source(state: SubState) -> dict:
    return {"result": f"[{state['source_type'].upper()}] 5 resultados para '{state['query'][:25]}...'"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("search", search_source)
sub_builder.add_edge(START, "search")
sub_builder.add_edge("search", END)
search_sub = sub_builder.compile()

def parallel_search(state: ParentState) -> dict:
    configs = [
        {"query": state["topic"], "source_type": "news", "result": ""},
        {"query": state["topic"], "source_type": "social", "result": ""},
        {"query": state["topic"], "source_type": "academic", "result": ""},
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        futures = {pool.submit(search_sub.invoke, c): c["source_type"] for c in configs}
        results = {}
        for f in concurrent.futures.as_completed(futures):
            name = futures[f]
            results[name] = f.result()["result"]

    return {
        "news": results["news"],
        "social": results["social"],
        "academic": results["academic"],
        "log": [f"parallel: {len(results)} fuentes completadas"],
    }

def summarize(state: ParentState) -> dict:
    summary = (
        f"RESUMEN sobre '{state['topic']}':\n"
        f"  {state['news']}\n"
        f"  {state['social']}\n"
        f"  {state['academic']}"
    )
    return {"summary": summary, "log": ["resumen generado"]}

builder = StateGraph(ParentState)
builder.add_node("search", parallel_search)
builder.add_node("summarize", summarize)
builder.add_edge(START, "search")
builder.add_edge("search", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "LangGraph multi-agent patterns",
    "news": "", "social": "", "academic": "",
    "summary": "", "log": [],
})

print(result["summary"])
print(f"\nLog: {result['log']}")
# Output esperado:
# RESUMEN sobre 'LangGraph multi-agent patterns':
#   [NEWS] 5 resultados para 'LangGraph multi-agent pat...'
#   [SOCIAL] 5 resultados para 'LangGraph multi-agent pat...'
#   [ACADEMIC] 5 resultados para 'LangGraph multi-agent pat...'
#
# Log: ['parallel: 3 fuentes completadas', 'resumen generado']

Ejercicio 3: Registry con dispatch y manejo de errores (Medio)

Crea un registry de 3 subagents (translator, summarizer, formatter). Implementa una función safe_dispatch que invoque el subagent por nombre con try/except. El orquestador debe llamar a los 3 en secuencia, manejando errores individuales sin abortar todo el flujo.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class SubState(TypedDict):
    input_text: str
    result: str

def translate(state: SubState) -> dict:
    return {"result": f"[EN] Translated: {state['input_text'][:30]}..."}

def summarize(state: SubState) -> dict:
    return {"result": f"[SUMMARY] Key points from: {state['input_text'][:30]}..."}

def format_output(state: SubState) -> dict:
    raise ValueError("Formatter temporalmente no disponible")

def build_sub(fn):
    b = StateGraph(SubState)
    b.add_node("work", fn)
    b.add_edge(START, "work")
    b.add_edge("work", END)
    return b.compile()

REGISTRY = {
    "translator": build_sub(translate),
    "summarizer": build_sub(summarize),
    "formatter": build_sub(format_output),
}

def safe_dispatch(name: str, text: str) -> dict:
    try:
        result = REGISTRY[name].invoke({"input_text": text, "result": ""})
        return {"success": True, "agent": name, "result": result["result"]}
    except Exception as e:
        return {"success": False, "agent": name, "result": f"ERROR: {str(e)[:40]}"}

class OrcState(TypedDict):
    text: str
    results: list[dict]
    log: Annotated[list[str], operator.add]

def orchestrate(state: OrcState) -> dict:
    pipeline = ["translator", "summarizer", "formatter"]
    results = []
    current = state["text"]

    for agent_name in pipeline:
        outcome = safe_dispatch(agent_name, current)
        results.append(outcome)
        if outcome["success"]:
            current = outcome["result"]

    successes = [r["agent"] for r in results if r["success"]]
    failures = [r["agent"] for r in results if not r["success"]]

    return {
        "results": results,
        "log": [
            f"completados: {successes}",
            f"fallidos: {failures}",
        ],
    }

builder = StateGraph(OrcState)
builder.add_node("orchestrate", orchestrate)
builder.add_edge(START, "orchestrate")
builder.add_edge("orchestrate", END)

graph = builder.compile()

result = graph.invoke({"text": "Los agentes de IA están transformando la industria", "results": [], "log": []})

for r in result["results"]:
    status = "✅" if r["success"] else "❌"
    print(f"  {status} {r['agent']}: {r['result'][:50]}...")
print(f"\nLog: {result['log']}")
# Output esperado:
#   ✅ translator: [EN] Translated: Los agentes de IA están trans...
#   ✅ summarizer: [SUMMARY] Key points from: [EN] Translated: Los agent...
#   ❌ formatter: ERROR: Formatter temporalmente no disponib...
#
# Log: ['completados: ['translator', 'summarizer']', "fallidos: ['formatter']"]

Ejercicio 4: Subagents con agregación de resultados (Avanzado)

Crea un sistema de "voting" donde 3 subagents analizan el mismo texto de forma independiente y retornan una clasificación (positivo, negativo, neutro). El padre agrega los votos y decide la clasificación final por mayoría. Implementa los subagents en paralelo.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

import concurrent.futures
from collections import Counter
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class SubState(TypedDict):
    text: str
    analyst_id: int
    classification: str

class ParentState(TypedDict):
    text: str
    votes: list[str]
    final_classification: str
    confidence: float
    log: Annotated[list[str], operator.add]

ANALYST_BIASES = {
    1: ["innovación", "crecimiento", "oportunidad"],
    2: ["riesgo", "costo", "problema"],
    3: [],
}

def classify(state: SubState) -> dict:
    text_lower = state["text"].lower()
    analyst_id = state["analyst_id"]
    positive_kw = ["bueno", "éxito", "crecimiento", "innovación", "mejora"]
    negative_kw = ["malo", "fracaso", "riesgo", "problema", "costo"]

    bias = ANALYST_BIASES.get(analyst_id, [])
    pos_score = sum(1 for w in positive_kw + bias if w in text_lower)
    neg_score = sum(1 for w in negative_kw + bias if w in text_lower)

    if pos_score > neg_score:
        return {"classification": "positivo"}
    elif neg_score > pos_score:
        return {"classification": "negativo"}
    return {"classification": "neutro"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("classify", classify)
sub_builder.add_edge(START, "classify")
sub_builder.add_edge("classify", END)
classifier = sub_builder.compile()

def parallel_classify(state: ParentState) -> dict:
    configs = [
        {"text": state["text"], "analyst_id": i, "classification": ""}
        for i in range(1, 4)
    ]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        futures = [pool.submit(classifier.invoke, c) for c in configs]
        votes = [f.result()["classification"] for f in futures]

    counts = Counter(votes)
    winner = counts.most_common(1)[0]
    confidence = winner[1] / len(votes)

    return {
        "votes": votes,
        "final_classification": winner[0],
        "confidence": confidence,
        "log": [
            f"votos: {votes}",
            f"resultado: {winner[0]} ({confidence:.0%} confianza)",
        ],
    }

builder = StateGraph(ParentState)
builder.add_node("classify", parallel_classify)
builder.add_edge(START, "classify")
builder.add_edge("classify", END)

graph = builder.compile()

texts = [
    "Gran innovación en AI, crecimiento del 40% y mejora continua",
    "Alto riesgo y costo, el problema persiste sin solución clara",
    "El mercado se mantiene estable sin cambios significativos",
]

for text in texts:
    result = graph.invoke({
        "text": text, "votes": [],
        "final_classification": "", "confidence": 0.0, "log": [],
    })
    print(f"Texto: {text[:50]}...")
    print(f"  Votos: {result['votes']}")
    print(f"  Clasificación: {result['final_classification']} ({result['confidence']:.0%})\n")
# Output esperado:
# Texto: Gran innovación en AI, crecimiento del 40% y mejor...
#   Votos: ['positivo', 'positivo', 'positivo']
#   Clasificación: positivo (100%)
#
# Texto: Alto riesgo y costo, el problema persiste sin soluc...
#   Votos: ['negativo', 'negativo', 'negativo']
#   Clasificación: negativo (100%)
#
# Texto: El mercado se mantiene estable sin cambios signific...
#   Votos: ['neutro', 'neutro', 'neutro']
#   Clasificación: neutro (100%)

Resumen

En esta cápsula aprendiste:

  • Los subagents trabajan en contexto aislado — reciben solo la descripción de su tarea, no la conversación completa del padre. Esto previene el context bloat que ocurre con handoffs y estado compartido
  • El patrón básico es envolver un agente como @tool — create_agent crea el subagent, @tool lo expone al padre. El padre llama la tool, el subagent ejecuta en su contexto, y retorna solo el resultado
  • Subgrafos compilados dan más control — para subagents con flujos internos complejos, crea un StateGraph independiente con su propio SubState e invócalo desde un nodo del padre
  • La ejecución paralela es natural — como los subagents tienen contexto aislado, puedes ejecutar varios simultáneamente con ThreadPoolExecutor sin conflictos de estado
  • El single dispatch pattern escala mejor — un solo dispatch(agent_name, task) con un registry reemplaza N tools individuales cuando tienes muchos subagents
  • El manejo de errores es responsabilidad del padre — usa safe_invoke con try/except y timeouts para que un subagent fallido no tumbe todo el sistema
  • Subagents vs handoffs: usa subagents cuando necesitas aislamiento de contexto y ejecución paralela; usa handoffs cuando necesitas contexto compartido en flujo secuencial

Próxima cápsula: Diseño de estado en multi-agent — cómo diseñar el estado compartido entre agentes, manejar conflictos, y decidir qué información comparten vs qué mantienen aislado.

Recursos adicionales

  1. LangChain — Subagents — Documentación oficial del patrón subagents
  2. LangGraph — Use Subgraphs — Cómo usar subgrafos compilados como nodos
  3. LangGraph — Functional API — @task y @entrypoint para subagents con aislamiento automático
  4. Context Engineering for Agents — Diseño del flujo de contexto entre agentes
  5. LangGraph — Multi-Agent Systems — Visión general de patrones multi-agente

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

← AnteriorPattern HandoffsSiguiente →Pattern Router