Day4：多Agent协作

从"一个人扛"到"团队协作"

开发者痛点：你是否遇到过一个Agent既要写代码又要写文档，结果两样都做不好？或者想让AI自动完成复杂任务，但单Agent能力不足？今天教你用多Agent协作解决这些问题！

今日学习目标

掌握3种协作模式完整可运行代码3个实战场景代码审查+客服工单

一、为什么单Agent不够？

先来看一个真实场景：

产品经理的需求：

"我要一个能自动审核代码的Agent，需要检查代码规范、安全漏洞、性能问题，还要生成审查报告..."

单Agent的问题：

1️⃣ 审查标准混在一起，输出混乱

2️⃣ 专业深度不够，只能泛泛而谈

3️⃣ 审查+写报告，两头都做不好

多Agent协作的解决方案：

代码规范Agent → 检查命名、注释、结构

安全审计Agent → 扫描注入、XSS等漏洞

性能分析Agent → 检测慢查询、内存泄漏

报告生成Agent → 汇总结果，生成专业报告

核心洞察：复杂任务需要专业分工 + 协作，就像真实的开发团队

二、3种协作模式详解

多Agent协作有3种经典模式，根据场景选择：

模式1：串行流水线（Sequential）

Agent按顺序执行，前一个的输出是后一个的输入

适用场景：内容创作（大纲→写作→编辑→校对）代码：result = agent1.run(task) → agent2.run(result) → agent3.run(result)

模式2：并行投票（Parallel）

多个Agent同时处理，最后汇总结果

适用场景：多角度分析（代码审查、方案评估）代码：results = [agent.run(task) for agent in agents]

模式3：主从协调（Orchestrator）

一个"主管Agent"分析任务，分配给合适的"专家Agent"

适用场景：复杂任务分配（客服工单处理、智能客服）代码：expert = orchestrator.select_expert(task) → expert.run(task)

三、实战1：代码审查团队（并行模式）

场景：你是一个Tech Lead，每天要Review团队代码，耗时费力。用多Agent自动完成！

完整代码实现

import openai# 代码示例：待审查的Python函数 code_sample = """ def get_user_data(user_id):     query = f"SELECT * FROM users WHERE id = {user_id}"     result = db.execute(query)     return result"""# Agent 1: 代码规范检查员class StyleReviewer:     system_prompt = """你是代码规范专家。检查以下问题：- 变量命名是否符合PEP8- 函数是否有文档字符串- 代码缩进和格式是否正确输出格式：【问题】→ 【建议】"""def review(self, code):         resp = openai.chat.completions.create(             model="gpt-4",             messages=[                 {"role": "system", "content": self.system_prompt},                 {"role": "user", "content": f"请审查代码：\n{code}"}             ]         )return resp.choices[0].message.content# Agent 2: 安全审计员class SecurityReviewer:     system_prompt = """你是安全专家。扫描以下漏洞：- SQL注入风险- XSS漏洞- 敏感信息硬编码- 不安全的文件操作输出格式：【严重等级】漏洞描述 → 修复建议"""def review(self, code):# 与安全专家的Prompt类似...pass# 并行执行审查 style_reviewer = StyleReviewer() security_reviewer = SecurityReviewer()# 同时发起两个审查（实际用async更优雅） style_result = style_reviewer.review(code_sample) security_result = security_reviewer.review(code_sample)# 汇总报告print("=== 代码审查报告 ===")print(f"规范检查：{style_result}")print(f"安全检查：{security_result}") 

运行效果

=== 代码审查报告 ===

四、实战2：智能客服工单系统（主从模式）

场景：公司客服每天要处理各种工单（技术问题、账单疑问、产品咨询），人工分类耗时。用多Agent自动分流！

完整代码实现

# 工单路由Agent - 分析并分配给专业团队class TicketRouter:     categories = ["technical", "billing", "product", "complaint"]def __init__(self):         self.agents = {"technical": TechSupportAgent(),"billing": BillingAgent(),"product": ProductAgent(),         }def route(self, ticket):# 用LLM智能分类工单         classification = self.classify(ticket)# 分配给对应的专家Agent         agent = self.agents.get(classification["category"])if agent:             response = agent.handle(                 ticket, classification["priority"]             )return {"ticket": ticket,"category": classification["category"],"priority": classification["priority"],"response": response,"assigned_to": classification["category"]             }def classify(self, ticket):         prompt = f"""分析以下客服工单，输出JSON格式：工单内容：{ticket['content']}分类选项：technical(技术问题)、billing(账单问题)、product(产品咨询)、complaint(投诉)输出格式：{  "category": "分类结果",  "priority": "urgent/high/medium/low",  "keywords": ["关键词1", "关键词2"],  "sentiment": "positive/neutral/negative"}"""# 调用LLM分类...return json.loads(response)# 技术支援Agentclass TechSupportAgent:def handle(self, ticket, priority):# 技术分析逻辑...return"已为您排查问题，请按以下步骤操作..."# 测试工单 router = TicketRouter() ticket = {"id": "TKT-2024-001","content": "我的APP登录不上去了，一直显示网络错误，              但是其他APP都能正常使用，急！" } result = router.route(ticket) 

运行效果

工单处理结果：

五、实战3：内容创作流水线（串行模式）

场景：运营团队每周要产出多篇公众号文章，从选题到发布流程长。用多Agent流水线自动化！

完整代码实现

class ContentPipeline:"""内容创作流水线 - 串行模式"""def __init__(self):         self.agents = {"idea": IdeaAgent(),  # 选题"outline": OutlineAgent(),  # 大纲"write": WriterAgent(),  # 写作"edit": EditorAgent(),  # 润色"review": ReviewAgent(),  # 合规检查         }         self.workflow = ["idea", "outline", "write", "edit", "review"]def create_article(self, topic):# 数据在Agent间流转         data = {"topic": topic, "content": ""}for step in self.workflow:             agent = self.agents[step]print(f"[Pipeline] Step: {step}...")             data = agent.process(data)return data# 各个Agent的实现class IdeaAgent:"""选题Agent - 找热点、选角度"""def process(self, data):         data["angle"] = "从开发者痛点切入"return dataclass WriterAgent:"""写作Agent - 产出初稿"""def process(self, data):# 根据大纲写正文         data["content"] = "这里是生成的文章内容..."return data# 运行流水线 pipeline = ContentPipeline() article = pipeline.create_article("AI Agent开发教程")print(f"文章已生成：{article['content'][:100]}...") 

六、3种模式对比与选型

七、进阶：混合模式与优化技巧

实战技巧

八、今日作业

🎯 必做任务

选择你工作中的一个实际痛点（如周报生成、Bug分类、会议纪要整理），用多Agent实现自动化

🚀 进阶挑战

实现混合模式：代码审查用并行模式 + 报告生成用串行模式

💡 思考题

如果两个Agent对代码审查结果有分歧，如何设计仲裁机制？

明天我们将学习：

• RAG增强
  ：让Agent拥有自己的知识库
• 向量数据库 + 智能检索
• 企业知识库问答系统
• 实战：搭建专属知识库Agent

今天你已经掌握了3种协作模式和3个实战场景。明天，我们将让Agent拥有自己的知识库，成为领域专家！