八、完整实战从零构建多Agent协作系统8.1 实战目标构建一个技术方案评审多Agent系统: 用户输入: 一个技术方案描述 系统输出: 多角度的评审报告 Agent团队: 1. 架构师Agent: 评审架构设计的合理性 2. 安全专家Agent: 评审安全风险 3. 性能专家Agent: 评审性能瓶颈 4. 汇总Agent: 整合所有评审意见生成最终报告 架构模式: 管理者模式汇总Agent作为管理者 通信方式: 共享状态8.2 完整实现 完整实战: 技术方案评审多Agent系统 依赖: pip install openai fromopenaiimportOpenAIfromdataclassesimportdataclass,fieldfromdatetimeimportdatetimeimportjsonfromconcurrent.futuresimportThreadPoolExecutor,as_completed clientOpenAI()# ═══════════════════════════════════════════════════════════════════# Part 1: Agent定义# ═══════════════════════════════════════════════════════════════════dataclassclassReviewAgent:评审Agentname:strrole:strsystem_prompt:strreview_dimensions:list[str]# 评审维度defreview(self,proposal:str)-dict:对技术方案进行评审dimensions_text\n.join([f{i1}.{d}fori,dinenumerate(self.review_dimensions)])responseclient.chat.completions.create(modelgpt-4o-mini,messages[{role:system,content:self.system_prompt},{role:user,content:f请从你的专业角度评审以下技术方案:{proposal}请从以下维度进行评审:{dimensions_text}输出JSON格式: {{ score: 1-10的评分, summary: 一句话总结评审结论, issues: [ {{severity: high/medium/low, description: 问题描述, suggestion: 改进建议}} ], highlights: [方案的亮点1, 方案的亮点2] }}}],response_format{type:json_object},temperature0.3)try:resultjson.loads(response.choices[0].message.content)result[reviewer]self.name result[role]self.rolereturnresultexcept:return{reviewer:self.name,role:self.role,score:5,summary:评审过程出错,issues:[],highlights:[]}# ═══════════════════════════════════════════════════════════════════# Part 2: 评审团队# ═══════════════════════════════════════════════════════════════════classReviewTeam:评审团队: 管理多个评审Agentdef__init__(self):self.reviewersself._create_reviewers()def_create_reviewers(self)-list[ReviewAgent]:创建评审团队return[ReviewAgent(name架构师,role系统架构评审,system_prompt你是一位资深系统架构师有15年分布式系统设计经验。 你擅长评估系统架构的合理性、可扩展性和可维护性。 你的评审风格: 严谨但建设性指出问题的同时给出改进方案。,review_dimensions[架构分层是否合理,组件职责是否清晰,可扩展性设计,技术选型是否合适,是否存在过度设计或设计不足]),ReviewAgent(name安全专家,role安全风险评审,system_prompt你是一位信息安全专家专注于应用安全和基础设施安全。 你擅长发现潜在的安全漏洞、数据泄露风险和权限问题。 你的评审风格: 从攻击者视角思考找出最薄弱的环节。,review_dimensions[认证和授权机制,数据安全传输和存储,输入验证和注入防护,敏感信息保护,第三方依赖的安全性]),ReviewAgent(name性能专家,role性能瓶颈评审,system_prompt你是一位性能工程专家专注于高并发系统的性能优化。 你擅长识别性能瓶颈、评估系统容量和优化方案。 你的评审风格: 用数据说话关注P99延迟和吞吐量。,review_dimensions[是否存在性能瓶颈,数据库查询效率,缓存策略是否合理,并发处理能力,资源使用效率]),]defreview(self,proposal:str,parallel:boolTrue)-dict:执行评审支持并行print(f 开始评审共{len(self.reviewers)}位评审专家)print(f 并行模式:{是ifparallelelse否})reviews[]ifparallel:# 并行执行所有评审withThreadPoolExecutor(max_workerslen(self.reviewers))asexecutor:futures{executor.submit(reviewer.review,proposal):reviewerforreviewerinself.reviewers}forfutureinas_completed(futures):reviewerfutures[future]resultfuture.result()reviews.append(result)print(f ✅{reviewer.name}评审完成 (评分:{result.get(score,?)}/10))else:# 串行执行forreviewerinself.reviewers:print(f →{reviewer.name}正在评审...)resultreviewer.review(proposal)reviews.append(result)print(f ✅{reviewer.name}评审完成 (评分:{result.get(score,?)}/10))# 汇总评审结果print(f\n 汇总评审结果...)summaryself._synthesize(proposal,reviews)return{individual_reviews:reviews,summary:summary,timestamp:datetime.now().isoformat()}def_synthesize(self,proposal:str,reviews:list[dict])-dict:汇总所有评审意见reviews_textforrinreviews:reviews_textf\n---{r[reviewer]}({r[role]}) 评分:{r.get(score,?)}/10 ---\nreviews_textf总结:{r.get(summary,)}\nifr.get(issues):reviews_text问题:\nforissueinr[issues]:reviews_textf [{issue.get(severity,)}]{issue.get(description,)}\nifr.get(highlights):reviews_textf亮点:{, .join(r.get(highlights,[]))}\nresponseclient.chat.completions.create(modelgpt-4o-mini,messages[{role:user,content:f请汇总以下多位专家的评审意见生成最终评审报告: 技术方案:{proposal[:500]}各专家评审:{reviews_text}输出JSON格式: {{ overall_score: 综合评分(1-10), verdict: 通过/有条件通过/需要修改/不通过, key_risks: [最关键的风险1, 风险2], must_fix: [必须修改的问题1, 问题2], nice_to_have: [建议优化的点1, 点2], final_comment: 给方案作者的综合建议2-3句话 }}}],response_format{type:json_object},temperature0.3)try:returnjson.loads(response.choices[0].message.content)except:return{overall_score:5,verdict:需要修改,final_comment:汇总过程出错}# ═══════════════════════════════════════════════════════════════════# Part 3: 使用示例# ═══════════════════════════════════════════════════════════════════if__name____main__:teamReviewTeam()proposal 技术方案: 订单服务重构 1. 架构: 从单体拆分为微服务使用gRPC通信 2. 数据库: 从MySQL迁移到PostgreSQL引入读写分离 3. 缓存: 使用Redis缓存热点订单数据TTL5分钟 4. 消息队列: 引入Kafka处理异步任务发送通知、更新统计 5. API网关: 使用Kong作为API网关统一认证和限流 6. 部署: Kubernetes部署HPA自动扩缩容 resultteam.review(proposal,parallelTrue)print(\n*60)print( 最终评审报告)print(*60)summaryresult[summary]print(f\n综合评分:{summary.get(overall_score,?)}/10)print(f评审结论:{summary.get(verdict,?)})print(f\n关键风险:)forriskinsummary.get(key_risks,[]):print(f ⚠️{risk})print(f\n必须修改:)forfixinsummary.get(must_fix,[]):print(f {fix})print(f\n建议优化:)forniceinsummary.get(nice_to_have,[]):print(f {nice})print(f\n综合建议:{summary.get(final_comment,)})