Pattern Handoffs

Descripción de la cápsula

En la cápsula anterior aprendiste el patrón Supervisor: un agente central que coordina a los demás, decide quién trabaja, y recopila resultados. Funciona bien cuando no sabes de antemano el orden de ejecución. Pero muchos flujos son predecibles: investigar → analizar → escribir siempre sigue el mismo orden. En estos casos, ¿para qué pasar por un supervisor si el researcher siempre le pasa al analyst, y el analyst siempre le pasa al writer?

Un handoff es cuando un agente transfiere el control directamente a otro agente. No hay supervisor intermediario. Agent A termina su parte, decide que Agent B es el siguiente, y le pasa el control con el contexto necesario. Agent B ejecuta, y puede pasar a Agent C o terminar. Es una cadena de responsabilidad donde cada agente sabe a quién delegarle.

La diferencia con el supervisor es arquitectónica: en el supervisor, toda decisión pasa por un nodo central. En handoffs, los agentes se coordinan entre sí. Esto reduce la latencia (menos roundtrips al supervisor), simplifica el flujo cuando es predecible, y hace que cada agente sea más autónomo.

Cuándo usar handoffs vs supervisor

La decisión no es "uno es mejor que el otro" — son herramientas para problemas diferentes:

Regla práctica:

• ✅ Usa supervisor cuando no sabes de antemano qué agente necesitas ni en qué orden
• ✅ Usa handoffs cuando el flujo es secuencial y predecible (research → analyze → write)
• ✅ Usa híbrido cuando tienes fases predecibles con decisiones dinámicas dentro de cada fase

El mecanismo de handoff en LangGraph

LangGraph implementa handoffs de dos formas:

1. A nivel de nodo: El nodo retorna un Command con goto que indica el siguiente nodo. El grafo sigue esa instrucción directamente.

2. A nivel de tool: Un agente (dentro de un nodo) llama una herramienta de handoff que retorna un Command. El Command navega al siguiente agente en el grafo padre usando graph=Command.PARENT.

Ambos usan el mismo objeto Command — la diferencia es dónde se origina la decisión. A nivel de nodo, la función del nodo decide. A nivel de tool, el LLM decide (al llamar la herramienta).

Handoff básico: Agente A → Agente B

El handoff más simple: un researcher investiga y transfiere directamente al analyst.

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 State(TypedDict):
    task: str
    research: str
    analysis: str
    log: Annotated[list[str], operator.add]

def researcher(state: State) -> Command:
    result = f"5 fuentes encontradas sobre '{state['task']}': papers, blogs, docs oficiales"
    return Command(
        goto="analyst",
        update={
            "research": result,
            "log": ["researcher_done → handoff to analyst"],
        },
    )

def analyst(state: State) -> dict:
    analysis = f"Análisis basado en: {state['research'][:50]}... → 3 tendencias identificadas"
    return {
        "analysis": analysis,
        "log": ["analyst_done"],
    }

builder = StateGraph(State)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)

builder.add_edge(START, "researcher")
builder.add_edge("analyst", END)

graph = builder.compile()

result = graph.invoke({
    "task": "Estado del mercado AI 2025",
    "research": "",
    "analysis": "",
    "log": [],
})

print(f"Research: {result['research']}")
print(f"Analysis: {result['analysis']}")
print(f"Log: {result['log']}")
# Output esperado:
# Research: 5 fuentes encontradas sobre 'Estado del mercado AI 2025': papers, blogs, docs oficiales
# Analysis: Análisis basado en: 5 fuentes encontradas sobre 'Estado del mercado AI 2... → 3 tendencias identificadas
# Log: ['researcher_done → handoff to analyst', 'analyst_done']

El researcher retorna Command(goto="analyst", update={...}). Esto hace dos cosas:

1. Actualiza el estado con los resultados de la investigación
2. Transfiere el control directamente al analyst — sin pasar por un supervisor

El analyst recibe el estado ya actualizado y trabaja con él. No hay nodo intermedio.

El flujo completo de un handoff

Un handoff real no es solo "el researcher pasa al analyst". Es una secuencia completa donde cada agente prepara el contexto para el siguiente:

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 PipelineState(TypedDict):
    topic: str
    sources: list[str]
    research_summary: str
    analysis: str
    report: str
    log: Annotated[list[str], operator.add]

def researcher(state: PipelineState) -> Command:
    sources = [
        f"Paper: '{state['topic']}' trends 2025",
        f"Blog: Industry analysis of {state['topic']}",
        f"Docs: Official {state['topic']} documentation",
    ]
    summary = f"Resumen de {len(sources)} fuentes sobre '{state['topic']}'"
    return Command(
        goto="analyst",
        update={
            "sources": sources,
            "research_summary": summary,
            "log": [f"researcher: {len(sources)} fuentes → handoff to analyst"],
        },
    )

def analyst(state: PipelineState) -> Command:
    analysis = (
        f"Análisis de {len(state['sources'])} fuentes: "
        f"tendencia alcista en '{state['topic']}', "
        f"2 riesgos identificados, 1 oportunidad clara"
    )
    return Command(
        goto="writer",
        update={
            "analysis": analysis,
            "log": ["analyst: análisis completo → handoff to writer"],
        },
    )

def writer(state: PipelineState) -> dict:
    report = (
        f"REPORTE EJECUTIVO\n"
        f"Tema: {state['topic']}\n"
        f"Fuentes: {len(state['sources'])}\n"
        f"Hallazgo principal: {state['analysis'][:60]}...\n"
        f"Recomendación: Invertir en esta área"
    )
    return {
        "report": report,
        "log": ["writer: reporte generado"],
    }

builder = StateGraph(PipelineState)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_node("writer", writer)

builder.add_edge(START, "researcher")
builder.add_edge("writer", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "LLMs en producción",
    "sources": [],
    "research_summary": "",
    "analysis": "",
    "report": "",
    "log": [],
})

print("=== Resultado del pipeline ===\n")
print(result["report"])
print(f"\n=== Flujo de handoffs ===")
for entry in result["log"]:
    print(f"  → {entry}")
# Output esperado:
# === Resultado del pipeline ===
#
# REPORTE EJECUTIVO
# Tema: LLMs en producción
# Fuentes: 3
# Hallazgo principal: Análisis de 3 fuentes: tendencia alcista en 'LLMs en producci...
# Recomendación: Invertir en esta área
#
# === Flujo de handoffs ===
#   → researcher: 3 fuentes → handoff to analyst
#   → analyst: análisis completo → handoff to writer
#   → writer: reporte generado

El flujo completo: researcher investiga → prepara contexto (fuentes + resumen) → transfiere al analyst → el analyst analiza → prepara contexto (análisis + hallazgos) → transfiere al writer → el writer genera el reporte final. Cada agente recibe exactamente lo que necesita del anterior.

Handoff tools: transferencia vía herramientas

En el ejemplo anterior, las funciones de nodo deciden directamente a quién transferir. Pero en un sistema real, los agentes son LLMs que toman decisiones dinámicas. El patrón de LangGraph para esto: handoff tools — herramientas que el agente llama para transferir el control.

Cuando un agente (creado con create_agent) decide que otro agente debe tomar el control, llama una herramienta de handoff. Esta herramienta retorna un Command que navega al agente destino en el grafo padre.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from langchain.tools import tool, ToolRuntime
from langchain.messages import AIMessage, ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class TeamState(AgentState):
    active_agent: str

@tool
def transfer_to_analyst(runtime: ToolRuntime) -> Command:
    """Transfiere el control al agente analista para análisis de datos."""
    last_ai = next(
        m for m in reversed(runtime.state["messages"])
        if isinstance(m, AIMessage)
    )
    return Command(
        goto="analyst_node",
        update={
            "active_agent": "analyst",
            "messages": [
                last_ai,
                ToolMessage(
                    content="Transferido al analista",
                    tool_call_id=runtime.tool_call_id,
                ),
            ],
        },
        graph=Command.PARENT,
    )

@tool
def transfer_to_writer(runtime: ToolRuntime) -> Command:
    """Transfiere el control al agente escritor para generar reportes."""
    last_ai = next(
        m for m in reversed(runtime.state["messages"])
        if isinstance(m, AIMessage)
    )
    return Command(
        goto="writer_node",
        update={
            "active_agent": "writer",
            "messages": [
                last_ai,
                ToolMessage(
                    content="Transferido al escritor",
                    tool_call_id=runtime.tool_call_id,
                ),
            ],
        },
        graph=Command.PARENT,
    )

researcher_agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[transfer_to_analyst],
    prompt=(
        "Eres un investigador. Cuando el usuario pide investigar algo, "
        "resume las fuentes principales y luego transfiere al analista."
    ),
)

analyst_agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[transfer_to_writer],
    prompt=(
        "Eres un analista. Analiza la información del investigador, "
        "identifica tendencias y riesgos, y transfiere al escritor."
    ),
)

writer_agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    prompt="Eres un escritor. Genera un reporte ejecutivo con la información disponible.",
)

def researcher_node(state: TeamState):
    return researcher_agent.invoke(state)

def analyst_node(state: TeamState):
    return analyst_agent.invoke(state)

def writer_node(state: TeamState):
    return writer_agent.invoke(state)

def route_after_agent(state: TeamState) -> Literal["researcher_node", "analyst_node", "writer_node", "__end__"]:
    messages = state.get("messages", [])
    if messages:
        last = messages[-1]
        if isinstance(last, AIMessage) and not last.tool_calls:
            return "__end__"
    active = state.get("active_agent", "researcher")
    return f"{active}_node"

builder = StateGraph(TeamState)
builder.add_node("researcher_node", researcher_node)
builder.add_node("analyst_node", analyst_node)
builder.add_node("writer_node", writer_node)

builder.add_conditional_edges(START, lambda _: "researcher_node")
builder.add_conditional_edges("researcher_node", route_after_agent, ["analyst_node", "writer_node", END])
builder.add_conditional_edges("analyst_node", route_after_agent, ["researcher_node", "writer_node", END])
builder.add_conditional_edges("writer_node", route_after_agent, ["researcher_node", "analyst_node", END])

graph = builder.compile()

result = graph.invoke({
    "messages": [{"role": "user", "content": "Investiga el estado de AI agents en 2025 y genera un reporte ejecutivo"}],
    "active_agent": "researcher",
})

for msg in result["messages"]:
    msg.pretty_print()
# Output esperado (varía según el modelo):
# ================================ Human Message =================================
# Investiga el estado de AI agents en 2025 y genera un reporte ejecutivo
# ================================== Ai Message ==================================
# He investigado las principales fuentes sobre AI agents...
# ================================== Tool Message ================================
# Transferido al analista
# ================================== Ai Message ==================================
# Basándome en la investigación, identifico 3 tendencias clave...
# ================================== Tool Message ================================
# Transferido al escritor
# ================================== Ai Message ==================================
# REPORTE EJECUTIVO: AI Agents en 2025
# ...

El mecanismo:

1. Researcher recibe la tarea del usuario
2. Investiga y decide transferir → llama transfer_to_analyst
3. La tool retorna Command(goto="analyst_node", graph=Command.PARENT) → el grafo padre navega al analyst
4. Analyst recibe el contexto, analiza, y llama transfer_to_writer
5. Writer recibe todo y genera el reporte final sin handoff tools (termina el flujo)

El ToolMessage es crítico: cuando un LLM llama una tool, espera una respuesta. Sin el ToolMessage con el tool_call_id correcto, el historial de conversación queda malformado.

Transferencia de estado: completa vs filtrada

Cuando Agent A le pasa el control a Agent B, ¿qué ve Agent B? Hay dos estrategias:

Handoff completo: Agent B ve todo el historial — cada mensaje, cada resultado de tool, cada decisión intermedia. Simple de implementar, pero el contexto crece con cada handoff.

Handoff filtrado: Agent B solo ve lo relevante — un resumen de lo anterior, o solo los datos que necesita. Más trabajo de implementación, pero el contexto se mantiene enfocado.

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 State(TypedDict):
    task: str
    raw_data: str
    cleaned_summary: str
    internal_notes: str
    analysis: str
    log: Annotated[list[str], operator.add]

def data_collector(state: State) -> Command:
    raw = (
        "Datos crudos: 500 registros de ventas, "
        "200 con errores de formato, "
        "50 duplicados, "
        "campos: fecha, producto, monto, región"
    )
    notes = "NOTA INTERNA: API de datos respondió en 2.3s, rate limit cerca del 80%"
    summary = "Datos limpios: 250 registros válidos de ventas Q4, 4 regiones, 12 productos"
    return Command(
        goto="analyst",
        update={
            "raw_data": raw,
            "cleaned_summary": summary,
            "internal_notes": notes,
            "log": ["collector: datos recopilados → handoff to analyst"],
        },
    )

def analyst_full_context(state: State) -> dict:
    """Analyst que ve TODO el estado (handoff completo)."""
    sees_raw = len(state["raw_data"]) > 0
    sees_notes = len(state["internal_notes"]) > 0
    sees_summary = len(state["cleaned_summary"]) > 0
    analysis = (
        f"Analyst (contexto completo) ve: "
        f"raw_data={'sí' if sees_raw else 'no'}, "
        f"internal_notes={'sí' if sees_notes else 'no'}, "
        f"cleaned_summary={'sí' if sees_summary else 'no'}"
    )
    return {"analysis": analysis, "log": ["analyst_full: análisis con todo el contexto"]}

def analyst_filtered_context(state: State) -> dict:
    """Analyst que solo ve el resumen limpio (handoff filtrado)."""
    analysis = f"Analyst (contexto filtrado) trabaja con: '{state['cleaned_summary']}'"
    return {"analysis": analysis, "log": ["analyst_filtered: análisis con contexto enfocado"]}


print("=== HANDOFF COMPLETO: analyst ve todo ===\n")

builder_full = StateGraph(State)
builder_full.add_node("collector", data_collector)
builder_full.add_node("analyst", analyst_full_context)
builder_full.add_edge(START, "collector")
builder_full.add_edge("analyst", END)

graph_full = builder_full.compile()
result_full = graph_full.invoke({
    "task": "Análisis Q4", "raw_data": "", "cleaned_summary": "",
    "internal_notes": "", "analysis": "", "log": [],
})
print(f"Analysis: {result_full['analysis']}")


print("\n=== HANDOFF FILTRADO: analyst solo ve el resumen ===\n")

def data_collector_filtered(state: State) -> Command:
    raw = "Datos crudos: 500 registros..."
    notes = "NOTA INTERNA: rate limit 80%"
    summary = "250 registros válidos de ventas Q4"
    return Command(
        goto="analyst",
        update={
            "raw_data": "",
            "cleaned_summary": summary,
            "internal_notes": "",
            "log": ["collector: datos filtrados → handoff to analyst"],
        },
    )

builder_filtered = StateGraph(State)
builder_filtered.add_node("collector", data_collector_filtered)
builder_filtered.add_node("analyst", analyst_filtered_context)
builder_filtered.add_edge(START, "collector")
builder_filtered.add_edge("analyst", END)

graph_filtered = builder_filtered.compile()
result_filtered = graph_filtered.invoke({
    "task": "Análisis Q4", "raw_data": "", "cleaned_summary": "",
    "internal_notes": "", "analysis": "", "log": [],
})
print(f"Analysis: {result_filtered['analysis']}")
# Output esperado:
# === HANDOFF COMPLETO: analyst ve todo ===
#
# Analysis: Analyst (contexto completo) ve: raw_data=sí, internal_notes=sí, cleaned_summary=sí
#
# === HANDOFF FILTRADO: analyst solo ve el resumen ===
#
# Analysis: Analyst (contexto filtrado) trabaja con: '250 registros válidos de ventas Q4'

En el handoff filtrado, el collector no pasa raw_data ni internal_notes al analyst. Solo pasa cleaned_summary. Esto mantiene el contexto del analyst enfocado en lo que realmente necesita.

Regla: Filtra el contexto cuando el agente destino no necesita los detalles intermedios. Pasa todo cuando el agente destino necesita el historial completo para tomar decisiones.

Cadenas de handoff: A → B → C → respuesta

Los handoffs se encadenan naturalmente. Cada agente procesa, transforma, y pasa al siguiente:

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 ChainState(TypedDict):
    query: str
    search_results: str
    fact_check: str
    draft: str
    edited: str
    log: Annotated[list[str], operator.add]

def searcher(state: ChainState) -> Command:
    results = f"3 resultados para '{state['query']}': [artículo científico, blog técnico, documentación]"
    return Command(
        goto="fact_checker",
        update={"search_results": results, "log": ["searcher → fact_checker"]},
    )

def fact_checker(state: ChainState) -> Command:
    check = f"Verificado: 2 de 3 fuentes confiables. Fuente descartada: blog sin referencias"
    return Command(
        goto="drafter",
        update={"fact_check": check, "log": ["fact_checker → drafter"]},
    )

def drafter(state: ChainState) -> Command:
    draft = (
        f"Borrador sobre '{state['query']}':\n"
        f"Basado en {state['fact_check'][:40]}...\n"
        f"Contenido: análisis detallado con 3 secciones"
    )
    return Command(
        goto="editor",
        update={"draft": draft, "log": ["drafter → editor"]},
    )

def editor(state: ChainState) -> dict:
    edited = f"VERSIÓN FINAL (editada): {state['draft'][:60]}... [Corregido estilo y gramática]"
    return {"edited": edited, "log": ["editor: publicación lista"]}

builder = StateGraph(ChainState)
builder.add_node("searcher", searcher)
builder.add_node("fact_checker", fact_checker)
builder.add_node("drafter", drafter)
builder.add_node("editor", editor)

builder.add_edge(START, "searcher")
builder.add_edge("editor", END)

graph = builder.compile()

result = graph.invoke({
    "query": "Mejores prácticas para RAG en producción",
    "search_results": "", "fact_check": "",
    "draft": "", "edited": "", "log": [],
})

print(f"Resultado final: {result['edited']}")
print(f"\nCadena de handoffs:")
for step in result["log"]:
    print(f"  {step}")
# Output esperado:
# Resultado final: VERSIÓN FINAL (editada): Borrador sobre 'Mejores prácticas para RAG en producción':
# Ba... [Corregido estilo y gramática]
#
# Cadena de handoffs:
#   searcher → fact_checker
#   fact_checker → drafter
#   drafter → editor
#   editor: publicación lista

Cuatro agentes, cero supervisores. Cada agente sabe exactamente a quién pasarle el trabajo. El flujo es predecible y cada handoff transfiere exactamente el contexto que el siguiente necesita.

Handoff con retorno: A → B → de vuelta a A

A veces un agente necesita delegar una subtarea y recuperar el control. El researcher necesita que un specialist verifique un dato, y después continúa con la investigación:

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 RoundTripState(TypedDict):
    task: str
    research: str
    verification: str
    final_report: str
    phase: str
    log: Annotated[list[str], operator.add]

def researcher(state: RoundTripState) -> Command:
    if state["phase"] == "initial":
        research = f"Investigación inicial sobre '{state['task']}': dato clave encontrado pero necesita verificación"
        return Command(
            goto="specialist",
            update={
                "research": research,
                "phase": "verifying",
                "log": ["researcher: investigación inicial → handoff to specialist"],
            },
        )
    else:
        report = (
            f"REPORTE FINAL\n"
            f"Investigación: {state['research'][:50]}...\n"
            f"Verificación: {state['verification'][:50]}...\n"
            f"Conclusión: datos confirmados, recomendación positiva"
        )
        return Command(
            goto="__end__",
            update={
                "final_report": report,
                "phase": "complete",
                "log": ["researcher: reporte final generado"],
            },
        )

def specialist(state: RoundTripState) -> Command:
    verification = f"Dato verificado contra 3 fuentes independientes: CONFIRMADO con 95% de confianza"
    return Command(
        goto="researcher",
        update={
            "verification": verification,
            "phase": "verified",
            "log": ["specialist: verificación completa → handoff back to researcher"],
        },
    )

builder = StateGraph(RoundTripState)
builder.add_node("researcher", researcher)
builder.add_node("specialist", specialist)

builder.add_edge(START, "researcher")

graph = builder.compile()

result = graph.invoke({
    "task": "Impacto de AI en productividad empresarial",
    "research": "", "verification": "", "final_report": "",
    "phase": "initial", "log": [],
})

print(result["final_report"])
print(f"\nFlujo round-trip:")
for step in result["log"]:
    print(f"  {step}")
# Output esperado:
# REPORTE FINAL
# Investigación: Investigación inicial sobre 'Impacto de AI en pro...
# Verificación: Dato verificado contra 3 fuentes independientes: C...
# Conclusión: datos confirmados, recomendación positiva
#
# Flujo round-trip:
#   researcher: investigación inicial → handoff to specialist
#   specialist: verificación completa → handoff back to researcher
#   researcher: reporte final generado

El researcher usa phase para saber en qué punto del flujo está. En la primera invocación, investiga y delega al specialist. Cuando el specialist devuelve el control (handoff de retorno), el researcher está en fase "verified" y genera el reporte final.

Este patrón es útil cuando un agente necesita validación externa antes de continuar. El researcher no pierde su contexto — todo persiste en el estado compartido.

Cuándo los handoffs se complican

Los handoffs son elegantes cuando la cadena es corta (2-4 agentes) y el flujo es predecible. Se complican cuando:

• ⚠️ Demasiados handoffs: A → B → C → D → E → F. Con 6+ agentes en cadena, debuggear un error es rastrear toda la cadena. ¿Quién corrompió el estado? ¿Quién recibió contexto incorrecto?
• ⚠️ Handoffs circulares sin exit condition: A → B → A → B → A... Si no hay una condición clara de terminación, el flujo nunca acaba (o llega a recursion_limit)
• ⚠️ Estado que crece sin control: Cada agente agrega datos al estado. Si no filtras, el último agente de la cadena recibe todo lo que acumularon los anteriores
• ⚠️ Handoffs condicionales complejos: Si Agent A puede transferir a B, C, o D dependiendo de 5 condiciones, estás reinventando un supervisor con más complejidad

Regla: Si tu cadena de handoffs necesita un "mapa" para entenderla, probablemente necesitas un supervisor.

Tabla comparativa: supervisor vs handoffs vs híbrido

El patrón híbrido es el más común en producción: un supervisor de alto nivel que coordina fases (research phase → analysis phase → writing phase), y dentro de cada fase, los agentes usan handoffs para la secuencia interna.

Troubleshooting

Problema 1: "El handoff no transfiere al agente correcto"

Síntoma: Command(goto="analyst") no navega al analyst, o el grafo termina prematuramente.

Causa: El nombre en goto no coincide con el nombre registrado en add_node().

Solución: Verifica que el nombre sea idéntico:

builder.add_node("analyst", analyst_fn)

Command(goto="analyst")

Problema 2: "ToolMessage faltante en handoff con tools"

Síntoma: El agente destino recibe un historial de mensajes malformado y produce errores o respuestas incoherentes.

Causa: No incluiste el ToolMessage al hacer el handoff. Cuando un LLM llama una tool, espera una respuesta con el mismo tool_call_id.

Solución: Siempre incluye el par AIMessage + ToolMessage en el update:

@tool
def transfer_to_analyst(runtime: ToolRuntime) -> Command:
    last_ai = next(
        m for m in reversed(runtime.state["messages"])
        if isinstance(m, AIMessage)
    )
    return Command(
        goto="analyst_node",
        update={
            "messages": [
                last_ai,
                ToolMessage(content="Transferido", tool_call_id=runtime.tool_call_id),
            ],
        },
        graph=Command.PARENT,
    )

Problema 3: "Handoff circular infinito (A → B → A → B...)"

Síntoma: El grafo ejecuta indefinidamente hasta llegar a recursion_limit.

Causa: No hay condición de terminación en el ciclo de handoffs.

Solución: Usa un campo de estado (phase, iteration_count) para controlar cuándo el ciclo debe terminar:

def agent_a(state: State) -> Command:
    if state["iteration"] >= 3:
        return Command(goto="__end__", update={"log": ["terminando"]})
    return Command(
        goto="agent_b",
        update={"iteration": state["iteration"] + 1},
    )

Problema 4: "El contexto del agente destino tiene datos innecesarios"

Síntoma: El analyst recibe datos internos del researcher (notas, intentos fallidos, datos crudos) que no necesita.

Causa: Estás usando handoff completo cuando deberías filtrar.

Solución: En el Command.update, solo incluye los campos relevantes para el agente destino. Limpia campos internos:

return Command(
    goto="analyst",
    update={
        "cleaned_summary": summary,
        "internal_notes": "",
        "raw_data": "",
    },
)

Ejercicios

Ejercicio 1: Handoff básico entre dos agentes (Fácil)

Crea un sistema con dos agentes: translator y reviewer. El translator traduce un texto de español a inglés, y hace handoff al reviewer que verifica la traducción. Usa Command para el handoff.

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 State(TypedDict):
    original_text: str
    translation: str
    review: str
    log: Annotated[list[str], operator.add]

def translator(state: State) -> Command:
    translated = f"Translation of '{state['original_text'][:30]}...': [English version here]"
    return Command(
        goto="reviewer",
        update={"translation": translated, "log": ["translator → reviewer"]},
    )

def reviewer(state: State) -> dict:
    review = f"Review: '{state['translation'][:40]}...' — Accuracy: 95%, fluency: good"
    return {"review": review, "log": ["reviewer: review complete"]}

builder = StateGraph(State)
builder.add_node("translator", translator)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "translator")
builder.add_edge("reviewer", END)

graph = builder.compile()
result = graph.invoke({
    "original_text": "Los agentes de IA están transformando la industria",
    "translation": "", "review": "", "log": [],
})

print(f"Translation: {result['translation']}")
print(f"Review: {result['review']}")
print(f"Log: {result['log']}")
# Output esperado:
# Translation: Translation of 'Los agentes de IA están trans...': [English version here]
# Review: Review: 'Translation of 'Los agentes de IA están t...' — Accuracy: 95%, fluency: good
# Log: ['translator → reviewer', 'reviewer: review complete']

Ejercicio 2: Cadena de 3 agentes con filtrado de contexto (Medio)

Crea una cadena: extractor → classifier → formatter. El extractor obtiene datos crudos (nombre, email, teléfono, notas internas). El classifier solo debe recibir nombre, email, teléfono (sin notas internas). El formatter recibe la clasificación y genera un output limpio. Implementa filtrado de contexto en los handoffs.

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 State(TypedDict):
    input_text: str
    name: str
    email: str
    phone: str
    internal_notes: str
    category: str
    formatted_output: str
    log: Annotated[list[str], operator.add]

def extractor(state: State) -> Command:
    return Command(
        goto="classifier",
        update={
            "name": "María García",
            "email": "maria@empresa.com",
            "phone": "+52 555 1234567",
            "internal_notes": "Lead caliente, contactar antes del viernes. Score interno: 85/100",
            "log": ["extractor: datos extraídos → handoff filtrado to classifier"],
        },
    )

def classifier(state: State) -> Command:
    has_notes = len(state.get("internal_notes", "")) > 0
    category = "enterprise" if "@empresa" in state["email"] else "individual"
    return Command(
        goto="formatter",
        update={
            "category": category,
            "internal_notes": "",
            "log": [f"classifier: categoría={category}, ve notas internas={has_notes} → handoff to formatter"],
        },
    )

def formatter(state: State) -> dict:
    output = (
        f"CONTACTO [{state['category'].upper()}]\n"
        f"  Nombre: {state['name']}\n"
        f"  Email: {state['email']}\n"
        f"  Teléfono: {state['phone']}"
    )
    return {"formatted_output": output, "log": ["formatter: output generado"]}

builder = StateGraph(State)
builder.add_node("extractor", extractor)
builder.add_node("classifier", classifier)
builder.add_node("formatter", formatter)

builder.add_edge(START, "extractor")
builder.add_edge("formatter", END)

graph = builder.compile()
result = graph.invoke({
    "input_text": "Datos del contacto...", "name": "", "email": "",
    "phone": "", "internal_notes": "", "category": "",
    "formatted_output": "", "log": [],
})

print(result["formatted_output"])
print(f"\nNotas internas al final: '{result['internal_notes']}'")
print(f"\nLog:")
for step in result["log"]:
    print(f"  {step}")
# Output esperado:
# CONTACTO [ENTERPRISE]
#   Nombre: María García
#   Email: maria@empresa.com
#   Teléfono: +52 555 1234567
#
# Notas internas al final: ''
#
# Log:
#   extractor: datos extraídos → handoff filtrado to classifier
#   classifier: categoría=enterprise, ve notas internas=True → handoff to formatter
#   formatter: output generado

Ejercicio 3: Handoff con retorno y verificación (Medio)

Crea un sistema donde un planner genera un plan, hace handoff a un validator, el validator verifica el plan y devuelve control al planner. Si la validación falla, el planner regenera. Si pasa, el planner genera el resultado final. Usa phase para controlar el flujo.

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 State(TypedDict):
    goal: str
    plan: str
    validation: str
    is_valid: bool
    result: str
    phase: str
    attempt: int
    log: Annotated[list[str], operator.add]

def planner(state: State) -> Command:
    attempt = state.get("attempt", 0)

    if state["phase"] == "initial" or state["phase"] == "retry":
        attempt += 1
        if attempt == 1:
            plan = f"Plan v1 para '{state['goal']}': enfoque básico"
        else:
            plan = f"Plan v{attempt} para '{state['goal']}': enfoque mejorado (fix: {state['validation'][:30]}...)"
        return Command(
            goto="validator",
            update={
                "plan": plan,
                "attempt": attempt,
                "phase": "validating",
                "log": [f"planner: plan v{attempt} → handoff to validator"],
            },
        )
    else:
        result = f"EJECUTADO: {state['plan']} — Validación: {state['validation']}"
        return Command(
            goto="__end__",
            update={
                "result": result,
                "phase": "complete",
                "log": ["planner: plan validado, resultado generado"],
            },
        )

def validator(state: State) -> Command:
    is_valid = state["attempt"] >= 2
    if is_valid:
        validation = "APROBADO: plan cumple todos los criterios"
        next_phase = "approved"
    else:
        validation = "RECHAZADO: falta manejo de errores y logging"
        next_phase = "retry"
    return Command(
        goto="planner",
        update={
            "validation": validation,
            "is_valid": is_valid,
            "phase": next_phase,
            "log": [f"validator: {validation[:30]}... → handoff back to planner"],
        },
    )

builder = StateGraph(State)
builder.add_node("planner", planner)
builder.add_node("validator", validator)
builder.add_edge(START, "planner")

graph = builder.compile()
result = graph.invoke({
    "goal": "Migrar base de datos a PostgreSQL",
    "plan": "", "validation": "", "is_valid": False,
    "result": "", "phase": "initial", "attempt": 0, "log": [],
})

print(f"Intentos: {result['attempt']}")
print(f"Resultado: {result['result'][:80]}...")
print(f"\nFlujo:")
for step in result["log"]:
    print(f"  {step}")
# Output esperado:
# Intentos: 2
# Resultado: EJECUTADO: Plan v2 para 'Migrar base de datos a PostgreSQL': enfoque mejorado (f...
#
# Flujo:
#   planner: plan v1 → handoff to validator
#   validator: RECHAZADO: falta manejo de err... → handoff back to planner
#   planner: plan v2 → handoff to validator
#   validator: APROBADO: plan cumple todos lo... → handoff back to planner
#   planner: plan validado, resultado generado

Ejercicio 4: Handoff dinámico basado en tipo de tarea (Medio)

Crea un router_agent que recibe una tarea y decide a quién hacer handoff: code_agent (para tareas de código), writing_agent (para tareas de escritura), o data_agent (para tareas de datos). El handoff se decide con palabras clave en la tarea. Usa Command con routing dinámico.

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 State(TypedDict):
    task: str
    routed_to: str
    result: str
    log: Annotated[list[str], operator.add]

CODE_KEYWORDS = ["código", "función", "bug", "implementar", "python", "api"]
DATA_KEYWORDS = ["datos", "análisis", "métricas", "dashboard", "sql"]

def router_agent(state: State) -> Command:
    task_lower = state["task"].lower()
    if any(kw in task_lower for kw in CODE_KEYWORDS):
        target = "code_agent"
    elif any(kw in task_lower for kw in DATA_KEYWORDS):
        target = "data_agent"
    else:
        target = "writing_agent"
    return Command(
        goto=target,
        update={
            "routed_to": target,
            "log": [f"router → {target}"],
        },
    )

def code_agent(state: State) -> dict:
    return {
        "result": f"[CODE] Implementación completada para: {state['task'][:40]}...",
        "log": ["code_agent: tarea completada"],
    }

def writing_agent(state: State) -> dict:
    return {
        "result": f"[WRITING] Contenido generado para: {state['task'][:40]}...",
        "log": ["writing_agent: tarea completada"],
    }

def data_agent(state: State) -> dict:
    return {
        "result": f"[DATA] Análisis completado para: {state['task'][:40]}...",
        "log": ["data_agent: tarea completada"],
    }

builder = StateGraph(State)
builder.add_node("router_agent", router_agent)
builder.add_node("code_agent", code_agent)
builder.add_node("writing_agent", writing_agent)
builder.add_node("data_agent", data_agent)

builder.add_edge(START, "router_agent")
builder.add_edge("code_agent", END)
builder.add_edge("writing_agent", END)
builder.add_edge("data_agent", END)

graph = builder.compile()

tasks = [
    "Implementar una función Python para validar emails",
    "Escribir un artículo sobre tendencias en AI",
    "Análisis de métricas de ventas Q4",
]

for task in tasks:
    result = graph.invoke({"task": task, "routed_to": "", "result": "", "log": []})
    print(f"Tarea: {task[:50]}...")
    print(f"  Resultado: {result['result']}")
    print(f"  Log: {result['log']}\n")
# Output esperado:
# Tarea: Implementar una función Python para validar emails...
#   Resultado: [CODE] Implementación completada para: Implementar una función Python para val...
#   Log: ['router → code_agent', 'code_agent: tarea completada']
#
# Tarea: Escribir un artículo sobre tendencias en AI...
#   Resultado: [WRITING] Contenido generado para: Escribir un artículo sobre tendencias en...
#   Log: ['router → writing_agent', 'writing_agent: tarea completada']
#
# Tarea: Análisis de métricas de ventas Q4...
#   Resultado: [DATA] Análisis completado para: Análisis de métricas de ventas Q4...
#   Log: ['router → data_agent', 'data_agent: tarea completada']

Ejercicio 5: Pipeline completo con handoffs y estado acumulativo (Avanzado)

Crea un pipeline de investigación con 4 agentes encadenados: sourcer (encuentra fuentes) → reader (lee y extrae info clave) → synthesizer (sintetiza en hallazgos) → presenter (genera presentación ejecutiva). Cada agente debe acumular su resultado en el estado Y pasar un resumen filtrado al siguiente. Al final, imprime el estado completo mostrando qué generó cada agente.

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 ResearchState(TypedDict):
    topic: str
    sources: list[str]
    key_extracts: list[str]
    synthesis: str
    presentation: str
    handoff_context: str
    log: Annotated[list[str], operator.add]

def sourcer(state: ResearchState) -> Command:
    sources = [
        "arxiv.org/paper-llm-agents-2025",
        "blog.langchain.dev/multi-agent-patterns",
        "docs.anthropic.com/agent-design",
    ]
    context = f"{len(sources)} fuentes académicas y técnicas identificadas"
    return Command(
        goto="reader",
        update={
            "sources": sources,
            "handoff_context": context,
            "log": [f"sourcer: {len(sources)} fuentes → reader"],
        },
    )

def reader(state: ResearchState) -> Command:
    extracts = [
        "LLM agents mejoran productividad en 40%",
        "Multi-agent supera single-agent en tareas complejas",
        "Context window management es el principal desafío",
    ]
    context = f"{len(extracts)} hallazgos clave extraídos de {len(state['sources'])} fuentes"
    return Command(
        goto="synthesizer",
        update={
            "key_extracts": extracts,
            "handoff_context": context,
            "log": [f"reader: {len(extracts)} extractos → synthesizer"],
        },
    )

def synthesizer(state: ResearchState) -> Command:
    synthesis = (
        f"SÍNTESIS: De {len(state['key_extracts'])} hallazgos, "
        f"la tendencia principal es que AI agents están madurando rápidamente. "
        f"Hallazgo clave: '{state['key_extracts'][0]}'"
    )
    context = "Síntesis lista: 1 tendencia principal, 1 hallazgo clave, 1 desafío"
    return Command(
        goto="presenter",
        update={
            "synthesis": synthesis,
            "handoff_context": context,
            "log": ["synthesizer: síntesis completa → presenter"],
        },
    )

def presenter(state: ResearchState) -> dict:
    presentation = (
        f"PRESENTACIÓN EJECUTIVA\n"
        f"{'=' * 40}\n"
        f"Tema: {state['topic']}\n"
        f"Fuentes consultadas: {len(state['sources'])}\n"
        f"Hallazgos clave: {len(state['key_extracts'])}\n"
        f"\n{state['synthesis']}\n"
        f"{'=' * 40}\n"
        f"Recomendación: Invertir en capacidades multi-agent"
    )
    return {
        "presentation": presentation,
        "log": ["presenter: presentación generada"],
    }

builder = StateGraph(ResearchState)
builder.add_node("sourcer", sourcer)
builder.add_node("reader", reader)
builder.add_node("synthesizer", synthesizer)
builder.add_node("presenter", presenter)

builder.add_edge(START, "sourcer")
builder.add_edge("presenter", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "AI Agents en producción empresarial",
    "sources": [], "key_extracts": [],
    "synthesis": "", "presentation": "",
    "handoff_context": "", "log": [],
})

print(result["presentation"])
print(f"\nPipeline completo:")
for step in result["log"]:
    print(f"  → {step}")
# Output esperado:
# PRESENTACIÓN EJECUTIVA
# ========================================
# Tema: AI Agents en producción empresarial
# Fuentes consultadas: 3
# Hallazgos clave: 3
#
# SÍNTESIS: De 3 hallazgos, la tendencia principal es que AI agents están madurando rápidamente. Hallazgo clave: 'LLM agents mejoran productividad en 40%'
# ========================================
# Recomendación: Invertir en capacidades multi-agent
#
# Pipeline completo:
#   → sourcer: 3 fuentes → reader
#   → reader: 3 extractos → synthesizer
#   → synthesizer: síntesis completa → presenter
#   → presenter: presentación generada

Resumen

En esta cápsula aprendiste:

• Un handoff es una transferencia directa de control entre agentes, sin pasar por un supervisor. Agent A decide que Agent B debe continuar y le transfiere el control con el contexto necesario
• Command(goto="target", update={...}) es el mecanismo de LangGraph para implementar handoffs a nivel de nodo. El nodo retorna un Command que indica al grafo dónde ir y qué actualizar
• Los handoff tools permiten que agentes con LLMs (creados con create_agent) decidan dinámicamente a quién transferir. La tool retorna Command con graph=Command.PARENT para navegar en el grafo padre
• El ToolMessage es obligatorio en handoffs con tools: completa el ciclo request-response que el LLM espera al llamar una herramienta
• El contexto se puede filtrar durante el handoff — no siempre necesitas pasar todo el estado. Pasa solo lo que el agente destino necesita
• Los handoffs se encadenan (A → B → C) para pipelines predecibles, y soportan retorno (A → B → A) para delegación con recuperación de control
• Handoffs vs supervisor: usa handoffs cuando el flujo es predecible y secuencial; usa supervisor cuando necesitas coordinación dinámica
• Cuidado con las cadenas largas: más de 4-5 handoffs hacen el debugging difícil. Si necesitas un "mapa" para entender el flujo, considera un supervisor

Próxima cápsula: Pattern Subagents — agentes que se ejecutan en contexto aislado. A diferencia de los handoffs donde el estado se comparte, los subagents trabajan en su propio espacio y solo devuelven el resultado. Esto previene el context bloat cuando tienes muchas subtareas.

Recursos adicionales

1. LangChain — Handoffs — Documentación oficial del patrón handoffs
2. LangGraph — Command — Referencia del objeto Command para navegación en grafos
3. create_handoff_tool API Reference — Convenience API para crear handoff tools
4. LangGraph — Multi-Agent Systems — Visión general de patrones multi-agente
5. Context Engineering for Agents — Cómo diseñar el flujo de contexto entre agentes

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