Agentic Architectures — Article 6: Multi-Agent Orchestration Patterns

I spent about three weeks convinced that a single agent with a really well-engineered harness could handle anything I threw at it. Article 5 was basically that conviction written out in code. Then I got handed a project that needed to analyze a large codebase, cross-reference it against security advisories, generate remediation suggestions, and validate those suggestions against the existing test suite all in one run.

The agent kept running out of context. Or it would get so deep into one part of the codebase that it lost track of the security advisories it had retrieved twenty thousand tokens ago. Or it would finish the analysis and have no tokens left to actually generate useful remediations.

I tried every harness trick I knew. Better context injection, smarter summarization, tighter tool scoping. None of it solved the fundamental problem: the task was genuinely too wide for one agent to hold in its head at once. Some problems aren’t a prompting issue. They need multiple agents working together.

This article is about what I built instead, and the patterns I’ve found actually work in production on AWS Bedrock with LangGraph.

What You’ll Find Here

Before jumping into code, here are the questions this article directly answers. If you’ve been building agents for a while, some of these will be familiar frustrations.

When does a single agent actually stop being enough? Not every use case needs multiple agents. The first section breaks down the three distinct ceilings you’ll hit and helps you recognize which one you’re actually facing before over-engineering a solution.

Which orchestration pattern fits my problem? Supervisor-Worker, Pipeline, Parallel Fan-out, and Debate are four genuinely different patterns with different cost profiles, failure modes, and use cases. The article covers all four with working code so you can compare them against your specific task structure.

How do I pass context between agents without blowing up the token budget? The naive approach of forwarding full message histories breaks down fast. There’s a concrete handoff pattern here that separates results from reasoning traces and a compression step that keeps pipeline costs under control.

What breaks in production that doesn’t break in a demo? Cost explosions when supervisors over-decompose, circular delegation loops, partial failures in parallel runs, and the observability problem of correlating traces across twelve agents. The production hardening section covers all of these with implementation-ready code.

How does this connect to what came before? Everything from Article 5 — the harness, JWT auth at tool boundaries, circuit breakers, loop detection — carries forward here. This article extends those patterns into multi-agent territory rather than replacing them.

Why a Single Agent Hits a Ceiling

Context limits are the obvious constraint, but they’re not the only one. There are three distinct reasons a single agent architecture eventually breaks down.

Context exhaustion is what everyone talks about. Even with 200k token windows, long-running tasks that accumulate tool outputs, intermediate reasoning, and conversation history will eventually run out of room. The agent starts forgetting things it retrieved earlier. Its reasoning quality degrades as the context fills up.

Specialization vs. generalization tradeoff is subtler. A single agent prompted to be good at everything ends up being mediocre at everything. An agent that needs to write Python, review security implications, and evaluate test coverage simultaneously is juggling three different cognitive modes. In practice, you get better results from a security specialist agent reviewing the output of a code generation agent than from one agent trying to do both.

Parallelization is the third and most underappreciated constraint. Some tasks are just independent of each other and can run concurrently. Running them sequentially in a single agent is leaving performance on the table.

The good news is that LangGraph makes multi-agent topologies first-class citizens. The graph model naturally represents agent coordination patterns in a way that sequential frameworks don’t.

A Taxonomy of Multi-Agent Patterns

Before writing any code, I want to lay out the four patterns this article covers. They’re not mutually exclusive, and real production systems often combine them.

+--------------------+--------------------------------------------------+------------------+
| Pattern            | Structure                                        | Best For         |
+--------------------+--------------------------------------------------+------------------+
| Supervisor-Worker  | One orchestrator decomposes + delegates          | Complex tasks    |
|                    | N workers execute specialized subtasks           | needing expert   |
|                    |                                                  | decomposition    |
+--------------------+--------------------------------------------------+------------------+
| Pipeline           | Agent A -> Agent B -> Agent C (sequential)       | Transformation   |
|                    | Each processes the previous output               | chains, ETL-like |
|                    |                                                  | workflows        |
+--------------------+--------------------------------------------------+------------------+
| Parallel Fan-out   | Orchestrator sends same/related task to N agents | Research,        |
|                    | Results aggregated into single output            | analysis tasks   |
|                    |                                                  | that parallelize |
+--------------------+--------------------------------------------------+------------------+
| Debate / Critique  | Agent A produces solution, Agent B critiques     | High-stakes      |
|                    | Agent C synthesizes or adjudicates               | outputs needing  |
|                    |                                                  | adversarial QA   |
+--------------------+--------------------------------------------------+------------------+

The choice between patterns depends on your task structure, not on which pattern sounds most impressive. I’ve seen teams reach for Debate when a simple Supervisor-Worker would have been faster and cheaper.

Pattern 1: Supervisor-Worker

This is the pattern I reach for most often. A supervisor agent breaks a complex task into subtasks, assigns them to specialized workers, collects their outputs, and synthesizes a final result. The supervisor doesn’t do the actual work it manages the work.

          ┌─────────────────────────────┐
          │       SUPERVISOR AGENT      │
          │  (task decomposition +      │
          │   result synthesis)         │
          └──────┬───────────┬──────────┘
                 │           │
        ┌────────▼──┐   ┌────▼──────────┐  ┌──────────────┐
        │  Worker A  │  │   Worker B    │  │   Worker C   │
        │  (code     │  │  (security    │  │  (test       │
        │   analysis)│  │   review)     │  │   coverage)  │
        └────────────┘  └───────────────┘  └──────────────┘

The implementation in LangGraph leans on the concept of a “supervisor node” that routes messages to worker subgraphs based on the current task state. Each worker is its own compiled graph they can have their own internal tools, verification loops, and error handling from Article 5.

# harness/multi_agent/supervisor.py
from typing import Literal, TypedDict, Annotated, List
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_aws import ChatBedrock
from pydantic import BaseModel
import boto3

class SupervisorState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    task_spec: str
    subtasks: List[dict]           # decomposed work items
    worker_results: dict           # keyed by subtask ID
    current_worker: str            # which worker is active
    synthesis_complete: bool
    agent_run_id: str

class SubtaskAssignment(BaseModel):
    """Structured output from supervisor decomposition."""
    subtask_id: str
    worker_type: Literal["code_analyst", "security_reviewer", "test_evaluator"]
    description: str
    depends_on: List[str]          # subtask IDs this depends on
    priority: int

SUPERVISOR_SYSTEM_PROMPT = """
You are an orchestrator agent. You do not write code or perform analysis yourself.
Your job is to:

1. Break the task into discrete subtasks
2. Assign each subtask to the correct specialist worker
3. Track dependencies between subtasks
4. Synthesize worker outputs into a coherent final result

Available workers:
- code_analyst: Reads and analyzes code structure, dependencies, patterns
- security_reviewer: Evaluates security implications, checks against CVEs
- test_evaluator: Assesses test coverage, identifies gaps

When decomposing, be specific. A subtask description like "analyze the auth module"
is useful. "analyze the code" is not.

Respond in JSON when asked to decompose. Respond in prose when asked to synthesize.
"""

def build_supervisor(region: str = "us-east-1") -> callable:
    bedrock = boto3.client("bedrock-runtime", region_name=region)

    # Supervisor uses the heavier model — it's doing strategic reasoning
    supervisor_model = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
        model_kwargs={
            "temperature": 0.2,
            "max_tokens": 8000,
            "thinking": {"type": "enabled", "budget_tokens": 5000}
        }
    )

    def supervisor_node(state: SupervisorState) -> SupervisorState:
        if not state.get("subtasks"):
            # First pass: decompose the task
            response = supervisor_model.invoke([
                SystemMessage(content=SUPERVISOR_SYSTEM_PROMPT),
                HumanMessage(content=f"Decompose this task into subtasks:\n{state['task_spec']}")
            ])
            import json
            try:
                subtasks = json.loads(response.content)
                if isinstance(subtasks, dict) and "subtasks" in subtasks:
                    subtasks = subtasks["subtasks"]
            except json.JSONDecodeError:
                subtasks = []

            return {**state, "subtasks": subtasks}

        # All workers done: synthesize
        results_summary = "\n\n".join([
            f"=== {worker} ===\n{result}"
            for worker, result in state["worker_results"].items()
        ])
        synthesis_prompt = f"""
Original task: {state['task_spec']}

Worker results:
{results_summary}

Synthesize these into a coherent final report. Highlight conflicts between
worker findings and make clear recommendations.
"""
        response = supervisor_model.invoke([
            SystemMessage(content=SUPERVISOR_SYSTEM_PROMPT),
            HumanMessage(content=synthesis_prompt)
        ])
        return {
            **state,
            "messages": state["messages"] + [response],
            "synthesis_complete": True,
        }

    return supervisor_node

def route_to_worker(state: SupervisorState) -> str:
    """
    Routes to the next worker with unfinished subtasks.
    Returns END when all subtasks are complete and synthesis is done.
    """
    if state.get("synthesis_complete"):
        return END

    # Find next unfinished subtask whose dependencies are met
    completed = set(state.get("worker_results", {}).keys())
    for subtask in state.get("subtasks", []):
        sid = subtask["subtask_id"]
        if sid in completed:
            continue
        deps = set(subtask.get("depends_on", []))
        if deps.issubset(completed):
            return subtask["worker_type"]   # route to this worker

    # All subtasks done, back to supervisor for synthesis
    return "supervisor"

The worker agents are deliberately thin. They receive a specific subtask description and return a result. They don’t need to know about the supervisor or the other workers.

# harness/multi_agent/workers.py
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3

CODE_ANALYST_PROMPT = """
You are a code analysis specialist. You receive specific, bounded analysis tasks.
Focus only on what you were asked to analyze. Do not expand scope.
Return structured findings: what you found, confidence level, and specific evidence.
"""

SECURITY_REVIEWER_PROMPT = """
You are a security review specialist. You look for vulnerabilities, insecure patterns,
and CVE-relevant code. Reference specific CWE numbers when applicable.
Return structured findings with severity levels (CRITICAL, HIGH, MEDIUM, LOW).
"""

TEST_EVALUATOR_PROMPT = """
You are a test coverage specialist. You assess test quality and identify gaps.
Focus on: coverage percentage where available, missing edge cases, and untested paths.
Return structured findings with specific test gaps and suggested test cases.
"""

WORKER_PROMPTS = {
    "code_analyst": CODE_ANALYST_PROMPT,
    "security_reviewer": SECURITY_REVIEWER_PROMPT,
    "test_evaluator": TEST_EVALUATOR_PROMPT,
}

def build_worker(worker_type: str, region: str = "us-east-1") -> callable:
    bedrock = boto3.client("bedrock-runtime", region_name=region)

    # Workers use the faster model — they execute a specific, bounded task
    worker_model = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        model_kwargs={"temperature": 0, "max_tokens": 4000}
    )

    system_prompt = WORKER_PROMPTS[worker_type]

    def worker_node(state: SupervisorState) -> SupervisorState:
        # Find the subtask assigned to this worker type
        completed = set(state.get("worker_results", {}).keys())
        current_subtask = None
        for subtask in state["subtasks"]:
            if subtask["worker_type"] == worker_type and subtask["subtask_id"] not in completed:
                deps = set(subtask.get("depends_on", []))
                if deps.issubset(completed):
                    current_subtask = subtask
                    break

        if not current_subtask:
            return state

        # Pass relevant prior results as context if this task has dependencies
        context = ""
        if current_subtask.get("depends_on"):
            for dep_id in current_subtask["depends_on"]:
                if dep_id in state.get("worker_results", {}):
                    context += f"\nPrevious analysis ({dep_id}):\n{state['worker_results'][dep_id]}\n"

        prompt = f"Task: {current_subtask['description']}"
        if context:
            prompt = f"Prior context:{context}\n\n{prompt}"

        response = worker_model.invoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=prompt)
        ])

        updated_results = {**state.get("worker_results", {})}
        updated_results[current_subtask["subtask_id"]] = response.content

        return {**state, "worker_results": updated_results}

    return worker_node

Wire it all together in LangGraph:

# harness/multi_agent/supervisor_graph.py
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from harness.multi_agent.supervisor import SupervisorState, build_supervisor, route_to_worker
from harness.multi_agent.workers import build_worker

def build_supervisor_graph(region: str = "us-east-1"):
    graph = StateGraph(SupervisorState)

    graph.add_node("supervisor", build_supervisor(region))
    graph.add_node("code_analyst", build_worker("code_analyst", region))
    graph.add_node("security_reviewer", build_worker("security_reviewer", region))
    graph.add_node("test_evaluator", build_worker("test_evaluator", region))

    graph.add_edge(START, "supervisor")

    graph.add_conditional_edges(
        "supervisor",
        route_to_worker,
        {
            "code_analyst": "code_analyst",
            "security_reviewer": "security_reviewer",
            "test_evaluator": "test_evaluator",
            END: END,
        }
    )

    # All workers route back to supervisor after completing their subtask
    for worker in ["code_analyst", "security_reviewer", "test_evaluator"]:
        graph.add_edge(worker, "supervisor")

    return graph.compile(checkpointer=MemorySaver())

Reference: LangGraph supervisor multi-agent tutorial: https://langchain-ai.github.io/langgraph/tutorials/multi_agent/agent_supervisor/

Pattern 2: Pipeline Orchestration

Pipelines are sequential by design. Agent A produces output, Agent B transforms it, Agent C evaluates it. Each agent has a narrow, well-defined job.

  ┌───────────────┐     ┌───────────────┐     ┌───────────────┐
  │   Agent A     │     │   Agent B     │     │   Agent C     │
  │  (extraction) │────>│ (enrichment)  │────>│ (validation)  │
  └───────────────┘     └───────────────┘     └───────────────┘
       output A               output B               output C
    becomes input B        becomes input C         final result

The main challenge in pipelines is context compression. By the time you’re three agents deep, you don’t want Agent C carrying the full message history from Agents A and B that’s thousands of tokens of intermediate reasoning it doesn’t need. You want the result of Agent A’s work, not the trace of how Agent A got there.

# harness/multi_agent/pipeline.py
from typing import TypedDict, Annotated, List, Optional, Any
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_aws import ChatBedrock
import boto3

class PipelineState(TypedDict):
    original_input: str
    stage_outputs: List[dict]       # accumulates each stage's compressed result
    current_stage: int
    final_output: Optional[str]
    agent_run_id: str

def compress_for_handoff(full_output: str, model: ChatBedrock) -> str:
    """
    Compresses a stage's full output into a structured handoff summary.
    This is the core of pipeline context management — the next agent gets
    the substance, not the reasoning trace.
    """
    response = model.invoke([
        SystemMessage(content="""
Compress the following agent output into a structured handoff summary.
Include: key findings, decisions made, artifacts produced, and what the
next stage needs to know. Discard reasoning traces and intermediate steps.
Target length: 20% of original. Use bullet points for clarity.
"""),
        HumanMessage(content=f"Compress this:\n\n{full_output}")
    ])
    return response.content

def build_pipeline_stage(
    stage_name: str,
    system_prompt: str,
    region: str = "us-east-1"
) -> callable:
    bedrock = boto3.client("bedrock-runtime", region_name=region)
    model = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        model_kwargs={"temperature": 0, "max_tokens": 6000}
    )
    # Cheaper model for compression — this is mechanical, not creative
    compressor = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-haiku-3-5",
        model_kwargs={"temperature": 0, "max_tokens": 2000}
    )

    def stage_node(state: PipelineState) -> PipelineState:
        # Build context from compressed prior stage outputs only
        prior_context = ""
        for past_stage in state.get("stage_outputs", []):
            prior_context += f"\n### {past_stage['stage']} output:\n{past_stage['compressed_output']}\n"

        prompt = f"Original task: {state['original_input']}\n"
        if prior_context:
            prompt += f"\nPrior stage results:\n{prior_context}\n"
        prompt += f"\nNow perform your stage: {stage_name}"

        response = model.invoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=prompt)
        ])
        full_output = response.content

        # Compress before storing — next stage won't see raw output
        compressed = compress_for_handoff(full_output, compressor)

        updated_outputs = list(state.get("stage_outputs", []))
        updated_outputs.append({
            "stage": stage_name,
            "full_output": full_output,
            "compressed_output": compressed,
        })

        return {
            **state,
            "stage_outputs": updated_outputs,
            "current_stage": state.get("current_stage", 0) + 1,
        }

    return stage_node

One thing I want to flag on pipelines: the compression step is important but it’s also a fidelity tradeoff. You will lose information. For some pipelines that’s acceptable. For others especially ones where Agent C needs to trace Agent A’s exact reasoning — you need to think more carefully about what gets compressed versus what gets passed through in full.

Pattern 3: Parallel Fan-out / Fan-in

The parallel pattern is straightforward in concept: split a task, run agents concurrently, merge the results. In practice, getting the merge right is where most implementations fall apart.

                    ┌──────────────────┐
                    │   ORCHESTRATOR   │
                    │  (task splitter) │
                    └──┬───┬───┬───┬──┘
                       │   │   │   │
            ┌──────────▼┐ ┌▼─┐ ┌▼──┐ ┌▼──────────┐
            │  Agent 1  │ │A2│ │A3 │ │  Agent 4  │
            │ (region A)│ │  │ │   │ │ (region D)│
            └──────────┬┘ └┬─┘ └┬──┘ └┬──────────┘
                       │   │    │     │
                    ┌──▼───▼────▼─────▼──┐
                    │     AGGREGATOR     │
                    │  (result merger)   │
                    └────────────────────┘

The implementation uses Python’s asyncio to run agents concurrently. On Bedrock, this is straightforward each concurrent agent call is a separate API request.

# harness/multi_agent/fanout.py
import asyncio
from typing import List, TypedDict, Annotated, Optional
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3

class FanoutResult(TypedDict):
    agent_id: str
    input_slice: str
    output: str
    success: bool
    error: Optional[str]

async def run_agent_async(
    agent_id: str,
    input_slice: str,
    system_prompt: str,
    model: ChatBedrock,
) -> FanoutResult:
    """Run a single agent asynchronously."""
    try:
        response = await model.ainvoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=input_slice)
        ])
        return FanoutResult(
            agent_id=agent_id,
            input_slice=input_slice,
            output=response.content,
            success=True,
            error=None,
        )
    except Exception as e:
        return FanoutResult(
            agent_id=agent_id,
            input_slice=input_slice,
            output="",
            success=False,
            error=str(e),
        )

async def fan_out(
    task_slices: List[str],
    system_prompt: str,
    region: str = "us-east-1",
    max_concurrent: int = 5,   # don't hammer Bedrock rate limits
) -> List[FanoutResult]:
    """
    Runs agents concurrently with a semaphore to cap parallelism.
    max_concurrent protects against Bedrock throttling — you
    will hit rate limits if you fire 20 concurrent requests.
    """
    bedrock = boto3.client("bedrock-runtime", region_name=region)
    model = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        model_kwargs={"temperature": 0, "max_tokens": 4000}
    )

    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_run(agent_id, slice_content):
        async with semaphore:
            return await run_agent_async(agent_id, slice_content, system_prompt, model)

    tasks = [
        bounded_run(f"agent_{i}", slice_content)
        for i, slice_content in enumerate(task_slices)
    ]
    return await asyncio.gather(*tasks)

def aggregate_results(
    results: List[FanoutResult],
    aggregator_model: ChatBedrock,
    aggregation_strategy: str = "synthesize",
) -> str:
    """
    Merges parallel agent outputs.

    aggregation_strategy options:
    - "synthesize": ask a model to merge findings coherently
    - "concat": simple concatenation (fast, no model call needed)
    - "vote": majority-vote for classification tasks
    """
    if aggregation_strategy == "concat":
        successful = [r for r in results if r["success"]]
        return "\n\n---\n\n".join(r["output"] for r in successful)

    failed = [r for r in results if not r["success"]]
    successful = [r for r in results if r["success"]]

    if failed:
        # Log partial failures but don't crash — partial results are usually useful
        for f in failed:
            print(f"Agent {f['agent_id']} failed: {f['error']}")

    outputs_for_synthesis = "\n\n".join([
        f"[Agent {r['agent_id']}]:\n{r['output']}"
        for r in successful
    ])

    response = aggregator_model.invoke([
        SystemMessage(content="""
You are a results aggregator. You receive outputs from multiple parallel agents
that each analyzed a different slice of the same problem. Your job is to:
1. Identify common findings across agents
2. Surface unique findings from individual agents
3. Flag any contradictions between agents and explain which to trust
4. Produce a single coherent output as if one expert had analyzed everything

Do not simply concatenate. Actively synthesize.
"""),
        HumanMessage(content=f"Synthesize these {len(successful)} agent outputs:\n\n{outputs_for_synthesis}")
    ])
    return response.content

# Convenience wrapper for synchronous callers
def run_parallel_analysis(
    task_slices: List[str],
    system_prompt: str,
    region: str = "us-east-1",
) -> str:
    results = asyncio.run(fan_out(task_slices, system_prompt, region))
    bedrock = boto3.client("bedrock-runtime", region_name=region)
    aggregator = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
        model_kwargs={"temperature": 0, "max_tokens": 8000}
    )
    return aggregate_results(results, aggregator)

One honest note on cost: fan-out multiplies your Bedrock costs linearly. Four parallel agents running the same task costs roughly four times as much as one agent. That’s usually fine when you’re parallelizing genuinely independent work, but it’s a bad idea when the task could be handled sequentially. I’ve seen teams reach for fan-out because it feels more powerful, not because the problem actually requires it.

Reference: LangGraph parallel node execution: https://langchain-ai.github.io/langgraph/how-tos/branching/

Pattern 4: Debate / Critique

Two agents independently approach the same problem, then a third evaluates their outputs and produces a synthesis. This is expensive and slow, which means it’s only worth it for outputs where quality matters more than speed high-stakes decisions, security-sensitive code, legal document review.

  ┌─────────────┐          ┌─────────────┐
  │  Proposer   │          │  Challenger │
  │  (solution  │          │  (solution  │
  │   attempt 1)│          │   attempt 2)│
  └──────┬──────┘          └──────┬──────┘
         │                        │
         └──────────┬─────────────┘
                    ▼
          ┌─────────────────┐
          │    ADJUDICATOR  │
          │  (critique +    │
          │   synthesis)    │
          └─────────────────┘

# harness/multi_agent/debate.py
from dataclasses import dataclass
from typing import Optional
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3

@dataclass
class DebateResult:
    proposer_solution: str
    challenger_solution: str
    adjudication: str
    final_recommendation: str
    agreement_level: str        # HIGH | MEDIUM | LOW | CONTRADICTION

PROPOSER_PROMPT = """
You are a solution proposer. Approach the problem carefully and produce your best
solution. Explain your reasoning. Do not hedge excessively — commit to a specific answer.
"""

CHALLENGER_PROMPT = """
You are a solution challenger. You will receive a problem that another agent has
already attempted. Produce your own independent solution WITHOUT seeing their work.
Approach this fresh. Your goal is not to contradict — it's to find the best solution.
"""

ADJUDICATOR_PROMPT = """
You are an adjudicator reviewing two independent solutions to the same problem.
Your job:
1. Identify where the two solutions agree (these are likely correct)
2. Identify where they diverge (these need careful evaluation)
3. For each divergence, evaluate which solution is better and why
4. Produce a final synthesis that takes the best of both

Be direct about contradictions. Do not smooth over genuine disagreements —
surface them clearly so the human reviewer can make a judgment call.

Rate the agreement level: HIGH (minor differences), MEDIUM (some significant
divergences), LOW (fundamentally different approaches), or CONTRADICTION
(mutually exclusive conclusions).
"""

def run_debate(
    problem: str,
    region: str = "us-east-1",
) -> DebateResult:
    bedrock = boto3.client("bedrock-runtime", region_name=region)

    heavy_model = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
        model_kwargs={
            "temperature": 0.3,        # slight temperature for independent solutions
            "max_tokens": 6000,
            "thinking": {"type": "enabled", "budget_tokens": 4000}
        }
    )

    # Proposer works the problem
    proposer_response = heavy_model.invoke([
        SystemMessage(content=PROPOSER_PROMPT),
        HumanMessage(content=problem)
    ])
    proposer_solution = proposer_response.content

    # Challenger works the same problem independently
    # Note: Challenger does NOT see Proposer's solution
    challenger_response = heavy_model.invoke([
        SystemMessage(content=CHALLENGER_PROMPT),
        HumanMessage(content=problem)
    ])
    challenger_solution = challenger_response.content

    # Adjudicator sees both and synthesizes
    adjudicator_response = heavy_model.invoke([
        SystemMessage(content=ADJUDICATOR_PROMPT),
        HumanMessage(content=f"""
Problem: {problem}

Solution A (Proposer):
{proposer_solution}

Solution B (Challenger):
{challenger_solution}

Adjudicate and synthesize.
""")
    ])

    adjudication = adjudicator_response.content

    # Extract agreement level from adjudication
    agreement_level = "MEDIUM"
    for level in ["CONTRADICTION", "LOW", "HIGH", "MEDIUM"]:
        if level in adjudication.upper():
            agreement_level = level
            break

    return DebateResult(
        proposer_solution=proposer_solution,
        challenger_solution=challenger_solution,
        adjudication=adjudication,
        final_recommendation=adjudication,
        agreement_level=agreement_level,
    )

One thing the Debate pattern reveals that single-agent evaluation doesn’t: when two independently-reasoning models arrive at the same conclusion, you have much stronger grounds for confidence. When they diverge, that divergence itself is signal it tells you the problem has genuine ambiguity that needs human judgment.

Inter-Agent Handoff and Context Passing

Regardless of which pattern you use, you need to think carefully about what gets passed between agents. This is where the context window problem from single-agent systems reappears in a different form.

The naive approach is to pass the full message history. That works fine for two agents but breaks down quickly as chains get longer. Instead, I use a structured handoff object that separates results from reasoning traces.

# harness/multi_agent/handoff.py
from dataclasses import dataclass, field
from typing import Any, Optional
from langchain_core.messages import HumanMessage

@dataclass
class AgentHandoff:
    """
    Structured context passed between agents.

    The split between result and trace is deliberate: downstream agents
    need the result, not the full reasoning history. Keeping them separate
    lets each agent decide how much context it wants to consume.
    """
    source_agent: str
    task_completed: str
    result_summary: str              # compressed, structured result
    artifacts: dict = field(default_factory=dict)  # files, code, structured data
    reasoning_trace: Optional[str] = None          # full trace, passed only if downstream needs it
    confidence: str = "MEDIUM"       # HIGH | MEDIUM | LOW
    flags: list = field(default_factory=list)      # NEEDS_REVIEW, PARTIAL_RESULT, etc.

    def to_context_message(self, include_trace: bool = False) -> HumanMessage:
        """
        Converts handoff to a HumanMessage for injection into next agent's context.
        include_trace=True only when the downstream agent genuinely needs the reasoning.
        """
        content = f"""
[HANDOFF FROM: {self.source_agent}]
Task completed: {self.task_completed}
Confidence: {self.confidence}
Flags: {', '.join(self.flags) if self.flags else 'none'}

Result summary:
{self.result_summary}
"""
        if self.artifacts:
            content += f"\nArtifacts available:\n"
            for key, value in self.artifacts.items():
                if isinstance(value, str) and len(value) < 500:
                    content += f"  {key}: {value}\n"
                else:
                    content += f"  {key}: [available, {type(value).__name__}]\n"

        if include_trace and self.reasoning_trace:
            content += f"\nFull reasoning trace:\n{self.reasoning_trace}"

        return HumanMessage(content=content)

Production Hardening

Multi-agent systems introduce failure modes that single-agent systems don’t have. Some of them are subtle. I’ve hit all of these in production.

Cost Explosion Prevention

When agents can spawn sub-agents, costs can spiral fast. A supervisor that decomposes aggressively can spawn 20 workers from a single user request. You need hard limits.

# harness/multi_agent/budget.py
import boto3
import time
from decimal import Decimal

class AgentBudgetGuard:
    """
    Tracks cumulative cost and agent count per run.
    Hard-stops execution when limits are exceeded.
    """

    def __init__(
        self,
        max_agents_per_run: int = 10,
        max_total_tokens: int = 500_000,
        table_name: str = "agent-budget-state",
        region: str = "us-east-1",
    ):
        self.max_agents = max_agents_per_run
        self.max_tokens = max_total_tokens
        self.table = boto3.resource("dynamodb", region_name=region).Table(table_name)

    def register_agent_spawn(self, run_id: str, agent_id: str) -> bool:
        """
        Returns True if spawn is allowed, False if budget exceeded.
        Call this before spawning any sub-agent.
        """
        response = self.table.update_item(
            Key={"run_id": run_id},
            UpdateExpression="SET agent_count = if_not_exists(agent_count, :z) + :inc",
            ExpressionAttributeValues={":z": 0, ":inc": 1},
            ReturnValues="UPDATED_NEW",
        )
        new_count = int(response["Attributes"]["agent_count"])

        if new_count > self.max_agents:
            raise AgentBudgetExceededError(
                f"Run {run_id} attempted to spawn agent #{new_count}, "
                f"but max_agents_per_run is {self.max_agents}. "
                f"Either the supervisor is over-decomposing, or there is a spawn loop."
            )
        return True

    def record_token_usage(self, run_id: str, tokens_used: int):
        response = self.table.update_item(
            Key={"run_id": run_id},
            UpdateExpression="SET total_tokens = if_not_exists(total_tokens, :z) + :inc",
            ExpressionAttributeValues={":z": 0, ":inc": tokens_used},
            ReturnValues="UPDATED_NEW",
        )
        total = int(response["Attributes"]["total_tokens"])
        if total > self.max_tokens:
            raise AgentBudgetExceededError(
                f"Run {run_id} consumed {total:,} tokens, exceeding limit of {self.max_tokens:,}."
            )

class AgentBudgetExceededError(Exception):
    pass

Deadlock Detection

Circular delegation is harder to hit than you’d think, but when you do hit it, it’s spectacular. Supervisor delegates to Worker A, Worker A decides it needs to delegate back to the Supervisor, which delegates again. Most of the time LangGraph’s graph structure prevents this, but if you’re doing dynamic routing it’s possible.

# harness/multi_agent/deadlock.py
import boto3
import time
from typing import List

class DeadlockDetector:
    """
    Tracks the delegation chain per run and detects cycles.
    Stored in DynamoDB so it works across parallel agent branches.
    """

    def __init__(self, table_name: str = "agent-delegation-chain", region: str = "us-east-1"):
        self.table = boto3.resource("dynamodb", region_name=region).Table(table_name)

    def record_delegation(self, run_id: str, from_agent: str, to_agent: str):
        """
        Records a delegation event and checks for cycles.
        Raises DeadlockDetectedError if a cycle is found.
        """
        self.table.put_item(Item={
            "run_id": run_id,
            "delegation_id": f"{from_agent}->{to_agent}-{int(time.time())}",
            "from_agent": from_agent,
            "to_agent": to_agent,
            "timestamp": int(time.time()),
        })

        chain = self._get_delegation_chain(run_id)
        if self._has_cycle(chain):
            cycle_description = self._describe_cycle(chain)
            raise DeadlockDetectedError(
                f"Delegation cycle detected in run {run_id}: {cycle_description}. "
                f"Check supervisor decomposition logic for circular dependencies."
            )

    def _get_delegation_chain(self, run_id: str) -> List[tuple]:
        response = self.table.query(
            KeyConditionExpression="run_id = :rid",
            ExpressionAttributeValues={":rid": run_id}
        )
        return [(item["from_agent"], item["to_agent"]) for item in response.get("Items", [])]

    def _has_cycle(self, chain: List[tuple]) -> bool:
        graph = {}
        for from_a, to_a in chain:
            graph.setdefault(from_a, set()).add(to_a)

        visited, rec_stack = set(), set()

        def dfs(node):
            visited.add(node)
            rec_stack.add(node)
            for neighbor in graph.get(node, []):
                if neighbor not in visited:
                    if dfs(neighbor): return True
                elif neighbor in rec_stack:
                    return True
            rec_stack.discard(node)
            return False

        return any(dfs(node) for node in graph if node not in visited)

    def _describe_cycle(self, chain: List[tuple]) -> str:
        return " -> ".join(f"{f}->{t}" for f, t in chain[-5:])

class DeadlockDetectedError(Exception):
    pass

Cross-Agent Observability

When a user request touches five agents, you need a single trace view that shows the full graph, not five disconnected traces. LangSmith handles this through run trees each sub-agent run is a child of the parent run, so you can navigate the full hierarchy.

# harness/multi_agent/observability.py
import os
from contextlib import contextmanager
from langsmith import Client
from langsmith.run_trees import RunTree

class MultiAgentTracer:
    """
    Maintains a run tree across all agents in a multi-agent system.
    Pass the parent_run_id to each agent so their traces nest correctly.
    """

    def __init__(self):
        self.client = Client()

    @contextmanager
    def agent_span(self, parent_run_id: str, agent_name: str, inputs: dict):
        """
        Context manager for an individual agent's trace span.

        Usage:
            with tracer.agent_span(parent_run_id, "security_reviewer", {...}) as span:
                result = run_security_review(...)
                span.end(outputs={"result": result})
        """
        run = self.client.create_run(
            name=agent_name,
            run_type="chain",
            inputs=inputs,
            parent_run_id=parent_run_id,
        )
        try:
            yield run
        except Exception as e:
            self.client.update_run(run.id, error=str(e))
            raise
        finally:
            self.client.update_run(run.id, end_time=None)  # auto-sets end time

Reference: LangSmith run trees for multi-agent tracing: https://docs.smith.langchain.com/how_to_guides/tracing/trace_with_langgraph

Production Reality Check

Multi-agent systems are the point in this series where complexity stops being exciting and starts being something you have to manage carefully.

Debugging becomes exponentially harder. When a single-agent run fails, you have one trace to read. When a supervisor-worker system fails, you might have twelve traces to correlate. Invest in your observability setup before you invest in your agent topology. If you can’t see what all the agents are doing in a single view, you will spend days debugging things that should take minutes.

Cost scales with agent count. The reasoning sandwich from Article 5 gets multiplied across every agent in your system. A five-agent supervisor-worker setup running extended thinking models can burn through a surprisingly large token budget on a single user request. Set hard budget limits from day one. The AgentBudgetGuard above is not optional it's the difference between a controlled cost model and a surprise invoice.

Partial failure is the norm, not the exception. In single-agent systems, either the run succeeds or it fails. In multi-agent systems, five of six workers might succeed and one might fail. You need to decide ahead of time what partial success means for your application. Does a security review with one failed agent return partial results or fail completely? This is a product decision, not a technical one but it needs to be made explicitly, not left to default behavior.

Model upgrades get more complicated. In Article 5 I mentioned that harnesses are model-specific. With multiple agents potentially running different models, a model upgrade can mean re-tuning multiple system prompts and verification loops. Keep a test suite for each agent type. It’s the only way to upgrade models with any confidence.

Coordinator agents are a single point of failure. In the Supervisor-Worker pattern, the supervisor going down means everything stops. In production, think about supervisor state persistence if the supervisor crashes mid-run, can it resume? LangGraph’s checkpointing handles this, but you need to configure it deliberately.

Reference Architecture

                  ┌───────────────────────────────────────────┐
                  │            User Request                   │
                  └──────────────────┬────────────────────────┘
                                     │
                  ┌──────────────────▼────────────────────────┐
                  │           AgentHarness Runtime            │
                  │  (budget guard, deadlock detector, tracer)│
                  └──────────────────┬────────────────────────┘
                                     │
                  ┌──────────────────▼────────────────────────┐
                  │         SUPERVISOR / ORCHESTRATOR         │
                  │   Claude 3.7 + extended thinking          │
                  │   Task decomposition + result synthesis   │
                  └────┬──────────────┬──────────────┬────────┘
                       │              │              │
           ┌───────────▼──┐  ┌────────▼──┐  ┌───────▼───────┐
           │   Worker A   │  │  Worker B  │  │   Worker C   │
           │  Claude 3.5  │  │ Claude 3.5 │  │  Claude 3.5  │
           │  specialist  │  │ specialist │  │  specialist  │
           └───────┬──────┘  └─────┬─────┘  └──────┬────────┘
                   │               │                │
           ┌───────▼───────────────▼────────────────▼────────┐
           │              Tool Execution Layer               │
           │     (auth, retry, circuit breaker from Art.5)   │
           └───────────────────────┬─────────────────────────┘
                                   │
           ┌───────────────────────▼──────────────────────────┐
           │                  AWS Services                    │
           │  Bedrock │ DynamoDB (budget+deadlock+loop+circ)  │
           │  Secrets Manager │ Knowledge Bases               │
           └──────────────────────────────────────────────────┘

  Observability: LangSmith run trees — full hierarchy per user request

Reference Infrastructure Stack

+-----------------------------+---------------------+------------------------------+
| Component                   | Technology          | Role                         |
+-----------------------------+---------------------+------------------------------+
| Orchestration               | LangGraph 0.2+      | Multi-agent graph, routing,  |
|                             |                     | supervisor-worker topology   |
+-----------------------------+---------------------+------------------------------+
| Supervisor Model            | Claude 3.7 Sonnet   | Task decomposition,          |
|                             | (extended thinking) | result synthesis             |
+-----------------------------+---------------------+------------------------------+
| Worker Models               | Claude 3.5 Sonnet   | Specialized execution,       |
|                             |                     | bounded tasks                |
+-----------------------------+---------------------+------------------------------+
| Parallel Execution          | asyncio + Bedrock   | Concurrent agent runs with   |
|                             |                     | semaphore-gated concurrency  |
+-----------------------------+---------------------+------------------------------+
| Context Compression         | Claude Haiku 3.5    | Pipeline stage handoffs,     |
|                             |                     | summary generation           |
+-----------------------------+---------------------+------------------------------+
| Budget Guard                | DynamoDB            | Agent count + token limits   |
|                             |                     | per run                      |
+-----------------------------+---------------------+------------------------------+
| Deadlock Detection          | DynamoDB            | Delegation cycle detection   |
+-----------------------------+---------------------+------------------------------+
| Loop Detection              | DynamoDB (Art. 5)   | Per-resource edit tracking   |
+-----------------------------+---------------------+------------------------------+
| Circuit Breaker State       | DynamoDB (Art. 5)   | Shared across all agents     |
|                             |                     | in a run                     |
+-----------------------------+---------------------+------------------------------+
| Cross-Agent Observability   | LangSmith run trees | Full hierarchy per request   |
+-----------------------------+---------------------+------------------------------+
| Auth Propagation            | CredentialManager   | JWT passed to all workers    |
|                             | (Art. 5)            | via execution context        |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative       | Ollama + Docker     | All patterns testable        |
|                             | Compose             | without Bedrock costs        |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code      | Terraform           | DynamoDB tables, IAM roles   |
+-----------------------------+---------------------+------------------------------+

A Note on Where We’re Headed

There’s a problem that surfaces naturally as agent systems get more complex, and this article has touched it several times without naming it directly. The pipeline compression step, the handoff summaries, the supervisor’s inability to remember what it learned in a previous run all of these point at the same underlying issue.

Agents forget. Each run starts fresh. A supervisor that decomposed and solved a complex problem last Tuesday has no memory of it when a similar task arrives on Wednesday. A worker that learned a useful pattern for handling a tricky edge case carries that knowledge for exactly one run.

The next article is about that problem specifically: how to give agents memory that persists across runs, and what the different tiers of memory look like working memory, episodic memory, semantic memory and how to implement each of them on AWS.

Other Articles in This Series:

• Agentic Architectures — Article 1: The Agentic AI Maturity Model
• Agentic Architectures — Article 2: Advanced Coordination and Reasoning Patterns
• Agentic Architectures — Article 3: AgentOps
• Agentic Architectures — Article 4: Agentic Protocols (MCP and A2A)
• Agentic Architectures — Article 5: Harness Engineering and the Agent Runtime Layer

Tags: AgenticAI, AWSBedrock, LangGraph, MultiAgent, SoftwareArchitecture, LLMOps, AIEngineering, CloudArchitecture, PythonProgramming, ArtificialIntelligence