Multi-Agent代码审查系统实战教程

Multi-Agent 代码审查系统：从零到一实战教程用多个 AI Agent 模拟真实代码审查团队——安全专家找漏洞、架构师评设计、性能工程师查瓶颈，最终输出一份专业的 Code Review 报告。前言你将学到什么理解 Multi-Agent 协作的核心原理用CrewAI搭建多角色协作的代码审查系统用LangGraph构建图状工作流的审查流水线集成GitHub API，实现 PR 自动审查搭建Web 界面，可视化审查结果部署上线，接入 CI/CD 流程技术栈组件选型说明Agent 框架CrewAI + LangGraph两种方案对比实现大模型DeepSeek V3便宜好用，支持 Function Calling代码托管GitHub API自动获取 PR 代码差异Web 框架FastAPI后端 API 服务前端Vue 3 + Element Plus可视化审查结果部署Docker + Docker Compose一键部署项目结构预览code-review-agent/ ├── agents/ # Agent 定义 │ ├── security_agent.py # 安全审查 Agent │ ├── architecture_agent.py # 架构审查 Agent │ ├── performance_agent.py # 性能审查 Agent │ └── summary_agent.py # 汇总 Agent ├── tools/ # 工具定义 │ ├── code_parser.py # 代码解析工具 │ ├── github_tool.py # GitHub API 工具 │ └── static_analysis.py # 静态分析工具 ├── workflows/ # 工作流 │ ├── crew_workflow.py # CrewAI 版工作流 │ └── graph_workflow.py # LangGraph 版工作流 ├── api/ # FastAPI 后端 │ ├── main.py │ ├── routes.py │ └── models.py ├── frontend/ # Vue 3 前端 ├── config.py # 配置文件 ├── .env # 环境变量 ├── requirements.txt ├── Dockerfile └── docker-compose.yml目录索引：第一章： 概述与架构设计 — 为什么用 Multi-Agent、系统架构图、5 个 Agent 角色定义、协作模式第二章： 环境搭建 — Python 环境、依赖安装、环境变量配置、GitHub Token 生成、验证脚本第三章： 基础工具层 — 代码解析器、GitHub API 工具、静态分析（Pylint/Bandit/Radon）、配置文件第四章： CrewAI 多 Agent 协作 — 5 个 Agent 角色定义、审查任务定义、Crew 组装、运行入口第五章： LangGraph 工作流 — State 定义、6 个节点函数、图构建、条件分支、流程可视化第六章： GitHub 集成 — PR 审查 API、Webhook 自动触发、GitHub Actions CI/CD第七章： Web 前端界面 — Vue 3 + Element Plus、代码/PR 输入、Markdown 报告渲染第八章： 进阶优化 — 结构化输出、审查缓存、并行执行、质量评估、Token 优化第九章： 部署上线 — Dockerfile、Docker Compose、Nginx 反向代理、Celery 任务队列附录： 常见问题排查、扩展方向、完整项目文件清单第一章：概述与架构设计1.1 为什么用 Multi-Agent 做代码审查？单 Agent 的局限：一个 Agent 身兼数职，容易顾此失彼——既要找安全漏洞，又要评架构设计，还要查性能问题，结果哪个都做不好。Multi-Agent 的优势：单 Agent： 一个全能选手 → 样样通，样样松 Multi-Agent：多个专家协作 → 各司其职，术业专攻 ├── 安全专家：专注漏洞扫描 ├── 架构师：专注设计评审 ├── 性能工程师：专注性能分析 └── 项目经理：汇总输出报告这就像真实的 Code Review 流程——不同角色从不同角度审查同一份代码。1.2 系统架构┌──────────────────┐ │ 用户 / GitHub │ │ 提交代码审查请求 │ └────────┬─────────┘ │ ▼ ┌──────────────────┐ │ API 网关 │ │ (FastAPI) │ └────────┬─────────┘ │ ▼ ┌──────────────────┐ │ 工作流编排器 │ │ CrewAI/LangGraph │ └────────┬─────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ 安全审查 │ │ 架构审查 │ │ 性能审查 │ │ Agent │ │ Agent │ │ Agent │ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │ │ │ └───────────────┼───────────────┘ ▼ ┌────────────────┐ │ 汇总 Agent │ │ 生成最终报告 │ └────────────────┘ │ ▼ ┌────────────────┐ │ 返回结果 │ │ / Web 界面展示 │ └────────────────┘1.3 各 Agent 角色定义角色职责关注点输出安全审查 Agent扫描安全漏洞SQL 注入、XSS、硬编码密钥、权限问题安全问题列表 + 严重等级架构审查 Agent评估代码设计设计模式、SOLID 原则、模块耦合、可扩展性架构建议 + 重构方案性能审查 Agent分析性能瓶颈时间复杂度、内存泄漏、N+1 查询、缓存策略性能问题 + 优化建议风格审查 Agent检查代码规范命名规范、注释质量、代码重复、可读性规范问题 + 修改建议汇总 Agent综合所有审查结果问题去重、优先级排序、生成报告最终审查报告1.4 协作模式模式一：并行审查（CrewAI）代码输入 │ ┌───────┼───────┬───────────┐ ▼ ▼ ▼ ▼ 安全 架构 性能 风格 Agent Agent Agent Agent │ │ │ │ └───────┴───────┴───────────┘ │ ▼ 汇总 Agent │ ▼ 审查报告四个 Agent 同时工作，最后汇总。模式二：串行流水线（LangGraph）代码输入 │ ▼ 代码预处理 Agent（解析、分模块） │ ▼ 安全审查 Agent │ ▼ 架构审查 Agent │ ▼ 性能审查 Agent │ ▼ 汇总 Agent（综合所有结果 + 去重 + 排序） │ ▼ 审查报告每个 Agent 按顺序执行，后面 Agent 可以看到前面的结果。第二章：环境搭建2.1 Python 环境# 创建虚拟环境conda create-ncode-reviewpython=3.10-yconda activate code-review2.2 安装依赖# 核心框架pipinstallcrewai crewai-tools pipinstalllangchain langchain-openai langchain-community pipinstalllanggraph# GitHub APIpipinstallPyGithub# 静态分析pipinstallpylint bandit radon# Web 服务pipinstallfastapi uvicorn pydantic# 其他工具pipinstallpython-dotenv tiktoken2.3 配置环境变量创建.env文件：# DeepSeek API DEEPSEEK_API_KEY=your-deepseek-api-key DEEPSEEK_BASE_URL=https://api.deepseek.com DEEPSEEK_MODEL=deepseek-chat # GitHub Token（可选，用于 PR 审查） GITHUB_TOKEN=your-github-personal-access-token # 应用配置 APP_PORT=8000 MAX_ITERATIONS=10 TEMPERATURE=0.12.4 GitHub Token 生成登录 GitHub → Settings → Developer settings → Personal access tokens → Tokens (classic)点击 “Generate new token”勾选权限：repo（完整仓库访问）复制 token 到.env文件2.5 验证环境# test_env.pyimportosfromdotenvimportload_dotenv load_dotenv()# 测试 DeepSeek APIfromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model=os.getenv("DEEPSEEK_MODEL","deepseek-chat"),openai_api_key=os.getenv("DEEPSEEK_API_KEY"),openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),temperature=0)response=llm.invoke("你好，请回复'环境配置成功'")print(response.content)# 测试 GitHub TokenfromgithubimportGithub g=Github(os.getenv("GITHUB_TOKEN"))user=g.get_user()print(f"GitHub 用户：{user.login}")python test_env.py输出类似：环境配置成功 GitHub 用户：your-username第三章：基础工具层实现在构建 Agent 之前，先准备好它们需要的"工具"。3.1 代码解析工具# tools/code_parser.py"""代码解析工具：将代码 diff 解析为结构化数据"""fromdataclassesimportdataclass,field@dataclassclassCodeChange:"""单个代码变更"""file_path:str# 文件路径change_type:str# 新增/修改/删除 (added/modified/deleted)added_lines:list# 新增的行removed_lines:list# 删除的行full_content:str# 文件完整内容language:str# 编程语言diff_text:str# 原始 diff 文本@dataclassclassReviewContext:"""审查上下文"""changes:list=field(default_factory=list)repo_name:str=""pr_number:int=0pr_title:str=""pr_description:str=""defdetect_language(file_path:str)-str:"""根据文件扩展名识别编程语言"""ext_map={".py":"Python",".js":"JavaScript",".ts":"TypeScript",".java":"Java",".go":"Go",".rs":"Rust",".cpp":"C++",".c":"C",".rb":"Ruby",".php":"PHP",".vue":"Vue",".jsx":"React JSX",".tsx":"React TSX",}forext,langinext_map.items():iffile_path.endswith(ext):returnlangreturn"Unknown"defparse_diff(diff_text:str)-list[CodeChange]:"""解析 unified diff 格式的代码差异"""changes=[]current_file=Noneadded_lines=[]removed_lines=[]forlineindiff_text.split("\n"):# 检测文件路径ifline.startswith("+++ b/"):ifcurrent_fileand(added_linesorremoved_lines):changes.append(CodeChange(file_path=current_file,change_type="modified"ifremoved_linesandadded_lineselse"added"ifadded_lineselse"deleted",added_lines=added_lines.copy(),removed_lines=removed_lines.copy(),full_content="",language=detect_language(current_file),diff_text=""))current_file=line[6:]added_lines=[]removed_lines=[]elifline.startswith("+")andnotline.startswith("+++"):added_lines.append(line[1:])elifline.startswith("-")andnotline.startswith("---"):removed_lines.append(line[1:])# 处理最后一个文件ifcurrent_fileand(added_linesorremoved_lines):changes.append(CodeChange(file_path=current_file,change_type="modified"ifremoved_linesandadded_lineselse"added"ifadded_lineselse"deleted",added_lines=added_lines.copy(),removed_lines=removed_lines.copy(),full_content="",language=detect_language(current_file),diff_text=""))returnchangesdefformat_code_for_review(changes:list[CodeChange])-str:"""将代码变更格式化为适合 Agent 审查的文本"""output=[]fori,changeinenumerate(changes,1):output.append(f"{'='*60}")output.append(f"文件{i}:{change.file_path}")output.append(f"类型:{change.change_type}| 语言:{change.language}")output.append(f"{'='*60}")ifchange.added_lines:output.append("\n--- 新增代码 ---")forline_num,lineinenumerate(change.added_lines,1):output.append(f"+{line}")ifchange.removed_lines:output.append("\n--- 删除代码 ---")forline_num,lineinenumerate(change.removed_lines,1):output.append(f"-{line}")ifchange.full_content:output.append(f"\n--- 文件完整内容 ---\n{change.full_content}")output.append("")return"\n".join(output)3.2 GitHub API 工具# tools/github_tool.py"""GitHub API 工具：获取 PR 信息和代码差异"""importosfromgithubimportGithubfrom.code_parserimportCodeChange,ReviewContext,detect_languageclassGitHubTool:"""GitHub 操作工具"""def__init__(self,token:str=None):self.token=tokenoros.getenv("GITHUB_TOKEN")self.github=Github(self.token)defget_pr_info(self,repo_name:str,pr_number:int)-dict:"""获取 PR 基本信息"""repo=self.github.get_repo(repo_name)pr=repo.get_pull(pr_number)return{"title":pr.title,"description":pr.bodyor"","author":pr.user.login,"state":pr.state,"base_branch":pr.base.ref,"head_branch":pr.head.ref,"changed_files":pr.changed_files,"additions":pr.additions,"deletions":pr.deletions,}defget_pr_diff(self,repo_name:str,pr_number:int)-list[CodeChange]:"""获取 PR 的代码变更"""repo=self.github.get_repo(repo_name)pr=repo.get_pull(pr_number)changes=[]forfileinpr.get_files():# 跳过非代码文件ifnotself._is_code_file(file.filename):continuechange=CodeChange(file_path=file.filename,change_type=file.status,added_lines=[l[1:]forlin(file.patchor"").split("\n")ifl.startswith("+")andnotl.startswith("+++")],removed_lines=[l[1:]forlin(file.patchor"").split("\n")ifl.startswith("-")andnotl.startswith("---")],full_content=self._get_file_content(repo,file.filename,pr.head.sha),language=detect_language(file.filename),diff_text=file.patchor"")changes.append(change)returnchangesdefget_pr_review_context(self,repo_name:str,pr_number:int)-ReviewContext:"""获取完整的审查上下文"""pr_info=self.get_pr_info(repo_name,pr_number)changes=self.get_pr_diff(repo_name,pr_number)returnReviewContext(changes=changes,repo_name=repo_name,pr_number=pr_number,pr_title=pr_info["title"],pr_description=pr_info["description"])defpost_review_comment(self,repo_name:str,pr_number:int,body:str,commit_sha:str=None):"""在 PR 上发布审查评论"""repo=self.github.get_repo(repo_name)pr=repo.get_pull(pr_number)ifcommit_sha:commit=repo.get_commit(commit_sha)pr.create_review_comment(body=body,commit=commit)else:pr.create_issue_comment(body)def_get_file_content(self,repo,file_path:str,ref:str)-str:"""获取文件完整内容"""try:content=repo.get_contents(file_path,ref=ref)returncontent.decoded_content.decode("utf-8")exceptException:return""def_is_code_file(self,file_path:str)-bool:"""判断是否为代码文件"""code_extensions={".py",".js",".ts",".java",".go",".rs",".cpp",".c",".h",".rb",".php",".vue",".jsx",".tsx",".css",".scss",".sql"}returnany(file_path.endswith(ext)forextincode_extensions)3.3 静态分析工具# tools/static_analysis.py"""静态分析工具：对代码进行自动化检查"""importsubprocessimporttempfileimportosimportjsonfrompathlibimportPathclassStaticAnalyzer:"""静态代码分析器"""defrun_pylint(self,code:str,file_path:str="code.py")-dict:"""运行 Pylint 代码质量检查"""withtempfile.NamedTemporaryFile(mode="w",suffix=".py",delete=False,encoding="utf-8")asf:f.write(code)temp_path=f.nametry:result=subprocess.run(["pylint",temp_path,"--output-format=json","--disable=C"],capture_output=True,text=True,timeout=30)issues=json.loads(result.stdout)ifresult.stdoutelse[]return{"tool":"pylint","issues":[{"line":i.get("line",0),"type":i.get("type","unknown"),"message":i.get("message",""),"symbol":i.get("symbol","")}foriinissues],"score":self._extract_pylint_score(result.stderr)}exceptExceptionase:return{"tool":"pylint","error":str(e),"issues":[]}finally:os.unlink(temp_path)defrun_bandit(self,code:str)-dict:"""运行 Bandit 安全扫描"""withtempfile.NamedTemporaryFile(mode="w",suffix=".py",delete=False,encoding="utf-8")asf:f.write(code)temp_path=f.nametry:result=subprocess.run(["bandit","-f","json",temp_path],capture_output=True,text=True,timeout=30)data=json.loads(result.stdout)ifresult.stdoutelse{}return{"tool":"bandit","issues":[{"line":r.get("line_number",0),"severity":r.get("issue_severity","UNKNOWN"),"confidence":r.get("issue_confidence","UNKNOWN"),"message":r.get("issue_text",""),"test_id":r.get("test_id","")}