上下文预算动态分配器：多 Agent 协作系统中的 Token 经济学架构-酒店常州论坛

上下文预算动态分配器：多 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 = ( f"<verification_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%。

多智能体架构的精髓，从来不是制造一个相互冗余的信息回音壁，而是通过冷峻的上下文契约，让每个专长节点在最清晰的视野下完成最锋利的一击。

企业官网建设流程全解析