AI 研发团队的代码审查流程与质量门禁设计

AI 研发团队的代码审查流程与质量门禁设计

AI 项目的代码审查有它的特殊性：除了常规的代码质量，你还需要关注 Prompt 版本管理、模型行为的可测试性、推理成本的合理性。本文分享一套经过实践验证的 AI 研发团队代码审查流程，以及配套的质量门禁（Quality Gate）设计方案。

一、AI 项目 Code Review 的特殊挑战

传统软件的 Code Review 主要关注逻辑、性能、安全。AI 项目在此基础上额外增加了：

二、代码审查清单：AI 项目专项检查项

在常规 Code Review 清单基础上，增加以下 AI 专项：

2.1 LLM 调用层检查

## LLM 调用 Review Checklist

### Prompt 质量
– [ ] Prompt 是否有版本号/注释说明上次修改原因？
– [ ] System Prompt 是否有足够的边界约束（拒绝不合理请求）？
– [ ] 是否避免了模糊指令（"写一段代码" vs "写一个 Python 函数，接受…返回…"）？
– [ ] Few-shot 示例质量是否经过验证？

### 错误处理
– [ ] 是否处理了 API Rate Limit 错误（429）并实现了指数退避？
– [ ] 是否处理了模型输出格式错误（非 JSON、截断输出等）？
– [ ] 是否处理了内容过滤触发（Moderation）的情况？
– [ ] 网络超时是否有合理设置（推荐 30-60s）？

### 成本控制
– [ ] max_tokens 是否设置了合理上限？
– [ ] 是否有 Token 消耗的监控埋点？
– [ ] 是否避免了不必要的重复调用（能缓存的结果有没有缓存）？
– [ ] 模型选择是否合理（简单任务用便宜模型）？

### 安全
– [ ] 用户输入是否经过清洗，防止 Prompt 注入？
– [ ] 模型输出是否经过验证，才用于后续业务逻辑？
– [ ] 是否避免了把敏感信息（密钥、PII）送入 LLM？

2.2 Agent / 工具调用检查

## Agent 代码 Review Checklist

### 工具设计
– [ ] 工具描述是否准确（模型看描述就能正确使用工具）？
– [ ] 工具参数类型是否有验证（Pydantic 或 JSONSchema）？
– [ ] 工具执行是否有超时保护？
– [ ] 工具失败是否有降级策略？

### Agent 循环
– [ ] 是否有最大迭代次数限制（防止无限循环）？
– [ ] 是否有总时长/成本上限？
– [ ] 关键操作是否有人工确认步骤？

### 状态管理
– [ ] Agent 状态是否持久化？（服务重启后能恢复）
– [ ] 多 Agent 共享状态是否有并发保护？

三、质量门禁（Quality Gate）设计

质量门禁是 CI/CD 流程中的自动化检查节点，代码不过门禁不允许合并。

3.1 整体架构

PR 提交
↓
[Gate 1: 基础代码质量]
– 代码格式 (Black, isort)
– 静态分析 (Pylint, mypy)
– 安全扫描 (Bandit)

↓ Pass

[Gate 2: AI 专项检查]
– Prompt 版本追踪
– Token 成本估算
– 模型调用安全检查

↓ Pass

[Gate 3: 测试覆盖]
– 单元测试 (pytest)
– AI 行为测试 (自定义评估)
– 覆盖率 >80%

↓ Pass

[Gate 4: 性能回归]
– P95 延迟对比
– Token 消耗对比（与 main 分支比较）

↓ Pass

允许合并

3.2 Gate 1：基础代码质量（GitHub Actions 实现）

# .github/workflows/quality-gate.yml
name: Quality Gate

on: [pull_request]

jobs:
code-quality:
runs-on: ubuntu–latest
steps:
– uses: actions/checkout@v4

– name: Setup Python
uses: actions/setup–python@v5
with:
python-version: '3.12'

– name: Install dependencies
run: |
pip install black isort pylint mypy bandit pytest pytest-cov
pip install -r requirements.txt

# 代码格式检查
– name: Black format check
run: black ––check ––line–length 120 src/

– name: Import sort check
run: isort ––check–only ––profile black src/

# 类型检查
– name: Type check
run: mypy src/ ––ignore–missing–imports ––strict

# 安全扫描
– name: Security scan
run: bandit –r src/ –ll # 只报告中级以上问题

# 基础测试
– name: Run tests
run: pytest tests/unit/ –v ––cov=src ––cov–report=xml
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

# 覆盖率门禁
– name: Coverage check
run: |
coverage report –fail-under=80

3.3 Gate 2：AI 专项检查脚本

# scripts/ai_quality_check.py
"""
AI 项目专项质量检查脚本
在 CI 中运行，检查 LLM 相关代码的质量问题
"""

import ast
import sys
import re
from pathlib import Path
from dataclasses import dataclass, field
from typing import List

@dataclass
class AIQualityIssue:
file: str
line: int
severity: str # "error" | "warning" | "info"
message: str

class AICodeChecker:

def __init__(self, source_dir: str):
self.source_dir = Path(source_dir)
self.issues: List[AIQualityIssue] = []

def check_all(self) –> List[AIQualityIssue]:
"""运行所有检查"""
for py_file in self.source_dir.rglob("*.py"):
self._check_file(py_file)
return self.issues

def _check_file(self, file_path: Path):
"""检查单个文件"""
try:
source = file_path.read_text(encoding="utf-8")
tree = ast.parse(source)
except Exception:
return

self._check_llm_calls(source, file_path, tree)
self._check_prompt_versioning(source, file_path)
self._check_error_handling(source, file_path, tree)
self._check_sensitive_data(source, file_path)

def _check_llm_calls(self, source: str, file_path: Path, tree):
"""检查 LLM 调用质量"""
lines = source.split("\\n")

for i, line in enumerate(lines, 1):
# 检查是否设置了 max_tokens
if "client.chat.completions.create" in line or \\
"client.messages.create" in line:
# 向下找 max_tokens
context = "\\n".join(lines[i:i+15])
if "max_tokens" not in context:
self.issues.append(AIQualityIssue(
file=str(file_path),
line=i,
severity="warning",
message="LLM 调用未设置 max_tokens，可能导致不必要的 Token 消耗"
))

# 检查超时设置
if "openai.OpenAI(" in line or "anthropic.Anthropic(" in line:
if "timeout" not in line:
self.issues.append(AIQualityIssue(
file=str(file_path),
line=i,
severity="warning",
message="LLM 客户端初始化未设置超时，建议设置 timeout=60"
))

def _check_prompt_versioning(self, source: str, file_path: Path):
"""检查 Prompt 是否有版本/变更注释"""
lines = source.split("\\n")

for i, line in enumerate(lines, 1):
# 找到长 Prompt 字符串
if '"""' in line or "'''" in line:
# 检查前后是否有 PROMPT_VERSION 或 # v 注释
context = "\\n".join(lines[max(0, i–3):i+1])
if ("system_message" in context.lower() or
"system_prompt" in context.lower() or
"SYSTEM" in context) and \\
"# v" not in context and \\
"PROMPT_VERSION" not in context and \\
"# updated" not in context.lower():
self.issues.append(AIQualityIssue(
file=str(file_path),
line=i,
severity="info",
message="System Prompt 建议添加版本注释（# v1.2 – 2024-xx）方便追踪变更"
))
break

def _check_sensitive_data(self, source: str, file_path: Path):
"""检查是否有敏感数据被送入 LLM"""
sensitive_patterns = [
(r'password\\s*=', "password 变量可能被送入 LLM"),
(r'api_key\\s*=', "API Key 可能被送入 LLM"),
(r'secret\\s*=', "secret 变量可能被送入 LLM"),
]

lines = source.split("\\n")
for i, line in enumerate(lines, 1):
for pattern, msg in sensitive_patterns:
if re.search(pattern, line, re.IGNORECASE):
# 检查附近是否有 LLM 调用
nearby = "\\n".join(lines[max(0, i–5):min(len(lines), i+5)])
if "messages" in nearby and ("invoke" in nearby or "create" in nearby):
self.issues.append(AIQualityIssue(
file=str(file_path),
line=i,
severity="error",
message=f"⚠️ 潜在安全问题：{msg}"
))

def _check_error_handling(self, source: str, file_path: Path, tree):
"""检查 LLM 调用是否有错误处理"""
lines = source.split("\\n")
for i, line in enumerate(lines, 1):
if ".invoke(" in line or ".create(" in line:
# 检查是否在 try-except 块中
# 简化检查：看前10行是否有 try:
context_before = "\\n".join(lines[max(0, i–10):i])
if "try:" not in context_before and "try :" not in context_before:
self.issues.append(AIQualityIssue(
file=str(file_path),
line=i,
severity="warning",
message="LLM 调用缺少 try-except 错误处理，建议捕获 API 异常"
))

def main():
checker = AICodeChecker("src/")
issues = checker.check_all()

errors = [i for i in issues if i.severity == "error"]
warnings = [i for i in issues if i.severity == "warning"]

for issue in issues:
icon = {"error": "❌", "warning": "⚠️", "info": "ℹ️"}[issue.severity]
print(f"{icon} {issue.file}:{issue.line} – {issue.message}")

print(f"\\n检查完成：{len(errors)} 个错误，{len(warnings)} 个警告")

if errors:
print("❌ 存在严重问题，质量门禁不通过")
sys.exit(1)

print("✅ 质量门禁通过")

if __name__ == "__main__":
main()

3.4 Gate 3：AI 行为测试框架

# tests/ai_behavior/test_llm_responses.py
"""
AI 行为测试：测试 LLM 调用的输出质量
使用 LLM-as-Judge 方式评估，而非固定断言
"""

import pytest
import json
from unittest.mock import patch, MagicMock
from src.agents.code_reviewer import CodeReviewAgent

class TestCodeReviewAgent:

@pytest.fixture
def agent(self):
return CodeReviewAgent(model="claude-3-5-sonnet-20241022")

def test_detects_sql_injection(self, agent):
"""测试：SQL 注入代码必须被检测出来"""
vulnerable_code = """
def get_user(user_id):
query = f"SELECT * FROM users WHERE id = {user_id}"
return db.execute(query)
"""
result = agent.review(vulnerable_code, language="python")

# 断言：结果中必须包含 SQL 注入相关信息
assert result["score"] < 60, "有 SQL 注入漏洞的代码评分不应超过 60"

security_issues_text = " ".join(result.get("security_issues", [])).lower()
assert any(keyword in security_issues_text
for keyword in ["sql", "injection", "注入", "sql注入"]), \\
"必须检测出 SQL 注入漏洞"

def test_format_consistency(self, agent):
"""测试：输出格式必须符合规范"""
simple_code = "def add(a, b): return a + b"
result = agent.review(simple_code, language="python")

# 验证返回结构
assert isinstance(result, dict), "返回值必须是字典"
assert "score" in result, "必须包含 score 字段"
assert "security_issues" in result, "必须包含 security_issues 字段"
assert isinstance(result["score"], int), "score 必须是整数"
assert 0 <= result["score"] <= 100, "score 必须在 0-100 之间"

def test_handles_api_error_gracefully(self, agent):
"""测试：API 错误时有优雅降级"""
with patch.object(agent._client, "messages") as mock_client:
mock_client.create.side_effect = Exception("API Error: 429 Too Many Requests")

# 不应该抛出异常，应该返回错误信息
result = agent.review("def foo(): pass", language="python")
assert "error" in result or result.get("score") is not None, \\
"API 错误时应该返回错误信息而不是抛出异常"

@pytest.mark.parametrize("good_code,min_score", [
("def add(a: int, b: int) -> int:\\n \\"\\"\\"Add two numbers.\\"\\"\\"\\n return a + b", 80),
("import hashlib\\n\\ndef hash_password(pwd: str) -> str:\\n return hashlib.sha256(pwd.encode()).hexdigest()", 75),
])
def test_good_code_gets_high_score(self, agent, good_code, min_score):
"""测试：质量良好的代码应该得到高分"""
result = agent.review(good_code, language="python")
assert result["score"] >= min_score, \\
f"良好代码评分 {result['score']} 低于预期 {min_score}"

四、Prompt 版本管理方案

4.1 用代码管理 Prompt（推荐）

# src/prompts/code_review_prompts.py
"""
代码审查相关 Prompt 配置
所有 Prompt 修改必须更新版本号并注明修改原因
"""

# v1.3 – 2025-03-15 – 增加对 async/await 模式的检查
CODE_REVIEW_SYSTEM_PROMPT = """你是一位资深代码审查专家，专注于代码质量、安全漏洞和性能优化。

审查重点：
1. 安全漏洞（SQL 注入、XSS、命令注入、敏感信息泄露）
2. 代码质量（可读性、可维护性、单一职责原则）
3. 性能问题（不必要的循环、N+1 查询、内存泄漏）
4. Python 最佳实践（类型注解、异常处理、async/await 规范使用）

返回格式（严格 JSON）：
{
"score": <0-100的整数>,
"security_issues": ["问题1（严重度：高/中/低）"],
"suggestions": ["建议1", "建议2"]
}

评分标准：
– 90-100：优秀，无明显问题
– 70-89：良好，有少量改进空间
– 50-69：一般，存在明显问题
– 0-49：不合格，存在严重问题（如安全漏洞）"""

# v1.1 – 2025-02-01 – 初始版本
CODE_REVIEW_USER_TEMPLATE = """请审查以下 {language} 代码：

```{language}
{code}

只返回 JSON，不要其他解释。“”"

PROMPT_VERSIONS = {
“code_review_system”: “1.3”,
“code_review_user”: “1.1”
}

### 4.2 Prompt 变更追踪钩子

```bash
# .git/hooks/pre-commit
#!/bin/bash
# 检查 Prompt 文件变更是否更新了版本号

changed_files=$(git diff –cached –name-only | grep "prompts/")

for file in $changed_files; do
if git diff –cached "$file" | grep -q "SYSTEM_PROMPT\\|USER_TEMPLATE"; then
if ! git diff –cached "$file" | grep -q "# v"; then
echo "⚠️ 警告：$file 中的 Prompt 被修改，但未更新版本注释"
echo "请在修改的 Prompt 上方添加版本注释，例如："
echo "# v1.x – YYYY-MM-DD – 修改原因"
exit 1
fi
fi
done

exit 0

五、Token 成本监控与告警

# src/monitoring/token_monitor.py
"""
Token 消耗监控中间件
"""
import functools
import time
from dataclasses import dataclass
from typing import Optional
import logging

logger = logging.getLogger(__name__)

# 价格配置（美元/1M Token）
TOKEN_PRICES = {
"claude-3-5-sonnet-20241022": {"input": 3.0, "output": 15.0},
"claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
"gpt-4o": {"input": 2.5, "output": 10.0},
"gpt-4o-mini": {"input": 0.15, "output": 0.6},
}

# 告警阈值（每次调用）
COST_ALERT_THRESHOLD_USD = 0.5 # 单次超 $0.5 告警

@dataclass
class TokenUsage:
model: str
input_tokens: int
output_tokens: int
latency_ms: float
cost_usd: float

def monitor_llm_call(func):
"""LLM 调用监控装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
latency_ms = (time.time() – start_time) * 1000

# 提取 Token 使用信息
model = kwargs.get("model", "unknown")
usage = getattr(result, "usage", None)

if usage:
prices = TOKEN_PRICES.get(model, {"input": 0, "output": 0})
cost_usd = (
usage.input_tokens / 1_000_000 * prices["input"] +
usage.output_tokens / 1_000_000 * prices["output"]
)

token_usage = TokenUsage(
model=model,
input_tokens=usage.input_tokens,
output_tokens=usage.output_tokens,
latency_ms=latency_ms,
cost_usd=cost_usd
)

# 记录到监控系统
logger.info(f"[TOKEN_USAGE] model={model} "
f"input={usage.input_tokens} "
f"output={usage.output_tokens} "
f"cost=${cost_usd:.4f} "
f"latency={latency_ms:.0f}ms")

# 成本告警
if cost_usd > COST_ALERT_THRESHOLD_USD:
logger.warning(f"⚠️ 单次 LLM 调用成本过高: ${cost_usd:.4f}")

return result
return wrapper

六、总结

AI 研发团队的 Code Review 和质量门禁，核心原则是：

参考文献