ProgRouter：多Agent工作流在线进度引导与成本控制

多 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 执行完之后观察到“测试通过率从 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 的自评。你问一个 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_factory=list) # 每轮日志 progress_scores: List[float] = field(default_factory=list) 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", score=score) # 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, })

这个骨架的关键点有三个：

1. 预算保护是最外层防线。进度判断再准，也不能代替硬性预算上限。max_total_calls和max_total_cost是生产环境必备的保险丝。

2. 进度估计器与路由逻辑完全解耦。estimate方法留成接口，你可以把测试通过率、验证器结果或裁判模型评分从外部注入，而不是把打分逻辑写死在路由器里。

3. 停止条件采用三层组合：到达目标分、边际增益不足、全局预算耗尽。后两者是成本控制的主要工具。

这里没有引入复杂的概率模型或强化学习，因为工程落地第一步应该是用规则跑通闭环，再考虑学习型策略。

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_by_cost_budget，说明预算约束太紧；集中在stop_by_progress且质量分不够，说明进度估计偏乐观；集中在stop_by_call_budget，说明单 Agent 的循环太多，应考虑优化切换策略。

如果进度分数曲线呈现先升后降，要警惕 Agent 在“破坏性修改”——后一轮输出把之前正确的内容改错了。这种时候，路由策略应当倾向于保留历史最佳版本（best-so-far），而不是无条件信任最新输出。这也是生产环境最常见但也最容易被忽略的问题。

9. 工程落地的几条建议

从设计到上线，进度引导编排不是只要写好 router 就完事。结合多 Agent 系统落地的通用经验，这里有五条建议。

第一，进度信号先于路由策略投入建设。很多人一上来就写路由策略，结果发现进度分数不可靠，策略再好也白搭。正确的顺序是：先跑一个固定工作流，积累真实任务数据和人工评分，用这些数据训练或校准进度估计器，再开始做动态路由。

第二，为每个任务保留“历史最佳版本”。动态路由必然存在误停或破坏性修改的风险，系统应该在内存中维护最佳输出快照。最终交付时，优先选择历史最佳版本而不是最后一版，能显著降低路由失误的代价。

第三，成本预估要采用真实 token 计数，而不是配置里的平均值。不同模型、不同提示词长度的实际成本差异很大，预算控制应该基于每次调用的实际消耗动态累加。

第四，路由决策必须全量可审计。每次路由动作（继续、换手、停止）的原因要写入日志，并且要能支持回放。上线后如果质量出问题，你需要在十分钟内定位到“是某次误停导致的”，而不是对着黑盒系统猜。

第五，渐进式灰度。不要一次性把所有流量切到新路由策略，先让固定流水线和进度引导策略并行运行一段时间，用真实流量的质量成本数据做对比，确认稳定后再调整流量比例。高风险的业务场景可以加一个人工确认环节：路由决定停止时，把最终结果发送给人工审核，确认通过才交付。

10. 总结与进一步的方向

ProgRouter 类型的方案真正解决的是多 Agent 工作流里的一个核心矛盾：质量需要投入，成本需要控制，而两者之间的平衡点只有在执行过程中才能找到。把编排从静态流程变成在线决策，用进度信号引导继续、换手与停止，是这类系统的共同设计主线。它看似只是加了一个“何时停止”的判断，实际改变的是工作流的决策机制：从“按计划执行”变成“按状态执行”。

如果你要动手实践，建议从最小闭环开始：先定义你任务域里的进度信号，用固定工作流积累一批带真实评分的数据，再实现一个带预算保护的路由器，离线回放对比效果，最后灰度上线。不要一上来就追求学习型策略，规则策略跑通后，你会对进度信号的质量有更深的理解，这时候再考虑用学习模型替代手工阈值也不迟。

后续值得深入的方向包括：进度信号的自动校准、让路由策略根据历史任务动态调整停止阈值、把进度不确定性纳入决策（低置信度时多跑一轮，高置信度时提前停止），以及多目标权衡中如何引入用户可配置的偏好参数。这些方向都建立在同一个基础之上：你能可靠地回答“任务现在进行到哪了”，而这恰恰是大多数多 Agent 系统目前最薄弱的地方。