做过多步智能体项目的人应该都有同感单步调用模型的时候一切正常一旦把多个步骤串起来系统就开始“原形毕露”。跑一次完整任务短则几十秒长则拖到几分钟中间还动不动冒出一些说不清的错——今天漏了一个字段明天同一个分支又走岔了后天整个链路卡在半路怎么也恢复不了。很多人第一反应是模型不够强、提示词写得不够细、上下文给得不够全可把日志翻烂之后往往才发现真正的瓶颈根本不在模型层而在编排层说得再具体一点就是控制流编排出了问题。这篇文章我打算把这块彻底讲透多步智能体为什么慢、为什么容易错控制流编排在其中到底扮演什么角色以及一套我自己在多个项目里实践过、验证过、能直接落地的编排方案。无论你是在做客服工单处理、批量文档分析还是搭各类自动化Agent这套思路和代码骨架基本都能复用。我会从问题拆解讲到方案设计再讲到参数调优和踩坑实录尽量把每个“为什么”都说清楚。1. 先拆解问题慢和错到底发生在哪个环节1.1 多步智能体的典型执行链路现在主流的多步智能体结构其实大同小异接收任务拆解成计划逐个执行子任务汇总中间结果最后输出答案。听起来很简单但每个环节下面都可能藏着多次模型调用以及工具调用、知识检索、中间校验、人工确认等一堆动作。举个例子一个文档审阅Agent的典型流程可能是这样读取文件 → 拆分章节 → 对每个章节做摘要 → 合并所有章节摘要 → 生成整体审阅意见 → 做一次一致性校验 → 输出最终报告。这条链路看似不长但数一下就会发现光模型调用就有七八次更别提每次调用之前的提示词构造、上下文拼接、返回结果解析和异常判断。每一步都是不确定的任何一步出错都可能让最终结果跑偏。我见过很多团队在这个阶段的做法是把所有步骤全都写在一个巨大的函数里一个函数从上到下依次调用模型错误处理就是最外层的 try-except。这种写法在只有两三个步骤时还能凑合一旦步骤超过五六个复杂度就完全失控了。慢和错其实从这一刻就注定了跟模型本身关系不大。1.2 “慢”的三大来源串行等待、空转等待、重复劳动先说串行等待。如果执行链路上所有步骤都严格按顺序执行七八次模型调用哪怕每次都不算慢整体延迟也会被简单相加。而模型接口的响应时间波动本来就大高峰期一次调用可能就要等十几秒串行下去总耗时直接不可用。第二个是空转等待。很多步骤之间其实并没有严格的先后依赖关系比如上面例子里不同章节的摘要互相之间完全独立但代码里却写成了“上一章摘要结束才能开始下一章”。明明可以并行的部分被无谓地排成了队列这一步是最冤枉的不花一分钱就能省出来的时间就这样被浪费了。第三个是重复劳动。同一个上下文片段、同一个中间结果被反复传给模型或者反复查询外部数据。比如每章摘要之前都重新加载一次整份文档哪怕模型根本不需要那么多内容。这不光拖时间、浪费token还变相增加了出错概率因为喂给模型的输入越杂模型就越容易关注到不该关注的部分。1.3 “易错”的四大高发点分支、状态、失败、回滚关于“易错”我总结了四个最容易出事的环节。首先是分支判断。很多团队喜欢让模型自己决定下一步做什么结果模型一误判整个流程就走偏了后面的步骤再怎么努力都救不回来。这种“自由度拉满”的设计在小规模Demo里很爽到了真实业务里就是灾难。其次是状态不一致。多步骤之间共享一个可变状态对象一个分支不小心改坏了某个字段其他并行分支读到的就是脏数据。这种问题特别隐蔽因为它不是每次都触发而是看执行顺序和并发时序复现都困难。第三是失败没有兜底。某一步超时或者返回格式异常之后系统没有降级方案直接整条链路失败。用户看到的就是“任务出错请重试”完全不可用。最后是回滚困难。任务走到后半段才发现前置步骤的结果有问题但系统没有恢复机制只能整条从头跑又慢又烧钱。这四类问题单独揪出来都不难解决但它们混在一起时就会把排查难度拉满。问题的核心其实都指向同一个地方控制流编排。2. 控制流编排的本质把随机性关进笼子2.1 控制流不是“流程代码”而是系统的骨架控制流这个概念听起来很抽象说白了就是回答四个问题什么步骤在什么条件下执行这些步骤谁先谁后哪些可以并发到多大程度失败之后怎么处理在多步智能体里模型是随机性的来源而控制流就是约束随机性的骨架。我用一个类比帮助团队理解这件事把智能体想成一条流水线模型是流水线上的工人而控制流是流水线本身的设计。工人再熟练如果传送带设计得乱七八糟一会儿堵塞一会儿漏件产线照样开不动。反过来传送带设计得合理哪怕工人水平普通一些也能稳定地产出合格品。很多团队把精力花在调提示词、换更强的模型上却忽视了对流水线的设计。表现就是换一个更强的模型准确率提升了一两个点但系统整体还是慢、还是时不时崩。这时候就该问问自己是不是骨架本身就撑不住了。2.2 几种常见编排模式与适用边界我梳理了几种最常用的控制流模式各有各的适用场景用错了就会出问题。线性串联是最直觉的模式适合步骤之间存在强依赖的场景比如必须先把信息补全才能生成结论。问题在于很多人把所有步骤都默认成线性串联明明没有依赖关系也要排队执行延迟被白白放大。并行扇出/扇入适合多个相互独立的子任务比如多章节并行分析、多数据源同时查询。收益非常大但代价是要处理好结果合并和顺序约定否则扇入阶段很容易乱。条件分支适合路由场景核心原则是分支条件能写成规则的绝对不要交给模型判断。循环反思适合质量要求高的生成任务让模型自己审自己的成果但要设置迭代上限否则很容易出现“反思循环停不下来”的场面。人工审核门适合高影响决策比如退款审批、对外发报告不过每个审核点都会引入人工等待延迟能少用就尽量少用千万不要给每个小步骤都加一道人工确认。2.3 “确定性骨架随机叶子”的设计原则我在实践中反复验证过的一个核心原则可以总结成八个字确定性骨架随机性叶子。意思是把流程中的决策尽量变成可计算的确定性规则只在真正需要语义理解的地方才调用模型。举个例子判断一份文档属于“合同类”还是“公告类”完全可以先用文件名、关键词、格式特征做规则匹配匹配不了的再让模型兜底。这样既稳定又便宜还快。再比如多步依赖关系用有向无环图DAG描述执行器按拓扑序调度能并行的并行该分支的走分支整条链路的顺序、重试、合并全部由确定性的编排器控制而不是交给模型自行发挥。模型调用在这里只是编排器控制下的一个“叶子节点”。叶子内部可以有一些随机性但叶子之外的骨架必须是确定的。这条原则看起来简单却帮我解决了大量“看着是模型问题、其实根本不是”的疑难杂症。3. 从零搭一套可控的编排层以DAG为核心3.1 第一步把业务拆成有向无环图动手写代码之前我建议先把业务流程画成一张有向无环图。还是拿文档审阅Agent来说任务可以拆成这样读取文档唯一入口节点拆分章节依赖读取结果每章关键信息抽取多个实例并行彼此独立都依赖拆分结果生成统一摘要依赖所有抽取结果完成生成审阅意见依赖统一摘要一致性校验依赖审阅意见输出报告依赖校验结果画出来之后你会发现一件很重要的事哪些步骤可以并行是一眼就能看出来的压根不需要靠模型“智能调度”。很多人觉得并行很难其实难的不是并行而是没想清楚依赖关系。依赖关系一理清并行就是顺理成章的事。拆图的时候还有一条经验节点粒度不要太大也不要太小。粒度太大会损失并行机会比如把“所有章节处理”堆成一个节点粒度太小会引入大量编排开销比如连“提取标题”都要单独建一个节点。合理粒度是“一个步骤完成一个完整、可独立验证的业务动作”比如“抽取某章节的合同金额”就是一个很好的粒度。3.2 第二步设计步骤协议与执行器确定好DAG之后下一步就是定义每个步骤的协议。我给每个步骤定义四样东西输入Schema、输出Schema、执行函数、元策略超时时间、重试次数、是否幂等、是否需要人工确认。这四样缺一不可。为什么一定要有输入输出Schema因为模型的输出天然不可控如果不在步骤边界做结构校验错误的数据就会像脏水一样流向后续所有步骤到最后一步才爆炸。到那时你已经分不清是哪个环节出的问题。我习惯在每步执行后立刻做Schema校验不合格就按失败策略处理这样可以尽早暴露问题、尽早重试而不是等到下游一个莫名其妙的报错出现才开始排查。执行器本身不需要多复杂的框架。我用Python写过一版非常轻量的实现核心思路是解析DAG → 拓扑排序 → 按依赖关系调度 → 并行节点放进线程池 → 收集结果 → 持久化每个步骤的输出快照。调度逻辑可以抽象成下面这样from dataclasses import dataclass from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Callable, Optional dataclass class Step: name: str deps: list[str] fn: Callable[..., Any] timeout: int 30 retries: int 2 idempotent: bool True output_schema: Optional[Callable[[Any], bool]] None class DagExecutor: def __init__(self, steps: list[Step], max_workers: int 4): self.steps {s.name: s for s in steps} self.max_workers max_workers self.results {} self.logs [] def run(self, inputs: dict[str, Any]) - dict[str, Any]: # 简单实现逐轮找出所有依赖已满足且未执行的节点 remaining set(self.steps.keys()) while remaining: ready [] for name in remaining: step self.steps[name] if all(dep in self.results for dep in step.deps): ready.append(step) if not ready: raise RuntimeError(检测到循环依赖或缺失依赖无法继续执行) with ThreadPoolExecutor(max_workersself.max_workers) as pool: futures {pool.submit(self._exec, s): s for s in ready} for fut in as_completed(futures): s futures[fut] self.results[s.name] fut.result() remaining - {s.name for s in ready} return self.results def _exec(self, step: Step) - Any: args {dep: self.results[dep] for dep in step.deps} for attempt in range(step.retries 1): try: out step.fn(**args) if step.output_schema and not step.output_schema(out): raise ValueError(f输出不满足Schema约束: {step.name}) return out except Exception as e: if attempt step.retries: raise self.logs.append(f{step.name} 第{attempt 1}次失败: {e}) return None这段代码当然不是生产级完整版但骨架已经能表达我想说的核心依赖关系是显式声明的并发是自动发生的重试和校验是有统一策略的。在此基础上再补上状态持久化、trace日志和人工审核回调就是一套非常够用的编排层。3.3 第三步并行、分支、循环的落地细节并行落地的时候最容易犯的错误是一味追求并发数。线程池开得太大模型接口被限流的概率急剧上升原本想提速反而被限流拖得更慢。我的做法是先根据接口的配额和单次调用的平均耗时估算一个理论上限再打七八折作为实际并发数。比如单次调用平均2秒、接口每分钟允许600次那理论上并发上限是20实际跑18就差不多了留一点余量给突发波动。分支逻辑落地的第一原则前面提过能算则不判。判断条件如果可以用字符串匹配、数值比较、正则表达式实现就坚决不上模型。实在需要模型做判断时不要让模型自由发挥而是给它一个选项白名单让它输出结构化结果比如 JSON 里的 category 字段只能从给定枚举里选一个然后再做一次白名单校验。模型输出不在白名单里时就重试或者走默认分支。循环逻辑比如反思、自我校验最核心的约束是必须设上限。我见过一个项目里反思循环没有设最大迭代次数某天模型连续几次都觉得“还能更好”一口气反思了二十多轮等团队发现时已经烧掉了几百块的token。设置 max_iterations 之外还要设计收敛条件每轮迭代后对比本轮结果和上轮结果变化小于阈值就自动退出不要傻乎乎地每次跑满上限。3.4 第四步失败策略与断点续跑失败策略这块我的原则是能重试的重试不能重试的补偿补偿不了的再人工介入。判断能不能重试的标准是幂等性——这个步骤重复执行会不会产生额外的副作用。一个纯粹的信息抽取步骤是幂等的重试没风险而一个“发送通知”的步骤就不幂等重试之前必须先检查上一次是否已经发送成功。重试的退避策略我一般用指数退避第一次失败等1秒第二次等2秒第三次等4秒最多三次。这个策略在大多数接口上足够友好也不会因为频繁重试造成二次限流。断点续跑是我最想强调的一个能力。很多团队的任务失败之后只能整条链路从头跑浪费的不仅是时间还有钱。正确做法是每个步骤执行完后立刻把输出结果持久化失败时记录失败点和失败原因下次恢复时直接从失败点对应的最近可用快照开始重跑失败步骤及其下游即可。我实际用的是“结果快照 重放”的组合。每个步骤的输出按步骤名存下来恢复时检查哪些步骤的结果快照还在、哪些已经失效失效的才重新执行。效果非常明显一个正常要跑两三分钟的链路在某个叶子节点失败后恢复通常只需要重跑那一个节点和它的下游总耗时能压到十几秒。4. 关键参数怎么调耗时、并发、重试的实测经验4.1 串行与并行优化后的耗时对比为了让上面的方案更直观我放一组我们团队在某文档分析项目里的平均实测数据。注意这是特定场景下的相对参考不同业务、不同模型差异会很大但趋势是一致的。执行方式完整链路耗时模型调用次数失败重跑耗时全串行执行85秒885秒全部重跑DAG并行优化31秒8约14秒只重跑失败分支并行中间结果缓存19秒约5约8秒从串行到DAG并行耗时从85秒降到31秒提升非常可观而且这一步几乎不增加任何成本只是把原本没有依赖关系的步骤从“排队”改成了“并发”。再加上中间结果缓存命中之后连模型调用都能省掉才是真正省钱的开始。4.2 超时、重试与并发度的取舍超时设置太短会误杀正常慢响应太长又会拖垮整体。我的经验是先统计模型接口在业务高峰期的 p95 耗时也就是最慢的5%请求之外的耗时上限然后在此基础上加30%到50%的缓冲。比如 p95 是8秒那超时设成12秒左右比较合适。注意超时是给单次调用设的不是给整个步骤设的步骤层还需要额外的总时长上限避免多次重试叠加后把整个任务拖死。重试参数前面说过指数退避这里补充一个细节重试之前先查一次目标步骤的是否已成功如果结果快照里已经有成功记录了就直接复用不要再调一次模型。这个细节在“超时了但服务端其实已经处理完”的场景下特别重要可以有效避免重复副作用。并发度的取舍上我见过两种极端一种是把并发数设成1浪费所有并行机会另一种是盲目开到几十上百结果接口限流报错一堆重试又进一步放大请求量。正确的做法是从接口配额反推设置一个安全并发上限并且给线程池配上信号量防止多个步骤同时启动时瞬时请求量超过上限。4.3 上下文精简与缓存复用带来的真实收益很多多步智能体越跑越慢问题出在上下文膨胀上。每完成一个步骤就把这个步骤的完整输出追加到上下文里传给下一步的模型。几轮下来上下文里塞满了中间环节的原文又长又杂模型推理速度变慢费用飙升还容易受无关信息干扰。我改成了两个策略一是只传必要字段每步从上游结果里挑出下一步真正依赖的数据二是对长文本做摘要压缩保留结构和关键结论丢弃原始过程信息。缓存复用这块我比较喜欢做的是“步骤级的结果缓存”。在规则上只要某个步骤的输入完全一致它的模型调用结果就可以安全复用。比如多份文档里都引用了同一段公司简介抽取公司信息这个步骤的输入是一样的直接命中缓存跳过一次模型调用。实测中合理设计上下文精简和缓存之后总耗时还能再降三成左右相比之下单纯的并行优化只能解决一部分问题上下文膨胀才是那个藏在背后的“隐形杀手”。5. 常见问题排查与避坑实录5.1 高频问题速查表我把项目里频繁遇到的控制流编排问题整理成了一张速查表排查的时候可以按图索骥。问题现象常见原因解决方向任务跑到一半失败但不知道哪一步缺乏步骤级日志和trace链路每一步都记录输入输出、耗时、重试次数用trace_id贯穿并行分支的结果互相污染共享了可变状态对象改为步骤间只传递不可变数据快照模型误判分支导致流程走偏分支决策交给模型自由发挥改为白名单枚举校验规则能判断就不用模型反思循环停不下来烧掉大量token缺少迭代上限和收敛条件设置max_iterations增加变化阈值收敛判断流程卡在某个步骤没有响应单步超时设置过长细化超时策略按p95耗时加缓冲恢复任务时整条链路从零重跑没有断点续跑机制增加结果快照持久化从失败点恢复多步调用后被接口限流并发数超出配额根据配额计算安全并发加信号量控制这张表不能覆盖所有情况但覆盖了我见过的八成以上问题。大部分团队拿着一套日志工具和数据追踪需求去对照很快就能定位到瓶颈层。5.2 定位“坏在哪一步”的调试思路调试多步智能体和调试普通程序最大的区别在于普通程序出错是确定的同样的输入基本复现同样的错误而智能体出错是概率性的同一条链路可能这次成功下次失败。正因为这样调试时一定要先看编排日志而不是先看模型日志。我习惯在所有关键节点埋点每个步骤开始、结束、重试、失败、结果快照落盘都打一条结构化日志里面带上trace_id、步骤名、耗时、输入输出的摘要信息。这样一旦任务失败顺着trace_id就能看到它在哪个步骤停了、重试了几次、返回了什么内容。这一步是最基础也最容易被忽视的。做最小化复现也是非常重要的手段。如果怀疑某个分支有问题单独把那个分支抽出来固定输入反复跑几十次统计失败率和失败模式。能复现出来问题就好解决的就怕那种偶发、无法复现的基本都是状态污染或者并发时序问题。这类问题我建议直接检查共享状态和并发变量而不是去猜模型能力。5.3 我踩过的几个真实典型的坑最后分享三个我在项目里踩过的坑每个都花了不少代价才走出来。第一个坑是让模型自由决定下一步。某个客服工单Agent早期版本里模型可以自己选择“查订单、查库存、发起退款”等动作。看起来非常灵活实际上模型经常选错动作而且选错以后没有任何机制纠偏。后来我把动作选择改成了“选项白名单规则预筛”只有规则判不了的情况才让模型选同时强制校验输出必须在白名单内。改完之后误判率直接降了一个数量级。第二个坑是并行分支共享状态对象。有一次我把多个并行抽取步骤的结果写进同一个字典以为每个步骤只写自己的key就没事。结果某一天两个分支的模型调用顺序错乱写入了同一个key后面的分支覆盖了前面的结果最后汇总时数据张冠李戴。排查了很久才发现是共享状态在并发下被竞争修改。后来我改成每个步骤的输出都是独立快照互不修改只允许执行器统一合并。第三个坑是反思循环没有耗竭保护。某个生成类Agent带自我校验设计时忘了设上限。某次模型连续审了多轮都觉得自己写得“还不够好”循环一直没退出最后烧掉了大量token任务还没完成。加了 max_iterations 上限和收敛阈值之后质量没有变差反而更稳定了省下的钱更不用说。写在最后的实践建议这篇文章写到这里核心内容已经讲完了。最后再分享一点个人体会控制流编排这件事门槛其实不在技术而在思维方式。很多人习惯把所有决策都交给模型觉得这样才“智能”但真实的工程恰恰相反——尽可能多的确定性只在必要处保留随机性才是多步智能体稳定高效的真正秘诀。如果你正在建设一个新的多步智能体我的建议是从第一步就把DAG思路引入先把依赖关系图画清楚再写任何业务代码。如果已经有一个跑得又慢又爱出错的存量系统也不要急着推翻重写按文中的顺序逐步改造即可先加步骤级日志再做依赖分析再上并行和快照恢复。每一步都能立竿见影地感受到变化。这个内容后续还可以往很多方向扩展比如加入更复杂的人工审核工作流或者在编排层之上做自动的参数调优。但无论怎么扩展骨架稳定、边界清晰、失败可控这三点是永远不变的地基。希望这些经验能帮你少走一些弯路。
做过多步智能体项目的人应该都有同感单步调用模型的时候一切正常一旦把多个步骤串起来系统就开始“原形毕露”。跑一次完整任务短则几十秒长则拖到几分钟中间还动不动冒出一些说不清的错——今天漏了一个字段明天同一个分支又走岔了后天整个链路卡在半路怎么也恢复不了。很多人第一反应是模型不够强、提示词写得不够细、上下文给得不够全可把日志翻烂之后往往才发现真正的瓶颈根本不在模型层而在编排层说得再具体一点就是控制流编排出了问题。这篇文章我打算把这块彻底讲透多步智能体为什么慢、为什么容易错控制流编排在其中到底扮演什么角色以及一套我自己在多个项目里实践过、验证过、能直接落地的编排方案。无论你是在做客服工单处理、批量文档分析还是搭各类自动化Agent这套思路和代码骨架基本都能复用。我会从问题拆解讲到方案设计再讲到参数调优和踩坑实录尽量把每个“为什么”都说清楚。1. 先拆解问题慢和错到底发生在哪个环节1.1 多步智能体的典型执行链路现在主流的多步智能体结构其实大同小异接收任务拆解成计划逐个执行子任务汇总中间结果最后输出答案。听起来很简单但每个环节下面都可能藏着多次模型调用以及工具调用、知识检索、中间校验、人工确认等一堆动作。举个例子一个文档审阅Agent的典型流程可能是这样读取文件 → 拆分章节 → 对每个章节做摘要 → 合并所有章节摘要 → 生成整体审阅意见 → 做一次一致性校验 → 输出最终报告。这条链路看似不长但数一下就会发现光模型调用就有七八次更别提每次调用之前的提示词构造、上下文拼接、返回结果解析和异常判断。每一步都是不确定的任何一步出错都可能让最终结果跑偏。我见过很多团队在这个阶段的做法是把所有步骤全都写在一个巨大的函数里一个函数从上到下依次调用模型错误处理就是最外层的 try-except。这种写法在只有两三个步骤时还能凑合一旦步骤超过五六个复杂度就完全失控了。慢和错其实从这一刻就注定了跟模型本身关系不大。1.2 “慢”的三大来源串行等待、空转等待、重复劳动先说串行等待。如果执行链路上所有步骤都严格按顺序执行七八次模型调用哪怕每次都不算慢整体延迟也会被简单相加。而模型接口的响应时间波动本来就大高峰期一次调用可能就要等十几秒串行下去总耗时直接不可用。第二个是空转等待。很多步骤之间其实并没有严格的先后依赖关系比如上面例子里不同章节的摘要互相之间完全独立但代码里却写成了“上一章摘要结束才能开始下一章”。明明可以并行的部分被无谓地排成了队列这一步是最冤枉的不花一分钱就能省出来的时间就这样被浪费了。第三个是重复劳动。同一个上下文片段、同一个中间结果被反复传给模型或者反复查询外部数据。比如每章摘要之前都重新加载一次整份文档哪怕模型根本不需要那么多内容。这不光拖时间、浪费token还变相增加了出错概率因为喂给模型的输入越杂模型就越容易关注到不该关注的部分。1.3 “易错”的四大高发点分支、状态、失败、回滚关于“易错”我总结了四个最容易出事的环节。首先是分支判断。很多团队喜欢让模型自己决定下一步做什么结果模型一误判整个流程就走偏了后面的步骤再怎么努力都救不回来。这种“自由度拉满”的设计在小规模Demo里很爽到了真实业务里就是灾难。其次是状态不一致。多步骤之间共享一个可变状态对象一个分支不小心改坏了某个字段其他并行分支读到的就是脏数据。这种问题特别隐蔽因为它不是每次都触发而是看执行顺序和并发时序复现都困难。第三是失败没有兜底。某一步超时或者返回格式异常之后系统没有降级方案直接整条链路失败。用户看到的就是“任务出错请重试”完全不可用。最后是回滚困难。任务走到后半段才发现前置步骤的结果有问题但系统没有恢复机制只能整条从头跑又慢又烧钱。这四类问题单独揪出来都不难解决但它们混在一起时就会把排查难度拉满。问题的核心其实都指向同一个地方控制流编排。2. 控制流编排的本质把随机性关进笼子2.1 控制流不是“流程代码”而是系统的骨架控制流这个概念听起来很抽象说白了就是回答四个问题什么步骤在什么条件下执行这些步骤谁先谁后哪些可以并发到多大程度失败之后怎么处理在多步智能体里模型是随机性的来源而控制流就是约束随机性的骨架。我用一个类比帮助团队理解这件事把智能体想成一条流水线模型是流水线上的工人而控制流是流水线本身的设计。工人再熟练如果传送带设计得乱七八糟一会儿堵塞一会儿漏件产线照样开不动。反过来传送带设计得合理哪怕工人水平普通一些也能稳定地产出合格品。很多团队把精力花在调提示词、换更强的模型上却忽视了对流水线的设计。表现就是换一个更强的模型准确率提升了一两个点但系统整体还是慢、还是时不时崩。这时候就该问问自己是不是骨架本身就撑不住了。2.2 几种常见编排模式与适用边界我梳理了几种最常用的控制流模式各有各的适用场景用错了就会出问题。线性串联是最直觉的模式适合步骤之间存在强依赖的场景比如必须先把信息补全才能生成结论。问题在于很多人把所有步骤都默认成线性串联明明没有依赖关系也要排队执行延迟被白白放大。并行扇出/扇入适合多个相互独立的子任务比如多章节并行分析、多数据源同时查询。收益非常大但代价是要处理好结果合并和顺序约定否则扇入阶段很容易乱。条件分支适合路由场景核心原则是分支条件能写成规则的绝对不要交给模型判断。循环反思适合质量要求高的生成任务让模型自己审自己的成果但要设置迭代上限否则很容易出现“反思循环停不下来”的场面。人工审核门适合高影响决策比如退款审批、对外发报告不过每个审核点都会引入人工等待延迟能少用就尽量少用千万不要给每个小步骤都加一道人工确认。2.3 “确定性骨架随机叶子”的设计原则我在实践中反复验证过的一个核心原则可以总结成八个字确定性骨架随机性叶子。意思是把流程中的决策尽量变成可计算的确定性规则只在真正需要语义理解的地方才调用模型。举个例子判断一份文档属于“合同类”还是“公告类”完全可以先用文件名、关键词、格式特征做规则匹配匹配不了的再让模型兜底。这样既稳定又便宜还快。再比如多步依赖关系用有向无环图DAG描述执行器按拓扑序调度能并行的并行该分支的走分支整条链路的顺序、重试、合并全部由确定性的编排器控制而不是交给模型自行发挥。模型调用在这里只是编排器控制下的一个“叶子节点”。叶子内部可以有一些随机性但叶子之外的骨架必须是确定的。这条原则看起来简单却帮我解决了大量“看着是模型问题、其实根本不是”的疑难杂症。3. 从零搭一套可控的编排层以DAG为核心3.1 第一步把业务拆成有向无环图动手写代码之前我建议先把业务流程画成一张有向无环图。还是拿文档审阅Agent来说任务可以拆成这样读取文档唯一入口节点拆分章节依赖读取结果每章关键信息抽取多个实例并行彼此独立都依赖拆分结果生成统一摘要依赖所有抽取结果完成生成审阅意见依赖统一摘要一致性校验依赖审阅意见输出报告依赖校验结果画出来之后你会发现一件很重要的事哪些步骤可以并行是一眼就能看出来的压根不需要靠模型“智能调度”。很多人觉得并行很难其实难的不是并行而是没想清楚依赖关系。依赖关系一理清并行就是顺理成章的事。拆图的时候还有一条经验节点粒度不要太大也不要太小。粒度太大会损失并行机会比如把“所有章节处理”堆成一个节点粒度太小会引入大量编排开销比如连“提取标题”都要单独建一个节点。合理粒度是“一个步骤完成一个完整、可独立验证的业务动作”比如“抽取某章节的合同金额”就是一个很好的粒度。3.2 第二步设计步骤协议与执行器确定好DAG之后下一步就是定义每个步骤的协议。我给每个步骤定义四样东西输入Schema、输出Schema、执行函数、元策略超时时间、重试次数、是否幂等、是否需要人工确认。这四样缺一不可。为什么一定要有输入输出Schema因为模型的输出天然不可控如果不在步骤边界做结构校验错误的数据就会像脏水一样流向后续所有步骤到最后一步才爆炸。到那时你已经分不清是哪个环节出的问题。我习惯在每步执行后立刻做Schema校验不合格就按失败策略处理这样可以尽早暴露问题、尽早重试而不是等到下游一个莫名其妙的报错出现才开始排查。执行器本身不需要多复杂的框架。我用Python写过一版非常轻量的实现核心思路是解析DAG → 拓扑排序 → 按依赖关系调度 → 并行节点放进线程池 → 收集结果 → 持久化每个步骤的输出快照。调度逻辑可以抽象成下面这样from dataclasses import dataclass from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Callable, Optional dataclass class Step: name: str deps: list[str] fn: Callable[..., Any] timeout: int 30 retries: int 2 idempotent: bool True output_schema: Optional[Callable[[Any], bool]] None class DagExecutor: def __init__(self, steps: list[Step], max_workers: int 4): self.steps {s.name: s for s in steps} self.max_workers max_workers self.results {} self.logs [] def run(self, inputs: dict[str, Any]) - dict[str, Any]: # 简单实现逐轮找出所有依赖已满足且未执行的节点 remaining set(self.steps.keys()) while remaining: ready [] for name in remaining: step self.steps[name] if all(dep in self.results for dep in step.deps): ready.append(step) if not ready: raise RuntimeError(检测到循环依赖或缺失依赖无法继续执行) with ThreadPoolExecutor(max_workersself.max_workers) as pool: futures {pool.submit(self._exec, s): s for s in ready} for fut in as_completed(futures): s futures[fut] self.results[s.name] fut.result() remaining - {s.name for s in ready} return self.results def _exec(self, step: Step) - Any: args {dep: self.results[dep] for dep in step.deps} for attempt in range(step.retries 1): try: out step.fn(**args) if step.output_schema and not step.output_schema(out): raise ValueError(f输出不满足Schema约束: {step.name}) return out except Exception as e: if attempt step.retries: raise self.logs.append(f{step.name} 第{attempt 1}次失败: {e}) return None这段代码当然不是生产级完整版但骨架已经能表达我想说的核心依赖关系是显式声明的并发是自动发生的重试和校验是有统一策略的。在此基础上再补上状态持久化、trace日志和人工审核回调就是一套非常够用的编排层。3.3 第三步并行、分支、循环的落地细节并行落地的时候最容易犯的错误是一味追求并发数。线程池开得太大模型接口被限流的概率急剧上升原本想提速反而被限流拖得更慢。我的做法是先根据接口的配额和单次调用的平均耗时估算一个理论上限再打七八折作为实际并发数。比如单次调用平均2秒、接口每分钟允许600次那理论上并发上限是20实际跑18就差不多了留一点余量给突发波动。分支逻辑落地的第一原则前面提过能算则不判。判断条件如果可以用字符串匹配、数值比较、正则表达式实现就坚决不上模型。实在需要模型做判断时不要让模型自由发挥而是给它一个选项白名单让它输出结构化结果比如 JSON 里的 category 字段只能从给定枚举里选一个然后再做一次白名单校验。模型输出不在白名单里时就重试或者走默认分支。循环逻辑比如反思、自我校验最核心的约束是必须设上限。我见过一个项目里反思循环没有设最大迭代次数某天模型连续几次都觉得“还能更好”一口气反思了二十多轮等团队发现时已经烧掉了几百块的token。设置 max_iterations 之外还要设计收敛条件每轮迭代后对比本轮结果和上轮结果变化小于阈值就自动退出不要傻乎乎地每次跑满上限。3.4 第四步失败策略与断点续跑失败策略这块我的原则是能重试的重试不能重试的补偿补偿不了的再人工介入。判断能不能重试的标准是幂等性——这个步骤重复执行会不会产生额外的副作用。一个纯粹的信息抽取步骤是幂等的重试没风险而一个“发送通知”的步骤就不幂等重试之前必须先检查上一次是否已经发送成功。重试的退避策略我一般用指数退避第一次失败等1秒第二次等2秒第三次等4秒最多三次。这个策略在大多数接口上足够友好也不会因为频繁重试造成二次限流。断点续跑是我最想强调的一个能力。很多团队的任务失败之后只能整条链路从头跑浪费的不仅是时间还有钱。正确做法是每个步骤执行完后立刻把输出结果持久化失败时记录失败点和失败原因下次恢复时直接从失败点对应的最近可用快照开始重跑失败步骤及其下游即可。我实际用的是“结果快照 重放”的组合。每个步骤的输出按步骤名存下来恢复时检查哪些步骤的结果快照还在、哪些已经失效失效的才重新执行。效果非常明显一个正常要跑两三分钟的链路在某个叶子节点失败后恢复通常只需要重跑那一个节点和它的下游总耗时能压到十几秒。4. 关键参数怎么调耗时、并发、重试的实测经验4.1 串行与并行优化后的耗时对比为了让上面的方案更直观我放一组我们团队在某文档分析项目里的平均实测数据。注意这是特定场景下的相对参考不同业务、不同模型差异会很大但趋势是一致的。执行方式完整链路耗时模型调用次数失败重跑耗时全串行执行85秒885秒全部重跑DAG并行优化31秒8约14秒只重跑失败分支并行中间结果缓存19秒约5约8秒从串行到DAG并行耗时从85秒降到31秒提升非常可观而且这一步几乎不增加任何成本只是把原本没有依赖关系的步骤从“排队”改成了“并发”。再加上中间结果缓存命中之后连模型调用都能省掉才是真正省钱的开始。4.2 超时、重试与并发度的取舍超时设置太短会误杀正常慢响应太长又会拖垮整体。我的经验是先统计模型接口在业务高峰期的 p95 耗时也就是最慢的5%请求之外的耗时上限然后在此基础上加30%到50%的缓冲。比如 p95 是8秒那超时设成12秒左右比较合适。注意超时是给单次调用设的不是给整个步骤设的步骤层还需要额外的总时长上限避免多次重试叠加后把整个任务拖死。重试参数前面说过指数退避这里补充一个细节重试之前先查一次目标步骤的是否已成功如果结果快照里已经有成功记录了就直接复用不要再调一次模型。这个细节在“超时了但服务端其实已经处理完”的场景下特别重要可以有效避免重复副作用。并发度的取舍上我见过两种极端一种是把并发数设成1浪费所有并行机会另一种是盲目开到几十上百结果接口限流报错一堆重试又进一步放大请求量。正确的做法是从接口配额反推设置一个安全并发上限并且给线程池配上信号量防止多个步骤同时启动时瞬时请求量超过上限。4.3 上下文精简与缓存复用带来的真实收益很多多步智能体越跑越慢问题出在上下文膨胀上。每完成一个步骤就把这个步骤的完整输出追加到上下文里传给下一步的模型。几轮下来上下文里塞满了中间环节的原文又长又杂模型推理速度变慢费用飙升还容易受无关信息干扰。我改成了两个策略一是只传必要字段每步从上游结果里挑出下一步真正依赖的数据二是对长文本做摘要压缩保留结构和关键结论丢弃原始过程信息。缓存复用这块我比较喜欢做的是“步骤级的结果缓存”。在规则上只要某个步骤的输入完全一致它的模型调用结果就可以安全复用。比如多份文档里都引用了同一段公司简介抽取公司信息这个步骤的输入是一样的直接命中缓存跳过一次模型调用。实测中合理设计上下文精简和缓存之后总耗时还能再降三成左右相比之下单纯的并行优化只能解决一部分问题上下文膨胀才是那个藏在背后的“隐形杀手”。5. 常见问题排查与避坑实录5.1 高频问题速查表我把项目里频繁遇到的控制流编排问题整理成了一张速查表排查的时候可以按图索骥。问题现象常见原因解决方向任务跑到一半失败但不知道哪一步缺乏步骤级日志和trace链路每一步都记录输入输出、耗时、重试次数用trace_id贯穿并行分支的结果互相污染共享了可变状态对象改为步骤间只传递不可变数据快照模型误判分支导致流程走偏分支决策交给模型自由发挥改为白名单枚举校验规则能判断就不用模型反思循环停不下来烧掉大量token缺少迭代上限和收敛条件设置max_iterations增加变化阈值收敛判断流程卡在某个步骤没有响应单步超时设置过长细化超时策略按p95耗时加缓冲恢复任务时整条链路从零重跑没有断点续跑机制增加结果快照持久化从失败点恢复多步调用后被接口限流并发数超出配额根据配额计算安全并发加信号量控制这张表不能覆盖所有情况但覆盖了我见过的八成以上问题。大部分团队拿着一套日志工具和数据追踪需求去对照很快就能定位到瓶颈层。5.2 定位“坏在哪一步”的调试思路调试多步智能体和调试普通程序最大的区别在于普通程序出错是确定的同样的输入基本复现同样的错误而智能体出错是概率性的同一条链路可能这次成功下次失败。正因为这样调试时一定要先看编排日志而不是先看模型日志。我习惯在所有关键节点埋点每个步骤开始、结束、重试、失败、结果快照落盘都打一条结构化日志里面带上trace_id、步骤名、耗时、输入输出的摘要信息。这样一旦任务失败顺着trace_id就能看到它在哪个步骤停了、重试了几次、返回了什么内容。这一步是最基础也最容易被忽视的。做最小化复现也是非常重要的手段。如果怀疑某个分支有问题单独把那个分支抽出来固定输入反复跑几十次统计失败率和失败模式。能复现出来问题就好解决的就怕那种偶发、无法复现的基本都是状态污染或者并发时序问题。这类问题我建议直接检查共享状态和并发变量而不是去猜模型能力。5.3 我踩过的几个真实典型的坑最后分享三个我在项目里踩过的坑每个都花了不少代价才走出来。第一个坑是让模型自由决定下一步。某个客服工单Agent早期版本里模型可以自己选择“查订单、查库存、发起退款”等动作。看起来非常灵活实际上模型经常选错动作而且选错以后没有任何机制纠偏。后来我把动作选择改成了“选项白名单规则预筛”只有规则判不了的情况才让模型选同时强制校验输出必须在白名单内。改完之后误判率直接降了一个数量级。第二个坑是并行分支共享状态对象。有一次我把多个并行抽取步骤的结果写进同一个字典以为每个步骤只写自己的key就没事。结果某一天两个分支的模型调用顺序错乱写入了同一个key后面的分支覆盖了前面的结果最后汇总时数据张冠李戴。排查了很久才发现是共享状态在并发下被竞争修改。后来我改成每个步骤的输出都是独立快照互不修改只允许执行器统一合并。第三个坑是反思循环没有耗竭保护。某个生成类Agent带自我校验设计时忘了设上限。某次模型连续审了多轮都觉得自己写得“还不够好”循环一直没退出最后烧掉了大量token任务还没完成。加了 max_iterations 上限和收敛阈值之后质量没有变差反而更稳定了省下的钱更不用说。写在最后的实践建议这篇文章写到这里核心内容已经讲完了。最后再分享一点个人体会控制流编排这件事门槛其实不在技术而在思维方式。很多人习惯把所有决策都交给模型觉得这样才“智能”但真实的工程恰恰相反——尽可能多的确定性只在必要处保留随机性才是多步智能体稳定高效的真正秘诀。如果你正在建设一个新的多步智能体我的建议是从第一步就把DAG思路引入先把依赖关系图画清楚再写任何业务代码。如果已经有一个跑得又慢又爱出错的存量系统也不要急着推翻重写按文中的顺序逐步改造即可先加步骤级日志再做依赖分析再上并行和快照恢复。每一步都能立竿见影地感受到变化。这个内容后续还可以往很多方向扩展比如加入更复杂的人工审核工作流或者在编排层之上做自动的参数调优。但无论怎么扩展骨架稳定、边界清晰、失败可控这三点是永远不变的地基。希望这些经验能帮你少走一些弯路。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
