你的 Agent 系统在收"上下文税"

有一个很少被提及的成本，正在悄悄吃掉多 Agent 系统的预算。

不是模型推理本身的费用，而是重复注入的上下文。

你有一个任务，派发给 8 个并发子 Agent。每个 Agent 的 prompt 里都带着完整的任务背景、用户需求、项目约束——5000 tokens 的信息。这意味着你为同一份信息付了 8 次钱，40000 tokens 的"上下文税"，对任务执行零贡献。

这还只是钱的问题。真正麻烦的是：当 Agent 数量和任务复杂度上升，每个 Agent 的上下文窗口开始被无关信息占满，模型的注意力被稀释，质量下降——但你不一定能直接看出来是"上下文污染"导致的，只会觉得"这次输出怎么有点奇怪"。

更反常识的是：给 Agent 更多信息，不一定让它更聪明，有时候反而会让它更糊涂。

这篇文章聊的，就是怎么从工程角度真正解决这个问题——不是"控制 token 消耗"的表面优化，而是重新思考多 Agent 系统里上下文的所有权和传递方式。

三种让系统悄悄变坏的模式

先把问题说清楚。多 Agent 系统里，上下文管理失控有三种典型模式，每一种都不是明显的 bug，但会让系统慢慢变烂：

模式一：广播式注入

Orchestrator 向每个子 Agent 派发任务时，把全局背景整个复制一份丢进去。子 Agent 越多，同样的信息被注入越多次。最终系统的 token 消耗里，真正有用的推理 token 可能只占 20%，其余 80% 是重复注入的背景信息。

这种模式在系统规模小时完全不痛，是个很舒适的陷阱。

模式二：状态漂移

Agent A 执行完某步骤后把结果写到某个地方，但 Agent B 拿到的还是旧版本的上下文，继续基于过期状态推理和执行。这种 bug 在同步系统里会立刻报错，在 Agent 系统里却会产生"合理"但错误的输出——Agent B 的推理逻辑完全没问题，只是基于了错误的前提。

发现这种问题的唯一方式，通常是最终结果出了问题之后的人工复查。

模式三：上下文焦虑

Anthropic 在最新的多 Agent 研究报告里专门命名了这个现象（Context Anxiety）：当模型的上下文窗口接近上限，它会察觉到"快满了"，开始提前收尾、草草了事，产出质量明显下降。

在多 Agent 系统里，这个问题会比单 Agent 场景更早触发——每个子 Agent 都在携带大量不必要的全局背景，窗口填满速度更快。

📌 ⚡ 一个反直觉的结论：Anthropic 的研究发现，在长时程任务中，完全重置上下文（清空窗口，重新注入结构化检查点）比增量压缩效果更好。这打破了"上下文越完整越好"的直觉——压缩后的历史摘要本身携带"主观解读噪声"，有时反而干扰新决策。

解决方案不是"压缩"，是重新设计所有权

很多团队遇到上下文问题的第一反应是：“加一个压缩步骤”。这思路方向没错，但治标不治本。更根本的问题是：上下文没有明确的所有权模型。

在单进程程序里，状态管理有清晰的作用域——函数局部、类成员、全局变量，各司其职，编译器帮你约束边界。多 Agent 系统天然是分布式的，没有这层约束，每个 Agent 有自己的"视角"，上下文该归谁管、该怎么传、过期了谁来清理，这些问题如果在设计阶段没想清楚，实现时一定会乱。

我倾向于把多 Agent 系统的上下文分成四层，并给每一层明确的所有权和访问规则：

层级	内容	所有者	访问规则
L0：任务元数据	任务ID、全局目标、约束条件	系统	只读，按需引用（不复制进 prompt）
L1：共享知识	用户需求、领域背景	Orchestrator	只读，向量检索按需拉取
L2：执行状态	子任务完成情况、中间产物	状态存储	带版本号的可写 KV
L3：私有推理	Agent 内部思考过程	Agent 本身	私有，不暴露给外部

关键原则只有一条：L0 和 L1 永远不要复制到 prompt 里，只传引用，子 Agent 自己按需拉取。

把上下文访问建模成工具调用

这是整套方案里最关键的一个设计决策，也是最反习惯的一个。

传统做法是：Orchestrator 组装好 prompt，把一切背景塞进去，然后发给子 Agent 执行。

更好的做法是：Orchestrator 只给子 Agent 传递任务描述，背景知识的访问被建模成工具——子 Agent 需要了解某个信息时，调用工具去拿。

# ❌ 传统做法：Orchestrator 提前注入所有背景
def bad_dispatch(subtask, full_background):
    prompt = f"""
背景：{full_background}   # 5000 tokens，很多可能用不到
任务：{subtask}
"""
    return agent.run(prompt)

# ✅ 正确做法：上下文访问工具化，Agent 按需拉取
def good_dispatch(subtask, context_ref):
    prompt = f"""
任务：{subtask}
任务引用ID：{context_ref.task_id}

如需了解项目背景，调用 get_background(query="你的问题")
如需查看完整需求文档，调用 get_requirement()
如需读取其他 Agent 的执行结果，调用 get_state(key="xxx")
"""
    return agent.run(prompt, tools=[context_ref.as_tools()])

这个设计有两个具体好处：

• 天然过滤：Agent 只会拉取它真正需要的信息。一个专注于"生成数据库 schema"的 Agent，不会去拉取 UI 设计规范，因为它根本不会想着去调这个工具。

• 可观测性：每次工具调用都是可记录的，你能清楚看到每个子 Agent 实际访问了哪些上下文，这对 debug 极其有价值。

实现这套工具层，代码量不大：

class TaskContextTools:
    def __init__(self, task_id: str, vector_db, kv_store):
        self.task_id = task_id
        self._vec = vector_db
        self._kv = kv_store
    
    def get_background(self, query: str, top_k: int = 3) -> str:
        """向量检索：只返回与当前问题相关的背景片段"""
        chunks = self._vec.search(
            query, 
            namespace=self.task_id, 
            top_k=top_k
        )
        return "\n---\n".join(c.text for c in chunks)
    
    def get_state(self, key: str) -> tuple[any, int]:
        """读取执行状态，同时返回版本号（写回时用）"""
        entry = self._kv.get(f"{self.task_id}:{key}")
        if not entry:
            return None, 0
        return entry["value"], entry["version"]
    
    def write_state(self, key: str, value, expected_version: int) -> bool:
        """乐观锁写入：expected_version 不匹配则拒绝，返回 False"""
        full_key = f"{self.task_id}:{key}"
        current = self._kv.get(full_key)
        current_ver = current["version"] if current else 0
        
        if current_ver != expected_version:
            return False  # 并发冲突，Agent 需要重读再决策
        
        self._kv.set(full_key, {
            "value": value,
            "version": expected_version + 1,
            "writer": self.caller_agent_id,
            "ts": time.time()
        })
        return True
    
    def as_tools(self) -> list:
        """转换为 LLM 工具调用格式"""
        return [
            tool(self.get_background, description="查询与问题相关的项目背景"),
            tool(self.get_state, description="读取某个执行状态字段"),
            tool(self.write_state, description="写回执行结果（需要提供读取时的版本号）"),
        ]

状态一致性：乐观锁才是正解

L2 层（执行状态）是多 Agent 系统里最容易出 bug 的地方。并发的子 Agent 可能同时读取和写入状态，如果没有保护机制，结果是不可预测的。

很多团队的第一反应是加互斥锁。这不是好选择——Agent 任务执行时间不稳定，持有锁的 Agent 如果执行超时或失败，会导致其他 Agent 无限等待。

乐观锁是更合适的选择，原因是多 Agent 任务中真正产生写冲突的概率其实不高——大多数情况下是任务设计层面的问题（两个 Agent 不应该写同一个状态），而不是并发竞争问题。乐观锁的代价是偶尔的重读重试，收益是不会产生死锁。

上面代码里已经实现了 write_state 的乐观锁逻辑，关键是 expected_version 参数——子 Agent 读取状态时记录版本号，写回时携带它。版本不匹配说明被并发修改，Agent 需要重读最新状态再决策。

对于少数真正不能并发的操作（比如"认领子任务"），用 Redis SET NX 做原子锁即可：

# 原子认领：防止同一子任务被两个 Agent 同时认领
def claim_subtask(redis_client, agent_id: str, subtask_id: str) -> bool:
    lock_key = f"claim:{subtask_id}"
    # SET NX EX：原子操作，5 分钟超时（防止 Agent 失败后永久占用）
    return bool(redis_client.set(lock_key, agent_id, nx=True, ex=300))

Orchestrator 应该是信息过滤器，不是广播站

这里有个认知上的反转需要明确说清楚：Orchestrator 的职责不是"给子 Agent 提供尽可能全的信息"，而是"给每个子 Agent 提供且只提供它需要的信息"。

过于"慷慨"的 Orchestrator 是很多系统上下文问题的根源。

class FilteringOrchestrator:
    
    # 每种角色只能看到它需要的状态字段
    ROLE_STATE_ACCESS = {
        "schema_designer": ["requirements.data_model", "constraints.db_type"],
        "api_developer":   ["schema.tables", "requirements.api_spec"],
        "ui_developer":    ["requirements.ui_spec", "api.endpoints"],
        "qa_agent":        ["*"],  # QA 需要全量视角
    }
    
    def dispatch(self, role: str, subtask: str) -> SubtaskContract:
        allowed_fields = self.ROLE_STATE_ACCESS.get(role, [])
        initial_state = self._filter_state(
            self.state_store.snapshot(), 
            allowed_fields
        )
        
        return SubtaskContract(
            subtask_id=uuid4().hex,
            description=subtask,
            initial_state=initial_state,     # 只包含该角色需要的字段
            expected_outputs=self._infer_outputs(role),
            acceptance_criteria=self._build_criteria(role, subtask),
            max_tokens=8000                    # token 预算约束
        )
    
    def aggregate_results(self, results: list) -> dict:
        """聚合时处理冲突，而不是静默覆盖"""
        merged = {}
        conflicts = []
        for r in results:
            for k, v in r.writes.items():
                if k in merged and merged[k] != v:
                    conflicts.append((k, merged[k], v, r.agent_id))
                merged[k] = v
        if conflicts:
            # 冲突由 Orchestrator 显式仲裁，而不是让后写者直接覆盖
            merged = self._arbitrate(merged, conflicts)
        return merged

这里引入了 SubtaskContract（冲刺合同）的概念，借鉴自 Anthropic 的多 Agent 研究。本质是在子任务开始前双方对"起点"和"终点"达成一致，防止 Generator 和 Evaluator 之间因信息不对称产生的状态漂移。

压缩 vs 重置：什么时候该清空窗口

Anthropic 的研究报告里有个让我印象深刻的结论：在许多长时程任务里，完全重置上下文窗口比增量压缩效果更好。

这听起来很反直觉——历史信息不是越多越好吗？

答案是：取决于任务类型。

• 探索型任务（Agent 在不断试错、调整方向）：保留历史推理有价值，用压缩。模型需要知道"之前试过什么路不通"。

• 执行型任务（有明确的步骤清单，逐项完成）：历史推理是噪声，用重置。模型只需要知道"已完成了什么，接下来做什么"。

class ContextManager:
    
    RESET_THRESHOLD = 0.8   # 上下文占用超过 80% 时触发
    COMPACT_THRESHOLD = 0.7  # 上下文占用超过 70% 时触发压缩
    
    async def manage(self, messages: list, task_type: str, 
                    state_store, max_tokens: int) -> list:
        usage = self._token_count(messages) / max_tokens
        
        if usage  list:
        """检查点必须是结构化 JSON，不是自由文本摘要"""
        # 自由文本摘要 = 携带模型主观解读的噪声
        # 结构化 JSON = 客观事实，干净
        checkpoint_msg = {
            "role": "user",
            "content": f"[上下文重置] 接手正在进行的任务。\n已完成：{json.dumps(checkpoint['done'])}\n当前状态：{json.dumps(checkpoint['state'])}\n剩余工作：{checkpoint['remaining']}"
        }
        return system_msgs + [checkpoint_msg]

这里有个细节值得强调：重置时的检查点必须是结构化数据，而不是自由文本摘要。文本摘要是模型对历史的"主观理解"，会携带偏差；结构化 JSON 是客观事实，干净。这是 Anthropic 研究中重置效果优于压缩的根本原因。

框架对比：不要高估高层封装

说了这么多原则，对应到具体框架时的情况如何？

框架	上下文引用传递	版本化状态管理	上下文压缩/重置	适合场景
LangGraph	✅ State 天然支持	✅ Checkpointer 内置	⚠️ 需自定义节点	生产级复杂流程
CrewAI	❌ 默认全量传递	⚠️ 基础 Memory	❌ 基本没有	快速原型
AutoGen	⚠️ GroupChat 共享	⚠️ 依赖对话历史	⚠️ 需手动实现	对话式协作
自定义实现	✅ 完全控制	✅ 完全控制	✅ 完全控制	复杂生产场景

直说我的判断：CrewAI 的上下文管理是不够的，在任务规模一大就会出问题。它的默认行为接近于"广播式注入"，对 token 成本不敏感的原型项目可以用，但别指望它能在生产环境里扛住压力。

LangGraph 的 State Graph 设计是目前框架里上下文管理最工程化的，但它的学习曲线也陡。如果你的系统需要精细的状态管理，LangGraph 或者自己实现一套，都比 CrewAI 或者 OpenAI Swarm 更值得投入。

📌 一个务实的建议：如果你现在在用某个框架，先用真实任务测量"实际使用的上下文 tokens / 传入的上下文 tokens"这个比率。如果低于 30%，说明你正在为 70% 的背景信息付钱但没有实际使用，这是值得优化的信号。

下一个没人认真想过的问题

上面这些模式，基本解决了多 Agent 系统里上下文工程的"已知问题"。但还有一个更深层的问题，目前几乎没有人给出过令人满意的答案：

上下文的信任模型。

在多 Agent 系统里，状态存储是各个 Agent 共享的"世界"。一个出错的子 Agent（或者被 prompt injection 攻击的 Agent）可能往状态存储里写入错误信息——而这个错误信息会被后续所有 Agent 读取，当作事实来推理和执行。

这是一种"状态污染"，比单 Agent 场景下的 prompt injection 更危险，因为它会在系统内部传播。

目前的工程实践基本靠两个手段缓解：一是限制子 Agent 的写权限（只能写自己负责的字段），二是引入 Evaluator Agent 在关键路径上验证写入的合理性。但这两个手段都是事后补救，没有一个系统性的解法。

这是我觉得接下来值得深入研究的方向——不是"怎么让 Agent 更聪明"，而是"怎么让 Agent 系统在局部失效时仍然保持全局可信"。这是分布式系统领域的老问题，但在 Agent 场景下有它特有的复杂性：失效的不是网络节点，而是推理过程本身。

在做多 Agent 系统？欢迎留言交流你们遇到的上下文问题 👇