构建 Gerrit AI 代码审查机器人：从零到自动部署

背景

在团队开发中，代码审查（Code Review）是保证代码质量的重要环节。但随着团队规模扩大，每一次提交都需要等待人工审查，效率成为瓶颈。能否让 AI 自动审查每一次提交，在人工介入前先完成一轮自动化评审？

本文将分享如何构建一个基于大模型的 Gerrit 自动代码审查机器人，它能够：

实时监听 Gerrit 事件流
自动获取提交的代码变更
调用大模型 API 进行代码审查
将审查结果以评分配至 Gerrit 评论区

系统架构

整个系统围绕三个核心模块构建：

Gerrit 事件流监听 → 获取变更文件 → 调用 LLM 审查 → 提交审查结果至 Gerrit

核心组件

组件	职责
事件监听器	通过 SSH 连接 Gerrit 的 `stream-events` 接口，实时接收 `patchset-created` 事件
文件获取器	通过 Gerrit REST API 获取提交中变更的文件内容（Base64 解码）
LLM 审查引擎	将文件内容发送至大模型，获取结构化的审查报告
结果提交器	将审查结果以 Code-Review 打分 + Markdown 评论的形式提交回 Gerrit

关键实现细节

1. 实时事件监听

使用 paramiko 库通过 SSH 连接到 Gerrit，执行 gerrit stream-events 命令，这是一个长连接，Gerrit 会在有新事件时推送 JSON 数据：

ssh.connect(host, port=29418, username=user, timeout=15)
stdin, stdout, stderr = ssh.exec_command("gerrit stream-events")
while True:
    line = stdout.readline()
    if not line:
        continue
    event = json.loads(line)
    if event.get("type") == "patchset-created":
        handle_patchset(event)

关键点：通过 AutoAddPolicy 自动接受 SSH 主机密钥，并利用 while True 循环实现断线自动重连。

2. 获取提交文件内容

通过 Gerrit 的 REST API 获取变更文件列表，再逐个获取 Base64 编码的文件内容并解码：

# 获取文件列表
files_resp = requests.get(f"{API_URL}/changes/{change_id}/revisions/{revision_id}/files")

# 跳过 COMMIT_MSG 和 MERGE_LIST 等元文件
for file_path in files:
    content_resp = requests.get(f"{API_URL}/.../files/{file_path}/content")
    b64_content = content_resp.text.strip()
    plain_content = base64.b64decode(b64_content).decode("utf-8", errors="ignore")

关键点：Gerrit API 返回的 JSON 带有 )]}' 前缀，需要先去除；文件内容以 Base64 编码传输，需解码为明文。

3. 大模型审查提示词设计

审查引擎的核心是一份精心设计的 System Prompt，它定义了：

审查范围：语法错误、空指针、逻辑缺陷、代码风格、命名规范等
风险等级：高风险（编译失败/运行崩溃）、中风险（编码规范）、低风险（格式问题）
输出格式：严格的 JSON 结构，包含 score（整数评分）和 message（Markdown 格式报告）
Markdown 模板：包含问题清单、风险图标（🔴🟡🟢）、代码示例等

评分规则：

+1：无问题，可通过
0：存在低/中风险，建议优化
-1：存在高风险，必须修复

4. 审查结果提交

将 LLM 的审查结果通过 Gerrit API 提交评论，并附带 Code-Review 打分：

payload = {
   
    "message": formatted_message,
    "labels": {
   "Code-Review": score}
}
requests.post(review_url, json=payload, auth=(user, password))

遇到的挑战与解决方案

挑战 1：大模型 API 超时

代码审查涉及大量文件内容，API 请求的读取阶段可能耗时较长。

解决方案：将超时设置为 (10, 280)，即连接超时 10 秒，读取超时 280 秒，给大模型足够的推理时间。

挑战 2：SSH 连接不稳定

与 Gerrit 的 SSH 连接可能因网络波动而断开。

解决方案：外层 while True 循环配合 try-except，在连接断开时等待 5 秒后自动重连。

挑战 3：JSON 格式严格性

大模型输出偶尔会包含 ```json 代码块标记，导致解析失败。

解决方案：在提示词中强调"禁止在 JSON 外包裹代码块"，并使用 json.loads() 严格解析。实际生产中建议增加失败重试或正则清理逻辑。

部署与运行

环境要求

Python 3.8+
依赖库：paramiko、requests

启动方式

系统提供两种运行模式：

1. 手动测试模式 — 审查指定的 Change：

manual_test()

2. 自动监听模式 — 持续监听 Gerrit 事件流：

watch_events()

建议先在手动测试模式下验证配置和审查效果，确认无误后再切换到自动监听模式。

效果展示

当有新的 PatchSet 提交时，机器人会自动：

打印审核开始信息（提交标题、Change ID）
获取变更文件列表
调用大模型生成审查报告
在 Gerrit 评论区生成类似以下格式的审查结果：

Code-Review： ✅ +1 通过

📋 总结

高风险问题：0 个

中风险问题：1 个

低风险问题：2 个

最终建议： 建议优化后合并

扩展方向

这个机器人还有很多可以优化的方向：

增量审查 — 只审查变更的行而非整个文件，减少 Token 消耗
多模型支持 — 根据文件类型选择不同的审查模型
审查缓存 — 对未变更的文件跳过重复审查
通知集成 — 审查完成后通过企业微信/钉钉/邮件通知开发者
自定义规则引擎 — 结合静态分析工具（如 SonarQube）提升审查准确度
Webhook 替代 SSH — 使用 Gerrit Webhook 插件替代 SSH 流式监听，更稳定可靠

总结

通过大模型与 Gerrit API 的结合，我们构建了一个全自动的 AI 代码审查机器人。它能够 7×24 小时不间断地审查每一次代码提交，在人工审查前提前发现潜在问题，显著提升代码审查效率。

当然，AI 审查并不能完全替代人工审查，尤其是在业务逻辑理解、架构设计评估等方面。但它可以作为第一道防线，过滤掉明显的低级错误和规范问题，让人工审查能够更专注于更高层次的代码质量评估。

代码审查的未来，不是人被替换，而是人和 AI 各司其职。

本文基于实际项目经验撰写，代码中的服务器地址、密钥等敏感信息已脱敏处理。

完整代码：

import paramiko
import json
import requests
import time
import base64

# ==================== 【配置区】 ====================
GERRIT_HOST = "***REDACTED***" # Gerrit 服务器地址，示例: "gerrit.example.com"
GERRIT_SSH_PORT = 29418
GERRIT_USER = "***REDACTED***" # Gerrit 用户名，需具备审核权限
GERRIT_HTTP_PASS = "***REDACTED***" # gerrit 后台获取
GERRIT_API_URL = "https://***REDACTED***/a" # Gerrit API 基础 URL，末尾需加 /a 以启用认证访问

# OpenAI 大模型 API 配置（按要求留空）
OPENAI_BASE_URL = "http://***REDACTED***/v1/chat/completions" # 大模型 API URL，示例: "https://api.openai.com/v1/chat/completions"
OPENAI_API_KEY = "***REDACTED***" # 大模型 API Key，示例: "sk-xxxxxx"
OPENAI_MODEL = "***REDACTED***" # 大模型名称，示例: "gpt-4.0" 或 "minimax/minimax-m3"
# 大模型请求超时：(连接超时秒数, 读取超时秒数)，审查大文件时读取阶段可能较久
OPENAI_TIMEOUT = (10, 280)


AI_PROMPT_TEMPLATE = """
(提示词下面给出)
"""


# =====================================================

def llm_code_review(files_content: dict) -> dict:
    """
    调用大模型 API 执行代码审查。
    返回格式：{"score": 分数(int), "message": Markdown 格式审查报告(str)}
    """
    if not OPENAI_API_KEY or not OPENAI_BASE_URL:
        return {
   "score": 0, "message": "大模型API配置缺失，审核终止"}

    prompt = AI_PROMPT_TEMPLATE + json.dumps(files_content, ensure_ascii=False, indent=2)

    print(prompt)

    headers = {
   
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
   
        "model": OPENAI_MODEL,
        "messages": [{
   "role": "user", "content": prompt}],
        "max_tokens": 4096,
        "stream": False,
        "temperature": 0.7,
        "top_p": 0.9
    }

    try:
        resp = requests.post(
            url=f"{OPENAI_BASE_URL}",
            headers=headers,
            json=payload,
            timeout=OPENAI_TIMEOUT
        )
        resp.raise_for_status()
        llm_data = resp.json()
        result_content = llm_data["choices"][0]["message"]["content"].strip()
        return json.loads(result_content)
    except Exception as e:
        print(f"大模型调用异常: {str(e)}")
        return {
   "score": 0, "message": f"大模型接口调用失败：{str(e)}"}


def get_change_files(change_id: str, revision_id: str) -> dict:
    """获取本次提交所有文件内容，并对Base64内容解码为明文"""
    try:
        url = f"{GERRIT_API_URL}/changes/{change_id}/revisions/{revision_id}/files"
        resp = requests.get(url, auth=(GERRIT_USER, GERRIT_HTTP_PASS), verify=False)
        files = json.loads(resp.text.lstrip(")]}'"))

        file_data = {
   }
        for file_path in files:
            if file_path in ("/COMMIT_MSG", "/MERGE_LIST"):
                continue
            content_url = f"{GERRIT_API_URL}/changes/{change_id}/revisions/{revision_id}/files/{file_path}/content"
            content_resp = requests.get(content_url, auth=(GERRIT_USER, GERRIT_HTTP_PASS), verify=False)
            b64_content = content_resp.text.strip()
            plain_content = base64.b64decode(b64_content).decode("utf-8", errors="ignore")
            file_data[file_path] = plain_content
        return file_data
    except Exception as e:
        print(f"获取文件异常: {e}")
        return {
   }


def send_review(change_id: str, revision_id: str, llm_result: dict):
    """将大模型审核结果提交至Gerrit"""
    review_url = f"{GERRIT_API_URL}/changes/{change_id}/revisions/{revision_id}/review"
    score = llm_result.get("score", 0)
    message = llm_result.get("message", "暂无审核信息")

    score_label = {
   1: "✅ +1 通过", 0: "⚠️ 0 建议优化", -1: "❌ -1 必须修复"}.get(score, f"评分 {score}")
    final_msg = f"**Code-Review：** {score_label}\n\n---\n\n{message}"
    payload = {
   
        "message": final_msg,
        "labels": {
   
            "Code-Review": score
        }
    }
    resp = requests.post(review_url, json=payload, auth=(GERRIT_USER, GERRIT_HTTP_PASS), verify=False)
    print(f"Gerrit接口返回码: {resp.status_code} | 最终打分: {score}")


def run_review_logic(change_id: str, revision_id: str, subject: str):
    """核心审核逻辑"""
    print(f"\n【开始审核】提交标题: {subject} | ChangeID: {change_id}")
    files = get_change_files(change_id, revision_id)
    if not files:
        print("未获取到文件，终止审核")
        return

    # 仅调用大模型完成审核
    llm_result = llm_code_review(files)
    print(f"大模型原始审核结果: {llm_result}")

    # 推送结果到Gerrit
    send_review(change_id, revision_id, llm_result)


def handle_patchset(event):
    """事件触发入口（自动流程）"""
    change_id = event["change"]["id"]
    revision_id = event["patchSet"]["id"]
    subject = event["change"]["subject"]
    run_review_logic(change_id, revision_id, subject)


def watch_events():
    """监听 Gerrit 事件流，断线自动重连"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    while True:
        try:
            ssh.connect(GERRIT_HOST, port=GERRIT_SSH_PORT, username=GERRIT_USER, timeout=15)
            print("✅ 已连接 Gerrit 事件流，等待提交...")
            stdin, stdout, stderr = ssh.exec_command("gerrit stream-events")
            while True:
                line = stdout.readline()
                if not line:
                    time.sleep(0.1)
                    continue
                try:
                    event = json.loads(line)
                    if event.get("type") == "patchset-created":
                        handle_patchset(event)
                except Exception:
                    continue
        except Exception as e:
            print(f"连接断开: {e}，5秒后重连...")
            time.sleep(5)


def manual_test():
    """手动测试入口"""
    test_change_id = "***REDACTED***"
    test_revision_id = "***REDACTED***"
    test_subject = "***REDACTED***"
    run_review_logic(test_change_id, test_revision_id, test_subject)


if __name__ == "__main__":
    requests.packages.urllib3.disable_warnings()

    # 二选一运行：
    # 1. 先跑手动测试
    manual_test()

    # 2. 测试通了再开启自动监听
    # watch_events()

提示词

你是资深后端代码审查工程师，严格依据 Google Java Style Guide、Java 官方语法规范、工程编码准则，对提交的代码文件进行静态审查。

审查范围

语法错误、编译错误、空指针风险、逻辑缺陷、无效代码、代码风格、文件格式、命名规范、不可达代码等。

风险等级

高：编译失败、运行崩溃、空指针、语法非法、严重逻辑错误（必须整改）
中：违反编码规范、文件格式异常、代码冗余（建议整改）
低：缩进、换行、命名风格等格式问题（按需优化）

输出格式（极其重要）

最外层必须是纯 JSON 字符串，禁止在 JSON 外包裹 ```json 代码块或任何说明文字。
JSON 仅包含两个字段：score（整数）、message（字符串）。
message 字段内容必须使用 Markdown 排版（Gerrit 评论支持 Markdown 渲染），换行用 \n，引号需正确转义以保证 JSON 合法。
score 规则：无问题 = 1；仅低/中风险 = 0；存在高风险 = -1。

message 字段 Markdown 模板（严格按此结构输出）

🤖 AI 代码审查报告

审查文件： 文件路径
审查依据： Google Java Style Guide / Java 语言规范
整体结论： 一句话总结（通过 / 需修改 / 必须修复）

问题清单

问题 1 · 🔴 高风险

项目	内容
说明	问题细节及违反的规范
影响	编译报错 / 运行异常 / 等
建议	明确修复方案

// 建议修改示例（如有）

问题 2 · 🟡 中风险

项目	内容
说明	...
影响	...
建议	...

（无问题时，删除「问题清单」区块，改为：）

✅ 未发现需要整改的问题，代码规范合规。

📋 总结

高风险问题：N 个
中风险问题：N 个
低风险问题：N 个
最终建议： 合并前必须修复 / 建议优化后合并 / 可直接合并

排版要求

标题用 ## / ### / ####，关键信息用粗体，文件路径和符号用 行内代码
代码示例用 ```java 代码块
风险图标：高 🔴、中 🟡、低 🟢
内容简洁专业，避免重复堆砌

待审核文件（文件路径: 代码内容）：

构建 Gerrit AI 代码审查机器人：从零到自动部署

构建 Gerrit AI 代码审查机器人：从零到自动部署

背景

系统架构

核心组件

关键实现细节

1. 实时事件监听

2. 获取提交文件内容

3. 大模型审查提示词设计

4. 审查结果提交

遇到的挑战与解决方案

挑战 1：大模型 API 超时

挑战 2：SSH 连接不稳定

挑战 3：JSON 格式严格性

部署与运行

环境要求

启动方式

效果展示

📋 总结

扩展方向

总结

完整代码：

提示词

审查范围

风险等级

输出格式（极其重要）

message 字段 Markdown 模板（严格按此结构输出）

🤖 AI 代码审查报告

问题清单

问题 1 · 🔴 高风险

问题 2 · 🟡 中风险

📋 总结

排版要求

热门文章

最新文章

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

构建 Gerrit AI 代码审查机器人：从零到自动部署

构建 Gerrit AI 代码审查机器人：从零到自动部署

背景

系统架构

核心组件

关键实现细节

1. 实时事件监听

2. 获取提交文件内容

3. 大模型审查提示词设计

4. 审查结果提交

遇到的挑战与解决方案

挑战 1：大模型 API 超时

挑战 2：SSH 连接不稳定

挑战 3：JSON 格式严格性

部署与运行

环境要求

启动方式

效果展示

📋 总结

扩展方向

总结

完整代码：

提示词

审查范围

风险等级

输出格式（极其重要）

message 字段 Markdown 模板（严格按此结构输出）

🤖 AI 代码审查报告

问题清单

问题 1 · 🔴 高风险

问题 2 · 🟡 中风险

📋 总结

排版要求

热门文章

最新文章

相关电子书