基于LLM的多智能体PR审查系统设计与工程实践

在实际软件开发团队中代码审查Code Review是保障代码质量、统一编码风格和传播知识的关键环节。然而随着项目迭代加速和团队规模扩大人工审查的瓶颈日益凸显耗时、易受主观因素影响、难以覆盖所有细节。近年来基于大语言模型LLM的智能体AI Agent技术为自动化代码审查提供了新的可能性。一个设计良好的 AI 智能体不仅能检查语法错误还能理解代码意图、识别潜在的设计缺陷和安全漏洞。但构建一个真正可用的 AI PR 审查智能体远不止是调用一个 API 那么简单。它涉及到系统设计层面的诸多考量如何让多个智能体分工协作如何设计评估机制确保审查质量如何将智能体无缝集成到现有的开发工作流中本文将以构建一个多智能体 PR 审查系统为例深入探讨其系统设计。我们将从核心概念入手逐步完成环境准备、智能体角色定义、工作流编排、评估体系搭建最终实现一个可运行的原型。无论你是希望将 AI 引入现有开发流程的团队负责人还是对智能体系统设计感兴趣的后端工程师本文都将提供一条从理论到实践的清晰路径。1. 理解 AI 智能体与多智能体系统设计在开始动手之前我们需要厘清几个核心概念这决定了我们如何设计系统而不仅仅是拼接代码。1.1 什么是 AI 智能体一个 AI 智能体AI Agent通常指一个能够感知环境、自主决策并执行行动以实现特定目标的软件实体。在代码审查的上下文中这个“环境”就是 PRPull Request的代码变更、提交信息、相关 Issue 等上下文“决策”是分析代码并生成审查意见“行动”则是发布评论或给出批准/拒绝建议。与简单的代码分析工具不同一个真正的智能体应具备目标导向性其核心目标是提升代码质量而非仅仅运行静态检查。自主性给定 PR 上下文后能自主规划审查步骤如先看架构再查安全。反应性能根据代码的具体内容环境变化调整审查的重点。持续性可以与开发者进行多轮交互例如针对回复的评论进一步追问。1.2 为何需要多智能体单一智能体试图处理所有任务代码风格、逻辑错误、安全漏洞、性能问题、架构合理性往往会力不从心导致审查意见肤浅或遗漏重点。多智能体系统通过“分而治之”来解决这个问题专业化每个智能体专注于一个特定领域如安全专家、性能专家、架构守护者利用更精准的提示词Prompt和知识提供更深度的分析。协作与制衡智能体之间可以交换信息或进行“辩论”。例如架构智能体认为一个改动破坏了分层设计而实现智能体则认为这是必要的性能优化系统可以综合双方意见给出更平衡的建议。可扩展性新增审查维度如新增合规性检查只需引入新的智能体无需重构核心逻辑。1.3 系统设计的关键挑战设计这样一个系统我们需要解决几个工程挑战编排Orchestration如何有序地启动、调度多个智能体并管理它们之间的数据流上下文管理每个智能体需要哪些信息如 diff、文件全量内容、项目结构如何高效地提供并控制上下文长度评估与反馈Evals如何评估智能体生成的审查意见的质量如何利用反馈持续改进智能体这正是“demystifying evals for ai agents”要解决的核心问题。与开发流程集成如何以最小侵入的方式接入 GitHub/GitLab 等平台如何处理异步事件如 PR 更新后重新审查接下来我们将围绕这些挑战构建一个具体的系统。2. 环境准备与核心技术选型我们将使用 Python 作为主要开发语言因为它拥有最丰富的 AI 开发生态。以下是构建原型所需的核心组件。2.1 基础环境与依赖首先确保你的开发环境已就绪。# 推荐使用 Python 3.10 或以上版本 python --version # 创建并激活虚拟环境 python -m venv venv # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate # 安装核心依赖 pip install openai langchain langgraph github-webhook flask关键依赖说明openai: 用于调用 GPT 等模型。也可替换为anthropicClaude或本地模型客户端。langchain: 一个强大的框架用于简化基于 LLM 的应用程序开发特别是其LangGraph库非常适合构建多智能体工作流。langgraph: LangChain 的子库专门用于构建有状态、多参与者的图工作流是多智能体编排的理想选择。github-webhookflask: 用于接收 GitHub 的 PR 事件 Webhook构建一个简单的 HTTP 服务。2.2 项目结构规划一个清晰的项目结构有助于管理复杂度。multi-agent-pr-reviewer/ ├── agents/ # 智能体定义目录 │ ├── __init__.py │ ├── base_agent.py # 智能体基类 │ ├── security_agent.py # 安全审查智能体 │ ├── style_agent.py # 代码风格智能体 │ ├── logic_agent.py # 业务逻辑智能体 │ └── architect_agent.py # 架构审查智能体 ├── core/ # 核心逻辑目录 │ ├── __init__.py │ ├── orchestrator.py # 基于LangGraph的编排器 │ ├── context_builder.py # 构建智能体上下文的模块 │ └── models.py # 数据模型PR数据、审查意见等 ├── evals/ # 评估模块目录 │ ├── __init__.py │ └── evaluator.py # 评估智能体输出质量的模块 ├── webhook/ # Webhook服务目录 │ ├── __init__.py │ └── server.py # Flask应用处理GitHub事件 ├── config.py # 配置文件API密钥、模型设置 ├── requirements.txt # 项目依赖 └── README.md2.3 配置管理将敏感信息和可变配置外置是生产环境的基本要求。创建config.py# config.py import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Config: # LLM 配置 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_MODEL os.getenv(OPENAI_MODEL, gpt-4-turbo-preview) # 可根据情况选择 gpt-3.5-turbo # GitHub 配置 (用于主动获取仓库信息非必须) GITHUB_TOKEN os.getenv(GITHUB_TOKEN) # Webhook 配置 WEBHOOK_SECRET os.getenv(WEBHOOK_SECRET) # GitHub Webhook 的密钥用于验证请求 SERVER_PORT int(os.getenv(SERVER_PORT, 5000)) # 智能体配置 AGENT_TIMEOUT int(os.getenv(AGENT_TIMEOUT, 30)) # 每个智能体超时时间秒 # 评估配置 EVALUATION_ENABLED os.getenv(EVALUATION_ENABLED, False).lower() true在项目根目录创建.env文件切记将其加入.gitignoreOPENAI_API_KEYsk-your-openai-api-key-here OPENAI_MODELgpt-4-turbo-preview GITHUB_TOKENghp_your_github_token_here WEBHOOK_SECRETyour_webhook_secret_here SERVER_PORT50003. 构建核心多智能体编排与实现系统的核心是一个协调多个智能体工作的编排器。我们将使用LangGraph来实现一个基于图的工作流。3.1 定义数据模型首先在core/models.py中定义在整个系统中流转的核心数据结构。# core/models.py from typing import List, Dict, Any, Optional from pydantic import BaseModel class FileChange(BaseModel): 表示一个文件的变更 filename: str status: str # added, modified, removed patch: Optional[str] # Git diff 格式的补丁 full_content: Optional[str] # 变更后的文件完整内容可选 class PullRequestContext(BaseModel): PR 的完整上下文信息 repo_owner: str repo_name: str pr_number: int title: str description: str base_branch: str head_branch: str file_changes: List[FileChange] # 可以扩展更多字段如提交历史、关联Issue等 class ReviewComment(BaseModel): 单个审查意见 agent_name: str # 提出意见的智能体名称 file_path: str line_number: Optional[int] # 关联的行号 comment: str # 审查意见正文 category: str # 如 security, style, logic, architecture severity: str # 如 info, warning, error, critical class AgentResponse(BaseModel): 单个智能体的响应 agent_name: str comments: List[ReviewComment] summary: str # 该智能体的审查摘要 metadata: Dict[str, Any] {} # 额外元数据如耗时、token使用量 class ReviewSummary(BaseModel): 整个PR的审查总结 pr_context: PullRequestContext agent_responses: List[AgentResponse] overall_decision: str # APPROVE, REQUEST_CHANGES, COMMENT overall_summary: str3.2 实现智能体基类与具体智能体所有智能体都应遵循统一的接口。在agents/base_agent.py中定义基类。# agents/base_agent.py from abc import ABC, abstractmethod from core.models import PullRequestContext, AgentResponse from langchain.chat_models import ChatOpenAI from langchain.schema import SystemMessage, HumanMessage import logging logger logging.getLogger(__name__) class BaseReviewAgent(ABC): 审查智能体基类 def __init__(self, name: str, model_name: str None): self.name name self.llm ChatOpenAI( model_namemodel_name, temperature0.1, # 审查任务需要低随机性保持稳定 request_timeout30 ) self.system_prompt self._get_system_prompt() abstractmethod def _get_system_prompt(self) - str: 返回该智能体的系统提示词。子类必须实现。 pass abstractmethod async def review(self, pr_context: PullRequestContext) - AgentResponse: 执行审查的核心方法。 参数: pr_context: PR上下文信息 返回: AgentResponse 对象 pass def _format_messages(self, pr_context: PullRequestContext) - list: 将PR上下文格式化为LLM的消息列表。 # 这里可以构建一个更友好的提示例如只包含变更的文件内容 human_prompt f 请审查以下 Pull Request: 仓库: {pr_context.repo_owner}/{pr_context.repo_name} PR #{pr_context.pr_number}: {pr_context.title} 描述: {pr_context.description} 变更的文件: {self._format_file_changes(pr_context.file_changes)} return [ SystemMessage(contentself.system_prompt), HumanMessage(contenthuman_prompt) ] def _format_file_changes(self, file_changes): # 简化实现拼接文件名和变更状态 return \n.join([f- {fc.filename} ({fc.status}) for fc in file_changes])接下来实现一个具体的智能体例如安全审查智能体。# agents/security_agent.py from agents.base_agent import BaseReviewAgent from core.models import PullRequestContext, AgentResponse, ReviewComment import json class SecurityReviewAgent(BaseReviewAgent): 专注于安全漏洞的审查智能体 def __init__(self, model_name: str None): super().__init__(nameSecurityExpert, model_namemodel_name) def _get_system_prompt(self) - str: return 你是一个资深的安全工程师负责代码安全审查。你的任务是仔细分析代码变更识别潜在的安全漏洞。 重点检查以下方面 1. **注入攻击**SQL注入、命令注入、LDAP注入等。 2. **敏感信息泄露**硬编码的密钥、密码、API令牌错误的日志记录。 3. **身份验证与授权**缺失的访问控制、权限绕过、会话管理问题。 4. **不安全的反序列化**。 5. **使用已知存在漏洞的依赖库**如果上下文提供了依赖信息。 6. **跨站脚本XSS**、跨站请求伪造CSRF等Web安全漏洞。 请严格、专业地审查。对于每个发现的问题请提供 - 文件路径和行号如果可能。 - 问题的清晰描述。 - 潜在的风险等级CRITICAL, HIGH, MEDIUM, LOW, INFO。 - 具体的修复建议或代码示例。 如果未发现安全问题请明确说明“本次变更未发现明显安全问题”。 async def review(self, pr_context: PullRequestContext) - AgentResponse: messages self._format_messages(pr_context) try: response await self.llm.agenerate([messages]) content response.generations[0][0].text # 解析LLM的返回将其结构化为一组ReviewComment。 # 这是一个简化示例。更健壮的做法是让LLM以JSON格式输出或使用LangChain的输出解析器。 comments self._parse_response_to_comments(content, pr_context) return AgentResponse( agent_nameself.name, commentscomments, summaryf安全审查完成共发现 {len(comments)} 个问题。, metadata{raw_llm_output: content[:500]} # 记录部分原始输出用于调试 ) except Exception as e: logger.error(fSecurity agent failed: {e}) return AgentResponse( agent_nameself.name, comments[], summaryf安全审查执行失败: {e}, metadata{error: str(e)} ) def _parse_response_to_comments(self, content: str, pr_context) - list: # 这是一个非常简单的解析器。实际应用中你需要设计更稳定的解析逻辑 # 例如要求LLM固定以JSON格式输出或使用正则表达式匹配。 comments [] # 假设LLM的回复中每个问题以“- [文件]行号: 描述”的格式列出 lines content.split(\n) for line in lines: if line.strip().startswith(-): # 简化的解析逻辑实际项目需要更健壮 comment ReviewComment( agent_nameself.name, file_path待解析, # 应从line中解析 line_numberNone, commentline.strip(), categorysecurity, severitywarning # 应从描述中判断 ) comments.append(comment) return comments类似地你可以创建StyleReviewAgent检查代码风格、命名规范、LogicReviewAgent检查业务逻辑错误、边界条件、ArchitectReviewAgent检查架构一致性、设计模式违反等。3.3 使用 LangGraph 构建编排器LangGraph允许我们将工作流定义为一个有向图节点是函数智能体或处理逻辑边定义了执行顺序。在core/orchestrator.py中实现编排器。# core/orchestrator.py from typing import Dict, Any, List from langgraph.graph import StateGraph, END from core.models import PullRequestContext, ReviewSummary, AgentResponse from agents.security_agent import SecurityReviewAgent from agents.style_agent import StyleReviewAgent from agents.logic_agent import LogicReviewAgent import asyncio import logging logger logging.getLogger(__name__) # 定义图的状态结构。这里我们用一个字典来传递状态。 class ReviewState(Dict[str, Any]): LangGraph 工作流的状态定义 pr_context: PullRequestContext agent_responses: List[AgentResponse] final_summary: ReviewSummary None class MultiAgentOrchestrator: def __init__(self): self.agents { security: SecurityReviewAgent(), style: StyleReviewAgent(), logic: LogicReviewAgent(), # 可以动态添加更多智能体 } self.graph self._build_graph() def _build_graph(self): 构建 LangGraph 工作流图 workflow StateGraph(ReviewState) # 添加节点每个节点对应一个智能体的审查任务 for agent_name in self.agents.keys(): workflow.add_node(agent_name, self._run_agent_task) # 定义入口点并行执行所有智能体 # 这里采用并行执行适用于智能体间无依赖的场景。 # 如果智能体间有依赖例如架构智能体先分析逻辑智能体再深入则需要设计更复杂的边。 workflow.add_edge(security, END) workflow.add_edge(style, END) workflow.add_edge(logic, END) # 注意当所有节点都指向END时LangGraph会并行执行它们。 # 我们需要一个“入口”节点来触发并行。 # 更准确的做法是使用一个起始节点然后分支到各个智能体。 # 重新构建使用一个起始节点然后分支。 workflow StateGraph(ReviewState) # 起始节点准备数据 workflow.add_node(start, self._prepare_context) for agent_name in self.agents.keys(): workflow.add_node(agent_name, self._run_agent_task) workflow.add_node(aggregate, self._aggregate_results) # 设置边start - 所有智能体并行 - aggregate - END workflow.set_entry_point(start) for agent_name in self.agents.keys(): workflow.add_edge(start, agent_name) # 所有智能体完成后都指向 aggregate 节点 for agent_name in self.agents.keys(): workflow.add_edge(agent_name, aggregate) workflow.add_edge(aggregate, END) # 设置汇聚条件当所有智能体节点都完成后才执行 aggregate。 # LangGraph 默认是流式一个节点完成后就触发下一个。 # 我们需要在 aggregate 节点上定义触发器或者使用更高级的构造。 # 简化处理这里假设智能体执行很快顺序执行。生产环境应考虑真正的并行和等待。 # 下面改为更简单的线性流程用于演示。 return self._build_linear_graph() # 回退到线性流程以简化演示 def _build_linear_graph(self): 构建一个简单的线性执行图依次执行 workflow StateGraph(ReviewState) agent_names list(self.agents.keys()) if not agent_names: workflow.add_node(noop, lambda state: state) workflow.set_entry_point(noop) workflow.add_edge(noop, END) return workflow.compile() # 添加节点 workflow.add_node(start, self._prepare_context) for name in agent_names: workflow.add_node(name, self._run_agent_task) workflow.add_node(aggregate, self._aggregate_results) # 设置线性边: start - agent1 - agent2 - ... - aggregate - END workflow.set_entry_point(start) workflow.add_edge(start, agent_names[0]) for i in range(len(agent_names) - 1): workflow.add_edge(agent_names[i], agent_names[i 1]) workflow.add_edge(agent_names[-1], aggregate) workflow.add_edge(aggregate, END) return workflow.compile() async def _prepare_context(self, state: ReviewState): 起始节点可以在这里进行一些上下文预处理如获取文件全量内容 logger.info(fPreparing context for PR #{state[pr_context].pr_number}) # 这里可以添加逻辑例如根据 patch 获取文件完整内容 return state async def _run_agent_task(self, state: ReviewState, node_name: str): 运行单个智能体任务 agent self.agents.get(node_name) if not agent: logger.error(fAgent {node_name} not found.) return state logger.info(fRunning agent: {node_name}) try: response await agent.review(state[pr_context]) if agent_responses not in state: state[agent_responses] [] state[agent_responses].append(response) except Exception as e: logger.error(fAgent {node_name} execution failed: {e}) error_response AgentResponse( agent_namenode_name, comments[], summaryf执行失败: {e}, metadata{error: str(e)} ) state[agent_responses].append(error_response) return state async def _aggregate_results(self, state: ReviewState): 汇总所有智能体的结果生成最终审查总结 responses state.get(agent_responses, []) all_comments [] for resp in responses: all_comments.extend(resp.comments) # 简单的决策逻辑如果任何智能体提出了严重critical问题则请求修改 # 否则如果有警告warning级别问题则发表评论。否则批准。 overall_decision APPROVE for comment in all_comments: if comment.severity critical: overall_decision REQUEST_CHANGES break if comment.severity warning and overall_decision APPROVE: overall_decision COMMENT summary_text f本次 PR 由 {len(self.agents)} 个智能体完成审查。共生成 {len(all_comments)} 条意见。 if overall_decision REQUEST_CHANGES: summary_text 存在严重问题需要修改。 elif overall_decision COMMENT: summary_text 存在一些建议请查看。 else: summary_text 未发现重大问题建议批准。 final_summary ReviewSummary( pr_contextstate[pr_context], agent_responsesresponses, overall_decisionoverall_decision, overall_summarysummary_text ) state[final_summary] final_summary logger.info(fAggregation complete. Decision: {overall_decision}) return state async def orchestrate_review(self, pr_context: PullRequestContext) - ReviewSummary: 主入口编排整个审查流程 initial_state ReviewState({ pr_context: pr_context, agent_responses: [] }) # 执行图工作流 final_state await self.graph.ainvoke(initial_state) return final_state.get(final_summary)4. 集成与运行连接 GitHub Webhook为了让系统响应 GitHub PR 事件我们需要创建一个简单的 Webhook 服务器。4.1 创建 Webhook 服务器在webhook/server.py中创建 Flask 应用。# webhook/server.py from flask import Flask, request, jsonify import hmac import hashlib import logging from core.orchestrator import MultiAgentOrchestrator from core.models import PullRequestContext, FileChange import asyncio import json app Flask(__name__) logger logging.getLogger(__name__) orchestrator MultiAgentOrchestrator() def verify_webhook_signature(data, signature): 验证 GitHub Webhook 签名 webhook_secret bytes(app.config[WEBHOOK_SECRET], utf-8) mac hmac.new(webhook_secret, msgdata, digestmodhashlib.sha256) expected_signature sha256 mac.hexdigest() return hmac.compare_digest(expected_signature, signature) app.route(/webhook, methods[POST]) def handle_webhook(): 处理 GitHub Webhook 请求 signature request.headers.get(X-Hub-Signature-256) if not signature: logger.warning(Missing X-Hub-Signature-256 header) return jsonify({error: Missing signature}), 401 if not verify_webhook_signature(request.data, signature): logger.warning(Invalid webhook signature) return jsonify({error: Invalid signature}), 401 event_type request.headers.get(X-GitHub-Event) payload request.json # 只处理 PR 相关事件 if event_type pull_request: action payload.get(action) # 当 PR 被打开、同步新提交或重新打开时触发审查 if action in [opened, synchronize, reopened]: logger.info(fProcessing PR event: {action}) # 在新线程或任务中运行异步审查避免阻塞Webhook响应 asyncio.run(process_pull_request(payload)) return jsonify({status: review triggered}), 202 else: logger.info(fIgnoring PR action: {action}) return jsonify({status: ignored}), 200 else: logger.info(fIgnoring event type: {event_type}) return jsonify({status: ignored}), 200 async def process_pull_request(payload): 异步处理 PR 审查 try: pr_info payload[pull_request] repo_info payload[repository] # 构建 PullRequestContext # 注意GitHub Webhook 的 payload 中不直接包含文件 diff。 # 我们需要使用 GitHub API 主动获取 diff。 # 此处为简化假设我们从 payload 中提取了必要信息。 # 实际项目中你需要调用 GitHub API 获取文件变更列表。 file_changes [] # 示例从 webhook 的 commits 等信息中模拟构建 file_changes # 这里需要调用 GitHub API 获取详细的 files 信息 # 伪代码: files github_api.get_pr_files(...) # for file in files: # file_changes.append(FileChange(filenamefile[filename], statusfile[status], patchfile.get(patch))) # 为了演示我们创建一个模拟的 FileChange sample_change FileChange( filenamesrc/main.py, statusmodified, patch -10,7 10,7 def get_user_input():\n user_input input(\Enter your name: \)\n # 模拟一个潜在的安全问题直接拼接 SQL\n- query f\SELECT * FROM users WHERE name {user_input}\\n query \SELECT * FROM users WHERE name %s\\n return query ) file_changes [sample_change] pr_context PullRequestContext( repo_ownerrepo_info[owner][login], repo_namerepo_info[name], pr_numberpr_info[number], titlepr_info[title], descriptionpr_info[body] or , base_branchpr_info[base][ref], head_branchpr_info[head][ref], file_changesfile_changes ) logger.info(fStarting review for PR #{pr_context.pr_number}) review_summary await orchestrator.orchestrate_review(pr_context) # 将审查结果发布回 GitHub PR await post_review_to_github(pr_context, review_summary) logger.info(fReview completed for PR #{pr_context.pr_number}. Decision: {review_summary.overall_decision}) except Exception as e: logger.error(fFailed to process pull request: {e}) async def post_review_to_github(pr_context: PullRequestContext, review_summary: ReviewSummary): 将审查结果以评论形式发布到 GitHub PR # 这里需要实现 GitHub API 调用 # 使用 GITHUB_TOKEN 进行认证 # 构建评论体汇总所有智能体的意见 comment_body f## AI PR 审查报告\n\n**总体建议: {review_summary.overall_decision}**\n\n{review_summary.overall_summary}\n\n---\n for agent_resp in review_summary.agent_responses: comment_body f\n### {agent_resp.agent_name}\n{agent_resp.summary}\n for comment in agent_resp.comments: comment_body f- **{comment.file_path}** (L{comment.line_number}): {comment.comment}\n # 伪代码: github_api.create_pr_review(...) logger.info(fGenerated review comment:\n{comment_body[:200]}...) # 实际调用示例需要安装 PyGithub: # from github import Github # g Github(app.config[GITHUB_TOKEN]) # repo g.get_repo(f{pr_context.repo_owner}/{pr_context.repo_name}) # pull repo.get_pull(pr_context.pr_number) # pull.create_review(bodycomment_body, eventreview_summary.overall_decision.lower()) if __name__ __main__: # 从 config 加载配置 from config import Config app.config[WEBHOOK_SECRET] Config.WEBHOOK_SECRET # 生产环境应使用 Gunicorn 等 WSGI 服务器 app.run(host0.0.0.0, portConfig.SERVER_PORT, debugFalse)4.2 运行与测试启动 Webhook 服务器cd /path/to/your/project python -m webhook.server服务器将在http://localhost:5000启动监听/webhook路径。配置 GitHub Webhook进入你的 GitHub 仓库的Settings-Webhooks-Add webhook。Payload URL: 填写你的服务器公网地址如https://your-domain.com/webhook。Content type: 选择application/json。Secret: 填写你在.env中设置的WEBHOOK_SECRET。Which events...: 选择Let me select individual events然后勾选Pull requests。点击Add webhook。触发测试在你的仓库创建一个新的 Pull Request。观察你的服务器日志应该能看到处理请求的记录。检查 PR 页面应该能看到由你的 AI 智能体发布的审查评论。5. 评估、排错与最佳实践一个没有评估和迭代的 AI 系统是不可靠的。同时在实际部署中会遇到各种问题。5.1 构建评估体系评估Evals的目标是量化智能体审查意见的质量以便持续改进。在evals/evaluator.py中创建一个简单的评估模块。# evals/evaluator.py from core.models import ReviewSummary, ReviewComment from typing import List, Tuple import asyncio class ReviewEvaluator: 评估智能体审查意见的质量 async def evaluate_comment_relevance(self, comment: ReviewComment, ground_truth: List[str]) - Tuple[float, str]: 评估单条评论的相关性。 ground_truth: 人工标注的该代码片段真实存在的问题列表。 返回: (得分, 理由) # 简化评估使用另一个LLM作为裁判判断评论是否指出了ground_truth中的问题。 # 生产环境需要更严谨的评估集和指标。 evaluation_prompt f 你是一个代码审查评估员。请判断以下AI生成的审查意见是否准确指出了代码中真实存在的问题。 代码变更上下文简化: [此处应注入代码片段] AI审查意见: {comment.comment} 人工标注的真实问题列表: {ground_truth} 请从以下方面评估 1. **相关性**意见是否针对代码变更是/否 2. **准确性**指出的问题是否真实存在且描述正确是/部分/否 3. **严重性**对问题严重性的判断是否合理高估/合理/低估 请给出一个综合得分0-10分并简要说明理由。 # 调用LLM进行评估... # score, reasoning await self._llm_judge(evaluation_prompt) # return score, reasoning return 7.5, 意见相关且准确但严重性判断稍显保守。 # 模拟返回 async def evaluate_agent_performance(self, summary: ReviewSummary, human_review_result) - dict: 评估整个智能体在本次PR审查中的表现。 human_review_result: 人工审查的结果可包含确认的有效意见、误报、漏报等。 # 计算精确率、召回率、F1分数等 # 精确率 AI正确指出的问题 / AI提出的所有意见 # 召回率 AI正确指出的问题 / 人工发现的所有问题 metrics { precision: 0.0, recall: 0.0, f1: 0.0, false_positive_count: 0, false_negative_count: 0, } # ... 实现具体的对比逻辑 return metrics def generate_evaluation_report(self, metrics_history: List[dict]) - str: 根据历史评估数据生成报告用于指导提示词或智能体策略的优化。 # 分析趋势找出表现不佳的智能体或问题类型 report 智能体性能评估报告\n report *50 \n # ... 生成报告逻辑 return report评估数据的收集人工标注定期抽样 PR由资深开发者进行人工审查将结果作为ground_truth。A/B测试将新旧版本的智能体例如不同提示词应用于相同的 PR 集对比结果。开发者反馈在 AI 评论旁添加“有用/无用”按钮收集直接反馈。5.2 常见问题与排查路径在开发和运行过程中你可能会遇到以下问题问题现象可能原因检查方式处理建议Webhook 请求失败返回 401签名验证失败1. 检查WEBHOOK_SECRET环境变量是否与 GitHub 配置一致。2. 检查服务器时间是否与 NTP 同步影响 HMAC。重新生成并配置 Secret。确保服务器时间准确。智能体无输出或输出混乱1. LLM API 调用失败。2. 提示词Prompt设计不佳。3. 上下文过长被截断。1. 查看日志中 LLM 调用的错误信息。2. 检查agent_responses中的metadata字段查看原始 LLM 输出。3. 计算并打印发送给 LLM 的 Token 数量。1. 检查 API 密钥、网络、配额。2. 迭代优化提示词使其指令更清晰。3. 精简上下文只发送必要的代码 diff。审查意见不准确或遗漏1. 智能体专业领域知识不足。2. 缺少项目特定上下文如架构图、API 文档。3. 代码变更过于复杂。1. 分析评估报告看哪个智能体或哪类问题表现差。2. 检查提供给智能体的上下文是否完整。1. 在提示词中加入项目特定的编码规范、架构约束。2. 考虑使用 RAG检索增强生成为智能体提供项目文档。3. 将大 PR 拆分成更小的变更集进行审查。系统响应缓慢1. 智能体串行执行。2. LLM API 调用延迟高。3. 获取 GitHub 文件内容耗时。1. 记录每个智能体和步骤的耗时。2. 监控网络延迟和 LLM 服务状态。1. 将编排器改为真正的并行执行使用asyncio.gather。2. 为 LLM 调用设置合理的超时和重试机制。3. 缓存 GitHub 仓库的某些元数据。无法发布评论到 GitHub1. GitHub Token 权限不足。2. API 调用频率超限。3. PR 已关闭或合并。1. 检查 Token 的 scope 是否包含repo或pull_requests。2. 查看 GitHub API 返回的错误信息。1. 生成具有足够权限的 Fine-grained Token 或 Classic Token。2. 实现简单的请求队列和重试逻辑。3. 在发布前检查 PR 状态。5.3 生产环境最佳实践提示词工程分而治之为每个智能体设计高度专业化、具体的提示词而不是一个通用提示词。提供示例在提示词中包含少量高质量审查意见的示例Few-shot Learning。输出格式化严格要求 LLM 以指定格式如 JSON输出便于后续解析。可以使用 LangChain 的OutputParser。持续迭代建立评估闭环根据评估结果不断优化提示词。系统健壮性错误处理与降级单个智能体失败不应导致整个系统崩溃。记录错误并提供降级结果如“安全审查暂不可用”。速率限制与重试对 LLM API 和 GitHub API 的调用实施速率限制和指数退避重试。异步与队列使用消息队列如 Redis、RabbitMQ处理 Webhook 事件实现异步、解耦的处理避免 HTTP 超时。可观测性记录详细的日志包括每个智能体的输入、输出、耗时和 Token 使用量。集成监控和告警。安全与权限最小权限原则GitHub Token 只授予必要的仓库和权限如只读访问代码、写评论。代码隔离在安全的沙箱环境中处理不可信的代码 diff尽管 PR 代码通常来自协作成员但仍需谨慎。敏感信息过滤确保智能体的输出不会意外泄露在训练数据中见过的敏感信息。人机协作明确标识AI 评论应清晰标明由哪个智能体生成并说明其局限性。提供上下文评论应引用具体的代码行并尽可能解释“为什么”这是个问题。支持交互允许开发者对 AI 评论进行反馈“有用”/“无用”这些反馈是优化系统最重要的数据。作为辅助而非替代明确告知团队AI 审查是辅助工具最终责任仍在开发者身上。构建一个有效的多智能体 PR 审查系统是一个持续迭代的过程。从今天构建的最小可行原型出发你可以逐步引入更专业的智能体、更强大的编排逻辑、基于 RAG 的项目知识库以及自动化的评估流水线。关键在于建立“执行-评估-优化”的闭环让系统在实际的代码审查场景中不断学习与进化最终成为一个真正能提升团队研发效能与代码质量的智能伙伴。