Agent 上下文窗口管理长任务不能靠无限堆 token一、Agent 跑到一半说我不知道前面发生了什么上下文溢出的真实代价你构建了一个代码审查 Agent。它依次读取三个模块的代码分析依赖关系然后给出重构建议。在第一个模块上它表现完美到第三个模块时它开始遗漏明显的命名冲突——因为前两个模块的代码早已被挤出了上下文窗口的注意力范围。更严重的是静默模式下的错误。Agent 没有报上下文溢出它只是生成了一份看起来连贯但事实错误的审查报告。等你发现的时候它已经影响了两次代码合并。这就是 Agent 长任务场景的核心困境上下文窗口是有限的物理资源但 Agent 的任务边界没有自然限制。二、上下文窗口的注意力衰减曲线与信息丢失机制LLM 的上下文窗口并非均匀利用。大量实验表明模型对窗口两端的注意力显著高于中间部分。Lost in the Middle现象会随着任务链长度加剧。flowchart TB subgraph Phase1[阶段一窗口充足 50% 容量] A[Token 1-1000br/系统 Prompt] -- C[完整推理] B[Token 1001-3000br/任务上下文] -- C C -- D[输出质量正常] end subgraph Phase2[阶段二窗口临界50%-80% 容量] E[新任务输入] -- F[旧信息被挤出br/注意力核心区] F -- G[中间段记忆衰减] G -- H[输出出现遗漏] end subgraph Phase3[阶段三窗口溢出 80% 容量] I[继续追加输入] -- J[幻觉概率上升] J -- K[静默错误模式] K -- L[需重启或压缩] end Phase1 -- Phase2 -- Phase3 style Phase3 fill:#ffcdd2 style D fill:#c8e6c9 style L fill:#ffcdd2上下文管理需要处理三种信息衰减位置衰减窗口中间位置的信息编码后激活值分布更扁平模型更难定位。语义衰减语义上不直接与当前查询相关的信息即使位置靠前也会被注意力机制淡化。时间衰减在多轮交互中越早轮次的信息权重越低——这与 transformer 的无状态设计有关。三、可配置的上下文窗口管理器实现以下是上下文管理器的核心实现支持三种策略滑动窗口、摘要压缩、优先级分层。import tiktoken from typing import List, Dict, Optional, Callable from dataclasses import dataclass, field from enum import Enum class RetentionStrategy(Enum): 内容保留策略枚举 SLIDING_WINDOW sliding # 保留最近 N token SUMMARIZE summarize # 压缩旧内容为摘要 PRIORITY priority # 按优先级逐出低优内容 dataclass class ContextChunk: 上下文分块——每个上下文来源被抽象为一个 Chunk 设计动机统一管理 system prompt、工具调用结果、历史对话等 不同类型的内容源在逐出时按统一优先级规则排队 content: str priority: int 0 # 0系统级 1任务级 2结果级 3临时级 source: str unknown token_count: int 0 # 缓存 token 计数避免重复计算 def __post_init__(self): # 自动计算 token 数量在构造时一次性完成 # 后续排序和逐出决策都依赖此值 if self.token_count 0 and self.content: # 使用 cl100k_base 编码器GPT-4/ChatGPT 默认 encoding tiktoken.get_encoding(cl100k_base) self.token_count len(encoding.encode(self.content)) class ContextWindowManager: 上下文窗口管理器 核心设计基于优先级的逐出策略。 系统 Promptpriority0除非极端溢出否则不会被逐出 中间计算结果priority2优先丢弃。 这确保了 Agent 在多轮任务中的信息完整性 def __init__( self, max_tokens: int 4096, reserve_ratio: float 0.2, # 预留 20% 给模型输出 strategy: RetentionStrategy RetentionStrategy.PRIORITY, summarizer: Optional[Callable] None # 外部摘要函数 ): self.max_tokens max_tokens self.effective_limit int(max_tokens * (1 - reserve_ratio)) self.strategy strategy self.summarizer summarizer self.chunks: List[ContextChunk] [] property def total_tokens(self) - int: 当前上下文窗口总 token 消耗 return sum(chunk.token_count for chunk in self.chunks) def add(self, content: str, priority: int 0, source: str ) - None: 添加内容到上下文窗口 为什么添加后立即做逐出而非等到调用前 因为在 Agent 循环中每次工具调用后都需要确保上下文合规 等到最终 API 调用时再做逐出会导致单次延迟不可预测 chunk ContextChunk(contentcontent, prioritypriority, sourcesource) self.chunks.append(chunk) self._maybe_evict() def _maybe_evict(self): 按策略逐出超限内容 while self.total_tokens self.effective_limit: if self.strategy RetentionStrategy.SLIDING_WINDOW: self._evict_oldest() elif self.strategy RetentionStrategy.SUMMARIZE: self._evict_with_summary() elif self.strategy RetentionStrategy.PRIORITY: self._evict_low_priority() else: self._evict_oldest() def _evict_oldest(self): 移除最早的非系统级内容 # 保护系统级内容不被逐出 for i, chunk in enumerate(self.chunks): if chunk.priority 0: self.chunks.pop(i) return # 紧急情况全是系统级内容逐出最老的 self.chunks.pop(0) def _evict_low_priority(self): 按优先级逐出高 priority 值低优先先被丢弃 逐出逻辑按 priority 降序排列淘汰最高值最不重要的内容。 相同优先级取最早的。这样在 token 预算紧张时 Agent 的中间计算步骤会先被压缩而系统指令和关键结果会被保留 self.chunks.sort(keylambda c: (-c.priority, self.chunks.index(c))) self.chunks.pop() # 移除优先级最低的 def _evict_with_summary(self): 将最老的非系统内容替换为摘要 if not self.summarizer: self._evict_oldest() return for i, chunk in enumerate(self.chunks): if chunk.priority 0: summary self.summarizer(chunk.content) self.chunks[i] ContextChunk( contentf[摘要] {summary}, prioritychunk.priority 1, # 摘要降低优先级 sourcechunk.source ) return def get_context(self) - str: 获取当前的完整上下文字符串 # 最终打包前再检查一次防止并发追加导致超限 self._maybe_evict() return \n---\n.join(chunk.content for chunk in self.chunks) def get_stats(self) - Dict: 获取窗口统计信息用于监控和告警 return { total_tokens: self.total_tokens, limit: self.effective_limit, usage_ratio: round(self.total_tokens / self.effective_limit, 3), chunk_count: len(self.chunks), priority_distribution: { p: sum(1 for c in self.chunks if c.priority p) for p in sorted(set(c.priority for c in self.chunks)) } } # # 使用示例 # if __name__ __main__: manager ContextWindowManager(max_tokens4096, reserve_ratio0.2) # 系统 Promptpriority0几乎不会被逐出 manager.add(你是一个代码审查助手请分析以下代码的质量问题。, priority0, sourcesystem) # 任务上下文priority1 manager.add(审查目标auth_module.py关注安全漏洞和性能问题。, priority1, sourcetask) # 模拟 Agent 逐步加载多个文件 for i in range(1, 6): manager.add( f# auth_module.py 内容第{i}段\n def authenticate(user, password):...\n * 100, priority2, sourceffile_part_{i} ) stats manager.get_stats() print(f加载第{i}段后: 使用率 {stats[usage_ratio]}) ## 四、上下文管理策略的边界场景与架构权衡 **摘要压缩的精度损失**。 当使用 summarizer 压缩旧内容时压缩必然带来信息损失。如果 Agent 的任务依赖已被压缩内容中的精确数值或代码片段摘要化的上下文可能产生看似完整实则残缺的信息环境。建议对摘要内容标注 [摘要] 前缀让模型意识到这是经过压缩的信息而非原始内容降低过度依赖摘要结论的风险。 **逐出策略的止损点**。 三种逐出策略各有盲区滑动窗口可能丢弃关键早期信息如任务目标摘要策略可能丢失精确细节优先级策略需要正确的优先级赋值——而优先级的定义本身就是另一个系统设计问题。见证奇迹的时刻是当你的 Agent 在长任务中保持 95% 的正确率到第 20 轮而它背后是上下文管理器在大约第 14 轮做了一次关键的保留高优先级、逐出低优先级决策。 **窗口使用率监控**。 建议为上下文管理器接入 get_stats() 的定时上报。当使用率持续在 90% 以上时说明当前任务可能需要拆分为多个子任务或接入外部分段存储。统计告警不应该在溢出后才触发——应在使用率达到 75% 时发出预警。 ## 五、总结 Agent 上下文窗口管理的核心不是怎么把更多东西塞进去而是怎么在有限空间里做最优的信息取舍。三个实践准则 1. **优先级分级是基础**系统 Prompt 任务定义 关键结果 中间过程逐出按反向顺序。 2. **压缩策略必须标注**摘要化后的信息要带 [摘要] 标记防止模型将其视为原始精确信息。 3. **监控先行于溢出**在 75% 使用率时预警在 90% 时启动更激进的逐出策略。