7. Advanced Orchestration

Overview

You already know the four fundamental multi-agent patterns: Supervisor, which coordinates workers; Handoffs, which transfer control directly; Subagents, which isolate context; and Router, which classifies and directs. In the previous capsule you designed shared vs isolated state and understood how information flows between agents. Each pattern solves one type of problem. But the reality of production is that no single pattern is enough. A real system needs supervisors that coordinate other supervisors, workers that execute in parallel, agents that produce contradictory answers and someone to resolve the conflict.

This capsule is where everything comes together. You're going to implement agent hierarchies, use LangGraph's Send API for real fan-out/fan-in, design consensus mechanisms, and face the failure modes that show up when the system grows: agent hangs, infinite supervisor loops, cascading context overflow.

Connection to the module: This is the last technical capsule before the final project. What you implement here — hierarchies, parallelism, consensus, failure handling — is exactly what you need for capsule 08.

Hierarchical Agents

The single-level problem

A Supervisor with 3 workers works well. A Supervisor with 12 workers doesn't. The LLM has to choose among too many options, the prompt becomes enormous, and routing quality degrades. The solution: hierarchy. An L1 Supervisor coordinates L2 Supervisors, and each L2 coordinates its own workers.

                        ┌─────────────────────┐
                        │   SUPERVISOR L1     │
                        │   (Coordinator)     │
                        └──────┬──────┬───────┘
                               │      │
                    ┌──────────┘      └──────────┐
                    ▼                             ▼
          ┌─────────────────┐           ┌─────────────────┐
          │  SUPERVISOR L2  │           │  SUPERVISOR L2  │
          │  (Research)     │           │  (Content)      │
          └──┬─────────┬────┘           └──┬─────────┬────┘
             │         │                   │         │
             ▼         ▼                   ▼         ▼
         ┌───────┐ ┌───────┐          ┌───────┐ ┌───────┐
         │ Web   │ │Papers │          │Writer │ │Editor │
         │Search │ │Search │          │       │ │       │
         └───────┘ └───────┘          └───────┘ └───────┘

Implementation: a supervisor of supervisors

Each L2 Supervisor is a compiled subgraph. The L1 invokes it as just another node:

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field

model = ChatOpenAI(model="gpt-4o-mini")

def make_worker(name: str, instruction: str):
    def node(state):
        response = model.invoke([SystemMessage(content=instruction)] + state["messages"])
        return {"messages": [HumanMessage(content=f"[{name}] {response.content}", name=name)]}
    return node

# --- Supervisor L2: Research Team ---
class L2State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_worker: str
    iteration_count: int

class L2Decision(BaseModel):
    next_worker: Literal["web_search", "papers_search", "FINISH"] = Field(description="Next")

l2_structured = model.with_structured_output(L2Decision)

def research_supervisor(state: L2State) -> dict:
    iteration = state.get("iteration_count", 0) + 1
    if iteration > 4:
        return {"next_worker": "FINISH", "iteration_count": iteration,
                "messages": [HumanMessage(content="[research_sup] Limit", name="research_sup")]}
    decision = l2_structured.invoke(
        [SystemMessage(content="You coordinate web_search and papers_search. FINISH if there's enough.")]
        + state["messages"])
    return {"next_worker": decision.next_worker, "iteration_count": iteration,
            "messages": [HumanMessage(
                content=f"[research_sup] → {decision.next_worker}", name="research_sup")]}

research_graph = StateGraph(L2State)
research_graph.add_node("supervisor", research_supervisor)
research_graph.add_node("web_search", make_worker("web_search", "Search for information on the web."))
research_graph.add_node("papers_search", make_worker("papers_search", "Search for academic papers."))
research_graph.add_edge(START, "supervisor")
research_graph.add_conditional_edges("supervisor",
    lambda s: "end" if s["next_worker"] == "FINISH" else s["next_worker"],
    {"web_search": "web_search", "papers_search": "papers_search", "end": END})
research_graph.add_edge("web_search", "supervisor")
research_graph.add_edge("papers_search", "supervisor")
research_team = research_graph.compile()

The L1 Supervisor uses research_team as if it were a worker:

class L1State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    next_team: str
    task_complete: bool

class L1Decision(BaseModel):
    next_team: Literal["research", "content", "FINISH"] = Field(description="Next team")

l1_structured = model.with_structured_output(L1Decision)

def l1_supervisor(state: L1State) -> dict:
    decision = l1_structured.invoke(
        [SystemMessage(content="You coordinate research and content. FINISH when everything is ready.")]
        + state["messages"])
    return {"next_team": decision.next_team,
            "task_complete": decision.next_team == "FINISH",
            "messages": [HumanMessage(content=f"[L1] → {decision.next_team}", name="L1")]}

def research_node(state: L1State) -> dict:
    result = research_team.invoke({
        "messages": state["messages"], "next_worker": "", "iteration_count": 0})
    return {"messages": [HumanMessage(
        content=f"[research_team] {result['messages'][-1].content}", name="research_team")]}

l1_graph = StateGraph(L1State)
l1_graph.add_node("supervisor", l1_supervisor)
l1_graph.add_node("research", research_node)
l1_graph.add_node("content", make_worker("content", "Write content based on the research."))
l1_graph.add_edge(START, "supervisor")
l1_graph.add_conditional_edges("supervisor",
    lambda s: "end" if s.get("task_complete") else s["next_team"],
    {"research": "research", "content": "content", "end": END})
l1_graph.add_edge("research", "supervisor")
l1_graph.add_edge("content", "supervisor")
full_system = l1_graph.compile()

When hierarchy is justified

Each level adds latency (one extra LLM call). If you have 3 workers and good routing, a single Supervisor is better.

Parallel Execution

The problem with sequential execution

In a classic Supervisor, the flow is sequential: Supervisor → Worker A → Supervisor → Worker B. If Worker A and Worker B are independent, you're making them wait for no reason.

SEQUENTIAL:     Supervisor → Researcher(3s) → Supervisor → Analyst(3s) → Writer(3s)  ≈15s
PARALLEL:       Supervisor → [Researcher(3s) || Analyst(3s)] → Writer(3s)              ≈9s

Fan-out with LangGraph's Send API

The Send API is LangGraph's native mechanism for parallel execution. Instead of returning a string with the next node, you return a list of Send objects — each one creates a parallel execution:

from langgraph.types import Send

class FanOutState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    tasks: list[str]
    results: Annotated[list[str], operator.add]

def planner(state: FanOutState) -> dict:
    response = model.invoke([
        SystemMessage(content="Break this down into independent sub-tasks. One per line. Maximum 3."),
        state["messages"][-1],
    ])
    tasks = [t.strip() for t in response.content.strip().split("\n") if t.strip()][:3]
    return {"tasks": tasks}

def fan_out(state: FanOutState) -> list[Send]:
    return [Send("worker", {"messages": [HumanMessage(content=task)], "results": []})
            for task in state["tasks"]]

def worker(state: FanOutState) -> dict:
    task = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="Execute this specific task. Be concise."),
        HumanMessage(content=task),
    ])
    return {"results": [f"[{task[:30]}] {response.content}"]}

def aggregator(state: FanOutState) -> dict:
    response = model.invoke([
        SystemMessage(content="Synthesize these results into a coherent answer."),
        HumanMessage(content="\n\n".join(state["results"])),
    ])
    return {"messages": [response]}

graph = StateGraph(FanOutState)
graph.add_node("planner", planner)
graph.add_node("worker", worker)
graph.add_node("aggregator", aggregator)
graph.add_edge(START, "planner")
graph.add_conditional_edges("planner", fan_out)
graph.add_edge("worker", "aggregator")
graph.add_edge("aggregator", END)
parallel_system = graph.compile()

Each Send creates an instance of the worker node with its own partial state. The workers execute in parallel. When they all finish, the results are aggregated with operator.add and the flow continues to the aggregator.

Send vs asyncio.gather

For compiled subgraphs you execute outside the graph, use asyncio.gather with return_exceptions=True:

import asyncio

async def parallel_research(state: dict) -> dict:
    results = await asyncio.gather(*[
        subagent.ainvoke({"messages": [HumanMessage(content=q)]})
        for q in state["sub_queries"]
    ], return_exceptions=True)
    return {"results": [
        r["messages"][-1].content if not isinstance(r, Exception) else f"[ERROR]: {r}"
        for r in results
    ]}

Consensus Between Agents

The problem: divergent answers

Two analysts evaluate whether a startup is a good investment. One says "buy" and the other says "avoid". Which one is right? Consensus is the mechanism for resolving that.

Three strategies

1. Voting (simple majority)

Useful when the answer is categorical (classification, yes/no):

def create_voter(perspective: str):
    def vote(state: dict) -> dict:
        response = model.invoke([
            SystemMessage(content=f"Perspective: {perspective}. "
                          "Classify: BUY, HOLD or SELL. Answer with the classification ONLY."),
            state["messages"][-1],
        ])
        return {"votes": [response.content.strip().upper()]}
    return vote

def tally_votes(state: dict) -> dict:
    from collections import Counter
    counts = Counter(state["votes"])
    winner = counts.most_common(1)[0]
    confidence = winner[1] / len(state["votes"])
    return {"messages": [HumanMessage(
        content=f"Result: {winner[0]} ({winner[1]}/{len(state['votes'])}, {confidence:.0%})")]}

2. Debate (agents argue)

The agents see each other's positions and can change their minds across multiple rounds:

def debate_round(state: dict) -> dict:
    round_num = state.get("round_number", 0) + 1
    positions = state.get("positions", [])
    positions_text = "\n".join([f"- {p['agent']}: {p['position']}" for p in positions]
                               ) if positions else "First round."

    new_positions = []
    for agent_id in ["alpha", "beta", "gamma"]:
        response = model.invoke([
            SystemMessage(content=f"You are {agent_id}. Round {round_num}.\n"
                          f"Current positions:\n{positions_text}\n"
                          "If someone convinced you, you may change. 2-3 sentences."),
            state["messages"][0],
        ])
        new_positions.append({"agent": agent_id, "position": response.content})
    return {"positions": new_positions, "round_number": round_num}

def check_convergence(state: dict) -> str:
    return "adjudicate" if state.get("round_number", 0) >= 3 else "debate"

3. Adjudication (external judge)

An independent agent that didn't participate acts as the judge:

def adjudicator(state: dict) -> dict:
    positions_text = "\n\n".join([
        f"**{p['agent']}:** {p['position']}" for p in state["positions"]])
    response = model.invoke([
        SystemMessage(content="Impartial judge. Pick the best answer or synthesize one."),
        HumanMessage(content=f"Question: {state['messages'][0].content}\n\n{positions_text}"),
    ])
    return {"messages": [response]}

When to use each one

Conflict Resolution

Consensus assumes the agents eventually converge. Conflict resolution handles genuinely contradictory results where there's no clear "correct" answer.

Strategy 1: the supervisor arbitrates

def resolve_conflict(state: dict) -> dict:
    results = state["worker_results"]
    response = model.invoke([
        SystemMessage(content="Contradictory results. If you can determine which one "
                      "is correct from the evidence, pick that one. If not, answer 'ESCALATE'."),
        HumanMessage(content=f"A: {results[0]}\n\nB: {results[1]}"),
    ])
    if "ESCALATE" in response.content:
        return {"needs_human": True, "conflict_summary": response.content}
    return {"resolved_result": response.content, "needs_human": False}

Strategy 2: re-execution with cross context

The conflicting agents see each other's result and re-execute:

def cross_retry(state: dict) -> dict:
    a, b = state["result_a"], state["result_b"]
    prompt = ("Your analysis differed from another agent's. "
              "Review it taking theirs into account. If there's an error in your analysis, correct it.")
    revised_a = model.invoke([SystemMessage(content=prompt),
        HumanMessage(content=f"Yours: {a}\nTheirs: {b}")]).content
    revised_b = model.invoke([SystemMessage(content=prompt),
        HumanMessage(content=f"Yours: {b}\nTheirs: {a}")]).content
    return {"revised_a": revised_a, "revised_b": revised_b}

Strategy 3: human escalation

from langgraph.types import interrupt

def escalate_to_human(state: dict) -> dict:
    human_decision = interrupt({
        "type": "conflict_resolution",
        "message": f"Conflict:\n{state['conflict_summary']}",
        "options": ["Use result A", "Use result B", "Discard both"],
    })
    return {"resolved_result": human_decision, "resolved_by": "human"}

Decision tree

Contradictory results?
├── No → Consensus (voting/adjudication)
└── Yes → Can it be settled with evidence?
    ├── Yes → The supervisor arbitrates
    └── No → Is re-execution acceptable?
        ├── Yes → Cross context
        └── No → Escalate to a human

Multi-Agent Failure Modes

1. Agent Hang

A worker with external tools ends up waiting indefinitely.

import asyncio

async def worker_with_timeout(agent, input_state: dict, timeout_s: int = 30):
    try:
        return {"success": True,
                "result": await asyncio.wait_for(agent.ainvoke(input_state), timeout=timeout_s)}
    except asyncio.TimeoutError:
        return {"success": False, "error": f"Timeout after {timeout_s}s"}

2. Supervisor Infinite Loop

The Supervisor sends to the Researcher, receives a result, and... sends to the Researcher again. Triple guard:

MAX_ITERATIONS = 8
MAX_SAME_CONSECUTIVE = 2

def supervisor_with_guards(state: dict) -> dict:
    iteration = state.get("iteration_count", 0) + 1
    called = state.get("workers_called", [])

    if iteration > MAX_ITERATIONS:
        return force_finish(state, "Iteration limit")

    if len(called) >= MAX_SAME_CONSECUTIVE:
        if len(set(called[-MAX_SAME_CONSECUTIVE:])) == 1:
            return force_finish(state, f"'{called[-1]}' repeated {MAX_SAME_CONSECUTIVE}x")

    recent = [m.content for m in state["messages"][-4:]]
    if len(recent) >= 4 and len(set(recent)) <= 2:
        return force_finish(state, "No new progress")

    return route_normally(state, iteration, called)

def force_finish(state, reason):
    return {"next_worker": "FINISH", "task_complete": True,
            "messages": [HumanMessage(content=f"[sup] FORCED: {reason}", name="sup")]}

3. Cascading Context Overflow

Each hierarchical level adds messages. Trim before passing down to the child:

def trim_for_child(messages: list, max_msgs: int = 10) -> list:
    if len(messages) <= max_msgs:
        return messages
    return [messages[0]] + messages[-(max_msgs - 1):]

4. Cascade Failure — Circuit Breaker

class CircuitBreaker:
    def __init__(self, max_failures: int = 3):
        self.failures: dict[str, int] = {}
        self.max_failures = max_failures

    def can_execute(self, name: str) -> bool:
        return self.failures.get(name, 0) < self.max_failures

    def record_failure(self, name: str):
        self.failures[name] = self.failures.get(name, 0) + 1

    def record_success(self, name: str):
        self.failures[name] = 0

circuit = CircuitBreaker()

def safe_delegate(worker_name: str, agent, state: dict) -> dict:
    if not circuit.can_execute(worker_name):
        return {"messages": [HumanMessage(
            content=f"[{worker_name}] CIRCUIT OPEN", name=worker_name)]}
    try:
        result = agent.invoke({"messages": state["messages"]})
        circuit.record_success(worker_name)
        return result
    except Exception as e:
        circuit.record_failure(worker_name)
        return {"messages": [HumanMessage(
            content=f"[{worker_name}] ERROR: {e}", name=worker_name)]}

Summary table

Debugging Multi-Agent

Per-agent logging

import json
from datetime import datetime

class AgentLogger:
    def __init__(self):
        self.logs: list[dict] = []

    def log(self, agent: str, event: str, data: dict):
        entry = {"ts": datetime.now().isoformat()[:19], "agent": agent, "event": event, **data}
        self.logs.append(entry)
        print(f"[{entry['ts']}] {agent}: {event}")

    def wrap_node(self, name: str, fn):
        def wrapped(state):
            self.log(name, "START", {"msgs": len(state.get("messages", []))})
            result = fn(state)
            self.log(name, "END", {"keys": list(result.keys())})
            return result
        return wrapped

logger = AgentLogger()
graph.add_node("researcher", logger.wrap_node("researcher", researcher_fn))

Tracing the supervisor's decisions

The most valuable thing: understanding why the Supervisor chose each worker. With structured output and reasoning:

class TracedDecision(BaseModel):
    next_worker: Literal["researcher", "analyst", "writer", "FINISH"]
    reasoning: str = Field(description="Why you chose this worker")
    confidence: float = Field(description="Confidence 0-1")

def traced_supervisor(state: dict) -> dict:
    decision = model.with_structured_output(TracedDecision).invoke(
        [SystemMessage(content="...")] + state["messages"])
    logger.log("supervisor", "DECISION", {
        "next": decision.next_worker,
        "why": decision.reasoning,
        "conf": decision.confidence,
    })
    return {"next_worker": decision.next_worker}

In production, send these logs to LangSmith or Datadog. Being able to filter by agent_name and see the sequence of decisions is invaluable.

Connection to the Project

In the next capsule you're going to build a complete multi-agent system. Everything from this capsule applies directly:

• Hierarchy: Start flat, scale to hierarchical if routing degrades with many workers
• Parallel execution: Use Send or asyncio.gather for independent workers
• Consensus: Implement at least adjudication for critical decisions
• Failure modes: At minimum: a timeout per worker, max iterations in the supervisor, context trimming
• Debugging: Use AgentLogger or LangSmith from the start, not after something fails

Troubleshooting

Problem 1: The L1 Supervisor always picks the same L2 team

Cause: The L1's prompt doesn't describe what each team does, or the L2's return messages don't indicate that its part is finished.

Solution: Include an explicit description of each team in the prompt. Make sure the L2 returns "Research complete" when it finishes. Add workers_called to the L1's prompt.

Problem 2: The Send API doesn't aggregate results correctly

Cause: The field that accumulates results has no reducer.

Solution: Change results: list[str] to results: Annotated[list[str], operator.add]. Without a reducer, each worker overwrites the previous result.

Problem 3: The debating agents never converge

Cause: The prompt incentivizes defending positions instead of converging.

Solution: Limit the rounds (maximum 3). Force adjudication at the end. Change the prompt to "if the other's evidence is convincing, update your position".

Problem 4: A timeout in parallel execution kills the whole group

Cause: asyncio.gather without return_exceptions=True cancels everything if one task fails.

Solution: Always use return_exceptions=True. Filter successful results from errors. A partial result is better than none.

Problem 5: Context overflow in a deep hierarchical system

Cause: Each level passes all the messages down to the next. With 3 levels they pile up fast.

Solution: Each level filters messages before passing them along. The L1 sends only the task to the L2. Results come back up as concise messages, not the full history.

Exercises

Exercise 1: Fan-out with 3 analysts (Basic)

Implement a system where a planner breaks a question into 3 sub-questions, 3 workers answer them in parallel using Send, and a synthesizer combines the results.

class ParallelState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sub_questions: list[str]
    answers: Annotated[list[str], operator.add]

def planner(state: ParallelState) -> dict:
    response = model.invoke([
        SystemMessage(content="Break this into 3 sub-questions. One per line."),
        state["messages"][-1]])
    return {"sub_questions": [q.strip() for q in response.content.split("\n") if q.strip()][:3]}

def fan_out(state: ParallelState) -> list[Send]:
    return [Send("analyst", {"messages": [HumanMessage(content=q)], "answers": []})
            for q in state["sub_questions"]]

def analyst(state: ParallelState) -> dict:
    q = state["messages"][-1].content
    r = model.invoke([SystemMessage(content="Answer concisely. 3 sentences max."),
                      HumanMessage(content=q)])
    return {"answers": [f"Q: {q}\nA: {r.content}"]}

def synthesizer(state: ParallelState) -> dict:
    r = model.invoke([SystemMessage(content="Integrate into a coherent answer."),
                      HumanMessage(content="\n\n".join(state["answers"]))])
    return {"messages": [r]}

graph = StateGraph(ParallelState)
graph.add_node("planner", planner)
graph.add_node("analyst", analyst)
graph.add_node("synthesizer", synthesizer)
graph.add_edge(START, "planner")
graph.add_conditional_edges("planner", fan_out)
graph.add_edge("analyst", "synthesizer")
graph.add_edge("synthesizer", END)
system = graph.compile()

Exercise 2: Voting with 3 classifiers (Intermediate)

3 agents classify a text as "positive", "negative" or "neutral". If there's no unanimity, a judge decides.

class VotingState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    votes: Annotated[list[str], operator.add]

def create_voter(name: str, bias: str):
    def vote(state: VotingState) -> dict:
        r = model.invoke([
            SystemMessage(content=f"{bias}\nClassify: POSITIVE, NEGATIVE or NEUTRAL. The word only."),
            state["messages"][-1]])
        v = r.content.strip().upper()
        return {"votes": [v if v in {"POSITIVE", "NEGATIVE", "NEUTRAL"} else "NEUTRAL"]}
    return vote

def evaluate(state: VotingState) -> dict:
    from collections import Counter
    counts = Counter(state["votes"])
    winner, count = counts.most_common(1)[0]
    if count == len(state["votes"]):
        return {"messages": [HumanMessage(content=f"Unanimous: {winner}")]}
    r = model.invoke([
        SystemMessage(content="Impartial judge. Decide the correct classification."),
        HumanMessage(content=f"Text: {state['messages'][0].content}\nVotes: {state['votes']}")])
    return {"messages": [HumanMessage(content=f"Judge ({state['votes']}): {r.content}")]}

graph = StateGraph(VotingState)
graph.add_node("v1", create_voter("optimist", "Look for the positive."))
graph.add_node("v2", create_voter("pessimist", "Look for problems."))
graph.add_node("v3", create_voter("neutral", "Evaluate objectively."))
graph.add_node("evaluate", evaluate)
graph.add_edge(START, "v1"); graph.add_edge(START, "v2"); graph.add_edge(START, "v3")
graph.add_edge("v1", "evaluate"); graph.add_edge("v2", "evaluate"); graph.add_edge("v3", "evaluate")
graph.add_edge("evaluate", END)
system = graph.compile()

Exercise 3: Circuit breaker for workers (Intermediate)

Implement a CircuitBreaker that "opens" after 3 consecutive failures of a worker, returning an immediate error without executing.

class CB:
    def __init__(self, max_f=3):
        self.failures: dict[str, int] = {}
        self.max_f = max_f
    def ok(self, n): return self.failures.get(n, 0) < self.max_f
    def fail(self, n): self.failures[n] = self.failures.get(n, 0) + 1
    def success(self, n): self.failures[n] = 0

cb = CB(max_f=3)

def protected(name, fn):
    def node(state):
        if not cb.ok(name):
            return {"messages": [HumanMessage(content=f"[{name}] CIRCUIT OPEN", name=name)]}
        try:
            result = fn(state)
            cb.success(name)
            return result
        except Exception as e:
            cb.fail(name)
            return {"messages": [HumanMessage(content=f"[{name}] FAIL: {e}", name=name)]}
    return node

import random
def flaky(state):
    if random.random() < 0.5: raise ConnectionError("timeout")
    return {"messages": [HumanMessage(content="[flaky] OK", name="flaky")]}

graph = StateGraph(WorkerState)
graph.add_node("worker", protected("flaky", flaky))
graph.add_edge(START, "worker")
graph.add_edge("worker", END)
system = graph.compile()

for i in range(5):
    r = system.invoke({"messages": [HumanMessage(content="test")]})
    print(f"Run {i+1}: {r['messages'][-1].content}")

Exercise 4: Hierarchical supervisor L1 → L2 (Hard)

Build an L1 that coordinates two L2 teams. A Research team (web, papers) and a Content team (writer, editor). Each L2 is a subgraph with its own supervisor.

def make_l2(team_name: str, workers: dict[str, str]):
    names = list(workers.keys())
    l2m = model.with_structured_output(L2Decision)

    def l2_sup(state: L2State) -> dict:
        it = state.get("iteration_count", 0) + 1
        if it > 3:
            return {"next_worker": "FINISH", "iteration_count": it,
                    "messages": [HumanMessage(content=f"[{team_name}] DONE", name=team_name)]}
        d = l2m.invoke([SystemMessage(content=f"You coordinate {names}. FINISH if ready.")]
                       + state["messages"])
        return {"next_worker": d.next_worker, "iteration_count": it,
                "messages": [HumanMessage(content=f"[{team_name}] → {d.next_worker}", name=team_name)]}

    b = StateGraph(L2State)
    b.add_node("sup", l2_sup)
    for wn, wi in workers.items():
        b.add_node(wn, make_worker(wn, wi))
    b.add_edge(START, "sup")
    routes = {wn: wn for wn in names}
    routes["FINISH"] = END
    b.add_conditional_edges("sup", lambda s: "FINISH" if s["next_worker"] == "FINISH"
                            else s["next_worker"], routes)
    for wn in names:
        b.add_edge(wn, "sup")
    return b.compile()

research = make_l2("research", {"web": "Search the web.", "papers": "Search for papers."})
content = make_l2("content", {"writer": "Write.", "editor": "Review and improve."})

def invoke_l2(team, state):
    r = team.invoke({"messages": state["messages"], "next_worker": "", "iteration_count": 0})
    return {"messages": [HumanMessage(content=r["messages"][-1].content, name="team")]}

top = StateGraph(L1State)
top.add_node("sup", l1_supervisor)
top.add_node("research", lambda s: invoke_l2(research, s))
top.add_node("content", lambda s: invoke_l2(content, s))
top.add_edge(START, "sup")
top.add_conditional_edges("sup", lambda s: "end" if s.get("task_complete") else s["next_team"],
    {"research": "research", "content": "content", "end": END})
top.add_edge("research", "sup"); top.add_edge("content", "sup")
hierarchy = top.compile()

Exercise 5: Parallel + consensus + synthesis (Hard)

Break the task into 2 sub-tasks. For each sub-task, run 2 workers with different perspectives. Apply adjudication if they differ. Synthesize at the end.

class FullState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sub_tasks: list[str]
    pair_results: Annotated[list[dict], operator.add]

def decompose(state: FullState) -> dict:
    r = model.invoke([SystemMessage(content="Break this into 2 sub-tasks. One per line."),
                      state["messages"][-1]])
    return {"sub_tasks": [t.strip() for t in r.content.split("\n") if t.strip()][:2]}

def fan_out_pairs(state: FullState) -> list[Send]:
    return [Send("pair", {"messages": [HumanMessage(content=t)], "pair_results": []})
            for t in state["sub_tasks"]]

def pair(state: FullState) -> dict:
    task = state["messages"][-1].content
    a = model.invoke([SystemMessage(content="Detailed technical perspective."),
                      HumanMessage(content=task)]).content
    b = model.invoke([SystemMessage(content="Concise practical perspective."),
                      HumanMessage(content=task)]).content
    if a[:50] == b[:50]:
        return {"pair_results": [{"task": task, "answer": a, "method": "agreement"}]}
    judge = model.invoke([SystemMessage(content="Pick the best one. Answer with it only."),
        HumanMessage(content=f"Task: {task}\n\nA: {a}\n\nB: {b}")]).content
    return {"pair_results": [{"task": task, "answer": judge, "method": "judge"}]}

def synthesize(state: FullState) -> dict:
    parts = "\n\n".join([f"### {p['task']}\n{p['answer']}" for p in state["pair_results"]])
    r = model.invoke([SystemMessage(content="Integrate into a coherent answer."),
                      HumanMessage(content=parts)])
    return {"messages": [r]}

graph = StateGraph(FullState)
graph.add_node("decompose", decompose)
graph.add_node("pair", pair)
graph.add_node("synthesize", synthesize)
graph.add_edge(START, "decompose")
graph.add_conditional_edges("decompose", fan_out_pairs)
graph.add_edge("pair", "synthesize")
graph.add_edge("synthesize", END)
system = graph.compile()

Summary

In this capsule you implemented the advanced multi-agent orchestration patterns:

• Hierarchical agents solve the scale problem: when a Supervisor has too many workers, you split into levels. The L1 coordinates teams (L2), and each L2 coordinates its workers. Each level is a subgraph with its own logic and anti-loop protection
• Parallel execution reduces latency when the workers are independent. The Send API implements native fan-out/fan-in — each Send creates a parallel execution whose result is aggregated automatically. asyncio.gather is the alternative for compiled subgraphs
• Consensus handles divergent answers: voting for classifications, debate for iterative convergence, adjudication with an external judge for the final decision
• Conflict resolution goes further: when results are genuinely contradictory, the supervisor arbitrates, you re-execute with cross context, or you escalate to a human
• Failure modes are new categories of failure: agent hangs, supervisor loops, context overflow, cascade failures — each with its specific prevention
• Multi-agent debugging requires per-agent logging, state inspection, and tracing of the supervisor's decisions with explicit reasoning

Next capsule: Final project — you're going to build a complete multi-agent system that integrates this module's patterns.

Additional Resources

1. LangGraph — Send API and Map-Reduce — Official documentation of the fan-out/fan-in pattern with Send
2. LangGraph — Multi-Agent Architectures — Multi-agent system concepts including hierarchies
3. LangGraph — Hierarchical Agent Teams — A tutorial on hierarchical teams with supervisors of supervisors
4. LangSmith — Tracing — Observability for debugging multi-agent systems
5. Circuit Breaker Pattern (Martin Fowler) — The circuit breaker pattern for distributed systems