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 万字符,超了就按文件切分,一个文件一次请求
Shell
# 只取暂存区,忽略生成物
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 里填内容。

JSON
{
  "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

Python
#!/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:

Shell
#!/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 事件触发,并发取消,结果写成评论:

YAML
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 截断按文件切分,单文件一次请求
返回空 contenttool_choice 设成了 auto改成强制指定工具

降级的原则是审查可以缺失,合并不能阻塞。钩子里网络失败就直接放行并打印警告;CI 里失败就跳过评论。这类基础设施最怕的是「因为审查服务挂了,全组今天不能提交」。

结语

一条能长期活下去的 AI 审查流水线,特征不是「报得多」,而是「报得少但都对」。技术上真正关键的只有三点:用 tool_choice 锁死输出结构、用 diff 而不是整文件控制成本和噪音、用 severity 决定谁阻塞谁只提示。剩下的都是规则调优,而规则调优靠的是每次误报后往 RULES 里加一条,不是换更贵的模型。

用 Opus 跑同一份 diff,单次成本是 Sonnet 的两倍,误报率未必降一半——先把规则和 diff 构造做好,再考虑升级模型。价格随时会变,价格页 上是实时挂单价;/v1/messages 的字段与流式用法见 Claude 文档。

注册后自带测试额度,review.py 可以直接对着自己的仓库跑一轮;确认效果再决定日常用哪个模型,按量计费,余额不过期。第一次接入的完整步骤在 快速开始。

延伸阅读

#claude api 代码审查#git hook#github actions#messages api#prompt caching
免费创建 API Key

相关阅读

DeepSeek 免费档能干什么:5 个生产级用法

平台上有一个按次计费 0 元的 deepseek-v4-flash-free 免费档,每天限流。本文给出 5 个真正能上生产的用法:意图分类、内容打标签、结构化抽取、会话摘要、批量预处理,并附用量测算与降级策略。

流式输出 SSE 原理与调试:从 curl 到前端

大模型 API 的流式输出本质是一条 SSE 长连接。本文拆解 text/event-stream 报文格式、stream:true 的分片结构、curl 与前端 ReadableStream 的调试方法,以及代理缓冲、[DONE]、usage 丢失等高频坑位。