AI agents · 11 min read

How to build a multi-agent research system in Python: planner, researchers, source checker

The design behind an agent team that writes sourced reports: what to split, how agents pass work along, and how every sentence gets checked.

By Homingo Learn · 2 November 2026

A single agent with a search tool can answer a question. Ask it for a sourced report on a broad topic and it runs out of context, repeats searches and cites pages that never said what it claims. A multi-agent system fixes this by splitting the work: a planner breaks the topic into sub-questions, researchers work on them in parallel, a writer drafts from what they found, and a source checker tests every sentence before the report ships. This guide walks through that design in plain Python (3.9 or newer), with the contracts that keep it from falling apart.

When to split into agents (and when not to)

Split when the work has independent parts that can run in parallel, when one context window cannot hold all the material, or when a result needs an independent check (a reviewer should not be the author).
Do not split when the steps are strictly sequential and each needs the full previous output, or when the task fits comfortably in one call. Every handoff adds cost and a place to lose information.

Research reports pass the test: sub-questions are independent, sources are many, and claims need checking by someone other than the writer.

1. The planner returns a contract, not prose

Ask the planner for JSON in a fixed shape and validate it in code. Duplicates are dropped, the number of tasks is capped, and a bad reply gets the exact error back for one retry:

import json
from dataclasses import dataclass

class PlanError(ValueError):
    pass

@dataclass
class Task:
    id: str
    question: str

def parse_plan(reply: str, max_tasks: int = 6) -> list[Task]:
    try:
        data = json.loads(reply)
    except json.JSONDecodeError as e:
        raise PlanError(f"reply is not JSON: {e}") from e
    questions = data.get("sub_questions") if isinstance(data, dict) else None
    if not isinstance(questions, list) or not questions:
        raise PlanError('expected {"sub_questions": ["...", ...]} with at least one item')
    seen, tasks = set(), []
    for q in questions:
        if not isinstance(q, str) or not q.strip():
            raise PlanError("every sub-question must be a non-empty string")
        key = " ".join(q.lower().split())
        if key in seen:
            continue  # same question twice: keep one
        seen.add(key)
        tasks.append(Task(id=f"T{len(tasks) + 1}", question=q.strip()))
    if len(tasks) > max_tasks:
        raise PlanError(f"at most {max_tasks} sub-questions, got {len(tasks)}")
    return tasks

PLANNER_PROMPT = """Split this research topic into 3 to 6 distinct sub-questions that can be
researched independently. Answer with JSON only: {{"sub_questions": ["...", "..."]}}
Topic: {topic}"""  # doubled braces are literal braces for str.format

def plan(topic: str, call_model) -> list[Task]:
    prompt = PLANNER_PROMPT.format(topic=topic)
    reply = call_model(prompt)
    try:
        return parse_plan(reply)
    except PlanError as e:  # one retry, with the exact reason
        return parse_plan(call_model(f"{prompt}\n\nYour last reply was rejected: {e}. Answer again."))

2. Handoffs: scoped context, validated returns, isolated failures

Each researcher gets an envelope with only what it needs: its own question, the topic and the rules for its answer, not the whole conversation. Its reply is checked against a return contract (for example, a list of findings, each with a URL and an exact quote). Inside each researcher, an invalid reply or a transient error gets one retry with the reason as feedback. And one failing researcher must never stop the run:

import asyncio

async def run_researchers(tasks, research):
    """research(task) -> validated findings; may raise."""
    results = await asyncio.gather(*(research(t) for t in tasks), return_exceptions=True)
    done, failed = [], []
    for task, result in zip(tasks, results):
        (failed if isinstance(result, BaseException) else done).append((task, result))
    return done, failed  # failures are reported in the run trace, never fatal

# Tool calls are blocking HTTP: inside research(), call them with
#   await asyncio.to_thread(toolbox.call, agent, "wikipedia_search", query=q)
# so the researchers really run at the same time.

3. One door for every tool call

Researchers call real search tools; the Wikipedia and arXiv APIs are public and a good place to start. Route every call through one object that checks the arguments, shares a cache across the team, caps real calls per agent, and turns errors into observations the agent can act on instead of crashes:

import json
import urllib.parse
import urllib.request

def wikipedia_search(query: str, limit: int = 5) -> list[dict]:
    url = "https://en.wikipedia.org/w/api.php?" + urllib.parse.urlencode(
        {"action": "query", "list": "search", "srsearch": query, "srlimit": limit, "format": "json"})
    req = urllib.request.Request(url, headers={"User-Agent": "agent-team-tutorial/0.1 (you@example.com)"})
    with urllib.request.urlopen(req, timeout=15) as r:
        hits = json.load(r)["query"]["search"]
    return [{"title": h["title"],
             "url": "https://en.wikipedia.org/wiki/" + urllib.parse.quote(h["title"].replace(" ", "_"))}
            for h in hits]

TOOLS = {"wikipedia_search": {"required": ["query"], "fn": wikipedia_search}}

class ToolBox:
    def __init__(self, tools: dict, cap_per_agent: int = 8):
        self.tools, self.cap, self.cache, self.calls = tools, cap_per_agent, {}, {}

    def call(self, agent: str, name: str, **args):
        spec = self.tools.get(name)
        if spec is None:
            return {"error": f"unknown tool {name}"}
        missing = [k for k in spec["required"] if k not in args]
        if missing:
            return {"error": f"missing arguments: {missing}"}
        try:
            key = (name, json.dumps(args, sort_keys=True))
        except TypeError as e:
            return {"error": f"arguments must be JSON values: {e}"}
        if key in self.cache:  # shared by the team; cached answers do not count against the cap
            return self.cache[key]
        if self.calls.get(agent, 0) >= self.cap:
            return {"error": f"call cap of {self.cap} reached for {agent}"}
        self.calls[agent] = self.calls.get(agent, 0) + 1
        try:
            result = spec["fn"](**args)
        except Exception as e:  # the agent sees the error and can try something else
            return {"error": f"{name} failed: {e}"}
        self.cache[key] = result
        return result

4. Shared memory: one source of truth

Findings from all researchers land in one memory. Every page gets one canonical URL and one stable id ([S1], [S2] …), so the same page found twice is cited once. A note is kept only if its quote really appears on the fetched page:

from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

def canonical_url(url: str) -> str:
    p = urlsplit(url.strip())
    host = p.netloc.lower().removeprefix("www.")
    query = urlencode([(k, v) for k, v in parse_qsl(p.query) if not k.startswith("utm_")])
    return urlunsplit(("https", host, p.path.rstrip("/") or "/", query, ""))

class Memory:
    def __init__(self):
        self.ids: dict[str, str] = {}
        self.notes: list[dict] = []

    def source_id(self, url: str) -> str:
        key = canonical_url(url)
        if key not in self.ids:
            self.ids[key] = f"S{len(self.ids) + 1}"
        return self.ids[key]

    def add(self, task_id: str, url: str, claim: str, quote: str, page_text: str):
        q = " ".join(quote.split())
        if len(q) < 20 or q not in " ".join(page_text.split()):
            return None  # no real quote, or not on the page: drop the note
        note = {"task": task_id, "source": self.source_id(url), "claim": claim, "quote": quote}
        if note not in self.notes:
            self.notes.append(note)
        return note["source"]

5. The writer and the source checker

The writer drafts from the memory only and cites ids like [S3], one id per bracket ([S1][S3] for two). Then the source checker reads the draft sentence by sentence and puts each into one of four classes: ok (cited, and the cited quotes support it), uncited, unknown_id (cites an id that is not in memory) or unsupported (the quotes do not back it). The checker’s list drives one revision; after it, every sentence that is still not ok is removed.

import re

CITE = re.compile(r"\[(S\d+)\]")

def check_sentence(sentence: str, memory: Memory, supports) -> str:
    """supports(sentence, quotes) -> bool is a separate model call acting as judge."""
    ids = CITE.findall(sentence)
    if not ids:
        return "uncited"
    if any(i not in memory.ids.values() for i in ids):
        return "unknown_id"
    quotes = [n["quote"] for n in memory.notes if n["source"] in ids]
    return "ok" if supports(sentence, quotes) else "unsupported"

6. Run it like production: budgets, traces, evals

Budgets: a total token cap for the run, a cap per agent and a cap on tool calls, checked before every model call, so a looping agent stops instead of spending.
Traces: one JSON line per step (agent, task, tool, tokens, time, result class). When a report is wrong, the trace shows which agent went wrong.
Evals: a fixed set of topics and a score (for example, the share of sentences the checker marks ok), run in CI so a prompt change that makes reports worse fails the build.

Building with Claude Code too? Our guide to Claude Code skills shows how to package a workflow so the agent follows it every time.