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

Paperclip 是给无人公司准备的编排层这点我得先说明白。过去两年我做过好多个跑在模型 API 上的自动化流程最后都死在同一件事上不是模型不聪明而是流程没有纪律。你可以让一个智能体自己查资料、写代码、发邮件但这只是单兵作战。一旦把它放进一个真正要它自己跑起来、有上下游交接、有异常要处理、有人要兜底的业务系统里你需要的是一层比模型本身更靠得住的骨架。Paperclip 就是冲着这个去的。这套设计不依赖某个厂商的 agent 框架核心是一套任务拆解、上下文传递、状态恢复和人工接管的通用机制。如果你正在搭多智能体系统或者想把某条业务流水线变成AI 员工闭环这篇文章里的架构拆分、最小代码骨架和踩坑记录应该能帮你省下一轮弯路。1. 为什么无人公司真正的卡点不是模型是编排1.1 单体 Agent 扛不住一个完整业务流程很多人第一次做多智能体项目时最容易犯的错是把一整条业务链路写进一个大 prompt寄希望于底层的长上下文模型能自己把每个环节执行完。我实测过的结果是——任务链一旦超过四五个环节单体智能体的整体成功率会断崖式下降。原因很朴素大语言模型本质上没有状态保持和过程控制的能力它擅长的是在某个具体节点上给出一个高质量输出。你让它一口气完成接收客户订单、查库存、计算优惠、生成采购单、通知仓库、更新财务记录越到后面前面环节的细节越容易被遗忘甚至会在第五步把第三步已经算好的金额重新算一遍还算错了。这就像让一个新员工独自负责整个公司所有部门的事他迟早会乱。更好理解的方式是把它当作分部门协作。每个智能体只负责一段明确定义的工作做完之后把结果交给下一个环节。这样每个 Agent 的上下文窗口不必装下整条流水线只需要理解自己这一段的输入和预期输出。于是核心问题就从怎么把大模型训得更聪明变成了怎么把多个 Agent 的协作流程组织起来——这就是编排层存在的根本理由。1.2 编排层到底编排的是什么编排层不关心单个 Agent 内部用什么模型、怎么写 prompt它关心的是五个要素任务Task流程要被执行的最小单元必须有明确的输入、输出和验收条件工具ToolAgent 执行任务时的外部动作通道比如查数据库、调 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_typetask.slice_type, contextcontext) 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, errorstr(e), retryableFalse) except TimeoutError as e: state_store.mark_failed(task_id, errorstr(e), retryableTrue)注意worker 本身不感知整个业务不需要知道退款计算还是工单分类背后发生了什么。它只做三件事取上下文、跑 Agent、校验输出。真正干活的 AgentRunner 你可以自己随意实现可以用 OpenAI 的函数调用也可以用本地模型甚至某些环节直接写死规则返回编排层一概不管。这个解耦非常关键它保证你在换模型或者调整 prompt 的时候不用碰调度逻辑。3.3 状态机转移表一张表写清所有恢复策略很多编排项目做到后面失控就是因为失败处理逻辑散落在各处 if-else 里。Paperclip 把所有转移路径明确定义成一张表新增一个状态流转之前必须先在表里找到对应位置当前状态事件下一状态处理动作pending前置依赖完成running投递 worker 队列running执行成功succeeded校验通过发布 ContextBus 消息running输出格式错误failed记录错误不重试人工介入running调用工具超时retrying指数退避retry_count1retrying重试成功succeeded校验通过retrying超过最大重试waiting_human转到人工队列waiting_human人工确认继续running重置 retry_count重新投递failed人工修正输入pending重新排队succeeded下游补偿需要compensated执行反向操作如撤销退款有这张表之后新增业务流的开发成本会显著下降。团队只需要约定好每个 TaskSlice 属于哪种失败类型剩下就交给编排层统一处置。4. 实战验证一个无人客服订单流程的压测记录4.1 场景设定与任务流转为了验证这套东西不是玩具我搭了一条无人客服订单处理链路业务规则是客户提交退款申请系统自动判断资格、计算金额、执行退款、发送通知。没有人工参与除非出现异常。任务流转拆成五个 TaskSliceparse_ticket解析工单、check_eligibility检查退款条件、calculate_amount计算退款金额、execute_refund执行退款、notify_user通知用户。每个 Agent 使用同一个基础模型不单独微调只通过不同的 system prompt 限定角色和输出格式。真正跑起来之前我先在单 Agent 模式下做了一次对照组同一个流程塞进一个大 prompt 让一个 Agent 从头跑到尾。对照组跑 50 单成功完成全部五个环节的只有 34 单成功率 68%失败大多发生在执行退款后忘了通知用户或者金额计算和检查资格的阶段字段对不上。这个数据和我之前的预估基本一致环节越多单体模式越不稳定。4.2 编排模式下的数据表现切到 Paperclip 编排模式后同样 50 单压测结果如下指标单 Agent 直跑Paperclip 编排全流程成功率68%91%平均单次耗时4 分 20 秒3 分 05 秒需人工介入单数无记录2 单数据不一致单数6 单1 单成功率提升的背后主要得益于三个改进。第一个改进是上下文隔离每个 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. 上线半年后沉淀的配置建议与个人体会跑了一段时间后我把很多凭感觉定的参数都调成了固定的推荐值。下面这组配置可以当成起点再根据你的业务微调配置项推荐值说明单任务超时90 秒超过 90 秒的一律转 retrying不盲目等待工具调用超时20 秒第三方 API 超过 20 秒基本是网络问题重试最大重试次数3 次超过 3 次转人工防止 Agent 无限折腾空转检测阈值连续 3 次无进展统计工具调用和输出变化无变化即空转上下文投影字段数不超过 20 个每个 TaskSlice 给 Agent 的上下文字段超过 20 个就要考虑拆分任务人工介入响应超时1 小时人工超过 1 小时未处理发企业微信通知关于模型选择我用过 GPT-4o、Claude、以及一些开源模型跑同样的编排链路。结论是编排层的封装让模型之间的差异变得没那么大——只要输出校验能兜住格式问题业务结构化做得足够好开源模型也能顶住大部分环节。当然复杂推理环节比如多条件优惠金额计算还是大模型更有优势但这属于质量调优不属于架构层面的硬伤。最后再说一点个人体会。做无人系统最大的反直觉点在于真正花费精力的地方往往不是模型推理而是围绕模型建立的过程可信度。Paperclip 这类编排层真正解决的不是让 AI 更聪明而是让 AI 聪明的时候可以被控制、犯错的时候可以被发现、被纠正的时候不需要推倒重来。如果你正打算把一条人工流程改造成AI 员工流程我强烈建议先把任务流转表和状态机画清楚再动手写代码。这一步省下的时间远比 mold 训练 or 调 prompt 的收益更直接。另外如果你后续想在 Paperclip 基础上扩展可以考虑给 ContextBus 加一个全局事实银行让多个 Agent 共享一份实时更新的事实库避免每个 Agent 都从自己的局部上下文里猜测业务状态。这个改法能进一步降低多环节长流程的错误率也是我下一轮迭代准备做的东西。