Marco Liu代码评审这件事，大部分团队都卡在同一个矛盾里：评审越认真，人越累；评审越随意，线上事故越多。当一个仓库一天合并几十个 PR、评审人同时挂着三四个待审列表时，"认真看"几乎成了一种奢侈。Uber 工程团队公开分享过他们在这方面的尝试——uReview，一个把代码评审拆成多个专职智能体、再由编排层汇总意见的多智能体（Multi-Agent）评审引擎。本文不讨论某一家公司的内部实现细节，而是围绕"如何自己搭一套多智能体代码评审流水线"这个目标，把架构拆解、环境准备、核心代码、端到端跑通、常见报错和工程化建议完整讲一遍。适合已经会写 Python、了解 Git diff、并且想把大模型真正接进研发流程的后端和平台工程同学；零基础读者也可以跟着代码一步步跑起来，只是阅读提示词工程那一节时建议多回看几遍。
1. 背景与核心概念
1.1 传统 Code Review 为什么容易失效
传统的代码评审依赖人，而人有三个绕不过去的限制。第一是注意力预算有限，一个评审人一次只能保持二三十分钟的高质量专注，超过这个时长，眼睛扫过的代码和真正读进去的代码完全是两回事。第二是知识覆盖面有限，一个后端工程师可能对并发和事务很敏感，但对前端模板里的 XSS 风险未必能第一时间反应过来；反过来也一样。第三是标准不统一，同一个仓库里，A 评审人认为"这里必须加注释"，B 评审人认为"注释是噪音"，新人在两种声音之间无所适从。
这三个限制叠加起来，就出现了很典型的现象：低价值问题（命名、格式、缩进）占据大量评审时间，高价值问题（边界条件、空指针、事务范围、注入风险）反而漏掉。而一旦线上出事，回头翻 PR 记录，往往会发现"当时有人看出来了，但没提"或者"当时没人往那个方向想"。
1.2 uReview 是什么，它解决什么问题
uReview 的定位是"评审引擎"，不是"评审机器人"。这两者的区别在于：机器人是"在 PR 里自动留一条评论"，引擎是"有一整套采集、分发、推理、汇总、反馈的流水线，最终产出一份结构化的评审意见"。
从公开分享的思路来看，它的核心主张是把一次代码评审拆成若干互不干扰的子任务，每个子任务交给一个专职智能体（Agent）去完成。比如一个智能体只负责安全视角，一个只负责正确性视角，一个只负责性能视角。每个智能体拿到的提示词、上下文范围、输出格式都是专门定制的，互不干扰。最后由一个聚合层把多份意见合并、去重、排序，形成一份人能直接读的评审报告。
这样设计的直接收益是：单个智能体的职责边界清晰，提示词可以写得很具体，不容易"什么都想说但什么都说不深"；同时不同视角可以并行执行，延迟不会线性叠加。间接收益是工程上可维护——你想加一个"数据库索引缺失检测"的视角，只需要新增一个智能体，不用动其他任何代码。
1.3 多智能体与单体 Prompt 的差别
很多同学第一反应是："为什么不直接写一个很长的 Prompt，让模型一次性把安全、性能、正确性都看一遍？" 这在只有几十行 diff 的时候确实能用，但一旦代码量上来，问题立刻暴露。
单体 Prompt 的典型失败模式有三种。第一种是"注意力稀释"，模型面对一大段指令加一大段代码，最终往往只抓住了最显眼的一两个问题，其余视角形同虚设。第二种是"视角污染"，安全性要求"宁可误报不可漏报"，可维护性要求"不要吹毛求疵"，这两个倾向放在同一个 Prompt 里会互相打架，最后输出风格飘忽。第三种是"无法并行、无法降级"，一次调用失败整次评审失败，也没法只重跑某一个视角。
多智能体的做法是把这些冲突显式化。每个智能体有独立的系统提示词、独立的严重级别门槛、独立的重试策略，甚至可以用不同的模型（安全视角用能力更强的模型，格式视角用便宜的小模型）。代价是工程复杂度上升，需要考虑并发、去重、成本控制，这也正是本文后面几节要解决的问题。
1.4 几个容易混淆的概念
在动手之前，先厘清几个词，后面会反复用到。
智能体（Agent）：在这里不是一个能自主规划、自我反思的复杂实体，而是一个"封装了提示词 + 上下文 + 输出解析的评审单元"。它无状态，输入一个代码块，输出一组问题。把它理解成一个"带模型调用的函数"更准确。
编排器（Orchestrator）：负责决定哪个智能体看哪段代码、并发跑多少个、失败了怎么重试。
聚合器（Aggregator）：把多个智能体对同一位置提出的相似问题合并成一条，避免同一行被三个智能体各骂一遍。
上下文打包（Context Packing）：把 Git diff 转换成带行号的、有长度上限的文本块。这一步做得好不好，直接决定模型能不能给出准确的行号。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 架构拆解：评审引擎的五个部件
2.1 整体分层
一套能上生产的评审引擎，基本可以分成五层，从上到下依次是触发层、采集层、编排层、推理层、汇总层。用 ASCII 图表示大致是这样：
text复制┌──────────────────────────────────────────────┐
│ 触发层 CI Job / PR 事件 / 本地命令行 │
└───────────────────┬──────────────────────────┘
│
┌───────────────────▼──────────────────────────┐
│ 采集层 git diff → 文件过滤 → 分块 → 行号渲染 │
└───────────────────┬──────────────────────────┘
│
┌───────────────────▼──────────────────────────┐
│ 编排层 任务矩阵（智能体 × 代码块）→ 并发执行 │
└───────────────────┬──────────────────────────┘
│
┌───────────────────▼──────────────────────────┐
│ 推理层 安全 / 正确性 / 性能 / 可维护性 智能体 │
└───────────────────┬──────────────────────────┘
│
┌───────────────────▼──────────────────────────┐
│ 汇总层 去重 → 严重级别排序 → Markdown 报告 │
└──────────────────────────────────────────────┘
这个分层最重要的价值是"每层可以单独替换"。比如你从本地命令行切到 CI，只动触发层；你从一家模型服务切到另一家，只动推理层的适配代码；你想改报告格式，只动汇总层。很多团队做这类工具做到后期维护不下去，根本原因就是把所有逻辑糊在一个几百行的脚本里。
2.2 上下文采集层要解决什么
采集层看着简单——不就是 git diff 吗？实际上下面的坑最多。
第一个坑是范围选择。用 git diff HEAD 拿到的是工作区未提交改动，用 git diff main...feature 拿到的是分叉点之后的改动，两者语义完全不同。CI 里跑评审，通常应该用三点语法 base...head，也就是"从共同祖先到当前提交"的差异，这样不会把主干上别人的提交也算进来。
第二个坑是噪音文件。锁文件、构建产物、自动生成的代码、国际化资源文件，这些内容既长又没有评审价值，还会吃掉大量 token 预算。必须有路径过滤规则。
第三个坑是行号锚定。Git diff 里的行号是相对 hunk 的偏移量，不是文件真实行号。如果不做转换，模型返回的行号会全部错位，评审意见贴到 PR 上就是"评论在错误的位置"，用户第一次看到就会失去信任。
第四个坑是超长文件。一个自动生成的 Protobuf 文件可能几千行，直接塞进去必然超上下文。需要在采集层就按 hunk 或按行数切块，而不是把一个文件当成一个单位。
2.3 智能体编排层的心智模型
编排层的本质是计算一个笛卡尔积：智能体列表 × 代码块列表，每个格子是一次独立的模型调用。这个视角很有用，因为它把"评审"变成了一个可度量的批处理任务——你能算出总共多少次调用、预计花多少钱、并发度设多少能在超时之前跑完。
同时它带来两个必须处理的问题。一是并发与限流，模型服务通常有速率限制，并发开太大就全是 429。二是失败隔离，某一个代码块在某个智能体上失败了，不应该让整次评审失败，而应该记录降级信息继续往下走。这两点在 4.6 节的代码里会具体体现。
2.4 结果聚合层为什么要单独存在
如果只用一个智能体，聚合层确实可有可无。但只要有三个以上视角，"同一行被重复评论"就会变成主要噪音来源。举个常见例子：一段拼接 SQL 的代码，安全智能体会说"存在注入风险"，正确性智能体会说"参数未校验可能导致异常"，性能智能体会说"这条 SQL 没有走索引"。三条意见指向同一行，但其实是同一个根因的三个侧面。
聚合层的任务是：按文件 + 行号临近 + 文本相似度做合并，保留最高严重级别，把来源智能体记录下来，最终一条意见里能体现"这条问题被安全、正确性两个视角同时发现"。这种"多视角共同确认"本身就是很强的信号，可以直接用来给意见加分。
2.5 反馈闭环是长期可用的关键
绝大多数自建评审工具死在第三个月：误报太多，大家开始无视评论，最后关掉。要避免这个结局，必须从第一天就设计反馈机制。
最轻量的做法是：在每条评论后面附一个可点击的"无效"标记，把点击记录落到一张表里，定期统计各智能体的有效率。更进一步的做法是把"被标记无效的意见"整理成反例，定期回灌到该智能体的提示词里，形成"少样本纠偏"。这个过程不需要多复杂的技术，一张表加一次人工复盘就够了，但它决定了这个工具能不能活过半年。
3. 环境准备与项目结构
3.1 运行环境与依赖
本文示例使用 Python 3.10 及以上版本，原因很简单：代码里用到了 str | None 这种联合类型标注语法，3.9 及以下会直接语法报错。如果你所在的环境必须是 3.8，把所有 X | None 改成 Optional[X] 即可，逻辑完全一致。
模型服务方面，本文统一使用 OpenAI 兼容的 /chat/completions 接口协议。这是目前兼容性最广的一种约定，主流自托管推理框架和大部分云服务都支持。具体用哪家、哪个模型，需要根据你的预算和评测结果决定，本文不会写死任何模型名称，示例里全部通过环境变量注入。版本方面，requests 库建议使用 2.x 的稳定版本，PyYAML 建议 6.x，实际以你项目的依赖锁定文件为准。
依赖清单如下：
text复制# requirements.txt
requests>=2.28
PyYAML>=6.0
安装命令：
bash复制python -m venv .venv
source .venv/bin/activate # Windows 用 .venv\Scripts\activate
pip install -r requirements.txt
3.2 项目目录结构
建议从一开始就按职责拆分目录，不要把所有逻辑写进一个 main.py。下面这套结构在后续扩展智能体和切换模型时基本不需要重构：
text复制ureview/
├── config/
│ ├── agents.yaml # 智能体开关、严重级别门槛
│ └── rules.yaml # 全局评审规则、忽略路径
├── src/
│ └── ureview/
│ ├── __init__.py
│ ├── models.py # Issue / Severity / Category
│ ├── llm.py # 模型调用适配层
│ ├── diff_parser.py # git diff 解析
│ ├── context_packer.py# 上下文渲染与分块
│ ├── agents/
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── security.py
│ │ ├── correctness.py
│ │ ├── performance.py
│ │ └── maintainability.py
│ ├── orchestrator.py # 任务矩阵与并发
│ ├── aggregator.py # 去重与排序
│ └── report.py # Markdown / JSON 输出
├── main.py
└── requirements.txt
config 和 src 分开的好处是，规则调整不需要改代码，运营同学也能维护；而 agents 单独成目录，新增一个视角只需要加一个文件并在配置里打开开关。
3.3 配置文件设计
先看智能体配置。这里的关键设计是"每个智能体有自己的严重级别门槛"——安全视角只报 major 以上，可维护性视角可以报到 minor，这样能天然控制不同视角的噪音量：
yaml复制# config/agents.yaml
llm:
timeout: 90
max_retries: 3
temperature: 0.0
budget:
max_chars_per_chunk: 8000
max_hunks_per_chunk: 6
concurrency: 4
agents:
- name: security
enabled: true
min_severity: major
- name: correctness
enabled: true
min_severity: minor
- name: performance
enabled: true
min_severity: minor
- name: maintainability
enabled: true
min_severity: nit
再看规则配置。global_rules 会被拼进每个智能体的系统提示词，是跨视角的通用约束；ignore_paths 在采集层生效，避免无用文件浪费预算；severity_floor 用来对测试文件降级，因为测试代码里的硬编码和魔法数字本身是可接受的：
yaml复制# config/rules.yaml
global_rules:
- "只评论本次 diff 中新增或修改的行，不要评论未改动代码"
- "不要评论纯格式问题（空格、缩进、引号风格），交给格式化工具有处理"
- "每条意见必须能对应到具体行号，无法定位的不要输出"
- "如果代码没有明显问题，返回空数组，不要为了凑数而编造问题"
- "建议必须具体，说明改成什么，而不是只说'建议优化'"
ignore_paths:
- "**/*.lock"
- "**/dist/**"
- "**/build/**"
- "**/*.min.js"
- "**/generated/**"
severity_floor:
"**/test_*.py": nit
"**/*_test.py": nit
3.4 密钥与权限边界
模型调用密钥必须通过环境变量注入，不允许写进配置文件提交到仓库。这一点看起来是常识，但每年仍然能看到大量泄露案例。推荐做法是：本地开发用 .env 并把它加进 .gitignore，CI 环境用平台的 Secret 管理功能注入。
涉及的三个环境变量如下：
bash复制export UREVIEW_LLM_BASE_URL="https://your-llm-endpoint/v1"
export UREVIEW_LLM_API_KEY="sk-xxxxxxxxxxxxxxxx"
export UREVIEW_LLM_MODEL="your-model-name"
另外要强调权限边界。这个工具需要读取代码仓库，如果需要回写 PR 评论，还需要写权限。生产环境务必遵守最小权限原则：评审用的 Token 只给它需要访问的那几个仓库，不要复用管理员账号；如果只是先在 CI 里输出报告不写评论，那就干脆不要申请写权限。任何涉及往主干仓库写入的操作，都应该先在测试仓库验证，并做好配置备份。
4. 核心实现：从零写一个多智能体评审流水线
4.1 数据结构定义
先定义贯穿全流程的数据结构。把严重级别和问题分类做成枚举，好处是排序和过滤时不会出现字符串拼写错误，输出报告时也容易做图标映射：
python复制# 文件路径：src/ureview/models.py
from dataclasses import dataclass, field
from enum import Enum
class Severity(str, Enum):
BLOCKER = "blocker"
MAJOR = "major"
MINOR = "minor"
NIT = "nit"
class Category(str, Enum):
CORRECTNESS = "correctness"
SECURITY = "security"
PERFORMANCE = "performance"
MAINTAINABILITY = "maintainability"
SEVERITY_ORDER = {
Severity.NIT: 0,
Severity.MINOR: 1,
Severity.MAJOR: 2,
Severity.BLOCKER: 3,
}
@dataclass
class Issue:
agent: str
file: str
line: int
severity: Severity
category: Category
message: str
suggestion: str = ""
evidence: str = ""
confidence: float = 0.8
agents: list[str] = field(default_factory=list)
def __post_init__(self) -> None:
if not self.agents:
self.agents = [self.agent]
def to_dict(self) -> dict:
return {
"agent": self.agent,
"agents": self.agents,
"file": self.file,
"line": self.line,
"severity": self.severity.value,
"category": self.category.value,
"message": self.message,
"suggestion": self.suggestion,
"evidence": self.evidence,
"confidence": self.confidence,
}
agents 字段和 agent 字段并存是有意为之：agent 记录首次发现者，agents 记录所有发现者，聚合阶段会往里追加。confidence 让模型可以对自己不确定的意见表态，后续可以做阈值过滤——实测下来，让模型显式输出置信度能显著减少"为了凑数而报"的情况。
4.2 模型调用适配层
适配层要做三件事：统一接口协议、处理重试与退避、把错误包装成可识别的异常。这里用标准库 requests 直接发请求，不做多余封装，方便你替换成任何兼容服务：
python复制# 文件路径：src/ureview/llm.py
import os
import time
import requests
class LLMError(RuntimeError):
"""模型调用失败时抛出，便于上层做降级处理。"""
class LLMClient:
def __init__(
self,
base_url: str | None = None,
api_key: str | None = None,
model: str | None = None,
timeout: int = 90,
max_retries: int = 3,
temperature: float = 0.0,
) -> None:
self.base_url = (base_url or os.environ["UREVIEW_LLM_BASE_URL"]).rstrip("/")
self.api_key = api_key or os.environ["UREVIEW_LLM_API_KEY"]
self.model = model or os.environ.get("UREVIEW_LLM_MODEL", "")
self.timeout = timeout
self.max_retries = max_retries
self.temperature = temperature
def chat(self, system: str, user: str, temperature: float | None = None) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"temperature": self.temperature if temperature is None else temperature,
}
last_error: Exception | None = None
for attempt in range(1, self.max_retries + 1):
try:
resp = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
json=payload,
timeout=self.timeout,
)
if resp.status_code == 429 or resp.status_code >= 500:
raise LLMError(f"HTTP {resp.status_code}: {resp.text[:200]}")
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]
except Exception as exc: # noqa: BLE001
last_error = exc
if attempt < self.max_retries:
time.sleep(min(2 ** attempt, 10))
raise LLMError(f"重试 {self.max_retries} 次后仍失败: {last_error}")
这里的退避策略是 2^attempt 秒并封顶 10 秒，实践中足够应对短时限流。需要注意的是 temperature 默认设为 0.0，代码评审是确定性任务，不需要创造性；如果发现同一段代码两次评审结果差异很大，先把温度降下来再看。
4.3 Diff 解析与上下文打包
这一步是准确性的地基。先写解析器，把原始 diff 拆成文件 → hunk → 行的三级结构：
python复制# 文件路径：src/ureview/diff_parser.py
import re
from dataclasses import dataclass, field
HUNK_RE = re.compile(r"^@@ -(\d+)(?:,\d+)? \+(\d+)(?:,\d+)? @@")
LANGUAGE_MAP = {
".py": "python",
".java": "java",
".go": "go",
".ts": "typescript",
".js": "javascript",
".sql": "sql",
".xml": "xml",
".yaml": "yaml",
".yml": "yaml",
}
@dataclass
class Hunk:
old_start: int
new_start: int
lines: list[str] = field(default_factory=list)
@dataclass
class FileDiff:
path: str
hunks: list[Hunk] = field(default_factory=list)
language: str = "text"
is_binary: bool = False
def _extract_path(line: str) -> str:
if " b/" in line:
return line.split(" b/", 1)[1].strip()
return line.rsplit(" ", 1)[-1].strip()
def _guess_language(path: str) -> str:
for suffix, lang in LANGUAGE_MAP.items():
if path.endswith(suffix):
return lang
return "text"
def parse_diff(raw: str) -> list[FileDiff]:
files: list[FileDiff] = []
current: FileDiff | None = None
hunk: Hunk | None = None
for line in raw.splitlines():
if line.startswith("diff --git "):
current = FileDiff(path=_extract_path(line))
files.append(current)
hunk = None
continue
if current is None:
continue
if line.startswith("Binary files "):
current.is_binary = True
continue
if line.startswith("--- ") or line.startswith("+++ "):
continue
matched = HUNK_RE.match(line)
if matched:
hunk = Hunk(old_start=int(matched.group(1)), new_start=int(matched.group(2)))
current.hunks.append(hunk)
continue
if hunk is not None:
hunk.lines.append(line)
for item in files:
item.language = _guess_language(item.path)
return files
解析完成后需要把 hunk 渲染成"带真实行号"的文本。这是让模型输出准确行号的关键——只要你在提示词里明确"每行前面的数字是文件真实行号"，模型绝大多数时候能照着引用：
python复制# 文件路径：src/ureview/context_packer.py
from dataclasses import dataclass
from .diff_parser import FileDiff, Hunk
@dataclass
class Chunk:
path: str
language: str
start_line: int
end_line: int
text: str
def render_hunk(hunk: Hunk) -> tuple[str, int, int]:
rendered: list[str] = []
new_lineno = hunk.new_start
start = hunk.new_start
for line in hunk.lines:
if line.startswith("\\"):
continue
if line.startswith("+"):
rendered.append(f"{new_lineno:>5} + {line[1:]}")
new_lineno += 1
elif line.startswith("-"):
rendered.append(f"{'':>5} - {line[1:]}")
else:
body = line[1:] if line.startswith(" ") else line
rendered.append(f"{new_lineno:>5} {body}")
new_lineno += 1
return "\n".join(rendered), start, max(new_lineno - 1, start)
def pack(file_diffs: list[FileDiff], max_chars: int, max_hunks: int) -> list[Chunk]:
chunks: list[Chunk] = []
for file_diff in file_diffs:
if file_diff.is_binary:
continue
bucket: list[str] = []
start_line = 0
end_line = 0
size = 0
count = 0
def flush() -> None:
nonlocal bucket, size, count
if bucket:
chunks.append(
Chunk(
path=file_diff.path,
language=file_diff.language,
start_line=start_line,
end_line=end_line,
text="\n".join(bucket),
)
)
bucket = []
size = 0
count = 0
for hunk in file_diff.hunks:
text, h_start, h_end = render_hunk(hunk)
if len(text) > max_chars:
flush()
chunks.append(
Chunk(
path=file_diff.path,
language=file_diff.language,
start_line=h_start,
end_line=h_end,
text=text[:max_chars],
)
)
continue
if bucket and (size + len(text) > max_chars or count >= max_hunks):
flush()
if not bucket:
start_line = h_start
end_line = h_end
bucket.append(text)
size += len(text)
count += 1
flush()
return chunks
打包策略里有几个细节值得说明。max_chars_per_chunk 设成 8000 字符是个经验值，大约对应两千多 token 的输入，既能装下足够上下文，又不至于让模型开始"看花眼"。超长 hunk 会被截断，这会损失一部分准确性，所以实践中更好的做法是按行切分而不是直接截断——如果你们仓库经常出现超长 hunk，建议把这里改造成按行滑窗。
4.4 专职智能体基类
基类把"调用模型 → 解析 JSON → 校正行号 → 过滤严重级别"这条固定流程封装起来，子类只需要关心系统提示词和分类：
python复制# 文件路径：src/ureview/agents/base.py
import json
import re
from ..context_packer import Chunk
from ..llm import LLMClient, LLMError
from ..models import SEVERITY_ORDER, Category, Issue, Severity
def extract_json(raw: str) -> dict:
text = raw.strip()
if text.startswith("```"):
text = re.sub(r"^```[a-zA-Z]*\n", "", text)
text = re.sub(r"\n```$", "", text.strip())
try:
return json.loads(text)
except json.JSONDecodeError:
pass
start, end = text.find("{"), text.rfind("}")
if start != -1 and end > start:
try:
return json.loads(text[start:end + 1])
except json.JSONDecodeError:
return {}
return {}
class BaseAgent:
name = "base"
category = Category.MAINTAINABILITY
min_severity = Severity.MINOR
def __init__(self, llm: LLMClient, global_rules: list[str] | None = None) -> None:
self.llm = llm
self.global_rules = global_rules or []
def system_prompt(self) -> str:
raise NotImplementedError
def build_system(self) -> str:
rules = "\n".join(f"- {item}" for item in self.global_rules)
return f"{self.system_prompt()}\n\n【全局约束】\n{rules}"
def review(self, chunk: Chunk) -> list[Issue]:
user_prompt = (
f"文件：{chunk.path}\n"
f"语言：{chunk.language}\n"
f"行号范围：{chunk.start_line}-{chunk.end_line}\n\n"
f"代码：\n```{chunk.language}\n{chunk.text}\n```\n\n"
"请按约定的 JSON 格式输出问题列表。"
)
try:
raw = self.llm.chat(self.build_system(), user_prompt)
except LLMError as exc:
print(f"[warn] agent={self.name} file={chunk.path} 调用失败: {exc}")
return []
return self.parse(raw, chunk)
def parse(self, raw: str, chunk: Chunk) -> list[Issue]:
data = extract_json(raw)
items = data.get("issues", []) if isinstance(data, dict) else []
issues: list[Issue] = []
for item in items:
if not isinstance(item, dict):
continue
try:
line = int(item.get("line") or chunk.start_line)
except (TypeError, ValueError):
line = chunk.start_line
if not (chunk.start_line <= line <= chunk.end_line):
line = chunk.start_line
try:
severity = Severity(str(item.get("severity", "minor")).lower())
except ValueError:
severity = Severity.MINOR
if SEVERITY_ORDER[severity] < SEVERITY_ORDER[self.min_severity]:
continue
message = str(item.get("message", "")).strip()
if not message:
continue
issues.append(
Issue(
agent=self.name,
file=chunk.path,
line=line,
severity=severity,
category=self.category,
message=message,
suggestion=str(item.get("suggestion", "")).strip(),
evidence=str(item.get("evidence", "")).strip(),
confidence=float(item.get("confidence", 0.8) or 0.8),
)
)
return issues
parse 里做了三重防御：行号越界就回退到块起始行、严重级别非法就降级为 minor、消息为空就丢弃。这些看起来啰嗦，但真实模型输出的脏数据比想象中多，没有这层兜底，后面聚合阶段一定会因为 KeyError 或类型错误崩掉。
4.5 四个智能体的实现
安全性智能体只盯一类问题：注入、越权、敏感信息泄露、不安全的反序列化、缺失的输入校验。提示词里要明确"不要评论风格"，否则它会忍不住说变量名起得不好：
python复制# 文件路径：src/ureview/agents/security.py
from ..models import Category, Severity
from .base import BaseAgent
class SecurityAgent(BaseAgent):
name = "security"
category = Category.SECURITY
min_severity = Severity.MAJOR
def system_prompt(self) -> str:
return (
"你是一名专注应用安全的代码评审专家。你只从安全视角审查代码，"
"关注点包括：SQL 拼接与注入、命令注入、路径穿越、越权访问、"
"敏感信息硬编码、不安全的反序列化、缺失的鉴权与输入校验、"
"日志中打印敏感字段。\n"
"输出要求：严格返回 JSON，格式为 "
'{"issues": [{"line": 12, "severity": "major", "message": "...", '
'"suggestion": "...", "evidence": "...", "confidence": 0.9}]}。\n'
"severity 取值只能是 blocker / major / minor / nit。"
"无法定位到具体行的问题不要输出。没有问题时返回 {\"issues\": []}。"
)
正确性智能体关注逻辑闭环：空值、边界条件、异常吞掉、并发共享状态、资源未释放、事务边界。这一类问题的特点是"业务语义强"，所以提示词里要鼓励模型结合上下文推断，而不是机械匹配：
python复制# 文件路径：src/ureview/agents/correctness.py
from ..models import Category, Severity
from .base import BaseAgent
class CorrectnessAgent(BaseAgent):
name = "correctness"
category = Category.CORRECTNESS
min_severity = Severity.MINOR
def system_prompt(self) -> str:
return (
"你是一名严谨的后端代码评审专家，只关注逻辑正确性。"
"重点检查：空指针与空集合、数组与字符串边界、除零、"
"循环中的提前返回、异常被静默吞掉、资源未关闭、"
"并发场景下的共享可变状态、事务范围是否正确、"
"重试逻辑是否幂等。\n"
"对每一条问题，必须在 evidence 字段引用触发问题的具体代码片段，"
"并在 suggestion 中给出可直接采用的修改方向。\n"
"严格返回 JSON：{\"issues\": [{\"line\": 1, \"severity\": \"major\", "
"\"message\": \"...\", \"suggestion\": \"...\", \"evidence\": \"...\", "
"\"confidence\": 0.85}]}。没有问题时返回 {\"issues\": []}。"
)
性能智能体关注可量化的开销：循环内 IO、N+1 查询、缺索引的全表扫描、不必要的深拷贝、正则重复编译、大对象序列化：
python复制# 文件路径：src/ureview/agents/performance.py
from ..models import Category, Severity
from .base import BaseAgent
class PerformanceAgent(BaseAgent):
name = "performance"
category = Category.PERFORMANCE
min_severity = Severity.MINOR
def system_prompt(self) -> str:
return (
"你是一名性能优化方向的代码评审专家。只关注可量化的性能问题："
"循环内的数据库查询或远程调用、疑似 N+1 查询、"
"缺少分页的大结果集读取、循环里的字符串拼接、"
"重复编译正则、不必要的大对象拷贝、"
"在事务中执行耗时操作。\n"
"不要提出没有依据的优化建议，例如在没有执行计划的情况下断言索引缺失，"
"这类情况请在 message 中说明前提条件。\n"
"严格返回 JSON：{\"issues\": [{\"line\": 1, \"severity\": \"minor\", "
"\"message\": \"...\", \"suggestion\": \"...\", \"confidence\": 0.7}]}。"
"没有问题时返回 {\"issues\": []}。"
)
可维护性智能体最容易变成噪音源，所以它的提示词要主动收敛：明确列出"不要报"的类别，并要求只报会造成实际理解成本的问题：
python复制# 文件路径：src/ureview/agents/maintainability.py
from ..models import Category, Severity
from .base import BaseAgent
class MaintainabilityAgent(BaseAgent):
name = "maintainability"
category = Category.MAINTAINABILITY
min_severity = Severity.NIT
def system_prompt(self) -> str:
return (
"你是一名关注长期可维护性的代码评审专家。"
"只报告会造成实际理解成本或后续修改风险的问题："
"职责过多的函数、超过三层的嵌套、被复制粘贴的重复逻辑、"
"含义不明的魔法数字、容易误导的变量命名、"
"缺失的关键注释（仅限复杂算法或非显然的业务规则）。\n"
"明确不要报告：格式与缩进、引号风格、行长度、"
"个人命名偏好、缺少文档字符串。\n"
"严格返回 JSON：{\"issues\": [{\"line\": 1, \"severity\": \"nit\", "
"\"message\": \"...\", \"suggestion\": \"...\"}]}。"
"没有问题时返回 {\"issues\": []}。"
)
四个智能体写完之后，如果想让它们各自使用不同模型（比如安全视角用更强的模型），只需在 agents.yaml 里给每个智能体加一个 model 字段，然后在工厂函数里按名字创建不同的 LLMClient。这类扩展点要在设计之初就留好，否则后面改动会牵连一大片代码。
4.6 编排器与并发控制
编排器负责组装任务矩阵并并发执行。这里用 ThreadPoolExecutor 就够了，因为瓶颈在等待网络 IO，不需要引入异步框架增加复杂度：
python复制# 文件路径：src/ureview/orchestrator.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from .agents.base import BaseAgent
from .context_packer import Chunk
from .models import Issue
class Orchestrator:
def __init__(self, agents: list[BaseAgent], concurrency: int = 4) -> None:
self.agents = agents
self.concurrency = max(1, concurrency)
def run(self, chunks: list[Chunk]) -> tuple[list[Issue], list[str]]:
issues: list[Issue] = []
failures: list[str] = []
tasks = [(agent, chunk) for agent in self.agents for chunk in chunks]
if not tasks:
return issues, failures
with ThreadPoolExecutor(max_workers=self.concurrency) as pool:
future_map = {
pool.submit(agent.review, chunk): (agent.name, chunk.path)
for agent, chunk in tasks
}
for future in as_completed(future_map):
agent_name, path = future_map[future]
try:
issues.extend(future.result())
except Exception as exc: # noqa: BLE001
failures.append(f"{agent_name}@{path}: {exc}")
return issues, failures
failures 单独返回而不是抛异常，是一个重要的工程决策：一次评审里某个代码块失败，你应该照常输出其余结果，同时在报告末尾标注"本次评审有 N 个片段未完成"。这比整次失败要友好得多，也让问题更容易定位。并发度默认 4，如果你的模型服务限流较严，把它降到 2；如果用了多副本部署且额度充足，可以提到 8 到 16，但要配合压测观察 429 的比例。
4.7 结果聚合与去重
聚合器做三件事：按位置聚类、按文本相似度判重、保留最高严重级别：
python复制# 文件路径：src/ureview/aggregator.py
import difflib
from .models import SEVERITY_ORDER, Issue
def _similar(a: str, b: str) -> float:
return difflib.SequenceMatcher(None, a, b).ratio()
def aggregate(issues: list[Issue], threshold: float = 0.75) -> list[Issue]:
ordered = sorted(
issues,
key=lambda item: (item.file, item.line, -SEVERITY_ORDER[item.severity]),
)
merged: list[Issue] = []
for issue in ordered:
duplicated = False
for exist in merged:
if exist.file != issue.file:
continue
if abs(exist.line - issue.line) > 3:
continue
if _similar(exist.message, issue.message) < threshold:
continue
if issue.agent not in exist.agents:
exist.agents.append(issue.agent)
if SEVERITY_ORDER[issue.severity] > SEVERITY_ORDER[exist.severity]:
exist.severity = issue.severity
if not exist.suggestion and issue.suggestion:
exist.suggestion = issue.suggestion
exist.confidence = max(exist.confidence, issue.confidence)
duplicated = True
break
if not duplicated:
merged.append(issue)
return sorted(merged, key=lambda item: (item.file, item.line))
阈值 0.75 是经过权衡的：太低会把"同一行的两个不同问题"误合并，太高则放过了"表述不同但根因相同"的重复。实际使用中建议把阈值做成配置项，并根据你们团队的反馈数据调整。另外注意 abs(exist.line - issue.line) > 3 这个窗口——同一段逻辑的问题常常分布在相邻几行，允许三行浮动能明显提升去重率。
4.8 报告渲染
报告要同时输出 Markdown 和 JSON。Markdown 给人看，JSON 给流水线后续消费，比如做统计或者推到其他系统：
python复制# 文件路径：src/ureview/report.py
import json
from .models import Issue
SEVERITY_LABEL = {
"blocker": "阻断",
"major": "严重",
"minor": "一般",
"nit": "建议",
}
def render_markdown(issues: list[Issue], failures: list[str], max_issues: int = 50) -> str:
lines = ["## 代码评审结果", ""]
if not issues:
lines.append("本次评审未发现需要处理的问题。")
else:
lines.append(f"共发现 **{len(issues)}** 条意见，按文件和行号排序如下：")
lines.append("")
for issue in issues[:max_issues]:
label = SEVERITY_LABEL.get(issue.severity.value, issue.severity.value)
sources = " / ".join(issue.agents)
lines.append(f"### `{issue.file}:{issue.line}` [{label}] {issue.category.value}")
lines.append("")
lines.append(f"- 来源视角：{sources}")
lines.append(f"- 问题描述：{issue.message}")
if issue.suggestion:
lines.append(f"- 修改建议：{issue.suggestion}")
if issue.evidence:
lines.append(f"- 相关代码：`{issue.evidence[:120]}`")
lines.append("")
if failures:
lines.append("### 未完成的片段")
lines.append("")
for item in failures:
lines.append(f"- {item}")
lines.append("")
return "\n".join(lines)
def render_json(issues: list[Issue], failures: list[str]) -> str:
return json.dumps(
{
"total": len(issues),
"issues": [item.to_dict() for item in issues],
"failures": failures,
},
ensure_ascii=False,
indent=2,
)
5. 端到端实战：跑通一次完整评审
5.1 准备示例仓库
先造一个小仓库，里面放一个带有典型问题的文件，用来验证整条链路：
bash复制mkdir -p /tmp/ureview-demo && cd /tmp/ureview-demo
git init -q
cat > order_service.py <<'PY'
import sqlite3
def query_orders(conn, user_id, keyword):
sql = "SELECT * FROM orders WHERE user_id = %s AND title LIKE '%" + keyword + "%'"
cursor = conn.cursor()
cursor.execute(sql % user_id)
rows = cursor.fetchall()
result = []
for row in rows:
detail = conn.execute("SELECT * FROM order_items WHERE order_id = ?", (row[0],)).fetchall()
result.append({"order": row, "items": detail})
return result
PY
git add . && git commit -qm "init: add order service"
再提交一次修改，制造可供评审的 diff：
bash复制cat >> order_service.py <<'PY'
def export_orders(conn, user_id, path):
rows = query_orders(conn, user_id, "")
with open(path, "w") as f:
for row in rows:
f.write(str(row) + "\n")
return len(rows)
PY
git add . && git commit -qm "feat: add export"
5.2 编写主入口
主入口把前面所有部件串起来，同时负责读取配置、过滤忽略路径：
python复制# 文件路径：main.py
import argparse
import fnmatch
import json
import os
import subprocess
import sys
import yaml
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src"))
from ureview.aggregator import aggregate
from ureview.agents.correctness import CorrectnessAgent
from ureview.agents.maintainability import MaintainabilityAgent
from ureview.agents.performance import PerformanceAgent
from ureview.agents.security import SecurityAgent
from ureview.context_packer import pack
from ureview.diff_parser import parse_diff
from ureview.llm import LLMClient
from ureview.orchestrator import Orchestrator
from ureview.report import render_json, render_markdown
AGENT_REGISTRY = {
"security": SecurityAgent,
"correctness": CorrectnessAgent,
"performance": PerformanceAgent,
"maintainability": MaintainabilityAgent,
}
def collect_diff(base: str, head: str) -> str:
result = subprocess.run(
["git", "diff", "--unified=3", f"{base}...{head}"],
capture_output=True,
text=True,
check=True,
)
return result.stdout
def should_ignore(path: str, patterns: list[str]) -> bool:
return any(fnmatch.fnmatch(path, pattern) for pattern in patterns)
def severity_floor_for(path: str, mapping: dict) -> str | None:
for pattern, floor in mapping.items():
if fnmatch.fnmatch(path, pattern):
return floor
return None
def build_agents(agents_cfg: dict, rules_cfg: dict, llm: LLMClient) -> list:
agents = []
for item in agents_cfg.get("agents", []):
if not item.get("enabled", True):
continue
cls = AGENT_REGISTRY.get(item["name"])
if cls is None:
print(f"[warn] 未知智能体: {item['name']}")
continue