上下文预算动态分配器多 Agent 协作系统中的 Token 经济学架构在由规划者Planner、执行者Worker与反思校验者Critic/Verifier组成的多 Agent 协作系统中一个极为普遍的系统级故障是上下文雪崩Context Avalanche。当多个智能体频繁调用工具、相互广播长文本日志和推理结论时系统的总 Token 消耗会呈几何级数爆炸。更致命的是各智能体的上下文窗口迅速被其他节点的冗长废话填满导致注意力分散核心指令严重钝化。许多早期 Agent 框架直接将整条交互总线Message Bus的所有历史消息机械广播给每一个节点。这种粗放的做法在简单两步问答中尚可应付但在面对跨度数十步的复杂长周期任务时往往导致 Critic 在还没开始校验前其上下文就已经被 Worker 的前置试错输出占满最终因超长序列的注意力稀释给出草率的盲目判定。要打破这一恶性循环必须建立一套上下文预算动态分配器Dynamic Context Budget Allocation。多 Agent 动态上下文预算分流架构 [全局交互上下文总线 (80k Token)] │ ▼ ┌────────────────────────────────────────────────────────┐ │ 预算控制器按拓扑角色与当前任务复杂度动态计算配额 │ └────────────────────────────────────────────────────────┘ │ │ │ ▼ (配额: 15k) ▼ (配额: 50k) ▼ (配额: 15k) ┌────────────────┐ ┌────────────────┐ ┌────────────────┐ │ Planner 视角 │ │ Worker 视角 │ │ Critic 视角 │ │ 聚合全局目标树 │ │ 保留细粒度堆栈 │ │ 仅保留契约输入 │ │ 与里程碑状态摘要│ │ 与局部工具反馈 │ │ 与最终输出断言 │ └────────────────┘ └────────────────┘ └────────────────┘一、多 Agent 协作中的 Token 经济学法则在多智能体流水线中不同角色的智能体对上下文的“认知粒度”有着本质区别Planner规划者需要的是高层语义拓扑而非执行碎屑规划者关心的是整体目标被拆分成了哪些子任务、各个子任务的完成状态以及全局依赖图。它根本不需要知道 Worker 在执行第三步时到底调用了哪条具体正则、捕获了什么底层系统错误。Worker执行者需要的是局部精确上下文而非全局宏观论述执行者只关心当前被指派的具体原子任务、上游直接输入契约以及最近几步工具调用的确切返回值。过多的全局战略信息只会挤占其调用工具时的精确参数注意力。Critic评审者需要的是对抗性证据链而非思考流水账评审者需要直接对照用户原始需求Contract Inputs与 Worker 的最终交付结果Artifacts并挂载轻量级的校验规则。如果把 Worker 数十次试错重试的思考过程全部塞给 Critic反而会诱发同理心偏见Sycophancy导致 Critic 误认为“既然你这么努力尝试了那就算你通过吧”。二、动态预算分配算法与信息熵自适应裁决为了在动态交互流中精确控制各 Agent 的输入体积调度器引入了基于信息增益Information Gain的自适应截断方程$$B_i B_{\text{total}} \times \frac{w_i \cdot \mathcal{H}(S_i)}{\sum_j w_j \cdot \mathcal{H}(S_j)}$$其中$B_{\text{total}}$ 为当前会话设定的全局 Token 预算上限$w_i$ 为角色权重矩阵例如当前处于代码编写阶段时$w_{\text{worker}}$ 权重大幅提高处于验收交付阶段时$w_{\text{critic}}$ 占主导$\mathcal{H}(S_i)$ 为该角色接收信息流的条件信息熵。调度器通过分析消息流中唯一实体、异常堆栈与关键指标的变化率动态决定是否实施局部折叠。三、上下文预算分配器的核心实现以下是我们在多智能体调度编排内核中落地的动态预算分配引擎实现支持不同角色视图的差异化编译与裁剪import math from typing import List, Dict, Any class AgentContextAllocator: def __init__(self, total_budget: int 16384): self.total_budget total_budget # 各协作角色的基准权重 self.base_weights { planner: 0.20, worker: 0.55, critic: 0.25 } def _estimate_tokens(self, text: str) - int: 快速轻量估算 Token 数量按平均 1.8 字符/token 估算 return max(1, int(len(text) / 1.8)) def distill_for_planner(self, messages: List[Dict[str, str]], budget: int) - str: 为 Planner 提炼全局任务状态树与里程碑折叠底层工具细节 distilled [] for msg in messages: role msg[role] content msg[content] if role system_goal: distilled.append(f[Global Goal] {content}) elif role task_status: distilled.append(f[Milestone] {content}) elif role worker_summary: distilled.append(f[Worker Result] {content[:200]}...) full_text \n.join(distilled) if self._estimate_tokens(full_text) budget: # 仅保留首部目标与最新的三次里程碑 return \n.join(distilled[:1] distilled[-3:]) return full_text def distill_for_worker(self, messages: List[Dict[str, str]], budget: int) - str: 为 Worker 保留当前子任务定义与高保真工具调用堆栈 distilled [] for msg in messages: if msg[role] in {current_task, tool_feedback, schema_contract}: distilled.append(f{msg[role]}\n{msg[content]}\n/{msg[role]}) # 严格控制在预算之内超限时优先丢弃早期冗余工具反馈 tokens_used sum(self._estimate_tokens(m) for m in distilled) while tokens_used budget and len(distilled) 2: # 丢弃最早的一个工具反馈 distilled.pop(1) tokens_used sum(self._estimate_tokens(m) for m in distilled) return \n\n.join(distilled) def distill_for_critic(self, messages: List[Dict[str, str]], budget: int) - str: 为 Critic 提取需求基线与最终交付成果完全剔除中间尝试过程 baseline deliverable for msg in messages: if msg[role] user_requirement: baseline msg[content] elif msg[role] final_candidate: deliverable msg[content] critic_context ( fverification_contract\n f【原始基准需求】:\n{baseline}\n\n f【待校验交付物】:\n{deliverable}\n f/verification_contract\n f请严格对比上述交付物是否逐条满足基准需求列出所有不合规项。 ) return critic_context def compile_agent_prompts( self, global_history: List[Dict[str, str]], active_stage: str execution ) - Dict[str, str]: 根据流水线活跃阶段动态分配各 Agent 上下文 # 根据执行阶段动态调整预算偏置 weights self.base_weights.copy() if active_stage planning: weights[planner] 0.60 weights[worker] 0.20 weights[critic] 0.20 elif active_stage evaluation: weights[planner] 0.15 weights[worker] 0.25 weights[critic] 0.60 budget_planner int(self.total_budget * weights[planner]) budget_worker int(self.total_budget * weights[worker]) budget_critic int(self.total_budget * weights[critic]) return { planner_prompt: self.distill_for_planner(global_history, budget_planner), worker_prompt: self.distill_for_worker(global_history, budget_worker), critic_prompt: self.distill_for_critic(global_history, budget_critic) }四、生产指标与协同效率跃升在某大型软件自动化重构与修复 Agent 系统的实测中未引入上下文动态分配之前一次包含 15 次工具调用的多 Agent 交互累积产生的全量上下文通常暴涨到 9.5 万 Token。每次批判者介入时由于输入文本过长不仅调用耗时长达 20 秒更是频繁出现“因未看清前置边界条件而放行缺陷代码”的严重漏检。上线上下文预算动态分配器后每个角色接收到的上下文被强行压制在最精炼的 8k~16k 预算内。单轮会话中多智能体协作的总 Token 消耗降低了 68%端到端任务执行时间缩短了 54%。更核心的收益在于决策确定性批判者对代码缺陷的捕获准确率由原先的 67% 直接拉升至 93.5%。多智能体架构的精髓从来不是制造一个相互冗余的信息回音壁而是通过冷峻的上下文契约让每个专长节点在最清晰的视野下完成最锋利的一击。