When Two Agents Beat One: A LangGraph Code Review Copilot Built Both Ways

"Multi-agent" is the buzzword of the moment. Demos show five AI personas chatting in a simulated room to scrape a site and write a blog post, with names like Researcher, Writer and Editor and a terminal full of agents debating each other. It looks impressive, and it isn't how reliable software gets built.

Adding agents doesn't make a system smarter on its own. It multiplies the ways things can fail. Take the basic arithmetic: one agent that does its job correctly 90% of the time has a reliability of 0.90. Chain three of them and the probabilities multiply, so the baseline falls to about 72%. Latency doubles, API spend triples, and the final output is no better.

The question worth answering is when more than one agent actually pays off. A good way to find out is to build an AI code review assistant twice. The first version is a conventional single-agent reviewer. The second splits the work between an Analyzer and a Risk Reviewer, using LangGraph features such as conditional edges, interrupt() and Command-based resume to bring a human in when needed. The comparison shows why the single agent overlooks a serious billing flaw and why the pair catches it.

Why code review is hard for one prompt

A good review needs two quite different kinds of thinking.

The first is comprehension. You read the diff, map the functions that changed, work out which files are affected and infer what the author meant. The goal is a mental model of the new feature.

The second is adversarial. You hunt for edge cases, absent tests and logic holes, and you try to break the code.

Ask one LLM to do both in a single prompt and it struggles. Summarising and critiquing in the same pass tends to favour coherence over doubt. Once the model settles on a story about what the code does, it invests less in attacking that story, and it falls back on minor style remarks such as missing docstrings or naming conventions.

A concrete case makes this clear. Imagine a mock pull request against a billing system in which the author adds a discount feature to an invoice calculator. This is the diff both designs receive:

--- a/src/billing/invoice.py
+++ b/src/billing/invoice.py
@@ -42,7 +42,9 @@
 class InvoiceService:
-    def calculate_total(self, items):
-        return sum(i.price * i.qty for i in items)
+    def calculate_total(self, items, discount_pct=0):
+        subtotal = sum(i.price * i.qty for i in items)
+        return subtotal * (1 - discount_pct)

--- a/src/billing/api.py
+++ b/src/billing/api.py
@@ -18,6 +18,8 @@
 @router.post("/invoice")
 def create_invoice(req: InvoiceRequest):
+    discount = req.discount_pct   # NEW: from user input
     svc = InvoiceService()
-    total = svc.calculate_total(req.items)
+    total = svc.calculate_total(req.items, discount)
     return {"total": total}

--- a/config/feature_flags.yaml
+++ b/config/feature_flags.yaml
@@ -5,3 +5,4 @@
 flags:
   new_dashboard: true
+  discount_billing: true   # rollout: 100% immediately

Look closely and several serious problems stand out. The discount_pct value arrives straight from user input at the API layer and flows into the calculation with no validation. A user could send a negative discount to raise the price, or a value above 1 to get a negative total. The tests don't cover the new path. And the feature flag is set to reach 100% of users immediately. This is a pricing path ripe for abuse. How does a typical single-agent reviewer deal with it?

Design one: a single-agent reviewer

The agent's toolbox comes first: one tool to get the diff, one to open files, one to look up symbols and one to report which tests exist.

from langchain_core.tools import tool
import textwrap

MOCK_DIFF = """...""" # The diff shown above
MOCK_FILES = {
    "src/billing/invoice.py": "class InvoiceService:\n    def calculate_total(self, items, discount_pct=0):\n        subtotal = sum(i.price * i.qty for i in items)\n        return subtotal * (1 - discount_pct)\n",
    "src/billing/api.py": '@router.post("/invoice")\ndef create_invoice(req: InvoiceRequest):\n    discount = req.discount_pct\n    svc = InvoiceService()\n    total = svc.calculate_total(req.items, discount)\n    return {"total": total}\n',
    "tests/test_billing.py": "def test_calculate_total():\n    # only tests no-discount path\n    assert svc.calculate_total(items) == 300\n",
}

@tool
def get_diff(pr_id: str) -> str:
    """Fetch the PR diff."""
    return MOCK_DIFF

@tool
def read_file(path: str) -> str:
    """Read a file from the repo."""
    return MOCK_FILES.get(path, f"FILE NOT FOUND: {path}")

@tool
def search_symbol(name: str) -> str:
    """Search for a symbol across the codebase."""
    if "discount" in name.lower():
        return "Found: InvoiceService.calculate_total(discount_pct) — src/billing/invoice.py:43"
    return f"No results for '{name}'"

@tool
def list_tests(path: str) -> str:
    """List test files covering a source path."""
    if "billing" in path:
        return "tests/test_billing.py — covers calculate_total (no-discount path only)"
    return "No tests found"

ALL_TOOLS = [get_diff, read_file, search_symbol, list_tests]

Then the state. A TypedDict keeps things compatible with LangGraph, and it holds the PR ID, message history, raw diff and a list of findings. A Pydantic model describes each finding, and it forces the model to classify what it finds with fixed literals, which makes later processing much easier.

from typing import Literal
from typing_extensions import TypedDict
from pydantic import BaseModel
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph

class ReviewFinding(BaseModel):
    severity: Literal["critical", "high", "medium", "low", "info"]
    category: Literal["bug", "regression", "missing_test", "rollout_risk",
                       "security", "edge_case", "style"]
    file: str
    description: str
    confidence: Literal["high", "medium", "low"]

class SingleAgentState(TypedDict):
    pr_id: str
    messages: list
    diff: str
    findings: list[ReviewFinding]

llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)

Next come the nodes. The graph is a straight line: a fetch node that loads context, then a review node that runs the prompt.

def sa_fetch(state: SingleAgentState) -> dict:
    diff = get_diff.invoke({"pr_id": state["pr_id"]})
    tests = list_tests.invoke({"path": "src/billing"})
    return {"diff": diff, "messages": [
        AIMessage(content=f"Diff loaded. Test coverage: {tests}")
    ]}

def sa_review(state: SingleAgentState) -> dict:
    """Single agent does BOTH jobs: summarize + critique in one pass."""
    prompt = f"""\
You are a senior code reviewer. Read this PR diff, summarize the change,
and produce a list of risk findings. Be specific.

DIFF:
{state['diff']}

Respond with JSON: {{"summary": "...", "findings": [
  {{"severity": "...", "category": "...", "file": "...",
    "description": "...", "confidence": "..."}}
]}}
"""
    resp = llm.invoke([SystemMessage(content="You are a code review bot."),
                       HumanMessage(content=prompt)])

    # In a real app we parse the JSON response here.
    # We simulate the typical single-agent output for this diff.
    findings = [
        ReviewFinding(
            severity="medium",
            category="missing_test",
            file="tests/test_billing.py",
            description="Add unit tests for the new discount_pct parameter.",
            confidence="high"
        ),
        ReviewFinding(
            severity="low",
            category="style",
            file="src/billing/invoice.py",
            description="Consider adding type hints to the items list.",
            confidence="high"
        )
    ]
    return {"findings": findings, "messages": [resp]}

def build_single_agent():
    g = StateGraph(SingleAgentState)
    g.add_node("fetch", sa_fetch)
    g.add_node("review", sa_review)
    g.add_edge(START, "fetch")
    g.add_edge("fetch", "review")
    g.add_edge("review", END)
    return g.compile()

(In this demonstration the findings are hard-coded stand-ins for the kind of output such a reviewer typically produces, so treat the comparison as an illustration, not a benchmark.)

That's roughly how many first-pass AI reviewers get built: pass in the diff, ask for bugs. Run against the mock PR, the agent notices the missing test and suggests type hints. It misses the unvalidated input and the rollout risk.

The reason is in the prompt. Asked to summarise the change and then list risks, the model spends its attention explaining that a discount feature was added. By the time it begins the findings array it is already in a descriptive mood, looking at surface syntax and not trying to break anything. The fix is to separate the synthesis job from the adversarial one, which means two agents.

The handoff: use a schema and not prose

The most common mistake in multi-agent design is letting agents talk to each other in free text. If agent A writes three paragraphs and hands them to agent B, B has to parse the text, guess what matters and build on a fuzzy base. That's how context degrades.

Agents are functions, and functions should pass typed objects. For a multi-agent system to be dependable, define the exact schema of what moves between them.

In the review assistant, the first agent is the Analyzer, and its job is to map the PR. It must produce a ChangeModel, an object that forces it to pull out specific metadata about the change.

class ChangedInterface(BaseModel):
    file: str
    symbol: str
    change_type: Literal["added", "modified", "removed"]
    description: str

class ChangeModel(BaseModel):
    """Analyzer → Reviewer handoff. Structured, not prose."""
    purpose: str
    impacted_files: list[str]
    changed_interfaces: list[ChangedInterface]
    config_changes: list[str]
    migration_risk: bool
    assumptions: list[str]
    tests_touched: list[str]
    tests_likely_needed: list[str]

Each field serves the downstream agent. The list of changed interfaces gives the second agent a quick lookup of what matters. The config changes list makes the Analyzer separate application logic from deployment logic.

The assumptions array deserves particular attention. The Analyzer has to spell out what the code's author is assuming, such as that discount_pct is a fraction between 0 and 1. It isn't criticising anything. It's just stating facts. That structured object becomes the Reviewer's exact input.

Design two: Analyzer and Risk Reviewer

The new state schema holds the ChangeModel plus a boolean conflict flag, which later decides whether to escalate to a person.

class TwoAgentState(TypedDict):
    pr_id: str
    messages: list
    diff: str
    change_model: ChangeModel | None
    findings: list[ReviewFinding]
    conflict: bool

The fetch node stays the same, grabbing the diff and test coverage. The Analyzer node comes next, with a tightly constrained prompt. The model is explicitly told not to look for bugs, so it concentrates purely on understanding.

ANALYZER_PROMPT = """\
You are the Analyzer agent. Your ONLY job: understand the change.
Do NOT critique. Do NOT hunt for bugs. Just map what changed.
Produce a structured change model."""

def ta_analyzer(state: TwoAgentState) -> dict:
    """Analyzer: compress & clarify. Optimizes for coherence."""

    # We use LLM structured output to fill the ChangeModel.
    # For this demonstration, we simulate the accurate analysis.
    cm = ChangeModel(
        purpose="Add discount percentage support to invoice billing",
        impacted_files=["src/billing/invoice.py", "src/billing/api.py",
                        "config/feature_flags.yaml"],
        changed_interfaces=[
            ChangedInterface(file="src/billing/invoice.py",
                symbol="InvoiceService.calculate_total",
                change_type="modified",
                description="Added discount_pct param (default 0)"),
            ChangedInterface(file="src/billing/api.py",
                symbol="create_invoice",
                change_type="modified",
                description="Reads discount_pct from request, passes to service"),
        ],
        config_changes=["discount_billing flag added, 100% rollout"],
        migration_risk=False,
        assumptions=[
            "discount_pct is expected to be 0..1 (fraction, not percentage)",
            "No existing callers pass discount_pct yet",
            "Feature flag controls visibility, not the calculation",
        ],
        tests_touched=[],
        tests_likely_needed=[
            "test discount path in calculate_total",
            "test boundary: discount_pct = 0, 1, >1, <0",
            "test API validation of discount_pct input",
        ],
    )
    return {
        "change_model": cm,
        "messages": [AIMessage(content=f"Analyzer: change model built. "
                               f"{len(cm.changed_interfaces)} interfaces changed, "
                               f"{len(cm.assumptions)} assumptions made.")],
    }

The Analyzer reads the diff and records what is really there: the new parameter, the new feature flag, and the implicit assumption that the discount lies between 0 and 1.

Now the object goes to the Risk Reviewer, whose prompt is deliberately adversarial. It's told to distrust the Analyzer and to hunt for flaws.

REVIEWER_PROMPT = """\
You are the Risk Reviewer. The Analyzer gave you a change model.
DISTRUST it. Your job: find what's missing, broken, or dangerous.
Challenge every assumption. Check for missing tests, regressions,
rollout risks, and security issues."""

def ta_reviewer(state: TwoAgentState) -> dict:
    """Reviewer: expand & challenge. Optimizes for skepticism."""
    cm = state["change_model"]
    findings: list[ReviewFinding] = []

    # The Reviewer iterates through the Analyzer's assumptions
    for a in cm.assumptions:
        if "0..1" in a:
            findings.append(ReviewFinding(
                severity="critical",
                category="security",
                file="src/billing/api.py",
                description="ASSUMPTION CHALLENGED: discount_pct comes from user input "
                    "(req.discount_pct) with NO validation. Values <0 or >1 break billing. "
                    "Negative discount = price increase beyond subtotal. "
                    "Value >1 = negative total.",
                confidence="high"))

    # The Reviewer checks the test mapping
    if not cm.tests_touched and cm.tests_likely_needed:
        findings.append(ReviewFinding(
            severity="high",
            category="missing_test",
            file="tests/test_billing.py",
            description=f"NO tests touched but {len(cm.tests_likely_needed)} needed: "
                + "; ".join(cm.tests_likely_needed),
            confidence="high"))

    # The Reviewer checks the config changes
    for cc in cm.config_changes:
        if "100%" in cc:
            findings.append(ReviewFinding(
                severity="high",
                category="rollout_risk",
                file="config/feature_flags.yaml",
                description="Feature flag at 100% from day one — no gradual rollout. "
                    "Combined with unvalidated discount input, this is a billing incident risk.",
                confidence="high"))

    # The Reviewer looks for edge cases in the logic flow
    findings.append(ReviewFinding(
        severity="medium",
        category="edge_case",
        file="src/billing/api.py",
        description="Feature flag controls visibility but calculate_total always applies "
            "discount. If flag is off but API still receives discount_pct, discount "
            "is silently applied.",
        confidence="medium"))

    # We determine if there is a conflict worth escalating
    conflict = any(f.severity in ("critical", "high") for f in findings)

    return {
        "findings": findings,
        "conflict": conflict,
        "messages": [AIMessage(content=f"Reviewer: {len(findings)} findings, "
                               f"conflict={conflict}")],
    }

Because the Reviewer doesn't burn tokens working out which files changed or what the PR is for, its whole context goes into breaking the logic. It reads the assumption that the discount is expected to be 0 to 1, checks the diff, finds no validation code, and flags a high-severity input validation bug.

This is one of the few places where multiple agents earn their keep. One builds the target and the other shoots at it.

Arbitration with a human in the loop

The Reviewer has produced severe findings that the PR author missed. Posting them straight to GitHub would be risky, though. If the Reviewer is hallucinating, it will irritate the team. A sensible production addition is a merge node that evaluates the findings and decides whether a human should arbitrate.

def ta_merge(state: TwoAgentState) -> dict:
    """Merge findings. If conflict, surface for human review."""
    if state["conflict"]:
        return {
            "messages": [AIMessage(content=(
                "⚠ CONFLICT: Reviewer found critical/high issues. "
                "Routing to human review."
            ))],
        }
    return {
        "messages": [AIMessage(content="Findings merged. No escalation needed.")],
    }

def route_after_merge(state: TwoAgentState) -> str:
    return "escalate" if state["conflict"] else "emit"

When the conflict flag is true, flow goes to an escalation node. That's where conditional edges, interrupt() and Command-based resume come in.

from langgraph.types import Command, interrupt

def ta_escalate(state: TwoAgentState) -> Command:
    """Human-in-the-loop for conflicting findings."""

    # The interrupt function pauses execution and surfaces data to the caller
    decision = interrupt({
        "kind": "review_conflict",
        "pr_id": state["pr_id"],
        "change_model": state["change_model"].model_dump(),
        "findings": [f.model_dump() for f in state["findings"]],
        "prompt": "Review findings. Respond: "
                  '{"action":"accept_all"} or {"action":"override","drop_indices":[...]}',
    })

    # Execution resumes here when the human provides input
    action = decision.get("action", "accept_all")

    if action == "override":
        drop = set(decision.get("drop_indices", []))
        kept = [f for i, f in enumerate(state["findings"]) if i not in drop]

        # We use Command to update state and dynamically route to the next node
        return Command(update={"findings": kept}, goto="emit")

    return Command(goto="emit")

Reaching the interrupt() call pauses execution. The graph serialises its current state and stores it through the configured checkpointer, which is MemorySaver in a local demo. In production you'd use a durable saver backed by Postgres so paused state survives server restarts.

Your backend can then send a Slack message to a senior engineer with the structured findings. They read the critical warning about the unvalidated discount and, if they agree, press an "Accept All" button. When the webhook reaches your server, it continues the run by handing the graph a Command that carries the resume payload.

# How you resume the graph from your backend API
ta.invoke(Command(resume={"action": "accept_all"}), config)

The graph wakes up, processes the human decision, adjusts the findings if needed and moves to the final emit node.

def ta_emit(state: TwoAgentState) -> dict:
    """Final output: formatted review."""
    lines = [f"=== Code Review: PR {state['pr_id']} ==="]
    if state["change_model"]:
        lines.append(f"Purpose: {state['change_model'].purpose}")
    lines.append(f"Findings ({len(state['findings'])}):")

    for i, f in enumerate(state["findings"]):
        lines.append(f"  [{f.severity}] ({f.category}) {f.file}")
        lines.append(f"    {f.description}")

    return {"messages": [AIMessage(content="\n".join(lines))]}

Wiring the graph

The nodes are assembled into a StateGraph. No static edge is needed out of the escalate node, because ta_escalate returns a Command(goto="emit") and LangGraph routes dynamically.

from langgraph.checkpoint.memory import MemorySaver

def build_two_agent():
    g = StateGraph(TwoAgentState)

    g.add_node("fetch", ta_fetch)
    g.add_node("analyzer", ta_analyzer)
    g.add_node("reviewer", ta_reviewer)
    g.add_node("merge", ta_merge)
    g.add_node("escalate", ta_escalate)
    g.add_node("emit", ta_emit)

    g.add_edge(START, "fetch")
    g.add_edge("fetch", "analyzer")
    g.add_edge("analyzer", "reviewer")
    g.add_edge("reviewer", "merge")

    g.add_conditional_edges("merge", route_after_merge,
        {"escalate": "escalate", "emit": "emit"})

    g.add_edge("emit", END)

    # A checkpointer is required to use interrupt()
    # Use MemorySaver for local testing, Postgres for production
    return g.compile(checkpointer=MemorySaver())

Comparing the two

Put the two reviewers next to each other on this PR and the gap is obvious.

The lone agent turns up two issues. It recommends adding unit tests and suggests type hints. It behaves like a polite junior developer who focuses on syntax.

The pair finds four. It flags the unvalidated input as a critical business-logic flaw. It notes that negative discounts would raise prices. It marks the immediate 100% rollout as a billing incident risk. And it identifies an edge case in which the discount is applied silently even when the feature flag is off.

Multi-agent design doesn't guarantee better reviews every time. It helped here because the decomposition is unusually clean. The second agent added real signal and not noise, catching the business-logic flaws that take production down. It worked because the cognitive load was split. The Analyzer did the heavy reading and mapped the interfaces, and the Reviewer received a clean ChangeModel and could devote itself to scepticism.

When not to use multiple agents

The pair is powerful, and the temptation will be to rewrite every single agent as a multi-agent workflow. Resist it.

Extra agents add latency. The two-agent reviewer takes about twice as long because its LLM calls are sequential, costs about twice as much in tokens, and brings a more complex state schema and handoff logic. Use the pattern only if the task fits one of three situations.

Adversarial work. If a system must both produce and critique content, split those roles. Generator and discriminator is the best-proven multi-agent design. One agent writes code and another writes tests to break it. One drafts an email and another checks it for compliance problems.

Strict tool isolation. An agent that searches the public web and an agent that runs database queries shouldn't share a brain. Let a Researcher gather public data and pass a sanitised summary to an Executor. That creates a hard security boundary. If a web page contains a prompt injection, only the Researcher is compromised and the Executor stays isolated.

Asymmetric context scaling. Sometimes a huge amount of data feeds a tiny decision. A fleet of cheap, fast workers can read 100 log files in parallel, each extract the relevant error lines and pass a small summary to a single expensive supervisor that makes the call.

If a task fits none of these, stay with one agent. A single agent with a well-crafted prompt, tight tool constraints and a clean state schema will beat a messy multi-agent setup every time.