Chapter 11 · Multi-Agent Systems
One agent is smart. Three agents arguing about the answer are smarter. This chapter is about making agents collaborate without making a mess.
Here is a single agent, given a hard problem:
User: Design a production-ready microservice architecture for a food
delivery app. Cover data stores, service boundaries, communication
patterns, and failure modes.
Agent: Here's a solid architecture:
Services: Order Service, Restaurant Service, Delivery Service,
User Service, Payment Service.
Data: PostgreSQL for orders and users, Redis for caching,
MongoDB for restaurant menus.
Communication: REST between services, Kafka for order events.
Failure modes: Circuit breakers on all external calls, retry
with exponential backoff, dead letter queues for failed events.
This should handle your needs.
Decent. Covers the basics. But it is generic. No trade-off analysis. No security considerations. No cost model. The kind of answer that looks right in a slide deck and falls apart in production.
Now give the same problem to three agents -- an architect, a security reviewer, and a cost optimizer -- and let them debate:
ARCHITECT: I propose six services: Order, Restaurant, Delivery, User,
Payment, and Notification. PostgreSQL for transactional data,
MongoDB for restaurant catalogs, Redis for session and cache,
Elasticsearch for search. Async communication via Kafka for
order state changes. Sync REST for direct queries.
SECURITY: I see three problems. First, Payment Service handling raw card
data -- that's PCI scope you don't want. Tokenize at the edge
and pass tokens, not card numbers. Second, User Service storing
PII alongside auth credentials -- separate them. Third, no
mention of API gateway rate limiting. A scraper will take you
down in minutes.
COST: The architect's design runs about $12K/month on AWS at moderate
scale. My concerns: Elasticsearch is expensive and you don't
need it yet -- PostgreSQL full-text search handles restaurant
search fine at your scale. Kafka is overkill for a v1 -- SQS
or Redis Streams costs 80% less. And six services means six
deployments, six monitoring dashboards, six on-call rotations.
Start with three: Order, Restaurant, User. Split later.
ARCHITECT: Security's PCI point is correct. Revised: Payment Service never
sees card data. Stripe tokenization at the client. User Service
split into Auth Service (credentials only) and Profile Service
(PII). API gateway with rate limiting added. On cost: fair.
Drop Elasticsearch, use PostgreSQL full-text search. Replace
Kafka with Redis Streams for v1, with a migration path to
Kafka documented. Merge Notification into a shared event
consumer. Three core services, not six.
SECURITY: The revision addresses my concerns. One addition: the API
gateway should enforce per-user rate limits, not just global.
And auth tokens should be short-lived with refresh rotation.
COST: Three services with Redis Streams brings it to ~$3.5K/month.
That's production-ready at a reasonable burn rate. Approved.
The multi-agent answer is not just better. It is in a different category. The architect's blind spots -- security, cost -- were invisible to a single agent. The debate surfaced them, forced revision, and produced an architecture that accounts for real-world constraints. This is the power of multi-agent systems.
What This Chapter Covers
Multi-agent systems are the most powerful pattern in agentic AI -- and the easiest to get wrong. This chapter covers the orchestration patterns that work, the ones that do not, and the principles that keep multi-agent systems from descending into chaos. You will build a debate system, a hierarchical system, and a complete multi-agent research pipeline. By the end, you will know when to use multiple agents, which pattern to pick, and how to avoid the failure modes that make most multi-agent systems worse than a single well-prompted agent.
Section 1: Why Multiple Agents?
The fundamental argument is simple: specialization beats generalization. A single agent with a massive system prompt trying to do everything -- architecture, security, cost analysis, code review, documentation -- produces mediocre results across the board. The prompt becomes a novel. The agent loses the thread. Context pollution sets in: security instructions bleed into architecture reasoning, cost constraints color design decisions, and the output is a bland compromise that satisfies every constraint weakly and none strongly.
Multiple agents, each with a focused role and narrow system prompt, produce clarity. The architect thinks like an architect. The security reviewer thinks like an attacker. The cost optimizer thinks like a CFO. Each agent's prompt is short, specific, and uncontaminated by competing concerns. The quality difference is not marginal. It is categorical.
The cognitive science parallel is not an analogy -- it is the same mechanism. Human teams outperform individuals on complex tasks for the same reasons: divided attention has a cost, specialized expertise produces better reasoning within a domain, and structured debate surfaces assumptions that a single mind never questions. An agent with a 5,000-token system prompt is a generalist stretched thin. Three agents with 500-token prompts each are specialists operating at full focus.
When multi-agent helps:
- Complex analysis requiring diverse expertise (architecture + security + cost)
- Creative problem-solving where multiple perspectives generate better solutions
- Tasks benefiting from debate and verification (fact-checking, code review, strategy)
- Parallelizable work (research across multiple sources, independent sub-tasks)
- High-stakes decisions where a single agent's blind spots are unacceptable
When multi-agent hurts:
- Simple tasks where coordination overhead exceeds the benefit (a single well-prompted agent is faster and cheaper)
- Tasks requiring tight consistency across outputs (multiple agents will diverge)
- Latency-sensitive applications (sequential agent calls multiply latency)
- Budget-constrained applications (N agents = N times the API cost per turn)
The rule: start with one agent. Add a second only when you can name the specific blind spot it addresses. Add a third only when the second proves insufficient. Every agent you add must earn its seat at the table.
Section 2: Orchestration Patterns
Multi-agent systems are defined not by how many agents you have, but by how they are organized. The orchestration pattern determines information flow, decision authority, and failure modes. Here are the five patterns you will actually use.
Sequential: The Pipeline
+----------+ +----------+ +----------+
| Agent A |---->| Agent B |---->| Agent C |
+----------+ +----------+ +----------+
Each agent processes the output of the previous. No debate. No voting. Pure staged processing.
Best for: Pipelines where each stage transforms the output -- research then analyze then write, or extract then validate then format.
Code:
def sequential_pipeline(task: str, agents: list[callable]) -> str:
"""Run agents in sequence, each receiving the previous output."""
result = task
for i, agent in enumerate(agents):
print(f"\n--- Stage {i+1} ---")
result = agent(result)
print(result[:200] + "..." if len(result) > 200 else result)
return result
# Usage: research -> analyze -> write
def researcher(topic): ...
def analyst(findings): ...
def writer(analysis): ...
report = sequential_pipeline(
"Electric vehicle market trends 2024-2026",
[researcher, analyst, writer]
)
Sequential is the simplest pattern. It is also the most fragile -- an error in stage 2 poisons everything downstream. Use it when the stages are truly independent transformations, not when later stages need to question earlier ones.
Hierarchical: The Manager
+--------------+
| Manager |
+--------------+
/ | \
v v v
+----------+ +----------+ +----------+
|Specialist| |Specialist| |Specialist|
+----------+ +----------+ +----------+
\ | /
v v v
+--------------+
| Manager |
| (synthesizes) |
+--------------+
A manager agent decomposes the task, delegates to specialists, and synthesizes their outputs. Specialists never talk to each other. The manager is the single point of coordination and the single point of failure.
Best for: Complex task decomposition where sub-tasks require different expertise and the manager can judge output quality.
Code:
def hierarchical(task: str, manager, specialists: dict) -> str:
"""Manager decomposes task, delegates, synthesizes."""
# Step 1: Manager creates a plan
plan = manager(f"Create a plan to: {task}. "
f"Available specialists: {list(specialists.keys())}")
sub_tasks = parse_plan(plan) # Extract list of (specialist, sub_task)
# Step 2: Delegate to specialists
results = {}
for specialist_name, sub_task in sub_tasks:
print(f"\n--- Delegating to {specialist_name} ---")
results[specialist_name] = specialists[specialist_name](sub_task)
# Step 3: Manager synthesizes
synthesis = manager(
f"Synthesize these results into a final answer for: {task}\n\n" +
"\n\n".join(f"{name}: {output}" for name, output in results.items())
)
return synthesis
The hierarchical pattern is the workhorse of multi-agent systems. It scales well, handles heterogeneous sub-tasks, and keeps coordination centralized. The weakness: the manager is a bottleneck. If the manager makes a bad plan or a bad synthesis, the whole system fails.
Debate: The Adversarial Pattern
+----------+ +----------+ +----------+
| Agent 1 | | Agent 2 | | Agent 3 |
+----------+ +----------+ +----------+
| | |
+-------+-------+-------+-------+
| |
Critique each Revise based
other's answers on critique
| |
+-------+-------+
|
+-------------+
| Judge |
+-------------+
Multiple agents answer independently, critique each other's answers, revise, and converge. A judge (or voting) selects the final answer.
Best for: High-stakes decisions, fact-checking, creative work where multiple perspectives improve quality, any task where a single agent's blind spots are costly.
Swarm: The Parallel Pattern
+----------+ +----------+ +----------+ +----------+
| Agent 1 | | Agent 2 | | Agent 3 | | Agent N |
+----------+ +----------+ +----------+ +----------+
| | | |
+--------------+--------------+--------------+
|
+-------------+
| Aggregator |
+-------------+
Many simple agents operate independently on sub-tasks. Results are aggregated. No inter-agent communication. Pure parallelism.
Best for: Parallel research across multiple sources, data gathering, independent verification, any embarrassingly parallel task.
Code:
from concurrent.futures import ThreadPoolExecutor, as_completed
def swarm(task: str, agents: list[callable], aggregator: callable) -> str:
"""Run all agents in parallel, aggregate results."""
with ThreadPoolExecutor() as executor:
futures = {executor.submit(agent, task): agent.__name__
for agent in agents}
results = {}
for future in as_completed(futures):
name = futures[future]
results[name] = future.result()
print(f"--- {name} complete ---")
return aggregator(results)
Voting: The Consensus Pattern
+----------+ +----------+ +----------+
| Agent 1 | | Agent 2 | | Agent N |
+----------+ +----------+ +----------+
| | |
+--------------+--------------+
|
+-------------+
| Aggregator | --> Majority or consensus
+-------------+
N agents answer independently. Majority vote or consensus determines the result. No debate. No revision. Just independent answers and a tally.
Best for: Reducing hallucination, increasing reliability on factual questions, classification tasks, any task where independence is more important than iteration.
Code:
def voting(task: str, agents: list[callable], threshold: float = 0.5) -> str:
"""Run N agents, return majority answer if it meets threshold."""
answers = [agent(task) for agent in agents]
from collections import Counter
tally = Counter(answers)
winner, count = tally.most_common(1)[0]
if count / len(answers) >= threshold:
return f"CONSENSUS ({count}/{len(agents)}): {winner}"
else:
return f"NO CONSENSUS. Votes: {dict(tally)}"
Pattern selection is the most important decision you will make in a multi-agent system. Pick the wrong pattern and you get chaos, cost overruns, or worse results than a single agent. The pattern is not about how many agents you have. It is about how information flows and who has authority.
Section 3: The Debate Pattern -- Deep Dive
Debate is the most powerful multi-agent pattern for answer quality. It works because it forces agents to confront their blind spots. A single agent produces an answer and stops. In a debate, that answer gets attacked by agents with different perspectives, and the original agent must either defend or revise. Weak reasoning collapses. Strong reasoning survives. The final answer is battle-tested.
How Debate Works
- Pose the question to all agents simultaneously.
- Answer: Each agent produces an answer with reasoning.
- Critique: Each agent reviews every other agent's answer and provides critique.
- Revise: Each agent revises their answer based on the critiques they received.
- Judge: A judge agent (or voting) selects the best final answer.
Complete Implementation
import openai
from dataclasses import dataclass
from typing import Optional
@dataclass
class DebateAgent:
name: str
role: str
model: str = "gpt-4o"
def answer(self, question: str) -> str:
"""Produce an initial answer with reasoning."""
client = openai.OpenAI()
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
f"You are {self.name}, a {self.role}. "
f"Answer the question with clear reasoning. "
f"Be specific. Cite evidence where possible. "
f"Consider trade-offs. Do not hedge -- take a position."
)},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content
def critique(self, question: str, my_answer: str,
other_name: str, other_answer: str) -> str:
"""Critique another agent's answer."""
client = openai.OpenAI()
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
f"You are {self.name}, a {self.role}. "
f"You are reviewing {other_name}'s answer to the same "
f"question you answered. Identify specific weaknesses, "
f"missing considerations, factual errors, or flawed "
f"assumptions. Be direct. Be specific. Your goal is to "
f"improve the final answer, not to be polite."
)},
{"role": "user", "content": (
f"Question: {question}\n\n"
f"Your answer was:\n{my_answer}\n\n"
f"{other_name}'s answer:\n{other_answer}\n\n"
f"Provide your critique of {other_name}'s answer."
)}
]
)
return response.choices[0].message.content
def revise(self, question: str, original_answer: str,
critiques: list[str]) -> str:
"""Revise answer based on critiques received."""
client = openai.OpenAI()
critiques_text = "\n\n".join(
f"Critique {i+1}:\n{c}" for i, c in enumerate(critiques)
)
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
f"You are {self.name}, a {self.role}. "
f"You received critiques of your answer. "
f"Address valid criticisms. Reject invalid ones. "
f"Produce a revised, stronger answer. "
f"If a critique is correct, incorporate it. "
f"If it is wrong, explain why you stand by your position."
)},
{"role": "user", "content": (
f"Question: {question}\n\n"
f"Your original answer:\n{original_answer}\n\n"
f"Critiques you received:\n{critiques_text}\n\n"
f"Produce your revised answer."
)}
]
)
return response.choices[0].message.content
class DebateOrchestrator:
def __init__(self, agents: list[DebateAgent],
judge_model: str = "gpt-4o"):
self.agents = agents
self.judge_model = judge_model
def run(self, question: str, max_rounds: int = 2) -> dict:
"""Run a full debate and return the results."""
trace = {"question": question, "rounds": []}
# Round 1: Initial answers
print("\n" + "="*60)
print("ROUND 1: INITIAL ANSWERS")
print("="*60)
initial_answers = {}
for agent in self.agents:
print(f"\n--- {agent.name} ({agent.role}) ---")
answer = agent.answer(question)
initial_answers[agent.name] = answer
print(answer[:300] + "..." if len(answer) > 300 else answer)
trace["rounds"].append({"phase": "initial", "answers": initial_answers})
for round_num in range(1, max_rounds + 1):
# Critique phase
print(f"\n{'='*60}")
print(f"ROUND {round_num+1}: CRITIQUE")
print(f"{'='*60}")
all_critiques = {a.name: [] for a in self.agents}
for critic in self.agents:
for target in self.agents:
if critic.name == target.name:
continue
print(f"\n--- {critic.name} critiques {target.name} ---")
critique = critic.critique(
question,
initial_answers[critic.name],
target.name,
initial_answers[target.name]
)
all_critiques[target.name].append(
f"[From {critic.name}]: {critique}"
)
print(critique[:200] + "..."
if len(critique) > 200 else critique)
# Revise phase
print(f"\n{'='*60}")
print(f"ROUND {round_num+1}: REVISION")
print(f"{'='*60}")
revised_answers = {}
for agent in self.agents:
print(f"\n--- {agent.name} revises ---")
revised = agent.revise(
question,
initial_answers[agent.name],
all_critiques[agent.name]
)
revised_answers[agent.name] = revised
print(revised[:300] + "..." if len(revised) > 300 else revised)
trace["rounds"].append({
"phase": f"debate_round_{round_num}",
"critiques": all_critiques,
"revised": revised_answers
})
initial_answers = revised_answers
# Judge phase
print(f"\n{'='*60}")
print("JUDGE: SELECTING BEST ANSWER")
print(f"{'='*60}")
verdict = self._judge(question, initial_answers)
trace["verdict"] = verdict
print(f"\nVERDICT: {verdict}")
return trace
def _judge(self, question: str, final_answers: dict) -> str:
"""Judge selects the best final answer."""
client = openai.OpenAI()
answers_text = "\n\n---\n\n".join(
f"{name}'s final answer:\n{answer}"
for name, answer in final_answers.items()
)
response = client.chat.completions.create(
model=self.judge_model,
messages=[
{"role": "system", "content": (
"You are a judge evaluating multiple answers to the same "
"question. Select the best answer based on: accuracy, "
"completeness, specificity, and quality of reasoning. "
"Name the winner and explain your reasoning. "
"If multiple answers are equally good, say so and explain."
)},
{"role": "user", "content": (
f"Question: {question}\n\n{answers_text}\n\n"
f"Which answer is best? Name the winner and explain why."
)}
]
)
return response.choices[0].message.content
Running the Debate
Here is the debate system run on a hard question: "What is the best database for a social media app with 100M users?"
agents = [
DebateAgent("Alice", "database architect specializing in high-scale systems"),
DebateAgent("Bob", "performance engineer focused on query optimization"),
DebateAgent("Carol", "DevOps engineer specializing in operational costs"),
]
orchestrator = DebateOrchestrator(agents)
result = orchestrator.run(
"What is the best database for a social media app with 100M users? "
"The app has: user profiles, posts with text and media, a social graph "
"(followers/following), a newsfeed, real-time messaging, and search. "
"Consider read/write patterns, consistency requirements, operational "
"complexity, and cost."
)
The debate transcript (condensed):
ROUND 1: INITIAL ANSWERS
Alice (Architect): PostgreSQL with read replicas for the core relational
data (users, profiles, social graph). The social graph is fundamentally
a graph problem, but at 100M users, a dedicated graph DB adds operational
complexity without enough benefit -- PostgreSQL recursive CTEs handle
follower queries at this scale. For the newsfeed, use Redis sorted sets
for the hot feed (last 7 days) and Cassandra for the cold feed. Messages
go to Cassandra for write throughput. Search: Elasticsearch. Total: 4 data
stores. This is the standard pattern at this scale.
Bob (Performance): You need Cassandra as the primary store. Social media
is write-heavy (posts, likes, messages) and read-heavy (newsfeed, timeline)
with simple access patterns -- exactly what Cassandra is designed for.
PostgreSQL will choke on the newsfeed fan-out writes at 100M users. A
single user with 50M followers posting one update means 50M rows written.
PostgreSQL cannot do that. Cassandra can. Use Elasticsearch for search
and Redis for caching hot data. Three data stores, not four.
Carol (Ops): Start with PostgreSQL for everything. At 100M users you
have a team of maybe 20-50 engineers. Four data stores means four
teams to run them, four failure modes, four backup strategies, four
migration paths. The operational cost of a multi-store architecture
often exceeds the infrastructure cost. PostgreSQL with careful schema
design, read replicas, and aggressive caching handles far more than
people think. Add Elasticsearch only when search actually breaks.
Add Cassandra only when write throughput actually breaks. Premature
optimization is the most expensive kind.
ROUND 2: CRITIQUE
Alice critiques Bob: "Cassandra as primary is a mistake. The social
graph requires JOINs -- 'show me posts from people I follow' is a JOIN
between follows and posts. Cassandra cannot JOIN. You would need to
denormalize everything, which means every follow/unfollow rewrites
massive amounts of data. Cassandra is great for the newsfeed and
messages. It is terrible for the social graph."
Bob critiques Alice: "PostgreSQL recursive CTEs for follower queries
at 100M users? That query will run for seconds, not milliseconds.
The newsfeed needs to render in under 200ms. You are hand-waving
past the performance problem. Also, four data stores is operationally
irresponsible for a team that is not Google."
Carol critiques Alice: "Four data stores. Four. Each one needs
monitoring, backup, failover, upgrades, security patches, and an
expert on the team who understands its failure modes. Your architecture
is technically correct and operationally dangerous."
Alice critiques Carol: "Starting with PostgreSQL for everything and
'adding later' is how you get a monolith that cannot be split. The
migration from 'PostgreSQL for everything' to 'PostgreSQL + Cassandra
+ Elasticsearch' while serving 100M users is a multi-year project
that will cause multiple outages. The architecture needs to be
designed for the target scale, not the starting scale."
ROUND 3: REVISION
Alice (revised): "Bob is right about newsfeed latency -- recursive CTEs
won't cut it for the hot path. Revised: PostgreSQL for users, profiles,
and the social graph (with materialized views for follower counts and
common queries). Redis sorted sets for the hot newsfeed (pre-computed
on write, not computed on read). Cassandra for messages and the cold
newsfeed archive. Elasticsearch for search. Four stores, but each has
a clear, non-overlapping responsibility. The social graph stays in
PostgreSQL because the write pattern (follow/unfollow) is low-frequency
and the read pattern benefits from JOINs."
Bob (revised): "Alice convinced me on the social graph -- Cassandra
is wrong for JOIN-heavy queries. Revised: Cassandra for posts, likes,
and messages (high-volume, simple access patterns). PostgreSQL for
users, profiles, and the social graph (relational, JOIN-heavy).
Redis for the hot newsfeed and session data. Elasticsearch for search.
This is four stores, but each is the right tool for its specific job."
Carol (revised): "I was too conservative. At 100M users, you cannot
run everything on PostgreSQL. But Alice and Bob are both proposing
four data stores, which is still too many for most teams. My revised
position: PostgreSQL for users, profiles, social graph. Cassandra for
posts, likes, messages, and newsfeed materialization. Elasticsearch
for search. Three stores. Redis is a cache, not a data store -- treat
it as infrastructure, not a separate system. Three data stores with
clear boundaries. This is the Instagram model circa 2015 and it worked."
JUDGE VERDICT: Carol's revised answer is the best. It balances technical
correctness with operational reality. Three data stores with clear
boundaries is the right number for a team of 20-50 engineers. Alice
and Bob both converged on four stores, which is technically optimal
but operationally excessive. Carol's "Redis is infrastructure, not a
data store" framing is the key insight -- it reduces the perceived
complexity without reducing capability. Winner: Carol.
The debate produced a better answer than any single agent. Alice's initial answer was technically sound but operationally naive. Bob's was performance-focused but wrong about the social graph. Carol's was operationally wise but too conservative. The debate forced each to confront their weaknesses, and the final answer -- Carol's revised position -- was stronger than any initial answer.
Why debate works: It surfaces blind spots that a single agent cannot see. It challenges assumptions. It forces rigor. The final answer is not the smartest agent's answer. It is the answer that survived attack from multiple perspectives.
Section 4: The Hierarchical Pattern -- Deep Dive
The hierarchical pattern is the workhorse for complex, multi-step tasks. A manager agent decomposes the task, delegates to specialists, evaluates their outputs, and synthesizes a final result. If the synthesis is insufficient, the manager re-plans and re-delegates. This is the pattern behind most production multi-agent systems.
How Hierarchy Works
- Plan: Manager receives the task and creates a plan with sub-tasks.
- Delegate: Manager assigns each sub-task to the appropriate specialist.
- Execute: Specialists execute their sub-tasks and return results.
- Evaluate: Manager evaluates each result. If any are insufficient, re-delegate with more specific instructions.
- Synthesize: Manager combines all results into a final output.
Complete Implementation
import openai
import json
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Specialist:
name: str
role: str
model: str = "gpt-4o"
def execute(self, task: str) -> str:
client = openai.OpenAI()
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
f"You are {self.name}, a {self.role}. "
f"Execute the assigned task thoroughly. "
f"Provide specific, actionable output. "
f"Include data, examples, and citations where relevant. "
f"Do not summarize -- produce the full deliverable."
)},
{"role": "user", "content": task}
]
)
return response.choices[0].message.content
class Manager:
def __init__(self, model: str = "gpt-4o"):
self.model = model
def plan(self, task: str, specialists: dict[str, Specialist]) -> list[dict]:
"""Decompose task into sub-tasks with specialist assignments."""
client = openai.OpenAI()
specialist_list = "\n".join(
f"- {name}: {spec.role}" for name, spec in specialists.items()
)
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
"You are a project manager. Decompose the given task "
"into sub-tasks and assign each to the most appropriate "
"specialist. Output a JSON array of objects with keys: "
"'specialist' (name), 'sub_task' (detailed instructions "
"for the specialist), 'priority' (1=highest). "
"Each sub-task should be self-contained and produce a "
"complete deliverable. Order by dependency."
)},
{"role": "user", "content": (
f"Task: {task}\n\n"
f"Available specialists:\n{specialist_list}\n\n"
f"Output the plan as a JSON array."
)}
]
)
plan_text = response.choices[0].message.content
# Extract JSON from response (handle markdown code blocks)
if "```" in plan_text:
plan_text = plan_text.split("```")[1]
if plan_text.startswith("json"):
plan_text = plan_text[4:]
return json.loads(plan_text.strip())
def evaluate(self, sub_task: str, result: str) -> tuple[bool, str]:
"""Evaluate a specialist's output. Returns (acceptable, feedback)."""
client = openai.OpenAI()
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
"You are a quality reviewer. Evaluate whether the "
"deliverable meets the requirements. Be strict. "
"Respond with a JSON object: "
'{"acceptable": true/false, "feedback": "..."}'
)},
{"role": "user", "content": (
f"Sub-task:\n{sub_task}\n\n"
f"Deliverable:\n{result}\n\n"
f"Is this acceptable? If not, what is missing?"
)}
]
)
eval_text = response.choices[0].message.content
if "```" in eval_text:
eval_text = eval_text.split("```")[1]
if eval_text.startswith("json"):
eval_text = eval_text[4:]
evaluation = json.loads(eval_text.strip())
return evaluation["acceptable"], evaluation["feedback"]
def synthesize(self, task: str, results: dict[str, str]) -> str:
"""Synthesize all specialist outputs into a final deliverable."""
client = openai.OpenAI()
results_text = "\n\n---\n\n".join(
f"[{name}]:\n{output}" for name, output in results.items()
)
response = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": (
"You are a senior editor. Synthesize the following "
"specialist outputs into a single, coherent, "
"comprehensive final deliverable. Resolve any "
"contradictions. Fill any gaps. The final output "
"should read as one unified document, not a "
"collection of separate reports."
)},
{"role": "user", "content": (
f"Original task: {task}\n\n"
f"Specialist outputs:\n{results_text}\n\n"
f"Produce the final synthesized deliverable."
)}
]
)
return response.choices[0].message.content
class HierarchicalOrchestrator:
def __init__(self, manager: Manager,
specialists: dict[str, Specialist],
max_retries: int = 2):
self.manager = manager
self.specialists = specialists
self.max_retries = max_retries
def run(self, task: str) -> dict:
"""Execute the full hierarchical workflow."""
trace = {"task": task, "plan": None, "executions": [], "synthesis": None}
# Phase 1: Plan
print("\n" + "="*60)
print("PHASE 1: MANAGER CREATES PLAN")
print("="*60)
plan = self.manager.plan(task, self.specialists)
trace["plan"] = plan
for item in plan:
print(f" [{item['priority']}] {item['specialist']}: "
f"{item['sub_task'][:100]}...")
# Phase 2: Execute with retry
results = {}
for item in sorted(plan, key=lambda x: x["priority"]):
spec_name = item["specialist"]
sub_task = item["sub_task"]
specialist = self.specialists[spec_name]
for attempt in range(1, self.max_retries + 2):
print(f"\n{'='*60}")
print(f"PHASE 2: {spec_name} (attempt {attempt})")
print(f"{'='*60}")
output = specialist.execute(sub_task)
print(output[:300] + "..." if len(output) > 300 else output)
acceptable, feedback = self.manager.evaluate(sub_task, output)
trace["executions"].append({
"specialist": spec_name,
"attempt": attempt,
"output": output,
"acceptable": acceptable,
"feedback": feedback
})
if acceptable:
print(f"\n [ACCEPTED]")
results[spec_name] = output
break
else:
print(f"\n [REJECTED] {feedback[:150]}...")
sub_task = (f"{sub_task}\n\nYour previous output was "
f"rejected. Feedback: {feedback}\n"
f"Address this feedback in your revision.")
# Phase 3: Synthesize
print(f"\n{'='*60}")
print("PHASE 3: MANAGER SYNTHESIZES")
print(f"{'='*60}")
synthesis = self.manager.synthesize(task, results)
trace["synthesis"] = synthesis
print(synthesis[:500] + "..." if len(synthesis) > 500 else synthesis)
return trace
Running the Hierarchical System
specialists = {
"researcher": Specialist("Researcher",
"market research analyst who finds data, trends, and statistics"),
"analyst": Specialist("Analyst",
"strategic analyst who identifies patterns, opportunities, and threats"),
"writer": Specialist("Writer",
"business writer who produces clear, compelling reports"),
}
manager = Manager()
orchestrator = HierarchicalOrchestrator(manager, specialists)
result = orchestrator.run(
"Write a comprehensive market analysis of the electric vehicle "
"industry. Cover: market size and growth, key players and market "
"share, technology trends (battery, autonomous driving, charging "
"infrastructure), regulatory landscape, consumer adoption trends, "
"and 5-year outlook. Target audience: institutional investors."
)
The execution trace (condensed):
PHASE 1: MANAGER CREATES PLAN
[1] researcher: Research EV market: size, growth rates, key players,
market share data, technology trends, regulatory changes...
[2] analyst: Analyze research findings: identify strategic patterns,
competitive dynamics, investment implications, risk factors...
[3] writer: Synthesize analysis into investor-grade report with
executive summary, data visualizations, and recommendations...
PHASE 2: researcher (attempt 1)
[Produces 15-page research document with market data from BloombergNEF,
IEA, company filings. Covers global EV sales (14M units in 2023, +35%
YoY), market share (BYD 20%, Tesla 13%, VW 8%), battery technology
(LFP vs NMC, solid-state timeline), charging infrastructure (US$100B
needed globally by 2030), regulatory (EU 2035 ban, US IRA incentives),
consumer adoption (price parity expected 2025-2027).]
[ACCEPTED]
PHASE 2: analyst (attempt 1)
[Produces strategic analysis: BYD's vertical integration is the
underappreciated moat. Tesla's valuation depends on FSD, not cars.
Legacy automakers face the "innovator's dilemma" -- their ICE
profits fund the transition but also create organizational
resistance. Battery supply chain is the bottleneck -- China controls
70% of refining capacity. Regulatory risk is asymmetric: stricter
rules help incumbents who already comply.]
[ACCEPTED]
PHASE 2: writer (attempt 1)
[Produces draft report with executive summary, five sections, and
investment thesis. But the executive summary is too long and the
recommendations lack specificity.]
[REJECTED] Executive summary should be 3-4 paragraphs max. Investment
recommendations need specific tickers, price targets, and catalysts...
PHASE 2: writer (attempt 2)
[Revised report with tight executive summary, specific investment
recommendations (BYD for battery supply chain exposure, Tesla as
FSD option-value play, lithium miners for commodity exposure),
risk factors, and 5-year scenario analysis.]
[ACCEPTED]
PHASE 3: MANAGER SYNTHESIZES
[Produces final 12-page investor report combining all three outputs
into a single coherent document with consistent voice, cross-
referenced data, and a clear narrative arc.]
The hierarchical pattern produced a complete, investor-grade report that no single agent could have produced in one pass. The manager decomposed the task, the specialists each did their focused work, the writer was sent back for revision, and the final synthesis was coherent and comprehensive.
The manager is the system's intelligence. A good manager produces a good plan, delegates to the right specialists, evaluates output honestly, and synthesizes without losing information. A bad manager produces a bad plan, delegates to the wrong specialists, accepts mediocre output, and produces a synthesis that is worse than any individual contribution. Invest in your manager prompt.
Section 5: Agent Communication
Agents need to talk to each other. How they talk determines what information flows where, who sees what, and how the system handles contention. There are three communication architectures, and you will use all three in different contexts.
Shared Message Bus
+----------+ +----------+ +----------+
| Agent A | | Agent B | | Agent C |
+----------+ +----------+ +----------+
| | |
+--------------+---------------+
|
+---------------+
| Message Bus |
| (all messages)|
+---------------+
All agents read from and write to a shared conversation. Every agent sees every message. This is the simplest architecture and the default for most multi-agent systems.
class MessageBus:
def __init__(self):
self.messages: list[dict] = []
def post(self, sender: str, content: str,
msg_type: str = "message"):
self.messages.append({
"sender": sender,
"type": msg_type,
"content": content,
"timestamp": len(self.messages)
})
def get_all(self) -> list[dict]:
return self.messages
def get_since(self, index: int) -> list[dict]:
return self.messages[index:]
def get_by_sender(self, sender: str) -> list[dict]:
return [m for m in self.messages if m["sender"] == sender]
# Usage in an agent loop
bus = MessageBus()
def agent_with_bus(name: str, role: str, bus: MessageBus):
client = openai.OpenAI()
history = bus.get_all()
history_text = "\n".join(
f"[{m['sender']}]: {m['content'][:200]}" for m in history
)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": (
f"You are {name}, {role}. You see all messages on the "
f"shared bus. Respond to the current situation.\n\n"
f"Conversation so far:\n{history_text}"
)},
{"role": "user", "content": "What is your next contribution?"}
]
)
reply = response.choices[0].message.content
bus.post(name, reply)
return reply
The shared bus is simple and transparent. Every agent has full context. The downside: context pollution. Agents see information irrelevant to their role, which can distract or confuse them. Use the shared bus when agents need full context and the conversation is short. Switch to direct messaging when context needs to be scoped.
Direct Messaging
+----------+ msg +----------+ msg +----------+
| Agent A |------->| Agent B |------->| Agent C |
+----------+ +----------+ +----------+
^ |
| msg |
+---------------------------------------+
Agents send messages to specific recipients. Each agent only sees messages addressed to it. This is the architecture for hierarchical and sequential patterns.
class DirectMessaging:
def __init__(self):
self.inboxes: dict[str, list[dict]] = {}
def send(self, sender: str, recipient: str, content: str,
msg_type: str = "message"):
if recipient not in self.inboxes:
self.inboxes[recipient] = []
self.inboxes[recipient].append({
"sender": sender,
"recipient": recipient,
"type": msg_type,
"content": content
})
def receive(self, agent_name: str) -> list[dict]:
return self.inboxes.get(agent_name, [])
def receive_unread(self, agent_name: str,
last_read: int = 0) -> list[dict]:
inbox = self.inboxes.get(agent_name, [])
return inbox[last_read:]
# Usage: Manager delegates to specialists
comm = DirectMessaging()
def manager_with_dm(task: str, specialists: list[str],
comm: DirectMessaging):
# Delegate sub-tasks
for i, spec in enumerate(specialists):
comm.send("manager", spec,
f"Sub-task {i+1}: Research {task} from angle {i+1}")
# Collect results
results = {}
for spec in specialists:
msgs = comm.receive("manager")
# In practice, specialists would send results back to manager
results[spec] = msgs[-1]["content"] if msgs else "No response"
return results
Direct messaging keeps context clean. Each agent only sees what is relevant. The downside: agents lose the broader context. A specialist working on sub-task 3 does not know what specialists 1 and 2 discovered, which can lead to duplicated work or contradictory conclusions. Mitigation: the manager includes relevant context from other sub-tasks when delegating.
Blackboard
+---------------------------------------------------+
| BLACKBOARD |
| shared_state = { |
| "findings": [...], |
| "hypotheses": [...], |
| "decisions": [...], |
| "current_phase": "analysis" |
| } |
+---------------------------------------------------+
^ ^ ^ ^
| | | |
+----------+ +----------+ +----------+ +----------+
| Agent A | | Agent B | | Agent C | | Agent D |
+----------+ +----------+ +----------+ +----------+
Agents read from and write to a shared state object. No direct messaging. Each agent checks the blackboard, does its work, and updates the blackboard. This is the architecture for swarm and collaborative patterns.
import threading
class Blackboard:
def __init__(self):
self.state: dict = {}
self.lock = threading.Lock()
self.history: list[dict] = []
def read(self, key: str, default=None):
with self.lock:
return self.state.get(key, default)
def write(self, key: str, value, agent: str):
with self.lock:
old = self.state.get(key)
self.state[key] = value
self.history.append({
"agent": agent,
"key": key,
"old_value": old,
"new_value": value
})
def append(self, key: str, value, agent: str):
"""Append to a list-valued key."""
with self.lock:
if key not in self.state:
self.state[key] = []
self.state[key].append(value)
self.history.append({
"agent": agent,
"key": key,
"action": "append",
"value": value
})
def snapshot(self) -> dict:
with self.lock:
return dict(self.state)
# Usage: Swarm of researchers writing findings to a shared blackboard
blackboard = Blackboard()
def researcher_agent(name: str, topic: str, blackboard: Blackboard):
client = openai.OpenAI()
# Read what others have found
existing = blackboard.read("findings", [])
existing_text = "\n".join(
f"- {f}" for f in existing[-5:] # Last 5 findings for context
)
response = client.chat.completions.create(
model="gpt-4o-mini", # Cheaper model for swarm agents
messages=[
{"role": "system", "content": (
f"You are {name}, a researcher. Find one specific, "
f"novel insight about {topic} that is not already "
f"covered. Existing findings:\n{existing_text}"
)},
{"role": "user", "content": f"Research: {topic}"}
]
)
finding = response.choices[0].message.content
blackboard.append("findings", finding, name)
return finding
The blackboard is the most flexible architecture. Agents are decoupled -- they do not need to know about each other. The blackboard is the single source of truth. The downside: no coordination. Two agents might work on the same sub-problem. The blackboard can become a bottleneck if many agents write simultaneously. Use it for parallel, independent work where coordination is not critical.
The Telephone Game Problem
Information degrades as it passes through multiple agents. Agent A summarizes its findings for Agent B. Agent B summarizes that summary for Agent C. By the time Agent C acts, the information has been compressed, simplified, and distorted. This is the "telephone game" problem, and it is the most common failure mode in sequential multi-agent systems.
Mitigations:
- Keep original sources accessible. Do not pass summaries -- pass references. Agent C should be able to read Agent A's original output, not just Agent B's summary of it.
- Structured handoffs. Instead of free-text summaries, use structured formats:
{"findings": [...], "confidence": 0.85, "sources": [...], "gaps": [...]}. Structure resists degradation better than prose. - Verification steps. After a handoff, have the receiving agent restate what it understood and have the sending agent confirm. This catches degradation early.
- Limit chain length. Three agents in sequence is the practical maximum. Beyond that, information loss is inevitable regardless of mitigations.
Section 6: Multi-Agent Pitfalls
Multi-agent systems fail in predictable ways. Here are the five failure modes and how to prevent them.
1. Infinite Debate
Agents never converge. Round 1 produces answers. Round 2 produces critiques. Round 3 produces revisions. Round 4 produces counter-critiques. The debate continues until you run out of API credits.
Fix: Hard cap on debate rounds (2 is usually enough). A judge with final authority. If agents have not converged by the cap, the judge picks the best answer and the system moves on. Debate is a tool for improving answers, not a philosophical exercise.
2. Groupthink
Agents agree too quickly, especially when they use the same model. Three GPT-4o agents with similar prompts will produce similar answers, similar critiques, and similar revisions. The debate looks productive but produces no real improvement.
Fix: Diverse prompts that assign genuinely different perspectives. Different models for different agents (GPT-4o for the architect, Claude for the security reviewer, Gemini for the cost optimizer). Adversarial roles that explicitly require agents to find flaws. If all agents agree on round 1, your prompts are not diverse enough.
3. Cost Explosion
Five agents times ten turns each equals fifty LLM calls. At GPT-4o pricing, that is real money. Multi-agent systems multiply your API costs by the number of agents times the number of turns.
Fix: Use cheaper models for simple roles (GPT-4o-mini for swarm agents, GPT-4o only for the manager and judge). Limit turns aggressively. Cache results -- if the same sub-task appears twice, reuse the cached output. Track cost per task and set budgets. A simple cost tracker:
class CostTracker:
def __init__(self):
self.calls = 0
self.total_input_tokens = 0
self.total_output_tokens = 0
# Approximate pricing per 1M tokens
self.pricing = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
}
def record(self, model: str, input_tokens: int,
output_tokens: int):
self.calls += 1
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
def cost(self) -> float:
# Simplified -- in practice, track per-model
return (self.total_input_tokens / 1_000_000 * 2.50 +
self.total_output_tokens / 1_000_000 * 10.00)
def report(self) -> str:
return (f"Calls: {self.calls}, "
f"Input: {self.total_input_tokens:,} tokens, "
f"Output: {self.total_output_tokens:,} tokens, "
f"Est. cost: ${self.cost():.2f}")
4. Coordination Overhead
Agents spend more time coordinating than working. The manager writes a 500-word plan. Each specialist reads the plan, asks clarifying questions, and waits for responses. The manager re-plans based on specialist feedback. Three rounds of coordination before any real work happens.
Fix: Clear, minimal protocols. The manager's plan should be a list of (specialist, task) pairs, not a strategy document. Specialists should execute, not negotiate. If a specialist needs clarification, it should make a reasonable assumption and proceed, flagging the assumption in its output. Coordination is overhead -- minimize it.
5. The "Too Many Cooks" Problem
More agents does not mean better results. Three well-prompted agents with clear roles outperform ten agents with vague roles. Every additional agent adds coordination overhead, cost, and the risk of contradictory outputs.
Fix: Start with two agents. Add a third only when you can name the specific gap it fills. Add a fourth only when the third proves insufficient. Every agent must earn its seat. The test: remove the agent and see if quality drops. If it does not, the agent was dead weight.
Section 7: Building a Multi-Agent Research System from Scratch
You now understand the patterns, the communication architectures, and the pitfalls. Let us build a complete multi-agent research system. This is not a toy. It is the architecture you would use to build a production research assistant.
System Design
+------------------+
| Orchestrator |
| (plans, manages) |
+------------------+
/ | | \
v v v v
+--------+ +--------+ +--------+ +----------+
|Searcher| |Searcher| |Analyst | | Writer |
| #1 | | #2 | | | | |
+--------+ +--------+ +--------+ +----------+
\ | | /
v v v v
+------------------+
| Reviewer |
| (fact-checks, |
| improves) |
+------------------+
Five agents. The Orchestrator plans and manages. Two Searcher agents work in parallel (swarm pattern) to gather information from different angles. The Analyst identifies patterns and gaps. The Writer synthesizes into a report. The Reviewer fact-checks and improves the report.
Complete Code
import openai
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import Optional
# --- Agent Definitions ---
@dataclass
class ResearchAgent:
name: str
role: str
model: str
def run(self, task: str, context: str = "") -> str:
client = openai.OpenAI()
messages = [
{"role": "system", "content": (
f"You are {self.name}, {self.role}. "
f"{context}\n"
f"Be thorough. Be specific. Cite sources where possible. "
f"Produce a complete deliverable, not a summary."
)},
{"role": "user", "content": task}
]
response = client.chat.completions.create(
model=self.model, messages=messages
)
return response.choices[0].message.content
class ResearchOrchestrator:
def __init__(self):
# Use cheaper models for parallel searchers
self.searchers = [
ResearchAgent("Searcher-Alpha",
"web researcher who finds specific facts, data, and sources",
"gpt-4o-mini"),
ResearchAgent("Searcher-Beta",
"web researcher who finds alternative perspectives, "
"contrarian views, and underreported angles",
"gpt-4o-mini"),
]
self.analyst = ResearchAgent("Analyst",
"strategic analyst who identifies patterns, contradictions, "
"gaps, and implications in research data",
"gpt-4o")
self.writer = ResearchAgent("Writer",
"technical writer who produces clear, well-structured, "
"authoritative reports",
"gpt-4o")
self.reviewer = ResearchAgent("Reviewer",
"fact-checker and editor who verifies claims, identifies "
"weak arguments, and improves clarity and accuracy",
"gpt-4o")
self.manager_model = "gpt-4o"
def run(self, question: str) -> dict:
trace = {"question": question, "phases": {}}
# Phase 1: Orchestrator creates research plan
print("\n" + "="*60)
print("PHASE 1: RESEARCH PLAN")
print("="*60)
plan = self._create_plan(question)
trace["phases"]["plan"] = plan
print(json.dumps(plan, indent=2))
# Phase 2: Parallel search
print(f"\n{'='*60}")
print("PHASE 2: PARALLEL SEARCH")
print(f"{'='*60}")
search_results = self._parallel_search(plan)
trace["phases"]["search_results"] = search_results
for name, result in search_results.items():
print(f"\n--- {name} ---")
print(result[:300] + "..." if len(result) > 300 else result)
# Phase 3: Analysis
print(f"\n{'='*60}")
print("PHASE 3: ANALYSIS")
print(f"{'='*60}")
analysis = self._analyze(question, search_results)
trace["phases"]["analysis"] = analysis
print(analysis[:400] + "..." if len(analysis) > 400 else analysis)
# Phase 4: Draft report
print(f"\n{'='*60}")
print("PHASE 4: DRAFT REPORT")
print(f"{'='*60}")
draft = self._write(question, search_results, analysis)
trace["phases"]["draft"] = draft
print(draft[:400] + "..." if len(draft) > 400 else draft)
# Phase 5: Review and revise
print(f"\n{'='*60}")
print("PHASE 5: REVIEW")
print(f"{'='*60}")
review = self._review(question, draft, search_results)
trace["phases"]["review"] = review
print(review[:400] + "..." if len(review) > 400 else review)
# Phase 6: Final report
print(f"\n{'='*60}")
print("PHASE 6: FINAL REPORT")
print(f"{'='*60}")
final = self._revise(draft, review)
trace["phases"]["final_report"] = final
print(final[:500] + "..." if len(final) > 500 else final)
return trace
def _create_plan(self, question: str) -> dict:
client = openai.OpenAI()
response = client.chat.completions.create(
model=self.manager_model,
messages=[
{"role": "system", "content": (
"You are a research director. Create a research plan "
"for the given question. Output JSON with: "
"'angles' (list of 2-3 research angles to investigate "
"in parallel), 'analysis_focus' (what patterns to look "
"for), 'report_structure' (sections for the final "
"report)."
)},
{"role": "user", "content": question}
]
)
text = response.choices[0].message.content
if "```" in text:
text = text.split("```")[1]
if text.startswith("json"):
text = text[4:]
return json.loads(text.strip())
def _parallel_search(self, plan: dict) -> dict:
results = {}
with ThreadPoolExecutor() as executor:
futures = {}
for i, angle in enumerate(plan.get("angles", [])):
searcher = self.searchers[i % len(self.searchers)]
future = executor.submit(
searcher.run,
f"Research this angle thoroughly: {angle}",
f"You are a {searcher.role}. Find specific facts, "
f"data points, statistics, and sources."
)
futures[future] = f"{searcher.name}: {angle[:60]}"
for future in as_completed(futures):
label = futures[future]
results[label] = future.result()
return results
def _analyze(self, question: str, search_results: dict) -> str:
combined = "\n\n---\n\n".join(
f"{label}:\n{content}" for label, content in search_results.items()
)
return self.analyst.run(
f"Analyze these research findings for the question: {question}\n\n"
f"Identify: key patterns, contradictions between sources, "
f"gaps in the research, implications, and the strongest "
f"evidence-based conclusions.\n\nFindings:\n{combined}",
"You are a strategic analyst. Be incisive. Identify what "
"matters and what does not."
)
def _write(self, question: str, search_results: dict,
analysis: str) -> str:
combined_findings = "\n\n".join(
f"{label}:\n{content}" for label, content in search_results.items()
)
return self.writer.run(
f"Write a comprehensive report answering: {question}\n\n"
f"Research findings:\n{combined_findings}\n\n"
f"Analysis:\n{analysis}\n\n"
f"Structure: Executive Summary, Background, Key Findings, "
f"Analysis, Implications, Conclusion. Be authoritative. "
f"Cite specific data points. Acknowledge uncertainty where "
f"it exists.",
"You are a technical writer producing an authoritative report."
)
def _review(self, question: str, draft: str,
search_results: dict) -> str:
combined_findings = "\n\n".join(
f"{label}:\n{content}" for label, content in search_results.items()
)
return self.reviewer.run(
f"Review this report for the question: {question}\n\n"
f"Report:\n{draft}\n\n"
f"Source research (fact-check against this):\n{combined_findings}\n\n"
f"Check for: factual errors, unsupported claims, logical "
f"gaps, missing important findings, clarity issues, bias. "
f"Provide specific, actionable feedback. For each issue, "
f"cite the specific passage and suggest a fix.",
"You are a rigorous fact-checker and editor. Be direct. "
"Be specific. Your job is to make this report bulletproof."
)
def _revise(self, draft: str, review: str) -> str:
return self.writer.run(
f"Revise this report based on the review feedback.\n\n"
f"Original report:\n{draft}\n\n"
f"Review feedback:\n{review}\n\n"
f"Address every issue raised. Produce the final, "
f"publication-ready version.",
"You are a technical writer producing the final version."
)
# --- Run the system ---
if __name__ == "__main__":
orchestrator = ResearchOrchestrator()
result = orchestrator.run(
"What is the current state and near-term outlook for "
"nuclear fusion energy? Cover: major projects (ITER, SPARC, "
"private ventures), key technical milestones achieved and "
"remaining, investment trends, timeline to commercial viability, "
"and the biggest skeptics' arguments."
)
The Execution Trace
Running this on the nuclear fusion question produces a six-phase trace:
Phase 1 -- Research Plan:
{
"angles": [
"Major fusion projects: ITER status, timeline, budget; SPARC and CFS
approach; private ventures (Helion, TAE, Zap, General Fusion) and
their claimed timelines",
"Investment trends: total private fusion investment, major investors,
government funding programs, year-over-year growth, comparison to
other clean energy sectors",
"Technical milestones: Q>1 achievements (NIF 2022), remaining
challenges (tritium breeding, materials, sustained operation),
expert assessments of timeline credibility"
],
"analysis_focus": "Gap between claimed timelines and technical reality;
which approaches have the strongest empirical basis; investment vs.
progress ratio",
"report_structure": [
"Executive Summary",
"The Fusion Landscape: Projects and Approaches",
"Technical Status: What Has Been Achieved",
"Investment and Funding Trends",
"Timeline to Commercial Viability",
"Skeptics' Case: Why Fusion Might Not Deliver",
"Conclusions and Outlook"
]
}
Phase 2 -- Parallel Search: Two searchers run simultaneously. Searcher-Alpha covers angles 1 and 2. Searcher-Beta covers angle 3 and seeks contrarian views. Combined output: ~3,000 words of research with specific data points, project statuses, investment figures, and expert quotes.
Phase 3 -- Analysis: The Analyst identifies the key pattern: "The gap between claimed timelines (CFS: 2030s, Helion: 2028) and demonstrated technical progress (no sustained Q>1, no tritium breeding at scale) is the central tension. Private fusion companies have raised $6B+ but none have demonstrated net energy. The NIF achievement was a physics demonstration, not an engineering milestone -- the laser approach is not commercially viable. The most credible near-term path is tokamak-based (SPARC/ITER), but even that faces unresolved materials and fuel cycle challenges."
Phase 4 -- Draft Report: The Writer produces a 2,500-word report with all seven sections, specific data, and a balanced assessment.
Phase 5 -- Review: The Reviewer identifies three issues: (1) the report overstates Helion's credibility without noting their approach (field-reversed configuration) has never achieved significant plasma performance, (2) the investment section conflates total raised with annual investment -- the trend is actually flat since 2022, and (3) the skeptics' section is too short and does not engage with the strongest argument (fusion will always be too expensive per kWh compared to solar + storage).
Phase 6 -- Final Report: The Writer addresses all three issues. The final report is accurate, balanced, and authoritative.
This is the pattern that works. Not "throw more agents at the problem." Not "let agents chat freely." A designed pipeline with clear phases, parallel work where possible, structured handoffs, and a review step that catches errors before the output reaches the user.
The Turn
You now understand that multi-agent systems are not about "more agents." They are about the right orchestration pattern for the task.
Debate for quality. When the cost of a wrong answer is high, make agents challenge each other. The answer that survives attack is stronger than any single agent's answer.
Hierarchy for complex decomposition. When a task is too large for one agent, have a manager break it down and delegate. The manager is the intelligence -- invest in its prompt.
Swarm for parallel work. When sub-tasks are independent, run them simultaneously. Speed without sacrificing quality.
Sequential for pipelines. When each stage transforms the output of the previous, chain them. But keep the chain short -- information degrades with every handoff.
Voting for reliability. When you need to reduce hallucination on factual questions, run multiple agents independently and take the consensus.
The pattern matters more than the number of agents. Two agents with the right pattern outperform ten agents with the wrong one. Start with one agent. Add a second only when you can name the blind spot it addresses. Add a third only when the second proves insufficient. Every agent must earn its seat.
Close: Revelation
Your agents can now collaborate, debate, divide work, and produce output that no single agent could match. They can challenge each other's assumptions, catch each other's errors, and synthesize multiple perspectives into coherent results.
But they are still working from training data and web search. They know what was in their training corpus. They can search the public internet. For many real applications, that is not enough.
Your company's internal documents. Your team's design decisions. Your customer's support history. Your proprietary data that exists nowhere on the public web. Your agents cannot see any of it.
For that, you need to ground them in YOUR data. You need to give them access to your documents, your knowledge base, your company's information -- and you need them to retrieve exactly the right information at exactly the right moment.
That is Retrieval-Augmented Generation. That is the next chapter.