Multi-Agent Systems

Real-World Scenario

A software engineering team uses a multi-agent system for code review: one agent checks security vulnerabilities, another reviews performance, a third reviews code style, and a fourth synthesizes their findings into a prioritized report. All three review agents run in parallel — cutting review time from 3 minutes to 45 seconds. The orchestrator doesn’t do the analysis itself; it delegates, collects, and synthesizes.

Orchestrator + Subagent Pattern

import anthropic
import json
from concurrent.futures import ThreadPoolExecutor, as_completed

client = anthropic.Anthropic()

# ─── Subagent: executes a specific task ──────────────────────────────────

def run_subagent(task: str, persona: str, model: str = "claude-haiku-4-5-20251001") -> str:
    """A focused subagent with a specific role."""
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        system=persona,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text


# ─── Code review: three parallel subagents ───────────────────────────────

REVIEWERS = {
    "security": """You are a security engineer. Review code for:
- SQL injection, XSS, command injection
- Hardcoded secrets or credentials
- Insecure authentication/authorization
- OWASP Top 10 vulnerabilities
Format each finding as: [SEVERITY] Description — Recommendation""",

    "performance": """You are a performance engineer. Review code for:
- Inefficient algorithms (O(n²) when O(n) is possible)
- Missing database indexes
- N+1 query problems
- Unnecessary memory allocation
- Missing caching opportunities
Format each finding as: [SEVERITY] Description — Recommendation""",

    "maintainability": """You are a staff engineer reviewing code quality. Check:
- Code duplication (DRY violations)
- Missing error handling
- Unclear variable/function names
- Functions that do too much (single responsibility)
- Missing input validation
Format each finding as: [SEVERITY] Description — Recommendation""",
}

# ─── Synthesizer: combines findings ──────────────────────────────────────

SYNTHESIZER_SYSTEM = """You are a tech lead synthesizing code review feedback.
Given findings from multiple reviewers, produce a prioritized review report:

1. List CRITICAL issues first (must fix before merge)
2. List HIGH issues (should fix)
3. List MEDIUM issues (consider fixing)
4. Provide a one-paragraph overall assessment

Be concise and actionable."""


def parallel_code_review(code: str) -> dict:
    """Run three review agents in parallel, then synthesize."""
    review_task = f"Review this code:\n\n```python\n{code}\n```"

    findings = {}
    # Run all reviewers concurrently
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = {
            executor.submit(run_subagent, review_task, persona): role
            for role, persona in REVIEWERS.items()
        }
        for future in as_completed(futures):
            role = futures[future]
            findings[role] = future.result()
            print(f"  ✓ {role} review complete")

    # Synthesize findings with a more capable model
    synthesis_prompt = f"""Code under review:
```python
{code}
```text

Security review findings:
{findings['security']}

Performance review findings:
{findings['performance']}

Maintainability review findings:
{findings['maintainability']}

Synthesize these into a final prioritized code review report."""

    synthesis = run_subagent(
        task=synthesis_prompt,
        persona=SYNTHESIZER_SYSTEM,
        model="claude-sonnet-4-6"
    )

    return {"findings": findings, "synthesis": synthesis}


# Test it
sample_code = """
def get_user_orders(user_id, status=None):
    query = f"SELECT * FROM orders WHERE user_id = {user_id}"
    if status:
        query += f" AND status = '{status}'"
    
    orders = []
    results = db.execute(query).fetchall()
    for row in results:
        order = dict(row)
        # Get items for each order
        items = db.execute(f"SELECT * FROM items WHERE order_id = {order['id']}").fetchall()
        order['items'] = [dict(i) for i in items]
        orders.append(order)
    
    return orders
"""

print("Running parallel code review...")
result = parallel_code_review(sample_code)
print("\n=== SYNTHESIS ===")
print(result["synthesis"])

Orchestrator with Dynamic Subagent Spawning

import anthropic
import json
from typing import Optional

client = anthropic.Anthropic()

ORCHESTRATOR_SYSTEM = """You are a research orchestrator. When given a research task:
1. Break it into 3-5 specific subtasks
2. Return them as a JSON array of objects with 'subtask' and 'specialist' fields
3. specialist can be: 'researcher', 'analyst', 'writer', 'fact_checker'

Return ONLY valid JSON, no other text."""

SPECIALISTS = {
    "researcher": "You are a thorough researcher. Find all relevant facts, data, and context about the given topic. Be comprehensive.",
    "analyst":    "You are a data analyst. Identify trends, patterns, and key insights from the information provided.",
    "writer":     "You are a technical writer. Synthesize research findings into clear, structured prose.",
    "fact_checker":"You are a fact-checker. Verify claims and flag any that seem uncertain or need citation.",
}

def orchestrate_research(topic: str) -> str:
    # Step 1: Orchestrator plans the work
    plan_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        system=ORCHESTRATOR_SYSTEM,
        messages=[{"role": "user", "content": f"Plan research for: {topic}"}]
    )
    
    subtasks = json.loads(plan_response.content[0].text)
    print(f"Plan: {len(subtasks)} subtasks")
    for i, task in enumerate(subtasks, 1):
        print(f"  {i}. [{task['specialist']}] {task['subtask']}")

    # Step 2: Run subtasks in parallel
    results = {}
    with ThreadPoolExecutor(max_workers=len(subtasks)) as executor:
        futures = {}
        for i, subtask_info in enumerate(subtasks):
            specialist = subtask_info["specialist"]
            persona = SPECIALISTS.get(specialist, SPECIALISTS["researcher"])
            future = executor.submit(run_subagent, subtask_info["subtask"], persona)
            futures[future] = (i, subtask_info["subtask"])

        for future in as_completed(futures):
            idx, task_desc = futures[future]
            results[idx] = {"task": task_desc, "result": future.result()}
            print(f"  ✓ Subtask {idx+1} complete")

    # Step 3: Synthesize into final report
    results_text = "\n\n".join([
        f"### Subtask {i+1}: {r['task']}\n{r['result']}"
        for i, r in sorted(results.items())
    ])

    final_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system="You are a research director. Synthesize the following research into a comprehensive, well-structured report.",
        messages=[{
            "role": "user",
            "content": f"Topic: {topic}\n\nResearch findings:\n\n{results_text}"
        }]
    )
    return final_response.content[0].text


result = orchestrate_research("The impact of transformer architecture on modern NLP")
print("\n=== FINAL REPORT ===")
print(result[:1000])  # print first 1000 chars

Human-in-the-Loop

import anthropic
import json

client = anthropic.Anthropic()

def human_approval_agent(task: str, risky_actions: list[str]) -> str:
    """Agent that pauses for human approval before risky actions."""

    tools = [
        {
            "name": "send_email",
            "description": "Send an email to a customer.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "to":      {"type": "string"},
                    "subject": {"type": "string"},
                    "body":    {"type": "string"},
                },
                "required": ["to", "subject", "body"]
            }
        },
        {
            "name": "read_database",
            "description": "Read customer records from the database.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    ]

    messages = [{"role": "user", "content": task}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return next(b.text for b in response.content if hasattr(b, "text"))

        messages.append({"role": "assistant", "content": response.content})
        tool_results = []

        for block in response.content:
            if block.type != "tool_use":
                continue

            # Check if this action requires human approval
            if block.name in risky_actions:
                print(f"\n⚠️  APPROVAL REQUIRED")
                print(f"Tool:  {block.name}")
                print(f"Input: {json.dumps(block.input, indent=2)}")
                approval = input("Approve? (y/n): ").strip().lower()

                if approval != "y":
                    result = "Action denied by human operator."
                    print("❌ Denied")
                else:
                    result = f"Action '{block.name}' executed successfully."
                    print("✅ Approved and executed")
            else:
                # Safe actions run automatically
                result = f"Database query executed. Found 3 records matching: {block.input.get('query', '')}"

            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })

        messages.append({"role": "user", "content": tool_results})


# send_email requires human approval; read_database runs automatically
human_approval_agent(
    task="Find all customers with overdue payments and send them a reminder email.",
    risky_actions=["send_email"]
)