Claude API 代码审查流水线:从 hook 到 CI
用 Claude Messages API 搭一条三层代码审查流水线:pre-commit 拦住本地问题、GitHub Actions 审查 PR diff、夜间扫全仓。含完整 review.py、误报压制方法与真实成本账。
lint 管格式,类型检查管接口,单元测试管行为——但它们都管不了「这段代码在并发下会不会丢数据」「这个 catch 是不是把异常吞了」。这类问题以前只能靠人看,现在可以交给模型,前提是你别把它做成一个只会刷屏的机器人。
这条流水线分三层:pre-commit 拦住本地能发现的问题,PR 阶段审查完整 diff,夜间扫一遍没人动的老代码。三层共用同一个 review.py,区别只是 diff 的来源和严重级别的门槛。
为什么不用现成的 review bot
现成的 PR 机器人有两个通病:把所有 diff 一股脑塞进上下文,导致长 PR 被截断;以及什么都报,包括「建议加注释」这种没人会改的东西。两周之后团队就会习惯性忽略它。
自建的好处是能控制三件事:只发变更行、只报六类问题、只在 high 级别阻塞。代价是要写一百多行代码,下面就是这一百多行。
第一步:构造一个便宜的 diff
diff 的构造方式直接决定成本。三个原则:
--unified=3而不是默认的 3——行,够模型理解上下文,又不至于把整个文件拖进来- 排除锁文件、构建产物、压缩后的资源,它们会让 token 数翻倍且毫无审查价值
- 单次请求的 diff 上限设 6 万字符,超了就按文件切分,一个文件一次请求
# 只取暂存区,忽略生成物
git diff --cached --unified=3 --no-color -- . \
':(exclude)*.lock' ':(exclude)package-lock.json' \
':(exclude)dist/*' ':(exclude)*.min.js'一个真实的 PR 通常在 200–800 行 diff、3000 tokens 上下。这个量级对 Claude Sonnet 5 来说很轻,单次成本两三分美元(后面有账)。
第二步:用 tool use 强制结构化输出
Anthropic 的 Messages API 没有 response_format。想要稳定的 JSON,最可靠的办法是定义一个只有 input_schema 的工具,然后用 tool_choice 强制调用它。模型没有别的出口,只能在 schema 里填内容。
{
"name": "report_findings",
"description": "提交代码审查结果",
"input_schema": {
"type": "object",
"properties": {
"findings": {
"type": "array",
"items": {
"type": "object",
"properties": {
"file": {"type": "string"},
"line": {"type": "integer"},
"severity": {"type": "string", "enum": ["high", "medium", "low"]},
"issue": {"type": "string"},
"suggestion": {"type": "string"}
},
"required": ["file", "line", "severity", "issue", "suggestion"]
}
}
},
"required": ["findings"]
}
}tool_choice: {"type": "tool", "name": "report_findings"} 这一步是关键。设成 auto 的话,模型在「没什么可报」时会选择不调用工具,你拿到一个空的 content,得额外判断;强制调用则稳定返回 findings: []。
完整 review.py
#!/usr/bin/env python3
"""ai-review:用 Claude Messages API 审查 git diff。"""
import argparse, os, subprocess, sys, time
import requests
BASE = os.environ.get("ZAIMOD_BASE", "https://api.zaimod.com").rstrip("/")
MODEL = os.environ.get("REVIEW_MODEL", "claude-sonnet-5")
KEY = os.environ["ZAIMOD_API_KEY"]
MAX_CHARS = 60_000
IGNORE = ("*.lock", "package-lock.json", "dist/*", "*.min.js", "*.svg")
RULES = """你是资深代码审查者。规则:
1. 只评论 diff 中新增或修改的行,不评论未改动的历史代码。
2. 不报纯风格问题(缩进、命名习惯、注释缺失、import 顺序)。
3. 只报六类:正确性缺陷、边界条件、并发与事务、安全、资源泄漏、错误处理缺失。
4. 每条必须能落到具体行;说不出具体行为后果的,不要报。
5. severity=high 仅用于会导致数据错误、线上故障或安全漏洞的问题。
6. 没有问题就返回空数组,不要为了凑数而报。
常见误报(不要报):日志文案、TODO 注释、测试文件里的硬编码、示例代码中的密钥占位符。
团队约定:所有外部调用必须有超时;数据库写操作必须显式事务;错误必须带上下文重新抛出。"""
TOOL = {
"name": "report_findings",
"description": "提交代码审查结果",
"input_schema": {
"type": "object",
"properties": {"findings": {"type": "array", "items": {
"type": "object",
"properties": {
"file": {"type": "string"}, "line": {"type": "integer"},
"severity": {"type": "string", "enum": ["high", "medium", "low"]},
"issue": {"type": "string"}, "suggestion": {"type": "string"}},
"required": ["file", "line", "severity", "issue", "suggestion"]}}},
"required": ["findings"]},
}
def collect_diff(staged=False, path=None):
if path:
with open(path, encoding="utf-8", errors="replace") as fh:
return fh.read()
cmd = ["git", "diff", "--unified=3", "--no-color"]
if staged:
cmd.append("--cached")
cmd += ["--", "."] + [f":(exclude){p}" for p in IGNORE]
return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout
def review(diff_text):
if not diff_text.strip():
return []
body = {
"model": MODEL, "max_tokens": 2048, "temperature": 0,
"system": [{"type": "text", "text": RULES,
"cache_control": {"type": "ephemeral"}}],
"tools": [TOOL],
"tool_choice": {"type": "tool", "name": "report_findings"},
"messages": [{"role": "user",
"content": "审查以下 diff:\n\n```diff\n"
+ diff_text[:MAX_CHARS] + "\n```"}],
}
for attempt in range(3):
r = requests.post(f"{BASE}/v1/messages",
headers={"x-api-key": KEY, "anthropic-version": "2023-06-01",
"content-type": "application/json"},
json=body, timeout=120)
if r.status_code in (429, 500, 502, 503, 504):
time.sleep(2 ** attempt)
continue
r.raise_for_status()
for block in r.json().get("content", []):
if block.get("type") == "tool_use":
return block["input"].get("findings", [])
return []
raise SystemExit("review failed: retries exhausted")
def to_markdown(findings):
if not findings:
return "### AI 审查\n\n未发现需要处理的问题。"
out = [f"### AI 审查:{len(findings)} 条", ""]
for f in findings:
out.append(f"- **{f['severity']}** `{f['file']}:{f['line']}` — {f['issue']}")
out.append(f" - 建议:{f['suggestion']}")
return "\n".join(out)
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--staged", action="store_true")
ap.add_argument("--file")
ap.add_argument("--fail-on", default="high", choices=["high", "medium", "none"])
ap.add_argument("--markdown-out")
a = ap.parse_args()
findings = review(collect_diff(a.staged, a.file))
for f in findings:
print(f"{f['severity'].upper():6} {f['file']}:{f['line']} {f['issue']}")
print(f" → {f['suggestion']}")
print(f"\n共 {len(findings)} 条")
if a.markdown_out:
with open(a.markdown_out, "w", encoding="utf-8") as fh:
fh.write(to_markdown(findings))
if a.fail_on == "high" and any(f["severity"] == "high" for f in findings):
sys.exit(1)注意三个细节。temperature: 0 让同一份 diff 的输出基本可复现;max_tokens 给 2048 够报十几条,再多说明规则写得太宽;system 用数组形式并挂 cache_control,这是 Anthropic 的缓存语法,重复调用时能省下 system 部分的输入费——但有个前提,下面「成本」一节会讲。
端点约定与字段细节在 Claude 文档 里,x-api-key、anthropic-version: 2023-06-01 两个头不能省。
第三步:pre-commit 拦住本地问题
本地这一层要快,超过 20 秒开发者就会开始找 --no-verify。只审暂存区,只阻塞 high:
#!/usr/bin/env bash
# .git/hooks/pre-commit
set -euo pipefail
[ "${SKIP_AI_REVIEW:-0}" = "1" ] && exit 0
command -v python3 >/dev/null || exit 0
python3 scripts/review.py --staged || {
echo ""
echo "AI 审查发现 high 级问题,提交已被阻止。"
echo "确认无误可跳过:SKIP_AI_REVIEW=1 git commit ..."
exit 1
}把 review.py 放进仓库的 scripts/ 目录,hook 用 git config core.hooksPath scripts/hooks 指向版本化的 hook 目录——这样团队新成员 clone 下来就生效,不用每人手动复制。
本地这层最容易翻车的地方是「改一行要等 8 秒」。可接受的做法是加缓存:把 git diff --cached 的哈希作为 key,命中就跳过。同一份 diff 反复 commit 失败重试时,这个优化能把等待降到零。
第四步:GitHub Actions 审查 PR
CI 这层不需要快,但需要稳定和可追溯。只在 PR 事件触发,并发取消,结果写成评论:
name: ai-review
on:
pull_request:
types: [opened, synchronize, reopened]
concurrency:
group: ai-review-${{ github.event.pull_request.number }}
cancel-in-progress: true
jobs:
review:
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
pull-requests: write
contents: read
steps:
- uses: actions/checkout@v4
with: { fetch-depth: 0 }
- uses: actions/setup-python@v5
with: { python-version: "3.12" }
- run: pip install requests
- name: Build diff
run: |
git diff "origin/${{ github.base_ref }}...HEAD" \
--unified=3 --no-color -- . > /tmp/pr.diff
wc -c /tmp/pr.diff
- name: Review
env:
ZAIMOD_API_KEY: ${{ secrets.ZAIMOD_API_KEY }}
run: python scripts/review.py --file /tmp/pr.diff --markdown-out review.md --fail-on none
- name: Comment
if: always()
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: gh pr comment "${{ github.event.pull_request.number }}" --body-file review.md--fail-on none 是刻意的:CI 阶段让审查永远不挂,只发评论。让一个概率性组件决定 PR 能不能合并,团队第一次遇到误报就会把它关掉。真正的门禁放在 pre-commit,那里人还在电脑前,能看到上下文。
把误报压下去
这套东西能不能活过两周,全看误报率。四个有效手段,按性价比排序:
一、规则里写死「不报什么」。 上面 RULES 里那段「常见误报」不是凑字数,它比正面规则更有效。团队每遇到一次误报,就往里加一条,两个月后误报会少一大半。
二、severity 分级要吝啬。 high 只留给「会导致数据错误、线上故障、安全漏洞」。全报 high 等于没有分级,pre-commit 会变成所有人的敌人。
三、只报 diff 行。 模型很乐意顺手指点旁边的老代码,但那不是这次 PR 的责任。规则第 1 条就是在拦这个。
四、基线抑制。 夜间全仓扫描时,把已确认不修的告警写进 .ai-review-baseline.json,报告里过滤掉。否则同一个老问题会每天出现在日报里,直到没人再看。
成本:一次 PR 多少钱
按 3000 输入 + 800 输出 token 估算,平台挂单价(美元 / 百万 tokens,default 分组):
| 模型 | 输入 | 输出 | 单次审查 | 每天 30 个 PR |
|---|---|---|---|---|
| gpt-5.3-codex | $1.75 | $14.00 | $0.0165 | $0.50 |
| claude-sonnet-5 | $4.00 | $20.00 | $0.0280 | $0.84 |
| qwen3-coder | $6.00 | $24.00 | $0.0372 | $1.12 |
| claude-sonnet-4-6 | $6.00 | $30.00 | $0.0420 | $1.26 |
| claude-opus-4-8 | $8.00 | $40.00 | $0.0560 | $1.68 |
一天 30 个 PR 的团队,月成本在 15–50 美元之间——比一个人一小时的人力便宜。真正需要算账的是夜间全仓扫描:一个中型仓库 5 万行代码,全量扫一次约 150 万输入 tokens,用 sonnet-5 大约 $6,一周一次就是 $24/月,和整个 PR 审查差不多。所以夜间扫描要按目录轮转,别每晚全量。
关于 cache_control:Anthropic 的缓存有最小长度门槛(Sonnet 约 1024 tokens),RULES 太短时 cache_creation_input_tokens 一直是 0,缓存根本没生效。把团队规范、常见误报、目录说明一起写进 system,凑到 1500 tokens 左右反而更省钱——这部分规则写得越细,单次成本越低。判断缓存是否真的生效,看响应里的 cache_read_input_tokens,方法在 base_url 迁移后的失效清单 里写过。
出错时的降级
三个必须处理的失败模式:
| 现象 | 原因 | 处理 |
|---|---|---|
| 429 | 并发超限 | 指数退避 1s / 2s / 4s,三次后跳过本次审查 |
| 请求超时 | diff 过大或网关 60s 截断 | 按文件切分,单文件一次请求 |
| 返回空 content | tool_choice 设成了 auto | 改成强制指定工具 |
降级的原则是审查可以缺失,合并不能阻塞。钩子里网络失败就直接放行并打印警告;CI 里失败就跳过评论。这类基础设施最怕的是「因为审查服务挂了,全组今天不能提交」。
结语
一条能长期活下去的 AI 审查流水线,特征不是「报得多」,而是「报得少但都对」。技术上真正关键的只有三点:用 tool_choice 锁死输出结构、用 diff 而不是整文件控制成本和噪音、用 severity 决定谁阻塞谁只提示。剩下的都是规则调优,而规则调优靠的是每次误报后往 RULES 里加一条,不是换更贵的模型。
用 Opus 跑同一份 diff,单次成本是 Sonnet 的两倍,误报率未必降一半——先把规则和 diff 构造做好,再考虑升级模型。价格随时会变,价格页 上是实时挂单价;/v1/messages 的字段与流式用法见 Claude 文档。
注册后自带测试额度,review.py 可以直接对着自己的仓库跑一轮;确认效果再决定日常用哪个模型,按量计费,余额不过期。第一次接入的完整步骤在 快速开始。