Multi-Agent Systems

Real-World Scenario

A team wants automated code review. One agent with one long prompt misses things and takes a while, so they split it: a security reviewer, a performance reviewer and a maintainability reviewer each look at the same diff in parallel, and an orchestrator merges their findings into one prioritized report. It works until the performance reviewer times out and the whole review disappears with it. Then the planning agent returns its plan wrapped in Markdown and the parser crashes. The orchestration is where multi-agent systems succeed or fail, and it’s ordinary code.

Every example below runs offline. Each subagent is a stand-in: a small function playing the part of a model call, with a time.sleep for the latency a real call has. The reviewers really do inspect the code they are given, with regular expressions, so their findings are real. Everything around them (fan-out, failure handling, plan parsing, approval) is the code you’d use with real models, and no API key is needed. The last section shows how to swap in a real provider.

Subagents: Three Reviewers

Save the code under review and the three reviewers as reviewers.py. synthesize is the orchestrator’s final step, merging all findings, worst first.

import re, time

CODE = '''def get_user_orders(user_id):
    orders = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}").fetchall()
    for order in orders:
        order["items"] = db.execute(f"SELECT * FROM items WHERE order_id = {order['id']}").fetchall()
    return orders'''

def security(code):
    """Stands in for a security-reviewer subagent (a real one is a model call, ~0.5 s)."""
    time.sleep(0.5)
    return [("CRITICAL", "SQL built with an f-string: injection risk")] if re.search(r'f".*SELECT', code) else []

def performance(code):
    """Stands in for a performance-reviewer subagent."""
    time.sleep(0.5)
    return [("HIGH", "query inside a for loop: N+1 queries")] if re.search(r"for .*\n.*execute", code) else []

def maintainability(code):
    """Stands in for a maintainability-reviewer subagent."""
    time.sleep(0.5)
    return [("MEDIUM", "SELECT * couples the code to every column")] if "SELECT *" in code else []

REVIEWERS = {"security": security, "performance": performance, "maintainability": maintainability}
SEVERITY = ["CRITICAL", "HIGH", "MEDIUM", "LOW"]

def synthesize(findings):
    """The orchestrator's last step: merge every subagent's findings, worst first."""
    merged = [(sev, role, text) for role, items in findings.items() for sev, text in items]
    return sorted(merged, key=lambda f: SEVERITY.index(f[0]))

Each reviewer has one narrow job, like a subagent with its own system prompt. None of them knows the others exist.

The Orchestrator: Fan Out in Parallel

Run the three reviewers one after another, then all at once in a thread pool, and time both. Save this as parallel.py.

import time
from concurrent.futures import ThreadPoolExecutor
from reviewers import CODE, REVIEWERS, synthesize

start = time.perf_counter()
one_by_one = {role: review(CODE) for role, review in REVIEWERS.items()}
print(f"sequential: {time.perf_counter() - start:.2f}s")

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=len(REVIEWERS)) as pool:
    futures = {role: pool.submit(review, CODE) for role, review in REVIEWERS.items()}
    parallel = {role: future.result() for role, future in futures.items()}
print(f"parallel:   {time.perf_counter() - start:.2f}s  same findings: {parallel == one_by_one}")

for severity, role, text in synthesize(parallel):
    print(f"[{severity:8}] {role:15} {text}")

$ python parallel.py
sequential: 1.50s
parallel:   0.50s  same findings: True
[CRITICAL] security        SQL built with an f-string: injection risk
[HIGH    ] performance     query inside a for loop: N+1 queries
[MEDIUM  ] maintainability SELECT * couples the code to every column

Sequentially, three half-second reviews take 1.50 s; in parallel they take 0.50 s, the time of the slowest one, with identical findings. Threads are the right tool here even with Python’s GIL, because a subagent spends nearly all its time waiting on the network, not computing. The orchestrator’s merge puts the SQL injection first regardless of which reviewer finished first. Collecting futures by role, rather than in completion order, keeps the report deterministic.

The Failure: One Subagent Takes Down the Review

Real model calls time out, hit rate limits and return errors. Replace the performance reviewer with one that fails, and collect results the naive way first. Save this as partial.py.

from concurrent.futures import ThreadPoolExecutor
from reviewers import CODE, REVIEWERS, synthesize

def flaky_performance(code):
    raise TimeoutError("performance subagent timed out after 30 s")

reviewers = REVIEWERS | {"performance": flaky_performance}

with ThreadPoolExecutor() as pool:
    futures = {role: pool.submit(review, CODE) for role, review in reviewers.items()}
    try:
        findings = {role: f.result() for role, f in futures.items()}
    except TimeoutError as e:
        print("naive gather lost every review:", e)

    findings, failed = {}, []
    for role, future in futures.items():
        try:
            findings[role] = future.result()
        except Exception as e:
            failed.append(f"{role} ({e})")
for severity, role, text in synthesize(findings):
    print(f"[{severity:8}] {role:15} {text}")
print("INCOMPLETE REVIEW - not run:", *failed)

$ python partial.py
naive gather lost every review: performance subagent timed out after 30 s
[CRITICAL] security        SQL built with an f-string: injection risk
[MEDIUM  ] maintainability SELECT * couples the code to every column
INCOMPLETE REVIEW - not run: performance (performance subagent timed out after 30 s)

future.result() re-raises the subagent’s exception in the orchestrator. In the naive version that exception escapes the dictionary comprehension, so the security review, which had succeeded, is thrown away along with everything else. The fix collects each future in its own try. Two of three reviews survive, and the report says plainly which one didn’t run. That last line matters as much as the findings. A review that silently omits the performance section reads as “no performance problems”, which is the one conclusion you can’t draw here. Whether to retry, wait, or ship a partial result is a policy decision, but it should be made in code, not by an unhandled exception.

Dynamic Spawning: Parsing the Orchestrator’s Plan

In the review, the subagents were fixed in advance. A research orchestrator instead asks a model to plan the subtasks and names a specialist for each, so the set of subagents is decided at run time. The model’s reply is text, and it isn’t always the text you asked for. Save this stand-in planner as planner.py.

import json

def plan(topic):
    """Stands in for an orchestrator LLM asked for 'ONLY a JSON array of subtasks'.
    Like real models, it sometimes wraps the JSON in a Markdown fence anyway."""
    return '```json\n' + json.dumps([
        {"specialist": "researcher", "subtask": f"Collect key dates for {topic}"},
        {"specialist": "analyst", "subtask": f"Count how {topic} changed benchmark scores"},
        {"specialist": "poet", "subtask": f"Write a sonnet about {topic}"},
        {"specialist": "fact_checker", "subtask": "Verify every date and number"},
    ], indent=1) + '\n```'

if __name__ == "__main__":
    reply = plan("transformers in NLP")
    print("first lines:", reply.splitlines()[:3])
    try:
        json.loads(reply)
    except json.JSONDecodeError as e:
        print("json.loads failed:", e)

$ python planner.py
first lines: ['```json', '[', ' {']
json.loads failed: Expecting value: line 1 column 1 (char 0)

The plan is valid JSON inside a Markdown code fence, and json.loads fails on the very first character, the backtick. Models do this even when the prompt says “return only JSON”, because fenced JSON is so common in their training data. The plan also contains a subtask for a poet, a specialist nobody defined. Parse it defensively, then spawn one subagent per valid step. Save this as orchestrate.py.

import json, re
from concurrent.futures import ThreadPoolExecutor
from planner import plan

SPECIALISTS = {   # each stands in for a subagent: a model call with its own system prompt
    "researcher":   lambda task: f"3 sources found for: {task}",
    "analyst":      lambda task: f"trend table built for: {task}",
    "fact_checker": lambda task: f"2 claims flagged for citation in: {task}",
}

def parse_plan(reply):
    body = re.sub(r"^```(?:json)?\s*|\s*```$", "", reply.strip())   # strip a Markdown fence
    steps = json.loads(body)
    unknown = [s for s in steps if s["specialist"] not in SPECIALISTS]
    for s in unknown:
        print(f"skipped: no specialist {s['specialist']!r} for {s['subtask']!r}")
    return [s for s in steps if s not in unknown]

if __name__ == "__main__":
    steps = parse_plan(plan("transformers in NLP"))
    with ThreadPoolExecutor(max_workers=len(steps)) as pool:      # one subagent per step
        results = pool.map(lambda s: SPECIALISTS[s["specialist"]](s["subtask"]), steps)
        for step, result in zip(steps, results):
            print(f"{step['specialist']:12} -> {result}")

$ python orchestrate.py
skipped: no specialist 'poet' for 'Write a sonnet about transformers in NLP'
researcher   -> 3 sources found for: Collect key dates for transformers in NLP
analyst      -> trend table built for: Count how transformers in NLP changed benchmark scores
fact_checker -> 2 claims flagged for citation in: Verify every date and number

Stripping the fence makes the plan parse. Checking each step against SPECIALISTS drops the poet subtask with a visible message instead of crashing on a KeyError in the middle of the fan-out, or worse, falling back to some default persona that writes a sonnet into a research report. The number of subagents came from the plan: three this time, and a different topic might produce five. pool.map returns results in the order of steps, so the report keeps the plan’s order even though the subagents run concurrently. Where a provider offers structured outputs or strict tool schemas, use them to constrain the plan’s shape, and keep this validation anyway, because the specialist names are still your code’s responsibility.

Human-in-the-Loop: Approval Before Risky Actions

Some tool calls can’t be undone: an email sent, money refunded. Those should wait for a person, and the check belongs in the code that runs tools, where the model can’t talk its way past it. Save this as approval.py.

RISKY = {"send_email", "issue_refund"}             # these wait for a person

TOOLS = {
    "read_database": lambda query: "3 customers overdue: ana@x.io, raj@x.io, lee@x.io",
    "send_email":    lambda to, subject: f"email sent to {to}",
    "issue_refund":  lambda to, amount: f"refunded {amount} to {to}",
}

def run(tool_calls, approve):
    """Runs the tool calls a model requested; risky ones need approve(name, args) == True."""
    log = []
    for name, args in tool_calls:
        if name in RISKY and not approve(name, args):
            result = "denied by operator"          # sent back to the model as the tool result
        else:
            result = TOOLS[name](**args)
        log.append((name, args.get("to", args.get("query")), result))
    return log

if __name__ == "__main__":
    calls = [("read_database", {"query": "overdue payments"}),       # what the stand-in model asked for
             ("send_email", {"to": "ana@x.io", "subject": "Payment reminder"}),
             ("issue_refund", {"to": "raj@x.io", "amount": 500}),
             ("send_email", {"to": "lee@x.io", "subject": "Payment reminder"})]
    operator = lambda name, args: name == "send_email"               # scripted human: emails yes, refunds no
    for name, target, result in run(calls, operator):
        print(f"{name:14} {target:17} -> {result}")

$ python approval.py
read_database  overdue payments  -> 3 customers overdue: ana@x.io, raj@x.io, lee@x.io
send_email     ana@x.io          -> email sent to ana@x.io
issue_refund   raj@x.io          -> denied by operator
send_email     lee@x.io          -> email sent to lee@x.io

The database read ran without asking, because reads are safe. Both emails went to the operator and were approved. The refund of 500 was denied, and the model receives “denied by operator” as that tool’s result, so it can tell the user rather than retry or assume success. approve is a parameter, not a call to input(), which is what made this runnable here. In production it becomes a Slack button, a ticket queue or a CLI prompt, and in tests it’s a scripted function like operator. Every decision is in log, which is the audit trail you’ll want when someone asks why a customer got an email.

Wiring This to a Real Provider

This section is reference code only. Unlike every example above, it was not executed, has no output block, and needs your own API key and the provider’s SDK (pip install anthropic or pip install openai). Each stand-in above is one function: a reviewer, a specialist, or the planner. To use a real model, replace its body with a model call that takes the subagent’s instructions as the system prompt and the subtask as the user message, and returns the text. The thread pool, the per-future error handling, parse_plan and the approval gate don’t change. Check the provider’s current documentation for model names before running it.

A subagent as an Anthropic Messages API call, reading ANTHROPIC_API_KEY from the environment:

import anthropic

client = anthropic.Anthropic()                     # reads ANTHROPIC_API_KEY

def run_subagent(instructions: str, task: str) -> str:
    response = client.messages.create(
        model="claude-opus-5-5",
        max_tokens=16000,
        system=instructions,                       # the subagent's role, e.g. "You are a security reviewer..."
        messages=[{"role": "user", "content": task}],
    )
    return "".join(block.text for block in response.content if block.type == "text")

# e.g. security = lambda code: run_subagent(SECURITY_INSTRUCTIONS, f"Review this code:\n{code}")

The same subagent with OpenAI’s Chat Completions API, reading OPENAI_API_KEY:

from openai import OpenAI

client = OpenAI()                                  # reads OPENAI_API_KEY

def run_subagent(instructions: str, task: str) -> str:
    response = client.chat.completions.create(
        model="YOUR_MODEL_NAME",
        messages=[{"role": "system", "content": instructions},
                  {"role": "user", "content": task}],
    )
    return response.choices[0].message.content

A single client can be shared across threads, so run_subagent drops straight into the ThreadPoolExecutor code above. With a real model, a reviewer returns prose rather than (severity, text) tuples, so either ask for a structured format and parse it as defensively as parse_plan does, or use the provider’s structured-output or tool-calling features to get a fixed shape. Real calls also fail in the ways partial.py simulates, with timeouts, rate limits and server errors, so the per-future error handling is not optional once there’s a network in the loop.

Practice

1. Add a fourth reviewer, docs, that reports LOW when the function has no docstring. Does the parallel run get slower?

   Solution

   import time
   from concurrent.futures import ThreadPoolExecutor
   from reviewers import CODE, REVIEWERS, synthesize
   
   def docs(code):
       """Stands in for a documentation-reviewer subagent."""
       time.sleep(0.5)
       return [] if '"""' in code else [("LOW", "function has no docstring")]
   
   reviewers = REVIEWERS | {"docs": docs}
   start = time.perf_counter()
   with ThreadPoolExecutor(max_workers=len(reviewers)) as pool:
       futures = {role: pool.submit(review, CODE) for role, review in reviewers.items()}
       findings = {role: future.result() for role, future in futures.items()}
   print(f"{len(reviewers)} reviewers in parallel: {time.perf_counter() - start:.2f}s")
   for severity, role, text in synthesize(findings):
       print(f"[{severity:8}] {role:15} {text}")

   $ python prac_docs.py
   4 reviewers in parallel: 0.50s
   [CRITICAL] security        SQL built with an f-string: injection risk
   [HIGH    ] performance     query inside a for loop: N+1 queries
   [MEDIUM  ] maintainability SELECT * couples the code to every column
   [LOW     ] docs            function has no docstring

   Still 0.50 s. Wall-clock time is set by the slowest subagent, not the number of subagents, as long as each one has a worker. The new finding sorts last because synthesize orders by severity, not by who reported it. The cost does grow, though: with real models, four reviewers are four billed calls, so a fourth reviewer is worth adding only if it finds things the others miss.

2. Make planning robust to worse replies than a fence: a reply cut off mid-JSON, steps with the wrong keys, and a plan with no usable step. Fall back to a single researcher step in each case.

   Solution

   import json
   from orchestrate import parse_plan
   
   DEFAULT = [{"specialist": "researcher", "subtask": "Research the whole topic"}]
   
   def safe_plan(reply):
       try:
           steps = parse_plan(reply)
       except (json.JSONDecodeError, KeyError, TypeError) as e:
           print(f"bad plan ({type(e).__name__}), using the default")
           return DEFAULT
       return steps or DEFAULT
   
   print(safe_plan('[{"specialist": "researcher", "subtask": "Collect dates"'))    # cut off
   print(safe_plan('[{"role": "researcher"}]'))                                     # wrong keys
   print(safe_plan('[{"specialist": "poet", "subtask": "A sonnet"}]'))              # nothing usable

   $ python prac_fallback.py
   bad plan (JSONDecodeError), using the default
   [{'specialist': 'researcher', 'subtask': 'Research the whole topic'}]
   bad plan (KeyError), using the default
   [{'specialist': 'researcher', 'subtask': 'Research the whole topic'}]
   skipped: no specialist 'poet' for 'A sonnet'
   [{'specialist': 'researcher', 'subtask': 'Research the whole topic'}]

   A truncated reply, typically from hitting the output token limit, fails to parse. Wrong keys fail with KeyError when parse_plan looks for specialist. A plan whose only step is unusable parses fine but leaves nothing to run, which steps or DEFAULT catches. In all three the orchestrator still does something sensible and says why. Another option is to send the error back to the planner and ask once for a corrected plan, with a limit on retries.

3. Approving every refund by hand doesn’t scale. Change the approval policy so refunds of 100 or less are approved automatically and only larger ones go to a person.

   Solution

   from approval import run
   
   def operator(name, args):
       if name == "issue_refund" and args["amount"] <= 100:
           return True                                 # small refunds: policy, no human needed
       print(f"  asking a human about {name} {args}")
       return name == "send_email"
   
   calls = [("issue_refund", {"to": "ana@x.io", "amount": 40}),
            ("issue_refund", {"to": "raj@x.io", "amount": 500}),
            ("send_email", {"to": "raj@x.io", "subject": "Refund update"})]
   for name, target, result in run(calls, operator):
       print(f"{name:14} {target:9} -> {result}")

   $ python prac_limit.py
     asking a human about issue_refund {'to': 'raj@x.io', 'amount': 500}
     asking a human about send_email {'to': 'raj@x.io', 'subject': 'Refund update'}
   issue_refund   ana@x.io  -> refunded 40 to ana@x.io
   issue_refund   raj@x.io  -> denied by operator
   send_email     raj@x.io  -> email sent to raj@x.io

   The refund of 40 went through without anyone being asked, and only the refund of 500 and the email reached a person. (The “asking” lines print first because run makes every decision before the loop prints the log.) The limit lives in operator, not in run or the prompt, so the policy can change without touching the agent. Note that the limit is per call: an agent issuing ten refunds of 90 passes it ten times, so a real policy also needs a running total per customer or per day.