Build a Multi-Agent Code Review Pipeline

Use AgentTeam and TeamRunner to orchestrate a Lead → Worker → Reviewer pipeline that automatically decomposes a goal, executes tasks in parallel, and quality-gates every result before marking it done.

In this tutorial you'll build an automated code review pipeline powered by three collaborating AI agents:

A Lead agent breaks a broad goal ("Review the auth module") into concrete, self-contained tasks
Worker agents claim tasks from a shared board and execute them in parallel
A Reviewer agent reads each result and either approves it (→ Done) or rejects it (→ back to the queue for retry)

The pipeline runs until every task is Done — or until max_rounds is hit, whichever comes first.

Install the SDK first: pip install a3s-code. Configure ~/.a3s/config.hcl with your LLM provider key.

How it works

Goal
  │
  ▼
┌─────────────┐     JSON task list
│  Lead LLM   │ ─────────────────────────────────────┐
└─────────────┘                                       │
                                                      ▼
                                             ┌─────────────────┐
                                             │  TeamTaskBoard  │
                                             │  ┌───────────┐  │
                                             │  │  task-1   │  │ Open
                                             │  │  task-2   │  │ Open
                                             │  │  task-3   │  │ Open
                                             │  └───────────┘  │
                                             └────────┬────────┘
                                                      │ claim()
              ┌──────────────┬──────────────┐         │
              ▼              ▼              ▼         │
        ┌──────────┐  ┌──────────┐  ┌──────────┐    │
        │ Worker 1 │  │ Worker 2 │  │ Worker 3 │◄───┘
        └────┬─────┘  └────┬─────┘  └────┬─────┘
             │              │              │ complete()
             └──────────────┴──────────────┘
                                      │
                                      ▼ InReview
                              ┌───────────────┐
                              │   Reviewer    │
                              └───────┬───────┘
                                      │
                      ┌───────────────┴───────────────┐
                      ▼                               ▼
                   APPROVED                        REJECTED
                   (→ Done)              (→ back to Open, re-claimed)

Walkthrough

Project structure

Two files: agent.hcl for the LLM provider config, and review.py for the pipeline logic.

No queue or external workers needed — the team runner handles all concurrency internally using Tokio tasks.

Agent config (HCL)

agent.hcl only needs the provider and model. Team coordination is pure in-process — no Lane queue required unless you want to rate-limit LLM calls.

The same config is reused for all three roles (Lead, Worker, Reviewer). Each role gets its own AgentSession with independent context.

Build the team

Create an AgentTeam, add members with their roles, then wrap it in a TeamRunner.

The team is consumed when you create the runner — any further team.add_member() calls after this point will raise an error.

max_rounds — how many reviewer polling rounds before giving up on pending tasks (default: 10)
poll_interval_ms — how often workers and reviewer check for new work (default: 200 ms)

Bind sessions

Each member needs a Session that acts as its LLM executor. Call agent.session(workspace) once per role.

bind_session validates that the member ID exists in the team and stores the session internally. Workers that share similar tasks can even share the same session object — the runner will call session.send() concurrently across Tokio tasks.

Run the workflow

runner.run_until_done(goal) blocks until all tasks are Done (or max_rounds expires).

Internally:

Lead receives a planning prompt and responds with {"tasks": ["...", "..."]} — the runner parses this JSON automatically
Worker tasks are spawned concurrently, each polling the board
The reviewer task runs in parallel, approving or rejecting each InReview task

Inspect results

result.done_tasks contains every task the reviewer approved. Each TeamTask has:

id — unique task ID (sequential, e.g., task-1)
description — the original task text posted by the Lead
result — the Worker's output text
status — always "done" in this list

result.rejected_tasks holds tasks still "rejected" after max_rounds — these could not be approved in time.

Low-level board access

You can post tasks manually, bypass the Lead entirely, and drive the board directly. Useful when you already know the task list and just want parallel execution + review.

board.stats() returns (open, in_progress, in_review, done, rejected) — useful for progress bars or dashboards.

Full pipeline

The complete review.py. Run it:

python review.py "Review the auth module for security issues"

You'll see the Lead decompose the goal, workers execute tasks in parallel, and the reviewer validate each result before marking it Done.

project structure

code-review/
├── agent.hcl   # LLM provider config
└── review.py   # pipeline: team setup → bind → run → inspect

Going further

Add role-specific instructions

Give each role a focused identity using SessionOptions:

from a3s_code import SessionOptions

sec_opts = SessionOptions()
sec_opts.role = "You are a security engineer specializing in authentication systems."
sec_opts.guidelines = (
    "Focus on: JWT validation, token expiry, CSRF, session fixation, "
    "timing attacks, and information leakage in error messages. "
    "Be specific — cite file paths and line numbers when possible."
)

perf_opts = SessionOptions()
perf_opts.role = "You are a performance engineer."
perf_opts.guidelines = (
    "Focus on: O(n²) patterns, missing caching, blocking I/O, "
    "N+1 queries, and unnecessary cryptographic operations."
)

rev_opts = SessionOptions()
rev_opts.role = "You are a senior engineering lead doing final sign-off."
rev_opts.guidelines = (
    "Approve if the result is specific, actionable, and covers the task. "
    "Reject if the result is vague, incomplete, or off-topic."
)

runner.bind_session("worker-sec",  agent.session(".", sec_opts))
runner.bind_session("worker-perf", agent.session(".", perf_opts))
runner.bind_session("reviewer",    agent.session(".", rev_opts))

Scale workers

Add more workers with the same "worker" role to process tasks faster. All workers share the same board — the claim() method is atomic, so each task is executed exactly once:

for i in range(4):
    team.add_member(f"worker-{i}", "worker")

runner = TeamRunner(team)
for i in range(4):
    runner.bind_session(f"worker-{i}", agent.session("."))

TypeScript / Node.js

The same pipeline works in Node.js with identical semantics:

import { Agent, Team, TeamRunner } from '@a3s-lab/code';

const agent = await Agent.create('agent.hcl');

const team = new Team('code-review', { maxRounds: 8, maxTasks: 30, channelBuffer: 128, pollIntervalMs: 150 });
team.addMember('lead',        'lead');
team.addMember('worker-sec',  'worker');
team.addMember('worker-perf', 'worker');
team.addMember('reviewer',    'reviewer');

const runner = new TeamRunner(team);
runner.bindSession('lead',        agent.session('.'));
runner.bindSession('worker-sec',  agent.session('.'));
runner.bindSession('worker-perf', agent.session('.'));
runner.bindSession('reviewer',    agent.session('.'));

const result = await runner.runUntilDone(
  'Review the authentication module for security issues'
);
console.log(`Done: ${result.doneTasks.length}, Rounds: ${result.rounds}`);

Rust

For maximum performance — embed the pipeline directly in your Rust application:

use a3s_code_core::{
    Agent, SessionOptions,
    agent_teams::{AgentTeam, TeamConfig, TeamRole, TeamRunner},
};
use std::sync::Arc;

let agent = Agent::new("agent.hcl").await?;

let config = TeamConfig {
    max_rounds: 8,
    poll_interval_ms: 150,
    ..TeamConfig::default()
};

let mut team = AgentTeam::new("code-review", config);
team.add_member("lead",        TeamRole::Lead);
team.add_member("worker-sec",  TeamRole::Worker);
team.add_member("worker-perf", TeamRole::Worker);
team.add_member("reviewer",    TeamRole::Reviewer);

let mut runner = TeamRunner::new(team);
runner.bind_session("lead",        Arc::new(agent.session(".", None)?))?;
runner.bind_session("worker-sec",  Arc::new(agent.session(".", None)?))?;
runner.bind_session("worker-perf", Arc::new(agent.session(".", None)?))?;
runner.bind_session("reviewer",    Arc::new(agent.session(".", None)?))?;

let result = runner.run_until_done(
    "Review the authentication module for security issues"
).await?;

println!("Done: {} tasks, {} rounds", result.done_tasks.len(), result.rounds);
for task in &result.done_tasks {
    println!("  ✅ {}: {}", task.id, task.description);
}

Key concepts

| Concept | Description | |---------|-------------| | AgentTeam | The team registry — holds members and the shared TeamTaskBoard. Consumed when passed to TeamRunner. | | TeamRunner | The execution engine — binds AgentSession to each member and drives the Lead → Worker → Reviewer loop. | | TeamTaskBoard | Thread-safe task queue. States: Open → InProgress → InReview → Done (or Rejected → Open on retry). | | TeamConfig.max_rounds | Maximum reviewer polling rounds. When exceeded, remaining tasks are returned as rejected_tasks. | | claim() | Atomic claim — returns the next Open or Rejected task and marks it InProgress. Safe to call from multiple concurrent workers. | | Lead prompt | Built-in. Instructs the LLM to return {"tasks": ["...", "..."]} and nothing else. You do not write this prompt. | | Reviewer prompt | Built-in. Instructs the reviewer to respond APPROVED: <reason> or REJECTED: <feedback>. |