AI项目实战（6）代码审查 Agent

"""P03 代码审查 Agent - 领域模型层"""
from __future__ import annotations

import json
from dataclasses import asdict, dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class Severity(str, Enum):
    """问题严重程度枚举。"""
    HIGH = "高危"
    MEDIUM = "中危"
    LOW = "低危"


class ReviewDimension(str, Enum):
    """审查维度枚举。"""
    SECURITY = "security"
    PERFORMANCE = "performance"
    STYLE = "style"


@dataclass
class ReviewIssue:
    """单个审查问题条目。"""
    file_path: str
    line_number: int | None = None
    severity: Severity = Severity.LOW
    dimension: ReviewDimension = ReviewDimension.STYLE
    description: str = ""
    suggestion: str = ""
    code_snippet: str = ""

    def validate(self) -> list[str]:
        """验证字段合法性，返回错误列表。"""
        errors: list[str] = []
        if not self.file_path:
            errors.append("file_path 不能为空")
        if not self.description:
            errors.append("description 不能为空")
        return errors

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["severity"] = self.severity.value
        data["dimension"] = self.dimension.value
        return data


@dataclass
class ReviewReport:
    """完整的审查报告。"""
    target_files: list[str] = field(default_factory=list)
    issues: list[ReviewIssue] = field(default_factory=list)
    score: int = 10
    reviewed_at: str = field(default_factory=lambda: datetime.now().isoformat())
    duration_seconds: float = 0.0

    @property
    def issue_count(self) -> dict[str, int]:
        counts: dict[str, int] = {"高危": 0, "中危": 0, "低危": 0}
        for issue in self.issues:
            counts[issue.severity.value] += 1
        return counts

    def calculate_score(self) -> int:
        base = 10
        for issue in self.issues:
            if issue.severity == Severity.HIGH:
                base -= 2
            elif issue.severity == Severity.MEDIUM:
                base -= 1
        return max(1, min(10, int(round(base))))


class ReviewSkill:
    """审查 Skill 基类。"""
    name: str = ""
    dimension: ReviewDimension = ReviewDimension.STYLE
    prompt_template: str = ""


class SecurityReviewSkill(ReviewSkill):
    """安全审查 Skill。"""
    name = "security"
    dimension = ReviewDimension.SECURITY
    prompt_template = """你是安全审计专家。审查以下代码的安全问题：
- SQL 注入（拼接 SQL 字符串）
- XSS 漏洞（直接输出用户输入未转义）
- 硬编码的密钥/密码/Token
- 命令注入（os.system / subprocess 拼接用户输入）
对每个发现的问题标注严重程度，并提供可直接执行的修复代码。"""


class PerformanceReviewSkill(ReviewSkill):
    """性能审查 Skill。"""
    name = "performance"
    dimension = ReviewDimension.PERFORMANCE
    prompt_template = """你是性能优化专家。审查以下代码的性能问题：
- N+1 查询（循环内执行数据库查询）
- 不必要的 O(n) 操作（可用 O(1) 替代）
- 大对象未释放（内存泄漏风险）
- 可并行化的串行操作"""


class StyleReviewSkill(ReviewSkill):
    """编码规范审查 Skill。"""
    name = "style"
    dimension = ReviewDimension.STYLE
    prompt_template = """你是代码规范专家。审查以下代码的规范问题：
- 函数/变量命名是否符合 PEP 8（snake_case）
- 缺少类型注解（函数参数和返回值）
- 函数过长（超过 50 行建议拆分）
- 缺少 docstring（模块、类、公共方法）
- 过深的嵌套层级（超过 3 层）"""


REVIEW_SKILLS: list[ReviewSkill] = [
    SecurityReviewSkill(),
    PerformanceReviewSkill(),
    StyleReviewSkill(),
]

"""P03 代码审查 Agent - Prompt 层"""
from __future__ import annotations

from .models import REVIEW_SKILLS, ReviewSkill

AGENT_SYSTEM_PROMPT = """你是高级代码审查 Agent，负责对代码变更进行全面审查。

## 核心职责
1. 读取待审查的文件内容
2. 从安全、性能、规范三个维度逐一审查
3. 输出结构化的审查报告（JSON 格式）

## 输出格式要求
必须输出 JSON 格式的审查报告，结构如下：
```json
{{
  "target_files": ["path/to/file.py"],
  "issues": [
    {{
      "file_path": "path/to/file.py",
      "line_number": 42,
      "severity": "高危",
      "dimension": "security",
      "description": "问题描述",
      "suggestion": "修复建议"
    }}
  ],
  "score": 8
}}

**核心代码讲解**：
- **集中管理**：所有提示词收在一个文件里，就像把老王的口头规矩统一抄到一本手册上，便于维护和版本控制
- **动态构建**：`build_system_prompt()` 自动遍历已注册的 Skill，挨个拼出维度说明——新增 Skill 时，提示词会自动跟上，不用手动改
- **格式契约**：白纸黑字写死了 JSON 输出结构，确保下游解析时不会因为格式漂移而炸场

---

### 5.4.4 tools.py - 工具层

```python
"""P03 代码审查 Agent - 工具层"""
from __future__ import annotations

import subprocess
from pathlib import Path

from langchain.tools import tool

from core.logging_conf import get_logger

logger = get_logger("p03.code_review.tools")
MAX_FILE_SIZE = 8000


@tool
def read_file(file_path: str) -> str:
    """读取指定文件的全部内容。"""
    logger.info("读取文件: %s", file_path)
    try:
        path = Path(file_path)
        if not path.exists():
            return f"文件不存在: {file_path}"
        content = path.read_text(encoding="utf-8")
        if len(content) > MAX_FILE_SIZE:
            content = content[:MAX_FILE_SIZE] + f"\n... (已截断至 {MAX_FILE_SIZE} 字符)"
        return content
    except Exception as e:
        logger.exception("读取文件失败: %s", e)
        return f"读取文件失败: {type(e).__name__}: {e}"


@tool
def grep_code(pattern: str, directory: str = ".") -> str:
    """在指定目录中搜索代码模式（正则表达式）。"""
    logger.info("代码搜索: pattern='%s'", pattern)
    try:
        result = subprocess.run(
            ["grep", "-rn", "--include=*.py", "-n", pattern, directory],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if not result.stdout.strip():
            return f"未找到匹配 '{pattern}' 的代码"
        return result.stdout
    except Exception as e:
        logger.exception("搜索失败: %s", e)
        return f"搜索失败: {type(e).__name__}: {e}"


@tool
def run_tests(repo_path: str = ".") -> str:
    """在指定目录运行 Python 测试套件。"""
    logger.info("运行测试: repo_path=%s", repo_path)
    try:
        result = subprocess.run(
            ["python", "-m", "pytest", "-x", "--tb=short", "-q"],
            capture_output=True,
            text=True,
            timeout=60,
            cwd=repo_path,
        )
        return result.stdout or result.stderr
    except Exception as e:
        logger.exception("测试运行失败: %s", e)
        return f"测试运行失败: {type(e).__name__}: {e}"

"""P03 代码审查 Agent - 服务层"""
from __future__ import annotations

import time

from core.logging_conf import get_logger

from .models import REVIEW_SKILLS, ReviewIssue, ReviewReport, Severity

logger = get_logger("p03.code_review.service")


class CodeReviewService:
    """代码审查服务。"""

    def __init__(self, skills=None):
        self.skills = skills or REVIEW_SKILLS
        logger.info("代码审查服务初始化完成，加载 %d 个审查 Skill", len(self.skills))

    def review_file(self, file_path: str, run_grep: bool = True) -> ReviewReport:
        """审查单个文件。"""
        start_time = time.time()
        logger.info("开始审查文件: %s", file_path)

        report = ReviewReport(target_files=[file_path])

        # 静态规则检查
        static_issues = self._static_rule_check(file_path)
        report.issues.extend(static_issues)

        report.score = report.calculate_score()
        report.duration_seconds = time.time() - start_time
        return report

    def _static_rule_check(self, file_path: str, content: str) -> list[ReviewIssue]:
        """静态规则检查（简单的字符串模式匹配）。"""
        issues: list[ReviewIssue] = []
        lines = content.split("\n")

        for line_no, line in enumerate(lines, 1):
            if "password" in line.lower() and "=" in line and "os.getenv" not in line:
                issues.append(ReviewIssue(
                    file_path=file_path,
                    line_number=line_no,
                    severity=Severity.HIGH,
                    description="可能存在硬编码密码",
                    suggestion="使用 os.getenv('PASSWORD') 从环境变量读取",
                    code_snippet=line.strip(),
                ))
        return issues

"""P03 代码审查 Agent - 项目入口层"""
from __future__ import annotations

from langchain.agents import create_agent

from core import BaseProject, build_chat_model, registry
from core.logging_conf import get_logger

from .prompts import build_system_prompt
from .tools import get_available_tools

logger = get_logger("p03.code_review.project")


class CodeReviewProject(BaseProject):
    """代码审查 Agent 项目。"""

    id = "p03_code_review"
    name = "代码审查 Agent"
    description = "自动审查代码安全、性能、规范，输出结构化报告。"
    capabilities = ["Harness", "Skill", "代码审查"]

    def build_agent(self) -> any:
        """构建代码审查 Agent。"""
        logger.info("正在构建代码审查 Agent...")
        model = build_chat_model()
        tools = get_available_tools()
        system_prompt = build_system_prompt()
        return create_agent(model=model, tools=tools, system_prompt=system_prompt)


registry.register(CodeReviewProject())

"""项目三：代码审查 Agent（生产级 Harness + Skill）"""
from __future__ import annotations

# 领域模型
from .models import (
    REVIEW_SKILLS,
    PerformanceReviewSkill,
    ReviewIssue,
    ReviewReport,
    ReviewSkill,
    SecurityReviewSkill,
    Severity,
    StyleReviewSkill,
)

# 提示词
from .prompts import build_system_prompt

# 工具
from .tools import grep_code, read_file, run_tests

# 服务
from .service import CodeReviewService

# 项目入口
from .project import CodeReviewProject

# 向后兼容
SYSTEM_PROMPT = build_system_prompt()

__all__ = [
    "Severity", "ReviewIssue", "ReviewReport", "ReviewSkill",
    "SecurityReviewSkill", "PerformanceReviewSkill", "StyleReviewSkill",
    "REVIEW_SKILLS", "build_system_prompt", "SYSTEM_PROMPT",
    "read_file", "grep_code", "run_tests",
    "CodeReviewService", "CodeReviewProject",
]