团队代码质量参差不齐，人工 Code Review 效率低且不一致？本文手把手教你构建一个基于 AI 的自动化 Code Review 系统，自动分析代码变更、识别问题、分级输出，并与 GitLab/GitHub 无缝集成。
一、项目背景与问题定义
1.1 团队痛点
在软件工程实践中，Code Review 是保障代码质量的重要环节，但随着团队规模扩大和问题积累，我们遇到了以下挑战：
1.2 解决方案
构建一个 AI 辅助 Code Review 系统，实现：
自动化分析：PR 创建/更新时自动触发 Review
多维度检查：安全漏洞、性能问题、代码规范、最佳实践
分级输出：Critical / Warning / Info 三级分类
知识积累：历史 Review 数据形成团队知识库
无缝集成：与 GitLab/GitHub 原生集成
二、系统架构
2.1 整体架构图
┌─────────────────────────────────────────────────────────────────────────────┐
│ AI Code Review 系统架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ GitLab / GitHub │ │
│ │ (Pull Request / Merge Request) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ │ Webhook │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ API 网关层 │ │
│ │ (FastAPI + Nginx) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 任务队列层 │ │
│ │ (Celery + Redis) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────┴──────────┐ │
│ ↓ ↓ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Worker 节点 1 │ │ Worker 节点 N │ │
│ │ (Diff 解析) │ │ (AI 分析) │ │
│ └──────────────────┘ └──────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 分析引擎层 │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ 安全扫描 │ │ 性能分析 │ │ 规范检查 │ │ AI 理解 │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ LLM 服务层 │ │
│ │ (GPT-4 / Claude / DeepSeek + vLLM 推理) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 知识库层 │ │
│ │ (向量数据库 + 历史 Review 数据) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 评论输出层 │ │
│ │ (GitLab/GitHub PR Comment API) │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘2.2 数据流图
┌─────────────────────────────────────────────────────────────────────────────┐
│ 数据流处理流程 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ PR 创建/更新 │
│ │ │
│ ↓ │
│ ┌───────────┐ 获取 Diff ┌───────────┐ 解析 Diff ┌───────────┐│
│ │ Webhook │ ──────────────→ │ Diff 获取 │ ─────────────→ │ 代码解析器 ││
│ └───────────┘ └───────────┘ └───────────┘│
│ │ │
│ ↓ │
│ ┌───────────┐ 存储上下文 ┌───────────┐ 召回知识 ┌───────────┐│
│ │ 知识库 │ ←───────────────│ 上下文组装 │ ←─────────────│ 向量检索 ││
│ └───────────┘ └───────────┘ └───────────┘│
│ │ │
│ ↓ │
│ ┌───────────┐ 分析结果 ┌───────────┐ 生成评论 ┌───────────┐│
│ │ 评论存储 │ ←───────────────│ 结果聚合 │ ←─────────────│ LLM 分析 ││
│ └───────────┘ └───────────┘ └───────────┘│
│ │ │
│ ↓ │
│ ┌───────────┐ 发布评论 ┌───────────┐ 格式化 ┌───────────┐ │
│ │ GitLab/ │ ←──────────────│ 评论生成 │ ←───────────│ 结果分级 │ │
│ │ GitHub │ └───────────┘ └───────────┘ │
│ └───────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘三、技术方案详解
3.1 Git Hook 触发 + Webhook 接收
# .gitlab-ci.yml - GitLab CI 配置
stages:
- review
ai_code_review:
stage: review
image: python:3.11-slim
services:
- redis:7-alpine
variables:
REDIS_URL: redis://redis:6379/0
GITLAB_TOKEN: $GITLAB_TOKEN
OPENAI_API_KEY: $OPENAI_API_KEY
script:
- pip install -r requirements.txt
- python -m app.main
only:
- merge_requests
trigger:
project: "your-project"
artifacts:
reports:
junit: reports/result.xml# app/webhooks/gitlab.py
"""
GitLab Webhook 处理
"""
import hmac
import hashlib
from typing import Optional, Dict, Any
from fastapi import APIRouter, Header, HTTPException, Request
from pydantic import BaseModel
from datetime import datetime
router = APIRouter(prefix="/webhooks", tags=["webhooks"])
class GitLabWebhookPayload(BaseModel):
"""GitLab Webhook 事件载荷"""
object_kind: str
event_type: str
project: Dict[str, Any]
object_attributes: Dict[str, Any]
user: Dict[str, Any]
changes: Optional[Dict[str, Any]] = None
class MergeRequestEvent(BaseModel):
"""Merge Request 事件"""
id: int
iid: int
title: str
source_branch: str
target_branch: str
state: str
url: str
diff_refs: Dict[str, str]
# Diff 信息
changes_count: Optional[str] = None
def verify_gitlab_webhook(
payload: bytes,
token: str,
signature: str
) -> bool:
"""
验证 GitLab Webhook 签名
GitLab 使用 X-Gitlab-Token 或 HMAC 签名
"""
# 方式1：简单 Token 验证
expected_token = token.encode()
return hmac.compare_digest(payload, expected_token)
# 方式2：HMAC 签名验证（如果配置了 secret token）
# secret = settings.GITLAB_WEBHOOK_SECRET.encode()
# expected = hmac.new(secret, payload, hashlib.sha256).hexdigest()
# return hmac.compare_digest(f"sha256={expected}", signature)
@router.post("/gitlab")
async def handle_gitlab_webhook(
request: Request,
x_gitlab_event: Optional[str] = Header(None),
x_gitlab_token: Optional[str] = Header(None)
):
"""
处理 GitLab Webhook
支持的事件：
- Merge Request 事件
- Push 事件
- Note (评论) 事件
"""
# 获取原始载荷
payload = await request.body()
# 验证签名（生产环境务必启用）
# if not verify_gitlab_webhook(payload, settings.GITLAB_WEBHOOK_SECRET, x_gitlab_token or ""):
# raise HTTPException(status_code=401, detail="Invalid webhook signature")
# 解析载荷
data = await request.json()
# 处理不同事件类型
event_type = data.get("object_kind") or data.get("event_type")
if event_type == "merge_request":
return await handle_merge_request(data)
elif event_type == "push":
return await handle_push_event(data)
elif event_type == "note":
return await handle_comment_event(data)
else:
return {"status": "ignored", "reason": f"Unsupported event type: {event_type}"}
async def handle_merge_request(data: Dict[str, Any]) -> Dict[str, Any]:
"""处理 Merge Request 事件"""
obj_attrs = data.get("object_attributes", {})
# 只处理打开或更新的 MR
if obj_attrs.get("state") not in ["opened", "update"]:
return {"status": "ignored", "reason": "MR not in opened or update state"}
mr_event = MergeRequestEvent(
id=obj_attrs.get("id"),
iid=obj_attrs.get("iid"),
title=obj_attrs.get("title", ""),
source_branch=obj_attrs.get("source_branch"),
target_branch=obj_attrs.get("target_branch"),
state=obj_attrs.get("state"),
url=obj_attrs.get("url"),
diff_refs=obj_attrs.get("diff_refs", {}),
changes_count=obj_attrs.get("changes_count")
)
# 触发异步 Review 任务
from app.tasks.review import analyze_merge_request
task = analyze_merge_request.delay(mr_event.dict())
return {
"status": "accepted",
"task_id": task.id,
"message": f"Review task queued for MR !{mr_event.iid}"
}
async def handle_push_event(data: Dict[str, Any]) -> Dict[str, Any]:
"""处理 Push 事件"""
# Push 事件通常不需要自动 Review
return {"status": "ignored", "reason": "Push events are not reviewed automatically"}
async def handle_comment_event(data: Dict[str, Any]) -> Dict[str, Any]:
"""处理评论事件"""
# 可以用于 /review 命令触发
return {"status": "ignored", "reason": "Comment events handled separately"}3.2 Diff 解析模块
# app/services/diff_parser.py
"""
代码 Diff 解析器
"""
import re
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from enum import Enum
class ChangeType(Enum):
"""变更类型"""
ADDED = "added" # 新增
DELETED = "deleted" # 删除
MODIFIED = "modified" # 修改
RENAMED = "renamed" # 重命名
@dataclass
class FileChange:
"""文件变更"""
path: str
change_type: ChangeType
# Diff 信息
diff: str
old_path: Optional[str] = None # 重命名时的旧路径
# 变更统计
additions: int = 0
deletions: int = 0
chunks: List["DiffChunk"] = field(default_factory=list)
# 解析后的代码
old_content: Optional[str] = None
new_content: Optional[str] = None
# 语言信息
language: Optional[str] = None
# 元数据
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class DiffChunk:
"""Diff 块（hunk）"""
old_start: int
old_lines: int
new_start: int
new_lines: int
lines: List["DiffLine"] = field(default_factory=list)
@dataclass
class DiffLine:
"""Diff 行"""
line_type: str # "+", "-", " ", "@@"
content: str
old_line_num: Optional[int] = None
new_line_num: Optional[int] = None
# 语义信息
is_comment_line: bool = False # 是否是注释
is_blank_line: bool = False # 是否是空行
@dataclass
class DiffResult:
"""Diff 解析结果"""
files: List[FileChange] = field(default_factory=list)
total_additions: int = 0
total_deletions: int = 0
total_files: int = 0
# 变更摘要
summary: str = ""
class DiffParser:
"""
Git Diff 解析器
支持：
- unified diff 格式
- 提取代码上下文
- 统计变更信息
"""
# 文件语言映射
EXTENSION_TO_LANGUAGE = {
".py": "python",
".js": "javascript",
".ts": "typescript",
".java": "java",
".go": "go",
".rs": "rust",
".cpp": "cpp",
".c": "c",
".h": "c",
".hpp": "cpp",
".cs": "csharp",
".rb": "ruby",
".php": "php",
".swift": "swift",
".kt": "kotlin",
".scala": "scala",
".vue": "vue",
".jsx": "jsx",
".tsx": "tsx",
}
def __init__(self):
self.files: List[FileChange] = []
def parse(self, diff_text: str) -> DiffResult:
"""
解析 Diff 文本
Args:
diff_text: Git diff 输出
Returns:
DiffResult: 解析结果
"""
# 按文件分割
file_diffs = self._split_by_file(diff_text)
for file_diff in file_diffs:
file_change = self._parse_file_diff(file_diff)
if file_change:
self.files.append(file_change)
# 计算统计
total_additions = sum(f.additions for f in self.files)
total_deletions = sum(f.deletions for f in self.files)
return DiffResult(
files=self.files,
total_additions=total_additions,
total_deletions=total_deletions,
total_files=len(self.files),
summary=self._generate_summary()
)
def _split_by_file(self, diff_text: str) -> List[str]:
"""按文件分割 Diff"""
# diff --git a/file1 b/file1 格式
pattern = r"diff --git a/.+? b/.+?(?=\ndiff --git|\Z)"
matches = re.finditer(pattern, diff_text, re.DOTALL)
return [m.group() for m in matches]
def _parse_file_diff(self, file_diff: str) -> Optional[FileChange]:
"""解析单个文件的 Diff"""
if not file_diff.strip():
return None
lines = file_diff.split("\n")
# 解析文件头
header_match = re.match(r"diff --git a/(.+?) b/(.+)", lines[0])
if not header_match:
return None
old_path = header_match.group(1)
new_path = header_match.group(2)
# 确定变更类型
change_type = self._detect_change_type(lines)
# 解析统计信息
additions, deletions = self._parse_stats(lines)
# 解析 Hunk 头
chunks = self._parse_chunks(lines)
# 检测语言
language = self._detect_language(new_path)
return FileChange(
path=new_path,
old_path=old_path if old_path != new_path else None,
change_type=change_type,
diff=file_diff,
additions=additions,
deletions=deletions,
chunks=chunks,
language=language,
metadata={
"old_path": old_path,
"new_path": new_path
}
)
def _detect_change_type(self, lines: List[str]) -> ChangeType:
"""检测变更类型"""
# 新文件
if any("new file mode" in line for line in lines):
return ChangeType.ADDED
# 删除文件
if any("deleted file mode" in line for line in lines):
return ChangeType.DELETED
# 重命名
if any("rename from" in line or "rename to" in line for line in lines):
return ChangeType.RENAMED
return ChangeType.MODIFIED
def _parse_stats(self, lines: List[str]) -> tuple:
"""解析变更统计"""
additions = 0
deletions = 0
# @@ -x,y +z,w @@ 格式
for line in lines:
match = re.match(r"@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@", line)
if match:
additions += int(match.group(1))
deletions += int(match.group(2)) if match.group(2) else 0
# 或者解析末尾统计行
stat_match = re.search(r"(\d+) insertions?\(\+\).*?(\d+) deletions?\(-\)", "\n".join(lines))
if stat_match:
additions = int(stat_match.group(1))
deletions = int(stat_match.group(2))
return additions, deletions
def _parse_chunks(self, lines: List[str]) -> List[DiffChunk]:
"""解析 Diff 块"""
chunks = []
current_chunk = None
current_lines = []
old_line_num = 0
new_line_num = 0
for line in lines:
# Hunk 头
hunk_match = re.match(r"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@(.*)", line)
if hunk_match:
# 保存上一个 chunk
if current_chunk:
current_chunk.lines = current_lines
chunks.append(current_chunk)
old_start = int(hunk_match.group(1))
old_lines = int(hunk_match.group(2)) if hunk_match.group(2) else 1
new_start = int(hunk_match.group(3))
new_lines = int(hunk_match.group(4)) if hunk_match.group(4) else 1
current_chunk = DiffChunk(
old_start=old_start,
old_lines=old_lines,
new_start=new_start,
new_lines=new_lines
)
current_lines = []
old_line_num = old_start
new_line_num = new_start
continue
# 跳过非变更内容
if not line or line.startswith(("diff --git", "index ", "---", "+++")):
continue
# 解析变更行
if line.startswith(("+", "-", " ")):
line_type = line[0]
content = line[1:]
is_blank = not content.strip()
diff_line = DiffLine(
line_type=line_type,
content=content,
old_line_num=old_line_num if line_type != "+" else None,
new_line_num=new_line_num if line_type != "-" else None,
is_blank_line=is_blank,
is_comment_line=self._is_comment_line(content, current_chunk.new_start if current_chunk else 0)
)
current_lines.append(diff_line)
# 更新行号
if line_type != "+":
old_line_num += 1
if line_type != "-":
new_line_num += 1
# 保存最后一个 chunk
if current_chunk:
current_chunk.lines = current_lines
chunks.append(current_chunk)
return chunks
def _is_comment_line(self, content: str, line_num: int) -> bool:
"""检测是否为注释行"""
content = content.strip()
# 单行注释
if content.startswith(("#", "//", "--")):
return True
# 块注释开始/结束
if content.startswith(("/*", "*/", "*/")) and content.endswith(("/*", "*/")):
return True
return False
def _detect_language(self, filepath: str) -> Optional[str]:
"""检测编程语言"""
import os
_, ext = os.path.splitext(filepath)
return self.EXTENSION_TO_LANGUAGE.get(ext.lower())
def _generate_summary(self) -> str:
"""生成变更摘要"""
if not self.files:
return "No changes"
summaries = []
for f in self.files[:5]: # 只显示前5个文件
type_icon = {
ChangeType.ADDED: "✨",
ChangeType.DELETED: "🗑️",
ChangeType.MODIFIED: "📝",
ChangeType.RENAMED: "📦"
}.get(f.change_type, "📄")
summaries.append(
f"{type_icon} {f.path} (+{f.additions}/-{f.deletions})"
)
if len(self.files) > 5:
summaries.append(f"... and {len(self.files) - 5} more files")
return "\n".join(summaries)3.3 AI 分析引擎
# app/services/ai_analyzer.py
"""
AI 代码分析引擎
"""
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from app.services.diff_parser import DiffResult, FileChange, ChangeType
from app.services.knowledge_base import KnowledgeBase
class IssueSeverity(Enum):
"""问题严重级别"""
CRITICAL = "critical" # 必须修复
WARNING = "warning" # 建议修复
INFO = "info" # 学习参考
@dataclass
class CodeIssue:
"""代码问题"""
severity: IssueSeverity
# 问题描述
title: str
description: str
suggestion: str
# 位置信息
file_path: str
line_start: Optional[int] = None
line_end: Optional[int] = None
code_snippet: Optional[str] = None
# 分类
category: str # "security", "performance", "style", "best_practice", "bug"
# 相关规则
rule_id: Optional[str] = None
rule_url: Optional[str] = None
# AI 置信度
confidence: float = 0.8
# 元数据
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class FileAnalysis:
"""单个文件的分析结果"""
file_path: str
language: Optional[str]
change_type: ChangeType
issues: List[CodeIssue] = field(default_factory=list)
# 统计
critical_count: int = 0
warning_count: int = 0
info_count: int = 0
# AI 整体评价
summary: str = ""
# 处理时间
processing_time_ms: float = 0
@dataclass
class ReviewResult:
"""完整的 Review 结果"""
mr_id: int
mr_iid: int
mr_title: str
files: List[FileAnalysis] = field(default_factory=list)
# 总体统计
total_files: int = 0
total_issues: int = 0
critical_count: int = 0
warning_count: int = 0
info_count: int = 0
# 总体评价
overall_summary: str = ""
# 知识库匹配
knowledge_matches: List[Dict[str, Any]] = field(default_factory=list)
# 处理信息
processing_time_ms: float = 0
tokens_used: int = 0
class AIAnalyzer:
"""
AI 代码分析引擎
分析维度：
1. 安全漏洞
2. 性能问题
3. 代码规范
4. 最佳实践
5. Bug 检测
"""
def __init__(
self,
llm_client,
knowledge_base: Optional[KnowledgeBase] = None
):
self.llm = llm_client
self.kb = knowledge_base
async def analyze(self, diff_result: DiffResult, context: Dict[str, Any]) -> ReviewResult:
"""
分析代码变更
Args:
diff_result: Diff 解析结果
context: 上下文信息（MR 标题、描述、评论等）
Returns:
ReviewResult: 分析结果
"""
import time
start_time = time.time()
# 查询相关知识
knowledge_matches = []
if self.kb:
query = f"{context.get('title', '')} {context.get('description', '')}"
knowledge_matches = await self.kb.search(query, top_k=5)
# 并行分析每个文件
file_analyses = await asyncio.gather(*[
self._analyze_file(file_change, context)
for file_change in diff_result.files
])
# 聚合结果
result = ReviewResult(
mr_id=context.get("mr_id"),
mr_iid=context.get("mr_iid"),
mr_title=context.get("title", ""),
files=file_analyses,
total_files=len(file_analyses),
total_issues=sum(f.total_issues for f in file_analyses),
critical_count=sum(f.critical_count for f in file_analyses),
warning_count=sum(f.warning_count for f in file_analyses),
info_count=sum(f.info_count for f in file_analyses),
knowledge_matches=knowledge_matches,
processing_time_ms=(time.time() - start_time) * 1000
)
# 生成总体评价
result.overall_summary = await self._generate_overall_summary(result, context)
return result
async def _analyze_file(
self,
file_change: FileChange,
context: Dict[str, Any]
) -> FileAnalysis:
"""分析单个文件"""
import time
start_time = time.time()
# 构建分析 Prompt
prompt = self._build_analysis_prompt(file_change, context)
# 调用 LLM 分析
response = await self.llm.agenerate(prompt)
# 解析结果
issues = self._parse_analysis_response(response, file_change)
# 统计
critical_count = sum(1 for i in issues if i.severity == IssueSeverity.CRITICAL)
warning_count = sum(1 for i in issues if i.severity == IssueSeverity.WARNING)
info_count = sum(1 for i in issues if i.severity == IssueSeverity.INFO)
return FileAnalysis(
file_path=file_change.path,
language=file_change.language,
change_type=file_change.change_type,
issues=issues,
critical_count=critical_count,
warning_count=warning_count,
info_count=info_count,
summary=f"检测到 {critical_count} 个严重问题，{warning_count} 个警告，{info_count} 个建议",
processing_time_ms=(time.time() - start_time) * 1000
)
def _build_analysis_prompt(
self,
file_change: FileChange,
context: Dict[str, Any]
) -> str:
"""构建分析 Prompt"""
# 提取变更的核心代码（只取新增和修改的行）
code_snippets = []
for chunk in file_change.chunks:
for line in chunk.lines:
if line.line_type in ["+", " "]:
if not line.is_blank_line and not line.is_comment_line:
code_snippets.append(f"Line {line.new_line_num}: {line.content}")
code_context = "\n".join(code_snippets[:100]) # 限制长度
return f"""你是一个资深的代码审查专家。请分析以下代码变更：
## 变更文件
路径: {file_change.path}
语言: {file_change.language or 'unknown'}
变更类型: {file_change.change_type.value}
新增行数: {file_change.additions}
删除行数: {file_change.deletions}
## MR 上下文
标题: {context.get('title', 'N/A')}
描述: {context.get('description', 'N/A')}
## 代码变更
{code_context}
## 分析要求
请从以下维度进行审查，每行代码最多识别一个问题：
1. **安全漏洞 (security)**
- SQL 注入、XSS、命令注入
- 敏感信息泄露（密码、密钥、Token）
- 不安全的依赖使用
- 权限控制问题
2. **性能问题 (performance)**
- 循环中的数据库查询
- 大数据量的内存问题
- 不必要的重复计算
- N+1 查询问题
3. **Bug 风险 (bug)**
- 空指针/空值检查缺失
- 边界条件处理不当
- 异常处理不当
- 逻辑错误
4. **代码规范 (style)**
- 命名不规范
- 注释缺失或过时
- 代码重复
- 函数过长
5. **最佳实践 (best_practice)**
- 缺少单元测试
- 配置硬编码
- 日志记录不足
- 不符合项目架构
## 输出格式
请以 JSON 格式输出分析结果：
{{
"issues": [
{{
"severity": "critical/warning/info",
"category": "security/performance/bug/style/best_practice",
"title": "问题简述",
"description": "问题详细说明",
"suggestion": "修复建议",
"line_start": 行号,
"code_snippet": "相关代码",
"confidence": 0.0-1.0
}}
],
"summary": "整体评价"
}}
注意：
- 只输出确实存在问题的情况，不要过度审查
- severity 判断标准：
- critical: 可能导致生产事故、严重安全风险、核心功能错误
- warning: 可能导致问题、需要改进
- info: 代码优化建议、学习参考
- 重点关注新增和修改的代码，不要审查未变更的部分"""
def _parse_analysis_response(
self,
response: str,
file_change: FileChange
) -> List[CodeIssue]:
"""解析 LLM 分析响应"""
import json
import re
issues = []
try:
# 尝试解析 JSON
# 提取 ```json ... ``` 包裹的内容
json_match = re.search(r"```(?:json)?\s*(.*?)\s*```", response, re.DOTALL)
if json_match:
data = json.loads(json_match.group(1))
else:
# 尝试直接解析
data = json.loads(response)
for item in data.get("issues", []):
issue = CodeIssue(
severity=IssueSeverity(item.get("severity", "info")),
title=item.get("title", "Unknown issue"),
description=item.get("description", ""),
suggestion=item.get("suggestion", ""),
file_path=file_change.path,
line_start=item.get("line_start"),
code_snippet=item.get("code_snippet"),
category=item.get("category", "best_practice"),
confidence=item.get("confidence", 0.8)
)
issues.append(issue)
except (json.JSONDecodeError, KeyError) as e:
# 解析失败，记录原始响应用于调试
print(f"Failed to parse analysis response: {e}")
print(f"Response: {response[:500]}")
return issues
async def _generate_overall_summary(
self,
result: ReviewResult,
context: Dict[str, Any]
) -> str:
"""生成总体评价"""
# 统计各类型问题
by_category = {}
for file_analysis in result.files:
for issue in file_analysis.issues:
if issue.category not in by_category:
by_category[issue.category] = 0
by_category[issue.category] += 1
# 构建评价 Prompt
prompt = f"""作为代码审查专家，请对以下 Review 结果给出总体评价：
MR 标题: {result.mr_title}
代码变更统计:
- 总文件数: {result.total_files}
- 总问题数: {result.total_issues}
- 严重问题: {result.critical_count}
- 警告: {result.warning_count}
- 建议: {result.info_count}
问题分类统计:
{chr(10).join(f"- {cat}: {count}" for cat, count in sorted(by_category.items()))}
请给出：
1. 总体评价（1-2句话）
2. 是否建议合并（建议/需要修改/阻止合并）
3. 需要关注的主要问题（最多3个）
请简洁回答。"""
response = await self.llm.agenerate(prompt)
return response
# ============== LLM 客户端封装 ==============
class LLMClient:
"""
LLM 客户端封装
支持多后端：OpenAI、Claude、DeepSeek 等
"""
def __init__(
self,
provider: str = "openai",
api_key: str = "",
model: str = "gpt-4o",
base_url: Optional[str] = None
):
self.provider = provider
self.api_key = api_key
self.model = model
self.base_url = base_url
self._client = self._init_client()
def _init_client(self):
"""初始化客户端"""
if self.provider == "openai":
from openai import AsyncOpenAI
return AsyncOpenAI(api_key=self.api_key)
elif self.provider == "anthropic":
from anthropic import AsyncAnthropic
return AsyncAnthropic(api_key=self.api_key)
elif self.provider == "deepseek":
from openai import AsyncOpenAI
return AsyncOpenAI(
api_key=self.api_key,
base_url=base_url or "https://api.deepseek.com"
)
else:
raise ValueError(f"Unsupported provider: {self.provider}")
async def agenerate(
self,
prompt: str,
temperature: float = 0.0,
max_tokens: int = 4096
) -> str:
"""
异步生成文本
Returns:
生成的文本
"""
if self.provider == "openai":
response = await self._client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=temperature,
max_tokens=max_tokens
)
return response.choices[0].message.content
elif self.provider == "anthropic":
response = await self._client.messages.create(
model=self.model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
elif self.provider == "deepseek":
response = await self._client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=temperature,
max_tokens=max_tokens
)
return response.choices[0].message.content
raise ValueError(f"Unsupported provider: {self.provider}")3.4 评论生成与输出
# app/services/comment_generator.py
"""
Review 评论生成与输出
"""
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from app.services.ai_analyzer import ReviewResult, FileAnalysis, CodeIssue, IssueSeverity
@dataclass
class Comment:
"""评论"""
body: str
is_inline: bool = False # 是否是行内评论
file_path: Optional[str] = None
line_number: Optional[int] = None
# Markdown 格式支持
is_markdown: bool = True
# 附件（如代码片段）
attachments: List[str] = field(default_factory=list)
@dataclass
class CommentSet:
"""一组评论"""
mr_iid: int
# 总评评论
summary_comment: Optional[Comment] = None
# 文件级评论
file_comments: List[Comment] = field(default_factory=list)
# 行内评论
inline_comments: List[Comment] = field(default_factory=list)
class CommentGenerator:
"""
Review 评论生成器
功能：
1. 将分析结果转换为 Markdown 评论
2. 按优先级排序
3. 生成摘要和详情
"""
def __init__(self):
self.severity_icons = {
IssueSeverity.CRITICAL: "🔴",
IssueSeverity.WARNING: "🟡",
IssueSeverity.INFO: "🔵"
}
self.severity_labels = {
IssueSeverity.CRITICAL: "严重",
IssueSeverity.WARNING: "警告",
IssueSeverity.INFO: "建议"
}
self.category_icons = {
"security": "🔒",
"performance": "⚡",
"bug": "🐛",
"style": "📝",
"best_practice": "✨"
}
def generate(self, result: ReviewResult) -> CommentSet:
"""
生成评论集
Args:
result: Review 结果
Returns:
CommentSet: 评论集
"""
comment_set = CommentSet(mr_iid=result.mr_iid)
# 生成总评评论
comment_set.summary_comment = self._generate_summary_comment(result)
# 生成文件评论
for file_analysis in result.files:
if file_analysis.issues:
file_comment = self._generate_file_comment(file_analysis)
comment_set.file_comments.append(file_comment)
# 生成行内评论
for file_analysis in result.files:
for issue in file_analysis.issues:
if issue.line_start:
inline = self._generate_inline_comment(issue)
comment_set.inline_comments.append(inline)
return comment_set
def _generate_summary_comment(self, result: ReviewResult) -> Comment:
"""生成总评评论"""
# 统计
total = result.total_issues
if total == 0:
status = "✅ **Looks Good!**"
verdict = "代码审查未发现问题，可以合并。"
elif result.critical_count > 0:
status = "🚨 **需要修改**"
verdict = f"发现 {result.critical_count} 个严重问题，建议修改后再合并。"
elif result.warning_count > 0:
status = "⚠️ **建议修改**"
verdict = f"发现 {result.warning_count} 个问题，建议修复后合并。"
else:
status = "💡 **有优化建议**"
verdict = f"发现 {result.info_count} 个优化建议，可以作为参考。"
body = f"""
## 🤖 AI Code Review 结果
{status}
### 📊 变更统计
| 指标 | 数值 |
|------|------|
| 修改文件数 | {result.total_files} |
| 总问题数 | {total} |
| 🔴 严重 | {result.critical_count} |
| 🟡 警告 | {result.warning_count} |
| 🔵 建议 | {result.info_count} |
### 📝 变更摘要
{result.overall_summary or '无'}
### ⏱️ 处理信息
- 分析耗时: {result.processing_time_ms:.0f}ms
---
*此评论由 AI Code Review 系统自动生成*
"""
return Comment(body=body.strip())
def _generate_file_comment(self, file_analysis: FileAnalysis) -> Comment:
"""生成文件级评论"""
severity_counts = {
IssueSeverity.CRITICAL: file_analysis.critical_count,
IssueSeverity.WARNING: file_analysis.warning_count,
IssueSeverity.INFO: file_analysis.info_count
}
issues_by_severity = {
IssueSeverity.CRITICAL: [],
IssueSeverity.WARNING: [],
IssueSeverity.INFO: []
}
for issue in file_analysis.issues:
issues_by_severity[issue.severity].append(issue)
# 构建问题列表
issue_sections = []
for severity in [IssueSeverity.CRITICAL, IssueSeverity.WARNING, IssueSeverity.INFO]:
issues = issues_by_severity[severity]
if not issues:
continue
icon = self.severity_icons[severity]
label = self.severity_labels[severity]
items = []
for issue in issues:
category_icon = self.category_icons.get(issue.category, "📄")
items.append(
f"- **{category_icon} {issue.title}**\n"
f" {issue.description}\n"
f" > 💡 {issue.suggestion}"
)
issue_sections.append(
f"#### {icon} {label} ({len(issues)}个)\n\n" + "\n\n".join(items)
)
body = f"""
## 📁 {file_analysis.path}
{file_analysis.summary}
---
{"".join(issue_sections)}
"""
return Comment(
body=body.strip(),
is_inline=False,
file_path=file_analysis.path
)
def _generate_inline_comment(self, issue: CodeIssue) -> Comment:
"""生成行内评论"""
icon = self.severity_icons[issue.severity]
label = self.severity_labels[issue.severity]
category_icon = self.category_icons.get(issue.category, "📄")
body = f"{icon} **{label}**: {issue.description}\n\n💡 {issue.suggestion}"
return Comment(
body=body,
is_inline=True,
file_path=issue.file_path,
line_number=issue.line_start
)
def to_github_format(self, comment: Comment) -> Dict[str, Any]:
"""转换为 GitHub API 格式"""
if comment.is_inline:
return {
"body": comment.body,
"path": comment.file_path,
"line": comment.line_number,
"side": "RIGHT" # 新代码行
}
else:
return {"body": comment.body}
def to_gitlab_format(self, comment: Comment, position: Optional[Dict] = None) -> Dict[str, Any]:
"""转换为 GitLab API 格式"""
if comment.is_inline and position:
return {
"body": comment.body,
"position": position # 需要计算具体位置
}
else:
return {"body": comment.body}3.5 GitLab API 集成
# app/services/gitlab_client.py
"""
GitLab API 客户端
"""
import httpx
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from app.services.comment_generator import CommentSet, Comment
@dataclass
class GitLabConfig:
"""GitLab 配置"""
url: str
token: str
project_id: int
class GitLabClient:
"""
GitLab API 客户端
功能：
1. 获取 MR 信息
2. 获取 Diff
3. 提交评论
"""
def __init__(self, config: GitLabConfig):
self.config = config
self.base_url = f"{config.url}/api/v4"
self.headers = {
"PRIVATE-TOKEN": config.token,
"Content-Type": "application/json"
}
async def get_merge_request(self, mr_iid: int) -> Dict[str, Any]:
"""获取 MR 信息"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/projects/{self.config.project_id}/merge_requests/{mr_iid}",
headers=self.headers
)
response.raise_for_status()
return response.json()
async def get_mr_changes(self, mr_iid: int) -> Dict[str, Any]:
"""获取 MR 的变更"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/projects/{self.config.project_id}/merge_requests/{mr_iid}/changes",
headers=self.headers
)
response.raise_for_status()
return response.json()
async def get_mr_diff(self, mr_iid: int) -> str:
"""获取 MR 的 Diff"""
# 获取完整的 unified diff
async with httpx.AsyncClient() as client:
# GitLab 没有直接的 diff API，使用 changes + 本地生成
changes = await self.get_mr_changes(mr_iid)
diff_lines = []
for change in changes.get("changes", []):
diff_lines.append(f"diff --git a/{change['old_path']} b/{change['new_path']}")
diff_lines.append(f"--- a/{change['old_path']}")
diff_lines.append(f"+++ b/{change['new_path']}")
if "diff" in change:
diff_lines.append(change["diff"])
diff_lines.append("")
return "\n".join(diff_lines)
async def post_comment(
self,
mr_iid: int,
body: str,
is_system: bool = True
) -> Dict[str, Any]:
"""提交 MR 评论"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/projects/{self.config.project_id}/merge_requests/{mr_iid}/notes",
headers=self.headers,
json={
"body": body,
"system": is_system # 系统评论会显示为机器人
}
)
response.raise_for_status()
return response.json()
async def post_inline_comment(
self,
mr_iid: int,
body: str,
file_path: str,
old_path: str,
new_path: str,
position: Dict[str, Any]
) -> Dict[str, Any]:
"""
提交行内评论（Diff 位置评论）
position 格式:
{
"base_sha": "...",
"start_sha": "...",
"head_sha": "...",
"position_type": "text",
"old_path": "...",
"new_path": "...",
"old_line": 10, # 或 None
"new_line": 10,
"width": 10,
"height": 10
}
"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/projects/{self.config.project_id}/merge_requests/{mr_iid}/discussions",
headers=self.headers,
json={
"body": body,
"position": {
"base_sha": position.get("base_sha"),
"start_sha": position.get("start_sha"),
"head_sha": position.get("head_sha"),
"position_type": "text",
"old_path": old_path,
"new_path": new_path,
"old_line": position.get("old_line"),
"new_line": position.get("new_line")
}
}
)
response.raise_for_status()
return response.json()
async def post_review(self, mr_iid: int, comment_set: CommentSet) -> Dict[str, Any]:
"""
发布完整 Review 结果
策略：
1. 先发总评评论
2. 再发文件级评论
3. 最后发行内评论
"""
results = {
"summary_posted": False,
"file_comments": 0,
"inline_comments": 0
}
# 1. 发送总评评论
if comment_set.summary_comment:
await self.post_comment(
mr_iid=mr_iid,
body=comment_set.summary_comment.body,
is_system=True
)
results["summary_posted"] = True
# 2. 发送文件级评论
for file_comment in comment_set.file_comments:
await self.post_comment(
mr_iid=mr_iid,
body=file_comment.body,
is_system=True
)
results["file_comments"] += 1
# 3. 发送行内评论
# 需要获取 MR 的 diff_refs
mr_info = await self.get_merge_request(mr_iid)
diff_refs = mr_info.get("diff_refs", {})
for inline_comment in comment_set.inline_comments:
try:
# 简化处理：使用文件的第一行位置
position = {
"base_sha": diff_refs.get("base_sha"),
"start_sha": diff_refs.get("start_sha"),
"head_sha": diff_refs.get("head_sha"),
"new_line": inline_comment.line_number
}
await self.post_inline_comment(
mr_iid=mr_iid,
body=inline_comment.body,
file_path=inline_comment.file_path,
old_path=inline_comment.file_path,
new_path=inline_comment.file_path,
position=position
)
results["inline_comments"] += 1
except Exception as e:
print(f"Failed to post inline comment: {e}")
return results四、FastAPI 后端完整代码
# app/main.py
"""
AI Code Review 系统 - FastAPI 主入口
"""
import os
from contextlib import asynccontextmanager
from typing import List, Optional
from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from datetime import datetime
from app.config import settings
from app.services.diff_parser import DiffParser
from app.services.ai_analyzer import AIAnalyzer, LLMClient, ReviewResult
from app.services.comment_generator import CommentGenerator
from app.services.gitlab_client import GitLabClient, GitLabConfig
from app.services.knowledge_base import KnowledgeBase
from app.tasks.review import analyze_merge_request, get_task_status
from app.database import get_db, ReviewRecord
# ============== API Models ==============
class MRWebhookRequest(BaseModel):
"""MR Webhook 请求"""
object_kind: str
event_type: str
project: dict
object_attributes: dict
user: dict
changes: Optional[dict] = None
class ReviewRequest(BaseModel):
"""手动 Review 请求"""
mr_iid: int
mr_id: int = Field(alias="mr_id")
title: str
description: str = ""
diff: str # 完整的 diff 文本
class ReviewResponse(BaseModel):
"""Review 响应"""
task_id: str
status: str
message: str
class ReviewStatusResponse(BaseModel):
"""Review 状态查询"""
task_id: str
status: str
result: Optional[dict] = None
error: Optional[str] = None
# ============== 依赖注入 ==============
def get_llm_client() -> LLMClient:
"""获取 LLM 客户端"""
return LLMClient(
provider=settings.LLM_PROVIDER,
api_key=settings.LLM_API_KEY,
model=settings.LLM_MODEL
)
def get_gitlab_client() -> GitLabClient:
"""获取 GitLab 客户端"""
return GitLabClient(
config=GitLabConfig(
url=settings.GITLAB_URL,
token=settings.GITLAB_TOKEN,
project_id=settings.GITLAB_PROJECT_ID
)
)
def get_knowledge_base() -> Optional[KnowledgeBase]:
"""获取知识库（可选）"""
if settings.ENABLE_KNOWLEDGE_BASE:
return KnowledgeBase()
return None
# ============== API 路由 ==============
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# 启动时
print("🚀 AI Code Review 系统启动")
yield
# 关闭时
print("👋 AI Code Review 系统关闭")
app = FastAPI(
title="AI Code Review System",
description="基于 AI 的自动化代码审查系统",
version="1.0.0",
lifespan=lifespan
)
# CORS 配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health")
async def health_check():
"""健康检查"""
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
@app.post("/webhooks/gitlab", response_model=ReviewResponse)
async def handle_gitlab_webhook(
request: MRWebhookRequest,
background_tasks: BackgroundTasks
):
"""
处理 GitLab Webhook
当 MR 创建或更新时自动触发 Review
"""
# 只处理 MR 事件
if request.object_kind != "merge_request":
return ReviewResponse(
task_id="",
status="ignored",
message=f"Ignored event type: {request.object_kind}"
)
# 只处理打开或更新的 MR
state = request.object_attributes.get("state")
if state not in ["opened", "update"]:
return ReviewResponse(
task_id="",
status="ignored",
message=f"Ignored MR state: {state}"
)
mr_iid = request.object_attributes.get("iid")
mr_id = request.object_attributes.get("id")
# 触发异步 Review
task = analyze_merge_request.delay({
"mr_iid": mr_iid,
"mr_id": mr_id,
"title": request.object_attributes.get("title", ""),
"description": request.object_attributes.get("description", ""),
"source_branch": request.object_attributes.get("source_branch"),
"target_branch": request.object_attributes.get("target_branch"),
"url": request.object_attributes.get("url"),
"diff_refs": request.object_attributes.get("diff_refs", {})
})
return ReviewResponse(
task_id=task.id,
status="queued",
message=f"Review task queued for MR !{mr_iid}"
)
@app.post("/review", response_model=ReviewResponse)
async def trigger_review(
request: ReviewRequest,
background_tasks: BackgroundTasks,
llm_client: LLMClient = Depends(get_llm_client),
gitlab_client: GitLabClient = Depends(get_gitlab_client)
):
"""
手动触发 Review
适用于测试或外部系统集成
"""
# 验证 MR 存在
try:
mr_info = await gitlab_client.get_merge_request(request.mr_iid)
except Exception as e:
raise HTTPException(status_code=404, detail=f"MR not found: {e}")
# 触发 Review
task = analyze_merge_request.delay({
"mr_iid": request.mr_iid,
"mr_id": request.mr_id,
"title": request.title,
"description": request.description,
"diff": request.diff
})
return ReviewResponse(
task_id=task.id,
status="queued",
message=f"Review task queued"
)
@app.get("/review/{task_id}", response_model=ReviewStatusResponse)
async def get_review_status(task_id: str):
"""
查询 Review 状态
返回任务状态和结果（如果完成）
"""
result = get_task_status(task_id)
if not result:
raise HTTPException(status_code=404, detail="Task not found")
return ReviewStatusResponse(
task_id=task_id,
status=result.get("status", "unknown"),
result=result.get("result"),
error=result.get("error")
)
@app.get("/review/history")
async def get_review_history(
limit: int = 20,
offset: int = 0,
mr_iid: Optional[int] = None
):
"""
获取 Review 历史记录
"""
db = next(get_db())
query = db.query(ReviewRecord)
if mr_iid:
query = query.filter(ReviewRecord.mr_iid == mr_iid)
total = query.count()
records = query.order_by(ReviewRecord.created_at.desc()).offset(offset).limit(limit).all()
return {
"total": total,
"records": [
{
"id": r.id,
"mr_iid": r.mr_iid,
"status": r.status,
"critical_count": r.critical_count,
"warning_count": r.warning_count,
"info_count": r.info_count,
"processing_time_ms": r.processing_time_ms,
"created_at": r.created_at.isoformat()
}
for r in records
]
}
# ============== 运行入口 ==============
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=8000,
reload=True
)五、Docker + K8s 部署配置
5.1 Docker 配置
# Dockerfile
FROM python:3.11-slim AS base
# 安装系统依赖
RUN apt-get update && apt-get install -y \
git \
curl \
&& rm -rf /var/lib/apt/lists/*
# 设置工作目录
WORKDIR /app
# 安装 Python 依赖
FROM base AS builder
RUN pip install --no-cache-dir poetry
COPY pyproject.toml poetry.lock* ./
RUN poetry config virtualenvs.create false \
&& poetry install --no-interaction --no-ansi --no-root
# 复制应用代码
FROM base AS runner
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY . .
# 非 root 用户运行
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
# 环境变量
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
# 端口
EXPOSE 8000
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]# docker-compose.yml
version: '3.8'
services:
api:
build: .
ports:
- "8000:8000"
environment:
- REDIS_URL=redis://redis:6379/0
- DATABASE_URL=postgresql://postgres:password@db:5432/ai_review
- GITLAB_URL=${GITLAB_URL}
- GITLAB_TOKEN=${GITLAB_TOKEN}
- GITLAB_PROJECT_ID=${GITLAB_PROJECT_ID}
- LLM_PROVIDER=${LLM_PROVIDER}
- LLM_API_KEY=${LLM_API_KEY}
- LLM_MODEL=${LLM_MODEL}
depends_on:
- redis
- db
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
worker:
build: .
command: celery -A app.celery_app worker --loglevel=info
environment:
- REDIS_URL=redis://redis:6379/0
- DATABASE_URL=postgresql://postgres:password@db:5432/ai_review
- LLM_PROVIDER=${LLM_PROVIDER}
- LLM_API_KEY=${LLM_API_KEY}
depends_on:
- redis
- db
restart: unless-stopped
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
restart: unless-stopped
db:
image: postgres:15-alpine
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=password
- POSTGRES_DB=ai_review
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
restart: unless-stopped
volumes:
redis_data:
postgres_data:5.2 Kubernetes 部署配置
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-code-review-api
labels:
app: ai-code-review
component: api
spec:
replicas: 2
selector:
matchLabels:
app: ai-code-review
component: api
template:
metadata:
labels:
app: ai-code-review
component: api
spec:
containers:
- name: api
image: your-registry/ai-code-review:latest
ports:
- containerPort: 8000
name: http
env:
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: redis_url
- name: LLM_API_KEY
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: llm_api_key
- name: GITLAB_TOKEN
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: gitlab_token
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: database_url
resources:
requests:
memory: "256Mi"
cpu: "250m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 30
timeoutSeconds: 5
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 3
volumeMounts:
- name: tmp
mountPath: /tmp
volumes:
- name: tmp
emptyDir: {}
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchLabels:
component: api
topologyKey: kubernetes.io/hostname
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-code-review-worker
labels:
app: ai-code-review
component: worker
spec:
replicas: 3
selector:
matchLabels:
app: ai-code-review
component: worker
template:
metadata:
labels:
app: ai-code-review
component: worker
spec:
containers:
- name: worker
image: your-registry/ai-code-review:latest
command: ["celery"]
args: ["-A", "app.celery_app", "worker", "--loglevel=info", "--concurrency=4"]
env:
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: redis_url
- name: LLM_API_KEY
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: llm_api_key
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: ai-code-review-secrets
key: database_url
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "1Gi"
cpu: "1000m"
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchLabels:
component: worker
topologyKey: kubernetes.io/hostname
---
apiVersion: v1
kind: Service
metadata:
name: ai-code-review-api-service
spec:
type: ClusterIP
ports:
- port: 80
targetPort: 8000
name: http
selector:
app: ai-code-review
component: api
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: ai-code-review-ingress
annotations:
nginx.ingress.kubernetes.io/rewrite-target: /
spec:
rules:
- host: ai-review.your-domain.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: ai-code-review-api-service
port:
number: 80
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: ai-code-review-api-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: ai-code-review-api
minReplicas: 2
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 805.3 配置管理
# config.yaml
# 应用配置
app:
name: "AI Code Review"
version: "1.0.0"
debug: false
# API 配置
api:
host: "0.0.0.0"
port: 8000
workers: 4
# CORS
cors:
allowed_origins:
- "http://localhost:3000"
- "https://your-gitlab.com"
allow_credentials: true
# GitLab 配置
gitlab:
url: "${GITLAB_URL}"
token: "${GITLAB_TOKEN}"
project_id: "${GITLAB_PROJECT_ID}"
# Webhook 配置
webhook:
secret: "${GITLAB_WEBHOOK_SECRET}"
events:
- merge_request
# LLM 配置
llm:
provider: "${LLM_PROVIDER}" # openai, anthropic, deepseek
api_key: "${LLM_API_KEY}"
model: "${LLM_MODEL}" # gpt-4o, claude-3-5-sonnet, deepseek-chat
# 推理配置
inference:
temperature: 0.0
max_tokens: 4096
timeout: 60
# Redis 配置
redis:
url: "${REDIS_URL}"
db: 0
# 数据库配置
database:
url: "${DATABASE_URL}"
pool_size: 10
max_overflow: 20
# 知识库配置
knowledge_base:
enabled: true
vector_db:
type: "chroma" # chroma, milvus, qdrant
persist_directory: "/data/knowledge_base"
embedding:
model: "bge-large-zh-v1.5"
dimension: 1024
# 分析配置
analysis:
# 问题检测配置
security:
enabled: true
check_sql_injection: true
check_xss: true
check_command_injection: true
check_secrets: true
performance:
enabled: true
check_n_plus_1: true
check_loop_queries: true
max_complexity: 15
style:
enabled: true
max_line_length: 120
max_function_lines: 50
# 文件过滤
exclude_patterns:
- "*.min.js"
- "*.css"
- "vendor/**"
- "node_modules/**"
- "dist/**"
# 语言配置
supported_languages:
- python
- javascript
- typescript
- java
- go
- rust
- cpp
# 告警配置
alerting:
enabled: true
# 问题阈值
thresholds:
critical_to_block_merge: 1
warning_review_required: 5
# 通知渠道
channels:
- type: "webhook"
url: "${SLACK_WEBHOOK_URL}"六、优缺点分析
6.1 系统优势
6.2 系统局限
6.3 适用场景
最适合：
代码规范检查（命名、格式、注释）
安全问题扫描（明显的安全漏洞）
重复性问题检测
最佳实践建议
快速初步 Review
不太适合：
需要深入业务逻辑的理解
架构设计和系统级问题
性能调优的精确建议
复杂算法的正确性验证
七、后续优化方向
7.1 技术优化
本地 LLM 部署
使用 vLLM 部署 DeepSeek/Qwen 模型
降低 API 成本，提升响应速度
保证代码数据安全
增量 Diff 分析
只分析新增代码，减少 token 消耗
增量 Review，快速反馈
多 Agent 协作
安全 Agent、性能 Agent、规范 Agent 分工
结果汇总，避免遗漏
7.2 功能增强
自适应阈值
根据项目历史自动调整问题阈值
差异化配置（核心模块更严格）
历史学习
学习团队既往的 Review 习惯
生成符合团队风格的建议
自动修复建议
不仅指出问题，还提供修复代码
一键应用修复
7.3 集成扩展
IDE 插件
VS Code / JetBrains 插件
本地开发时实时 Review
CI/CD 集成
作为 CI 检查的一环
Blocking merge 条件可配置
知识库完善
团队编码规范文档化
FAQ 问答库
八、总结
本文完整实现了一个 AI 辅助 Code Review 系统，包括：
架构设计：Webhook 触发 → 异步任务处理 → AI 分析 → 评论输出
核心模块：Diff 解析、AI 分析、评论生成、GitLab 集成
部署方案：Docker Compose 本地开发，Kubernetes 生产部署
配置管理：环境变量 + YAML 配置，灵活适配
系统可以在 MR 创建/更新时自动触发 Review，通过 AI 分析代码变更，识别安全漏洞、性能问题、代码规范等，并以分级评论形式输出到 PR 页面。
核心价值：
🚀 提升 Review 效率 50%+
🎯 问题发现率提升 30%+
📊 Review 标准统一
📚 知识持续积累
下一步建议：
先用云端 LLM API 快速验证效果
逐步调整 Prompt 和阈值优化准确率
考虑本地部署开源模型降低成本
根据团队反馈持续迭代
完整代码已开源，欢迎 Star 和贡献！
评论区