第12章：Workflow 编排模式

很多开发者一接到"构建 AI Agent"的需求，第一反应是去找框架、拆分 Agent、设计协作模式。但大多数时候，这是过早的复杂化。

这一章要解决的核心问题是：什么时候一个 Agent 够用，什么时候需要多个 Agent 协作，以及当引入多 Agent 时，有哪些可以信赖的编排模式。后两节讲 Human-in-the-Loop 和 Agent-User Interaction——当 Agent 自主决策走到需要人类拍板的边界时，系统设计应该怎么处理，以及这种自主性应该怎样被用户感知和纠偏。

12.1 单 Agent 系统：大多数场景的正确起点

核心直觉

一个 Agent，配上一组工具和一份指令，循环执行直到完成——这在绝大多数生产场景下已经够用，不要上来就做复杂的多 Agent 系统。

什么是单 Agent 系统

单 Agent 系统的结构极其简单：一个 LLM 实例，持有一段 System Prompt 描述它的角色和约束，配上若干工具（查数据库、搜索网页、写文件……），然后循环执行感知-推理-行动直到任务完成或触发停止条件。

这就是 Agent 循环的完整结构。没有中央编排者，没有消息总线，没有角色分工。

为什么先推单 Agent

OpenAI 的《A Practical Guide to Building Agents》和 Anthropic 的《Building Effective Agents》在这一点上立场一致 [1][2]：

"Start with a single agent. Add complexity only when a simpler solution demonstrably falls short."

理由是务实的。多 Agent 系统引入了 Agent 间通信、共享状态同步、错误传播链、延迟叠加等新的故障模式。每一层复杂度都是新的调试负担。如果单 Agent 能完成任务，花时间优化它的工具设计和 Prompt，比急于引入协作要有价值得多。

单 Agent 系统的结构特征

一个生产级的单 Agent 循环大概长这样：

python

import anthropic
import json
from typing import Any

client = anthropic.Anthropic()

def run_agent(task: str, tools: list[dict]) -> str:
    messages = [{"role": "user", "content": task}]
    
    while True:
        response = client.messages.create(
            model="claude-opus-4-7",
            max_tokens=8096,
            system="你是一个任务执行助手，使用工具完成用户给定的任务。完成后总结结果。",
            tools=tools,
            messages=messages
        )
        
        # 任务完成，退出循环
        if response.stop_reason == "end_turn":
            return next(
                block.text for block in response.content
                if hasattr(block, "text")
            )
        
        # 收集所有工具调用并执行
        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            result = dispatch_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
        
        # 追加到消息历史，继续循环
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

这段代码没有任何框架依赖，能跑完 90% 的单 Agent 任务场景。

工程权衡

单 Agent 系统的天花板在于三个地方：

上下文窗口：长任务会把工具调用历史堆满，上下文压缩是必要的（第 6 章 6.7 节讨论过）
专注度问题：工具数量多了，或者任务领域跨度大，同一个 Agent 同时处理多件差异很大的事，质量会下降
并行能力：单 Agent 天然串行，跨领域的独立子任务无法并行

这三个限制，就是引入多 Agent 的合理理由。但注意：先遇到这些限制，再考虑多 Agent，不要提前优化。

12.2 何时需要 Multi-Agent：单 Agent 的能力天花板在哪里

核心直觉

引入多 Agent 的理由只有一种：有可验证的证据表明，单 Agent 在某个维度上已经到了天花板，而拆分能解决那个具体问题。

三个合理理由

理由一：上下文窗口管理 当任务涉及大量工具调用历史、长文档分析、或者需要同时维护多个独立工作流的中间状态时，单个 Agent 的上下文窗口会成为瓶颈。把不同的子任务交给不同 Agent，每个 Agent 维护独立的、小的上下文，是有效的缓解策略。

理由二：专业化分工 研究任务和代码生成任务需要不同的指令风格和工具集。一个"研究员"Agent 和一个"编码员"Agent，各自持有针对性的 Prompt 和工具，比一个通用 Agent 兼顾两种任务的质量更稳定。Anthropic 在内部的 Coding Agent 实现中观察到：专用 Agent 在特定领域工具调用的准确率显著高于通用 Agent [1]。

理由三：并行执行 有依赖关系的任务必须串行，但如果两个子任务彼此独立，把它们交给两个 Agent 并行执行，总延迟由最慢的那个决定，而不是两者之和。报告生成场景中，市场分析和财务分析可以并行；软件开发场景中，前端和后端的修改往往也是独立的。

什么不是引入多 Agent 的好理由

有几种错误动机很常见：

"感觉应该有个 Orchestrator"：如果没有遇到上面三个具体问题，加一个 Orchestrator 只会增加调试难度
"模仿真实团队结构"：技术架构不需要映射组织结构，PM/Architect/Developer 的角色分工映射到 Multi-Agent 并不总是有效
"用了框架所以用多 Agent"：框架（LangGraph、CrewAI）让多 Agent 变容易了，但不代表就应该用

一个实用的判断问题：如果我不能说清楚每个 Agent 分别在解决单 Agent 的什么具体限制，那就不需要多 Agent。

从单到多的迁移信号

观察到的问题	对应的多 Agent 解法
上下文窗口频繁爆满，压缩后质量下降	拆分成上下文独立的子 Agent
一个 Agent 处理多种领域任务质量不稳定	专用 Agent + 路由层
串行执行导致总延迟不可接受	并行子 Agent
某个工具集太大，Agent 选择困难	按功能分组给不同 Agent

12.3 编排模式分类

确定需要多 Agent 之后，下一个问题是选哪种编排结构。三种主流模式各自有明确的适用场景。

Manager 模式（中央编排）

一句话：有一个 Orchestrator Agent 做决策，把任务委派给 Worker Agent，最后汇聚结果。

在 OpenAI Agents SDK 里，Manager 模式对应"Agents as tools"——Manager Agent 通过 Agent.as_tool() 调用 Worker Agent，就像调用普通工具一样，但内部是另一个完整的 Agent 执行循环 [3]：

python

from agents import Agent

# 定义专用 Worker
research_agent = Agent(
    name="Researcher",
    instructions="用网络搜索工具查找指定主题的信息，返回有来源的摘要。",
    tools=[web_search_tool],
)

coding_agent = Agent(
    name="Coder",
    instructions="根据需求编写 Python 代码，附上测试用例。",
    tools=[code_execution_tool],
)

# Manager 把 Worker 当工具调用
manager = Agent(
    name="Manager",
    instructions="根据任务需要，调用研究员或编码员来完成工作，最后汇总结果。",
    tools=[
        research_agent.as_tool(
            tool_name="do_research",
            tool_description="用于搜索信息和研究特定主题"
        ),
        coding_agent.as_tool(
            tool_name="write_code",
            tool_description="用于编写和测试代码"
        ),
    ],
)

适合场景：任务需要汇聚多个专业领域的结果，且有一个自然的"综合"步骤。Manager 拥有完整的执行控制权和输出所有权。

注意：Manager 模式下，所有子任务的执行都是由 Manager 发起的同步调用，Orchestrator 必须等待所有 Worker 完成。如果子任务可以并行，需要额外设计并发逻辑。

去中心化模式（Handoff）

一句话：没有中央控制者，当前 Agent 判断到该把控制权移交给另一个 Agent 时，直接 Handoff 过去，被交接的 Agent 接管后续对话。

在 OpenAI Agents SDK 里，Handoff 是一等公民概念：

python

from agents import Agent, handoff

support_agent = Agent(
    name="Support",
    instructions="处理一般客服问题，涉及退款时转交给退款专员。",
)

refund_agent = Agent(
    name="Refund Specialist",
    instructions="处理退款、换货等售后问题。需要核查订单信息后再确认。",
    tools=[check_order_tool, process_refund_tool],
)

# Triage Agent 带有 handoff 能力
triage_agent = Agent(
    name="Triage",
    instructions="识别用户意图，路由到合适的专员。",
    handoffs=[
        handoff(
            support_agent,
            tool_description_override="用于一般问题咨询",
        ),
        handoff(
            refund_agent,
            tool_description_override="用于退款和售后请求",
        ),
    ],
)

Handoff 和 Manager 模式的根本区别在于谁拥有最终回复权：

Manager 模式：Manager 始终持有最终回复权，Worker 只是内部工具
Handoff 模式：控制权转移后，新的 Agent 直接面向用户，原来的 Agent 退出

适合场景：分阶段的任务流，每个阶段的目标明确、边界清晰，或者对话型场景中需要不同专员分别处理不同问题。

常见误区：不要把 Handoff 用在需要合并多个结果的场景——那是 Manager 模式的工作。

流水线模式（Pipeline）

一句话：前一个 Agent 的输出直接成为下一个 Agent 的输入，顺序传递，每个 Agent 只做一件事。

流水线模式本质上是第 3 章里 Prompt Chaining 的 Multi-Agent 扩展版本——区别在于 Prompt Chaining 里每一步是单次 LLM 调用，而流水线里每一步是一个完整的 Agent（有自己的工具调用循环）。

python

import asyncio

async def run_pipeline(raw_input: str) -> str:
    # Stage 1: 提取结构化数据
    extracted = await extract_agent.run(raw_input)
    
    # Stage 2: 深度分析
    analysis = await analysis_agent.run(extracted.output)
    
    # Stage 3: 生成报告
    report = await report_agent.run(analysis.output)
    
    # Stage 4: 质量检查（不通过则返回错误信号）
    result = await qa_agent.run(report.output)
    if not result.passed:
        raise ValueError(f"Quality check failed: {result.reason}")
    
    return result.output

适合场景：任务有明确的阶段性，每个阶段的输入输出格式清晰，且前后阶段有强依赖关系。内容生产（大纲→扩写→润色→审核）、数据处理（抽取→清洗→分析→报告）都适合这个模式。

与 Handoff 的区别：流水线通常由外部代码编排，知道完整的流程；Handoff 是 Agent 自己决定移交。流水线更可控，Handoff 更灵活。

模式选型速查

模式	控制权	适合场景	核心代价
Manager	中央集权	需要汇聚多专域结果	Orchestrator 复杂度高，需要并发设计
Handoff	分布式移交	对话型、阶段性强的任务	中途错误时追溯困难
Pipeline	外部代码编排	阶段分明、格式明确的批处理	灵活性低，步骤写死

实际系统里，这三种模式经常混用：一个 Manager 把任务分给多个流水线，流水线内部的某个阶段用 Handoff 处理分类路由。

12.4 Human-in-the-Loop 设计

核心直觉

Agent 自主执行很好，但有些操作不应该自主执行——送错邮件、错删数据、错误转账，代价超出了自动化带来的收益。HITL 就是在 Agent 的执行路径上设置需要人类确认的关卡。

为什么 HITL 是系统设计问题，不是临时补丁

很多团队是在出了问题之后才加 HITL，把它当成"救火"手段。这个思路是反的。

正确的问题是：在系统设计阶段，把所有会影响不可逆状态的操作列出来，决定每一种操作的自主权级别。什么可以自动执行，什么需要通知，什么需要审批——这是架构决策，不是出了问题再补的防火墙。

OpenAI 在《Safety in building agents》中明确指出 [4]：

"Keep tool approvals on. When using MCP tools, always enable tool approvals so end users can review and confirm every operation, including reads and writes."

注意这句话的适用范围：它针对的是 MCP tools 这类连接外部系统、权限边界更复杂的场景，而不是所有工具的一般规则。对 MCP、生产数据连接器、高权限外部系统，保守做法是把 approvals 默认打开；对普通本地工具或低风险函数工具，仍然应该按下面的分级自主权来决定是否自动执行。

分级自主权（Calibrated Autonomy）

三个风险等级，对应三种响应：

R1（低风险）→ 自动执行
  - 读取操作（查数据库、搜索）
  - 生成草稿（报告、代码、邮件）
  - 无副作用的分析任务

R2（中风险）→ 执行后通知
  - 发送通知类消息
  - 写入非关键数据
  - 调用第三方只读 API

R3（高风险）→ 执行前审批
  - 不可逆操作（删除、发送正式邮件）
  - 涉及金钱（转账、退款、订单变更）
  - 影响大量用户的批量操作
  - 访问敏感数据

这个分级不是固定的——同样是"发邮件"，给用户发通知邮件可能是 R2，给监管机构发正式文件就是 R3。分级应该根据业务场景和不可逆程度来定。

审批检查点的放置策略

关键设计点：审批不是"Agent 停下来等"的阻塞模型，而是可序列化的状态暂停。执行状态被持久化到数据库，Agent 进程可以终止，审批可以在数小时或数天后进行，然后用持久化的状态恢复执行。

OpenAI Agents SDK 的 HITL 实现直接支持这种模式 [5]：

python

from agents import Agent, Runner, RunState, function_tool

@function_tool(needs_approval=True)  # 声明此工具需要审批
async def cancel_order(order_id: int) -> str:
    # 实际的取消逻辑
    return f"订单 {order_id} 已取消"

agent = Agent(
    name="Order Agent",
    instructions="处理订单相关请求。取消订单前必须获得用户确认。",
    tools=[cancel_order],
)

async def main():
    result = await Runner.run(agent, "帮我取消订单 12345")
    
    # 检查是否有等待审批的操作
    while result.interruptions:
        for interruption in result.interruptions:
            print(f"Agent 请求执行: {interruption.name}")
            print(f"参数: {interruption.arguments}")
            
            # 获取人类决策（实际场景中来自 UI、消息系统等）
            approved = await get_human_approval(interruption)
            
            # 更新状态并恢复
            state = result.to_state()
            if approved:
                state.approve(interruption)
            else:
                state.reject(interruption, rejection_message="用户拒绝了此操作")
        
        # 从审批后的状态继续执行
        result = await Runner.run(agent, state)
    
    print(result.final_output)

needs_approval=True 可以是静态声明（所有调用都需要审批），也可以是动态函数（根据参数决定是否需要审批）——比如"只有退款金额超过 1000 元才需要审批"：

python

async def requires_approval(ctx, params, call_id) -> bool:
    return params.get("amount", 0) > 1000

@function_tool(needs_approval=requires_approval)
async def process_refund(order_id: int, amount: float) -> str:
    ...

审批流的超时和降级

HITL 系统有一个常被忽略的问题：如果审批请求无人响应怎么办？

实践中推荐两种策略：

超时自动拒绝：设定最大等待时间（比如 24 小时），超时则视为拒绝，Agent 收到拒绝信号后选择安全的降级路径（跳过该操作，通知用户）
超时升级：先通知直接负责人，超时后升级通知其上级，再超时记录到审计日志并中止任务

两种策略根据业务风险选择——金融相关的审批建议用超时拒绝，内容发布等的审批可以用超时升级。

常见误区

误区一：HITL 只在"最终步骤"放。很多实现只在输出结果时加一个确认。但危险操作通常发生在中间步骤——Agent 在执行过程中的某次工具调用可能已经造成了损害。应该在工具层面而非输出层面设置审批。

误区二：审批阻塞主线程。正确的实现是状态持久化 + 异步恢复，不是让 Agent 进程挂起等待。这对长时运行任务尤其重要。

误区三：所有操作都需要审批。HITL 过度会让 Agent 失去自主性的价值，变成"每次操作都要点确认"的手动流程。审批点应该精确到真正的高风险操作。

12.5 Agent-User Interaction / UX 设计

核心直觉

用户对 Agent 的信任，不是来自它“看起来很聪明”，而是来自它在长任务里持续反馈、在高风险处停下来、出错时给人接管和纠偏的机会。

长任务里的信任建立

很多 Agent 任务不是 3 秒出结果，而是持续几十秒到几分钟。这个时候最糟糕的体验不是慢，而是用户不知道它在做什么、做到哪一步、出了问题没有。

所以长任务至少要暴露三类信息：

当前阶段：例如“正在搜索资料”“正在生成初稿”“正在等待审批”
最近动作：最近一次调用了什么工具，得到了什么关键结果
下一步意图：接下来准备做什么，是否存在风险操作

如果用户只能看到一个旋转中的 loading，系统再强也很难建立信任。

可暂停、可取消、可接管

Agent 不是传统的同步函数调用。它可能卡在外部 API、等待网页加载、等待审批，或者在错误路径上越走越远。UI 必须给用户三个控制选项：

暂停（Pause）：临时冻结执行，保留上下文和当前状态
取消（Cancel）：终止当前任务，避免继续消耗成本或造成副作用
接管（Take over）：用户直接修改中间结果、工具参数或执行路径

这三个按钮不是“锦上添花”，而是高自主系统的基本安全阀。

Agent 出错时怎么呈现

Agent 出错时，最差的 UX 是把内部异常原样暴露给用户：ToolTimeoutError、ValueError、HTTP 500、堆栈追踪。用户需要的不是这些实现细节，而是可操作的下一步。

一个好的错误呈现应该回答三个问题：

哪里失败了：例如“网页连续 3 次加载超时”
影响是什么：例如“订单尚未提交”
你现在可以做什么：例如“重试”“跳过这一步”“切换为人工处理”

工程日志应该完整保留，面向用户的界面则要把错误翻译成决策信息。

用户纠偏机制

用户很少一开始就把需求说得完全正确。好的 Agent 系统要允许低成本纠偏，而不是要求用户“重新来过”。

三种最常见的纠偏入口：

自然语言修正：例如“不要最便宜的，要评价最高的”
显式参数调整：例如直接修改预算、日期、收件人、搜索范围
重新规划确认：当修改会影响后续路径时，先展示新计划再执行

如果系统只能接受“重跑整个任务”，用户很快就会放弃让 Agent 做长流程工作。

不同自主等级，对应不同 UI

自主权不是一个开关，而是一条光谱。不同等级的 Agent，应该对应不同的界面设计：

自主等级	合适的 UI 形态	用户预期
建议型	给建议、不给执行按钮默认确认	我来做决定
草稿型	先出草稿，等待编辑/确认	它先帮我起稿
半自动型	默认执行低风险步骤，高风险前暂停	它能推进，但关键处要问我
高自主型	后台持续执行，只在异常/高风险时唤醒	它大部分时间自己完成

一个常见错误，是用同一种 UI 同时承载这四种模式。结果不是太吵，就是太危险。

一个实用原则

当你不确定某个 Agent 该怎么呈现时，问自己一句话：如果它现在做错了，用户能不能在 10 秒内理解发生了什么，并知道下一步怎么纠正？

如果答案是否定的，那不是模型能力问题，而是 UX 设计还没完成。

面试高频题

Q1：什么场景下单 Agent 已经够用？什么信号告诉你需要引入 Multi-Agent？

参考答案框架：

单 Agent 够用的场景：任务可以在一次连续的感知-推理-行动循环中完成，工具集不超过 10 个，任务执行时间在几分钟以内，上下文窗口能容纳所有必要信息。

需要 Multi-Agent 的信号（三个判断维度）：

上下文信号：任务执行中频繁需要压缩上下文，且压缩后质量明显下降
质量信号：同一个 Agent 处理不同领域任务时，工具选择准确率出现明显差异
性能信号：有多个独立子任务，串行执行的总延迟超过业务可接受上限

加分点：能举出判断依据——比如"我在监控里发现 Agent 在某类任务的工具选择 error rate 比其他任务高 2 倍，这是专业化拆分的信号"，而不是"感觉应该用 Multi-Agent"。

Q2：Manager 模式、Handoff 模式、流水线模式各自的适用场景和失败模式是什么？

参考答案框架：

Manager 模式适合需要汇聚多专域结果的任务；失败模式是 Orchestrator 上下文过载（它需要理解所有 Worker 的结果）、以及当子任务有依赖关系时并行调度变得复杂。

Handoff 模式适合对话型场景和阶段分明的任务；失败模式是中途发生 Handoff 后追溯失败原因困难，以及 Handoff 判断本身的准确率问题（分错了就全链路出错）。

流水线模式适合批处理型任务；失败模式是灵活性低，遇到异常情况（某一步输出不符合下一步预期格式）整条流水线需要重设计。

Q3：设计 Human-in-the-Loop 系统时，如何处理审批请求长时间未响应的情况？

参考答案框架：

核心是将"审批"从同步阻塞改为异步状态持久化。Agent 执行到需要审批的操作时，序列化当前状态到持久存储（数据库/队列），发出审批通知，然后终止当前执行进程。

超时处理两种策略：超时自动拒绝（金融、数据删除等高风险场景）或超时升级（内容发布、通知类场景）。

加分点：提到"审批决策本身也应该记录到审计日志，包括审批人、时间、决策理由"——这是合规要求，也是事后追溯的依据。

参考资料

[1] Building effective agents - Anthropic (2024.12)
https://www.anthropic.com/engineering/building-effective-agents

[2] A Practical Guide to Building Agents - OpenAI (2025.04)
https://cdn.openai.com/business-guides-and-resources/a-practical-guide-to-building-agents.pdf

[3] Agent orchestration - OpenAI Agents SDK
https://openai.github.io/openai-agents-python/multi_agent/

[4] Safety in building agents - OpenAI
https://developers.openai.com/api/docs/guides/agent-builder-safety

[5] Human-in-the-loop - OpenAI Agents SDK
https://openai.github.io/openai-agents-python/human_in_the_loop/

第12章：Workflow 编排模式 ​

12.1 单 Agent 系统：大多数场景的正确起点 ​

核心直觉 ​

什么是单 Agent 系统 ​

为什么先推单 Agent ​

单 Agent 系统的结构特征 ​

工程权衡 ​

12.2 何时需要 Multi-Agent：单 Agent 的能力天花板在哪里 ​

核心直觉 ​

三个合理理由 ​

什么不是引入多 Agent 的好理由 ​

从单到多的迁移信号 ​

12.3 编排模式分类 ​

Manager 模式（中央编排） ​

去中心化模式（Handoff） ​

流水线模式（Pipeline） ​

模式选型速查 ​

12.4 Human-in-the-Loop 设计 ​

核心直觉 ​

为什么 HITL 是系统设计问题，不是临时补丁 ​

分级自主权（Calibrated Autonomy） ​

审批检查点的放置策略 ​

审批流的超时和降级 ​

常见误区 ​

12.5 Agent-User Interaction / UX 设计 ​

核心直觉 ​

长任务里的信任建立 ​

可暂停、可取消、可接管 ​

Agent 出错时怎么呈现 ​

用户纠偏机制 ​

不同自主等级，对应不同 UI ​

一个实用原则 ​

面试高频题 ​

参考资料 ​

第12章：Workflow 编排模式

12.1 单 Agent 系统：大多数场景的正确起点

核心直觉

什么是单 Agent 系统

为什么先推单 Agent

单 Agent 系统的结构特征

工程权衡

12.2 何时需要 Multi-Agent：单 Agent 的能力天花板在哪里

核心直觉

三个合理理由

什么不是引入多 Agent 的好理由

从单到多的迁移信号

12.3 编排模式分类

Manager 模式（中央编排）

去中心化模式（Handoff）

流水线模式（Pipeline）

模式选型速查

12.4 Human-in-the-Loop 设计

核心直觉

为什么 HITL 是系统设计问题，不是临时补丁

分级自主权（Calibrated Autonomy）

审批检查点的放置策略

审批流的超时和降级

常见误区

12.5 Agent-User Interaction / UX 设计

核心直觉

长任务里的信任建立

可暂停、可取消、可接管

Agent 出错时怎么呈现

用户纠偏机制

不同自主等级，对应不同 UI

一个实用原则

面试高频题

参考资料