Skip to content

4. 关键代码验证

核心逻辑:AI 审查工作流

本节展示如何实现一个基础的 AI 代码审查系统,重点在于关键逻辑和集成方式。

1. GitHub Actions 工作流配置

使用 GitHub Actions 自动触发 AI 代码审查:

yaml
name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize, reopened]

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Get changed files
        id: changed-files
        uses: tj-actions/changed-files@v42
        with:
          files: |
            **/*.py
            **/*.js
            **/*.ts
            **/*.java
            **/*.go

      - name: Run AI Review
        if: steps.changed-files.outputs.any_changed == 'true'
        uses: villesau/ai-codereviewer@v1.0.0
        with:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          model: 'gpt-4'
          exclude_patterns: 'tests/,mocks/'
          max_files: 50

2. AI 审查核心逻辑(概念性伪代码)

以下展示 AI 审查系统的核心处理逻辑:

python
# AI 审查核心流程(概念性伪代码)

async def review_pull_request(pr_context: PRContext) -> List[ReviewComment]:
    """
    审查 Pull Request 的核心函数
    """
    comments = []

    # 阶段 1: 静态分析(确定性检查)
    static_issues = await run_static_analysis(pr_context.diff)
    comments.extend(static_issues)

    # 阶段 2: AI 审查(上下文感知)
    ai_issues = await run_ai_review(pr_context)
    comments.extend(ai_issues)

    # 阶段 3: 后处理(去重、排序)
    filtered_comments = post_process_comments(comments)

    # 阶段 4: 输出到 PR
    await post_comments_to_pr(pr_context, filtered_comments)

    return filtered_comments

async def run_ai_review(pr_context: PRContext) -> List[ReviewComment]:
    """
    使用 LLM 进行代码审查
    """
    comments = []
    changed_files = pr_context.get_changed_files()

    for file in changed_files:
        # 获取文件的完整上下文
        file_content = await pr_context.get_file_content(file.path)
        related_files = await pr_context.get_related_files(file.path)

        # 构建提示词
        prompt = build_review_prompt(
            diff=file.diff,
            file_content=file_content,
            related_files=related_files,
            team_guidelines=pr_context.team_guidelines
        )

        # 调用 LLM
        response = await llm_client.complete(prompt, model="gpt-4")

        # 解析响应
        file_comments = parse_review_response(response, file.path)
        comments.extend(file_comments)

    return comments

def build_review_prompt(
    diff: str,
    file_content: str,
    related_files: Dict[str, str],
    team_guidelines: str
) -> str:
    """
    构建审查提示词
    """
    prompt = f"""
请审查以下代码变更,重点关注:

1. **安全漏洞**:SQL 注入、XSS、硬编码机密、不安全的加密
2. **逻辑错误**:边缘情况、空值处理、异常处理
3. **性能问题**:N+1 查询、内存泄漏、低效算法
4. **代码质量**:可读性、可维护性、命名规范
5. **架构一致性**:是否符合现有架构模式

## 团队编码规范:
{team_guidelines}

## 主要文件变更:
```diff
{diff}

相关文件上下文:

请以 JSON 格式返回评论,每个评论包含:

  • line: 行号
  • severity: 'critical' | 'major' | 'minor' | 'info'
  • message: 问题描述
  • suggestion: 修改建议(可选)
  • rationale: 原因解释 """ return prompt

def post_process_comments(comments: List[ReviewComment]) -> List[ReviewComment]: """ 后处理:去重、过滤、排序 """ # 1. 去重:基于位置和问题类型 unique_comments = remove_duplicates(comments)

# 2. 过滤:移除置信度过低的评论
filtered_comments = [c for c in unique_comments if c.confidence > 0.7]

# 3. 排序:按严重性和位置
sorted_comments = sort_by_severity_and_position(filtered_comments)

# 4. 限制:避免评论过多
if len(sorted_comments) > 20:
    sorted_comments = sorted_comments[:20]

return sorted_comments

### 3. 模块化审查器(基于 Uber uReview 架构)

实现专业化审查器以提供更准确的审查:

```python
# 专业化审查器架构

class BaseReviewer(ABC):
    """审查器基类"""

    @abstractmethod
    async def review(self, context: ReviewContext) -> List[ReviewComment]:
        pass

class StandardReviewer(BaseReviewer):
    """标准审查器:一般代码质量"""

    async def review(self, context: ReviewContext) -> List[ReviewComment]:
        prompt = self._build_standard_prompt(context)
        response = await llm_client.complete(prompt, model="gpt-4")
        return self._parse_response(response)

class SecurityReviewer(BaseReviewer):
    """安全审查器:专注于安全漏洞"""

    async def review(self, context: ReviewContext) -> List[ReviewComment]:
        # 使用专门的安全模型或提示词
        prompt = self._build_security_prompt(context)
        response = await llm_client.complete(
            prompt,
            model="claude-3-5-sonnet"  # 使用在安全方面更强的模型
        )
        return self._parse_response(response)

class ArchitectureReviewer(BaseReviewer):
    """架构审查器:架构一致性"""

    async def review(self, context: ReviewContext) -> List[ReviewComment]:
        prompt = self._build_architecture_prompt(context)
        response = await llm_client.complete(prompt, model="gpt-4")
        return self._parse_response(response)

class PerformanceReviewer(BaseReviewer):
    """性能审查器:性能优化"""

    async def review(self, context: ReviewContext) -> List[ReviewComment]:
        prompt = self._build_performance_prompt(context)
        response = await llm_client.complete(prompt, model="gpt-4")
        return self._parse_response(response)

# 审查协调器
class ReviewOrchestrator:
    """协调多个审查器"""

    def __init__(self):
        self.reviewers = [
            StandardReviewer(),
            SecurityReviewer(),
            ArchitectureReviewer(),
            PerformanceReviewer()
        ]

    async def review(self, context: ReviewContext) -> List[ReviewComment]:
        all_comments = []

        # 并行运行所有审查器
        tasks = [reviewer.review(context) for reviewer in self.reviewers]
        results = await asyncio.gather(*tasks)

        # 合并所有评论
        for comments in results:
            all_comments.extend(comments)

        # 后处理
        return post_process_comments(all_comments)

4. 集成静态分析

在 AI 审查之前先运行确定性检查:

python
# 静态分析集成

async def run_static_analysis(diff: str) -> List[ReviewComment]:
    """
    运行静态分析工具
    """
    comments = []

    # 1. ESLint(JavaScript/TypeScript)
    eslint_output = await run_eslint(diff)
    comments.extend(parse_eslint_output(eslint_output))

    # 2. SonarQube
    sonar_output = await run_sonarqube(diff)
    comments.extend(parse_sonar_output(sonar_output))

    # 3. Custom Linter(团队特定规则)
    custom_output = await run_custom_linter(diff)
    comments.extend(parse_custom_output(custom_output))

    return comments

async def run_eslint(diff: str) -> str:
    """
    运行 ESLint
    """
    process = await asyncio.create_subprocess_exec(
        'eslint',
        '--format', 'json',
        '--stdin',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    stdout, stderr = await process.communicate(input=diff.encode())
    return stdout.decode()

def parse_eslint_output(output: str) -> List[ReviewComment]:
    """
    解析 ESLint 输出
    """
    results = json.loads(output)
    comments = []

    for result in results:
        for message in result['messages']:
            comment = ReviewComment(
                file=result['filePath'],
                line=message['line'],
                severity=map_eslint_severity(message['severity']),
                message=message['message'],
                rule_id=message['ruleId'],
                source='eslint'
            )
            comments.append(comment)

    return comments

5. 自定义规则配置

支持团队自定义审查规则:

yaml
# config/review-rules.yaml

version: "1.0"

# 审查器配置
reviewers:
  standard:
    enabled: true
    model: "gpt-4"
    max_comments: 10

  security:
    enabled: true
    model: "claude-3-5-sonnet"
    max_comments: 5

  architecture:
    enabled: true
    model: "gpt-4"
    max_comments: 5

  performance:
    enabled: true
    model: "gpt-4"
    max_comments: 5

# 团队编码规范
team_guidelines: |
  - 使用 TypeScript 严格模式
  - 所有函数必须有 JSDoc 注释
  - 使用函数式编程,避免副作用
  - 错误处理必须包含日志记录
  - 遵循 SOLID 原则

# 文件排除规则
exclude_patterns:
  - "tests/**/*"
  - "mocks/**/*"
  - "**/*.test.js"
  - "**/*.spec.ts"

# 安全规则
security_rules:
  - no_hardcoded_secrets
  - no_sql_injection
  - no_xss_vulnerabilities
  - validate_external_input

# 架构规则
architecture_rules:
  - enforce_layer_separation
  - no_circular_dependencies
  - use_dependency_injection
  - follow_design_patterns

6. 集成到现有 CI/CD 流水线

将 AI 审查集成到现有的 CI/CD 流水线:

yaml
# .github/workflows/ci-cd.yml

name: CI/CD Pipeline

on:
  pull_request:
    branches: [main, develop]

jobs:
  # 1. 代码质量检查(确定性)
  code-quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run ESLint
        run: npm run lint

      - name: Run SonarQube
        uses: sonarsource/sonarqube-scan-action@master
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

  # 2. 单元测试
  unit-tests:
    runs-on: ubuntu-latest
    needs: code-quality
    steps:
      - uses: actions/checkout@v4

      - name: Run tests
        run: npm test

  # 3. AI 代码审查
  ai-review:
    runs-on: ubuntu-latest
    needs: code-quality
    steps:
      - uses: actions/checkout@v4

      - name: Run AI Code Review
        uses: ./actions/ai-review
        with:
          github_token: ${{ secrets.GITHUB_TOKEN }}
          openai_api_key: ${{ secrets.OPENAI_API_KEY }}
          config_file: "config/review-rules.yaml"

  # 4. 人工审查门禁
  manual-review:
    runs-on: ubuntu-latest
    needs: [unit-tests, ai-review]
    steps:
      - name: Require review
        uses: trstringer/manual-approval@v1
        with:
          secret: ${{ secrets.GITHUB_TOKEN }}
          approvers: senior-developers

与现有系统集成

1. 与 Slack/Teams 集成

发送审查结果到团队聊天工具:

python
# 通知服务

async def notify_review_results(pr: PullRequest, comments: List[ReviewComment]):
    """
    发送审查结果到 Slack
    """
    summary = build_review_summary(pr, comments)

    await slack_client.chat_postMessage(
        channel=SLACK_CHANNEL,
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*🔍 AI Code Review Results*\n\n{summary}"
                }
            }
        ]
    )

def build_review_summary(pr: PullRequest, comments: List[ReviewComment]) -> str:
    """
    构建审查摘要
    """
    critical_count = sum(1 for c in comments if c.severity == 'critical')
    major_count = sum(1 for c in comments if c.severity == 'major')

    summary = f"""
PR: {pr.title}
Author: {pr.author}
Files changed: {len(pr.changed_files)}

Issues found:
- Critical: {critical_count}
- Major: {major_count}
- Total: {len(comments)}

< {pr.url} | View PR >
"""
    return summary

2. 与 Jira/Linear 集成

将审查结果关联到需求跟踪系统:

python
# 问题跟踪集成

async def create_tickets_for_critical_issues(
    pr: PullRequest,
    comments: List[ReviewComment]
):
    """
    为关键问题创建 Jira 工单
    """
    critical_issues = [c for c in comments if c.severity == 'critical']

    for issue in critical_issues:
        jira_ticket = await jira_client.create_issue({
            'project': {'key': 'SECURITY'},
            'summary': f'Critical issue in PR {pr.number}: {issue.message}',
            'description': f"""
Location: {issue.file}:{issue.line}
Message: {issue.message}
Suggestion: {issue.suggestion}

PR: {pr.url}
""",
            'issuetype': {'name': 'Bug'},
            'priority': {'name': 'Highest'}
        })

        # 关联到 PR
        await add_pr_comment(pr, f"Created Jira ticket: {jira_ticket.key}")

关键配置参数

LLM 模型配置

python
# LLM 客户端配置

LLM_CONFIG = {
    "gpt-4": {
        "temperature": 0.3,  # 降低温度以获得更确定性的输出
        "max_tokens": 2000,
        "timeout": 60
    },
    "claude-3-5-sonnet": {
        "temperature": 0.3,
        "max_tokens": 2000,
        "timeout": 90
    }
}

# 审查器配置
REVIEWER_CONFIG = {
    "standard": {
        "model": "gpt-4",
        "max_comments": 10,
        "min_confidence": 0.7
    },
    "security": {
        "model": "claude-3-5-sonnet",
        "max_comments": 5,
        "min_confidence": 0.8
    }
}

性能优化配置

python
# 性能优化

async def review_with_caching(context: ReviewContext) -> List[ReviewComment]:
    """
    使用缓存优化性能
    """
    cache_key = generate_cache_key(context)

    # 检查缓存
    cached_result = await cache.get(cache_key)
    if cached_result:
        return cached_result

    # 执行审查
    comments = await run_ai_review(context)

    # 缓存结果
    await cache.set(cache_key, comments, ttl=3600)

    return comments

参考资料