Multi-agent code review

Run parallel specialist agents against the same code from different angles, then fan in to a synthesizer that produces a consolidated, prioritized report.

A single agent reviewing code for everything at once produces shallow findings. It has to split attention across security, performance, readability, and testability without going deep on any one.

Specialization through parallelism solves this: run four focused reviewers concurrently, then aggregate. The durable shape is a fan-out Graph where each branch is an agent() task with a domain-specific system prompt, feeding a fan-in node that normalizes findings into a severity-ranked report.

The pattern

               ┌─── security_review ───┐
code_input ───►├─── performance_review ─┼──► aggregate ──► report
               ├─── style_review ──────┤
               └─── testing_review ────┘

Four specialist agents start in parallel. Each produces a structured JSON list of findings. A non-LLM aggregation task collects and counts; a summary task ranks by severity. The workflow is a durable Flux execution, so findings are checkpointed as agents complete and the full result survives worker restarts.

Durable shape

The skeleton has five components:

  1. Specialist tasks — one @task per concern, each creating an agent() with a domain-scoped system prompt
  2. Structured output format — a shared prompt suffix that asks the LLM for a JSON array of findings, so the aggregator can parse any specialist’s output with one function
  3. Fan-in task — a @task that accepts *review_outputs, counts findings by severity, and separates findings from test suggestions
  4. Summary task — pulls the top critical/high findings into a plain recommendations list
  5. Graph wiring — start_with on each specialist node, add_edge from each specialist to aggregate, single end_with on the report node

Multiple start_with() calls power the fan-out here: the four specialist nodes have no incoming edges, so the Graph runs them in parallel and waits for all of them before the aggregate node executes.

After the specialists finish, no LLM is involved. Aggregation and summary are plain Python tasks, so those steps are cheap and deterministic.

Complete solution

Install the ai extra if you have not already, then start Ollama and pull a capable model:

pip install "flux-core[ai]"
ollama pull llama3.2
from __future__ import annotations

import json
import re
from datetime import datetime
from typing import Any

from flux import ExecutionContext, task, workflow
from flux.tasks import Graph
from flux.tasks.ai import agent


# ---------------------------------------------------------------------------
# Shared output format
# ---------------------------------------------------------------------------

REVIEW_OUTPUT_FORMAT = """
Provide your review as a JSON array of findings. Each finding should have:
- severity: "critical" | "high" | "medium" | "low"
- issue: string (description)
- line: number | null
- recommendation: string

Example: [{"severity": "high", "issue": "SQL injection", "line": 42,
           "recommendation": "Use parameterized queries"}]

Respond with ONLY the JSON array, no other text."""

TESTING_OUTPUT_FORMAT = """
Provide your suggestions as a JSON array. Each suggestion should have:
- priority: "high" | "medium" | "low"
- test_case: string (description)
- verifies: string (what it tests)
- importance: string (why it matters)

Respond with ONLY the JSON array, no other text."""


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def parse_llm_json_response(content: str) -> list[dict[str, Any]]:
    """Parse JSON from an LLM response, stripping markdown fences if present."""
    content = content.strip()
    if content.startswith("```"):
        lines = content.split("\n")
        content = "\n".join(lines[1:-1]) if len(lines) > 2 else "\n".join(lines[1:])
    content = content.replace("```", "").strip()

    for attempt in [
        content,
        content.replace("\n", "\\n").replace("\r", "\\r").replace("\t", "\\t"),
    ]:
        try:
            result = json.loads(attempt)
            return result if isinstance(result, list) else [result]
        except json.JSONDecodeError:
            pass

    for match in re.findall(r"\[[\s\S]*\]|\{[\s\S]*\}", content):
        try:
            result = json.loads(match)
            return result if isinstance(result, list) else [result]
        except json.JSONDecodeError:
            continue

    raise json.JSONDecodeError(
        "Could not parse JSON from LLM response", content, 0
    )


def _build_prompt(
    code: str,
    file_path: str | None,
    context: str | None,
    output_format: str,
) -> str:
    parts = []
    if file_path:
        parts.append(f"File: {file_path}")
    if context:
        parts.append(f"Context: {context}")
    context_str = "\n".join(parts) if parts else "No additional context provided."
    return f"{context_str}\n\nCode to review:\n```\n{code}\n```\n\n{output_format}"


def _parse_review(
    agent_name: str, raw: str, key: str = "findings"
) -> dict[str, Any]:
    try:
        items = parse_llm_json_response(raw)
        return {"agent": agent_name, "status": "success", key: items}
    except json.JSONDecodeError as exc:
        return {"agent": agent_name, "status": "parse_error", "error": str(exc), key: []}


# ---------------------------------------------------------------------------
# Specialist agents (Graph nodes)
# ---------------------------------------------------------------------------

@task
async def run_security(input_data: dict[str, Any]) -> dict[str, Any]:
    reviewer = await agent(
        "You are a security code reviewer with expertise in finding vulnerabilities. "
        "Analyze code for: SQL injection, XSS, authentication issues, hardcoded secrets, "
        "input validation, command injection, path traversal, insecure cryptography, "
        "race conditions, and sensitive data exposure.",
        model="ollama/llama3.2",
        name="security_review",
    )
    prompt = _build_prompt(
        input_data["code"],
        input_data.get("file_path"),
        input_data.get("context"),
        REVIEW_OUTPUT_FORMAT,
    )
    return _parse_review("security", await reviewer(prompt))


@task
async def run_performance(input_data: dict[str, Any]) -> dict[str, Any]:
    reviewer = await agent(
        "You are a performance optimization expert. "
        "Review code for: algorithm efficiency, unnecessary loops, inefficient data "
        "structures, memory leaks, database query optimization, missing caching, "
        "redundant computations, I/O bottlenecks, and blocking operations.",
        model="ollama/llama3.2",
        name="performance_review",
    )
    prompt = _build_prompt(
        input_data["code"],
        input_data.get("file_path"),
        input_data.get("context"),
        REVIEW_OUTPUT_FORMAT,
    )
    return _parse_review("performance", await reviewer(prompt))


@task
async def run_style(input_data: dict[str, Any]) -> dict[str, Any]:
    reviewer = await agent(
        "You are a code quality and style expert. "
        "Review code for: readability, naming conventions, organization, documentation, "
        "DRY violations, function complexity, magic numbers, error handling, type hints, "
        "and PEP 8 compliance.",
        model="ollama/llama3.2",
        name="style_review",
    )
    prompt = _build_prompt(
        input_data["code"],
        input_data.get("file_path"),
        input_data.get("context"),
        REVIEW_OUTPUT_FORMAT,
    )
    return _parse_review("style", await reviewer(prompt))


@task
async def run_testing(input_data: dict[str, Any]) -> dict[str, Any]:
    reviewer = await agent(
        "You are a testing and quality assurance expert. "
        "Suggest: critical test cases, edge cases, error conditions, integration tests, "
        "mock requirements, test data, missing coverage, and regression tests.",
        model="ollama/llama3.2",
        name="testing_review",
    )
    prompt = _build_prompt(
        input_data["code"],
        input_data.get("file_path"),
        input_data.get("context"),
        TESTING_OUTPUT_FORMAT,
    )
    return _parse_review("testing", await reviewer(prompt), key="suggestions")


# ---------------------------------------------------------------------------
# Aggregation and summary (plain Python tasks — no LLM)
# ---------------------------------------------------------------------------

@task
async def collect_reviews(*review_outputs: dict[str, Any]) -> dict[str, Any]:
    """Fan-in: merge all specialist findings into one structure."""
    all_findings: list[dict[str, Any]] = []
    all_suggestions: list[dict[str, Any]] = []
    counts: dict[str, int] = {"critical": 0, "high": 0, "medium": 0, "low": 0}
    completed = failed = 0

    for review in review_outputs:
        agent_name = review.get("agent", "unknown")
        if review.get("status") == "success":
            completed += 1
            for f in review.get("findings", []):
                counts[f.get("severity", "low")] = counts.get(f.get("severity", "low"), 0) + 1
                all_findings.append({"agent": agent_name, **f})
            for s in review.get("suggestions", []):
                all_suggestions.append({"agent": agent_name, **s})
        else:
            failed += 1

    return {
        "all_findings": all_findings,
        "all_suggestions": all_suggestions,
        "counts": counts,
        "agents_completed": completed,
        "agents_failed": failed,
    }


@task
async def generate_report(aggregated: dict[str, Any]) -> dict[str, Any]:
    """Produce the final severity-ranked report."""
    counts = aggregated["counts"]
    top_issues = [
        f"{f['severity'].upper()}: {f['issue']}"
        + (f" (line {f['line']})" if f.get("line") else "")
        for f in aggregated["all_findings"]
        if f.get("severity") in ("critical", "high")
    ]
    return {
        "summary": {
            "total_issues": sum(counts.values()),
            **counts,
            "agents_completed": aggregated["agents_completed"],
            "agents_failed": aggregated["agents_failed"],
        },
        "top_issues": top_issues[:10],
        "test_suggestions_count": len(aggregated["all_suggestions"]),
    }


# ---------------------------------------------------------------------------
# Workflow
# ---------------------------------------------------------------------------

@workflow
async def multi_agent_code_review(ctx: ExecutionContext[dict[str, Any]]):
    """
    Parallel multi-agent code review.

    Input:
        {
            "code": "Source code string (required)",
            "file_path": "Optional path for context",
            "context":   "Optional description of what the code does"
        }
    """
    raw = ctx.input or {}
    if not raw.get("code"):
        return {"error": "No code provided", "execution_id": ctx.execution_id}

    graph = (
        Graph("code_review")
        .add_node("security",    run_security)
        .add_node("performance", run_performance)
        .add_node("style",       run_style)
        .add_node("testing",     run_testing)
        .add_node("aggregate",   collect_reviews)
        .add_node("report",      generate_report)
        .start_with("security")
        .start_with("performance")
        .start_with("style")
        .start_with("testing")
        .add_edge("security",    "aggregate")
        .add_edge("performance", "aggregate")
        .add_edge("style",       "aggregate")
        .add_edge("testing",     "aggregate")
        .add_edge("aggregate",   "report")
        .end_with("report")
    )

    start = datetime.now()
    report = await graph(raw)
    report["metadata"] = {
        "execution_id": ctx.execution_id,
        "execution_time_s": (datetime.now() - start).total_seconds(),
        "code_length": len(raw["code"]),
    }
    return report


if __name__ == "__main__":
    code_under_review = '''
def login_user(username, password):
    query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
    result = db.execute(query)
    return result.fetchone()
'''

    result = multi_agent_code_review.run({
        "code": code_under_review,
        "file_path": "auth.py",
        "context": "User authentication module",
    })

    if result.has_failed:
        print(f"Review failed: {result.output}")
    else:
        output = result.output
        summary = output["summary"]
        print(f"Total issues: {summary['total_issues']}")
        print(f"  Critical: {summary['critical']}")
        print(f"  High:     {summary['high']}")
        print(f"  Medium:   {summary['medium']}")
        print(f"  Low:      {summary['low']}")
        print(f"\nTop issues:")
        for issue in output["top_issues"]:
            print(f"  {issue}")

Why this holds up under load

Each specialist agent runs as a first-class Flux task:

collect_reviews and generate_report make no LLM calls, so they replay instantly if ever needed.

Connecting to skills

The skills_code_review.py source example shows a complementary pattern. Instead of separate specialist tasks, a single agent carries a SkillCatalog with a security skill and a performance skill, and the LLM decides which to activate based on the request:

from flux.tasks.ai import Skill, SkillCatalog, agent

security_skill = Skill(
    name="security-reviewer",
    description="Reviews code for security vulnerabilities including injection, XSS, "
                "authentication flaws, and OWASP Top 10 issues. Use when the user asks "
                "for a security review or mentions vulnerabilities.",
    instructions=(
        "You are a security expert. Review the provided code for security issues.\n\n"
        "Check for:\n"
        "1. SQL injection vulnerabilities\n"
        "2. Cross-site scripting (XSS)\n"
        "3. Authentication and authorization flaws\n"
        "4. Sensitive data exposure\n"
        "5. Input validation issues\n\n"
        "For each issue found, provide:\n"
        "- Severity (Critical/High/Medium/Low)\n"
        "- Description of the vulnerability\n"
        "- A concrete fix with code\n"
    ),
)

performance_skill = Skill(
    name="performance-reviewer",
    description="Reviews code for performance issues including algorithmic complexity, "
                "memory leaks, unnecessary allocations, and database query optimization.",
    instructions=(
        "You are a performance engineering expert. Review for:\n"
        "1. Algorithmic complexity (O(n^2) or worse)\n"
        "2. Unnecessary memory allocations\n"
        "3. N+1 query patterns\n"
        "4. Missing caching opportunities\n"
        "5. Blocking I/O in async contexts\n\n"
        "For each issue: Impact, Description, Optimized alternative with code."
    ),
)

catalog = SkillCatalog([security_skill, performance_skill])

reviewer = await agent(
    "You are a code review assistant. Use the appropriate skill for the type of "
    "review requested.",
    model="ollama/llama3.2",
    name="code-reviewer",
    skills=catalog,
)
review = await reviewer("Security review:\n\n```python\n...\n```")

The two patterns serve different needs. Skills let a single agent switch review modes based on what the caller asks for. The parallel Graph pattern covers all dimensions at once without waiting for any one reviewer to finish. Use skills when the review type comes from user input. Use the Graph pattern when every submission should get full coverage regardless of input.

Running the workflow

Register and run against a Flux server:

flux workflow register multi_agent_code_review.py
flux workflow run multi_agent_code_review '{"code": "def unsafe(u): return db.execute(f\"SELECT * FROM t WHERE id={u}\")", "file_path": "api.py"}'

Or run inline during development:

python multi_agent_code_review.py

Poll for the result once the workflow is dispatched to a worker:

flux workflow status multi_agent_code_review <execution_id>
flux execution show <execution_id> --detailed

Variations

What’s next