当两个智能体战胜一个：一款可双向使用的LangGraph代码审查辅助工具
在LangGraph中，先将一名代码审查员构建为单一智能体，再将其构建为兼具分析功能与风险审查功能的智能体，随后了解三种多智能体配置值得采用的场景。
来源: Multi-Agent Systems: When 2 Agents Beat 1 (and When They Don’t)
“多智能体”是当下的热门词汇。演示中展示五个人工智能角色在模拟房间内聊天，以抓取网站内容并撰写博客文章，这些角色的名字有研究员、作家和编辑等，还有一个终端里满是互相讨论的智能体。看起来很令人印象深刻，但这并非构建可靠软件的方式。
仅仅增加智能体并不会让系统变得更智能，反而会增加出错的可能性。以基本算术为例：一个智能体90%的时间能正确完成任务，其可靠性为0.90。如果将三个这样的智能体串联起来，概率会相乘，因此可靠性会降至约72%。延迟会翻倍，API使用成本会增加两倍，而最终输出的质量却并无提升。
值得探讨的问题是：何时会出现多个智能体共同完成工作的情况。一个有效的验证方法是将AI代码审查助手构建两次。第一个版本是传统的单智能体审查器；第二个版本则将任务分配给分析器和风险审查器，利用LangGraph中的条件边、interrupt()功能以及基于Command的恢复机制，在需要时引入人类介入。通过对比可以明白为何单智能体会忽略严重的计费错误，而双智能体组合却能发现这些问题。
为何单次提示难以完成代码审查
优质的代码审查需要两种截然不同的思维方式。
第一种是理解能力。你需要阅读代码差异，确定哪些函数发生了变化，判断哪些文件受到影响，并推断代码作者的意图。其目标是形成对新功能的心理模型。
第二种方法是对抗性测试。你需要寻找边缘情况、缺失的测试用例以及逻辑漏洞，试图破坏代码功能。
如果让一个大型语言模型在同一个提示词中同时完成这两项任务，它会显得力不从心。在同一轮处理中既进行总结又进行批判时，模型往往更倾向于保持内容的连贯性而非质疑其正确性。一旦模型确定了关于代码功能的某种解释，它就会减少对这一解释的质疑，转而关注一些较小的风格问题，比如缺少文档字符串或命名不规范等。
一个具体的例子可以阐明这一点。想象有一个针对计费系统的模拟拉取请求，提交者在发票计算器中添加了折扣功能。以下是两种设计所收到的差异对比：
--- 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
仔细观察就会发现几个严重的问题。discount_pct的值直接来自用户输入并传送到API层，随后被用于计算而未经过任何验证。用户可能会输入负数的折扣率来抬高价格，或者输入大于1的数值从而导致总金额出现负数。现有测试并未覆盖这一新路径。此外，功能开关被设置为立即向100%的用户开放。这显然是一个容易被滥用的定价流程。普通的单一代理审核员该如何处理这种情况？
设计一：单一代理审核员
首先是为该代理准备工具箱：一个用于获取差异的工具，一个用于打开文件的工具，一个用于查询符号信息的工具，以及一个用于报告现有测试的工具。
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]
接下来是状态部分。一个 TypedDict 用于确保与 LangGraph 兼容，其中存储着 PR ID、消息历史、原始差异内容以及一系列检测结果。Pydantic 模型用于描述每一项检测结果，并强制使用固定的字面值对检测结果进行分类，从而大大简化后续处理流程。
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)
接下来是各个节点。该流程呈直线结构：首先是加载上下文的获取节点，然后是执行提示词的审核节点。
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()
（在本次演示中，展示的结果是预先设定好的示例，用以模拟审核人员通常会产生的输出，因此请将此对比视为示例而非基准。）
第一代人工智能审核工具大致就是如此构建的：输入差异内容，要求找出错误。当用模拟的拉取请求进行测试时，该工具会注意到缺失的测试用例并建议添加类型提示，但却无法发现未经过验证的输入以及上线风险。
原因在于提示词的设计。当被要求先总结变更内容再列出风险时，模型会将注意力集中在解释新增了折扣功能这一事实上。等到开始列出问题时，它仍处于描述性思维状态，只关注表面语法而未尝试深入检测问题。解决办法是将内容生成任务与对抗性检测任务分开，也就是使用两个不同的模型。
交接方式：使用结构化格式而非文字描述
多智能体设计中最常见的错误就是让智能体之间用自由文本进行交流。如果智能体A撰写了三段文字并交给智能体B，B就必须解析这些文本，猜测哪些内容重要，然后在此基础上进行处理。这样一来，上下文质量就会下降。
智能体本质上是函数，而函数应当传递类型明确的对象。要想让多智能体系统具备可靠性，就必须明确界定它们之间传递数据的精确结构。
在代码审查助手中，第一个智能体是分析器，它的任务是解析拉取请求。它必须生成一个ChangeModel对象，该对象会要求它提取与更改相关的特定元数据。
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]
每个字段都为下游代理服务。已变更接口的列表能让第二个代理快速了解哪些内容是重要的。配置变更列表则使分析器能够将应用逻辑与部署逻辑分开。
assumptions数组值得特别关注。分析器需要明确说明代码作者所做出的假设，例如discount_pct是介于0和1之间的小数。这并非在批评什么，只是陈述事实。该结构化对象会成为审核人员的确切输入。
设计二：分析器与风险审核员
新的状态架构包含ChangeModel以及一个布尔类型的conflict标志，后者用于决定是否需要将问题上报给人工处理。
class TwoAgentState(TypedDict):
pr_id: str
messages: list
diff: str
change_model: ChangeModel | None
findings: list[ReviewFinding]
conflict: bool
获取节点保持不变，负责获取差异信息与测试覆盖率。接下来是分析节点，其输入提示有严格限制。模型被明确要求不要寻找错误，因此只需专注于理解内容。
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.")],
}
分析器会读取差异内容并记录实际存在的信息：新的参数、新的功能标志，以及折扣值介于0和1之间的隐含假设。
随后，该对象会被转交给风险审核员，而给他的指示则是带有故意的对抗性——要求他不要信任分析器的结果，要努力找出其中的缺陷。
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}")],
}
由于审核者在判断哪些文件发生了变化或该拉取请求的目的是什么时不会消耗令牌，这一整个流程都会导致逻辑出错。它假设折扣值应在0到1之间，检查差异后未发现任何验证代码，于是标记为高严重性的输入验证错误。
这是为数不多的需要多个角色协同工作的场景之一：一个负责构建目标，另一个则负责对其进行检查。
有人工介入的仲裁机制
审核者发现了拉取请求作者遗漏的严重问题。不过直接将这些问题发布到GitHub上存在风险。如果审核者出现了误判，将会惹恼团队。一个合理的解决方案是添加一个合并节点，该节点能够评估这些问题并决定是否需要人工进行仲裁。
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"
当 conflict 标志为真时，流程会进入升级节点。此时就需要用到条件边、interrupt() 以及基于 Command 的恢复机制。
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")
调用 interrupt() 会暂停执行流程。该图谱会序列化其当前状态，并通过配置的检查点存储起来，在本地演示中该检查点为 MemorySaver；在正式环境中，则会使用由 Postgres 支持的持久化存储机制，以确保暂停状态在服务器重启后依然存在。
随后，后端系统可以将结构化的分析结果通过 Slack 发送给资深工程师。工程师们会查看关于未经过验证的折扣的严重警告，若同意，则点击“全部接受”按钮。当 webhook 信号到达服务器后，服务器会向图谱发送一个包含继续执行指令的 Command，从而恢复执行流程。
# How you resume the graph from your backend API
ta.invoke(Command(resume={"action": "accept_all"}), config)
该图会启动，处理人类的决策，必要时调整分析结果，然后进入最终的输出节点。
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))]}
图的连接方式
这些节点被组合成一个StateGraph。由于ta_escalate会返回一个Command(goto="emit")，且LangGraph会进行动态路由，因此无需从escalate节点引出静态边。
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())
两者对比
在本次拉取请求中将两位评审者放在一起对比，差异十分明显。
那位单独工作的评审者指出了两个问题。他建议添加单元测试并提出使用类型提示。他的表现就像一个注重语法、态度谦逊的初级开发者。
那对评审者则发现了四个问题。他们将未经验证的输入视为严重的业务逻辑缺陷，指出负折扣会导致价格上升，认为立即100%推广该功能会带来计费风险，还发现了一种边缘情况：即使功能开关处于关闭状态，折扣仍会悄悄应用。
多智能体设计并不能每次都保证得到更好的评估结果。在这里它之所以有效，是因为任务分解得异常清晰。第二个智能体带来的是真实信号而非噪声，能够识别出导致系统故障的业务逻辑缺陷。其之所以能发挥作用，是因为认知负荷被分摊了——分析器负责大量数据读取并映射接口，而评估者则能收到结构清晰的ChangeModel，从而专注于进行批判性分析。
何时不应使用多个智能体
这种组合虽然强大，但人们很容易产生将每个智能体都改写为多智能体工作流的冲动，应当抵制这种想法。
额外的智能体会增加延迟。双智能体评估模式所需时间大约是单智能体模式的两倍，因为其大语言模型调用是顺序进行的，Token成本也高出约两倍，同时还涉及更复杂的状态架构和数据传递逻辑。只有当任务符合以下三种情况之一时，才应使用这种模式。
对抗式设计。如果一个系统需要同时生成内容与对内容进行评估，应将这两项功能分开。生成器与判别器是最经验证的多智能体设计方案：一个智能体负责编写代码，另一个则编写测试用例来破坏该代码；一个智能体负责起草邮件，另一个则检查其中是否存在合规问题。
严格的工具隔离。负责搜索公共网络的智能体与负责执行数据库查询的智能体不应共享同一套处理逻辑。让研究者收集公共数据，并将经过处理的摘要传递给执行者，这样就能形成坚实的安全边界。如果某个网页存在提示注入攻击，只有研究者会受到影响，而执行者仍能保持隔离状态。
非对称上下文扩展。有时大量数据仅用于支撑一个微小的决策。一群廉价且快速的工作者可以同时读取100个日志文件，各自提取相关的错误行，然后将简短的摘要传递给一个昂贵的主管，由它来做出最终决策。
如果某个任务不符合上述任何情况，那就使用单个智能体。一个经过精心设计的提示、严格的工具限制以及清晰的状态结构，总能比混乱的多智能体架构表现得更好。
相关阅读
- Claude Code Agent Teams：何时以及如何采用多智能体架构 — 介绍何时一个智能体就足够、增加智能体的四个触发条件、编排模式、通信方式、八种故障模式，以及Claude Code团队在原生方案与自行搭建方案之间的选择。