多 Agent LLM 工作流正在从一个“炫技概念”变成真实业务里的基础设施规划 Agent 拆任务编码 Agent 写代码审查 Agent 找问题如此循环。但凡是真正把这个流程跑上线的团队几乎都会撞到同一个矛盾Agent 配得越多、循环得越深质量提升的边际越来越小token 账单却还在稳定增长。更头疼的是任务的难度在一开始是未知的——同一个工作流处理“写个排序函数”和“重构一整个订单模块”的复杂度完全不同可固定流程不知道这个差异它只会一视同仁地把所有 Agent 全跑一遍。ProgRouter 这个研究方向就是冲着这个矛盾去的。从题目来看它做的是“在线进度引导的编排”Online Progress-Guided Orchestration核心思路很直接给编排器装上一块“进度面板”实时判断任务进行到哪一步、距离目标还有多远然后基于这个判断决定下一步是继续、换手还是停止。它要解决的问题不是单个 Agent 的生成质量而是整个多 Agent 工作流在质量与成本之间的动态权衡。这篇文章不打算照搬论文公式而是把 ProgRouter 背后的设计逻辑拆开讲为什么要做在线决策而非固定流程进度信号应该怎么定义、怎么度量路由决策如何和预算控制结合起来以及如果你想在项目里落地类似机制代码骨架、配置、验证方式和常见坑分别是什么。1. 多 Agent 工作流的成本失控与质量焦虑先回到一个基本问题为什么需要多 Agent单 Agent 已经能完成写作、编码、总结但复杂任务往往需要多个角色配合——一个 Agent 自己边规划边执行容易出现“自圆其说”的盲区拆成规划、执行、审查之后每一步都有独立上下文不同模型还可以按角色分配不同能力侧重。这是多 Agent 架构在工程上被接受的根本原因。可一旦上了多 Agent新的问题立刻出现。第一是固定流水线的浪费。最常见的工作流写法是把 5 个 Agent 串成一个固定 DAG拆题 → 设计 → 编码 → 审查 → 修复每个阶段强制执行。对复杂任务这套流程是必要的但对简单任务它把大量 token 花在了不必要的“仪式感”上。实际业务里混合着各种难度请求固定流程等于让所有任务都按最高规格买单。第二是循环不收敛。很多团队会把审查 Agent 的输出作为反馈让编码 Agent 继续修。这个设计本身没问题但它缺少一个明确的停机制审查者总是能挑出新的小问题编码者总是愿意“再改一版”于是工作流陷入 10 轮、20 轮的无限循环。到最后质量没有显著提升成本已经翻了几倍。第三是路由决策缺少状态感。有的系统做了一定程度的动态路由按任务类型分流或者按用户的模型偏好选路。但这种路由的决策依据是任务开始前的静态特征它不关心任务执行到一半时到底进展如何。真正需要判断的是“当前这个方案距离可用状态还有多远”而不是“这类任务一般用什么路径”。ProgRouter 的核心立场正是把编排问题从静态规划变成在线决策每次 Agent 执行完之后系统先估算当前进度再决定是否值得继续投入成本。这个“先看进度、再决定花不花钱”的模式是它区别于传统工作流引擎的关键。2. 在线进度引导编排的基本思想“Progress-Guided”这个词可以拆成两层理解Progress 是信号Guided 是控制。所谓 Progress指的不是任务经过了几个阶段而是“当前结果离最终质量目标的距离”。比如写代码任务进度可以是单元测试通过率写文档任务进度可以是内容完整度评分数据分析任务进度可以是结论是否达到可交付标准。它是一个 0 到 1 的连续值而不是“做了第几步”这样的是非值。所谓 Guided指的是编排器以进度为中心来生成路由动作。经典的编排是预先定义的条件满足就走分支 A否则走分支 B。进度引导的编排则是循环式每执行完一个 Agent重新评估进度然后从“继续当前 Agent”“切换到另一个 Agent”“整体停止”三个动作里选一个。这个模式与两种常见方案的差别可以用下面这个表说明编排方式决策时机决策依据典型问题固定流水线任务开始前一次性确定任务类型简单任务过度执行复杂任务可能路径过短静态规则路由任务开始前或少量分叉点分类特征、用户属性无法感知执行中状态变化进度引导编排每个 Agent 执行完成后实时进度信号、成本消耗需要可靠进度度量设计复杂度高一句话概括在线性固定流水线是一张提前画好的地图无论路上遇到什么都按原路走进度引导则像一个实时导航每过一个路口就重新看一次路况决定是继续直行、换道还是靠边停车。为什么在线如此重要因为任务难度的信息只有在执行过程中才会逐渐暴露。你无法在请求进来之前知道“这个代码任务测试会不会一遍通过”但你可以在一轮 Agent 执行完之后观察到“测试通过率从 40% 到了 90%”。用在线反馈替代预先假设是 ProgRouter 这类方案质量与成本同时可控的根本原因。需要说明的是这里描述的是从题目和系统设计语言推断出的通用机制。具体论文中的进度估计器可能采用人工规则、小模型打分或 LLM-as-Judge但无论哪种实现逻辑闭环是一致的观测状态 → 估算进度 → 路由决策 → 执行代价 → 更新状态。3. 质量-成本权衡的决策框架要理解 ProgRouter 为什么把质量与成本放在一起说需要先承认一个事实在多 Agent 工作流里质量不是一个可以直接最大化的指标因为每一步质量的提升都有价格。这个价格包括 token 费用、推理延迟、外部 API 调用次数以及失败重试带来的运维成本。如果我们把一次工作流执行看成一段连续的投资过程那么每一轮 Agent 调用都可以看作一笔投资投入成本 C期望换回质量增量 ΔQ。理性的编排策略应该是当 ΔQ 的边际收益大于 C 时继续投资当 ΔQ 趋近于 0 时停止。这就引出了停止条件的设计问题。常见做法有两种一种是绝对阈值。设定一个目标质量分比如“测试覆盖率 95%”、“审查通过”或“答案置信度 0.9”达到即停止。它的优点是直接、可解释缺点是对复杂任务可能永远达不到阈值需要额外兜底。另一种是边际增益阈值。记录连续几轮的质量提升幅度如果最新两轮之间的提升小于某个值比如 2%就认为已经进入收益递减区触发停止。这种方式更灵活能适应不同难度的任务但容易受到进度估计噪声的干扰——评分模型本身波动一下就可能误判为“已经没有提升了”。把二者结合是工程上更稳的做法先满足绝对目标就停止不满足目标时看边际增益是否低于阈值同时再叠加一个全局预算上限作为最后防线。这个三层停止策略就是质量-成本权衡落到代码里的具体形态。从数学模型上很多类似系统会给每次路由决策定义一个效用函数U Quality − λ × Cost其中 λ 是成本相对质量的权重。λ 大意味着系统更在意省钱λ 小意味着更愿意堆成本换质量。不同的业务场景对应不同的 λ给客户生成营销文案λ 可以大一些给生产环境改代码λ 就应该小一些。ProgRouter 的在线决策目标本质上是让每一笔 token 支出都落在效用函数的最优区间。4. 进度信号编排器最关键的“传感器”进度引导编排成不成立几乎完全取决于进度信号靠不靠谱。如果进度分数是错的再好的路由策略都是空中楼阁。这里值得单独讲。进度信号的理想特征有三个单调性、低成本、低噪声。单调性指随着任务推进分数大致上升不能出现严重倒挂低成本指获取这个信号的代价远低于继续调用昂贵 Agent 的代价否则监控比执行还贵低噪声指分数稳定不会因为提示词的微小扰动就剧烈跳动。现实中没有信号能同时满足三点所以工程上要做取舍。常见的进度信号包括信号类型获取方式评估成本主要风险子任务完成清单让规划 Agent 输出 check列表并逐项标记低Agent 自评偏向乐观单元测试/集成测试通过率真实执行测试套件中覆盖不全面时误判LLM-as-Judge 质量打分用裁判模型对当前输出评分中高裁判模型偏差、评分漂移外部验证器结果编译、静态检查、schema 校验低只覆盖形式正确性语义一致性收敛度对比多轮输出之间的相似度中文本相似不等于质量高这里最大的坑是过度信任 Agent 的自评。你问一个 Agent“任务完成了吗”它几乎总是回答“完成了”。这不是模型故意撒谎而是生成式模型缺乏对自身输出的客观校验能力它只能根据训练分布给出一个“看起来合理”的自信答复。因此进度信号最好来自外部客观来源测试结果、验证器、裁判模型的独立评分或者至少是多个信号的综合。另一个常见问题是进度分作用的错位进度分数被用来做“是否停止”的决策但这个分数本身也需要校准。比如某个任务的真实完成度是 0.8而进度估计器给的是 0.95系统就可能提前停止导致交付质量不足。反过来估计器给得保守系统就会在低价值循环里空转。生产环境里进度估计器需要像普通模型一样做定期评估与校准而不是写完一个启发式函数就永久使用。5. 简化实现一个 Progress-Guided Router 骨架理论说完落到代码。这里提供一个可运行的简化骨架帮助你理解路由器的核心逻辑。它不依赖任何特定框架只使用 Python 标准库与 dataclass核心思想可以无缝迁移到 LangGraph、AutoGen 或其他编排平台上。# progress_router.py from dataclasses import dataclass, field from enum import Enum from typing import List, Optional class RouterAction(str, Enum): CONTINUE continue # 继续当前 Agent SWITCH switch # 切换到其他 Agent STOP stop # 终止整个工作流 dataclass class AgentSpec: name: str description: str max_calls: int 3 avg_cost_per_call: float 1.0 dataclass class TaskState: task_id: str goal: str current_output: str history: List[dict] field(default_factorylist) # 每轮日志 progress_scores: List[float] field(default_factorylist) total_cost: float 0.0 total_calls: int 0 class ProgressEstimator: 进度估计器对外部信号的封装重点在于不要直接问 Agent 自己。 def __init__(self, min_gain: float 0.02, target_score: float 0.95): self.min_gain min_gain self.target_score target_score def estimate(self, state: TaskState, agent_output: str) - float: 实际项目中这里可以换成测试通过率、验证器结果或裁判模型评分。 当前实现返回由外部回调注入的分数保证路由逻辑与信号来源解耦。 raise NotImplementedError(请接入你的进度打分实现) def should_stop(self, scores: List[float]) - bool: if not scores: return False latest scores[-1] if latest self.target_score: return True if len(scores) 3: recent_gain latest - scores[-2] if recent_gain self.min_gain: return True return False class ProgressRouter: 在线进度引导路由器。 def __init__( self, agents: List[AgentSpec], estimator: ProgressEstimator, max_total_calls: int 10, max_total_cost: float 50.0, ): self.agents {a.name: a for a in agents} self.estimator estimator self.max_total_calls max_total_calls self.max_total_cost max_total_cost def run(self, state: TaskState) - TaskState: current_agent list(self.agents.keys())[0] while True: # 1. 全局预算保护 if state.total_calls self.max_total_calls: self._log(state, router, stop_by_call_budget) break if state.total_cost self.max_total_cost: self._log(state, router, stop_by_cost_budget) break # 2. 执行当前 Agent实际项目中这里调用 LLM agent self.agents[current_agent] output self._invoke_agent(agent, state) # 3. 估算进度 score self.estimator.estimate(state, output) state.progress_scores.append(score) state.current_output output state.total_calls 1 state.total_cost agent.avg_cost_per_call self._log(state, current_agent, executed, scorescore) # 4. 路由决策 if self.estimator.should_stop(state.progress_scores): self._log(state, router, stop_by_progress) break # 5. 选择下一个 Agent优先给“还未尽力”的角色机会 action self._next_action(state, current_agent) if action RouterAction.STOP: break if action RouterAction.SWITCH: current_agent self._select_next_agent(state, current_agent) return state def _invoke_agent(self, agent: AgentSpec, state: TaskState) - str: # 在你的系统中这里是实际的模型调用 return f[{agent.name}] 针对 {state.goal} 的第 {state.total_calls 1} 轮产出 def _next_action( self, state: TaskState, current_agent: str ) - RouterAction: agent self.agents[current_agent] call_count sum( 1 for h in state.history if h[agent] current_agent ) if call_count agent.max_calls: return RouterAction.SWITCH return RouterAction.CONTINUE def _select_next_agent(self, state: TaskState, current_agent: str) - str: candidates [ name for name, a in self.agents.items() if name ! current_agent and sum(1 for h in state.history if h[agent] name) a.max_calls ] return candidates[0] if candidates else current_agent def _log(self, state: TaskState, source: str, event: str, score: Optional[float] None) - None: state.history.append({ task_id: state.task_id, source: source, event: event, score: score, total_calls: state.total_calls, total_cost: state.total_cost, })这个骨架的关键点有三个预算保护是最外层防线。进度判断再准也不能代替硬性预算上限。max_total_calls和max_total_cost是生产环境必备的保险丝。进度估计器与路由逻辑完全解耦。estimate方法留成接口你可以把测试通过率、验证器结果或裁判模型评分从外部注入而不是把打分逻辑写死在路由器里。停止条件采用三层组合到达目标分、边际增益不足、全局预算耗尽。后两者是成本控制的主要工具。这里没有引入复杂的概率模型或强化学习因为工程落地第一步应该是用规则跑通闭环再考虑学习型策略。6. 工作流定义与配置示例路由逻辑写完之后下一步是让工作流可配置化。最好的做法是把 Agent 列表、调用上限、预算、停止阈值全部外置到配置文件中这样调整策略不需要改代码。# workflow.yaml workflow: id: code-generation-review goal: 生成功能代码并通过测试与审查 agents: - name: planner description: 拆解任务与设计实现方案 max_calls: 1 avg_cost_per_call: 0.5 - name: coder description: 编写或修改代码 max_calls: 3 avg_cost_per_call: 2.0 - name: reviewer description: 代码审查与修改建议 max_calls: 3 avg_cost_per_call: 1.5 - name: tester description: 执行测试套件并报告通过率 max_calls: 2 avg_cost_per_call: 0.3 router: type: progress_guided max_total_calls: 10 max_total_cost: 20.0 termination: target_score: 0.95 min_progress_gain: 0.02 min_rounds: 2配置中的avg_cost_per_call是一个平均值实际项目中可以从调用的 token 数动态计算。min_rounds是防止过早停止的护栏即使分数连续提升不足也要至少执行两轮再判断避免因单次评分抖动导致误停。运行时路由器会输出一份结构化的决策日志方便事后分析。典型的日志片段如下{ task_id: task_1024, rounds: [ { agent: planner, cost: 0.5, score: 0.35, action: continue }, { agent: coder, cost: 2.0, score: 0.62, action: continue }, { agent: tester, cost: 0.3, score: 0.78, action: continue }, { agent: coder, cost: 2.0, score: 0.91, action: continue }, { agent: tester, cost: 0.3, score: 0.94, action: switch }, { agent: reviewer, cost: 1.5, score: 0.96, action: stop } ], total_cost: 6.6, total_calls: 6, stop_reason: target_score_reached }这份日志的价值在于你可以回放每一次路由决策检查“当时为什么没停”“为什么把 reviewer 换上场”是否符合预期。没有日志的编排器在生产环境里几乎是不可维护的。7. 运行验证如何判断路由策略真的有效实现和配置都有了最后也是最关键的问题是怎么证明这套机制比固定流水线好这里给出一条可行的验证路径。第一步离线回放。如果你已经有历史工作流的执行记录可以拿这些记录做模拟把每轮的真实输出与评分喂给新的路由策略看它会在第几轮停止、总成本是多少、最终质量分是多少。这一步不需要真实调用模型成本几乎为零是上线前最有效的筛选手段。第二步质量-成本综合评价。不要只看“平均成本下降了 30%”这种单指标要同时看质量是否保持。常用的评价方式是计算质量成本比并画出 Pareto 曲线# eval.py def evaluate_strategy(results: list[dict]) - dict: results 中的每一项包含: task_id, quality_score, total_cost, stop_reason total_quality sum(r[quality_score] for r in results) total_cost sum(r[total_cost] for r in results) completed sum(1 for r in results if r[quality_score] 0.9) return { avg_quality: round(total_quality / len(results), 3), total_cost: round(total_cost, 2), quality_per_cost: round(total_quality / total_cost, 4), success_rate_90: round(completed / len(results), 3), dist_stop_reason: { reason: sum(1 for r in results if r[stop_reason] reason) for reason in set(r[stop_reason] for r in results) }, }第三设置对照组。最直接的实验是把同一批任务分别用固定流水线、预算上限版本、进度引导版本跑一遍然后对比三个指标平均质量、总成本、超时率。这里要注意评估集要包含不同难度的任务避免全部是简单任务——否则任何带停止机制的策略都会“看起来很好”。判断成功不能只看成本下降。更合理的标准是在质量不降级或降级幅度在业务可接受范围内的前提下成本显著下降同时复杂任务的完成率没有变差。如果进度引导把简单任务的成本降下来了却让复杂任务提前误停、质量崩了说明进度信号在困难样本上的校准还有问题。8. 常见问题与排查思路把类似机制搬到真实项目时以下问题出现频率最高问题现象可能原因排查方式解决方案任务频繁提前停止交付质量不足进度估计器打分偏高对比估计分与人工终审评分计算偏差重新校准打分模型或引入外部验证器信号成本没有明显下降停止条件过宽松或进度信号区分度低检查 stop_reason 分布调低目标分阈值或边际增益阈值检查预算是否过大路由在 Agent 之间反复横跳切换策略没有考虑“当前 Agent 是否接近完成”查看决策日志中 action 变化序列增加最少连续执行轮数约束进度分数震荡明显评分信号噪声大比如 LLM-as-Judge 抖动对同一输出多次评分看方差取多次评分均值或改用确定性验证器简单任务仍然走完整流程子任务完成清单被 Agent 自评污染对比 Agent 自评与外部测试结果用外部客观信号替代自评信号复杂任务超预算全局预算小于复杂任务实际需求按任务难度分层设置预算建立难度预估分配不同档位的 max_total_cost排查路径有个通用顺序先看停止原因分布再看每轮进度分曲线最后看单次路由决策的具体上下文。停止原因集中在stop_by_cost_budget说明预算约束太紧集中在stop_by_progress且质量分不够说明进度估计偏乐观集中在stop_by_call_budget说明单 Agent 的循环太多应考虑优化切换策略。如果进度分数曲线呈现先升后降要警惕 Agent 在“破坏性修改”——后一轮输出把之前正确的内容改错了。这种时候路由策略应当倾向于保留历史最佳版本best-so-far而不是无条件信任最新输出。这也是生产环境最常见但也最容易被忽略的问题。9. 工程落地的几条建议从设计到上线进度引导编排不是只要写好 router 就完事。结合多 Agent 系统落地的通用经验这里有五条建议。第一进度信号先于路由策略投入建设。很多人一上来就写路由策略结果发现进度分数不可靠策略再好也白搭。正确的顺序是先跑一个固定工作流积累真实任务数据和人工评分用这些数据训练或校准进度估计器再开始做动态路由。第二为每个任务保留“历史最佳版本”。动态路由必然存在误停或破坏性修改的风险系统应该在内存中维护最佳输出快照。最终交付时优先选择历史最佳版本而不是最后一版能显著降低路由失误的代价。第三成本预估要采用真实 token 计数而不是配置里的平均值。不同模型、不同提示词长度的实际成本差异很大预算控制应该基于每次调用的实际消耗动态累加。第四路由决策必须全量可审计。每次路由动作继续、换手、停止的原因要写入日志并且要能支持回放。上线后如果质量出问题你需要在十分钟内定位到“是某次误停导致的”而不是对着黑盒系统猜。第五渐进式灰度。不要一次性把所有流量切到新路由策略先让固定流水线和进度引导策略并行运行一段时间用真实流量的质量成本数据做对比确认稳定后再调整流量比例。高风险的业务场景可以加一个人工确认环节路由决定停止时把最终结果发送给人工审核确认通过才交付。10. 总结与进一步的方向ProgRouter 类型的方案真正解决的是多 Agent 工作流里的一个核心矛盾质量需要投入成本需要控制而两者之间的平衡点只有在执行过程中才能找到。把编排从静态流程变成在线决策用进度信号引导继续、换手与停止是这类系统的共同设计主线。它看似只是加了一个“何时停止”的判断实际改变的是工作流的决策机制从“按计划执行”变成“按状态执行”。如果你要动手实践建议从最小闭环开始先定义你任务域里的进度信号用固定工作流积累一批带真实评分的数据再实现一个带预算保护的路由器离线回放对比效果最后灰度上线。不要一上来就追求学习型策略规则策略跑通后你会对进度信号的质量有更深的理解这时候再考虑用学习模型替代手工阈值也不迟。后续值得深入的方向包括进度信号的自动校准、让路由策略根据历史任务动态调整停止阈值、把进度不确定性纳入决策低置信度时多跑一轮高置信度时提前停止以及多目标权衡中如何引入用户可配置的偏好参数。这些方向都建立在同一个基础之上你能可靠地回答“任务现在进行到哪了”而这恰恰是大多数多 Agent 系统目前最薄弱的地方。