第六章 Multi-Agent — 多个 Agent 协作
·
第6章：Multi-Agent — 多个 Agent 协作
📌 本章目标
- 理解为什么要用多个 Agent
- 掌握 Agent 角色设计方法
- 实现顺序编排（流水线模式）
- 实现对话式协作（你一言我一语）
6.1 单 Agent 的局限
一个 Agent 什么都能做，但也意味着它什么都不精：
用户: "帮我调研一下 Python Web 框架，写一篇对比文章，然后发公众号"
单 Agent 的困境:
1. 调研需要搜索、对比、验证 → 研究员思维
2. 写文章需要文笔、结构、读者视角 → 写手思维
3. 发公众号需要排版、标题优化 → 运营思维
一个 Agent 来回切换角色，容易混乱，输出质量不稳定
多 Agent 的思路：每个 Agent 只做一件事，做到最好。然后用一个编排器让它们协作。
6.2 定义 Agent 类
首先，我们抽象出一个通用的 Agent 类：
"""
通用 Agent 类 —— 每个 Agent 实例就是一个"员工"
使用前请先: pip install openai python-dotenv
并在 .env 文件中设置: DEEPSEEK_API_KEY=sk-你的密钥
"""
import json
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
# DeepSeek 客户端
client = OpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1"
)
def chat(messages: list, model: str = "deepseek-chat", temperature: float = 0.7) -> str:
"""封装 LLM 调用，返回文本回复"""
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
)
return response.choices[0].message.content
def chat_with_tools(messages: list, tools: list, model: str = "deepseek-chat") -> dict:
"""带工具调用的 LLM 调用，返回结构化结果"""
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
)
msg = response.choices[0].message
if msg.tool_calls:
return {
"has_tool_calls": True,
"content": None,
"tool_calls": [
{
"id": tc.id,
"name": tc.function.name,
"arguments": tc.function.arguments,
}
for tc in msg.tool_calls
],
"raw_message": msg,
}
else:
return {
"has_tool_calls": False,
"content": msg.content,
"tool_calls": [],
"raw_message": msg,
}
class Agent:
"""一个能独立工作的 AI Agent"""
def __init__(
self,
name: str,
role: str,
personality: str = "",
tools: dict[str, callable] = None,
tool_schemas: list[dict] = None,
):
"""
参数:
name: Agent 的名字（如"研究员小明"）
role: Agent 的职责描述（如"负责搜集和整理技术资料"）
personality: 性格/风格描述（如"严谨、注重细节"）
tools: 可用的工具 {函数名: 函数}
tool_schemas: 工具的 JSON Schema
"""
self.name = name
self.role = role
self.personality = personality
self.tools = tools or {}
self.tool_schemas = tool_schemas or []
# 构建 system prompt
self.system_prompt = self._build_system_prompt()
def _build_system_prompt(self) -> str:
"""根据角色信息构建 system prompt"""
prompt = f"你是 {self.name}。你的职责是 {self.role}。"
if self.personality:
prompt += f" 你的风格是 {self.personality}。"
prompt += "\n请你只做你职责范围内的事，不要越权。"
prompt += "\n给出直接有用的输出，不要多余的客套话。"
return prompt
def run(self, task: str, context: str = "") -> str:
"""
执行一个任务
参数:
task: 当前要完成的任务描述
context: 上游 Agent 的输出或其他上下文信息
返回:
任务执行结果
"""
messages = [{"role": "system", "content": self.system_prompt}]
# 如果有上下文，拼接进去
if context:
messages.append({
"role": "user",
"content": f"背景信息：\n{context}\n\n请完成以下任务：\n{task}"
})
else:
messages.append({"role": "user", "content": task})
# 如果有工具，走 tool calling 循环
if self.tools:
return self._run_with_tools(messages)
else:
return chat(messages)
def _run_with_tools(self, messages: list, max_steps: int = 5) -> str:
"""带工具调用的执行循环"""
for _ in range(max_steps):
response = chat_with_tools(messages, self.tool_schemas)
if response["has_tool_calls"]:
# 处理工具调用
messages.append(response["raw_message"])
for tc in response["tool_calls"]:
func = self.tools.get(tc["name"])
if func:
args = json.loads(tc["arguments"])
result = func(**args)
messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"content": str(result)
})
else:
return response["content"]
return chat(messages) # 达到最大步数，强制总结
# ── 测试：创建一个研究员 Agent ──
def mock_search(query: str) -> str:
"""模拟搜索工具"""
db = {
"FastAPI": "FastAPI 是一个高性能 Python Web 框架，基于 Starlette 和 Pydantic，支持异步，性能接近 Node.js",
"Django": "Django 是全栈 Web 框架，内置 ORM、Admin、认证等，适合快速开发，学习曲线较陡",
"Flask": "Flask 是轻量级 Web 框架，灵活自由，适合微服务和简单 API，需要自己组装扩展",
}
return db.get(query, f"未找到 {query} 的信息")
if __name__ == "__main__":
researcher = Agent(
name="研究员同学",
role="搜索和整理技术资料，输出结构化的调研报告",
personality="严谨，注重事实，用数据说话",
tools={"search": mock_search},
tool_schemas=[{
"type": "function",
"function": {
"name": "search",
"description": "搜索技术信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
}]
)
result = researcher.run("调研 FastAPI、Django、Flask 三个框架的优缺点")
print(result)
6.3 顺序编排：流水线模式
最常用的多 Agent 模式——一个接一个，像工厂流水线：
"""
流水线编排 —— 将本节代码追加到6.2节完整代码后面即可运行
依赖6.2节定义的 Agent 类、chat、chat_with_tools、client
"""
class SequentialPipeline:
"""
顺序编排：Agent A → Agent B → Agent C → 输出
适用场景：
- 调研 → 写作 → 审核
- 需求分析 → 代码编写 → 代码审查
- 翻译 → 润色 → 本地化
"""
def __init__(self):
self.agents: list[Agent] = []
self.tasks: list[str] = []
def add_stage(self, agent: Agent, task_template: str):
"""
添加一个阶段
task_template 中可以用 {input} 引用上一个阶段的输出
例如: "请根据以下资料写文章:\n{input}"
"""
self.agents.append(agent)
self.tasks.append(task_template)
def run(self, initial_input: str, verbose: bool = True) -> str:
"""按顺序执行所有阶段"""
current_output = initial_input
for i, (agent, task) in enumerate(zip(self.agents, self.tasks), 1):
if verbose:
print(f"\n{'='*50}")
print(f"🔹 阶段 {i}: {agent.name}")
print(f"{'='*50}")
# 把 {input} 替换为上一阶段输出
full_task = task.format(input=current_output)
current_output = agent.run(full_task)
if verbose:
print(f"输出预览: {current_output[:200]}...")
return current_output
# ── 完整示例：用多 Agent 写技术文章 ──
def demo_pipeline():
"""演示：研究员 → 写手 → 审核员，协作写一篇文章"""
# Agent 1: 研究员
researcher = Agent(
name="研究员",
role="搜索和整理技术资料，输出结构化的要点列表",
personality="严谨客观，每个观点都有依据",
)
# Agent 2: 写手
writer = Agent(
name="写手",
role="将研究资料写成通俗易懂的教程文章",
personality="文笔流畅，擅长用小例子解释复杂概念，面向初学者",
)
# Agent 3: 审核员
reviewer = Agent(
name="审核员",
role="审核文章质量，检查准确性、可读性、完整性",
personality="严格但建设性，每个问题都给出具体改进建议",
)
# 搭建流水线
pipeline = SequentialPipeline()
pipeline.add_stage(
researcher,
"请调研以下主题的关键知识点，列出5-8个要点：Python 装饰器是什么，怎么用，常见场景"
)
pipeline.add_stage(
writer,
"请根据以下研究资料，写一篇面向编程初学者的教程文章，500字左右：\n{input}"
)
pipeline.add_stage(
reviewer,
"请审核以下文章，然后给出最终改进后的版本：\n{input}"
)
# 运行
final_article = pipeline.run("Python 装饰器")
print(f"\n{'='*50}")
print(f"📄 最终文章:\n{final_article}")
return final_article
6.4 对话式协作：Agent 之间互相讨论
有时候我们希望 Agent 之间像开会一样讨论，而不是简单的流水线：
"""
对话式协作 —— 将本节代码追加到6.2节完整代码后面即可运行
依赖6.2节定义的 Agent 类、chat、chat_with_tools、client
"""
class ConversationOrchestrator:
"""
对话式编排：多个 Agent 在一起讨论一个问题
像圆桌会议一样
"""
def __init__(self, moderator_prompt: str = ""):
self.agents: dict[str, Agent] = {}
self.moderator_prompt = moderator_prompt or "你是会议主持人，负责引导讨论、总结共识"
def add_agent(self, agent: Agent):
"""添加一个参会 Agent"""
self.agents[agent.name] = agent
def discuss(self, topic: str, rounds: int = 2, verbose: bool = True) -> str:
"""
多 Agent 讨论一个话题
流程：
1. 主持人介绍话题
2. 每个 Agent 轮流发言
3. 重复 rounds 轮
4. 主持人总结共识
"""
transcript = [f"# 讨论主题：{topic}\n"]
for r in range(rounds):
if verbose:
print(f"\n{'='*50}")
print(f"🔵 第 {r+1} 轮讨论")
print(f"{'='*50}")
for name, agent in self.agents.items():
# 构建讨论上下文
discussion_context = "\n".join(transcript)
task = f"""你正在参加一个关于「{topic}」的小组讨论。
以下是目前的讨论记录：
{discussion_context}
请从你的角色角度发表看法。你可以：
- 对主题提出你的见解
- 回应其他人的观点（赞同或质疑都可以）
- 提出新的思路
请简明扼要地发言。"""
response = agent.run(task)
transcript.append(f"[{name}]: {response}")
if verbose:
print(f"🗣️ [{name}]: {response[:150]}...")
# 主持人总结
if verbose:
print(f"\n{'='*50}")
print("📋 总结中...")
moderator = Agent(
name="主持人",
role="总结讨论要点，形成最终结论"
)
summary = moderator.run(
f"请总结以下讨论，指出共识、分歧和最终建议：\n{chr(10).join(transcript)}"
)
return summary
# ── 演示：讨论一个技术决策 ──
def demo_discussion():
"""演示多个 Agent 讨论技术方案"""
# CTO：关心架构和扩展性
cto = Agent(
name="CTO",
role="从架构和技术战略角度评估方案，关心可扩展性、维护成本",
personality="眼光长远，注重技术选型的长期影响"
)
# 后端工程师：关心实现细节
backend_dev = Agent(
name="后端工程师",
role="从实现角度评估方案，关心开发效率、性能、稳定性",
personality="务实，注重落地方案的可操作性"
)
# 产品经理：关心用户体验
pm = Agent(
name="产品经理",
role="从用户价值和业务目标角度评估方案",
personality="以用户为中心，注重投入产出比"
)
orchestrator = ConversationOrchestrator()
orchestrator.add_agent(cto)
orchestrator.add_agent(backend_dev)
orchestrator.add_agent(pm)
conclusion = orchestrator.discuss(
"我们是否应该从单体架构迁移到微服务？团队有5个人，用户量大约10万"
)
print(f"\n📋 最终结论:\n{conclusion}")
6.5 常用多 Agent 模式总结
📝 本章小结
- 多 Agent 的本质：分工合作，每个 Agent 专注一个角色
- Agent = 角色 + System Prompt + 工具
- 流水线模式：一个接一个，适合有明确步骤的任务
- 对话模式：互相讨论，适合需要多方视角的决策
- 不要过度设计：2-3个 Agent 能解决的事，不要搞 10 个
✏️ 练习题
-
基础题：实现一个"代码审查员" Agent，它接收一段代码，审查后返回问题和改进建议。
-
进阶题：用流水线模式实现一个"翻译优化"流程：翻译员 → 润色员 → 本地化审核员（检查文化适配性）。
-
挑战题：实现一个简单的投票机制——3个 Agent 对同一个问题给出答案，取多数意见作为最终结果。
下一章预告：第7章：实战项目 — 个人知识库问答Agent —— 我们把前6章学到的所有东西，整合成一个能跑起来的完整应用！
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
24
0- 0
已为社区贡献2条内容
所有评论(0)