多智能体编排层：让AI流程可控可恢复的工程实践

Paperclip 是给"无人公司"准备的编排层，这点我得先说明白。过去两年我做过好多个跑在模型 API 上的自动化流程，最后都死在同一件事上：不是模型不聪明，而是流程没有纪律。你可以让一个智能体自己查资料、写代码、发邮件，但这只是"单兵作战"。一旦把它放进一个真正要它自己跑起来、有上下游交接、有异常要处理、有人要兜底的业务系统里，你需要的是一层比模型本身更靠得住的骨架。Paperclip 就是冲着这个去的。这套设计不依赖某个厂商的 agent 框架，核心是一套任务拆解、上下文传递、状态恢复和人工接管的通用机制。如果你正在搭多智能体系统，或者想把某条业务流水线变成"AI 员工闭环"，这篇文章里的架构拆分、最小代码骨架和踩坑记录应该能帮你省下一轮弯路。

1. 为什么"无人公司"真正的卡点不是模型，是编排

1.1 单体 Agent 扛不住一个完整业务流程

很多人第一次做多智能体项目时，最容易犯的错是：把一整条业务链路写进一个大 prompt，寄希望于底层的长上下文模型能自己把每个环节执行完。

我实测过的结果是——任务链一旦超过四五个环节，单体智能体的整体成功率会断崖式下降。原因很朴素：大语言模型本质上没有"状态保持"和"过程控制"的能力，它擅长的是在某个具体节点上给出一个高质量输出。你让它一口气完成"接收客户订单、查库存、计算优惠、生成采购单、通知仓库、更新财务记录"，越到后面，前面环节的细节越容易被遗忘，甚至会在第五步把第三步已经算好的金额重新算一遍还算错了。这就像让一个新员工独自负责整个公司所有部门的事，他迟早会乱。

更好理解的方式是把它当作"分部门协作"。每个智能体只负责一段明确定义的工作，做完之后把结果交给下一个环节。这样每个 Agent 的上下文窗口不必装下整条流水线，只需要理解自己这一段的输入和预期输出。于是核心问题就从"怎么把大模型训得更聪明"变成了"怎么把多个 Agent 的协作流程组织起来"——这就是编排层存在的根本理由。

1.2 编排层到底编排的是什么

编排层不关心单个 Agent 内部用什么模型、怎么写 prompt，它关心的是五个要素：

• 任务（Task）：流程要被执行的最小单元，必须有明确的输入、输出和验收条件；
• 工具（Tool）：Agent 执行任务时的外部动作通道，比如查数据库、调 API；
• 上下文（Context）：在 Agent 之间传递的信息，但不是把全量聊天记录扔给下一个，而是只传"完成任务所需的最小集合"；
• 状态（State）：整个流程跑到了哪一步，是否成功、需要重试还是人工介入；
• 人（Human）：异常处理时的兜底入口，无人公司不是真的无人，而是把人的角色从"执行者"变成"监督者和仲裁者"。

Paperclip 把五个要素做成了一套可插拔的运行时。它不是 SaaS 产品，也不是某个云平台上的托管服务，它更像一个开源脚手架，你按自己的业务规则把节点接上去。

所以后面所有内容，我会围绕这套运行时讲：它的任务模型怎么切、状态机怎么转、失败怎么兜底、人和系统之间怎么交接。拿你自己的场景换掉示例中的业务内容，这套骨架依然成立。

2. Paperclip 的项目定位与它盯上的三个硬骨头

2.1 任务拆解：从"目标"到"可执行任务片"

Paperclip 里最核心的抽象叫 TaskSlice，一个"任务片"。它不是 URL 层面的一个大流程，而是把业务目标拆成一个个可以被 Agent 独立处理、独立验收的切片。

为什么要强调"独立验收"？因为在多智能体协作里，最怕上游 Agent 给下游丢来一个"大概完成了"的结果。TaskSlice 强制要求每个任务片声明三样东西：

• 输入协议（InputSchema）：进来的数据必须长什么样；
• 输出协议（OutputSchema）：完成之后必须返回什么结构；
• 幂等键（IdempotencyKey）：同一个任务片如果被执行两次，必须能被识别并且只生效一次。

拆分的策略上，我建议遵循"最小可验证单元"原则。举个例子：你希望 Agent 自动处理退款工单，不要拆成一个"处理退款"大任务片，而要拆成"读取工单信息""判断是否满足退款条件""计算退款金额""执行退款操作""生成回执通知"五个小切片。每个切片单独跑的正确率都能做到 95%，五个串起来理论上仍然有 0.95 的 5 次方约等于 77%。但如果每个切片内部又叠加了多步复杂操作，正确率没这么高，累计之后就会跌到不能看。

拆完之后，最好维护一张任务流转表，把每个 TaskSlice 的前置任务、依赖工具、超时时间写清楚。我见过不少项目跳过这一步直接上手写代码，等到联调的时候才发现上下游字段都对不上，改起来极其痛苦。编排层的价值正是在这种"流程建模"阶段释放出来的——它逼你把工作流说清楚，而不是让模型临场发挥。

2.2 上下文传递：给每个 Agent 一张"干净的桌子"

多 Agent 协作容易犯的第二个毛病，是上下文贪多。很多人觉得：既然上下文越完整 Agent 决策越准确，那么把所有历史记录都拼给每一个 Agent 不就好了？真实运行后发现，上下文越长，Agent 注意力越容易跑偏。它可能在第三页对话里读到一句无关紧要的用户备注，然后莫名其妙改变了自己的操作。

Paperclip 对上下文的处理方式是建立一个 ContextBus（上下文总线）。它不是把所有消息广播给所有人，而是按 TaskSlice 的输入协议做投影解析，只把当前任务片需要的字段从全局上下文中抽出来，再组装成一段结构化描述传给 Agent。每个 Agent 看到的必须是"一张干净的桌子"，上面只放它完成当前任务需要的东西，不需要的统统不放。

交接的时候还要注意"摘要-明细分离"。全局事件流里保留明细数据，传给下游 Agent 的可以是上游 Agent 生成的结构化摘要。比如查库存的 Agent 返回了 20 条库存批次明细，传到下一个计算可用量的 Agent 时，可以先把它聚合成一个 JSON：可用总量、锁定总量、最早到期批次。这样可以保持后续 Agent 的上下文极短，也能防止明细中的脏数据干扰判断。

2.3 失败恢复：让出错不再意味着重头开始

没有编排层的 Agent 流程，最常见的失败处理方式是"重跑一遍"，最多加个指数退避。但真实业务里重跑整个流程代价很高：可能你已经给客户发了确认邮件，结果因为最后一步更新数据库失败，整单状态就乱了。

Paperclip 把每个 TaskSlice 都放进一个有限状态机里。一次任务的生命周期包括：pending、running、failed、retrying、waiting_human、succeeded、compensated。关键是每个失败都要被分类：是可重试的瞬时错误，还是不可重试的逻辑错误，还是需要人工判断的边界情况。这三类走的根本不是同一条恢复路径。

可重试的瞬时错误，比如第三方 API 超时，可以自动重试并退避。不可重试的逻辑错误，比如 Agent 输出格式不对导致解析失败，应回到"重新生成输出"环节，而不是盲目重跑整个人工流程。边界情况，如退款金额和订单金额对不上，必须立刻标记 waiting_human，转交人工处理，而不是让 Agent 自行补偿。这套分类机制决定了编排层是"有脑子的调度器"而不是"盲目的重试循环器"。

3. 编排层的最小可运行骨架：一个可移植的参考实现

3.1 技术选型与模块划分

Paperclip 的参考实现我建议直接用 Python 写，原因很直接：Agent 生态和数据处理相关的库大多在 Python 这边，后续接 LangChain、LlamaIndex 或者裸 OpenAI SDK 都方便。环境上用到的最少组合是四件套：FastAPI 做 HTTP 入口、PostgreSQL 存任务状态、Redis 做上下文总线的临时消息通道、一个普通的进程池做 worker 执行。

为什么需要 PostgreSQL 而不只是 Redis？因为任务状态是业务真相（source of truth），任何时刻系统崩溃了，重启后必须能从数据库里恢复所有任务的当前状态；Redis 只适合做瞬时消息队列和分布式锁，数据随时可以重建。这层区分在选型时就要想清楚。

模块上分成五个部分：scheduler（调度器）、worker_pool（执行池）、context_bus（上下文总线）、state_store（状态存储）、human_api（人工介入接口）。它们之间没有强耦合，消息通信用 Redis 队列就可以。

3.2 核心数据结构与调度循环

直接看代码比说概念快。下面是一个简化的调度器骨架，核心逻辑就是从任务队列里取出待执行任务，检查前置条件，然后丢给 worker 去跑。

# task_slice.py from pydantic import BaseModel from typing import Optional, Any from enum import Enum class TaskState(str, Enum): PENDING = "pending" RUNNING = "running" FAILED = "failed" RETRYING = "retrying" WAITING_HUMAN = "waiting_human" SUCCEEDED = "succeeded" COMPENSATED = "compensated" class TaskSlice(BaseModel): task_id: str slice_type: str # 例如 refund_calculate input_data: dict[str, Any] # 当前切片需要的输入 output_data: Optional[dict] # 成功后写回的输出 idempotency_key: str # 幂等键 depends_on: list[str] = [] # 依赖的上游 task_id state: TaskState = TaskState.PENDING retry_count: int = 0 max_retry: int = 3

调度器主要做一件事：轮询待执行任务，检查依赖是否已完成，再判断是否超过最大重试，然后投递到 Redis 队列。

# scheduler.py import redis import json from sqlmodel import Session, select def dispatch_due_tasks(session: Session, redis_client: redis.Redis): # 只取出 pending 状态且所有依赖已成功执行的任务 tasks = session.exec(select(TaskSlice).where(TaskSlice.state == TaskState.PENDING)).all() for t in tasks: deps = [session.get(TaskSlice, d) for d in t.depends_on] if any(d.state != TaskState.SUCCEEDED for d in deps): continue if t.retry_count >= t.max_retry: t.state = TaskState.WAITING_HUMAN session.add(t) continue t.state = TaskState.RUNNING session.add(t) # 把任务投入 worker 队列 redis_client.lpush("paperclip:jobs", json.dumps(t.task_id)) session.commit()

Worker 的执行逻辑比较机械：从队列里取任务，组装上下文，调 Agent，校验输出结构，根据结果更新状态。

# worker.py def run(task_id: str, ctx_bus: ContextBus, agent_runner, state_store): task = state_store.fetch(task_id) # 从上下文总线里只取本任务需要的字段 context = ctx_bus.project(task.slice_type, task.input_data) try: raw_output = agent_runner.run(slice_type=task.slice_type, context=context) validated_output = OutputValidator.validate(task.slice_type, raw_output) # 写回下游要用的输出结果 state_store.mark_succeeded(task_id, validated_output) ctx_bus.publish(task.slice_type, validated_output) except OutputSchemaError as e: state_store.mark_failed(task_id, error=str(e), retryable=False) except TimeoutError as e: state_store.mark_failed(task_id, error=str(e), retryable=True)

注意，worker 本身不感知整个业务，不需要知道"退款计算"还是"工单分类"背后发生了什么。它只做三件事：取上下文、跑 Agent、校验输出。真正干活的 AgentRunner 你可以自己随意实现，可以用 OpenAI 的函数调用，也可以用本地模型，甚至某些环节直接写死规则返回，编排层一概不管。这个解耦非常关键，它保证你在换模型或者调整 prompt 的时候，不用碰调度逻辑。

3.3 状态机转移表：一张表写清所有恢复策略

很多编排项目做到后面失控，就是因为失败处理逻辑散落在各处 if-else 里。Paperclip 把所有转移路径明确定义成一张表，新增一个状态流转之前必须先在表里找到对应位置：

有这张表之后，新增业务流的开发成本会显著下降。团队只需要约定好每个 TaskSlice 属于哪种失败类型，剩下就交给编排层统一处置。

4. 实战验证：一个无人客服订单流程的压测记录

4.1 场景设定与任务流转

为了验证这套东西不是玩具，我搭了一条无人客服订单处理链路，业务规则是：客户提交退款申请，系统自动判断资格、计算金额、执行退款、发送通知。没有人工参与，除非出现异常。

任务流转拆成五个 TaskSlice：parse_ticket（解析工单）、check_eligibility（检查退款条件）、calculate_amount（计算退款金额）、execute_refund（执行退款）、notify_user（通知用户）。每个 Agent 使用同一个基础模型，不单独微调，只通过不同的 system prompt 限定角色和输出格式。

真正跑起来之前，我先在单 Agent 模式下做了一次对照组：同一个流程塞进一个大 prompt 让一个 Agent 从头跑到尾。对照组跑 50 单，成功完成全部五个环节的只有 34 单，成功率 68%，失败大多发生在执行退款后忘了通知用户，或者金额计算和检查资格的阶段字段对不上。这个数据和我之前的预估基本一致：环节越多，单体模式越不稳定。

4.2 编排模式下的数据表现

切到 Paperclip 编排模式后，同样 50 单压测，结果如下：

成功率提升的背后，主要得益于三个改进。第一个改进是上下文隔离：每个 Agent 只看到自己那个环节的数据，parse_ticket 传出去的是结构化工单 JSON，金额计算环节的 Agent 不再被原始工单文本里的冗余表达干扰。第二个改进是输出校验：每个 TaskSlice 都有输出协议，模型输出的金额如果是个带单位的长句而不是数字，会触发 schema 校验失败，任务不会流到下游去污染下一步。第三个改进是失败隔离：某一步出错只需要重跑那一步，不需要把前面所有环节重新来一遍。

耗时变短也符合预期。单 Agent 直跑时，模型经常在中间环节反复自言自语，把已经完成过的步骤又检查一遍；编排模式下每个 Agent 的任务窗口很短，模型不需要在长上下文里"回想"，所以生成速度更快。

需要提醒的是，这组数据是在测试场景下得出的，生产环境如果涉及第三方支付接口，成功率大概率会被外部系统延迟拉低，但编排层带来的相对增益依然显著。

5. 把 Paperclip 跑起来之后，我踩过的五个坑

5.1 Agent 会死循环，编排层必须有自己的熔断器

一个非常容易忽略的问题：Agent 拿到工具调用权限后，可能会在一个失败动作上反复折腾。比如它调用库存查询接口失败，就自己换个参数再调一次，然后再换一次，根本停不下来。如果编排层只在任务级别设置一个很大的超时时间，整个 worker 就被这个 Agent 占死了。

Paperclip 的做法是在工具调用层和任务层都设熔断。单个工具调用超过 20 秒就强制中断，单个任务的重试超过 3 次就转人工；另外还要统计 Agent 的连续空转次数，如果连续三次输出都没有推进任务（比如都是"让我再想想"或者重复调用同一个工具），立刻标记 failed 并转到 waiting_human。这个"空转检测"能救回很多看起来在干活、实际在打转的情况。

5.2 大模型输出的 JSON 不可信，schema 校验不是可选配置

刚开始我也偷懒，让 Agent 在 prompt 里输出 JSON 字符串，直接用 json.loads 去解析。压测一开始就出问题：模型偶尔会在 JSON 后面追加一句"如果您需要进一步计算，请告诉我"，导致解析失败。还有一些更隐蔽的，比如金额字段用字符串返回："金额是 23.45 元"，而不是一个数字 23.45。

之后我把输出校验模块升级成严格模式：定义好每个字段的类型、校验规则、是否允许空值、枚举可选项；模型输出后先进一个 Pydantic 模型做解析，解析失败就通知 Agent 重新格式化，最多重试两次。这样做的代价是每轮多花几百毫秒，但换来的是下游环节的数据可信度。尤其像退款这种涉及钱的场景，宁可多校验一次，也不能让半结构化的文本流进财务流程。

5.3 上下文越长越蠢，会话总线需要主动剪枝

有段时间我为了让决策更准确，把全部历史事件都推给下一个环节，结果发现下游 Agent 反而频繁出错。把日志拉出来看，原来上一个环节的 Agent 在中间态写过一段错误的假设，虽然最终结果被修正了，但那段错误假设还是留在上下文里，污染了下游的判断。

所以 ContextBus 里我做了一条强制规则：下游 Agent 只能通过上游 TaskSlice 的 OutputSchema 派生数据，不能直接翻聊天记录。上游给下游传的数据必须是"经过校验的输出"，而不是"对话过程中所有内容"。这个规则和人的工作习惯一致——同事之间交接的是最终结论和必要备注，而不是把一周的脑内草稿全推给对方。

5.4 幂等键是无人跑批的命根子

在无人值守模式下，最怕的就是重复执行产生资损。没有幂等保护时，如果 execute_refund 这个 TaskSlice 第一次执行成功但数据库提交超时，编排层把它标记为 failed，触发重试，于是客户被扣款两次。这种问题在普通脚本时代靠人工盯着可能还能发现，在无人公司模式下根本防不住。

Paperclip 的幂等设计很简单但很有效：每个 TaskSlice 生成时，用业务键加环节名生成 idempotency_key，比如退款单号+环节序号。执行前先查状态存储里这个 key 是否已经有 succeeded 记录，如果是，直接跳过。这要求下游所有操作都先做"检查并写入"而不是"直接写入"，例如退款操作前先查是否已有同单号的退款流水。

5.5 人在回路不是最后一道防线，而是穿插流程里的"裁判站"

很多人理解的人工介入是"流程全崩了最后找个人看一眼"，但真到崩了才找人已经晚了。Paperclip 把人介入点设计成流程中的裁判站，而不是终点的救火队。比如计算出来的退款金额超过某个阈值、客户在工单里带有明显的情绪化关键词、或者多个环节连续重试失败，这些情况都提前转 waiting_human。

转移给人工时，编排层必须带着完整的证据包上去：任务片 ID、当前状态、输入数据、Agent 的输出原文、校验错误的详情。人工只需要在界面点"确认继续"或"修正参数"，系统会自动重跑。这个设计让我值守的工作量大幅降低：50 单里需要介入的 2 单，都是因为金额异常被提前拦下，Agent 自己大概率也能处理，但涉及钱的事，还是让人看一眼更稳妥。

6. 上线半年后沉淀的配置建议与个人体会

跑了一段时间后，我把很多"凭感觉定"的参数都调成了固定的推荐值。下面这组配置可以当成起点，再根据你的业务微调：

关于模型选择，我用过 GPT-4o、Claude、以及一些开源模型跑同样的编排链路。结论是：编排层的封装让模型之间的差异变得没那么大——只要输出校验能兜住格式问题，业务结构化做得足够好，开源模型也能顶住大部分环节。当然，复杂推理环节（比如多条件优惠金额计算）还是大模型更有优势，但这属于质量调优，不属于架构层面的硬伤。

最后再说一点个人体会。做无人系统最大的反直觉点在于：真正花费精力的地方往往不是模型推理，而是围绕模型建立的"过程可信度"。Paperclip 这类编排层真正解决的，不是让 AI 更聪明，而是让 AI 聪明的时候可以被控制、犯错的时候可以被发现、被纠正的时候不需要推倒重来。如果你正打算把一条人工流程改造成"AI 员工流程"，我强烈建议先把任务流转表和状态机画清楚再动手写代码。这一步省下的时间，远比 mold 训练 or 调 prompt 的收益更直接。

另外，如果你后续想在 Paperclip 基础上扩展，可以考虑给 ContextBus 加一个"全局事实银行"：让多个 Agent 共享一份实时更新的事实库，避免每个 Agent 都从自己的局部上下文里猜测业务状态。这个改法能进一步降低多环节长流程的错误率，也是我下一轮迭代准备做的东西。