作者：PySuper | 来源：zhengxingtao.com
更新日期：2026-09-01
前言
Code Review（代码审查）是保证代码质量的重要环节，但传统的人工 Review 面临诸多痛点：时间成本高、审查标准不一致、reviewer 疲劳导致漏检等。
大语言模型的出现为 Code Review 带来了新的可能性。本文将深入探讨如何用 AI 构建高效的 Code Review 系统，包括实现方案、实践技巧和完整代码示例。
一、传统 Code Review 的痛点
1.1 行业现状
┌─────────────────────────────────────────────────────────────────────┐
│ 传统 Code Review 痛点分析 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ⏰ 时间成本高 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ 一位工程师平均每天花费: │ │
│ │ ├── 写代码: 3-4 小时 │ │
│ │ ├── Code Review: 1-2 小时 ◄─── 被动消耗 │ │
│ │ └── 其他: 1-2 小时 │ │
│ │ │ │
│ │ 一周按 5 天算: │ │
│ │ Code Review 总耗时 = 5-10 小时/周 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 📊 标准不一致 │
│ ├── 不同 reviewer 风格不同 │
│ ├── 同一 reviewer 不同时间标准不同 │
│ ├── 经验丰富的 reviewer vs 新人 reviewer │
│ └── 业务高峰期 Review 质量下降 │
│ │
│ 👤 主观性问题 │
│ ├── "我觉得这个命名不好" → 没有客观标准 │
│ ├── "这里可以优化" → 缺少具体建议 │
│ └── "这段代码有问题" → 没有指出具体问题 │
│ │
│ 🔍 遗漏风险 │
│ ├── 安全漏洞（SQL注入、XSS等） │
│ ├── 性能问题（N+1查询、内存泄漏） │
│ ├── 并发问题（竞态条件、死锁） │
│ └── 边界情况未处理 │
│ │
└─────────────────────────────────────────────────────────────────────┘1.2 痛点量化分析
┌─────────────────────────────────────────────────────────────────────┐
│ Code Review 效率瓶颈分析 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 传统 Review 时间分布: │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ 理解代码逻辑 ████████████████████░░░░░░░ 40% │ │
│ │ │ │
│ │ 检查代码规范 ████████████░░░░░░░░░░░░░░░░ 25% │ │
│ │ │ │
│ │ 发现明显 Bug ████████░░░░░░░░░░░░░░░░░░░░ 15% │ │
│ │ │ │
│ │ 安全/性能检查 █████░░░░░░░░░░░░░░░░░░░░░░░ 10% │ │
│ │ │ │
│ │ 撰写 Review ████░░░░░░░░░░░░░░░░░░░░░░░░ 10% │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 💡 AI 介入后变化: │
│ │
│ ├── 理解代码逻辑: AI 完成（70%节省） │
│ ├── 检查代码规范: AI 完成（80%节省） │
│ ├── 安全/性能检查: AI 完成（60%节省） │
│ └── 人工聚焦: 业务逻辑、架构设计、关键决策 │
│ │
│ 📈 整体效率提升: 40-60% │
│ │
└─────────────────────────────────────────────────────────────────────┘二、AI Code Review 的实现方案
2.1 系统架构设计
┌─────────────────────────────────────────────────────────────────────┐
│ AI Code Review 架构图 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Git 仓库 │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Webhook 触发 │ │
│ │ (PR/MR 创建) │ │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Review Pipeline │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐│ │
│ │ │ 代码解析 │──►│ 差异提取 │──►│ LLM 分析 │──►│ 结果聚合 ││ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘│ │
│ │ │ │ │ │ │ │
│ │ ▼ ▼ ▼ ▼ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐│ │
│ │ │ AST 分析 │ │ 上下文 │ │ 安全扫描 │ │ 格式化 ││ │
│ │ │ │ │ 构建 │ │ 性能检测 │ │ 输出 ││ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘│ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ 结果输出 │ │
│ │ │ │
│ │ ┌───────────┐ │ │
│ │ │ PR 评论 │ │ │
│ │ ├───────────┤ │ │
│ │ │ Slack/钉钉│ │ │
│ │ ├───────────┤ │ │
│ │ │ Dashboard │ │ │
│ │ └───────────┘ │ │
│ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 技术选型
# tech_stack.yml
# AI Code Review 技术栈配置
# 核心框架
llm_framework:
primary: langchain
# 或选择 langgraph (复杂流程)
# LLM 选择
llm:
provider: anthropic # anthropic / openai / azure
model: claude-sonnet-4-5
fallback: deepseek-chat
# 本地部署选项 (隐私敏感场景)
# local:
# provider: ollama
# model: codellama:13b
# 代码解析
code_parser:
language: python
tools:
- tree-sitter # AST 解析
- jupyter-book # Notebook 支持
# Git 集成
git:
providers:
github:
enabled: true
webhook_secret: ${GITHUB_WEBHOOK_SECRET}
gitlab:
enabled: true
gitee:
enabled: false
# CI/CD 集成
cicd:
github_actions: true
gitlab_ci: false
jenkins: false
# 通知渠道
notification:
slack:
enabled: true
webhook_url: ${SLACK_WEBHOOK_URL}
dingtalk:
enabled: true
webhook_url: ${DINGTALK_WEBHOOK_URL}
feishu:
enabled: false
# 存储
storage:
database: postgresql://localhost:5432/codereview
cache: redis://localhost:6379/0
# 监控
monitoring:
prometheus: true
grafana: true三、大模型能 Review 什么
3.1 Review 能力矩阵
┌─────────────────────────────────────────────────────────────────────┐
│ AI Review 能力矩阵 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ AI 擅长领域 │ │
│ ├─────────────────────────────────────────────────────────────┤ │
│ │ │ │
│ │ ✅ 安全漏洞检测 │ │
│ │ ├── SQL 注入风险 │ │
│ │ ├── XSS 跨站脚本 │ │
│ │ ├── 命令注入 │ │
│ │ ├── 敏感信息泄露 │ │
│ │ ├── 不安全的加密使用 │ │
│ │ └── 认证/授权缺陷 │ │
│ │ │ │
│ │ ✅ 性能问题 │ │
│ │ ├── N+1 查询问题 │ │
│ │ ├── 循环中的数据库查询 │ │
│ │ ├── 大数据加载到内存 │ │
│ │ ├── 不必要的重复计算 │ │
│ │ └── 缺少必要的索引 │ │
│ │ │ │
│ │ ✅ 代码规范 │ │
│ │ ├── 命名不规范 │ │
│ │ ├── 函数过长 │ │
│ │ ├── 重复代码 │ │
│ │ ├── 缺少注释 │ │
│ │ └── import 顺序混乱 │ │
│ │ │ │
│ │ ✅ 错误处理 │ │
│ │ ├── 裸 except │ │
│ │ ├── 异常被静默吞掉 │ │
│ │ ├── 错误信息不友好 │ │
│ │ └── 缺少超时处理 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ AI 较弱领域 │ │
│ ├─────────────────────────────────────────────────────────────┤ │
│ │ │ │
│ │ ⚠️ 业务逻辑正确性 │ │
│ │ ├── 业务规则理解 │ │
│ │ ├── 边界条件判断 │ │
│ │ └── 需求完整性 │ │
│ │ │ │
│ │ ⚠️ 架构设计 │ │
│ │ ├── 模块划分是否合理 │ │
│ │ ├── 依赖关系是否清晰 │ │
│ │ └── 技术选型是否恰当 │ │
│ │ │ │
│ │ ⚠️ 上下文依赖 │ │
│ │ ├── 团队代码规范 │ │
│ │ ├── 项目特殊约定 │ │
│ │ └── 历史遗留代码影响 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘3.2 检测规则定义
# rules/security_rules.py
"""
安全检测规则定义
"""
SECURITY_RULES = {
# SQL 注入风险
"sql_injection": {
"severity": "critical",
"patterns": [
{
"type": "raw_sql",
"pattern": r'execute\s*\(\s*["\'].*%s.*["\']',
"message": "检测到原始 SQL 查询，可能存在 SQL 注入风险",
"suggestion": "使用参数化查询或 ORM 方法"
},
{
"type": "string_format_sql",
"pattern": r'cursor\.execute\s*\([^)]*\+[^)]+\)',
"message": "字符串拼接 SQL 查询",
"suggestion": "使用参数化查询"
},
{
"type": "f_string_sql",
"pattern": r'cursor\.execute\s*\([^)]*f["\']',
"message": "f-string 构造 SQL 查询",
"suggestion": "使用参数化查询"
}
],
"examples": [
{
"bad": 'cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")',
"good": 'cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))'
}
]
},
# XSS 风险
"xss_vulnerability": {
"severity": "high",
"patterns": [
{
"type": "direct_html_render",
"pattern": r'render\s*\([^)]*\.html\s*,\s*\{[^}]*\+=',
"message": "直接拼接用户输入到 HTML 响应"
},
{
"type": "inner_html_assignment",
"pattern": r'innerHTML\s*=\s*.*\+',
"message": "使用 innerHTML 拼接内容"
}
],
"sanitization": [
" bleach.clean() ",
" markupsafe.escape() ",
" html.escape() "
]
},
# 敏感信息泄露
"sensitive_data_exposure": {
"severity": "high",
"patterns": [
{
"type": "hardcoded_password",
"pattern": r'password\s*=\s*["\'][^"\']{8,}["\']',
"exclusions": ["password", "old_password", "hashed_password"]
},
{
"type": "api_key_in_code",
"patterns": [
r'api[_-]?key\s*=\s*["\'][A-Za-z0-9]{20,}["\']',
r'secret[_-]?key\s*=\s*["\'][A-Za-z0-9]{20,}["\']',
r'token\s*=\s*["\'][A-Za-z0-9]{20,}["\']'
]
},
{
"type": "private_key",
"pattern": r'-----BEGIN\s+(RSA\s+)?PRIVATE\s+KEY-----'
}
],
"suggestion": "使用环境变量或密钥管理服务"
},
# 命令注入
"command_injection": {
"severity": "critical",
"patterns": [
{
"type": "os_system",
"pattern": r'os\.system\s*\([^)]*\+[^)]+\)'
},
{
"type": "subprocess_shell",
"pattern": r'subprocess\.(run|call|Popen)\s*\([^)]*shell\s*=\s*True'
},
{
"type": "eval_usage",
"pattern": r'\beval\s*\('
},
{
"type": "exec_usage",
"pattern": r'\bexec\s*\('
}
]
}
}
# 性能检测规则
PERFORMANCE_RULES = {
"n_plus_one_query": {
"severity": "medium",
"patterns": [
{
"type": "loop_query",
"framework": "django",
"pattern": r'for\s+\w+\s+in\s+\w+:\s*\n\s+\w+\.\w+\.get\(|\w+\.\w+\.filter\('
},
{
"type": "loop_query",
"framework": "sqlalchemy",
"pattern": r'for\s+\w+\s+in\s+\w+:\s*\n\s+.*\.query\.'
}
],
"fix": "使用 prefetch_related() 或 select_related()"
},
"memory_leak": {
"severity": "medium",
"patterns": [
{
"type": "large_file_read",
"pattern": r'open\([^)]+\)\.read\(\)',
"threshold": "files > 1MB"
},
{
"type": "accumulator",
"pattern": r'while\s+True:\s*\n\s+\w+\.append\('
}
]
}
}
# 代码规范规则
CODE_STYLE_RULES = {
"function_length": {
"severity": "info",
"threshold": 50, # 行数
"suggestion": "函数过长，建议拆分为多个小函数"
},
"class_length": {
"severity": "info",
"threshold": 300, # 行数
"suggestion": "类过长，考虑拆分为多个类或使用模块化"
},
"duplicate_code": {
"severity": "warning",
"threshold": 5, # 重复行数
"min_similarity": 0.8 # 相似度
},
"naming": {
"severity": "info",
"rules": {
"function": r'^[a-z_][a-z0-9_]*$', # snake_case
"class": r'^[A-Z][a-zA-Z0-9]*$', # PascalCase
"constant": r'^[A-Z_][A-Z0-9_]*$' # UPPER_CASE
}
}
}四、大模型不能 Review 什么
4.1 局限性分析
┌─────────────────────────────────────────────────────────────────────┐
│ AI Code Review 局限性 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ⚠️ 业务逻辑正确性 │
│ ══════════════════════════════════════════════ │
│ │
│ 问题: AI 无法理解业务规则和领域知识 │
│ │
│ 示例: │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ # 业务规则：VIP用户月消费满1000元升级为SVIP │ │
│ │ def upgrade_vip_level(user): │ │
│ │ if user.monthly_spend >= 1000: │ │
│ │ user.level = 'SVIP' │ │
│ │ user.save() │ │
│ │ │ │
│ │ # 问题：这个代码在某些边界情况可能有问题， │ │
│ │ # 比如用户刚完成一笔大额订单但还未结算 │ │
│ │ # AI 无法知道这个业务规则 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 解决: 需要人工 Review + 业务规则文档化 │
│ │
│ ───────────────────────────────────────────────────────────── │
│ │
│ ⚠️ 上下文理解 │
│ ══════════════════════════════════════════════ │
│ │
│ 问题: AI 无法完全理解项目历史和团队约定 │
│ │
│ 示例: │
│ ├── 某些"坏味道"代码可能是为了兼容性 │
│ ├── 历史遗留代码可能有特殊原因 │
│ └── 团队内部约定 AI 无法获知 │
│ │
│ 解决: 提供项目 README + 架构文档给 AI │
│ │
│ ───────────────────────────────────────────────────────────── │
│ │
│ ⚠️ 架构决策 │
│ ══════════════════════════════════════════════ │
│ │
│ 问题: 架构级决策难以在单个 PR 中评估 │
│ │
│ 示例: │
│ ├── 模块划分是否合理 │
│ ├── 引入新框架是否必要 │
│ └── 技术债的取舍 │
│ │
│ 解决: 架构级 Review 需要单独流程 │
│ │
└─────────────────────────────────────────────────────────────────────┘4.2 人工 Review 要点
┌─────────────────────────────────────────────────────────────────────┐
│ 人工 Review 清单 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 🤖 AI Review 已覆盖（可跳过） │
│ ├── ✅ 安全漏洞 │
│ ├── ✅ 性能问题 │
│ ├── ✅ 代码规范 │
│ ├── ✅ 错误处理 │
│ └── ✅ 类型注解 │
│ │
│ 👤 人工 Review 必须关注 │
│ ├── ⬜ 业务逻辑正确性 │
│ │ ├── 业务规则是否符合需求文档 │
│ │ ├── 边界条件处理是否完整 │
│ │ └── 数据流转是否正确 │
│ │ │
│ ├── ⬜ 架构设计 │
│ │ ├── 是否符合项目架构 │
│ │ ├── 是否引入不必要的复杂度 │
│ │ └── 是否影响其他模块 │
│ │ │
│ ├── ⬜ 可测试性 │
│ │ ├── 是否便于单元测试 │
│ │ └── 是否需要集成测试 │
│ │ │
│ ├── ⬜ 可维护性 │
│ │ ├── 后续改动成本 │
│ │ └── 技术债 │
│ │ │
│ └── ⬜ 沟通确认 │
│ ├── 实现是否与 PR 描述一致 │
│ └── 是否有遗漏 │
│ │
└─────────────────────────────────────────────────────────────────────┘五、误报率控制策略
5.1 误报来源分析
┌─────────────────────────────────────────────────────────────────────┐
│ 误报率控制策略 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 常见误报类型: │
│ │
│ 1️⃣ 规则过于严格 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ "代码超过 50 行，建议拆分" │ │
│ │ │ │
│ │ 误报场景: │ │
│ │ - 配置字典 (合法的大数据结构) │ │
│ │ - 测试代码 (setup/teardown 天然较长) │ │
│ │ - 自动化生成代码 │ │
│ │ │ │
│ │ 解决: 添加豁免注释和规则配置 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 2️⃣ 上下文缺失 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ "使用 eval() 可能不安全" │ │
│ │ │ │
│ │ 误报场景: │ │
│ │ - 解析配置文件 (合理使用场景) │ │
│ │ - DSL 实现 (设计如此) │ │
│ │ - 测试代码中的动态执行 │ │
│ │ │ │
│ │ 解决: 上下文分析和人工确认 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ 3️⃣ 框架特性误解 │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ "循环中查询数据库" │ │
│ │ │ │
│ │ 误报场景: │ │
│ │ - Django ORM lazy loading (queryset 是 lazy 的) │ │
│ │ - Batch 操作已在外层 │ │
│ │ │ │
│ │ 解决: AST 分析和 LLM 理解结合 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘5.2 误报控制实现
# core/false_positive_control.py
"""
误报率控制模块
通过多维度分析降低误报率
"""
import re
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Callable
from enum import Enum
class Severity(Enum):
"""问题严重程度"""
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
INFO = "info"
@dataclass
class ReviewFinding:
"""审查发现"""
rule_id: str
severity: Severity
file: str
line: int
message: str
suggestion: str
confidence: float = 1.0 # 置信度 0-1
is_false_positive: bool = False
false_positive_reason: Optional[str] = None
evidence: List[str] = field(default_factory=list)
@dataclass
class Context:
"""代码上下文"""
file_path: str
file_content: str
language: str
framework: Optional[str] = None
project_type: Optional[str] = None
annotations: List[str] = field(default_factory=list) # 豁免注释
class FalsePositiveController:
"""
误报控制器
多层策略降低误报率
"""
def __init__(self):
# 豁免注释模式
self.exemption_patterns = [
r'#\s*no-review\s*:\s*(\w+)', # no-review: security
r'#\s*pragma\s*:\s*no-cover',
r'#\s* noqa\s*:\s*([A-Z]+)', # flake8 noqa
r'#\s*type:\\s*ignore',
]
# 安全使用的白名单
self.safe_patterns = {
"eval": [
r'eval\s*\(\s*ast\.literal_eval', # 安全解析
r'eval\s*\(\s*json\.loads', # JSON 解析
],
"os_system": [
r'os\.system\s*\(\s*["\'].*shlex\.quote', # 安全转义
],
"raw_sql": [
r'\.extra\s*\(\s*where\s*=', # Django ORM extra
]
}
# 豁免文件模式
self.exempt_file_patterns = [
r'tests?/.*',
r'_test\.py$',
r'conftest\.py',
r'fixtures?/.*',
r'migrations?/.*',
r'.*_generated\.py$',
r'.*\.min\.js$',
]
def should_exempt(
self,
finding: ReviewFinding,
context: Context
) -> bool:
"""
判断是否应该豁免
Args:
finding: 审查发现
context: 代码上下文
Returns:
bool: 是否豁免
"""
# 1. 检查豁免注释
if self._has_exemption_comment(finding, context):
finding.is_false_positive = True
finding.false_positive_reason = "检测到豁免注释"
return True
# 2. 检查豁免文件
if self._is_exempt_file(finding.file):
finding.is_false_positive = True
finding.false_positive_reason = "豁免文件类型"
return True
# 3. 检查安全使用模式
if self._is_safe_usage(finding, context):
finding.is_false_positive = True
finding.false_positive_reason = "检测到安全使用模式"
return True
# 4. 置信度阈值过滤
if finding.confidence < 0.6:
finding.is_false_positive = True
finding.false_positive_reason = "置信度低于阈值"
return True
return False
def _has_exemption_comment(
self,
finding: ReviewFinding,
context: Context
) -> bool:
"""检查是否有豁免注释"""
# 获取问题代码行及其上下文（前后5行）
lines = context.file_content.split('\n')
start_line = max(0, finding.line - 6)
end_line = min(len(lines), finding.line + 1)
relevant_lines = '\n'.join(lines[start_line:end_line])
for pattern in self.exemption_patterns:
if re.search(pattern, relevant_lines, re.IGNORECASE):
return True
return False
def _is_exempt_file(self, file_path: str) -> bool:
"""检查是否为豁免文件"""
for pattern in self.exempt_file_patterns:
if re.match(pattern, file_path):
return True
return False
def _is_safe_usage(
self,
finding: ReviewFinding,
context: Context
) -> bool:
"""检查是否为安全使用"""
rule_id = finding.rule_id
# 提取规则ID中的关键部分（如 sql_injection -> sql）
key = rule_id.split('_')[0] if '_' in rule_id else rule_id
if key not in self.safe_patterns:
return False
# 获取代码行内容
lines = context.file_content.split('\n')
if finding.line > len(lines):
return False
code_line = lines[finding.line - 1]
# 检查是否符合安全模式
for safe_pattern in self.safe_patterns[key]:
if re.search(safe_pattern, code_line):
return True
return False
class ConfidenceCalculator:
"""
置信度计算器
根据多维度因素计算发现的置信度
"""
@staticmethod
def calculate(
rule_id: str,
context: Context,
match_info: Dict
) -> float:
"""
计算置信度
Returns:
float: 置信度 0-1
"""
confidence = 0.8 # 基础置信度
# 1. 框架特定调整
if context.framework == "django":
if "query" in rule_id.lower():
# Django ORM 可能导致误报，降低置信度
confidence *= 0.7
elif "sql" in rule_id.lower():
# Django ORM 有 SQL 注入防护
confidence *= 0.8
# 2. 文件类型调整
if "test" in context.file_path.lower():
# 测试文件可能需要特殊处理
confidence *= 0.6
# 3. 代码特征调整
if match_info.get("has_type_hint"):
# 有类型注解的代码更可靠
confidence *= 1.1
if match_info.get("has_docstring"):
# 有文档字符串说明意图更清晰
confidence *= 1.1
# 4. 历史反馈调整（如果有反馈数据）
# false_positive_history = get_history(rule_id, context.file_path)
# if false_positive_history > 0.5:
# confidence *= 0.7
# 确保置信度在 0-1 范围内
return min(1.0, max(0.0, confidence))六、与 CI/CD 集成
6.1 GitHub Actions 配置
# .github/workflows/ai-code-review.yml
name: AI Code Review
on:
pull_request:
types: [opened, synchronize, reopened]
pull_request_review:
types: [submitted]
jobs:
ai-review:
name: AI Code Review
runs-on: ubuntu-latest
if: github.event_name == 'pull_request'
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
fetch-depth: 0 # 获取完整历史用于分析
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
- name: Run AI Code Review
env:
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
python -m ai_code_review.cli \
--repo ${{ github.repository }} \
--pr ${{ github.event.pull_request.number }} \
--sha ${{ github.sha }} \
--github-token ${{ secrets.GITHUB_TOKEN }}
- name: Upload artifacts
if: always()
uses: actions/upload-artifact@v4
with:
name: review-results
path: review_results/
retention-days: 7
- name: Post review comment
if: always()
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const results = JSON.parse(
fs.readFileSync('review_results/summary.json', 'utf8')
);
const comment = `
## 🤖 AI Code Review 结果
### 概览
- 📊 发现问题: ${results.total_issues} 个
- 🔴 严重: ${results.critical} | 🟠 高危: ${results.high} | 🟡 中: ${results.medium}
- ✅ 误报过滤: ${results.false_positives} 个
### 详情
${results.comments.map(c => `
**${c.severity.toUpperCase()}** - ${c.rule_name}
- 📁 ${c.file}:${c.line}
- 💬 ${c.message}
- 💡 建议: ${c.suggestion}
`).join('\n')}
---
*此评论由 AI Code Review 系统自动生成*
`;
github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: comment
});6.2 GitLab CI 配置
# .gitlab-ci.yml
stages:
- test
- review
ai_code_review:
stage: review
image: python:3.11-slim
only:
- merge_requests
variables:
ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
script: |
# 安装依赖
pip install -r requirements.txt
# 运行 AI Review
python -m ai_code_review.cli \
--mr-id ${CI_MERGE_REQUEST_IID} \
--gitlab-token ${GITLAB_TOKEN} \
--gitlab-url ${CI_API_V4_URL}
artifacts:
reports:
codequality: review_results/codequality.json
paths:
- review_results/
expire_in: 7 days
allow_failure: true # Review 结果不阻塞合并七、完整的 AI Code Review Pipeline 代码
7.1 核心 Pipeline 实现
# ai_code_review/pipeline.py
"""
AI Code Review Pipeline
完整的代码审查流程实现
"""
import asyncio
import hashlib
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict, Any
import json
from .code_parser import CodeParser
from .rule_engine import RuleEngine
from .llm_analyzer import LLMAnalyzer
from .false_positive_control import FalsePositiveController, Context
from .result_formatter import ResultFormatter
from .git_client import GitClient
from .notification import NotificationService
class ReviewStage(Enum):
"""审查阶段"""
PARSING = "parsing"
RULE_SCAN = "rule_scan"
LLM_ANALYSIS = "llm_analysis"
FALSE_POSITIVE_FILTER = "false_positive_filter"
AGGREGATION = "aggregation"
NOTIFICATION = "notification"
@dataclass
class ReviewConfig:
"""审查配置"""
# LLM 配置
llm_provider: str = "anthropic"
llm_model: str = "claude-sonnet-4-5"
llm_temperature: float = 0.3
# 审查范围
include_patterns: List[str] = field(default_factory=lambda: ["**/*.py"])
exclude_patterns: List[str] = field(default_factory=lambda: [
"**/test*.py",
"**/*_test.py",
"**/migrations/*.py",
"**/conftest.py"
])
# 规则配置
enabled_rules: List[str] = field(default_factory=list) # 空表示全部
disabled_rules: List[str] = field(default_factory=lambda: [
"function_length", # 函数长度规则可能过于敏感
"class_length"
])
# 误报控制
confidence_threshold: float = 0.5
# 通知配置
notify_on_critical: bool = True
notify_on_high: bool = False
notify_slack: bool = True
notify_pr_comment: bool = True
# 审查深度
max_files_to_analyze: int = 50
max_context_lines: int = 100
@dataclass
class ReviewRequest:
"""审查请求"""
repo_url: str
diff: Dict[str, Any] # Git diff 数据
base_sha: str
head_sha: str
pr_number: Optional[int] = None
pr_title: Optional[str] = None
author: Optional[str] = None
config: ReviewConfig = field(default_factory=ReviewConfig)
@dataclass
class ReviewResult:
"""审查结果"""
request: ReviewRequest
findings: List[Dict[str, Any]] = field(default_factory=list)
statistics: Dict[str, int] = field(default_factory=dict)
duration_ms: float = 0.0
timestamp: datetime = field(default_factory=datetime.now)
errors: List[str] = field(default_factory=list)
class ReviewPipeline:
"""
AI Code Review Pipeline
流程：
1. 代码解析 - 提取变更内容和上下文
2. 规则扫描 - 基于规则的安全/性能检测
3. LLM 分析 - 使用大模型进行深度分析
4. 误报过滤 - 多层策略降低误报率
5. 结果聚合 - 合并去重、优先级排序
6. 通知推送 - PR 评论、Slack 等
"""
def __init__(
self,
config: ReviewConfig,
git_client: GitClient,
llm_analyzer: LLMAnalyzer,
notification_service: NotificationService
):
self.config = config
self.git_client = git_client
self.llm_analyzer = llm_analyzer
self.notification_service = notification_service
# 初始化组件
self.parser = CodeParser()
self.rule_engine = RuleEngine(config)
self.fp_controller = FalsePositiveController()
self.formatter = ResultFormatter()
async def run(self, request: ReviewRequest) -> ReviewResult:
"""
执行完整的审查流程
Args:
request: 审查请求
Returns:
ReviewResult: 审查结果
"""
import time
start_time = time.time()
findings = []
errors = []
try:
# Stage 1: 代码解析
print(f"[{ReviewStage.PARSING.value}] 解析代码变更...")
parsed_changes = await self._parse_changes(request)
# Stage 2: 规则扫描
print(f"[{ReviewStage.RULE_SCAN.value}] 执行规则扫描...")
rule_findings = await self._rule_scan(parsed_changes)
findings.extend(rule_findings)
# Stage 3: LLM 分析
print(f"[{ReviewStage.LLM_ANALYSIS.value}] 执行 LLM 分析...")
llm_findings = await self._llm_analyze(parsed_changes)
findings.extend(llm_findings)
# Stage 4: 误报过滤
print(f"[{ReviewStage.FALSE_POSITIVE_FILTER.value}] 过滤误报...")
filtered_findings = await self._filter_false_positives(
findings, parsed_changes
)
# Stage 5: 结果聚合
print(f"[{ReviewStage.AGGREGATION.value}] 聚合结果...")
aggregated_findings = self._aggregate(findings)
# Stage 6: 通知推送
print(f"[{ReviewStage.NOTIFICATION.value}] 推送通知...")
await self._notify(aggregated_findings, request)
except Exception as e:
errors.append(str(e))
print(f"审查流程异常: {e}")
# 统计信息
statistics = self._calculate_statistics(aggregated_findings)
return ReviewResult(
request=request,
findings=aggregated_findings,
statistics=statistics,
duration_ms=(time.time() - start_time) * 1000,
errors=errors
)
async def _parse_changes(
self,
request: ReviewRequest
) -> List[Dict[str, Any]]:
"""解析代码变更"""
changes = []
for file_path, diff_info in request.diff.items():
# 检查是否在排除列表中
if self._should_exclude(file_path):
continue
# 解析变更
parsed = await self.parser.parse_diff(
file_path=file_path,
diff_content=diff_info.get("diff", ""),
old_content=diff_info.get("old_content", ""),
new_content=diff_info.get("new_content", "")
)
changes.append({
"file": file_path,
"language": self._detect_language(file_path),
"changes": parsed,
"context": diff_info
})
return changes
async def _rule_scan(
self,
changes: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""规则扫描"""
findings = []
for change in changes:
file_findings = await self.rule_engine.scan(
file_path=change["file"],
content=change["context"].get("new_content", ""),
language=change["language"]
)
for finding in file_findings:
finding["source"] = "rule"
finding["file"] = change["file"]
findings.append(finding)
return findings
async def _llm_analyze(
self,
changes: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""LLM 深度分析"""
findings = []
for change in changes:
# 限制分析的文件数量
if len(findings) >= self.config.max_files_to_analyze:
break
analysis = await self.llm_analyzer.analyze_code(
file_path=change["file"],
content=change["context"].get("new_content", ""),
diff=change["changes"],
language=change["language"]
)
for finding in analysis.get("findings", []):
finding["source"] = "llm"
finding["file"] = change["file"]
finding["confidence"] = analysis.get("confidence", 0.8)
findings.append(finding)
return findings
async def _filter_false_positives(
self,
findings: List[Dict[str, Any]],
changes: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""过滤误报"""
filtered = []
context_map = {c["file"]: c for c in changes}
for finding in findings:
context = Context(
file_path=finding["file"],
file_content=context_map.get(
finding["file"], {}
).get("context", {}).get("new_content", ""),
language=context_map.get(
finding["file"], {}
).get("language", "python")
)
# 转换为 ReviewFinding 格式
finding_obj = self._to_finding_object(finding)
# 检查是否应该豁免
if not self.fp_controller.should_exempt(finding_obj, context):
filtered.append(finding)
else:
finding["is_false_positive"] = True
finding["false_positive_reason"] = finding_obj.false_positive_reason
return filtered
def _aggregate(
self,
findings: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""聚合结果"""
# 按严重程度排序
severity_order = {
"critical": 0,
"high": 1,
"medium": 2,
"low": 3,
"info": 4
}
findings.sort(key=lambda f: (
severity_order.get(f.get("severity", "info"), 5),
f.get("file", ""),
f.get("line", 0)
))
return findings
async def _notify(
self,
findings: List[Dict[str, Any]],
request: ReviewRequest
):
"""发送通知"""
# 过滤需要通知的问题
critical_findings = [
f for f in findings
if f.get("severity") in ["critical", "high"]
and not f.get("is_false_positive", False)
]
if self.config.notify_pr_comment and request.pr_number:
await self.notification_service.post_pr_comment(
pr_number=request.pr_number,
findings=critical_findings,
statistics=self._calculate_statistics(findings)
)
if self.config.notify_slack and critical_findings:
await self.notification_service.send_slack(
findings=critical_findings,
repo=request.repo_url,
pr=request.pr_number
)
def _calculate_statistics(
self,
findings: List[Dict[str, Any]]
) -> Dict[str, int]:
"""计算统计信息"""
stats = {
"total": len(findings),
"critical": 0,
"high": 0,
"medium": 0,
"low": 0,
"info": 0,
"false_positives": 0,
"from_rule": 0,
"from_llm": 0
}
for f in findings:
severity = f.get("severity", "info")
if severity in stats:
stats[severity] += 1
if f.get("is_false_positive"):
stats["false_positives"] += 1
if f.get("source") == "rule":
stats["from_rule"] += 1
elif f.get("source") == "llm":
stats["from_llm"] += 1
return stats
def _should_exclude(self, file_path: str) -> bool:
"""检查是否应该排除"""
import fnmatch
for pattern in self.config.exclude_patterns:
if fnmatch.fnmatch(file_path, pattern):
return True
return False
def _detect_language(self, file_path: str) -> str:
"""检测编程语言"""
ext = file_path.split(".")[-1].lower()
lang_map = {
"py": "python",
"js": "javascript",
"ts": "typescript",
"java": "java",
"go": "go",
"rs": "rust",
"rb": "ruby"
}
return lang_map.get(ext, "unknown")
def _to_finding_object(
self,
finding: Dict[str, Any]
):
"""转换为 ReviewFinding 对象"""
from .false_positive_control import ReviewFinding, Severity
severity_map = {
"critical": Severity.CRITICAL,
"high": Severity.HIGH,
"medium": Severity.MEDIUM,
"low": Severity.LOW,
"info": Severity.INFO
}
return ReviewFinding(
rule_id=finding.get("rule_id", "unknown"),
severity=severity_map.get(
finding.get("severity", "info"),
Severity.INFO
),
file=finding.get("file", ""),
line=finding.get("line", 0),
message=finding.get("message", ""),
suggestion=finding.get("suggestion", ""),
confidence=finding.get("confidence", 1.0)
)7.2 LLM 分析器实现
# ai_code_review/llm_analyzer.py
"""
LLM 代码分析器
使用大模型进行深度代码分析
"""
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
LLM_ANALYSIS_PROMPT = """
## 任务
你是一个资深的代码审查专家。请分析以下代码变更，关注安全性、性能、代码质量和最佳实践。
## 代码信息
- 文件: {file_path}
- 编程语言: {language}
## 代码变更
```diff
{diff}完整代码（变更部分）
{code_snippet}分析要求
请从以下维度进行审查：
1. 安全性 (Security)
SQL 注入、命令注入、XSS 等安全漏洞
敏感信息泄露（密码、密钥、Token）
不安全的加密或哈希
认证/授权问题
2. 性能 (Performance)
N+1 查询问题
循环中的数据库/网络操作
大数据内存问题
不必要的重复计算
3. 代码质量 (Code Quality)
代码重复
函数/类过长
命名不规范
缺少错误处理
缺少日志记录
4. 最佳实践 (Best Practices)
使用推荐的 API/库
正确的资源管理
适当的抽象和封装
测试覆盖
输出格式
请按以下 JSON 格式输出分析结果：
{{
"findings": [
{{
"severity": "critical|high|medium|low|info",
"rule_id": "security_sql_injection",
"line": <行号>,
"message": "<问题描述>",
"suggestion": "<修复建议>",
"confidence": <0.0-1.0置信度>,
"category": "security|performance|quality|practice"
}}
],
"summary": "<总体评价>",
"strengths": ["<优点列表>"],
"recommendations": ["<改进建议列表>"]
}}注意事项
只报告你确信的问题，不要猜测
考虑代码的实际用途和上下文
提供具体可行的修复建议
如果代码没有问题，返回空的 findings 数组
"""
class LLMAnalyzer:
"""
LLM 代码分析器
"""
def __init__(
self,
llm_client, # 支持 LangChain 的 LLM 客户端
model: str = "claude-sonnet-4-5",
temperature: float = 0.3
):
self.llm_client = llm_client
self.model = model
self.temperature = temperature
async def analyze_code(
self,
file_path: str,
content: str,
diff: Dict[str, Any],
language: str
) -> Dict[str, Any]:
"""
分析代码变更
Args:
file_path: 文件路径
content: 完整文件内容
diff: 代码变更信息
language: 编程语言
Returns:
dict: 分析结果
"""
# 构建提示
prompt = self._build_prompt(
file_path=file_path,
content=content,
diff=diff,
language=language
)
try:
# 调用 LLM
response = await self._call_llm(prompt)
# 解析响应
result = self._parse_response(response)
return result
except Exception as e:
return {
"findings": [],
"summary": f"分析失败: {str(e)}",
"error": str(e)
}
def _build_prompt(
self,
file_path: str,
content: str,
diff: Dict[str, Any],
language: str
) -> str:
"""构建分析提示"""
# 提取代码片段（变更部分及上下文）
code_snippet = self._extract_code_snippet(content, diff)
# 格式化 diff
diff_str = self._format_diff(diff)
return LLM_ANALYSIS_PROMPT.format(
file_path=file_path,
language=language,
diff=diff_str,
code_snippet=code_snippet
)
async def _call_llm(self, prompt: str) -> str:
"""调用 LLM"""
# 使用 LangChain
from langchain.schema import HumanMessage
response = await self.llm_client.agenerate([
[HumanMessage(content=prompt)]
])
return response.generations[0][0].text
def _parse_response(self, response: str) -> Dict[str, Any]:
"""解析 LLM 响应"""
import json
import re
# 尝试提取 JSON
json_match = re.search(
r'```json\s*(.*?)\s*```',
response,
re.DOTALL
)
if json_match:
try:
return json.loads(json_match.group(1))
except json.JSONDecodeError:
pass
# 尝试直接解析
try:
return json.loads(response)
except json.JSONDecodeError:
# 如果无法解析 JSON，返回文本结果
return {
"findings": [],
"summary": response,
"raw_response": response
}
def _extract_code_snippet(
self,
content: str,
diff: Dict[str, Any]
) -> str:
"""提取代码片段"""
# 获取变更的行号范围
changed_lines = set()
for change in diff.get("changes", []):
changed_lines.add(change.get("old_line", 0))
changed_lines.add(change.get("new_line", 0))
if not changed_lines:
return content[:2000] # 默认返回前2000字符
# 获取变更及其上下文（前后各5行）
lines = content.split('\n')
context_lines = set()
for line_num in changed_lines:
for i in range(max(0, line_num - 6), min(len(lines), line_num + 6)):
context_lines.add(i)
snippet_lines = [
f"{i+1}: {lines[i]}"
for i in sorted(context_lines)
]
return '\n'.join(snippet_lines)
def _format_diff(self, diff: Dict[str, Any]) -> str:
"""格式化 diff"""
formatted = []
for change in diff.get("changes", []):
change_type = change.get("type", "modify")
old_line = change.get("old_line", 0)
new_line = change.get("new_line", 0)
content = change.get("content", "")
if change_type == "add":
formatted.append(f"+{new_line}: {content}")
elif change_type == "delete":
formatted.append(f"-{old_line}: {content}")
else:
formatted.append(f" {new_line}: {content}")
return '\n'.join(formatted)
---
## 八、人工 + AI 的混合 Review 最佳实践
### 8.1 分工策略
┌─────────────────────────────────────────────────────────────────────┐
│ 人工 + AI 混合 Review │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 审查分工矩阵 │ │
│ ├─────────────────────────────────────────────────────────────┤ │
│ │ │ │
│ │ 🤖 AI Review（自动执行） │ │
│ │ ├── 安全漏洞扫描 │ │
│ │ ├── 性能问题检测 │ │
│ │ ├── 代码规范检查 │ │
│ │ ├── 错误处理审查 │ │
│ │ └── 类型注解检查 │ │
│ │ │ │
│ │ 👤 人工 Review（必须执行） │ │
│ │ ├── 业务逻辑正确性 │ │
│ │ ├── 架构设计评估 │ │
│ │ ├── PR 描述与实现一致性 │ │
│ │ ├── 测试覆盖评估 │ │
│ │ └── 团队规范遵守 │ │
│ │ │ │
│ │ 👥 团队 Review（按需） │ │
│ │ ├── 复杂重构方案评审 │ │
│ │ ├── 新技术引入评审 │ │
│ │ └── 架构调整评审 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
### 8.2 Review 流程
```yaml
# review_process.yml
# Code Review 流程定义
review_process:
name: "AI-Assisted Code Review"
stages:
- name: "自动检查"
type: "automated"
assignee: "ai-reviewer"
tools:
- name: "AI Code Review"
config: "ai_review_config.yml"
- name: "静态分析"
tools:
- pylint
- ruff
- mypy
deadline: "PR 创建后 5 分钟内"
blocking: false # 不阻塞，可选
- name: "作者自检"
type: "author"
assignee: "PR 作者"
checklist:
- "代码符合 PR 描述"
- "添加了必要的测试"
- "更新了相关文档"
- "所有 CI 检查通过"
deadline: "PR 创建后 2 小时内"
blocking: true
- name: "同行审查"
type: "peer"
assignee: "至少 1 名团队成员"
checklist:
- "业务逻辑正确"
- "架构设计合理"
- "代码可维护"
- "测试充分"
min_reviewers: 1
deadline: "1 个工作日内"
blocking: true
- name: "技术 Lead 审批"
type: "lead"
assignee: "技术 Lead"
triggers:
- "文件变更 > 10 个"
- "涉及架构调整"
- "新增依赖"
deadline: "2 个工作日内"
blocking: true
- name: "合并"
type: "merge"
preconditions:
- "所有 blocking 检查通过"
- "至少 1 个 Approval"
- "无 open 讨论"
assignee: "PR 作者或 Reviewer"8.3 Review 报告模板
## Code Review 报告
### 基本信息
- **PR**: #[编号] [标题]
- **作者**: [姓名]
- **审查时间**: [日期]
- **审查方式**: AI + 人工
---
### AI Review 结果
#### 发现问题统计
| 严重程度 | 数量 | 状态 |
|---------|------|------|
| 🔴 Critical | 0 | 已修复 |
| 🟠 High | 1 | 已修复 |
| 🟡 Medium | 3 | 已修复 |
| 🟢 Low | 5 | 可选修复 |
| 🔵 Info | 8 | 已确认 |
#### 关键发现
- [问题描述及修复状态]
#### 误报过滤
- 过滤误报: 2 个
- 过滤原因: [豁免注释、安全模式]
---
### 人工 Review 结果
#### 审查清单
- [✅/❌] 业务逻辑正确性
- [✅/❌] 架构设计合理性
- [✅/❌] 代码可维护性
- [✅/❌] 测试覆盖充分
- [✅/❌] 文档更新完整
#### 讨论事项
- [讨论点及结论]
---
### 总体评价
**AI Review**: ✅ 通过
**人工 Review**: ✅ 通过
**最终结论**: ✅ 可以合并
---
### 后续跟进
- [ ] [跟进事项]九、总结
AI Code Review 是提升工程效能的有力工具，但需要理性看待其能力边界。
核心价值：
🚀 提升效率：自动处理大量重复性审查工作
🔍 覆盖全面：规则 + LLM 双引擎，检测面广
📊 标准一致：消除人工审查的主观差异
⚡ 快速反馈：PR 创建即可获得 AI 审查结果
最佳实践：
分层策略：AI 处理规则类问题，人工聚焦业务和架构
误报控制：多层过滤机制，保持 Review 体验
持续优化：根据反馈调整规则和阈值
透明沟通：让团队理解 AI Review 的定位和局限
关键提醒：
AI Review 是辅助工具，不能完全替代人工审查
业务逻辑和架构设计必须人工把关
持续优化 Prompt 和规则，降低误报率
关注开发者体验，避免噪声干扰
推荐阅读
LangGraph实战：构建复杂AI工作流
RAG开发者全栈指南：企业级知识库搭建
关注 PySuper，获取更多 AI + 工程实践干货！
评论区