Orquestación Avanzada

Descripción de la cápsula

Los sistemas reales rara vez usan un solo patrón. Un e-commerce no tiene "un supervisor" o "un router" — tiene un supervisor que coordina departamentos, cada departamento tiene su propio router interno, y algunos agentes se comunican por handoffs mientras otros trabajan como subagentes aislados. La complejidad real viene de combinar patrones, no de dominar uno solo.

En las cápsulas anteriores aprendiste los 4 patrones fundamentales: Supervisor, Handoffs, Subagents y Router. También entiendes estado compartido vs aislado y cómo cada patrón lo maneja. Ahora toca componer. Esta cápsula te muestra las 4 técnicas de orquestación avanzada que aparecen en sistemas multi-agente de producción: combinación de patrones, jerarquías de agentes, ejecución en paralelo, y consenso entre agentes.

También aborda las dos habilidades que separan un prototipo de un sistema real: HITL en contextos multi-agente (¿dónde pones la aprobación humana cuando hay 6 agentes?) y debugging multi-agente (¿cómo sabes cuál agente causó el problema?).

Combinando patrones: Supervisor + Router

El patrón más común en producción es un supervisor de alto nivel que coordina el flujo general, con routers especializados dentro de dominios específicos. El supervisor decide "esta tarea es de análisis" y el router de análisis decide "este tipo de análisis lo maneja el agente financiero."

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class OrchestratorState(TypedDict):
    task: str
    domain: str
    sub_domain: str
    result: str
    trace: Annotated[list[str], operator.add]

def supervisor(state: OrchestratorState) -> dict:
    """Supervisor de alto nivel: decide el dominio."""
    task = state["task"].lower()

    if any(w in task for w in ["buscar", "investigar", "encontrar", "search"]):
        domain = "research"
    elif any(w in task for w in ["analizar", "comparar", "evaluar", "analyze"]):
        domain = "analysis"
    elif any(w in task for w in ["escribir", "redactar", "generar reporte", "write"]):
        domain = "writing"
    else:
        domain = "research"

    return {
        "domain": domain,
        "trace": [f"[supervisor] Tarea asignada al dominio: {domain}"],
    }

def route_by_domain(state: OrchestratorState) -> str:
    return f"router_{state['domain']}"

def router_research(state: OrchestratorState) -> dict:
    """Router interno del dominio research."""
    task = state["task"].lower()

    if "paper" in task or "arxiv" in task or "académico" in task:
        sub = "academic_search"
    elif "noticias" in task or "news" in task or "reciente" in task:
        sub = "news_search"
    else:
        sub = "web_search"

    return {
        "sub_domain": sub,
        "result": f"[{sub}] Resultados para: {state['task']}",
        "trace": [f"[router_research] Sub-dominio: {sub}"],
    }

def router_analysis(state: OrchestratorState) -> dict:
    """Router interno del dominio analysis."""
    task = state["task"].lower()

    if "financiero" in task or "precio" in task or "costo" in task:
        sub = "financial_analyst"
    elif "comparar" in task or "versus" in task or "vs" in task:
        sub = "comparison_analyst"
    else:
        sub = "general_analyst"

    return {
        "sub_domain": sub,
        "result": f"[{sub}] Análisis de: {state['task']}",
        "trace": [f"[router_analysis] Sub-dominio: {sub}"],
    }

def router_writing(state: OrchestratorState) -> dict:
    """Router interno del dominio writing."""
    task = state["task"].lower()

    if "ejecutivo" in task or "resumen" in task:
        sub = "executive_writer"
    elif "técnico" in task or "detallado" in task:
        sub = "technical_writer"
    else:
        sub = "general_writer"

    return {
        "sub_domain": sub,
        "result": f"[{sub}] Documento sobre: {state['task']}",
        "trace": [f"[router_writing] Sub-dominio: {sub}"],
    }

def output_node(state: OrchestratorState) -> dict:
    return {
        "trace": [f"[output] Resultado final del dominio '{state['domain']}' / '{state['sub_domain']}'"],
    }

builder = StateGraph(OrchestratorState)
builder.add_node("supervisor", supervisor)
builder.add_node("router_research", router_research)
builder.add_node("router_analysis", router_analysis)
builder.add_node("router_writing", router_writing)
builder.add_node("output", output_node)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_by_domain, {
    "router_research": "router_research",
    "router_analysis": "router_analysis",
    "router_writing": "router_writing",
})
builder.add_edge("router_research", "output")
builder.add_edge("router_analysis", "output")
builder.add_edge("router_writing", "output")
builder.add_edge("output", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

tasks = [
    "Buscar papers académicos sobre transformers",
    "Analizar costos financieros del proyecto",
    "Escribir resumen ejecutivo del Q1",
]

for i, task in enumerate(tasks):
    config = {"configurable": {"thread_id": f"orch-{i}"}}
    result = graph.invoke(
        {"task": task, "domain": "", "sub_domain": "", "result": "", "trace": []},
        config,
    )
    print(f"\nTarea: {task}")
    for step in result["trace"]:
        print(f"  {step}")
    print(f"  Resultado: {result['result']}")
# Output esperado:
# Tarea: Buscar papers académicos sobre transformers
#   [supervisor] Tarea asignada al dominio: research
#   [router_research] Sub-dominio: academic_search
#   [output] Resultado final del dominio 'research' / 'academic_search'
#   Resultado: [academic_search] Resultados para: Buscar papers académicos sobre transformers
#
# Tarea: Analizar costos financieros del proyecto
#   [supervisor] Tarea asignada al dominio: analysis
#   [router_analysis] Sub-dominio: financial_analyst
#   [output] Resultado final del dominio 'analysis' / 'financial_analyst'
#   Resultado: [financial_analyst] Análisis de: Analizar costos financieros del proyecto
#
# Tarea: Escribir resumen ejecutivo del Q1
#   [supervisor] Tarea asignada al dominio: writing
#   [router_writing] Sub-dominio: executive_writer
#   [output] Resultado final del dominio 'writing' / 'executive_writer'
#   Resultado: [executive_writer] Documento sobre: Escribir resumen ejecutivo del Q1

El flujo tiene dos niveles de decisión: el supervisor elige el dominio (research/analysis/writing) y el router interno elige el agente especializado (academic_search/financial_analyst/executive_writer). El trace te muestra exactamente la cadena de decisiones.

Combinando patrones: Handoffs + Subagents

Otro patrón potente: una cadena de handoffs secuenciales donde cada agente usa subagentes para subtareas internas. El agente principal recibe el trabajo, delega sub-tareas a sus workers, consolida, y pasa el resultado al siguiente agente en la cadena.

from dotenv import load_dotenv
load_dotenv()

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

class PipelineState(TypedDict):
    topic: str
    raw_findings: list[str]
    analysis: str
    final_report: str
    trace: Annotated[list[str], operator.add]

def researcher_agent(state: PipelineState) -> dict:
    """Researcher usa 'subagentes' internos para buscar en múltiples fuentes."""
    topic = state["topic"]

    web_result = f"[web] 3 artículos sobre {topic}"
    arxiv_result = f"[arxiv] 2 papers relevantes sobre {topic}"
    news_result = f"[news] 1 noticia reciente sobre {topic}"

    combined = [web_result, arxiv_result, news_result]

    return {
        "raw_findings": combined,
        "trace": [
            f"[researcher] Inicio — tema: {topic}",
            f"[researcher:web_worker] {web_result}",
            f"[researcher:arxiv_worker] {arxiv_result}",
            f"[researcher:news_worker] {news_result}",
            f"[researcher] Handoff → analyst con {len(combined)} hallazgos",
        ],
    }

def analyst_agent(state: PipelineState) -> dict:
    """Analyst usa subagentes para diferentes tipos de análisis."""
    findings_count = len(state["raw_findings"])

    pattern_analysis = f"Patrón identificado: tendencia creciente en {state['topic']}"
    contradiction_check = "Sin contradicciones entre fuentes"

    analysis = f"{pattern_analysis}. {contradiction_check}. Basado en {findings_count} fuentes."

    return {
        "analysis": analysis,
        "trace": [
            f"[analyst] Recibido: {findings_count} hallazgos",
            f"[analyst:pattern_worker] {pattern_analysis}",
            f"[analyst:contradiction_worker] {contradiction_check}",
            f"[analyst] Handoff → writer con análisis completo",
        ],
    }

def writer_agent(state: PipelineState) -> dict:
    """Writer genera el reporte final."""
    report = (
        f"# Reporte: {state['topic']}\n\n"
        f"## Análisis\n{state['analysis']}\n\n"
        f"## Fuentes\n" + "\n".join(f"- {f}" for f in state["raw_findings"])
    )

    return {
        "final_report": report,
        "trace": [
            f"[writer] Recibido: análisis de {len(state['analysis'])} chars",
            f"[writer] Reporte generado: {len(report)} chars",
        ],
    }

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

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

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "handoff-subagent-001"}}
result = graph.invoke(
    {"topic": "AI multi-agent systems", "raw_findings": [], "analysis": "", "final_report": "", "trace": []},
    config,
)

print("=== Trace completo ===")
for step in result["trace"]:
    print(f"  {step}")
print(f"\n=== Reporte ===\n{result['final_report']}")
# Output esperado:
# === Trace completo ===
#   [researcher] Inicio — tema: AI multi-agent systems
#   [researcher:web_worker] [web] 3 artículos sobre AI multi-agent systems
#   [researcher:arxiv_worker] [arxiv] 2 papers relevantes sobre AI multi-agent systems
#   [researcher:news_worker] [news] 1 noticia reciente sobre AI multi-agent systems
#   [researcher] Handoff → analyst con 3 hallazgos
#   [analyst] Recibido: 3 hallazgos
#   [analyst:pattern_worker] Patrón identificado: tendencia creciente en AI multi-agent systems
#   [analyst:contradiction_worker] Sin contradicciones entre fuentes
#   [analyst] Handoff → writer con análisis completo
#   [writer] Recibido: análisis de 97 chars
#   [writer] Reporte generado: 236 chars
#
# === Reporte ===
# # Reporte: AI multi-agent systems
#
# ## Análisis
# Patrón identificado: tendencia creciente en AI multi-agent systems. Sin contradicciones entre fuentes. Basado en 3 fuentes.
#
# ## Fuentes
# - [web] 3 artículos sobre AI multi-agent systems
# - [arxiv] 2 papers relevantes sobre AI multi-agent systems
# - [news] 1 noticia reciente sobre AI multi-agent systems

El trace muestra los dos niveles: los handoffs entre agentes principales (researcher → analyst → writer) y los workers internos de cada agente (researcher:web_worker, analyst:pattern_worker). Cada agente es autónomo en cómo organiza su trabajo interno.

Agentes jerárquicos: supervisores de supervisores

Cuando el sistema crece más allá de 6 agentes, un solo supervisor se vuelve un cuello de botella. La solución: jerarquía. Un supervisor de alto nivel delega a supervisores de departamento, que a su vez coordinan workers especializados.

Top Supervisor
├── Research Supervisor
│   ├── Web Search Worker
│   ├── Academic Search Worker
│   └── News Search Worker
└── Analysis Supervisor
    ├── Pattern Analyst Worker
    └── Fact-Check Worker

Cuándo agregar jerarquía:

• ✅ >6 agentes en el sistema
• ✅ Dominios claramente distintos (research ≠ analysis ≠ writing)
• ✅ Diferentes niveles de aprobación por departamento
• ✅ Necesitas escalar un departamento sin tocar los demás
• ❌ <4 agentes — un supervisor plano es más simple
• ❌ Todos los agentes hacen tareas similares — no hay dominios claros

from dotenv import load_dotenv
load_dotenv()

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

class HierarchyState(TypedDict):
    query: str
    web_results: list[str]
    academic_results: list[str]
    news_results: list[str]
    all_findings: list[str]
    patterns: list[str]
    fact_check: str
    final_output: str
    trace: Annotated[list[str], operator.add]

def top_supervisor(state: HierarchyState) -> dict:
    return {
        "trace": [f"[top_supervisor] Tarea recibida: '{state['query']}' → delegando a research_supervisor"],
    }

def research_supervisor(state: HierarchyState) -> dict:
    return {
        "trace": [f"[research_supervisor] Coordinando 3 workers para: '{state['query']}'"],
    }

def web_worker(state: HierarchyState) -> dict:
    results = [f"Web resultado {i+1} sobre {state['query']}" for i in range(3)]
    return {
        "web_results": results,
        "trace": [f"[research:web_worker] {len(results)} resultados encontrados"],
    }

def academic_worker(state: HierarchyState) -> dict:
    results = [f"Paper {i+1} sobre {state['query']}" for i in range(2)]
    return {
        "academic_results": results,
        "trace": [f"[research:academic_worker] {len(results)} papers encontrados"],
    }

def news_worker(state: HierarchyState) -> dict:
    results = [f"Noticia sobre {state['query']}"]
    return {
        "news_results": results,
        "trace": [f"[research:news_worker] {len(results)} noticias encontradas"],
    }

def research_merge(state: HierarchyState) -> dict:
    all_findings = state["web_results"] + state["academic_results"] + state["news_results"]
    return {
        "all_findings": all_findings,
        "trace": [f"[research_supervisor] Merge: {len(all_findings)} hallazgos totales → delegando a analysis_supervisor"],
    }

def analysis_supervisor(state: HierarchyState) -> dict:
    return {
        "trace": [f"[analysis_supervisor] Analizando {len(state['all_findings'])} hallazgos"],
    }

def pattern_worker(state: HierarchyState) -> dict:
    patterns = [f"Patrón: consenso sobre {state['query']}", "Patrón: crecimiento en adopción"]
    return {
        "patterns": patterns,
        "trace": [f"[analysis:pattern_worker] {len(patterns)} patrones identificados"],
    }

def factcheck_worker(state: HierarchyState) -> dict:
    return {
        "fact_check": "Sin contradicciones detectadas entre las 6 fuentes",
        "trace": [f"[analysis:factcheck_worker] Verificación completada"],
    }

def analysis_merge(state: HierarchyState) -> dict:
    output = (
        f"Hallazgos: {len(state['all_findings'])} | "
        f"Patrones: {len(state['patterns'])} | "
        f"Fact-check: {state['fact_check']}"
    )
    return {
        "final_output": output,
        "trace": [f"[top_supervisor] Resultado final consolidado"],
    }

builder = StateGraph(HierarchyState)

builder.add_node("top_supervisor", top_supervisor)
builder.add_node("research_supervisor", research_supervisor)
builder.add_node("web_worker", web_worker)
builder.add_node("academic_worker", academic_worker)
builder.add_node("news_worker", news_worker)
builder.add_node("research_merge", research_merge)
builder.add_node("analysis_supervisor", analysis_supervisor)
builder.add_node("pattern_worker", pattern_worker)
builder.add_node("factcheck_worker", factcheck_worker)
builder.add_node("analysis_merge", analysis_merge)

builder.add_edge(START, "top_supervisor")
builder.add_edge("top_supervisor", "research_supervisor")
builder.add_edge("research_supervisor", "web_worker")
builder.add_edge("research_supervisor", "academic_worker")
builder.add_edge("research_supervisor", "news_worker")
builder.add_edge("web_worker", "research_merge")
builder.add_edge("academic_worker", "research_merge")
builder.add_edge("news_worker", "research_merge")
builder.add_edge("research_merge", "analysis_supervisor")
builder.add_edge("analysis_supervisor", "pattern_worker")
builder.add_edge("analysis_supervisor", "factcheck_worker")
builder.add_edge("pattern_worker", "analysis_merge")
builder.add_edge("factcheck_worker", "analysis_merge")
builder.add_edge("analysis_merge", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "hierarchy-001"}}
result = graph.invoke(
    {
        "query": "multi-agent AI systems",
        "web_results": [], "academic_results": [], "news_results": [],
        "all_findings": [], "patterns": [], "fact_check": "", "final_output": "",
        "trace": [],
    },
    config,
)

print("=== Flujo jerárquico ===")
for step in result["trace"]:
    print(f"  {step}")
print(f"\n  Final: {result['final_output']}")
# Output esperado:
# === Flujo jerárquico ===
#   [top_supervisor] Tarea recibida: 'multi-agent AI systems' → delegando a research_supervisor
#   [research_supervisor] Coordinando 3 workers para: 'multi-agent AI systems'
#   [research:web_worker] 3 resultados encontrados
#   [research:academic_worker] 2 papers encontrados
#   [research:news_worker] 1 noticias encontradas
#   [research_supervisor] Merge: 6 hallazgos totales → delegando a analysis_supervisor
#   [analysis_supervisor] Analizando 6 hallazgos
#   [analysis:pattern_worker] 2 patrones identificados
#   [analysis:factcheck_worker] Verificación completada
#   [top_supervisor] Resultado final consolidado
#
#   Final: Hallazgos: 6 | Patrones: 2 | Fact-check: Sin contradicciones detectadas entre las 6 fuentes

La jerarquía aporta claridad: cada supervisor sabe exactamente qué workers coordina, y cada worker tiene una responsabilidad aislada. Si necesitas agregar un nuevo worker al equipo de research (por ejemplo, patent_worker), solo tocas ese subgrafo — el analysis_supervisor ni se entera.

Ejecución paralela: fan-out / fan-in

Cuando los agentes trabajan en subtareas independientes, no hay razón para esperar secuencialmente. Fan-out envía la tarea a múltiples agentes simultáneamente, fan-in recopila los resultados y los fusiona.

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class ParallelState(TypedDict):
    query: str
    researcher_output: str
    analyst_output: str
    factchecker_output: str
    merged_result: str
    timing: Annotated[list[str], operator.add]

def distribute(state: ParallelState) -> dict:
    return {
        "timing": [f"[distribute] Enviando '{state['query']}' a 3 agentes en paralelo"],
    }

def researcher(state: ParallelState) -> dict:
    start = time.time()
    time.sleep(0.1)
    elapsed = (time.time() - start) * 1000
    output = f"5 fuentes encontradas sobre '{state['query']}'"
    return {
        "researcher_output": output,
        "timing": [f"[researcher] {elapsed:.0f}ms — {output}"],
    }

def analyst(state: ParallelState) -> dict:
    start = time.time()
    time.sleep(0.15)
    elapsed = (time.time() - start) * 1000
    output = f"Tendencia alcista identificada en '{state['query']}'"
    return {
        "analyst_output": output,
        "timing": [f"[analyst] {elapsed:.0f}ms — {output}"],
    }

def factchecker(state: ParallelState) -> dict:
    start = time.time()
    time.sleep(0.08)
    elapsed = (time.time() - start) * 1000
    output = f"4/5 fuentes verificadas como confiables"
    return {
        "factchecker_output": output,
        "timing": [f"[factchecker] {elapsed:.0f}ms — {output}"],
    }

def merge(state: ParallelState) -> dict:
    merged = (
        f"Investigación: {state['researcher_output']} | "
        f"Análisis: {state['analyst_output']} | "
        f"Verificación: {state['factchecker_output']}"
    )
    return {
        "merged_result": merged,
        "timing": [f"[merge] Resultados de 3 agentes combinados ({len(merged)} chars)"],
    }

builder = StateGraph(ParallelState)
builder.add_node("distribute", distribute)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_node("factchecker", factchecker)
builder.add_node("merge", merge)

builder.add_edge(START, "distribute")
builder.add_edge("distribute", "researcher")
builder.add_edge("distribute", "analyst")
builder.add_edge("distribute", "factchecker")
builder.add_edge("researcher", "merge")
builder.add_edge("analyst", "merge")
builder.add_edge("factchecker", "merge")
builder.add_edge("merge", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "parallel-001"}}
overall_start = time.time()
result = graph.invoke(
    {
        "query": "LLM agent architectures",
        "researcher_output": "", "analyst_output": "",
        "factchecker_output": "", "merged_result": "",
        "timing": [],
    },
    config,
)
overall_ms = (time.time() - overall_start) * 1000

print("=== Ejecución paralela ===")
for step in result["timing"]:
    print(f"  {step}")
print(f"\n  Tiempo total: {overall_ms:.0f}ms")
print(f"  (secuencial sería ~330ms, paralelo es ~150ms)")
# Output esperado:
# === Ejecución paralela ===
#   [distribute] Enviando 'LLM agent architectures' a 3 agentes en paralelo
#   [researcher] 100ms — 5 fuentes encontradas sobre 'LLM agent architectures'
#   [analyst] 150ms — Tendencia alcista identificada en 'LLM agent architectures'
#   [factchecker] 80ms — 4/5 fuentes verificadas como confiables
#   [merge] Resultados de 3 agentes combinados (165 chars)
#
#   Tiempo total: ~180ms
#   (secuencial sería ~330ms, paralelo es ~150ms)

LangGraph ejecuta los nodos researcher, analyst y factchecker en paralelo porque los tres tienen la misma dependencia (distribute) y no dependen entre sí. El nodo merge espera a que los tres terminen (fan-in). No necesitas threads manuales — el grafo lo resuelve.

Consenso entre agentes

Cuando la precisión es crítica, puedes hacer que múltiples agentes analicen los mismos datos y comparen conclusiones. Dos estrategias: votación (mayoría gana) y ponderación por confianza.

from dotenv import load_dotenv
load_dotenv()

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

class ConsensusState(TypedDict):
    data: str
    opinions: Annotated[list[dict], operator.add]
    consensus: str
    method: str
    trace: Annotated[list[str], operator.add]

def analyst_a(state: ConsensusState) -> dict:
    return {
        "opinions": [{"agent": "analyst_a", "conclusion": "bullish", "confidence": 0.85}],
        "trace": [f"[analyst_a] Conclusión: bullish (85% confianza)"],
    }

def analyst_b(state: ConsensusState) -> dict:
    return {
        "opinions": [{"agent": "analyst_b", "conclusion": "bullish", "confidence": 0.72}],
        "trace": [f"[analyst_b] Conclusión: bullish (72% confianza)"],
    }

def analyst_c(state: ConsensusState) -> dict:
    return {
        "opinions": [{"agent": "analyst_c", "conclusion": "bearish", "confidence": 0.68}],
        "trace": [f"[analyst_c] Conclusión: bearish (68% confianza)"],
    }

def voting_consensus(state: ConsensusState) -> dict:
    """Mayoría simple: la conclusión más votada gana."""
    from collections import Counter
    votes = Counter(op["conclusion"] for op in state["opinions"])
    winner = votes.most_common(1)[0]

    return {
        "consensus": winner[0],
        "method": "voting",
        "trace": [
            f"[consensus:voting] Votos: {dict(votes)}",
            f"[consensus:voting] Ganador: {winner[0]} ({winner[1]}/{len(state['opinions'])} votos)",
        ],
    }

def weighted_consensus(state: ConsensusState) -> dict:
    """Ponderación: la conclusión con mayor confianza acumulada gana."""
    scores: dict[str, float] = {}
    for op in state["opinions"]:
        conclusion = op["conclusion"]
        scores[conclusion] = scores.get(conclusion, 0) + op["confidence"]

    winner = max(scores, key=scores.get)

    return {
        "consensus": winner,
        "method": "weighted",
        "trace": [
            f"[consensus:weighted] Scores ponderados: {scores}",
            f"[consensus:weighted] Ganador: {winner} (score: {scores[winner]:.2f})",
        ],
    }

def build_consensus_graph(method: str = "voting"):
    builder = StateGraph(ConsensusState)
    builder.add_node("analyst_a", analyst_a)
    builder.add_node("analyst_b", analyst_b)
    builder.add_node("analyst_c", analyst_c)

    consensus_fn = voting_consensus if method == "voting" else weighted_consensus
    builder.add_node("consensus", consensus_fn)

    builder.add_edge(START, "analyst_a")
    builder.add_edge(START, "analyst_b")
    builder.add_edge(START, "analyst_c")
    builder.add_edge("analyst_a", "consensus")
    builder.add_edge("analyst_b", "consensus")
    builder.add_edge("analyst_c", "consensus")
    builder.add_edge("consensus", END)

    checkpointer = MemorySaver()
    return builder.compile(checkpointer=checkpointer)

print("=== Método 1: Votación ===")
graph_vote = build_consensus_graph("voting")
config1 = {"configurable": {"thread_id": "consensus-vote"}}
result1 = graph_vote.invoke(
    {"data": "Q1 financial report", "opinions": [], "consensus": "", "method": "", "trace": []},
    config1,
)
for step in result1["trace"]:
    print(f"  {step}")

print(f"\n=== Método 2: Ponderación por confianza ===")
graph_weighted = build_consensus_graph("weighted")
config2 = {"configurable": {"thread_id": "consensus-weighted"}}
result2 = graph_weighted.invoke(
    {"data": "Q1 financial report", "opinions": [], "consensus": "", "method": "", "trace": []},
    config2,
)
for step in result2["trace"]:
    print(f"  {step}")
# Output esperado:
# === Método 1: Votación ===
#   [analyst_a] Conclusión: bullish (85% confianza)
#   [analyst_b] Conclusión: bullish (72% confianza)
#   [analyst_c] Conclusión: bearish (68% confianza)
#   [consensus:voting] Votos: {'bullish': 2, 'bearish': 1}
#   [consensus:voting] Ganador: bullish (2/3 votos)
#
# === Método 2: Ponderación por confianza ===
#   [analyst_a] Conclusión: bullish (85% confianza)
#   [analyst_b] Conclusión: bullish (72% confianza)
#   [analyst_c] Conclusión: bearish (68% confianza)
#   [consensus:weighted] Scores ponderados: {'bullish': 1.57, 'bearish': 0.68}
#   [consensus:weighted] Ganador: bullish (score: 1.57)

En este caso ambos métodos coinciden, pero no siempre es así. Si analyst_a tuviera 0.51 de confianza y analyst_c tuviera 0.95, la votación seguiría diciendo "bullish" (2 vs 1), pero la ponderación podría favorecer "bearish" si la confianza acumulada lo justifica.

HITL en multi-agente: dónde poner la aprobación humana

Con un solo agente, la decisión era simple: ¿antes o después de esta acción? Con múltiples agentes, tienes 4 puntos posibles:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class HITLMultiState(TypedDict):
    task: str
    plan: dict
    research_result: str
    analysis_result: str
    final_output: str
    trace: Annotated[list[str], operator.add]

def supervisor_plan(state: HITLMultiState) -> dict:
    plan = {
        "agents": ["researcher", "analyst"],
        "estimated_cost": 2.50,
        "estimated_time": "30s",
    }
    return {
        "plan": plan,
        "trace": [f"[supervisor] Plan generado: {plan['agents']} (${plan['estimated_cost']})"],
    }

def hitl_approve_plan(state: HITLMultiState) -> dict:
    """HITL centralizado: aprobación antes de ejecutar cualquier agente."""
    plan = state["plan"]

    response = interrupt({
        "type": "plan_approval",
        "message": (
            f"El supervisor propone:\n"
            f"  Agentes: {plan['agents']}\n"
            f"  Costo estimado: ${plan['estimated_cost']}\n"
            f"  Tiempo estimado: {plan['estimated_time']}\n"
            f"¿Aprobar el plan completo?"
        ),
        "plan": plan,
    })

    action = response if isinstance(response, str) else response.get("action", "approve")

    if action == "cancel":
        return {"trace": [f"[hitl] Plan cancelado por el usuario"]}

    return {"trace": [f"[hitl] Plan aprobado — ejecutando agentes"]}

def should_continue(state: HITLMultiState) -> str:
    last_trace = state["trace"][-1] if state["trace"] else ""
    if "cancelado" in last_trace:
        return "end"
    return "researcher"

def researcher(state: HITLMultiState) -> dict:
    result = f"5 fuentes encontradas sobre '{state['task']}'"
    return {
        "research_result": result,
        "trace": [f"[researcher] {result}"],
    }

def analyst(state: HITLMultiState) -> dict:
    result = f"Análisis completo de: {state['research_result']}"
    return {
        "analysis_result": result,
        "trace": [f"[analyst] {result}"],
    }

def compile_output(state: HITLMultiState) -> dict:
    output = f"Reporte: {state['analysis_result']}"
    return {
        "final_output": output,
        "trace": [f"[supervisor] Output final compilado"],
    }

builder = StateGraph(HITLMultiState)
builder.add_node("plan", supervisor_plan)
builder.add_node("approve", hitl_approve_plan)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_node("compile", compile_output)

builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_conditional_edges("approve", should_continue, {
    "researcher": "researcher",
    "end": END,
})
builder.add_edge("researcher", "analyst")
builder.add_edge("analyst", "compile")
builder.add_edge("compile", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

print("=== HITL centralizado: aprobar plan ===")
config = {"configurable": {"thread_id": "hitl-multi-001"}}
graph.invoke(
    {"task": "AI trends 2026", "plan": {}, "research_result": "", "analysis_result": "", "final_output": "", "trace": []},
    config,
)

state = graph.get_state(config)
print(f"  Esperando aprobación. Siguiente: {state.next}")

result = graph.invoke(Command(resume="approve"), config)
print(f"\n=== Flujo completo ===")
for step in result["trace"]:
    print(f"  {step}")
# Output esperado:
# === HITL centralizado: aprobar plan ===
#   Esperando aprobación. Siguiente: ('approve',)
#
# === Flujo completo ===
#   [supervisor] Plan generado: ['researcher', 'analyst'] ($2.5)
#   [hitl] Plan aprobado — ejecutando agentes
#   [researcher] 5 fuentes encontradas sobre 'AI trends 2026'
#   [analyst] Análisis completo de: 5 fuentes encontradas sobre 'AI trends 2026'
#   [supervisor] Output final compilado

La decisión entre centralizar y distribuir HITL depende de tu contexto:

• ✅ Centralizado cuando quieres una sola interrupción y el humano confía en los agentes individuales
• ✅ Distribuido (post-agente) cuando cada agente puede fallar de formas diferentes y necesitas validar cada paso
• ❌ Evita HITL en cada agente Y al final — el usuario terminará aprobando 7 cosas para una sola tarea

Debugging multi-agente: la habilidad crítica

Debuggear un sistema multi-agente es significativamente más difícil que un solo agente. Con un agente, el problema está "en algún lugar del pipeline." Con 4 agentes, el problema puede estar en el agente, en la comunicación entre agentes, en el estado compartido, o en el supervisor que delegó mal.

Logging por agente

La regla más importante: cada agente loguea con su nombre como prefijo. Sin esto, un log de 200 líneas es imposible de leer.

from dotenv import load_dotenv
load_dotenv()

import time
import logging

class AgentLogger:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.logger = logging.getLogger(f"agent.{agent_name}")
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                f"%(asctime)s | %(levelname)-5s | [{agent_name}] %(message)s",
                datefmt="%H:%M:%S",
            )
            handler.setFormatter(formatter)
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.DEBUG)

    def info(self, msg: str):
        self.logger.info(msg)

    def error(self, msg: str):
        self.logger.error(msg)

    def debug(self, msg: str):
        self.logger.debug(msg)

    def warning(self, msg: str):
        self.logger.warning(msg)

researcher_log = AgentLogger("researcher")
analyst_log = AgentLogger("analyst")
writer_log = AgentLogger("writer")
supervisor_log = AgentLogger("supervisor")

supervisor_log.info("Delegando tarea a researcher")
researcher_log.info("Buscando en 3 fuentes")
researcher_log.debug("web_search: 200 OK (120ms)")
researcher_log.debug("arxiv_search: 200 OK (340ms)")
researcher_log.warning("news_search: timeout después de 5s, reintentando...")
researcher_log.debug("news_search: 200 OK (890ms, intento 2)")
researcher_log.info("3/3 fuentes completadas → handoff a analyst")

analyst_log.info("Recibido: 6 hallazgos de researcher")
analyst_log.debug("Identificando patrones...")
analyst_log.info("2 patrones, 0 contradicciones → handoff a writer")

writer_log.info("Generando reporte (formato: bullet_points)")
writer_log.info("Reporte listo: 450 chars")

supervisor_log.info("Pipeline completo en 1.2s")
# Output esperado:
# 14:30:01 | INFO  | [supervisor] Delegando tarea a researcher
# 14:30:01 | INFO  | [researcher] Buscando en 3 fuentes
# 14:30:01 | DEBUG | [researcher] web_search: 200 OK (120ms)
# 14:30:01 | DEBUG | [researcher] arxiv_search: 200 OK (340ms)
# 14:30:01 | WARN  | [researcher] news_search: timeout después de 5s, reintentando...
# 14:30:01 | DEBUG | [researcher] news_search: 200 OK (890ms, intento 2)
# 14:30:01 | INFO  | [researcher] 3/3 fuentes completadas → handoff a analyst
# 14:30:01 | INFO  | [analyst] Recibido: 6 hallazgos de researcher
# 14:30:01 | DEBUG | [analyst] Identificando patrones...
# 14:30:01 | INFO  | [analyst] 2 patrones, 0 contradicciones → handoff a writer
# 14:30:01 | INFO  | [writer] Generando reporte (formato: bullet_points)
# 14:30:01 | INFO  | [writer] Reporte listo: 450 chars
# 14:30:01 | INFO  | [supervisor] Pipeline completo en 1.2s

Con el prefijo [researcher], [analyst], [writer], puedes filtrar logs por agente: grep "[researcher]" logs.txt te da solo lo que hizo el researcher.

Tracing del flujo: quién manejó qué y cuándo

Además de logs individuales, necesitas una vista de alto nivel del flujo entre agentes:

from dotenv import load_dotenv
load_dotenv()

import time

class FlowTracer:
    def __init__(self):
        self.events: list[dict] = []
        self.start_time = time.time()

    def record(self, agent: str, event: str, data: dict | None = None):
        elapsed = (time.time() - self.start_time) * 1000
        entry = {
            "agent": agent,
            "event": event,
            "elapsed_ms": round(elapsed),
            "data": data or {},
        }
        self.events.append(entry)

    def print_timeline(self):
        print(f"\n{'Agent':<15} {'Event':<25} {'Time':>8}  Details")
        print("-" * 70)
        for e in self.events:
            details = ", ".join(f"{k}={v}" for k, v in e["data"].items()) if e["data"] else ""
            print(f"{e['agent']:<15} {e['event']:<25} {e['elapsed_ms']:>6}ms  {details}")

    def find_bottleneck(self) -> dict:
        agent_times: dict[str, list[int]] = {}
        for e in self.events:
            agent = e["agent"]
            if agent not in agent_times:
                agent_times[agent] = []
            agent_times[agent].append(e["elapsed_ms"])

        durations = {}
        for agent, times in agent_times.items():
            durations[agent] = max(times) - min(times)

        slowest = max(durations, key=durations.get)
        return {"agent": slowest, "duration_ms": durations[slowest]}

tracer = FlowTracer()

tracer.record("supervisor", "task_received", {"topic": "AI agents"})
tracer.record("supervisor", "delegated", {"to": "researcher"})
time.sleep(0.05)
tracer.record("researcher", "started", {"sources": 3})
time.sleep(0.1)
tracer.record("researcher", "completed", {"findings": 6})
tracer.record("researcher", "handoff", {"to": "analyst"})
time.sleep(0.02)
tracer.record("analyst", "started", {"input_size": 6})
time.sleep(0.15)
tracer.record("analyst", "completed", {"patterns": 2})
tracer.record("analyst", "handoff", {"to": "writer"})
time.sleep(0.01)
tracer.record("writer", "started", {"format": "bullets"})
time.sleep(0.05)
tracer.record("writer", "completed", {"report_chars": 450})
tracer.record("supervisor", "pipeline_done", {})

tracer.print_timeline()

bottleneck = tracer.find_bottleneck()
print(f"\n⚠️ Cuello de botella: {bottleneck['agent']} ({bottleneck['duration_ms']}ms)")
# Output esperado:
# Agent           Event                       Time  Details
# ----------------------------------------------------------------------
# supervisor      task_received                  0ms  topic=AI agents
# supervisor      delegated                      0ms  to=researcher
# researcher      started                       50ms  sources=3
# researcher      completed                    150ms  findings=6
# researcher      handoff                      150ms  to=analyst
# analyst         started                      170ms  input_size=6
# analyst         completed                    320ms  patterns=2
# analyst         handoff                      320ms  to=writer
# writer          started                      330ms  format=bullets
# writer          completed                    380ms  report_chars=450
# supervisor      pipeline_done                380ms
#
# ⚠️ Cuello de botella: analyst (150ms)

Visualización del grafo

draw_mermaid_png() genera una imagen del grafo completo que puedes compartir con tu equipo:

from dotenv import load_dotenv
load_dotenv()

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

class SimpleState(TypedDict):
    data: str

def supervisor(state: SimpleState) -> dict:
    return state

def researcher(state: SimpleState) -> dict:
    return state

def analyst(state: SimpleState) -> dict:
    return state

def writer(state: SimpleState) -> dict:
    return state

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

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

graph = builder.compile()

mermaid_code = graph.get_graph().draw_mermaid()
print("=== Diagrama Mermaid ===")
print(mermaid_code)

# Para generar una imagen PNG:
# png_data = graph.get_graph().draw_mermaid_png()
# with open("graph_diagram.png", "wb") as f:
#     f.write(png_data)
# print("Diagrama guardado en graph_diagram.png")
# Output esperado:
# === Diagrama Mermaid ===
# %%{init: {'flowchart': {'curve': 'linear'}}}%%
# graph TD;
# 	__start__([<p>__start__</p>]):::first
# 	supervisor(supervisor)
# 	researcher(researcher)
# 	analyst(analyst)
# 	writer(writer)
# 	__end__([<p>__end__</p>]):::last
# 	__start__ --> supervisor;
# 	supervisor --> researcher;
# 	researcher --> analyst;
# 	analyst --> writer;
# 	writer --> __end__;

Failure modes comunes

Troubleshooting

Problema 1: "Los nodos paralelos no se ejecutan en paralelo"

Síntoma: El tiempo total es la suma de los tiempos individuales, no el máximo.

Causa: Los edges no salen del mismo nodo. Para que LangGraph ejecute nodos en paralelo, deben compartir el mismo nodo padre.

Solución: Verifica que los edges van del mismo nodo a los workers:

builder.add_edge("distribute", "agent_a")
builder.add_edge("distribute", "agent_b")
builder.add_edge("distribute", "agent_c")

Problema 2: "El consenso siempre devuelve el mismo resultado"

Síntoma: El nodo de consenso siempre elige al primer agente.

Causa: La lista opinions usa Annotated[list, operator.add] pero los agentes no agregan — sobrescriben.

Solución: Cada agente debe retornar una lista con un solo elemento para que operator.add los acumule:

def analyst_a(state) -> dict:
    return {"opinions": [{"agent": "a", "conclusion": "bullish"}]}

Problema 3: "El supervisor delega en loop infinito"

Síntoma: El grafo nunca llega a END.

Causa: La condición de salida del supervisor siempre elige delegar más trabajo.

Solución: Agrega un contador al estado y una condición de salida:

class State(TypedDict):
    iteration: int

def should_continue(state) -> str:
    if state["iteration"] >= 3:
        return "output"
    return "delegate"

Problema 4: "El HITL interrupt se dispara para cada agente"

Síntoma: El sistema pide aprobación 5 veces para una sola tarea.

Causa: Cada agente tiene su propio interrupt() y todos se ejecutan.

Solución: Centraliza el HITL en el supervisor. Solo el supervisor interrumpe, los workers ejecutan sin pausar:

def supervisor(state):
    response = interrupt({"type": "plan_approval", ...})
    # Solo aquí, no en cada worker

Problema 5: "No encuentro cuál agente causó el error"

Síntoma: El output final es incorrecto pero no sabes qué agente falló.

Causa: Los agentes no loguean con prefijo y el estado compartido no registra quién escribió qué.

Solución: Usa AgentLogger con nombre y agrega source a cada dato del estado:

return {
    "findings": results,
    "trace": [f"[researcher] 5 hallazgos producidos"],
}

Ejercicios

Ejercicio 1: Supervisor + Router de dos dominios (Fácil)

Crea un sistema donde un supervisor clasifica tareas en "technical" o "business", y cada dominio tiene un router interno que selecciona entre 2 agentes especializados. Prueba con 3 tareas diferentes.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    domain: str
    agent: str
    result: str
    trace: Annotated[list[str], operator.add]

def supervisor(state: State) -> dict:
    task = state["task"].lower()
    domain = "technical" if any(w in task for w in ["code", "bug", "deploy", "api"]) else "business"
    return {"domain": domain, "trace": [f"[supervisor] → {domain}"]}

def route_domain(state: State) -> str:
    return f"router_{state['domain']}"

def router_technical(state: State) -> dict:
    task = state["task"].lower()
    agent = "debugger" if "bug" in task else "deployer"
    return {
        "agent": agent,
        "result": f"[{agent}] Procesado: {state['task']}",
        "trace": [f"[router_technical] → {agent}"],
    }

def router_business(state: State) -> dict:
    task = state["task"].lower()
    agent = "strategist" if "strategy" in task or "estrategia" in task else "analyst"
    return {
        "agent": agent,
        "result": f"[{agent}] Procesado: {state['task']}",
        "trace": [f"[router_business] → {agent}"],
    }

builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("router_technical", router_technical)
builder.add_node("router_business", router_business)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_domain, {
    "router_technical": "router_technical",
    "router_business": "router_business",
})
builder.add_edge("router_technical", END)
builder.add_edge("router_business", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

tasks = ["Fix the bug in auth API", "Define Q2 strategy", "Deploy new version"]
for i, task in enumerate(tasks):
    config = {"configurable": {"thread_id": f"ex1-{i}"}}
    r = graph.invoke({"task": task, "domain": "", "agent": "", "result": "", "trace": []}, config)
    print(f"{task}:")
    for s in r["trace"]:
        print(f"  {s}")
    print(f"  Resultado: {r['result']}\n")
# Output esperado:
# Fix the bug in auth API:
#   [supervisor] → technical
#   [router_technical] → debugger
#   Resultado: [debugger] Procesado: Fix the bug in auth API
#
# Define Q2 strategy:
#   [supervisor] → business
#   [router_business] → strategist
#   Resultado: [strategist] Procesado: Define Q2 strategy
#
# Deploy new version:
#   [supervisor] → technical
#   [router_technical] → deployer
#   Resultado: [deployer] Procesado: Deploy new version

Ejercicio 2: Fan-out con 4 agentes y merge (Medio)

Crea un grafo con 4 agentes que se ejecutan en paralelo desde un nodo distributor. Cada agente simula una API diferente con diferentes tiempos de respuesta (sleep 0.1s, 0.2s, 0.05s, 0.15s). El nodo merge debe consolidar los 4 resultados y reportar cuál fue el más lento.

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    query: str
    results: Annotated[list[dict], operator.add]
    slowest: str
    total_ms: float

def distribute(state: State) -> dict:
    return {}

def make_agent(name: str, delay: float):
    def agent_fn(state: State) -> dict:
        start = time.time()
        time.sleep(delay)
        ms = (time.time() - start) * 1000
        return {
            "results": [{"agent": name, "data": f"{name}: datos sobre {state['query']}", "ms": round(ms)}],
        }
    return agent_fn

def merge(state: State) -> dict:
    slowest = max(state["results"], key=lambda r: r["ms"])
    total = max(r["ms"] for r in state["results"])
    return {"slowest": slowest["agent"], "total_ms": total}

builder = StateGraph(State)
builder.add_node("distribute", distribute)
builder.add_node("api_a", make_agent("api_a", 0.1))
builder.add_node("api_b", make_agent("api_b", 0.2))
builder.add_node("api_c", make_agent("api_c", 0.05))
builder.add_node("api_d", make_agent("api_d", 0.15))
builder.add_node("merge", merge)

builder.add_edge(START, "distribute")
for api in ["api_a", "api_b", "api_c", "api_d"]:
    builder.add_edge("distribute", api)
    builder.add_edge(api, "merge")
builder.add_edge("merge", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "fanout-4"}}
result = graph.invoke({"query": "LLM benchmarks", "results": [], "slowest": "", "total_ms": 0}, config)

for r in result["results"]:
    print(f"  {r['agent']}: {r['ms']}ms")
print(f"\n  Más lento: {result['slowest']} ({result['total_ms']}ms)")
print(f"  (secuencial sería ~500ms, paralelo ~200ms)")
# Output esperado:
#   api_a: 100ms
#   api_b: 200ms
#   api_c: 50ms
#   api_d: 150ms
#
#   Más lento: api_b (200ms)
#   (secuencial sería ~500ms, paralelo ~200ms)

Ejercicio 3: Consenso con desempate (Medio)

Implementa un sistema de consenso con 4 agentes analistas. Dos dicen "bullish" y dos dicen "bearish" (empate). Implementa una estrategia de desempate basada en la confianza promedio de cada grupo.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    opinions: Annotated[list[dict], operator.add]
    consensus: str
    tiebreaker_used: bool

def analyst_1(state: State) -> dict:
    return {"opinions": [{"agent": "A1", "conclusion": "bullish", "confidence": 0.90}]}

def analyst_2(state: State) -> dict:
    return {"opinions": [{"agent": "A2", "conclusion": "bearish", "confidence": 0.85}]}

def analyst_3(state: State) -> dict:
    return {"opinions": [{"agent": "A3", "conclusion": "bullish", "confidence": 0.70}]}

def analyst_4(state: State) -> dict:
    return {"opinions": [{"agent": "A4", "conclusion": "bearish", "confidence": 0.95}]}

def consensus_with_tiebreak(state: State) -> dict:
    from collections import Counter
    votes = Counter(op["conclusion"] for op in state["opinions"])
    top_two = votes.most_common(2)

    if len(top_two) < 2 or top_two[0][1] > top_two[1][1]:
        return {"consensus": top_two[0][0], "tiebreaker_used": False}

    tied = [t[0] for t in top_two]
    avg_conf = {}
    for conclusion in tied:
        confs = [op["confidence"] for op in state["opinions"] if op["conclusion"] == conclusion]
        avg_conf[conclusion] = sum(confs) / len(confs)

    winner = max(avg_conf, key=avg_conf.get)
    print(f"  Empate {votes.most_common()} → Desempate por confianza: {avg_conf}")
    return {"consensus": winner, "tiebreaker_used": True}

builder = StateGraph(State)
for name, fn in [("a1", analyst_1), ("a2", analyst_2), ("a3", analyst_3), ("a4", analyst_4)]:
    builder.add_node(name, fn)
    builder.add_edge(START, name)
builder.add_node("consensus", consensus_with_tiebreak)
for name in ["a1", "a2", "a3", "a4"]:
    builder.add_edge(name, "consensus")
builder.add_edge("consensus", END)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "tie-001"}}
result = graph.invoke({"opinions": [], "consensus": "", "tiebreaker_used": False}, config)

for op in result["opinions"]:
    print(f"  {op['agent']}: {op['conclusion']} ({op['confidence']:.0%})")
print(f"\n  Consenso: {result['consensus']} (tiebreaker: {result['tiebreaker_used']})")
# Output esperado:
#   A1: bullish (90%)
#   A2: bearish (85%)
#   A3: bullish (70%)
#   A4: bearish (95%)
#   Empate [('bullish', 2), ('bearish', 2)] → Desempate por confianza: {'bullish': 0.8, 'bearish': 0.9}
#
#   Consenso: bearish (tiebreaker: True)

Ejercicio 4: FlowTracer con detección de anomalías (Medio)

Extiende el FlowTracer con un método detect_anomalies() que identifique: (a) agentes que tardaron más del doble del promedio, (b) handoffs donde el tiempo entre agentes fue >100ms, (c) agentes que se ejecutaron más de una vez (posible loop).

from dotenv import load_dotenv
load_dotenv()

import time

class FlowTracer:
    def __init__(self):
        self.events: list[dict] = []
        self.start_time = time.time()

    def record(self, agent: str, event: str, data: dict | None = None):
        elapsed = (time.time() - self.start_time) * 1000
        self.events.append({"agent": agent, "event": event, "elapsed_ms": round(elapsed), "data": data or {}})

    def detect_anomalies(self) -> list[str]:
        anomalies = []

        agent_durations: dict[str, float] = {}
        agent_starts: dict[str, float] = {}
        agent_counts: dict[str, int] = {}

        for e in self.events:
            agent = e["agent"]
            agent_counts[agent] = agent_counts.get(agent, 0) + 1

            if e["event"] == "started":
                agent_starts[agent] = e["elapsed_ms"]
            elif e["event"] == "completed" and agent in agent_starts:
                agent_durations[agent] = e["elapsed_ms"] - agent_starts[agent]

        if agent_durations:
            avg = sum(agent_durations.values()) / len(agent_durations)
            for agent, dur in agent_durations.items():
                if dur > avg * 2:
                    anomalies.append(f"SLOW: {agent} tardó {dur:.0f}ms (promedio: {avg:.0f}ms)")

        handoff_events = [e for e in self.events if e["event"] == "handoff"]
        for he in handoff_events:
            target = he["data"].get("to", "")
            target_start = next(
                (e["elapsed_ms"] for e in self.events if e["agent"] == target and e["event"] == "started"),
                None,
            )
            if target_start and (target_start - he["elapsed_ms"]) > 100:
                gap = target_start - he["elapsed_ms"]
                anomalies.append(f"GAP: {he['agent']} → {target} tardó {gap:.0f}ms")

        for agent, count in agent_counts.items():
            if count > 3:
                anomalies.append(f"LOOP: {agent} se ejecutó {count} veces")

        return anomalies

tracer = FlowTracer()

tracer.record("supervisor", "started")
tracer.record("supervisor", "completed")
tracer.record("supervisor", "handoff", {"to": "researcher"})

time.sleep(0.05)
tracer.record("researcher", "started")
time.sleep(0.3)
tracer.record("researcher", "completed")
tracer.record("researcher", "handoff", {"to": "analyst"})

time.sleep(0.15)
tracer.record("analyst", "started")
time.sleep(0.05)
tracer.record("analyst", "completed")

for _ in range(5):
    tracer.record("retry_agent", "attempt")

anomalies = tracer.detect_anomalies()
print("=== Anomalías detectadas ===")
for a in anomalies:
    print(f"  ⚠️ {a}")
# Output esperado:
# === Anomalías detectadas ===
#   ⚠️ SLOW: researcher tardó 300ms (promedio: 120ms)
#   ⚠️ GAP: researcher → analyst tardó 150ms
#   ⚠️ LOOP: retry_agent se ejecutó 5 veces

Ejercicio 5: HITL centralizado con threshold de costo (Avanzado)

Crea un sistema multi-agente donde el supervisor genera un plan con 3 agentes y un costo estimado. Si el costo total es <$1, ejecuta automáticamente. Si es >=$1, pide aprobación humana centralizada. Prueba con un plan barato y uno caro.

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    plan_cost: float
    approved: bool
    approval_source: str
    results: Annotated[list[str], operator.add]
    trace: Annotated[list[str], operator.add]

def plan(state: State) -> dict:
    cost = 0.10 if "simple" in state["task"].lower() else 5.00
    return {
        "plan_cost": cost,
        "trace": [f"[supervisor] Plan: costo ${cost:.2f}"],
    }

def gate(state: State) -> dict:
    if state["plan_cost"] < 1.0:
        return {
            "approved": True,
            "approval_source": "auto",
            "trace": [f"[gate] Auto-aprobado (${state['plan_cost']:.2f} < $1)"],
        }

    response = interrupt({
        "type": "cost_gate",
        "cost": state["plan_cost"],
        "message": f"Plan cuesta ${state['plan_cost']:.2f}. ¿Aprobar?",
    })
    approved = response in ("approve", "yes", True)
    return {
        "approved": approved,
        "approval_source": "human",
        "trace": [f"[gate] Humano {'aprobó' if approved else 'rechazó'} (${state['plan_cost']:.2f})"],
    }

def route_gate(state: State) -> str:
    return "execute" if state["approved"] else "end"

def execute(state: State) -> dict:
    return {
        "results": [f"Agente A completó", "Agente B completó", "Agente C completó"],
        "trace": [f"[execute] 3 agentes completaron su trabajo"],
    }

builder = StateGraph(State)
builder.add_node("plan", plan)
builder.add_node("gate", gate)
builder.add_node("execute", execute)

builder.add_edge(START, "plan")
builder.add_edge("plan", "gate")
builder.add_conditional_edges("gate", route_gate, {"execute": "execute", "end": END})
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

print("=== Tarea barata (auto-approve) ===")
config1 = {"configurable": {"thread_id": "hitl-cost-1"}}
r1 = graph.invoke(
    {"task": "Simple search", "plan_cost": 0, "approved": False, "approval_source": "", "results": [], "trace": []},
    config1,
)
for s in r1["trace"]:
    print(f"  {s}")

print("\n=== Tarea cara (requiere humano) ===")
config2 = {"configurable": {"thread_id": "hitl-cost-2"}}
graph.invoke(
    {"task": "Deep analysis", "plan_cost": 0, "approved": False, "approval_source": "", "results": [], "trace": []},
    config2,
)
state = graph.get_state(config2)
print(f"  Esperando aprobación... (next: {state.next})")
r2 = graph.invoke(Command(resume="approve"), config2)
for s in r2["trace"]:
    print(f"  {s}")
# Output esperado:
# === Tarea barata (auto-approve) ===
#   [supervisor] Plan: costo $0.10
#   [gate] Auto-aprobado ($0.10 < $1)
#   [execute] 3 agentes completaron su trabajo
#
# === Tarea cara (requiere humano) ===
#   Esperando aprobación... (next: ('gate',))
#   [supervisor] Plan: costo $5.00
#   [gate] Humano aprobó ($5.00)
#   [execute] 3 agentes completaron su trabajo

Ejercicio 6: Sistema jerárquico completo con logging (Avanzado)

Construye un sistema jerárquico con un top supervisor, 2 department supervisors (research y analysis), y 2 workers por departamento. Cada nodo debe loguear usando AgentLogger. Incluye un FlowTracer que registre todo el flujo y al final imprima la timeline completa y el cuello de botella.

from dotenv import load_dotenv
load_dotenv()

import time
import operator
import logging
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class AgentLogger:
    def __init__(self, name: str):
        self.name = name
        self.logger = logging.getLogger(f"agent.{name}")
        if not self.logger.handlers:
            h = logging.StreamHandler()
            h.setFormatter(logging.Formatter(f"%(asctime)s | [{name}] %(message)s", datefmt="%H:%M:%S"))
            self.logger.addHandler(h)
            self.logger.setLevel(logging.INFO)

    def info(self, msg: str):
        self.logger.info(msg)

class FlowTracer:
    def __init__(self):
        self.events: list[dict] = []
        self.t0 = time.time()

    def record(self, agent: str, event: str):
        self.events.append({"agent": agent, "event": event, "ms": round((time.time() - self.t0) * 1000)})

    def timeline(self):
        for e in self.events:
            print(f"  {e['ms']:>5}ms | {e['agent']:<20} | {e['event']}")

    def bottleneck(self):
        starts, ends = {}, {}
        for e in self.events:
            if "start" in e["event"]:
                starts[e["agent"]] = e["ms"]
            elif "done" in e["event"]:
                ends[e["agent"]] = e["ms"]
        durs = {a: ends[a] - starts[a] for a in starts if a in ends}
        if durs:
            slow = max(durs, key=durs.get)
            print(f"  Cuello de botella: {slow} ({durs[slow]}ms)")

tracer = FlowTracer()

class State(TypedDict):
    query: str
    research_data: Annotated[list[str], operator.add]
    analysis_data: Annotated[list[str], operator.add]
    output: str

def make_node(name: str, delay: float, field: str, value_fn):
    logger = AgentLogger(name)
    def node_fn(state: State) -> dict:
        tracer.record(name, "start")
        logger.info(f"Procesando: {state['query']}")
        time.sleep(delay)
        val = value_fn(state)
        logger.info(f"Completado: {val}")
        tracer.record(name, "done")
        return {field: [val]} if field in ("research_data", "analysis_data") else {field: val}
    return node_fn

builder = StateGraph(State)

builder.add_node("top_sup", make_node("top_sup", 0.01, "output", lambda s: ""))
builder.add_node("res_sup", make_node("res_sup", 0.01, "output", lambda s: ""))
builder.add_node("res_w1", make_node("res_w1", 0.1, "research_data", lambda s: f"web: {s['query']}"))
builder.add_node("res_w2", make_node("res_w2", 0.08, "research_data", lambda s: f"arxiv: {s['query']}"))
builder.add_node("ana_sup", make_node("ana_sup", 0.01, "output", lambda s: ""))
builder.add_node("ana_w1", make_node("ana_w1", 0.15, "analysis_data", lambda s: f"patterns in {len(s['research_data'])} sources"))
builder.add_node("ana_w2", make_node("ana_w2", 0.12, "analysis_data", lambda s: f"factcheck: OK"))
builder.add_node("final", make_node("final", 0.01, "output",
    lambda s: f"Research: {s['research_data']} | Analysis: {s['analysis_data']}"))

builder.add_edge(START, "top_sup")
builder.add_edge("top_sup", "res_sup")
builder.add_edge("res_sup", "res_w1")
builder.add_edge("res_sup", "res_w2")
builder.add_edge("res_w1", "ana_sup")
builder.add_edge("res_w2", "ana_sup")
builder.add_edge("ana_sup", "ana_w1")
builder.add_edge("ana_sup", "ana_w2")
builder.add_edge("ana_w1", "final")
builder.add_edge("ana_w2", "final")
builder.add_edge("final", END)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "hier-full"}}

result = graph.invoke(
    {"query": "AI agents", "research_data": [], "analysis_data": [], "output": ""},
    config,
)

print("\n=== Timeline ===")
tracer.timeline()
tracer.bottleneck()
# Output esperado:
# (logs con timestamps de AgentLogger)
#
# === Timeline ===
#      0ms | top_sup              | start
#     10ms | top_sup              | done
#     10ms | res_sup              | start
#     20ms | res_sup              | done
#     20ms | res_w1               | start
#     20ms | res_w2               | start
#    120ms | res_w1               | done
#    100ms | res_w2               | done
#    120ms | ana_sup              | start
#    130ms | ana_sup              | done
#    130ms | ana_w1               | start
#    130ms | ana_w2               | start
#    280ms | ana_w1               | done
#    250ms | ana_w2               | done
#    280ms | final                | start
#    290ms | final                | done
#   Cuello de botella: ana_w1 (150ms)

Resumen

En esta cápsula aprendiste:

• Supervisor + Router combina coordinación de alto nivel con routing especializado por dominio — el patrón más común en producción
• Handoffs + Subagents permite cadenas secuenciales donde cada agente tiene workers internos que descomponen su subtarea
• Jerarquías de agentes escalan cuando tienes >6 agentes con dominios distintos — un supervisor de supervisores con workers especializados por departamento
• Ejecución paralela (fan-out / fan-in) ejecuta agentes independientes simultáneamente — el tiempo total es el del agente más lento, no la suma de todos
• Consenso entre agentes — votación para decisiones simples, ponderación por confianza cuando la certeza de cada agente varía
• HITL en multi-agente tiene 4 estrategias: pre-delegación, post-agente, solo al final, y centralizado. La centralizada es la más limpia para el usuario
• Debugging multi-agente requiere tres herramientas: logging por agente con prefijos, tracing del flujo con timeline, y visualización del grafo con draw_mermaid()
• Los failure modes comunes son loops de delegación, corrupción de estado compartido, mensajes perdidos entre agentes, y supervisores ciegos al resultado de sus workers

Próxima cápsula: todo lo que aprendiste se integra en el proyecto del módulo — el Research Assistant se transforma de un solo agente (v4) a un sistema multi-agente con researcher, analyst, writer y supervisor (v5).

Recursos adicionales

1. LangGraph Multi-Agent — Conceptos oficiales de multi-agente en LangGraph
2. LangGraph Supervisor — Implementación del patrón supervisor
3. LangGraph Handoffs — Delegación entre agentes
4. LangGraph Subgraphs — Subgrafos como agentes internos
5. Multi-Agent Architectures (LangChain Blog) — Patrones de arquitectura multi-agente
6. LangGraph draw_mermaid — Visualización de grafos para debugging

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