第 31 章：实战：代码审查 Agent 流水线

需求/PR 描述 → 代码检索 → 多专家并行审查 → 聚合报告 → 人做最终决策

class CodeReviewPipeline:
    def __init__(self, retriever, reviewers, aggregator):
        self.retriever = retriever
        self.reviewers = reviewers
        self.aggregator = aggregator

    def run(self, pr_description: str, repo: str, base: str, head: str) -> dict:
        context = self.retriever.fetch(pr_description, repo, base, head)
        reviews = [r.review(context) for r in self.reviewers]
        return self.aggregator.aggregate(reviews, context)

def parse_pr_description(text: str) -> dict:
    prompt = f"""从以下 PR 描述中提取：
1. 变更目标
2. 影响范围
3. 需要重点审查的方面（安全、性能、兼容性等）

PR 描述：{text}
返回 JSON。"""
    return json.loads(llm_call(prompt))

class GitRetriever:
    def fetch(self, repo: str, base: str, head: str) -> list[str]:
        diff = subprocess.check_output(
            ["git", "-C", repo, "diff", f"{base}...{head}"],
            text=True,
        )
        return chunk_by_file(diff)

    def get_callers(self, repo: str, symbol: str) -> list[str]:
        # 简化：用 grep 近似，真实项目用 AST/索引
        return grep_symbol(repo, symbol)

class SecurityReviewer:
    def review(self, context: ReviewContext) -> Review:
        prompt = f"""你是一名安全专家，审查以下代码变更：
{context.diff}

输出 JSON：{{ "score": 1-10, "issues": [{{ "file", "line", "severity", "description", "suggestion" }}] }}"""
        return Review.parse(llm_call(prompt))

class StyleReviewer:
    def review(self, context: ReviewContext) -> Review:
        # 可结合 linter（flake8, eslint）结果，LLM 只做汇总
        lint_output = run_linter(context.files)
        return Review.parse(summarize_lint(lint_output))

class LogicReviewer:
    def review(self, context: ReviewContext) -> Review:
        prompt = f"""审查以下代码的业务逻辑正确性：...
{context.diff}"""
        return Review.parse(llm_call(prompt))

class ReportAggregator:
    def aggregate(self, reviews: list[Review], context: ReviewContext) -> dict:
        all_issues = []
        for r in reviews:
            all_issues.extend(r.issues)
        # 去重：按文件+行号+描述哈希
        seen = set()
        unique = []
        for issue in all_issues:
            key = (issue.file, issue.line, issue.description)
            if key not in seen:
                seen.add(key)
                unique.append(issue)
        unique.sort(key=lambda x: (x.severity != "high", x.severity != "medium"))
        return {
            "summary": self.summarize(unique),
            "issues": unique,
            "scores": {r.reviewer: r.score for r in reviews},
        }

async function styleReview(files: string[]): Promise<Issue[]> {
  const lint = await runESLint(files);
  const summary = await llmSummarize(lint);
  return summary.issues;
}

func runGoVet(paths []string) ([]byte, error) {
    cmd := exec.Command("go", append([]string{"vet"}, paths...)...)
    return cmd.CombinedOutput()
}