多 Agent 系统#
为什么需要多 Agent？#
单个 Agent 能力有限，面对复杂任务时会遇到：
上下文长度瓶颈：单次请求能处理的信息有限
专业化瓶颈：一个通用 Agent 无法在所有领域都表现优秀
速度瓶颈：串行执行效率低
解决方案：让多个专门化 Agent 协作，各司其职。
常见的多 Agent 模式#
模式 |
结构 |
适用场景 |
|---|---|---|
Orchestrator-Worker |
一个主 Agent 调度多个子 Agent |
需要规划和分工的复杂任务 |
Pipeline（流水线） |
每个 Agent 输出作为下一个的输入 |
有明确处理顺序的任务 |
并行 Worker |
多个 Agent 同时处理不同子任务 |
可拆分的独立子任务 |
投票表决 |
多个 Agent 独立给出答案，再裁决 |
高风险、需要可靠性的决策 |
import json
import os
import concurrent.futures
import litellm
litellm.drop_params = True
from dotenv import load_dotenv
load_dotenv()
MODEL = os.getenv("LLM_MODEL")
print(f"使用模型：{MODEL}")
# gpt-5/o系列不支持自定义temperature值，统一用安全wrapper
def _c(**kw):
_m = kw.get('model', MODEL)
if any(_m.startswith(p) for p in ('openai/gpt-5','openai/o1','openai/o3','openai/o4')):
kw.pop('temperature', None)
return litellm.completion(**kw)
使用模型：openai/gpt-5-mini
# ── 通用 Agent 辅助函数 ────────────────────────────────────
def agent(
system_prompt: str,
user_message: str,
temperature: float = 0.7,
max_tokens: int = 512,
**kwargs,
) -> str:
"""
单次调用 Agent 的辅助函数。
参数：
system_prompt: 定义 Agent 的角色和行为
user_message: 用户输入或上游 Agent 的输出
temperature: 创意度（0=确定性，1=创意）
max_tokens: 最大输出长度
**kwargs: 传递给 litellm.completion 的其他参数
返回：
Agent 的文本输出
"""
response = _c(
model=MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
temperature=temperature,
max_tokens=max_tokens,
**kwargs,
)
return response.choices[0].message.content
# 快速测试
test_reply = agent(
system_prompt="你是一个助手，用一句话回答。",
user_message="量子计算是什么？",
temperature=0,
)
print(f"测试输出：{test_reply}")
测试输出：量子计算是利用量子比特的叠加与纠缠特性，通过对量子态进行量子门操作和测量来执行计算，从而在某些问题上（如因数分解和搜索）显著超越经典计算机的计算能力。
Section 1：Orchestrator-Worker 模式#
核心思路：
Orchestrator（编排者）：理解任务，制定计划，决定调用哪些 Worker
Worker（执行者）：各有专长，接受指令执行具体工作
用户任务
↓
Orchestrator（规划 + 分配）
├→ ResearchAgent（调研）
├→ WriterAgent（写作）
└→ ReviewerAgent（审校）
↓
最终结果
# ── Worker Agents 定义 ────────────────────────────────────
def research_agent(topic: str) -> str:
"""调研 Agent：收集和整理关于某个话题的关键事实"""
print(f" [ResearchAgent] 调研主题：{topic}")
result = agent(
system_prompt=(
"你是一个专业的研究员。给定一个主题，提供 3-5 个关键事实和数据。"
"每个事实单独一行，以'-'开头。内容简洁准确。"
),
user_message=f"请调研：{topic}",
temperature=0.3,
)
print(f" [ResearchAgent] 完成，输出 {len(result)} 字")
return result
def writer_agent(research_result: str, style: str = "科普") -> str:
"""写作 Agent：将调研结果转化为可读的文章"""
print(f" [WriterAgent] 撰写 {style} 风格文章")
result = agent(
system_prompt=(
f"你是一个擅长写{style}文章的作家。"
"将提供的研究资料转化为流畅、易读的文章段落。"
"要有引人入胜的开头，清晰的主体，简洁的结尾。"
),
user_message=f"基于以下研究资料写一篇文章：\n\n{research_result}",
temperature=0.8,
max_tokens=600,
)
print(f" [WriterAgent] 完成，输出 {len(result)} 字")
return result
def reviewer_agent(article: str) -> str:
"""审校 Agent：检查文章质量并给出改进建议"""
print(f" [ReviewerAgent] 审校文章")
result = agent(
system_prompt=(
"你是一个严格的编辑。审校文章时重点检查："
"1) 逻辑连贯性 2) 事实准确性 3) 表达清晰度。"
"给出评分（1-10）和具体的改进建议，然后输出修改后的最终版本。"
),
user_message=f"请审校以下文章：\n\n{article}",
temperature=0.3,
max_tokens=800,
)
print(f" [ReviewerAgent] 完成")
return result
# ── Orchestrator Agent ────────────────────────────────────
def orchestrator_agent(task: str) -> dict:
"""
编排 Agent：分析任务，输出执行计划（JSON 格式）。
返回结构：
{
"plan": "任务分析...",
"research_topic": "调研主题",
"article_style": "文章风格",
"need_review": true/false
}
"""
print(f" [Orchestrator] 分析任务：{task}")
result = agent(
system_prompt=(
"你是一个任务编排者。分析用户的写作任务，输出执行计划。"
"只输出 JSON，格式："
'{"plan": "任务分析", "research_topic": "调研主题", '
'"article_style": "科普/学术/商业", "need_review": true}'
),
user_message=task,
temperature=0,
)
try:
content = result.strip()
if content.startswith("```"):
content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
plan = json.loads(content)
except json.JSONDecodeError:
# 解析失败时使用默认计划
plan = {
"plan": "直接执行",
"research_topic": task,
"article_style": "科普",
"need_review": True,
}
print(f" [Orchestrator] 计划：{plan}")
return plan
def run_orchestrator_pipeline(task: str) -> str:
"""完整的 Orchestrator-Worker 流程"""
import time
print(f"\n{'='*60}")
print(f"任务：{task}")
print(f"{'='*60}")
# Step 1: 编排者制定计划
print("\n[Phase 1] 编排者分析任务...")
plan = orchestrator_agent(task)
# Step 2: 调研
print("\n[Phase 2] 调研 Agent 工作...")
research = research_agent(plan["research_topic"])
print(f"调研结果：\n{research}")
# Step 3: 写作
print("\n[Phase 3] 写作 Agent 工作...")
article = writer_agent(research, plan.get("article_style", "科普"))
print(f"\n初稿：\n{article}")
# Step 4: 审校（如果需要）
if plan.get("need_review", True):
print("\n[Phase 4] 审校 Agent 工作...")
final = reviewer_agent(article)
print(f"\n最终版本：\n{final}")
return final
return article
# 运行演示
final_article = run_orchestrator_pipeline(
task="写一篇关于量子计算的简短科普文章，面向普通读者"
)
============================================================
任务：写一篇关于量子计算的简短科普文章，面向普通读者
============================================================
[Phase 1] 编排者分析任务...
[Orchestrator] 分析任务：写一篇关于量子计算的简短科普文章，面向普通读者
[Orchestrator] 计划：{'plan': '直接执行', 'research_topic': '写一篇关于量子计算的简短科普文章，面向普通读者', 'article_style': '科普', 'need_review': True}
[Phase 2] 调研 Agent 工作...
[ResearchAgent] 调研主题：写一篇关于量子计算的简短科普文章，面向普通读者
[ResearchAgent] 完成，输出 0 字
调研结果：
[Phase 3] 写作 Agent 工作...
[WriterAgent] 撰写 科普 风格文章
[WriterAgent] 完成，输出 0 字
初稿：
[Phase 4] 审校 Agent 工作...
[ReviewerAgent] 审校文章
[ReviewerAgent] 完成
最终版本：
你还没粘贴文章内容。请把要审校的全文粘到这里，或上传文件/给出链接。
在你提交文章前，请选择或告知（可选）：
- 目标读者与语气（例如：学术/通俗/商务/新闻/博客）；
- 是否需要保留原作者风格（还是更改为更正式/简洁）；
- 是否要同时检查格式、引用和标点（还是只关注逻辑、事实与表达）；
- 是否需要同时输出“逐句修改对照”（track changes）或仅提供最终修改后版本。
收到文章后我会按你要求：
1) 给出1–10分的总体评分并说明评分理由；
2) 针对逻辑连贯性、事实准确性、表达清晰度给出具体改进建议（逐条）；
3) 输出修改后的最终版本（根据你是否要保留原风格调整语言）。
Section 2：并行 Worker#
当多个子任务相互独立时，串行执行会浪费时间。
使用 concurrent.futures.ThreadPoolExecutor 可以让多个 Agent 并发运行。
┌─→ 调研主题 A ─┐
用户任务 → 拆分任务 ─┼─→ 调研主题 B ─┼─→ 综合 Agent → 最终结果
└─→ 调研主题 C ─┘
（三个 Agent 同时运行）
注意：线程并发对 I/O 密集型任务（API 调用）效果显著，对 CPU 密集型任务需用多进程。
import time
def parallel_research(topics: list[str], max_workers: int = 3) -> dict[str, str]:
"""
并行调研多个主题。
参数：
topics: 要调研的主题列表
max_workers: 最大并发线程数
返回：
{topic: research_result} 字典
"""
results = {}
def research_one(topic: str) -> tuple[str, str]:
print(f" [并行] 开始调研：{topic}")
result = agent(
system_prompt="你是一个专业研究员。用 2-3 句话概括这个主题的核心内容。",
user_message=f"简要介绍：{topic}",
temperature=0.3,
max_tokens=200,
)
print(f" [并行] 完成调研：{topic}")
return topic, result
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(research_one, t): t for t in topics}
for future in concurrent.futures.as_completed(futures):
try:
topic, result = future.result()
results[topic] = result
except Exception as e:
topic = futures[future]
results[topic] = f"调研失败：{e}"
return results
def synthesize_agent(research_dict: dict[str, str], synthesis_goal: str) -> str:
"""综合 Agent：将多个调研结果整合为连贯输出"""
combined = "\n\n".join(
f"【{topic}】\n{content}" for topic, content in research_dict.items()
)
return agent(
system_prompt=(
"你是一个善于综合归纳的分析师。"
"将多个来源的信息整合为结构清晰的综合报告。"
),
user_message=f"综合目标：{synthesis_goal}\n\n各主题调研结果：\n{combined}",
temperature=0.5,
max_tokens=600,
)
# ── 演示：并行调研 3 个主题 ──────────────────────────────
topics = [
"量子计算的基本原理",
"量子计算的当前应用",
"量子计算面临的技术挑战",
]
print("并行 Worker 演示")
print("="*50)
print(f"同时调研 {len(topics)} 个主题...\n")
start_time = time.time()
research_results = parallel_research(topics, max_workers=3)
parallel_time = time.time() - start_time
print(f"\n并行耗时：{parallel_time:.1f} 秒")
print("\n各主题调研结果：")
for topic, result in research_results.items():
print(f"\n[{topic}]\n{result}")
# 综合所有结果
print("\n综合 Agent 整合结果...")
synthesis = synthesize_agent(research_results, "量子计算综合报告")
print(f"\n综合报告：\n{synthesis}")
并行 Worker 演示
==================================================
同时调研 3 个主题...
[并行] 开始调研：量子计算的基本原理
[并行] 开始调研：量子计算的当前应用
[并行] 开始调研：量子计算面临的技术挑战
[并行] 完成调研：量子计算的基本原理
[并行] 完成调研：量子计算的当前应用
[并行] 完成调研：量子计算面临的技术挑战
并行耗时：4.0 秒
各主题调研结果：
[量子计算的基本原理]
[量子计算的当前应用]
[量子计算面临的技术挑战]
综合 Agent 整合结果...
综合报告：
Section 3：Pipeline（流水线）#
流水线模式中，每个 Agent 是一个处理阶段，前一个 Agent 的输出直接成为下一个的输入。
用户输入
→ [Planner] 制定大纲
→ [Researcher] 补充素材
→ [Writer] 撰写文章
→ [Editor] 润色优化
→ 最终输出
每个阶段都能看到前面所有阶段的工作，确保连贯性。
class Pipeline:
"""
流水线多 Agent 系统。
每个阶段接收上一阶段的输出，加工后传给下一阶段。
支持记录每个阶段的中间结果。
"""
def __init__(self):
self._stages = [] # [(name, system_prompt, temperature, max_tokens)]
self._results = {} # 记录每阶段输出
def add_stage(
self,
name: str,
system_prompt: str,
temperature: float = 0.7,
max_tokens: int = 512,
) -> "Pipeline":
"""添加流水线阶段（支持链式调用）"""
self._stages.append((name, system_prompt, temperature, max_tokens))
return self # 支持链式调用
def run(self, initial_input: str, verbose: bool = True) -> str:
"""运行完整流水线"""
self._results = {}
current_input = initial_input
if verbose:
print(f"\n{'='*60}")
print(f"流水线输入：{initial_input[:80]}")
print(f"阶段数：{len(self._stages)}")
print(f"{'='*60}")
for i, (name, system_prompt, temperature, max_tokens) in enumerate(self._stages):
if verbose:
print(f"\n[阶段 {i+1}/{len(self._stages)}] {name}")
print(f" 输入（前 100 字）：{current_input[:100]}..." if len(current_input) > 100 else f" 输入：{current_input}")
output = agent(
system_prompt=system_prompt,
user_message=current_input,
temperature=temperature,
max_tokens=max_tokens,
)
self._results[name] = output
current_input = output # 输出成为下一阶段的输入
if verbose:
print(f" 输出（前 150 字）：{output[:150]}..." if len(output) > 150 else f" 输出：{output}")
return current_input # 最后一个阶段的输出
def get_stage_output(self, stage_name: str) -> str:
"""获取指定阶段的输出"""
return self._results.get(stage_name, "该阶段未执行")
# ── 构建写作流水线 ─────────────────────────────────────────
writing_pipeline = (
Pipeline()
.add_stage(
name="Planner",
system_prompt=(
"你是一个内容规划师。根据用户的写作需求，制定一个详细的文章大纲。"
"大纲包含：标题、引言要点、3个主体段落的要点、结论要点。"
),
temperature=0.3,
max_tokens=400,
)
.add_stage(
name="Researcher",
system_prompt=(
"你是一个内容研究员。基于提供的文章大纲，为每个部分补充具体的事实、"
"数据、例子，让文章内容更加充实。输出增强后的大纲（保留原结构）。"
),
temperature=0.4,
max_tokens=600,
)
.add_stage(
name="Writer",
system_prompt=(
"你是一位优秀的科普作家。将提供的内容大纲扩展成一篇完整的文章。"
"文章要语言流畅、通俗易懂，避免过多专业术语。"
),
temperature=0.8,
max_tokens=800,
)
.add_stage(
name="Editor",
system_prompt=(
"你是一位资深编辑。润色文章：修正语法错误，改善句子流畅度，"
"确保逻辑连贯，去除冗余表达。输出最终优化版本。"
),
temperature=0.3,
max_tokens=800,
)
)
# 运行流水线
final_output = writing_pipeline.run(
initial_input="写一篇关于人工智能在医疗领域应用的科普文章，面向普通大众"
)
print("\n" + "="*60)
print("流水线完成！各阶段输出已记录。")
print(f"最终文章字数：{len(final_output)} 字")
============================================================
流水线输入：写一篇关于人工智能在医疗领域应用的科普文章，面向普通大众
阶段数：4
============================================================
[阶段 1/4] Planner
输入：写一篇关于人工智能在医疗领域应用的科普文章，面向普通大众
输出：
[阶段 2/4] Researcher
输入：
输出：
[阶段 3/4] Writer
输入：
输出（前 150 字）：我需要你把大纲贴过来，才能把它扩展开成完整文章。请把大纲内容粘贴在下一条消息里，同时告诉我以下几点偏好（可选，但有助于我更精准地写）：
- 主题/领域（如果大纲里没有明确写出）
- 目标读者（比如中学生、普通大众、大学生、科普爱好者等）
- 预计字数或篇幅（例如 500、1000、2000 字）
...
[阶段 4/4] Editor
输入（前 100 字）：我需要你把大纲贴过来，才能把它扩展开成完整文章。请把大纲内容粘贴在下一条消息里，同时告诉我以下几点偏好（可选，但有助于我更精准地写）：
- 主题/领域（如果大纲里没有明确写出）
- 目标读者（比如中...
输出：
============================================================
流水线完成！各阶段输出已记录。
最终文章字数：0 字
Section 4：投票表决（多 Agent 共识）#
对于高风险决策，单个 Agent 的答案可能有偏差。 让多个 Agent 独立回答，再由 Judge Agent 综合判断，能提高可靠性。
适用场景：
医疗建议
法律分析
重要的商业决策
试题解答验证
设置较高的 temperature 让每个 Agent 独立思考，避免趋同。
def voting_agents(
question: str,
num_voters: int = 3,
voter_temperature: float = 0.9,
voter_system: str = "你是一个专家，请独立、认真地回答问题。",
) -> dict:
"""
让多个 Agent 独立回答同一问题。
参数：
question: 要回答的问题
num_voters: 投票 Agent 数量
voter_temperature: 投票者的创意度（高温度 = 更多样化）
返回：
{"question": ..., "answers": [...], "judge_result": ...}
"""
print(f"\n{'='*60}")
print(f"问题：{question}")
print(f"投票 Agent 数：{num_voters}")
print(f"{'='*60}")
# 并行获取多个答案
answers = []
def get_answer(agent_id: int) -> str:
print(f" [Agent-{agent_id}] 独立思考中...")
result = agent(
system_prompt=voter_system,
user_message=question,
temperature=voter_temperature,
max_tokens=300,
)
print(f" [Agent-{agent_id}] 完成")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=num_voters) as executor:
futures = [executor.submit(get_answer, i + 1) for i in range(num_voters)]
for i, future in enumerate(concurrent.futures.as_completed(futures)):
answers.append(future.result())
# 打印各 Agent 答案
print("\n各 Agent 的独立答案：")
for i, ans in enumerate(answers):
print(f"\n[Agent-{i+1}]\n{ans}")
# Judge Agent 综合裁决
print("\n[JudgeAgent] 综合裁决中...")
answers_text = "\n\n".join(
f"答案 {i+1}：{ans}" for i, ans in enumerate(answers)
)
judge_result = agent(
system_prompt=(
"你是一个公正的裁判。评估多个专家对同一问题的回答，"
"指出各答案的优缺点，然后综合所有观点给出最佳答案。"
"格式：\n评估：（各答案分析）\n最佳答案：（综合后的答案）"
),
user_message=f"问题：{question}\n\n{answers_text}",
temperature=0.2,
max_tokens=500,
)
print(f"\n[JudgeAgent] 裁决结果：\n{judge_result}")
return {
"question": question,
"answers": answers,
"judge_result": judge_result,
}
# 演示：高风险决策场景
vote_result = voting_agents(
question="一家初创公司应该优先融资还是保持盈利？请给出建议和理由。",
num_voters=3,
voter_temperature=0.9,
voter_system="你是一个有丰富经验的创业顾问，根据你的独立判断回答。",
)
============================================================
问题：一家初创公司应该优先融资还是保持盈利？请给出建议和理由。
投票 Agent 数：3
============================================================
[Agent-1] 独立思考中...
[Agent-2] 独立思考中...
[Agent-3] 独立思考中...
[Agent-2] 完成
[Agent-3] 完成
[Agent-1] 完成
各 Agent 的独立答案：
[Agent-1]
[Agent-2]
[Agent-3]
[JudgeAgent] 综合裁决中...
[JudgeAgent] 裁决结果：
Section 5：Agent 间通信协议#
当 Agent 数量增多，需要一个标准化的消息格式来管理 Agent 间通信。
简单的消息格式：
{
"from": "agent_name",
"to": "agent_name",
"type": "request|response|error|broadcast",
"content": "消息内容",
"metadata": {"timestamp": ..., "priority": ...}
}
这是 AutoGen、CrewAI 等框架的基础思想。
import time as time_module
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class AgentMessage:
"""Agent 间通信的标准消息格式"""
from_agent: str
to_agent: str
msg_type: str # request | response | error | broadcast
content: str
metadata: dict = field(default_factory=dict)
timestamp: float = field(default_factory=time_module.time)
message_id: str = field(default_factory=lambda: f"msg_{int(time_module.time()*1000)}")
def to_dict(self) -> dict:
return {
"id": self.message_id,
"from": self.from_agent,
"to": self.to_agent,
"type": self.msg_type,
"content": self.content,
"metadata": self.metadata,
"timestamp": self.timestamp,
}
class MessageBus:
"""
简单的消息总线：Agent 通过消息总线通信。
所有消息都会被记录，便于调试和追溯。
"""
def __init__(self):
self._log: list[AgentMessage] = []
self._handlers: dict[str, callable] = {}
def register(self, agent_name: str, handler: callable):
"""注册 Agent 的消息处理函数"""
self._handlers[agent_name] = handler
print(f" [MessageBus] 注册 Agent：{agent_name}")
def send(self, message: AgentMessage) -> Optional[str]:
"""发送消息并返回目标 Agent 的响应"""
self._log.append(message)
print(f" [MessageBus] {message.from_agent} → {message.to_agent}: "
f"[{message.msg_type}] {message.content[:60]}")
if message.to_agent == "broadcast":
# 广播给所有 Agent
for name, handler in self._handlers.items():
if name != message.from_agent:
handler(message)
return None
handler = self._handlers.get(message.to_agent)
if handler:
return handler(message)
return f"错误：Agent '{message.to_agent}' 不存在"
def get_log(self) -> list[dict]:
"""获取完整消息日志"""
return [m.to_dict() for m in self._log]
# ── 构建基于消息总线的多 Agent 系统 ──────────────────────
bus = MessageBus()
# 定义各 Agent 的消息处理逻辑
def handle_analyst_request(msg: AgentMessage) -> str:
"""分析师 Agent：分析数据，返回洞察"""
analysis = agent(
system_prompt="你是一个数据分析师，从数据中提炼关键洞察。简洁回答。",
user_message=msg.content,
temperature=0.3,
max_tokens=200,
)
return analysis
def handle_writer_request(msg: AgentMessage) -> str:
"""写作 Agent：将分析结果写成报告"""
report = agent(
system_prompt="你是一个报告撰写专家，将数据洞察转化为清晰的报告段落。",
user_message=msg.content,
temperature=0.7,
max_tokens=300,
)
return report
def handle_manager_request(msg: AgentMessage) -> str:
"""经理 Agent：做出最终决策"""
decision = agent(
system_prompt="你是一个决策者，基于报告给出行动建议。用 1-3 个要点列出。",
user_message=msg.content,
temperature=0.3,
max_tokens=200,
)
return decision
# 注册 Agent
bus.register("Analyst", handle_analyst_request)
bus.register("Writer", handle_writer_request)
bus.register("Manager", handle_manager_request)
# 模拟 Agent 间通信流程
print("\nAgent 间通信协议演示")
print("="*50)
raw_data = "上月销售额：100万，本月：85万，环比下降 15%。主要下降来自华东区。"
# 1. 协调者发送分析请求给分析师
analysis = bus.send(AgentMessage(
from_agent="Coordinator",
to_agent="Analyst",
msg_type="request",
content=f"请分析以下销售数据：{raw_data}",
))
print(f"\n分析师回复：{analysis}")
# 2. 协调者将分析结果发给写作 Agent
report = bus.send(AgentMessage(
from_agent="Coordinator",
to_agent="Writer",
msg_type="request",
content=f"将以下分析写成简报：{analysis}",
))
print(f"\n写作 Agent 简报：{report}")
# 3. 协调者发给经理做决策
decision = bus.send(AgentMessage(
from_agent="Coordinator",
to_agent="Manager",
msg_type="request",
content=f"基于以下简报，给出行动建议：{report}",
))
print(f"\n经理决策：{decision}")
# 查看消息日志
print("\n消息总线日志：")
for log_entry in bus.get_log():
print(f" [{log_entry['from']} → {log_entry['to']}] {log_entry['type']}: {log_entry['content'][:50]}...")
[MessageBus] 注册 Agent：Analyst
[MessageBus] 注册 Agent：Writer
[MessageBus] 注册 Agent：Manager
Agent 间通信协议演示
==================================================
[MessageBus] Coordinator → Analyst: [request] 请分析以下销售数据：上月销售额：100万，本月：85万，环比下降 15%。主要下降来自华东区。
分析师回复：
[MessageBus] Coordinator → Writer: [request] 将以下分析写成简报：
写作 Agent 简报：
[MessageBus] Coordinator → Manager: [request] 基于以下简报，给出行动建议：
经理决策：
消息总线日志：
[Coordinator → Analyst] request: 请分析以下销售数据：上月销售额：100万，本月：85万，环比下降 15%。主要下降来自华东区。...
[Coordinator → Writer] request: 将以下分析写成简报：...
[Coordinator → Manager] request: 基于以下简报，给出行动建议：...
进阶：完整多 Agent 项目演示#
将以上所有模式组合，构建一个「智能内容生产系统」：
并行调研多个子主题
流水线处理（规划→研究→写作→编辑）
投票选出最佳标题
消息总线协调所有通信
def content_production_system(topic: str) -> str:
"""
完整的多 Agent 内容生产系统。
流程：
1. [规划 Agent] 将主题拆分为 3 个子主题
2. [并行调研] 同时调研所有子主题
3. [写作 Agent] 整合调研写成文章
4. [标题投票] 3 个 Agent 各提供标题，选最佳
5. 输出最终文章
"""
print(f"\n{'='*60}")
print(f"内容生产系统 - 主题：{topic}")
print(f"{'='*60}")
# Phase 1: 规划 - 拆分子主题
print("\n[Phase 1] 规划 Agent 拆分主题...")
subtopics_raw = agent(
system_prompt=(
"你是内容规划师。将主题拆分为 3 个子主题，以 JSON 数组输出。"
"示例：[\"子主题1\", \"子主题2\", \"子主题3\"]"
),
user_message=f"主题：{topic}",
temperature=0,
)
try:
content = subtopics_raw.strip()
if content.startswith("```"):
content = content.split("\n", 1)[1].rsplit("\n", 1)[0]
subtopics = json.loads(content)
if not isinstance(subtopics, list):
raise ValueError
except Exception:
subtopics = [f"{topic}的基础", f"{topic}的应用", f"{topic}的未来"]
print(f"子主题：{subtopics}")
# Phase 2: 并行调研所有子主题
print("\n[Phase 2] 并行调研子主题...")
research_data = parallel_research(subtopics, max_workers=3)
# Phase 3: 写作整合
print("\n[Phase 3] 写作 Agent 整合内容...")
combined_research = "\n\n".join(
f"【{sub}】\n{res}" for sub, res in research_data.items()
)
article_body = agent(
system_prompt="你是科普作家，将多个研究片段整合为连贯的文章正文（不含标题）。",
user_message=f"主题：{topic}\n\n研究资料：\n{combined_research}",
temperature=0.7,
max_tokens=700,
)
# Phase 4: 标题投票
print("\n[Phase 4] 标题投票...")
def get_title(agent_id: int) -> str:
return agent(
system_prompt="你是标题策划师，为文章起一个吸引眼球的标题（20字以内）。只输出标题本身。",
user_message=f"文章主题：{topic}\n文章正文（摘要）：{article_body[:300]}",
temperature=0.9,
max_tokens=50,
)
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
title_futures = [executor.submit(get_title, i) for i in range(3)]
candidate_titles = [f.result() for f in concurrent.futures.as_completed(title_futures)]
print(f"候选标题：{candidate_titles}")
best_title = agent(
system_prompt="你是编辑，从候选标题中选出最佳的一个。只输出标题本身。",
user_message=f"候选标题：{json.dumps(candidate_titles, ensure_ascii=False)}\n文章主题：{topic}",
temperature=0,
)
print(f"最佳标题：{best_title}")
# 组合最终文章
final_article = f"# {best_title.strip()}\n\n{article_body}"
print(f"\n{'='*60}")
print("最终文章：")
print(final_article)
return final_article
# 运行完整演示
result = content_production_system("人工智能对教育的影响")
============================================================
内容生产系统 - 主题：人工智能对教育的影响
============================================================
[Phase 1] 规划 Agent 拆分主题...
子主题：['个性化学习与教学技术', '教师角色转变与专业发展', '教育公平、隐私与伦理挑战']
[Phase 2] 并行调研子主题...
[并行] 开始调研：个性化学习与教学技术
[并行] 开始调研：教师角色转变与专业发展
[并行] 开始调研：教育公平、隐私与伦理挑战
[并行] 完成调研：个性化学习与教学技术
[并行] 完成调研：教育公平、隐私与伦理挑战
[并行] 完成调研：教师角色转变与专业发展
[Phase 3] 写作 Agent 整合内容...
[Phase 4] 标题投票...
候选标题：['', '', '']
最佳标题：
============================================================
最终文章：
#
练习题#
扩展 Orchestrator：让 Orchestrator 能够动态决定调用哪些 Worker，而不是固定流程。提示：让 Orchestrator 返回
{"workers": ["research", "writer"]}格式的计划。错误容忍的并行 Worker：当某个 Worker 失败时，Pipeline 应该能够跳过或用备选 Agent 重试。修改
parallel_research添加重试逻辑。加权投票：在投票表决中，根据每个 Agent 的历史准确率给它们不同的权重。如何追踪和更新 Agent 的历史表现？
自适应 Pipeline：设计一个 Pipeline，其中每个阶段可以决定「需要返回上一阶段修改」（类似人类审稿流程中的返修）。
总结：何时使用哪种模式？#
四种模式对比#
模式 |
适用场景 |
优点 |
缺点 |
|---|---|---|---|
Orchestrator-Worker |
需要规划和协调的复杂任务 |
灵活，可动态分配 |
Orchestrator 是单点故障 |
并行 Worker |
可拆分的独立子任务 |
速度快，效率高 |
结果需要额外整合 |
Pipeline |
有明确处理顺序的任务 |
逻辑清晰，易维护 |
串行，速度较慢 |
投票表决 |
高风险、需要可靠性的决策 |
减少单一偏差，更可靠 |
成本高（N 倍 API 调用） |
选择建议#
任务能否拆分为独立子任务？
是 → 用并行 Worker（速度优先）
否 ↓
任务有固定处理顺序？
是 → 用 Pipeline
否 ↓
需要高可靠性？
是 → 用投票表决
否 → 用 Orchestrator-Worker
生产建议#
设置超时和重试机制（网络不稳定时 Agent 调用可能失败）
记录所有 Agent 的输入输出（便于调试和成本分析）
控制并发数量（避免触发 API 速率限制）
使用
asyncio替代ThreadPoolExecutor获得更好的并发性能
# 模式选择速查工具
def recommend_pattern(task_description: str) -> str:
"""根据任务描述推荐多 Agent 模式"""
recommendation = agent(
system_prompt=(
"你是多 Agent 系统架构专家。根据任务描述，推荐最合适的 Agent 模式。"
"可选模式：Orchestrator-Worker、并行 Worker、Pipeline、投票表决。"
"给出推荐和理由（50字以内）。"
),
user_message=f"任务：{task_description}",
temperature=0.3,
)
return recommendation
# 测试几个典型场景
test_tasks = [
"分析多份财务报表，判断公司是否值得投资",
"将一篇英文论文翻译成中文，再进行技术校对",
"同时收集 10 个城市的天气数据做对比分析",
"生成一篇复杂的市场调研报告，需要规划、数据收集、分析、写作",
]
print("多 Agent 模式推荐工具")
print("="*50)
for task in test_tasks:
rec = recommend_pattern(task)
print(f"\n任务：{task}")
print(f"推荐：{rec}")
多 Agent 模式推荐工具
==================================================
任务：分析多份财务报表，判断公司是否值得投资
推荐：推荐：Orchestrator-Worker。理由：协调专长子Agent并综合多份报表分析结论。
任务：将一篇英文论文翻译成中文，再进行技术校对
推荐：推荐：Pipeline。理由：分阶段先翻译再技术校对，职责清晰，易衔接。
任务：同时收集 10 个城市的天气数据做对比分析
推荐：
任务：生成一篇复杂的市场调研报告，需要规划、数据收集、分析、写作
推荐：
# 知识检验：多 Agent 系统成本估算
def estimate_cost(pattern: str, num_agents: int, turns: int) -> dict:
"""估算不同多 Agent 模式的 API 调用次数"""
estimates = {
"Orchestrator-Worker": {
"调用次数公式": "1（编排）+ num_workers（执行）",
"调用次数": 1 + num_agents,
"并发性": "Worker 可并行",
},
"Pipeline": {
"调用次数公式": "num_stages（串行）",
"调用次数": num_agents,
"并发性": "串行",
},
"并行Worker": {
"调用次数公式": "num_workers + 1（整合）",
"调用次数": num_agents + 1,
"并发性": "全部并行",
},
"投票表决": {
"调用次数公式": "num_voters + 1（裁判）",
"调用次数": num_agents + 1,
"并发性": "投票者可并行",
},
}
return estimates.get(pattern, {"错误": "未知模式"})
print("各模式 API 调用次数对比（5个Agent，单轮任务）")
print("="*50)
for pattern in ["Orchestrator-Worker", "Pipeline", "并行Worker", "投票表决"]:
est = estimate_cost(pattern, num_agents=5, turns=1)
print(f"\n{pattern}")
for k, v in est.items():
print(f" {k}: {v}")
各模式 API 调用次数对比（5个Agent，单轮任务）
==================================================
Orchestrator-Worker
调用次数公式: 1（编排）+ num_workers（执行）
调用次数: 6
并发性: Worker 可并行
Pipeline
调用次数公式: num_stages（串行）
调用次数: 5
并发性: 串行
并行Worker
调用次数公式: num_workers + 1（整合）
调用次数: 6
并发性: 全部并行
投票表决
调用次数公式: num_voters + 1（裁判）
调用次数: 6
并发性: 投票者可并行
# 本章总结
print("="*60)
print("第 5 章 Agent 系统 - 学习路径总结")
print("="*60)
chapters = [
("01_react.ipynb", "ReAct 模式", "Thought→Action→Observation 循环"),
("02_memory.ipynb", "Agent 记忆", "滑动窗口/摘要/实体/向量记忆"),
("03_multi_agent.ipynb", "多 Agent 系统", "Orchestrator/Pipeline/投票/消息总线"),
]
for fn, title, desc in chapters:
print(f" {fn} {title}: {desc}")
print("\n下一步建议：")
print(" - 尝试将 ReAct + 记忆 + 多 Agent 组合成完整系统")
print(" - 学习 LangGraph、AutoGen、CrewAI 等成熟框架")
print(" - 为 Agent 添加人工审核节点（Human-in-the-loop）")
print(f"\n当前使用模型：{MODEL}")
============================================================
第 5 章 Agent 系统 - 学习路径总结
============================================================
01_react.ipynb ReAct 模式: Thought→Action→Observation 循环
02_memory.ipynb Agent 记忆: 滑动窗口/摘要/实体/向量记忆
03_multi_agent.ipynb 多 Agent 系统: Orchestrator/Pipeline/投票/消息总线
下一步建议：
- 尝试将 ReAct + 记忆 + 多 Agent 组合成完整系统
- 学习 LangGraph、AutoGen、CrewAI 等成熟框架
- 为 Agent 添加人工审核节点（Human-in-the-loop）
当前使用模型：openai/gpt-5-mini