Agent 架构模式 - 从单 Agent 到多 Agent 的工程范式 / Agent Architecture Patterns
📅 创建时间：2026-07-28 🏷️ 标签：#AgentArchitecture #PromptChaining #Routing #Orchestrator #MultiAgent 📚 前置知识：[[05-agent-workflow]] [[06-multi-agent]]
📋 本章目标
- 理解 Agent 系统的五种核心架构模式及其数据流
- 掌握每种模式的适用场景与局限性
- 能够根据任务特征选择合适的架构模式
- 理解复合模式：如何将多种模式组合成生产级系统
- 建立"先简单后复杂"的架构选型直觉
第0部分：为什么要关心架构模式？
0.1 从单个 Agent 循环到系统设计
你已经知道单个 Agent 的 while 循环是怎么工作的：
┌─────────────────────────────────────────────────────────────┐
│ 单 Agent 循环（回顾） │
├─────────────────────────────────────────────────────────────┤
│ │
│ messages = [system_prompt, user_question] │
│ │
│ ┌─────────────────┐ │
│ │ LLM API 调用 │←── messages + tools │
│ └────────┬────────┘ │
│ │ │
│ finish_reason? │
│ ┌────┴────┐ │
│ "stop" "tool_calls" │
│ │ │ │
│ 返回文本 执行工具 → 结果追加到 messages → 循环 │
│ │
└─────────────────────────────────────────────────────────────┘但当你面对真实业务问题时，单个 Agent 往往不够：
- 一个客服工单可能涉及退款审查、技术诊断、情绪安抚三个完全不同的能力域
- 一个代码仓库分析需要同时检查几十个文件，每个文件的分析互相独立
- 一份法律合同的撰写需要起草、审核、修改多轮迭代
把所有这些逻辑塞进一个 Agent 的 system prompt 里，结果就是 prompt 越来越长、行为越来越不可控、调试越来越困难。
0.2 Anthropic 的核心理念：从简单开始
Anthropic 在 2024 年底发布的 "Building effective agents" 博文中提出了一个关键原则：
┌─────────────────────────────────────────────────────────────┐
│ 架构复杂度递增原则 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 简单模式（Prompt Chaining / Routing） │
│ ↓ 当简单模式不够用时 │
│ 中等模式（Parallelization） │
│ ↓ 当需要动态决策时 │
│ 复杂模式（Orchestrator-Workers / Evaluator-Optimizer） │
│ │
│ 核心思想：不要一上来就上最复杂的架构。 │
│ 能用 prompt chaining 解决的，不要上 orchestrator。 │
│ │
└─────────────────────────────────────────────────────────────┘下面我们逐一拆解这五种模式，每种都包含数据流图、适用场景、代码骨架。
第1部分：Prompt Chaining —— 最简单的流水线
1.1 什么是 Prompt Chaining？
Prompt Chaining 是最直观的多步骤模式：前一步的输出作为后一步的输入，按固定顺序执行。
┌─────────────────────────────────────────────────────────────┐
│ Prompt Chaining 数据流 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Step A │────→│ Step B │────→│ Step C │──→ 最终输出 │
│ │ (提取) │ │ (翻译) │ │ (校对) │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │
│ 输出原文 输出译稿 输出校对稿 │
│ (结构化数据) (文本) (文本+批注) │
│ │
│ 每一步都是独立的 LLM 调用，有自己的 system prompt │
│ 步骤之间通过程序传递数据（不是 LLM 自动传递） │
│ │
└─────────────────────────────────────────────────────────────┘关键特征：
| 特征 | 说明 |
|---|---|
| 步骤顺序 | 固定，编译期确定 |
| 步骤数量 | 预先知道 |
| 步骤间依赖 | 单向，无循环 |
| 每步的 LLM | 可以不同（小模型提取，大模型翻译） |
| 失败处理 | 某步失败则链中断，可重试该步 |
1.2 适用与不适用场景
┌─────────────────────────────────────────────────────────────┐
│ Prompt Chaining 适用性判断 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ✅ 适用： │
│ • 任务可以清晰分解为固定步骤（文档翻译、数据ETL） │
│ • 每一步有明确的输入输出契约 │
│ • 需要在步骤间做确定性处理（格式转换、校验） │
│ • 希望每一步可独立调试和评估 │
│ • 不同步骤适合用不同模型（成本优化） │
│ │
│ ❌ 不适用： │
│ • 步骤数量或顺序无法预先确定 │
│ • 需要根据中间结果动态决定下一步做什么 │
│ • 步骤间需要反复迭代（用 Evaluator-Optimizer） │
│ • 任务本身很简单，单次 LLM 调用就能完成 │
│ │
└─────────────────────────────────────────────────────────────┘1.3 实例：文档翻译流水线
┌─────────────────────────────────────────────────────────────┐
│ 文档翻译 Prompt Chaining 实战 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入：一篇英文技术文档 │
│ │
│ Step 1: 提取原文 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ System: "提取以下 Markdown 中的正文段落，忽略代码块， │ │
│ │ 输出 JSON 数组，每段一个元素" │ │
│ │ Input: raw_markdown │ │
│ │ Output: ["段落1原文", "段落2原文", ...] │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Step 2: 逐段翻译 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ System: "你是技术文档翻译专家。将英文段落翻译为中文， │ │
│ │ 保留技术术语的英文原名（加括号标注中文）" │ │
│ │ Input: paragraphs[i] │ │
│ │ Output: "段落i中文翻译" │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Step 3: 校对 + 术语一致性检查 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ System: "对照原文和译文，检查：1)术语一致性 2)漏译 │ │
│ │ 3)错译 4)中文流畅度。输出校对报告+修正稿" │ │
│ │ Input: 原文 + 译文 │ │
│ │ Output: { "score": 95, "issues": [...], "final": "..."} │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.4 代码骨架
python
from dataclasses import dataclass
from typing import Any
@dataclass
class ChainStep:
"""流水线中的一个步骤"""
name: str
system_prompt: str
model: str = "gpt-4o-mini" # 提取/校对可用小模型
output_format: str = "text" # "text" | "json"
class PromptChain:
"""固定顺序的 Prompt 流水线"""
def __init__(self, steps: list[ChainStep]):
self.steps = steps
def run(self, initial_input: str) -> dict[str, Any]:
"""按顺序执行每个步骤，每步的输出作为下一步的输入"""
results = {}
current_input = initial_input
for step in self.steps:
print(f"[Chain] 执行: {step.name}")
response = self._call_llm(
system=step.system_prompt,
user_input=current_input,
model=step.model,
format=step.output_format
)
results[step.name] = response
current_input = response # ← 核心：输出变输入
return results
def _call_llm(self, system: str, user_input: str,
model: str, format: str) -> str:
# 实际调用 LLM API（简化示意）
...
# --- 使用示例：文档翻译 ---
chain = PromptChain(steps=[
ChainStep(
name="extract",
system_prompt="提取 Markdown 正文段落，输出 JSON 数组",
output_format="json"
),
ChainStep(
name="translate",
system_prompt="你是技术翻译专家，将英文译为中文",
model="gpt-4o" # 翻译用强模型
),
ChainStep(
name="review",
system_prompt="对照原文校对译文，检查术语一致性和流畅度",
output_format="json"
),
])
result = chain.run(initial_input=raw_markdown)
# result = {
# "extract": '["段落1", "段落2", ...]',
# "translate": "中文译文全文...",
# "review": '{"score": 95, "final": "修正后译文..."}'
# }1.5 关键设计决策
┌─────────────────────────────────────────────────────────────┐
│ Prompt Chaining 设计 Checklist │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 步骤粒度：每步只做一件事 │
│ ❌ 一步内完成"提取+翻译+校对" │
│ ✅ 三步分开，每步 prompt 短而精确 │
│ │
│ 2. 中间数据格式：约定好每步的输入输出 schema │
│ ❌ "把结果传给下一步"（模糊） │
│ ✅ "输出 JSON: {paragraphs: string[], metadata: {...}}" │
│ │
│ 3. 模型选择：不同步骤用不同模型 │
│ • 提取/分类 → gpt-4o-mini（便宜、快） │
│ • 翻译/生成 → gpt-4o / claude-sonnet（质量优先） │
│ │
│ 4. 错误处理：每步独立的 try-catch + 重试 │
│ • Step 2 失败不应影响 Step 1 的结果 │
│ • 记录每步的输入输出，方便事后调试 │
│ │
└─────────────────────────────────────────────────────────────┘第2部分：Routing —— 智能分流
2.1 什么是 Routing？
Routing 模式在流水线的最前端加一个分类器，根据输入特征将任务路由到不同的专门处理器。
┌─────────────────────────────────────────────────────────────┐
│ Routing 数据流 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ 分类器 (Classifier) │ │
│ │ "这是什么类型的问题？" │ │
│ │ 输出: {category, confidence, reason} │ │
│ └────────┬──────────┬──────────┬────────────────────┘ │
│ │ │ │ │
│ refund technical complaint │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 退款 Agent│ │ 技术 Agent│ │ 投诉 Agent│ │
│ │ │ │ │ │ │ │
│ │ 查订单 │ │ 诊断问题 │ │ 情绪安抚 │ │
│ │ 验政策 │ │ 给方案 │ │ 记录升级 │ │
│ │ 执行退款 │ │ 创建工单 │ │ 承诺回访 │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ └─────────────┴─────────────┘ │
│ │ │
│ ▼ │
│ 统一响应格式化 │
│ │ │
│ ▼ │
│ 返回给用户 │
│ │
└─────────────────────────────────────────────────────────────┘关键特征：
| 特征 | 说明 |
|---|---|
| 决策点 | 单一，在入口处 |
| 分支数量 | 有限且预先定义 |
| 分支间关系 | 互斥，一次请求只走一条分支 |
| 分类器 | 通常是轻量级 LLM 调用 + 结构化输出 |
| 可扩展性 | 加新分支 = 加新 Agent + 更新分类器 prompt |
2.2 适用与不适用场景
┌─────────────────────────────────────────────────────────────┐
│ Routing 适用性判断 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ✅ 适用： │
│ • 输入类型明显不同，需要不同的处理逻辑 │
│ • 每种类型的处理逻辑相对独立，可以单独优化 │
│ • 希望对不同类型的请求做不同的成本/延迟策略 │
│ • 简单问题快速回答，复杂问题深度处理（模型分层） │
│ │
│ ❌ 不适用： │
│ • 一个请求同时涉及多个类别（如"退款+投诉"）— 考虑并行 │
│ • 类别边界模糊，分类器准确率低 │
│ • 处理逻辑高度相似，分开反而增加维护成本 │
│ • 需要不同 Agent 之间协作完成一个任务 │
│ │
└─────────────────────────────────────────────────────────────┘2.3 实例：智能客服路由系统
┌─────────────────────────────────────────────────────────────┐
│ 智能客服 Routing 实战 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户: "我上周买的显示器有坏点，我要退款！" │
│ │
│ 分类器输出: │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ { │ │
│ │ "category": "refund", │ │
│ │ "sub_category": "quality_defect", │ │
│ │ "confidence": 0.95, │ │
│ │ "extracted_info": { │ │
│ │ "product": "显示器", │ │
│ │ "purchase_time": "上周", │ │
│ │ "issue": "坏点" │ │
│ │ } │ │
│ │ } │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ 路由到 refund_agent │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 退款 Agent 处理流程: │ │
│ │ 1. 根据 "显示器" + "上周" 查询订单 │ │
│ │ 2. 检查退款政策: 质量问题 → 7天内可退 │ │
│ │ 3. 生成退货单号，发送退货指引 │ │
│ │ 4. 输出: "已为您创建退货单 #R20260728，请..." │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘2.4 代码骨架
python
from enum import Enum
from dataclasses import dataclass
class Category(str, Enum):
REFUND = "refund"
TECHNICAL = "technical"
COMPLAINT = "complaint"
GENERAL = "general"
@dataclass
class RoutingDecision:
category: Category
confidence: float
reasoning: str
extracted_info: dict
class Router:
"""分类器 + 路由表"""
ROUTING_PROMPT = """分析用户消息，判断属于以下哪个类别:
- refund: 退款、退货、订单取消相关
- technical: 产品使用问题、故障排查、技术咨询
- complaint: 服务投诉、人员投诉、体验问题
- general: 一般咨询、产品信息询问
输出 JSON: {"category": "...", "confidence": 0.0-1.0,
"reasoning": "...", "extracted_info": {...}}"""
def __init__(self):
self.handlers: dict[Category, callable] = {}
def register(self, category: Category, handler: callable):
"""注册一个分类对应的处理函数"""
self.handlers[category] = handler
def route(self, user_message: str) -> str:
"""分类 → 路由 → 执行 → 返回"""
# Step 1: 分类
decision = self._classify(user_message)
print(f"[Router] → {decision.category.value} "
f"(confidence: {decision.confidence})")
# Step 2: 低置信度走兜底
if decision.confidence < 0.7:
decision.category = Category.GENERAL
# Step 3: 路由到对应 handler
handler = self.handlers.get(decision.category)
if not handler:
return "抱歉，我无法处理您的请求。"
return handler(user_message, decision.extracted_info)
def _classify(self, message: str) -> RoutingDecision:
"""调用 LLM 做分类（强制结构化输出）"""
response = llm.chat(
messages=[
{"role": "system", "content": self.ROUTING_PROMPT},
{"role": "user", "content": message}
],
response_format={"type": "json_object"}
)
data = json.loads(response.content)
return RoutingDecision(
category=Category(data["category"]),
confidence=data["confidence"],
reasoning=data["reasoning"],
extracted_info=data.get("extracted_info", {})
)
# --- 使用示例 ---
router = Router()
# 注册各分类的处理逻辑
router.register(Category.REFUND, handle_refund)
router.register(Category.TECHNICAL, handle_technical)
router.register(Category.COMPLAINT, handle_complaint)
router.register(Category.GENERAL, handle_general)
# 处理用户请求
response = router.route("我上周买的显示器有坏点，我要退款！")2.5 分类器设计要点
┌─────────────────────────────────────────────────────────────┐
│ 分类器设计 Checklist │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 类别定义要互斥 + 穷尽 │
│ • 加一个 "general" / "other" 兜底类别 │
│ • 每个类别有清晰的边界描述和正反例 │
│ │
│ 2. 分类器要提取上下文信息，不只是分类标签 │
│ ❌ output: "refund" │
│ ✅ output: {category, confidence, extracted_info} │
│ 下游 handler 不应该重新解析原始消息 │
│ │
│ 3. 设置置信度阈值 │
│ • confidence > 0.8 → 直接路由 │
│ • 0.5 < confidence < 0.8 → 追问确认 │
│ • confidence < 0.5 → 走兜底或转人工 │
│ │
│ 4. 分类器用便宜的小模型 │
│ • gpt-4o-mini 分类准确率通常 > 95% │
│ • 省钱 + 低延迟 │
│ │
└─────────────────────────────────────────────────────────────┘第3部分：Parallelization —— 并行执行
3.1 什么是 Parallelization？
当任务可以分解为互不依赖的子任务时，同时执行它们，然后汇总结果。
┌─────────────────────────────────────────────────────────────┐
│ Parallelization 数据流 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户: "对比 OpenAI、Anthropic、Google 最新模型的价格" │
│ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ 任务分解 (Sectioning) │ │
│ │ 生成子任务: │ │
│ │ [搜索OpenAI定价, 搜索Anthropic定价, 搜索Google定价] │ │
│ └───┬───────────────┬───────────────┬───────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────────┐ ┌────────────┐ │
│ │Worker A│ │ Worker B │ │ Worker C │ │
│ │搜索 │ │ 搜索 │ │ 搜索 │ │
│ │OpenAI │ │ Anthropic │ │ Google │ │
│ │定价 │ │ 定价 │ │ 定价 │ │
│ └───┬────┘ └─────┬──────┘ └─────┬──────┘ │
│ │ │ │ │
│ │ 并行执行 │ │ │
│ │ (asyncio / │ │ │
│ │ threads) │ │ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────────┐ ┌────────────┐ │
│ │结果A │ │ 结果B │ │ 结果C │ │
│ │GPT-4o: │ │Sonnet: │ │Gemini: │ │
│ │$2.5/1M │ │$3/1M │ │$1.25/1M │ │
│ └───┬────┘ └─────┬──────┘ └─────┬──────┘ │
│ │ │ │ │
│ └───────────────┴───────────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ 聚合 (Aggregation) │ │
│ │ LLM 将三个结果合并为对比表格 + 分析结论 │ │
│ └───────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 最终输出 │
│ │
└─────────────────────────────────────────────────────────────┘关键特征：
| 特征 | 说明 |
|---|---|
| 子任务关系 | 无依赖，这是并行的前提 |
| 执行方式 | 真正的并发（asyncio / 线程池） |
| 结果汇总 | 所有子任务完成后，聚合输出 |
| 容错 | 单个子任务失败不影响其他，可降级 |
| 加速比 | 理想情况下 N 倍（受最慢子任务限制） |
3.2 适用与不适用场景
┌─────────────────────────────────────────────────────────────┐
│ Parallelization 适用性判断 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ✅ 适用： │
│ • 子任务真正互相独立（无数据依赖） │
│ • 需要从多个数据源获取信息后汇总 │
│ • 需要对同一输入做多维度分析（安全、性能、风格） │
│ • 子任务耗时相近，并行能显著降低端到端延迟 │
│ • 对同一内容做多次采样/投票（提高可靠性） │
│ │
│ ❌ 不适用： │
│ • 子任务有依赖关系（B 需要 A 的输出）→ 用 Chaining │
│ • 子任务数量或内容无法预先确定 → 用 Orchestrator-Workers │
│ • 一个 LLM 调用就能完成（并行反而增加成本） │
│ • 需要子任务之间实时通信或协调 │
│ │
└─────────────────────────────────────────────────────────────┘3.3 两种并行范式
┌─────────────────────────────────────────────────────────────┐
│ 并行化的两种范式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 范式 A: Sectioning（分段并行） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 同一任务 → 拆成 N 个独立子任务 → 各自执行 → 汇总 │ │
│ │ │ │
│ │ 举例： │ │
│ │ • 搜索三个不同的数据源 │ │
│ │ • 分析代码库中的 N 个文件 │ │
│ │ • 翻译文档的 N 个段落 │ │
│ │ │ │
│ │ 特点：子任务做的是同类工作，只是数据不同 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 范式 B: Voting（投票并行） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 同一输入 → 发给 N 个不同 Agent/模型 → 投票/择优 │ │
│ │ │ │
│ │ 举例： │ │
│ │ • 代码审查：3 个 reviewer 各自审查，投票决定是否合并 │ │
│ │ • 内容审核：多个审核员独立判断，降低误判率 │ │
│ │ • 生成多个候选回答，用评估模型选最佳 │ │
│ │ │ │
│ │ 特点：同一输入，不同视角，投票保证可靠性 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.4 代码骨架
python
import asyncio
from dataclasses import dataclass
@dataclass
class SubTask:
id: str
system_prompt: str
user_input: str
model: str = "gpt-4o-mini"
@dataclass
class SubResult:
task_id: str
output: str
success: bool
error: str | None = None
class ParallelRunner:
"""并行执行多个独立的 LLM 调用"""
def __init__(self, max_concurrency: int = 5):
self.semaphore = asyncio.Semaphore(max_concurrency)
async def run(self, tasks: list[SubTask],
aggregator_prompt: str) -> str:
"""并行执行所有子任务，然后聚合结果"""
# Step 1: 并行执行
async def execute_one(task: SubTask) -> SubResult:
async with self.semaphore:
try:
output = await self._call_llm_async(
system=task.system_prompt,
user_input=task.user_input,
model=task.model
)
return SubResult(
task_id=task.id, output=output,
success=True
)
except Exception as e:
return SubResult(
task_id=task.id, output="",
success=False, error=str(e)
)
results: list[SubResult] = await asyncio.gather(
*[execute_one(t) for t in tasks]
)
# Step 2: 检查失败的子任务
failed = [r for r in results if not r.success]
succeeded = [r for r in results if r.success]
print(f"[Parallel] {len(succeeded)}/{len(tasks)} 成功"
+ (f", {len(failed)} 失败" if failed else ""))
# Step 3: 聚合
results_text = "\n---\n".join(
f"[{r.task_id}]\n{r.output}" for r in succeeded
)
aggregated = await self._call_llm_async(
system=aggregator_prompt,
user_input=results_text
)
return aggregated
async def _call_llm_async(self, system: str,
user_input: str,
model: str) -> str:
# 异步 LLM API 调用（简化示意）
...
# --- 使用示例：多数据源搜索对比 ---
async def compare_model_pricing():
runner = ParallelRunner(max_concurrency=3)
tasks = [
SubTask(
id="openai",
system_prompt="搜索 OpenAI 最新模型定价信息，输出结构化数据",
user_input="GPT-4o, GPT-4o-mini 当前 API 价格"
),
SubTask(
id="anthropic",
system_prompt="搜索 Anthropic 最新模型定价信息，输出结构化数据",
user_input="Claude Sonnet, Claude Haiku 当前 API 价格"
),
SubTask(
id="google",
system_prompt="搜索 Google 最新模型定价信息，输出结构化数据",
user_input="Gemini 1.5 Pro, Gemini 1.5 Flash 当前 API 价格"
),
]
aggregator_prompt = """将以下多个搜索结果合并为一个对比表格。
包含: 模型名、输入价格($/1M tokens)、输出价格($/1M tokens)、备注"""
result = await runner.run(tasks, aggregator_prompt)
print(result)3.5 并行化陷阱
┌─────────────────────────────────────────────────────────────┐
│ Parallelization 常见陷阱 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 陷阱1: 假并行 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ for task in tasks: │ │
│ │ result = await call_llm(task) ← 串行！ │ │
│ │ 必须用 asyncio.gather / ThreadPoolExecutor │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 陷阱2: 并发数过多导致限流 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 同时发 100 个请求 → 触发 API rate limit → 全部失败 │ │
│ │ 解决: Semaphore 控制并发数，配合重试+退避 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 陷阱3: 聚合质量依赖于每个子任务的质量 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 一个子任务返回垃圾数据 → 污染整个聚合结果 │ │
│ │ 解决: 每个子任务做输出校验，不合格的丢弃或重试 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 陷阱4: 成本线性增长 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ N 个并行任务 = N 倍的 token 消耗 │ │
│ │ 确保并行带来的价值 > 额外的成本 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘第4部分：Orchestrator-Workers —— 动态任务调度
4.1 什么是 Orchestrator-Workers？
这是最复杂也最强大的模式。一个中央 Orchestrator（编排器）动态分析任务，拆解为子任务，分配给多个 Worker 执行，并根据中间结果动态调整后续计划。
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator-Workers 数据流 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户: "分析这个代码仓库的安全性" │
│ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Orchestrator (编排器) │ │
│ │ │ │
│ │ 1. 分析任务: "安全审查" │ │
│ │ 2. 制定计划: │ │
│ │ - 扫描依赖漏洞 │ │
│ │ - 检查硬编码密钥 │ │
│ │ - 审查认证逻辑 │ │
│ │ - 检查 SQL 注入 │ │
│ │ 3. 分配 Worker: │ │
│ │ Worker A → 扫描依赖 │ │
│ │ Worker B → 搜索密钥 │ │
│ └───┬───────────────────────────────────────────────┘ │
│ │ │
│ │ 分发任务 │
│ │ │
│ ┌───▼──────────┐ ┌──────────────┐ │
│ │ Worker A │ │ Worker B │ ... │
│ │ 扫描依赖漏洞 │ │ 搜索硬编码 │ │
│ │ │ │ 密钥 │ │
│ │ 工具: │ │ 工具: │ │
│ │ - pip audit │ │ - grep │ │
│ │ - npm audit │ │ - truffleHog│ │
│ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │
│ │ 返回结果 │ │
│ ▼ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Orchestrator 再次评估 │ │
│ │ │ │
│ │ Worker A 报告: 发现 3 个高危依赖漏洞 │ │
│ │ → 动态决策: 创建 Worker C 深入分析漏洞利用路径 │ │
│ │ │ │
│ │ Worker B 报告: 发现 2 处硬编码 API Key │ │
│ │ → 动态决策: 已完成，无需更多分析 │ │
│ │ │ │
│ │ Worker C 返回: 漏洞详情 + 修复建议 │ │
│ │ → 动态决策: 所有子任务完成，开始汇总 │ │
│ └────────────────────┬──────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ 最终汇总报告 │ │
│ │ • 依赖漏洞: 3个（高危）+ 修复方案 │ │
│ │ • 密钥泄露: 2处 + 轮换建议 │ │
│ │ • 整体安全评分: 6.2/10 │ │
│ └───────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘关键特征：
| 特征 | 说明 |
|---|---|
| 任务分解 | 运行时动态决定，不是预先写死 |
| Worker 角色 | 可以同构（相同 prompt 不同数据）或异构（不同能力） |
| 控制流 | Orchestrator 集中决策，Worker 只执行不决策 |
| 反馈循环 | Worker 结果会触发 Orchestrator 重新规划 |
| 终止条件 | Orchestrator 判断任务完成，或达到最大步数 |
4.2 适用与不适用场景
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator-Workers 适用性判断 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ✅ 适用： │
│ • 任务无法预先分解，需要根据中间结果动态调整 │
│ • 需要多种不同能力的 Worker 协同（搜索、分析、执行） │
│ • 子任务之间有隐含依赖，需要编排器协调顺序 │
│ • 复杂且开放式的任务（代码审查、研究报告、系统诊断） │
│ │
│ ❌ 不适用： │
│ • 任务可以用 Prompt Chaining 固定步骤解决 │
│ • 不需要动态决策（用 Orchestrator 是杀鸡用牛刀） │
│ • 子任务完全独立且可预知（直接用 Parallelization） │
│ • 对延迟敏感、需要快速响应（O-W 的协调开销大） │
│ │
└─────────────────────────────────────────────────────────────┘4.3 Orchestrator 的核心循环
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator 决策循环详解 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────┐ │
│ │ │ │
│ │ PLAN → DISPATCH → COLLECT → DECIDE │ │
│ │ ↑ │ │ │
│ │ └─────────────── 循环 ──────────────┘ │ │
│ │ │ │
│ │ PLAN: 分析当前状态，制定下一步计划 │ │
│ │ DISPATCH: 分配任务给合适的 Worker │ │
│ │ COLLECT: 收集 Worker 返回的结果 │ │
│ │ DECIDE: 判断是继续还是结束 │ │
│ │ │ │
│ └─────────────────────────────────────────────┘ │
│ │
│ 终止条件: │
│ • 所有必要信息已收集齐 │
│ • 达到最大迭代次数（防止死循环） │
│ • Orchestrator 判断继续下去也不会有新信息 │
│ │
└─────────────────────────────────────────────────────────────┘4.4 代码骨架
python
import json
from dataclasses import dataclass, field
@dataclass
class Task:
"""Orchestrator 分配的任务"""
id: str
description: str
worker_type: str # 指定用哪种 Worker
context: dict # Worker 需要的上下文
@dataclass
class TaskResult:
"""Worker 返回的结果"""
task_id: str
output: str
findings: list[dict] # 结构化的发现
confidence: float
class Worker:
"""执行具体任务的 Worker"""
def __init__(self, name: str, system_prompt: str,
tools: list[dict] | None = None):
self.name = name
self.system_prompt = system_prompt
self.tools = tools or []
def execute(self, task: Task) -> TaskResult:
"""执行一个具体任务"""
user_message = (
f"任务: {task.description}\n"
f"上下文: {json.dumps(task.context, ensure_ascii=False)}"
)
response = self._run_agent_loop(
system=self.system_prompt,
user_input=user_message,
tools=self.tools
)
return TaskResult(
task_id=task.id,
output=response.content,
findings=self._extract_findings(response),
confidence=0.85
)
def _run_agent_loop(self, system, user_input, tools):
# 标准的单 Agent while 循环（见 01-function-calling.md）
...
def _extract_findings(self, response) -> list[dict]:
# 从 Worker 输出中提取结构化发现
...
class Orchestrator:
"""中央编排器：分析任务 → 分配 Worker → 收集结果 → 决策"""
PLANNER_PROMPT = """你是一个任务编排器。根据用户的总体目标和已收集的信息，
制定下一步计划。输出格式:
{
"status": "in_progress" | "done",
"reasoning": "基于当前信息的分析...",
"next_tasks": [
{
"id": "task_3",
"description": "具体任务描述",
"worker_type": "code_analyzer",
"priority": "high" | "medium" | "low"
}
],
"summary": "当前进度的摘要"
}"""
def __init__(self, workers: dict[str, Worker],
max_iterations: int = 10):
self.workers = workers
self.max_iterations = max_iterations
self.collected_results: list[TaskResult] = []
self.task_history: list[dict] = []
def execute(self, user_goal: str) -> str:
"""编排整个任务执行"""
iteration = 0
current_context = {"goal": user_goal, "findings": []}
while iteration < self.max_iterations:
iteration += 1
print(f"\n[Orchestrator] === 迭代 {iteration} ===")
# Step 1: PLAN — 让 LLM 分析当前状态，制定计划
plan = self._plan(current_context)
if plan["status"] == "done":
print("[Orchestrator] 任务完成，生成最终报告")
break
# Step 2: DISPATCH — 分配任务给 Worker
tasks = [
Task(
id=t["id"],
description=t["description"],
worker_type=t["worker_type"],
context=current_context
)
for t in plan.get("next_tasks", [])
]
# Step 3: EXECUTE — Worker 执行
for task in tasks:
worker = self.workers.get(task.worker_type)
if not worker:
print(f" ⚠ 找不到 Worker: {task.worker_type}")
continue
result = worker.execute(task)
self.collected_results.append(result)
self.task_history.append({
"task": task,
"result": result.output
})
# Step 4: UPDATE — 更新上下文
current_context["findings"] = [
{"task_id": r.task_id, "output": r.output,
"findings": r.findings}
for r in self.collected_results
]
# 生成最终报告
return self._synthesize(user_goal)
def _plan(self, context: dict) -> dict:
"""调用 LLM 制定下一步计划"""
response = llm.chat(
messages=[
{"role": "system", "content": self.PLANNER_PROMPT},
{"role": "user",
"content": json.dumps(context, ensure_ascii=False)}
],
response_format={"type": "json_object"}
)
return json.loads(response.content)
def _synthesize(self, goal: str) -> str:
"""汇总所有 Worker 结果，生成最终报告"""
all_results = "\n\n".join(
f"## {r.task_id}\n{r.output}"
for r in self.collected_results
)
response = llm.chat(
messages=[
{"role": "system", "content":
"汇总以下分析结果，生成面向用户的最终报告"},
{"role": "user", "content":
f"目标: {goal}\n\n分析结果:\n{all_results}"}
]
)
return response.content
# --- 使用示例：代码仓库安全审查 ---
security_workers = {
"dependency_scanner": Worker(
name="依赖扫描器",
system_prompt="你是依赖安全专家。检查依赖库的已知漏洞...",
tools=[pip_audit_tool, npm_audit_tool]
),
"secret_detector": Worker(
name="密钥检测器",
system_prompt="你是密钥泄露检测专家。搜索硬编码的密钥...",
tools=[grep_tool, trufflehog_tool]
),
"code_analyzer": Worker(
name="代码分析器",
system_prompt="你是代码安全审查专家。审查认证、授权、输入校验...",
tools=[read_file_tool, grep_tool]
),
}
orchestrator = Orchestrator(
workers=security_workers,
max_iterations=8
)
report = orchestrator.execute("分析 /path/to/repo 的安全性")
print(report)4.5 Orchestrator-Workers 的设计陷阱
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator-Workers 设计 Checklist │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. Orchestrator 会"方向漂移" │
│ → 在 Planner Prompt 中明确任务边界和禁止方向 │
│ → 每次迭代都回顾原始目标 │
│ │
│ 2. Worker 可能返回不完整或不准确的结果 │
│ → 让 Worker 输出置信度 + 证据引用 │
│ → Orchestrator 交叉验证多个 Worker 的结果 │
│ │
│ 3. 死循环风险 │
│ → max_iterations 硬限制 │
│ → 检测重复任务（同一任务不要分配两次） │
│ → 如果连续 N 次迭代无新发现，强制终止 │
│ │
│ 4. 成本爆炸 │
│ → 每次迭代 = 1 次 Planner 调用 + N 次 Worker 调用 │
│ → 限制 Worker 的输出长度（不要让 Worker 写长篇报告） │
│ → 为 Orchestrator 和 Worker 使用不同级别的模型 │
│ │
│ 5. Worker 能力边界要清晰 │
│ → 每个 Worker 的 system prompt 明确"你能做什么" │
│ → 不匹配的任务 Orchestrator 不应该分配给该 Worker │
│ │
└─────────────────────────────────────────────────────────────┘第5部分：Evaluator-Optimizer —— 迭代精炼
5.1 什么是 Evaluator-Optimizer？
两个角色循环协作：Generator 生成内容，Evaluator 按明确标准打分并给出修改建议，Generator 根据反馈修改，直到 Evaluator 满意或达到最大迭代次数。
┌─────────────────────────────────────────────────────────────┐
│ Evaluator-Optimizer 数据流 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户: "写一篇产品发布公告" │
│ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Generator (生成器) │ │
│ │ "你是文案专家，撰写产品发布公告" │ │
│ │ │ │
│ │ 输出: 公告 v1 │ │
│ └───────────────────────┬───────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Evaluator (评估器) │ │
│ │ 评估标准: │ │
│ │ • 清晰度 (1-10): 7 │ │
│ │ • 说服力 (1-10): 6 │ │
│ │ • 完整性 (1-10): 8 │ │
│ │ • 品牌调性 (1-10): 5 ← 不达标! │ │
│ │ │ │
│ │ 总体: 未通过 (阈值 8.0, 当前 6.5) │ │
│ │ 反馈: "品牌调性不符，缺少产品价值观表述..." │ │
│ └───────────────────────┬───────────────────────────┘ │
│ │ │
│ │ 反馈 + 原文 → Generator │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Generator (生成器) │ │
│ │ "根据反馈修改: 加强品牌调性..." │ │
│ │ │ │
│ │ 输出: 公告 v2 │ │
│ └───────────────────────┬───────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Evaluator (评估器) │ │
│ │ 品牌调性: 5 → 8.5 ✅ │ │
│ │ 总体: 8.3 ≥ 8.0 → 通过! ✅ │ │
│ └───────────────────────┬───────────────────────────┘ │
│ │ │
│ ▼ │
│ 输出: 公告 v2 │
│ │
└─────────────────────────────────────────────────────────────┘关键特征：
| 特征 | 说明 |
|---|---|
| 角色关系 | Generator 和 Evaluator 是明确分离的两个 LLM 调用 |
| 评估标准 | 必须在 Evaluator 的 prompt 中明确定义 |
| 反馈形式 | 具体、可操作的修改建议（不是笼统的"写得不好"） |
| 迭代次数 | 有上限，通常 3-5 轮足够 |
| 适用场景 | 输出质量有明确评判标准 |
5.2 适用与不适用场景
┌─────────────────────────────────────────────────────────────┐
│ Evaluator-Optimizer 适用性判断 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ✅ 适用： │
│ • 输出质量有明确、可量化的评估标准 │
│ • "好"与"不好"的判断可以在 prompt 中清晰描述 │
│ • 质量比延迟重要（允许多轮迭代） │
│ • Generator 有能力根据反馈做针对性修改 │
│ • 需要保持一致的输出质量而不仅是"一次生成" │
│ │
│ ❌ 不适用： │
│ • 评估标准模糊、主观性强（"写得有趣"很难让 LLM 评估） │
│ • 对延迟敏感、需要实时响应 │
│ • Generator 能力不足以根据反馈改进（小模型不行） │
│ • 任务本身不需要多轮精炼（如简单问答） │
│ │
└─────────────────────────────────────────────────────────────┘5.3 代码骨架
python
from dataclasses import dataclass, field
@dataclass
class EvalResult:
"""评估器返回的结果"""
passed: bool
overall_score: float # 0.0 - 10.0
dimension_scores: dict # {"清晰度": 7, "说服力": 6, ...}
feedback: str # 具体的改进建议
strengths: list[str] # 做得好的地方（保留）
issues: list[str] # 需要改进的地方
class Generator:
"""生成器：负责产出内容"""
def __init__(self, system_prompt: str, model: str = "gpt-4o"):
self.system_prompt = system_prompt
self.model = model
def generate(self, task: str,
feedback: str | None = None,
previous_output: str | None = None) -> str:
"""生成内容（首次）或根据反馈修改"""
if feedback and previous_output:
user_message = (
f"原始任务: {task}\n\n"
f"你之前的输出:\n```\n{previous_output}\n```\n\n"
f"评估反馈:\n{feedback}\n\n"
f"请根据反馈修改，保留反馈中提到的优点，改进指出的问题。"
)
else:
user_message = task
response = llm.chat(
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_message}
],
model=self.model
)
return response.content
class Evaluator:
"""评估器：按标准打分并给出反馈"""
EVAL_PROMPT = """你是内容质量评估专家。按以下标准评估给定内容:
评估维度（每项 1-10 分）:
{criteria}
输出 JSON:
{{
"overall_score": 0.0,
"dimension_scores": {{"维度1": 7, ...}},
"passed": true/false (overall_score >= {threshold}),
"strengths": ["优点1", ...],
"issues": ["问题1", ...],
"feedback": "具体的、可操作的修改建议..."
}}
重要: feedback 必须具体，指明哪里需要改、怎么改。
不要只说"需要改进"，要说"第2段缺少数据支撑，建议加入..."。"""
def __init__(self, criteria: dict[str, str],
threshold: float = 8.0,
model: str = "gpt-4o-mini"): # 评估可用小模型
self.criteria = criteria
self.threshold = threshold
self.model = model
# 构建评估 prompt
criteria_text = "\n".join(
f"- {name}: {desc}"
for name, desc in criteria.items()
)
self.system_prompt = self.EVAL_PROMPT.format(
criteria=criteria_text,
threshold=threshold
)
def evaluate(self, content: str, task: str) -> EvalResult:
"""评估一段内容"""
response = llm.chat(
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user",
"content": f"任务: {task}\n\n待评估内容:\n{content}"}
],
model=self.model,
response_format={"type": "json_object"}
)
data = json.loads(response.content)
return EvalResult(
passed=data["passed"],
overall_score=data["overall_score"],
dimension_scores=data["dimension_scores"],
feedback=data["feedback"],
strengths=data["strengths"],
issues=data["issues"]
)
class EvaluatorOptimizerLoop:
"""Evaluator-Optimizer 主循环"""
def __init__(self, generator: Generator, evaluator: Evaluator,
max_iterations: int = 5):
self.generator = generator
self.evaluator = evaluator
self.max_iterations = max_iterations
def run(self, task: str) -> dict:
"""执行生成-评估-修改循环"""
history = []
current_output = None
feedback = None
for i in range(self.max_iterations):
print(f"\n[E-O] === 迭代 {i+1}/{self.max_iterations} ===")
# Step 1: Generate
print("[E-O] Generator 生成中...")
current_output = self.generator.generate(
task=task,
feedback=feedback,
previous_output=current_output
)
# Step 2: Evaluate
print("[E-O] Evaluator 评估中...")
eval_result = self.evaluator.evaluate(current_output, task)
history.append({
"iteration": i + 1,
"output": current_output,
"evaluation": eval_result
})
print(f"[E-O] 得分: {eval_result.overall_score:.2f} "
f"(阈值: {self.evaluator.threshold})")
# Step 3: Check
if eval_result.passed:
print("[E-O] ✅ 通过！")
break
feedback = eval_result.feedback
print(f"[E-O] ❌ 未通过，反馈: {eval_result.feedback[:100]}...")
return {
"final_output": current_output,
"iterations": len(history),
"final_score": history[-1]["evaluation"].overall_score,
"history": history
}
# --- 使用示例：产品发布公告撰写 ---
generator = Generator(
system_prompt="""你是资深产品营销文案。撰写产品发布公告时:
- 开门见山，第一段说清产品是什么
- 用数据说话，不是空洞形容词
- 保持品牌调性：专业但不冰冷，热情但不浮夸
- 包含：产品名、核心功能、上线时间、获取方式""",
model="gpt-4o"
)
evaluator = Evaluator(
criteria={
"清晰度": "读者能否在30秒内理解产品是什么",
"说服力": "是否有足够的理由让读者想尝试",
"完整性": "是否包含所有必要信息（名称、功能、时间、获取方式）",
"品牌调性": "语气是否符合'专业但不冰冷，热情但不浮夸'",
"简洁度": "是否没有冗余废话，每句话有信息量"
},
threshold=8.0
)
loop = EvaluatorOptimizerLoop(
generator=generator,
evaluator=evaluator,
max_iterations=4
)
result = loop.run("为我们的新产品 'DataViz Pro' 写一篇发布公告。"
"它是一个 AI 驱动的数据可视化工具，"
"支持自然语言生成图表，下周一上线。")5.4 有效的评估标准设计
┌─────────────────────────────────────────────────────────────┐
│ 评估标准设计 Checklist │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 标准要具体、可操作 │
│ ❌ "文章质量好" │
│ ✅ "逻辑连贯：每个论点有论据支撑，段落间有过渡" │
│ │
│ 2. 每个标准要有明确的"好"和"差"的定义 │
│ 例: "清晰度 10分 = 不看第二遍就能复述核心观点" │
│ "清晰度 2分 = 读三遍仍不确定主旨" │
│ │
│ 3. 阈值要合理 │
│ • 太低了失去精炼意义（6.0 阈值 = 几乎不迭代） │
│ • 太高了永远达不到（9.5 阈值 = 死循环） │
│ • 推荐 7.5-8.5 │
│ │
│ 4. Evaluator 的反馈必须包含"怎么做"而不只是"哪里不好" │
│ ❌ "第3段逻辑不清晰" │
│ ✅ "第3段从数据直接跳到结论，缺少推理过程。 │
│ 建议: 加一句解释'数据X说明Y，因此我们得出Z'" │
│ │
│ 5. Evaluator 可以用小模型 │
│ • 评估比生成便宜（gpt-4o-mini 足够） │
│ • Generator 用强模型（gpt-4o / claude-sonnet） │
│ │
└─────────────────────────────────────────────────────────────┘第6部分：模式对比与选型指南
6.1 五种模式一图对比
┌─────────────────────────────────────────────────────────────┐
│ 五种架构模式全景对比 │
├──────────┬──────────┬────────────┬──────────┬───────────────┤
│ 模式 │ 决策方式 │ LLM 调用次数│ 适用复杂度 │ 核心风险 │
├──────────┼──────────┼────────────┼──────────┼───────────────┤
│ Chaining │ 固定顺序 │ 等于步骤数 │ 低-中 │ 步骤失败传播 │
│ Routing │ 入口分类 │ 1+N │ 低-中 │ 分类错误 │
│ Parallel │ 编译期分解│ N+1 │ 中 │ 聚合质量 │
│ Orch-Work│ 运行时动态│ 多次(可变) │ 高 │ 方向漂移 │
│ Eval-Opt │ 反馈循环 │ 2×迭代次数 │ 中-高 │ 无法收敛 │
└──────────┴──────────┴────────────┴──────────┴───────────────┘6.2 选型决策树
┌─────────────────────────────────────────────────────────────┐
│ 架构模式选型决策树 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题: 任务可以分解为固定步骤吗? │
│ │ │
│ ├── YES → 步骤间有依赖吗? │
│ │ ├── YES → Prompt Chaining │
│ │ └── NO → Parallelization │
│ │ │
│ └── NO → 需要根据输入类型分流吗? │
│ ├── YES → Routing │
│ └── NO → 需要多轮迭代改进吗? │
│ ├── YES → Evaluator-Optimizer │
│ └── NO → 需要动态规划吗? │
│ ├── YES → Orchestrator │
│ └── NO → 单 Agent 够了 │
│ │
└─────────────────────────────────────────────────────────────┘6.3 复合模式：真实系统的常见形态
实际生产系统很少只用一种模式。下面是两种常见的复合模式。
复合模式 A：Routing + Chaining + Eval-Opt
┌─────────────────────────────────────────────────────────────┐
│ 复合模式 A: 智能客服系统 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ [Routing] 分类器 → refund / technical / complaint │
│ │ │
│ ├─ refund: │
│ │ [Chaining] 查订单 → 验政策 → 执行退款 → 通知 │
│ │ │
│ ├─ technical: │
│ │ [Eval-Opt] 诊断 → 生成方案 → 评估方案质量 → 修改 │
│ │ │
│ └─ complaint: │
│ [Chaining] 情绪安抚 → 问题记录 → 升级 → 回访承诺 │
│ │
└─────────────────────────────────────────────────────────────┘复合模式 B：Orchestrator-Workers + Parallelization
┌─────────────────────────────────────────────────────────────┐
│ 复合模式 B: 代码审查系统 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Orchestrator 制定审查计划 │
│ │ │
│ ├─ [Parallel] │
│ │ Worker A: 安全审查 ─┐ │
│ │ Worker B: 性能审查 ─┼─ 同时执行 │
│ │ Worker C: 风格审查 ─┘ │
│ │ │
│ ├─ Orchestrator 分析结果，决定深入方向 │
│ │ │
│ └─ [Chaining] │
│ 汇总 → 去重 → 排序(按严重性) → 生成修复建议 │
│ │
└─────────────────────────────────────────────────────────────┘6.4 架构反模式
┌─────────────────────────────────────────────────────────────┐
│ 常见的架构反模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 反模式1: 过早抽象 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题: 上来就建一个通用的 Orchestrator 框架 │ │
│ │ 后果: 框架比业务逻辑还复杂，调试困难 │ │
│ │ 纠正: 先用 Chaining 实现，真正需要动态调度再重构 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 反模式2: 一个 Agent 做所有事 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题: 一个巨型 system prompt 包含所有逻辑 │ │
│ │ 后果: 行为不可预测，prompt 越写越长，难以维护 │ │
│ │ 纠正: 按职责拆分为多个 Agent，每个只管一件事 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 反模式3: 为了并行而并行 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题: 本来一次 LLM 调用能搞定的，拆成 5 个并行调用 │ │
│ │ 后果: 成本 5x，延迟取决于最慢的那个，收益不大 │ │
│ │ 纠正: 只在子任务真正独立且并行有明显收益时才用 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 反模式4: Evaluator 和 Generator 用同一个 prompt │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题: Generator 自我评估 → 永远觉得自己的输出很好 │ │
│ │ 后果: 没有真正的改进循环 │ │
│ │ 纠正: Evaluator 必须是独立的 LLM 调用，有独立的 │ │
│ │ 评估标准 prompt │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘核心总结
总结1：五种模式的本质
┌─────────────────────────────────────────────────────────────┐
│ │
│ Prompt Chaining = 固定流水线，A → B → C │
│ Routing = 智能分流，分类器 → 专门 Agent │
│ Parallelization = 同时执行，无依赖子任务 → 汇总 │
│ Orchestrator-Workers = 动态调度，中央编排 → 灵活分配 │
│ Evaluator-Optimizer = 迭代精炼，生成 → 评估 → 修改 │
│ │
└─────────────────────────────────────────────────────────────┘总结2：选型原则
从最简单开始 → 不够用再加复杂度:
单次 LLM 调用够用 → 不要加 Agent
Prompt Chaining 够用 → 不要加 Orchestrator
固定并行够用 → 不要加动态调度
一次生成够用 → 不要加 Evaluator-Optimizer总结3：关键设计决策对照
| 决策 | 简单方案 | 复杂方案 |
|---|---|---|
| 任务分解 | 编译期写死步骤 | 运行时 LLM 动态规划 |
| 任务分配 | 固定顺序 | Orchestrator 按需分配 |
| 质量保证 | 人工检查 | Evaluator 自动评估迭代 |
| 并行 | 无 / asyncio.gather | Orchestrator 管理依赖 |
| 错误处理 | 步骤失败 → 重试该步 | Orchestrator 重新规划 |
章节测试
测试1：模式识别
以下场景最适合哪种架构模式？ "用户上传一张图片，系统需要：识别图片内容、检测是否有违规内容、如果有违规则模糊处理并记录日志。"
测试2：Prompt Chaining
Prompt Chaining 和 Orchestrator-Workers 最关键的区别是什么？
测试3：Routing 设计
一个 Routing 系统的分类器准确率只有 70%，你如何设计兜底策略？
测试4：并行化判断
以下哪些任务适合并行化？ A. 翻译一篇文档的 10 个段落 B. 先搜索资料，再根据搜索结果写报告 C. 同时检查代码的安全性、性能、可读性 D. 生成初稿 → 评估 → 修改 → 再评估
测试5：Evaluator-Optimizer
为什么 Evaluator 不能用和 Generator 相同的 system prompt？
测试6：Orchestrator 死循环
Orchestrator-Workers 模式中，有哪些机制可以防止无限循环？
测试7：复合模式设计
为一个"智能简历优化系统"设计架构。需求：用户上传简历，系统分析简历质量，识别薄弱点，逐项优化，最终输出多个版本供选择。
参考答案
测试1答案
答案：Prompt Chaining（固定顺序流水线）
- 步骤明确：识别 → 检测 → 处理 → 记录
- 步骤间有顺序依赖（检测必须在识别之后）
- 不需要动态决策（什么情况模糊处理是预先定义的）
测试2答案
答案：
- Prompt Chaining：步骤顺序和内容在编译期（写代码时）就确定了
- Orchestrator-Workers：步骤在运行时由 LLM 动态分析任务后决定
- Chaining 适合已知工作流，O-W 适合开放式任务
测试3答案
答案：
- 设置置信度阈值：< 0.7 走通用兜底 Agent
- 低置信度时追问用户确认："您是想咨询退款吗？"
- 高风险类别宁可漏判不可误判（如投诉升级宁可转人工）
- 持续收集分类错误的 case，定期更新分类器 prompt
测试4答案
答案：A 和 C 适合并行化
- A: 每个段落独立翻译，无依赖 → 适合
- B: 写报告依赖搜索结果 → 不适合（用 Chaining）
- C: 三个维度互相独立 → 适合
- D: 有明确的迭代循环 → 不适合（用 Eval-Opt）
测试5答案
答案：
- Generator 的 prompt 是"如何写出好内容"（创造性）
- Evaluator 的 prompt 是"如何判断内容好坏"（批判性）
- 如果同一个 prompt 既生成又评估，LLM 会倾向于自我肯定
- 分离后才能实现真正的"客观评估 + 针对性改进"
测试6答案
答案：
max_iterations硬限制（如最多 10 轮）- 检测重复任务：同一任务不分配两次
- 连续 N 轮无新发现 → 强制终止
- Planner Prompt 中明确终止条件："当所有必要信息已获取，status 返回 done"
- Token 预算：总 token 消耗超过阈值时终止
测试7答案
答案：建议使用 Routing + Eval-Opt + Chaining 复合模式
[Routing] 分类器分析简历，识别薄弱维度
│
├─ [Eval-Opt] 逐项优化（每个维度独立循环）
│ 经验部分: Generator 重写 → Evaluator 评估
│ 技能部分: Generator 重组 → Evaluator 评估
│ 格式部分: Generator 调整 → Evaluator 评估
│
└─ [Chaining]
优化后各部分 → 组装完整简历 → 生成多个版本
(激进版、保守版、创意版)相关笔记
- [[05-agent-workflow]] - Agent 工作流与 Tool Calling 循环
- [[06-multi-agent]] - 多 Agent 系统与协作模式
- [[00-agent-overview]] - Agent 整体学习路线
- [[02-framework-evolution]] - 框架如何封装这些模式
下一步学习
- [ ] 阅读 07 - RAG 高级技巧
- [ ] 尝试用 Prompt Chaining 实现一个简单的文档处理流水线
- [ ] 阅读 Anthropic "Building effective agents" 原文
学习状态：🟡 开始学习