The Problem
Without a coordinator, multi-agent systems collapse into chaos. Each agent completes its task without knowing if downstream agents need its output in a specific format. When a worker fails, nothing handles the retry. There's no record of what was tried and why. The supervisor pattern solves all of this: one orchestrator agent acts as the control plane — routing tasks, handling failures, tracking state, and producing the audit trail.
The supervisor is the system's nervous system. Workers are its hands. The supervisor knows the goal; workers know how to execute one specific piece.
Supervisor architecture
One supervisor, N workers. The architecture is simple — the discipline is in the details.
The supervisor's responsibilities:
- Receive the top-level task
- Plan the workflow (which workers, in what order)
- Dispatch to appropriate workers
- Receive and evaluate results
- Decide on next steps (proceed, retry, escalate, complete)
- Handle failures with explicit strategies
- Synthesize the final output
The worker's responsibilities:
- Accept one task with clear context
- Execute using its specialized tools
- Return a structured result with status, confidence, and any errors
- Nothing else — workers don't coordinate with each other
From the supervisor's perspective, each worker is a tool. The supervisor calls researcher(task, context) the same way it would call a database query — except the "database" is an LLM with a specialized system prompt.
Structured dispatch protocol
Free-form communication between supervisor and workers creates parsing ambiguity. A structured protocol eliminates it.
Dispatch format (supervisor → worker):
{
"worker": "researcher",
"task": "Find recent papers on transformer attention mechanisms",
"context": {"session_id": "abc123", "topic": "attention"},
"max_steps": 5
}
Result format (worker → supervisor):
{
"status": "success", # "success" | "failed" | "partial"
"result": {...}, # The actual output
"steps_used": 3,
"confidence": 0.9, # 0–1: how confident is the worker in its result?
"error": None # Error message if status != success
}
The confidence field is critical. A supervisor that sees confidence: 0.4 knows to treat the result skeptically — perhaps retry, cross-check with a different worker, or flag for human review. confidence: 0.9 means proceed.
A supervisor dispatches a "research and summarize academic papers" task to the researcher agent. The researcher returns status="failed" with reason="rate_limit". What's the supervisor's correct response?
Rate limits are transient. The correct response is retry with backoff — the same worker will eventually succeed. Switching workers is wrong (the researcher is the right agent for this task). Only after persistent failures should the supervisor consider alternatives. Terminating immediately wastes all prior work and gives the user nothing.
Failure handling in supervisor systems
When a worker fails, the supervisor has a decision tree:
- Retry same worker — for transient failures (rate limits, timeouts, network errors). Use exponential backoff. Max 3 retries.
- Retry with different worker — for capability mismatch failures. "The summarizer can't handle this format" → try a different summarizer.
- Decompose the task — if the task is too large for the worker, split it into smaller pieces and retry.
- Escalate to human — when all retry strategies fail and a partial result would be harmful.
- Return partial results — when partial information is better than nothing. Always include a note about what's missing.
Cascading failures are the #1 bug in supervisor systems. Worker A fails to provide context, so Worker B gets no input and fails too. Design your dispatch so each worker can return useful partial results even without perfect input from upstream.
Circuit breakers at the worker level: if a worker fails 3 times in 5 minutes, mark it as "degraded" and route around it for the next 10 minutes. Don't keep hammering a worker that's clearly broken.
In the supervisor pattern, the supervisor agent knows the workflow is complete when:
The supervisor's termination condition is dual: (1) natural completion — it has all information needed to answer the original task; (2) hard budget cap — max_steps is exhausted. Always have both. A natural completion condition alone can loop forever if the model is uncertain. A budget cap alone returns incomplete results. Both together give you a reliable termination guarantee.
Build
Build it
Here's a small standalone simulation of the retry-with-backoff dispatch logic before the full supervisor implementation.
# Minimal supervisor dispatch retry simulation (no dependencies)
import random
random.seed(7)
def worker_call(attempt):
# Simulate a flaky worker that succeeds ~50% of the time, and always
# succeeds by the 3rd attempt for this demo.
if attempt == 3 or random.random() > 0.5:
return {"status": "success", "confidence": 0.9}
return {"status": "failed", "error": "rate_limit"}
def dispatch_with_retry(worker_name, max_retries=3):
for attempt in range(1, max_retries + 1):
result = worker_call(attempt)
print(f" attempt {attempt}: {result['status']}")
if result["status"] == "success":
return result
return {"status": "failed", "confidence": 0.0}
print(f"Dispatching to 'researcher'...")
final = dispatch_with_retry("researcher")
print(f"Final result: {final}")
The code below implements a complete supervisor system with three workers, structured dispatch, failure handling, and a full audit trail.
"""
lesson_04_02_supervisor.py
Supervisor / Orchestrator pattern: one supervisor coordinates multiple worker agents.
Includes structured dispatch, retry logic, failure handling, and audit logging.
"""
import os
import json
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional, Any
import anthropic
ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY")
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) if ANTHROPIC_KEY else None
@dataclass
class WorkerResult:
"""Structured result returned by every worker agent."""
status: str # "success" | "failed" | "partial"
result: Any # The actual output (dict, str, etc.)
steps_used: int
confidence: float # 0.0–1.0
error: Optional[str] = None
worker_name: str = ""
@dataclass
class DispatchRecord:
"""Audit trail entry for one supervisor dispatch."""
dispatch_id: str
timestamp: str
worker: str
task: str
result_status: str
confidence: float
retry_count: int
duration_ms: int
notes: str = ""
class WorkerAgent:
"""Base class for all worker agents."""
def __init__(self, name: str, system_prompt: str):
self.name = name
self.system_prompt = system_prompt
self._fail_count = 0 # For testing failure scenarios
def execute(self, task: str, context: dict) -> WorkerResult:
raise NotImplementedError
def _call_llm(self, user_message: str) -> str:
if not client:
return f"[Mock result from {self.name}: {user_message[:60]}...]"
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=400,
system=self.system_prompt,
messages=[{"role": "user", "content": user_message}],
)
return response.content[0].text
class ResearchWorker(WorkerAgent):
def __init__(self):
super().__init__(
name="researcher",
system_prompt=(
"You are a research specialist. Given a topic, provide a concise, "
"factual summary with 3–5 key points. Be precise and cite specifics. "
"Rate your confidence in the accuracy of your research (0.0–1.0)."
)
)
def execute(self, task: str, context: dict) -> WorkerResult:
start = time.time()
try:
content = self._call_llm(
f"Research this topic and provide 3-5 key factual points: {task}"
)
duration = int((time.time() - start) * 1000)
return WorkerResult(
status="success",
result={"summary": content, "source": "llm_research"},
steps_used=2,
confidence=0.82,
worker_name=self.name,
)
except Exception as e:
return WorkerResult(
status="failed",
result={},
steps_used=1,
confidence=0.0,
error=str(e),
worker_name=self.name,
)
class WriterWorker(WorkerAgent):
def __init__(self):
super().__init__(
name="writer",
system_prompt=(
"You are a technical writer. Transform research findings into a clear, "
"well-structured draft. Use precise language, logical flow, and appropriate "
"formatting. Produce ready-to-review content."
)
)
def execute(self, task: str, context: dict) -> WorkerResult:
research = context.get("research_summary", "No research provided.")
try:
content = self._call_llm(
f"Write a clear, structured draft based on this research.\n"
f"Topic: {task}\nResearch:\n{research}"
)
return WorkerResult(
status="success",
result={"draft": content},
steps_used=2,
confidence=0.88,
worker_name=self.name,
)
except Exception as e:
return WorkerResult(
status="failed", result={}, steps_used=1,
confidence=0.0, error=str(e), worker_name=self.name,
)
class FactCheckerWorker(WorkerAgent):
def __init__(self):
super().__init__(
name="fact_checker",
system_prompt=(
"You are a fact-checking specialist. Review drafts for factual accuracy. "
"Identify any claims that are incorrect, unverifiable, or misleading. "
"Provide a verdict: APPROVED or NEEDS_REVISION with specific issues."
)
)
def execute(self, task: str, context: dict) -> WorkerResult:
draft = context.get("draft", "No draft provided.")
try:
review = self._call_llm(
f"Fact-check this draft about '{task}':\n\n{draft}\n\n"
f"Give APPROVED or NEEDS_REVISION with specific issues if any."
)
approved = "APPROVED" in review.upper()
return WorkerResult(
status="success",
result={"verdict": "APPROVED" if approved else "NEEDS_REVISION", "review": review},
steps_used=2,
confidence=0.85,
worker_name=self.name,
)
except Exception as e:
return WorkerResult(
status="failed", result={}, steps_used=1,
confidence=0.0, error=str(e), worker_name=self.name,
)
class SupervisorAgent:
"""
Orchestrates worker agents to complete a multi-step task.
Manages:
- Dispatch planning
- Structured results
- Retry with backoff on failure
- Audit trail for every decision
- Hard step budget to prevent runaway loops
"""
MAX_RETRIES = 3
MAX_STEPS = 20
CONFIDENCE_THRESHOLD = 0.6
def __init__(self):
self.workers: dict[str, WorkerAgent] = {
"researcher": ResearchWorker(),
"writer": WriterWorker(),
"fact_checker": FactCheckerWorker(),
}
self.audit_log: list[DispatchRecord] = []
self._dispatch_counter = 0
def _dispatch(self, worker_name: str, task: str, context: dict) -> WorkerResult:
"""Dispatch to a worker with retry logic."""
self._dispatch_counter += 1
worker = self.workers.get(worker_name)
if not worker:
raise ValueError(f"Unknown worker: {worker_name}")
start = time.time()
retry_count = 0
result = None
for attempt in range(self.MAX_RETRIES):
result = worker.execute(task, context)
if result.status == "success":
break
retry_count += 1
if attempt < self.MAX_RETRIES - 1:
wait = 2 ** attempt # exponential backoff
print(f" [supervisor] Worker '{worker_name}' failed (attempt {attempt+1}). "
f"Retrying in {wait}s. Error: {result.error}")
time.sleep(wait)
duration_ms = int((time.time() - start) * 1000)
record = DispatchRecord(
dispatch_id=f"d{self._dispatch_counter:04d}",
timestamp=datetime.utcnow().isoformat(),
worker=worker_name,
task=task[:80],
result_status=result.status if result else "unknown",
confidence=result.confidence if result else 0.0,
retry_count=retry_count,
duration_ms=duration_ms,
)
self.audit_log.append(record)
return result
def run(self, task: str) -> dict:
"""
Execute a research → write → fact-check workflow.
Returns the final report or partial results with explanation.
"""
steps_used = 0
context: dict = {}
print(f"\n[supervisor] Starting task: {task}")
# Step 1: Research
if steps_used >= self.MAX_STEPS:
return {"status": "budget_exceeded", "result": context}
print(f"[supervisor] Dispatching to researcher...")
research_result = self._dispatch("researcher", task, context)
steps_used += research_result.steps_used
if research_result.status == "failed":
return {
"status": "partial",
"result": "Research failed after retries. Cannot proceed.",
"audit_log": [asdict(r) for r in self.audit_log],
}
context["research_summary"] = research_result.result.get("summary", "")
if research_result.confidence < self.CONFIDENCE_THRESHOLD:
print(f"[supervisor] Low research confidence ({research_result.confidence}). "
f"Adding caution note.")
context["low_confidence_note"] = True
# Step 2: Write
print(f"[supervisor] Dispatching to writer...")
write_result = self._dispatch("writer", task, context)
steps_used += write_result.steps_used
if write_result.status == "failed":
return {
"status": "partial",
"result": context.get("research_summary", ""),
"note": "Writing failed. Returning raw research.",
"audit_log": [asdict(r) for r in self.audit_log],
}
context["draft"] = write_result.result.get("draft", "")
# Step 3: Fact-check
print(f"[supervisor] Dispatching to fact_checker...")
check_result = self._dispatch("fact_checker", task, context)
steps_used += check_result.steps_used
verdict = check_result.result.get("verdict", "UNKNOWN") if check_result.status == "success" else "UNKNOWN"
print(f"[supervisor] Fact-check verdict: {verdict}")
return {
"status": "success",
"result": context["draft"],
"fact_check_verdict": verdict,
"fact_check_review": check_result.result.get("review", ""),
"steps_used": steps_used,
"audit_log": [asdict(r) for r in self.audit_log],
}
def print_audit_log(self):
print("\n--- AUDIT LOG ---")
for record in self.audit_log:
print(
f" [{record.dispatch_id}] {record.timestamp[:19]} "
f"worker={record.worker:<15} "
f"status={record.result_status:<8} "
f"confidence={record.confidence:.2f} "
f"retries={record.retry_count} "
f"ms={record.duration_ms}"
)
if __name__ == "__main__":
supervisor = SupervisorAgent()
result = supervisor.run("The current state of large language model reasoning capabilities")
print("\n" + "=" * 60)
print("FINAL RESULT")
print("=" * 60)
print(f"Status: {result['status']}")
print(f"Fact-check verdict: {result.get('fact_check_verdict', 'N/A')}")
print(f"\nDraft:\n{result['result'][:600]}...")
supervisor.print_audit_log()