Build an AI Agent Team in Python: Planner, Worker, Reviewer

An AI agent team in Python is three roles and a list: a planner that splits the goal into tasks, workers that do them, and a reviewer that says accept or reject. The list (a task board) is the shared state. Everything else is a loop with limits.

Here is the whole thing, about 90 lines, with a real run. One caveat up front: there is no live model in this post. I used a deterministic stub that returns canned answers, so the output shows the orchestration behaving, not any model's quality. Swap the stub for a real client and the control flow stays the same.

What does the team look like?

Three roles, each with one job and one output format.

The shared state is a Board holding Task objects with a status (todo, doing, done, failed), an attempt counter, the reviewer's last feedback, and the result. Agents never talk to each other directly. They read and write the board. That one rule makes handoffs inspectable: print the board and you know what happened.

For background on the pattern, see our multi-agent systems lesson and the earlier planner, executor, critic walkthrough. Public descriptions of the same split exist too. The swarms library documents a planner, a shared task queue, concurrent workers and a judge (swarms docs, read 2026-10-23), and the MadAgents paper has an orchestrator ask a reviewer for judgment after planning and before final reporting (arXiv 2601.21015, read 2026-10-23). This post does not reuse either design. It is the smallest version that still has real stop conditions.

The code

Standard library only. The StubLLM is the fake model: the planner returns three fixed tasks, the worker gives a too-short answer to t2 on its first try, and the worker always times out on t3 so we can watch failure handling.

from dataclasses import dataclass, field
import json


class StubLLM:
    """Deterministic stand-in for a model. Counts 'tokens' as words."""

    def __init__(self):
        self.calls = 0
        self.words = 0
        self.seen = {}

    def __call__(self, role, prompt):
        self.calls += 1
        self.words += len(prompt.split())
        if role == "planner":
            return json.dumps([
                {"id": "t1", "goal": "list three risks of the launch"},
                {"id": "t2", "goal": "draft a rollback plan"},
                {"id": "t3", "goal": "write a status email"},
            ])
        if role == "worker":
            tid = prompt.split("|")[0]
            n = self.seen[tid] = self.seen.get(tid, 0) + 1
            if tid == "t2" and n == 1:
                return "Rollback: revert the deploy."          # too thin on purpose
            if tid == "t3":
                raise TimeoutError("model timed out")           # always times out
            return f"Result for {tid} (attempt {n}): " + "detail " * 12
        if role == "reviewer":
            text = prompt.split("|", 1)[1]
            ok = len(text.split()) >= 10
            return json.dumps({"ok": ok, "reason": "ok" if ok else "too thin: add steps and an owner"})
        raise ValueError(role)


@dataclass
class Task:
    id: str
    goal: str
    status: str = "todo"   # todo, doing, done, failed
    attempts: int = 0
    feedback: str = ""
    result: str = ""


@dataclass
class Board:
    tasks: list = field(default_factory=list)
    log: list = field(default_factory=list)

    def note(self, msg):
        self.log.append(msg)
        print(msg)


def run_team(llm, goal, max_attempts=3, max_calls=12):
    board = Board()
    board.note(f"[planner] goal: {goal}")
    plan = json.loads(llm("planner", goal))
    board.tasks = [Task(p["id"], p["goal"]) for p in plan]
    board.note(f"[planner] {len(plan)} tasks")
    for t in board.tasks:
        while t.status in ("todo", "doing"):
            if llm.calls >= max_calls:               # stop condition: budget cap
                t.status = "failed"
                board.note(f"[stop] call budget {max_calls} hit at {t.id}")
                break
            if t.attempts >= max_attempts:           # stop condition: retries
                t.status = "failed"
                board.note(f"[stop] {t.id} failed after {t.attempts} attempts")
                break
            t.attempts += 1
            t.status = "doing"
            try:
                out = llm("worker", f"{t.id}|{t.goal}|{t.feedback}")
            except TimeoutError as e:                # failure handling
                board.note(f"[worker] {t.id} attempt {t.attempts} error: {e}")
                continue
            verdict = json.loads(llm("reviewer", f"{t.id}|{out}"))
            if verdict["ok"]:
                t.status, t.result = "done", out
                board.note(f"[reviewer] {t.id} accepted on attempt {t.attempts}")
            else:
                t.feedback = verdict["reason"]       # handoff back with feedback
                board.note(f"[reviewer] {t.id} rejected: {t.feedback}")
    return board

The driver is three lines: build a StubLLM, call run_team(llm, "prepare launch pack"), print each task's status. I ran it on 2026-10-23 with Python 3.13.

What did a full run produce?

Real output, pasted:

[planner] goal: prepare launch pack
[planner] 3 tasks
[reviewer] t1 accepted on attempt 1
[reviewer] t2 rejected: too thin: add steps and an owner
[reviewer] t2 accepted on attempt 2
[worker] t3 attempt 1 error: model timed out
[worker] t3 attempt 2 error: model timed out
[worker] t3 attempt 3 error: model timed out
[stop] t3 failed after 3 attempts
---
t1 done attempts= 1
t2 done attempts= 2
t3 failed attempts= 3
llm calls: 10 prompt words: 73

Read it like a trace:

• t1 passes first time. Two model calls (worker, reviewer).
• t2 is the case you built the reviewer for. The first draft was one thin sentence, the reviewer rejected it with a reason, the reason was passed to the worker as feedback, and attempt 2 passed. Four calls.
• t3 never produced anything, and the run still finished. The task is marked failed, the run does not hang, and a human can see exactly which task and why.
• Ten calls in total: one planner, then 2 + 4 + 3 for the tasks.

The stub's reviewer rule (at least 10 words) is deliberately silly. A real reviewer needs a rubric. See the checklist below.

How do you stop a team from looping forever?

Prompts do not stop loops. Code does. The script has three stops, and each one fired or could fire:

1. Attempts per task (max_attempts=3): this ended t3.
2. Call budget for the run (max_calls=12): the run used 10 calls. Drop it to 6 and the loop marks the task it is on as failed and says so. Real budgets should also count tokens and money. The token budgeting post shows how.
3. A terminal status: failed is a state, not an exception. Downstream code can check status and decide what to do.

One design choice worth copying: a reviewer rejection and a worker error both consume an attempt. Without that, a reviewer that always says no, or a worker that always errors, is an infinite loop with a polite log.

What should the reviewer actually check?

A reviewer that says "looks good" is a rubber stamp and costs you a call. Give it checks the worker can fail:

• Does the output answer the task's goal, not a nearby one?
• Is every required field or section present?
• Are claims traceable to something the worker was given? (Reject unsupported specifics.)
• Does it fit the format the next step expects, such as valid JSON?

Prefer deterministic checks in code (schema, length, banned strings, links resolve) and use a model reviewer only for judgment calls. Return a structured verdict with a reason, because the reason is the feedback the worker needs on retry. A reviewer from the same model family tends to share the worker's blind spots, so for high-stakes output a human or a different check is stronger.

Where does this need more than 90 lines?

The script is single-process and sequential. Be honest about what it lacks:

• Persistence. Kill the process and the board is gone. Write the board to JSON or SQLite after every status change.
• Parallelism. Workers run one at a time. For independent tasks, use a thread pool or asyncio and a lock around board updates.
• Real error types. It catches only TimeoutError. Real APIs raise rate-limit, auth and malformed-output errors that deserve different handling (back off, stop, retry with a stricter prompt).
• Tracing. Printing is fine for a demo. In production you want spans per call, as in the OpenTelemetry tracing post.

Adding all four yourself is the point where a framework starts to pay for itself.

Plain Python or a framework?

For a team this small, plain Python wins on readability and debugging: the whole control flow fits on one screen and a failing run is a stack trace you wrote. Frameworks win when you need durable state, branching graphs, interrupts for human approval or built-in tracing. The longer comparisons are in plain Python agent loop vs framework and the smolagents, CrewAI and LangGraph comparison. My rule: start in plain Python, and move when you catch yourself rewriting persistence or retries for the third time.

When does a single agent beat a team?

More often than the diagrams suggest. A team costs more calls (this toy run used ten for three tasks), loses context at every handoff, and adds failure modes you now have to monitor. Stay with one agent when:

• The task fits in one context window and takes a handful of steps.
• Steps depend on each other, so there is nothing to parallelise.
• You cannot yet say what the reviewer would check. A reviewer without a rubric is overhead.
• Latency matters. Every review pass is another round trip.

Add roles one at a time, in this order: a reviewer first (cheap and testable), then a planner when tasks vary, then parallel workers when tasks are truly independent. Our multi-agent workflows post goes through the trade-offs in more detail.

Wiring a real model in

Replace StubLLM with a function that takes (role, prompt) and calls your provider, with one system prompt per role and a JSON-only instruction for the planner and reviewer. Parse defensively: a malformed planner reply should count as a failed run, not a crash. Cheaper models often do the worker job fine while the reviewer is the role worth spending on, but measure that on your own tasks.

Disclosure: the AICredits and Hostinger links below are affiliate links. I ran only the stub-model script above; the real-model wiring is untested.

If you want a pay-in-rupees, OpenAI-compatible endpoint to point the worker at, AICredits takes INR via UPI from Rs 50 (base URL https://api.aicredits.in/v1). Any provider with an OpenAI-style API works the same way, and so does a local model through Ollama.

Running the team around the clock

A team that polls a queue or reacts to webhooks needs a host that stays up. A small VPS is enough for the orchestrator, since the heavy compute is the model API. Hostinger KVM 2 is one option; the Hostinger plan guide explains which size fits which workload, and any other VPS or a plain cron job on a machine you already own does the same job.