Advanced Orchestration

Capsule overview

Real systems rarely use a single pattern. An e-commerce platform doesn't have "a supervisor" or "a router" — it has a supervisor coordinating departments, each department has its own internal router, and some agents communicate through handoffs while others work as isolated subagents. Real complexity comes from combining patterns, not from mastering just one.

In the previous capsules you learned the 4 fundamental patterns: Supervisor, Handoffs, Subagents and Router. You also understand shared vs isolated state and how each pattern handles it. Now it's time to compose. This capsule shows you the 4 advanced orchestration techniques that show up in production multi-agent systems: combining patterns, agent hierarchies, parallel execution, and consensus between agents.

It also covers the two skills that separate a prototype from a real system: HITL in multi-agent contexts (where do you put the human approval when there are 6 agents?) and multi-agent debugging (how do you know which agent caused the problem?).

Combining patterns: Supervisor + Router

The most common pattern in production is a high-level supervisor that coordinates the overall flow, with specialized routers inside specific domains. The supervisor decides "this task belongs to analysis" and the analysis router decides "this type of analysis is handled by the financial agent."

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:
    """High-level supervisor: decides the domain."""
    task = state["task"].lower()

    if any(w in task for w in ["search", "research", "find", "look up"]):
        domain = "research"
    elif any(w in task for w in ["analyze", "compare", "evaluate", "assess"]):
        domain = "analysis"
    elif any(w in task for w in ["write", "draft", "generate report", "compose"]):
        domain = "writing"
    else:
        domain = "research"

    return {
        "domain": domain,
        "trace": [f"[supervisor] Task assigned to domain: {domain}"],
    }

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

def router_research(state: OrchestratorState) -> dict:
    """Internal router for the research domain."""
    task = state["task"].lower()

    if "paper" in task or "arxiv" in task or "academic" in task:
        sub = "academic_search"
    elif "news" in task or "headline" in task or "recent" in task:
        sub = "news_search"
    else:
        sub = "web_search"

    return {
        "sub_domain": sub,
        "result": f"[{sub}] Results for: {state['task']}",
        "trace": [f"[router_research] Sub-domain: {sub}"],
    }

def router_analysis(state: OrchestratorState) -> dict:
    """Internal router for the analysis domain."""
    task = state["task"].lower()

    if "financial" in task or "price" in task or "cost" in task:
        sub = "financial_analyst"
    elif "compare" in task or "versus" in task or "vs" in task:
        sub = "comparison_analyst"
    else:
        sub = "general_analyst"

    return {
        "sub_domain": sub,
        "result": f"[{sub}] Analysis of: {state['task']}",
        "trace": [f"[router_analysis] Sub-domain: {sub}"],
    }

def router_writing(state: OrchestratorState) -> dict:
    """Internal router for the writing domain."""
    task = state["task"].lower()

    if "executive" in task or "summary" in task:
        sub = "executive_writer"
    elif "technical" in task or "detailed" in task:
        sub = "technical_writer"
    else:
        sub = "general_writer"

    return {
        "sub_domain": sub,
        "result": f"[{sub}] Document about: {state['task']}",
        "trace": [f"[router_writing] Sub-domain: {sub}"],
    }

def output_node(state: OrchestratorState) -> dict:
    return {
        "trace": [f"[output] Final result from domain '{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 = [
    "Search academic papers about transformers",
    "Analyze the financial costs of the project",
    "Write the Q1 executive summary",
]

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"\nTask: {task}")
    for step in result["trace"]:
        print(f"  {step}")
    print(f"  Result: {result['result']}")
# Expected output:
# Task: Search academic papers about transformers
#   [supervisor] Task assigned to domain: research
#   [router_research] Sub-domain: academic_search
#   [output] Final result from domain 'research' / 'academic_search'
#   Result: [academic_search] Results for: Search academic papers about transformers
#
# Task: Analyze the financial costs of the project
#   [supervisor] Task assigned to domain: analysis
#   [router_analysis] Sub-domain: financial_analyst
#   [output] Final result from domain 'analysis' / 'financial_analyst'
#   Result: [financial_analyst] Analysis of: Analyze the financial costs of the project
#
# Task: Write the Q1 executive summary
#   [supervisor] Task assigned to domain: writing
#   [router_writing] Sub-domain: executive_writer
#   [output] Final result from domain 'writing' / 'executive_writer'
#   Result: [executive_writer] Document about: Write the Q1 executive summary

The flow has two decision levels: the supervisor picks the domain (research/analysis/writing) and the internal router picks the specialized agent (academic_search/financial_analyst/executive_writer). The trace shows you the exact chain of decisions.

Combining patterns: Handoffs + Subagents

Another powerful pattern: a chain of sequential handoffs where each agent uses subagents for its internal subtasks. The main agent receives the work, delegates sub-tasks to its workers, consolidates, and passes the result to the next agent in the chain.

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:
    """The researcher uses internal 'subagents' to search multiple sources."""
    topic = state["topic"]

    web_result = f"[web] 3 articles about {topic}"
    arxiv_result = f"[arxiv] 2 relevant papers about {topic}"
    news_result = f"[news] 1 recent news item about {topic}"

    combined = [web_result, arxiv_result, news_result]

    return {
        "raw_findings": combined,
        "trace": [
            f"[researcher] Start — topic: {topic}",
            f"[researcher:web_worker] {web_result}",
            f"[researcher:arxiv_worker] {arxiv_result}",
            f"[researcher:news_worker] {news_result}",
            f"[researcher] Handoff → analyst with {len(combined)} findings",
        ],
    }

def analyst_agent(state: PipelineState) -> dict:
    """The analyst uses subagents for different types of analysis."""
    findings_count = len(state["raw_findings"])

    pattern_analysis = f"Pattern identified: growing trend in {state['topic']}"
    contradiction_check = "No contradictions between sources"

    analysis = f"{pattern_analysis}. {contradiction_check}. Based on {findings_count} sources."

    return {
        "analysis": analysis,
        "trace": [
            f"[analyst] Received: {findings_count} findings",
            f"[analyst:pattern_worker] {pattern_analysis}",
            f"[analyst:contradiction_worker] {contradiction_check}",
            f"[analyst] Handoff → writer with the full analysis",
        ],
    }

def writer_agent(state: PipelineState) -> dict:
    """The writer generates the final report."""
    report = (
        f"# Report: {state['topic']}\n\n"
        f"## Analysis\n{state['analysis']}\n\n"
        f"## Sources\n" + "\n".join(f"- {f}" for f in state["raw_findings"])
    )

    return {
        "final_report": report,
        "trace": [
            f"[writer] Received: analysis of {len(state['analysis'])} chars",
            f"[writer] Report generated: {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("=== Full trace ===")
for step in result["trace"]:
    print(f"  {step}")
print(f"\n=== Report ===\n{result['final_report']}")
# Expected output:
# === Full trace ===
#   [researcher] Start — topic: AI multi-agent systems
#   [researcher:web_worker] [web] 3 articles about AI multi-agent systems
#   [researcher:arxiv_worker] [arxiv] 2 relevant papers about AI multi-agent systems
#   [researcher:news_worker] [news] 1 recent news item about AI multi-agent systems
#   [researcher] Handoff → analyst with 3 findings
#   [analyst] Received: 3 findings
#   [analyst:pattern_worker] Pattern identified: growing trend in AI multi-agent systems
#   [analyst:contradiction_worker] No contradictions between sources
#   [analyst] Handoff → writer with the full analysis
#   [writer] Received: analysis of 97 chars
#   [writer] Report generated: 236 chars
#
# === Report ===
# # Report: AI multi-agent systems
#
# ## Analysis
# Pattern identified: growing trend in AI multi-agent systems. No contradictions between sources. Based on 3 sources.
#
# ## Sources
# - [web] 3 articles about AI multi-agent systems
# - [arxiv] 2 relevant papers about AI multi-agent systems
# - [news] 1 recent news item about AI multi-agent systems

The trace shows both levels: the handoffs between main agents (researcher → analyst → writer) and each agent's internal workers (researcher:web_worker, analyst:pattern_worker). Each agent is autonomous in how it organizes its internal work.

Hierarchical agents: supervisors of supervisors

When the system grows past 6 agents, a single supervisor becomes a bottleneck. The solution: hierarchy. A high-level supervisor delegates to department supervisors, which in turn coordinate specialized workers.

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

When to add hierarchy:

• ✅ >6 agents in the system
• ✅ Clearly distinct domains (research ≠ analysis ≠ writing)
• ✅ Different approval levels per department
• ✅ You need to scale one department without touching the others
• ❌ <4 agents — a flat supervisor is simpler
• ❌ All the agents do similar tasks — there are no clear domains

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] Task received: '{state['query']}' → delegating to research_supervisor"],
    }

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

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

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

def news_worker(state: HierarchyState) -> dict:
    results = [f"News item about {state['query']}"]
    return {
        "news_results": results,
        "trace": [f"[research:news_worker] {len(results)} news items found"],
    }

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)} total findings → delegating to analysis_supervisor"],
    }

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

def pattern_worker(state: HierarchyState) -> dict:
    patterns = [f"Pattern: consensus on {state['query']}", "Pattern: growth in adoption"]
    return {
        "patterns": patterns,
        "trace": [f"[analysis:pattern_worker] {len(patterns)} patterns identified"],
    }

def factcheck_worker(state: HierarchyState) -> dict:
    return {
        "fact_check": "No contradictions detected across the 6 sources",
        "trace": [f"[analysis:factcheck_worker] Verification complete"],
    }

def analysis_merge(state: HierarchyState) -> dict:
    output = (
        f"Findings: {len(state['all_findings'])} | "
        f"Patterns: {len(state['patterns'])} | "
        f"Fact-check: {state['fact_check']}"
    )
    return {
        "final_output": output,
        "trace": [f"[top_supervisor] Final result consolidated"],
    }

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("=== Hierarchical flow ===")
for step in result["trace"]:
    print(f"  {step}")
print(f"\n  Final: {result['final_output']}")
# Expected output:
# === Hierarchical flow ===
#   [top_supervisor] Task received: 'multi-agent AI systems' → delegating to research_supervisor
#   [research_supervisor] Coordinating 3 workers for: 'multi-agent AI systems'
#   [research:web_worker] 3 results found
#   [research:academic_worker] 2 papers found
#   [research:news_worker] 1 news items found
#   [research_supervisor] Merge: 6 total findings → delegating to analysis_supervisor
#   [analysis_supervisor] Analyzing 6 findings
#   [analysis:pattern_worker] 2 patterns identified
#   [analysis:factcheck_worker] Verification complete
#   [top_supervisor] Final result consolidated
#
#   Final: Findings: 6 | Patterns: 2 | Fact-check: No contradictions detected across the 6 sources

Hierarchy brings clarity: each supervisor knows exactly which workers it coordinates, and each worker has an isolated responsibility. If you need to add a new worker to the research team (say, patent_worker), you only touch that subgraph — the analysis_supervisor never even finds out.

Parallel execution: fan-out / fan-in

When agents work on independent subtasks, there's no reason to wait sequentially. Fan-out sends the task to several agents simultaneously, fan-in collects the results and merges them.

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] Sending '{state['query']}' to 3 agents in parallel"],
    }

def researcher(state: ParallelState) -> dict:
    start = time.time()
    time.sleep(0.1)
    elapsed = (time.time() - start) * 1000
    output = f"5 sources found about '{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"Upward trend identified in '{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 sources verified as reliable"
    return {
        "factchecker_output": output,
        "timing": [f"[factchecker] {elapsed:.0f}ms — {output}"],
    }

def merge(state: ParallelState) -> dict:
    merged = (
        f"Research: {state['researcher_output']} | "
        f"Analysis: {state['analyst_output']} | "
        f"Verification: {state['factchecker_output']}"
    )
    return {
        "merged_result": merged,
        "timing": [f"[merge] Results from 3 agents combined ({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("=== Parallel execution ===")
for step in result["timing"]:
    print(f"  {step}")
print(f"\n  Total time: {overall_ms:.0f}ms")
print(f"  (sequential would be ~330ms, parallel is ~150ms)")
# Expected output:
# === Parallel execution ===
#   [distribute] Sending 'LLM agent architectures' to 3 agents in parallel
#   [researcher] 100ms — 5 sources found about 'LLM agent architectures'
#   [analyst] 150ms — Upward trend identified in 'LLM agent architectures'
#   [factchecker] 80ms — 4/5 sources verified as reliable
#   [merge] Results from 3 agents combined (165 chars)
#
#   Total time: ~180ms
#   (sequential would be ~330ms, parallel is ~150ms)

LangGraph runs the researcher, analyst and factchecker nodes in parallel because all three share the same dependency (distribute) and don't depend on each other. The merge node waits for all three to finish (fan-in). You don't need manual threads — the graph handles it.

Consensus between agents

When precision is critical, you can have several agents analyze the same data and compare conclusions. Two strategies: voting (majority wins) and confidence weighting.

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] Conclusion: bullish (85% confidence)"],
    }

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

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

def voting_consensus(state: ConsensusState) -> dict:
    """Simple majority: the most-voted conclusion wins."""
    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] Votes: {dict(votes)}",
            f"[consensus:voting] Winner: {winner[0]} ({winner[1]}/{len(state['opinions'])} votes)",
        ],
    }

def weighted_consensus(state: ConsensusState) -> dict:
    """Weighting: the conclusion with the highest accumulated confidence wins."""
    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] Weighted scores: {scores}",
            f"[consensus:weighted] Winner: {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("=== Method 1: Voting ===")
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=== Method 2: Confidence weighting ===")
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}")
# Expected output:
# === Method 1: Voting ===
#   [analyst_a] Conclusion: bullish (85% confidence)
#   [analyst_b] Conclusion: bullish (72% confidence)
#   [analyst_c] Conclusion: bearish (68% confidence)
#   [consensus:voting] Votes: {'bullish': 2, 'bearish': 1}
#   [consensus:voting] Winner: bullish (2/3 votes)
#
# === Method 2: Confidence weighting ===
#   [analyst_a] Conclusion: bullish (85% confidence)
#   [analyst_b] Conclusion: bullish (72% confidence)
#   [analyst_c] Conclusion: bearish (68% confidence)
#   [consensus:weighted] Weighted scores: {'bullish': 1.57, 'bearish': 0.68}
#   [consensus:weighted] Winner: bullish (score: 1.57)

In this case both methods agree, but that isn't always true. If analyst_a had 0.51 confidence and analyst_c had 0.95, voting would still say "bullish" (2 vs 1), but the weighting could favor "bearish" if the accumulated confidence justifies it.

HITL in multi-agent: where to put the human approval

With a single agent, the decision was simple: before or after this action? With several agents, you have 4 possible points:

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 generated: {plan['agents']} (${plan['estimated_cost']})"],
    }

def hitl_approve_plan(state: HITLMultiState) -> dict:
    """Centralized HITL: approval before running any agent."""
    plan = state["plan"]

    response = interrupt({
        "type": "plan_approval",
        "message": (
            f"The supervisor proposes:\n"
            f"  Agents: {plan['agents']}\n"
            f"  Estimated cost: ${plan['estimated_cost']}\n"
            f"  Estimated time: {plan['estimated_time']}\n"
            f"Approve the full plan?"
        ),
        "plan": plan,
    })

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

    if action == "cancel":
        return {"trace": [f"[hitl] Plan cancelled by the user"]}

    return {"trace": [f"[hitl] Plan approved — running the agents"]}

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

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

def analyst(state: HITLMultiState) -> dict:
    result = f"Full analysis of: {state['research_result']}"
    return {
        "analysis_result": result,
        "trace": [f"[analyst] {result}"],
    }

def compile_output(state: HITLMultiState) -> dict:
    output = f"Report: {state['analysis_result']}"
    return {
        "final_output": output,
        "trace": [f"[supervisor] Final output compiled"],
    }

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("=== Centralized HITL: approve the 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"  Waiting for approval. Next: {state.next}")

result = graph.invoke(Command(resume="approve"), config)
print(f"\n=== Full flow ===")
for step in result["trace"]:
    print(f"  {step}")
# Expected output:
# === Centralized HITL: approve the plan ===
#   Waiting for approval. Next: ('approve',)
#
# === Full flow ===
#   [supervisor] Plan generated: ['researcher', 'analyst'] ($2.5)
#   [hitl] Plan approved — running the agents
#   [researcher] 5 sources found about 'AI trends 2026'
#   [analyst] Full analysis of: 5 sources found about 'AI trends 2026'
#   [supervisor] Final output compiled

The choice between centralizing and distributing HITL depends on your context:

• ✅ Centralized when you want a single interruption and the human trusts the individual agents
• ✅ Distributed (post-agent) when each agent can fail in different ways and you need to validate every step
• ❌ Avoid HITL on every agent AND at the end — the user will end up approving 7 things for a single task

Multi-agent debugging: the critical skill

Debugging a multi-agent system is significantly harder than debugging a single agent. With one agent, the problem is "somewhere in the pipeline." With 4 agents, the problem could be in the agent, in the communication between agents, in the shared state, or in the supervisor that delegated badly.

Per-agent logging

The most important rule: every agent logs with its name as a prefix. Without that, a 200-line log is impossible to read.

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("Delegating task to researcher")
researcher_log.info("Searching across 3 sources")
researcher_log.debug("web_search: 200 OK (120ms)")
researcher_log.debug("arxiv_search: 200 OK (340ms)")
researcher_log.warning("news_search: timeout after 5s, retrying...")
researcher_log.debug("news_search: 200 OK (890ms, attempt 2)")
researcher_log.info("3/3 sources completed → handoff to analyst")

analyst_log.info("Received: 6 findings from researcher")
analyst_log.debug("Identifying patterns...")
analyst_log.info("2 patterns, 0 contradictions → handoff to writer")

writer_log.info("Generating report (format: bullet_points)")
writer_log.info("Report ready: 450 chars")

supervisor_log.info("Pipeline complete in 1.2s")
# Expected output:
# 14:30:01 | INFO  | [supervisor] Delegating task to researcher
# 14:30:01 | INFO  | [researcher] Searching across 3 sources
# 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 after 5s, retrying...
# 14:30:01 | DEBUG | [researcher] news_search: 200 OK (890ms, attempt 2)
# 14:30:01 | INFO  | [researcher] 3/3 sources completed → handoff to analyst
# 14:30:01 | INFO  | [analyst] Received: 6 findings from researcher
# 14:30:01 | DEBUG | [analyst] Identifying patterns...
# 14:30:01 | INFO  | [analyst] 2 patterns, 0 contradictions → handoff to writer
# 14:30:01 | INFO  | [writer] Generating report (format: bullet_points)
# 14:30:01 | INFO  | [writer] Report ready: 450 chars
# 14:30:01 | INFO  | [supervisor] Pipeline complete in 1.2s

With the [researcher], [analyst], [writer] prefixes, you can filter logs by agent: grep "[researcher]" logs.txt gives you only what the researcher did.

Tracing the flow: who handled what, and when

On top of individual logs, you need a high-level view of the flow between agents:

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⚠️ Bottleneck: {bottleneck['agent']} ({bottleneck['duration_ms']}ms)")
# Expected output:
# 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
#
# ⚠️ Bottleneck: analyst (150ms)

Graph visualization

draw_mermaid_png() generates an image of the full graph that you can share with your team:

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("=== Mermaid diagram ===")
print(mermaid_code)

# To generate a PNG image:
# png_data = graph.get_graph().draw_mermaid_png()
# with open("graph_diagram.png", "wb") as f:
#     f.write(png_data)
# print("Diagram saved to graph_diagram.png")
# Expected output:
# === Mermaid diagram ===
# %%{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__;

Common failure modes

Troubleshooting

Problem 1: "The parallel nodes aren't running in parallel"

Symptom: The total time is the sum of the individual times, not the maximum.

Cause: The edges don't come out of the same node. For LangGraph to run nodes in parallel, they have to share the same parent node.

Fix: Check that the edges go from the same node to the workers:

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

Problem 2: "Consensus always returns the same result"

Symptom: The consensus node always picks the first agent.

Cause: The opinions list uses Annotated[list, operator.add] but the agents aren't appending — they're overwriting.

Fix: Each agent must return a list with a single element so operator.add accumulates them:

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

Problem 3: "The supervisor delegates in an infinite loop"

Symptom: The graph never reaches END.

Cause: The supervisor's exit condition always chooses to delegate more work.

Fix: Add a counter to the state and an exit condition:

class State(TypedDict):
    iteration: int

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

Problem 4: "The HITL interrupt fires for every agent"

Symptom: The system asks for approval 5 times for a single task.

Cause: Every agent has its own interrupt() and all of them run.

Fix: Centralize HITL in the supervisor. Only the supervisor interrupts; the workers run without pausing:

def supervisor(state):
    response = interrupt({"type": "plan_approval", ...})
    # Only here, not in every worker

Problem 5: "I can't find which agent caused the error"

Symptom: The final output is wrong but you don't know which agent failed.

Cause: The agents don't log with a prefix and the shared state doesn't record who wrote what.

Fix: Use AgentLogger with a name and add source to every piece of data in the state:

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

Exercises

Exercise 1: Supervisor + Router across two domains (Easy)

Build a system where a supervisor classifies tasks as "technical" or "business", and each domain has an internal router that picks between 2 specialized agents. Test it with 3 different tasks.

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}] Processed: {state['task']}",
        "trace": [f"[router_technical] → {agent}"],
    }

def router_business(state: State) -> dict:
    task = state["task"].lower()
    agent = "strategist" if "strategy" in task or "roadmap" in task else "analyst"
    return {
        "agent": agent,
        "result": f"[{agent}] Processed: {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"  Result: {r['result']}\n")
# Expected output:
# Fix the bug in auth API:
#   [supervisor] → technical
#   [router_technical] → debugger
#   Result: [debugger] Processed: Fix the bug in auth API
#
# Define Q2 strategy:
#   [supervisor] → business
#   [router_business] → strategist
#   Result: [strategist] Processed: Define Q2 strategy
#
# Deploy new version:
#   [supervisor] → technical
#   [router_technical] → deployer
#   Result: [deployer] Processed: Deploy new version

Exercise 2: Fan-out with 4 agents and a merge (Medium)

Build a graph with 4 agents that run in parallel from a distributor node. Each agent simulates a different API with different response times (sleep 0.1s, 0.2s, 0.05s, 0.15s). The merge node must consolidate the 4 results and report which one was the slowest.

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}: data about {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  Slowest: {result['slowest']} ({result['total_ms']}ms)")
print(f"  (sequential would be ~500ms, parallel ~200ms)")
# Expected output:
#   api_a: 100ms
#   api_b: 200ms
#   api_c: 50ms
#   api_d: 150ms
#
#   Slowest: api_b (200ms)
#   (sequential would be ~500ms, parallel ~200ms)

Exercise 3: Consensus with a tiebreaker (Medium)

Implement a consensus system with 4 analyst agents. Two say "bullish" and two say "bearish" (a tie). Implement a tiebreaking strategy based on each group's average confidence.

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"  Tie {votes.most_common()} → Tiebreak by confidence: {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  Consensus: {result['consensus']} (tiebreaker: {result['tiebreaker_used']})")
# Expected output:
#   A1: bullish (90%)
#   A2: bearish (85%)
#   A3: bullish (70%)
#   A4: bearish (95%)
#   Tie [('bullish', 2), ('bearish', 2)] → Tiebreak by confidence: {'bullish': 0.8, 'bearish': 0.9}
#
#   Consensus: bearish (tiebreaker: True)

Exercise 4: FlowTracer with anomaly detection (Medium)

Extend FlowTracer with a detect_anomalies() method that identifies: (a) agents that took more than twice the average, (b) handoffs where the gap between agents was >100ms, (c) agents that ran more than once (a possible 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} took {dur:.0f}ms (average: {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} took {gap:.0f}ms")

        for agent, count in agent_counts.items():
            if count > 3:
                anomalies.append(f"LOOP: {agent} ran {count} times")

        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("=== Anomalies detected ===")
for a in anomalies:
    print(f"  ⚠️ {a}")
# Expected output:
# === Anomalies detected ===
#   ⚠️ SLOW: researcher took 300ms (average: 120ms)
#   ⚠️ GAP: researcher → analyst took 150ms
#   ⚠️ LOOP: retry_agent ran 5 times

Exercise 5: Centralized HITL with a cost threshold (Advanced)

Build a multi-agent system where the supervisor generates a plan with 3 agents and an estimated cost. If the total cost is <$1, it runs automatically. If it's >=$1, it asks for centralized human approval. Test it with a cheap plan and an expensive one.

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: cost ${cost:.2f}"],
    }

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

    response = interrupt({
        "type": "cost_gate",
        "cost": state["plan_cost"],
        "message": f"The plan costs ${state['plan_cost']:.2f}. Approve?",
    })
    approved = response in ("approve", "yes", True)
    return {
        "approved": approved,
        "approval_source": "human",
        "trace": [f"[gate] Human {'approved' if approved else 'rejected'} (${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"Agent A finished", "Agent B finished", "Agent C finished"],
        "trace": [f"[execute] 3 agents completed their work"],
    }

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("=== Cheap task (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=== Expensive task (needs a human) ===")
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"  Waiting for approval... (next: {state.next})")
r2 = graph.invoke(Command(resume="approve"), config2)
for s in r2["trace"]:
    print(f"  {s}")
# Expected output:
# === Cheap task (auto-approve) ===
#   [supervisor] Plan: cost $0.10
#   [gate] Auto-approved ($0.10 < $1)
#   [execute] 3 agents completed their work
#
# === Expensive task (needs a human) ===
#   Waiting for approval... (next: ('gate',))
#   [supervisor] Plan: cost $5.00
#   [gate] Human approved ($5.00)
#   [execute] 3 agents completed their work

Exercise 6: Full hierarchical system with logging (Advanced)

Build a hierarchical system with a top supervisor, 2 department supervisors (research and analysis), and 2 workers per department. Every node has to log using AgentLogger. Include a FlowTracer that records the whole flow and at the end prints the full timeline and the bottleneck.

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"  Bottleneck: {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"Processing: {state['query']}")
        time.sleep(delay)
        val = value_fn(state)
        logger.info(f"Completed: {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()
# Expected output:
# (logs with timestamps from 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
#   Bottleneck: ana_w1 (150ms)

Summary

In this capsule you learned:

• Supervisor + Router combines high-level coordination with specialized routing per domain — the most common pattern in production
• Handoffs + Subagents enables sequential chains where each agent has internal workers that decompose its subtask
• Agent hierarchies scale when you have >6 agents across distinct domains — a supervisor of supervisors with specialized workers per department
• Parallel execution (fan-out / fan-in) runs independent agents simultaneously — the total time is that of the slowest agent, not the sum of all of them
• Consensus between agents — voting for simple decisions, confidence weighting when each agent's certainty varies
• HITL in multi-agent has 4 strategies: pre-delegation, post-agent, only at the end, and centralized. The centralized one is the cleanest for the user
• Multi-agent debugging requires three tools: per-agent logging with prefixes, flow tracing with a timeline, and graph visualization with draw_mermaid()
• The common failure modes are delegation loops, shared-state corruption, lost messages between agents, and supervisors blind to their workers' results

Next capsule: everything you learned comes together in the module project — the Research Assistant goes from a single agent (v4) to a multi-agent system with researcher, analyst, writer and supervisor (v5).

Additional resources

1. LangGraph Multi-Agent — Official multi-agent concepts in LangGraph
2. LangGraph Supervisor — Implementing the supervisor pattern
3. LangGraph Handoffs — Delegation between agents
4. LangGraph Subgraphs — Subgraphs as internal agents
5. Multi-Agent Architectures (LangChain Blog) — Multi-agent architecture patterns
6. LangGraph draw_mermaid — Graph visualization for debugging

Module 10 — LangChain & LangGraph: From Chains to Agents