Most developers hit the same wall when scaling Claude-based automation: a single agent trying to do everything becomes a sprawling, unreliable mess. Multi-agent workflows with Claude solve this by splitting complex tasks across specialized agents that coordinate through well-defined interfaces — but the gap between a toy demo and something that holds up in production is substantial. This guide covers the architecture patterns, orchestration code, and failure modes I’ve run into shipping these systems for real.
Why Single-Agent Architectures Break at Scale
A single Claude agent handling a complex pipeline — say, ingesting a support ticket, querying a knowledge base, drafting a response, checking compliance, and logging the outcome — runs into a few hard limits fast.
- Context window exhaustion: Stuffing multiple tool calls, retrieval results, and conversation history into one context blows past 200K tokens quickly when you’re processing high volumes.
- Reliability cliff: The more steps you chain in a single prompt, the higher the probability of one step going wrong and poisoning the rest. There’s no clean retry boundary.
- Cost inefficiency: Running claude-3-5-sonnet for tasks that only need claude-3-haiku is waste you’ll feel at scale. A 10-step pipeline running on Sonnet when 7 of those steps are trivial formatting or classification tasks adds up fast.
- Debugging hell: When something breaks inside a monolithic agent, you’re trawling through a massive trace to find the failure point.
Multi-agent systems fix all of this — at the cost of coordination complexity. That’s the actual tradeoff you’re navigating.
The Three Core Orchestration Patterns
1. Hierarchical Orchestrator-Worker
One orchestrator agent breaks down a task and dispatches subtasks to specialized worker agents. The orchestrator receives results, validates them, and either continues the pipeline or requests a retry. This is the right pattern for complex, multi-step tasks where the subtasks aren’t known ahead of time.
import anthropic
import json
client = anthropic.Anthropic()
def run_orchestrator(task: str) -> dict:
"""Top-level orchestrator that plans and delegates subtasks."""
system_prompt = """You are an orchestrator agent. Given a complex task, break it into
discrete subtasks and return them as JSON. Each subtask should have:
- id: unique identifier
- agent_type: 'researcher' | 'writer' | 'reviewer'
- instruction: specific instruction for that agent
- depends_on: list of task IDs this task depends on (empty if none)
"""
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": f"Plan this task: {task}"}]
)
# Parse the plan from the orchestrator's response
plan = json.loads(response.content[0].text)
results = {}
# Execute tasks respecting dependency order
for subtask in topological_sort(plan["subtasks"]):
context = {dep: results[dep] for dep in subtask["depends_on"]}
results[subtask["id"]] = run_worker(subtask, context)
return results
def run_worker(subtask: dict, context: dict) -> str:
"""Execute a single subtask with the appropriate agent config."""
# Use cheaper model for simpler agent types
model = "claude-3-haiku-20240307" if subtask["agent_type"] == "reviewer" else "claude-3-5-sonnet-20241022"
context_str = "\n".join([f"{k}: {v}" for k, v in context.items()])
response = client.messages.create(
model=model,
max_tokens=2048,
messages=[{
"role": "user",
"content": f"Context from prior steps:\n{context_str}\n\nYour task: {subtask['instruction']}"
}]
)
return response.content[0].text
The topological_sort function handles dependency ordering — don’t skip this. Running tasks out of order is a silent failure mode that produces garbage results without throwing errors.
2. Sequential Pipeline with Handoffs
Each agent does its job and passes a structured output to the next. Simpler than hierarchical, easier to debug, and the right choice when your task flow is predictable. Think ETL-style pipelines: extract → transform → validate → store.
from dataclasses import dataclass
from typing import Optional
@dataclass
class PipelineState:
raw_input: str
extracted_data: Optional[dict] = None
transformed_data: Optional[dict] = None
validation_result: Optional[dict] = None
error: Optional[str] = None
def run_pipeline(raw_input: str) -> PipelineState:
state = PipelineState(raw_input=raw_input)
# Each stage updates state and can short-circuit on failure
stages = [extraction_agent, transformation_agent, validation_agent]
for stage in stages:
try:
state = stage(state)
if state.error:
# Log the error, optionally retry, then exit the pipeline
print(f"Pipeline halted at {stage.__name__}: {state.error}")
break
except Exception as e:
state.error = str(e)
break
return state
def extraction_agent(state: PipelineState) -> PipelineState:
response = client.messages.create(
model="claude-3-haiku-20240307", # Cheap model for structured extraction
max_tokens=512,
system="Extract structured data from the input. Return JSON only.",
messages=[{"role": "user", "content": state.raw_input}]
)
state.extracted_data = json.loads(response.content[0].text)
return state
3. Parallel Fan-Out with Aggregation
Dispatch multiple agents simultaneously for tasks that are independent, then aggregate results. Useful for things like running multiple research queries in parallel, or having several agents evaluate the same content from different perspectives.
import asyncio
import anthropic
async_client = anthropic.AsyncAnthropic()
async def run_parallel_agents(task: str, perspectives: list[str]) -> list[str]:
"""Run the same task from multiple agent perspectives simultaneously."""
async def single_agent(perspective: str) -> str:
response = await async_client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=f"You are evaluating this from the perspective of: {perspective}",
messages=[{"role": "user", "content": task}]
)
return response.content[0].text
# Fire all agents simultaneously - this is where you get the latency win
results = await asyncio.gather(*[single_agent(p) for p in perspectives])
return list(results)
# Aggregator then synthesizes results
async def aggregate_results(results: list[str]) -> str:
combined = "\n\n---\n\n".join([f"Perspective {i+1}:\n{r}" for i, r in enumerate(results)])
response = await async_client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
system="Synthesize multiple perspectives into a single coherent analysis.",
messages=[{"role": "user", "content": combined}]
)
return response.content[0].text
Production Error Handling That Actually Works
The retry logic in most tutorials is naive. Here’s what production error handling looks like for multi-agent workflows with Claude:
import time
from anthropic import RateLimitError, APIStatusError
def call_claude_with_retry(
model: str,
messages: list,
max_retries: int = 3,
base_delay: float = 1.0
) -> str:
"""Wrapper with exponential backoff and structured error handling."""
for attempt in range(max_retries):
try:
response = client.messages.create(
model=model,
max_tokens=2048,
messages=messages
)
return response.content[0].text
except RateLimitError:
# Rate limits need longer backoff — don't just retry immediately
delay = base_delay * (4 ** attempt)
print(f"Rate limited. Waiting {delay}s before retry {attempt + 1}/{max_retries}")
time.sleep(delay)
except APIStatusError as e:
if e.status_code >= 500:
# Server errors: retry with backoff
delay = base_delay * (2 ** attempt)
time.sleep(delay)
else:
# Client errors (4xx except 429): don't retry, raise immediately
raise
raise RuntimeError(f"Failed after {max_retries} retries")
One thing the docs underemphasize: rate limits in multi-agent systems compound. If you’re running 10 parallel agents and they all hit rate limits simultaneously, naive retry logic will make every single one wait and retry in near-unison. Implement jitter: delay = base_delay * (2 ** attempt) + random.uniform(0, 1). This spreads retries out and prevents thundering herd problems.
Cost Optimization: Model Routing in Practice
This is where you actually save money. Not every agent in your pipeline needs Sonnet. Here’s a routing pattern I use in production:
def select_model(task_type: str, complexity_score: float) -> str:
"""
Route to the cheapest model that can handle the task.
Haiku: ~$0.00025/1K input tokens
Sonnet: ~$0.003/1K input tokens
Opus: ~$0.015/1K input tokens
(Verify current pricing at anthropic.com/pricing)
"""
# Always use Haiku for these — they don't need reasoning
cheap_tasks = {"classification", "extraction", "formatting", "validation"}
if task_type in cheap_tasks:
return "claude-3-haiku-20240307"
# Use complexity score (e.g., from a quick Haiku pre-assessment) to route
if complexity_score < 0.4:
return "claude-3-haiku-20240307"
elif complexity_score < 0.75:
return "claude-3-5-sonnet-20241022"
else:
return "claude-3-opus-20240229"
def assess_complexity(task: str) -> float:
"""Use Haiku to quickly score task complexity before routing. Costs fractions of a cent."""
response = client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=10,
system="Rate task complexity 0.0-1.0. Return only the number.",
messages=[{"role": "user", "content": task}]
)
return float(response.content[0].text.strip())
In one pipeline I shipped handling roughly 50,000 tasks per month, routing classification and extraction steps to Haiku cut monthly API costs by around 60% with no measurable quality drop on those steps. The complexity assessment itself costs about $0.000015 per call — negligible.
Debugging Multi-Agent Systems Without Going Insane
When something breaks in a multi-agent workflow, you need observability baked in from the start. Retrofitting it is painful.
Structured Logging per Agent
import uuid
import logging
from datetime import datetime
def log_agent_call(agent_id: str, model: str, input_tokens: int,
output_tokens: int, duration_ms: float, trace_id: str):
"""Emit structured logs for every agent call."""
logging.info({
"event": "agent_call",
"trace_id": trace_id, # Ties all agents in one workflow run together
"agent_id": agent_id,
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"duration_ms": duration_ms,
"timestamp": datetime.utcnow().isoformat(),
# Estimated cost based on current pricing
"estimated_cost_usd": (input_tokens * 0.000003) + (output_tokens * 0.000015)
})
def run_traced_workflow(task: str) -> dict:
trace_id = str(uuid.uuid4()) # Single ID for the entire multi-agent run
# Pass trace_id through every agent call
return run_orchestrator(task, trace_id=trace_id)
Ship these logs to wherever you already aggregate (Datadog, CloudWatch, even a simple Postgres table). The trace_id lets you reconstruct the exact sequence of agent calls for any given workflow run — critical when you’re trying to understand why a specific customer’s request produced a bad output three days ago.
Prompt Versioning
Version your system prompts like code. Store them with a hash or version number in your logs. When you update a prompt and behavior changes, you need to know exactly which run used which version. I’ve been burned by this — a prompt update improved 95% of cases and silently broke the other 5%, which only surfaced in customer complaints a week later.
When to Use n8n or Make vs. Pure Code
If your multi-agent workflow is primarily moving data between services with Claude handling specific steps, n8n’s visual workflow builder can genuinely accelerate development. The Claude nodes work well for single-call steps. Where it breaks down: complex conditional routing between agents, stateful workflows that need to persist mid-execution, and anything that requires the kind of retry logic shown above.
My rule: use n8n for workflows where the majority of nodes are integrations (Slack, databases, webhooks) with Claude as one actor. Use pure Python when Claude agents are the primary logic layer.
Who Should Use Which Pattern
Solo founders and small teams building internal tools: start with sequential pipelines. They’re easy to debug, straightforward to extend, and you can add parallelism later when you have actual performance data showing you need it.
Teams building customer-facing products at any meaningful scale: hierarchical orchestration with structured logging from day one. The upfront investment in observability pays back immediately once you’re debugging production issues.
Anyone optimizing for cost: implement model routing before anything else. The Haiku/Sonnet split alone typically cuts costs 40-70% on pipelines with mixed task complexity, and it takes an afternoon to implement.
The patterns for multi-agent workflows with Claude in this guide aren’t theoretical — they’re distilled from shipping systems that process hundreds of thousands of tasks per month. Start with the sequential pipeline, add the retry wrapper, implement logging with trace IDs, and only reach for the hierarchical pattern when your task complexity actually demands it. Complexity you add before you need it just becomes technical debt with an LLM API bill attached.
Editorial note: API pricing, model capabilities, and tool features change frequently — always verify current details on the vendor’s website before building in production. Code examples are tested at time of writing; pin your dependency versions to avoid breaking changes. Some links in this article may be affiliate links — we may earn a commission if you sign up, at no extra cost to you.
