1. 项目概述当长程智能体遭遇“上下文之困”最近在折腾LLM驱动的自主智能体LLM-powered Autonomous Agents时一个绕不开的瓶颈越来越清晰地摆在面前上下文窗口。无论是构建一个能处理复杂工作流的自动化助手还是一个需要长期记忆和规划的探索型智能体我们都在追求更长的任务视野Long-Horizon。然而随着对话轮次和工具调用次数的增加上下文长度会像滚雪球一样膨胀最终触及模型的上限。传统的解决方案简单粗暴——压缩Compaction或直接截断Truncation。但这无异于给智能体做了“脑前额叶切除手术”丢失的关键信息可能导致后续决策完全跑偏陷入逻辑混乱或重复循环。这就引出了我们这次要深入探讨的核心“超越压缩的结构化上下文驱逐”Structured Context Eviction。这不仅仅是一个技术策略更是一种设计哲学。它的目标不是简单地腾出空间而是像一位经验丰富的图书管理员在有限的馆藏空间Token预算内决定哪些书籍记忆片段应该保留在触手可及的开架区活跃上下文哪些可以归档到密集书库外部记忆而哪些则可以安全地剔除。Lilian Weng等研究者对智能体系统的思考也指向了记忆与上下文管理的重要性。本文将从一个实践者的角度拆解这一策略背后的逻辑、实现的关键技术点并分享一套可落地的实操方案。2. 核心思路从“无差别丢弃”到“有策略保留”为什么传统的压缩或截断不够用因为它们是“无差别”的。假设智能体正在执行一个“策划并执行一场线上会议”的长程任务上下文里混杂了最初的用户指令、搜索到的参会者时间信息、生成的会议议程草稿、与日历API交互的日志、以及中途用户提出的细节修改。如果单纯从尾部截断可能把用户最新的修改要求给丢了如果进行通用文本压缩可能会模糊化那些对后续工具调用至关重要的结构化参数。结构化上下文驱逐的核心思路在于引入一个评估框架对上下文中的每一个信息片段进行价值评估和分类然后基于当前任务状态执行有策略的保留、转移或移除。这背后是三个关键转变从“文本流”到“信息单元”不再将上下文视为一个连续的文本字符串而是将其解析为结构化的信息单元Information Units。这些单元可以是一条用户消息、一个工具调用及其结果、一段系统提示、或智能体自身的推理过程。从“静态预算”到“动态配额”Token预算的分配不再是固定的。根据任务阶段我们可以为“核心指令”、“近期工具结果”、“历史摘要”等不同类别的信息单元设置动态配额。在规划阶段可能保留更多历史决策逻辑在执行阶段则优先保留最新的工具I/O。从“被动清理”到“主动管理”驱逐动作不是等到令牌超限时才触发而是作为一个持续的后台进程。每当新的信息单元产生时管理系统就会评估其重要性并可能触发对旧单元的降级或移除。2.1 信息单元的价值评估维度要实现有策略的保留首先得会“估价”。我们可以从以下几个维度对每个信息单元进行打分任务相关性该信息与当前最高层级任务目标的直接关联程度。例如在会议策划任务中“会议主题”的相关性始终很高而某次失败的“发送测试邮件”的详细错误日志在问题解决后相关性会急剧下降。时效性信息的新旧程度。通常越新的信息越可能影响下一步操作。但要注意一些早期设定的“约束条件”如“预算不超过1000元”虽旧却至关重要。信息密度该单元所承载的不可替代信息的浓度。一段冗长的、重复性的API响应可以被摘要替代低密度而一个包含具体日期、时间、链接的会议邀请结果则是高密度的。结构性依赖该信息是否是其他信息单元理解或产生的前提。例如一个工具调用的结果严重依赖于之前那条工具调用请求的参数。驱逐了请求结果就变得无法理解。在实际评分时我会采用一个加权公式例如Score w1 * Relevance w2 * Recency w3 * Density。权重w1, w2, w3可以根据智能体的类型进行调整。一个偏重规划的智能体Relevance权重更高一个偏重实时交互的智能体Recency权重更高。注意评估模型本身不宜过于复杂否则其计算开销会抵消上下文管理带来的收益。初期可以从基于规则的简单评分开始例如标记所有“用户输入”和“关键工具结果”为高优先级。3. 系统架构与核心组件设计一个完整的结构化上下文驱逐系统可以看作是在智能体主循环旁增加的一个“记忆管理”子系统。其核心组件与数据流如下图所示概念描述[新信息单元产生] ↓ [解析与标注组件] - 将原始文本解析为单元并打上类型、时间戳、依赖关系等元数据标签。 ↓ [价值评估组件] - 根据预设规则或轻量模型计算该单元的当前价值分数。 ↓ [上下文状态监控器] - 持续计算当前总令牌数及各分类配额使用情况。 ↓ [驱逐策略执行器] - [策略仓库] (包含多种驱逐算法) ↓ [动作执行] - 1. 保留至活跃上下文。 2. 摘要后保留摘要。 3. 转移至外部向量数据库。 4. 直接移除。 ↓ [更新后的活跃上下文] - 送入LLM进行下一轮推理。3.1 解析与标注组件这是所有工作的基础。我们需要从LLM的对话历史中准确地切分出信息单元。对于遵循标准框架如LangChain的AgentExecutor、AutoGPT的架构的智能体这相对容易因为工具调用、结果、思考过程通常有明确的格式如Thought:,Action:,Observation:。关键是要设计一套统一的元数据schemaclass InformationUnit: def __init__(self, content: str, unit_type: str, timestamp: float, parent_id: Optional[str] None): self.id str(uuid.uuid4()) # 唯一标识 self.content content # 原始文本内容 self.type unit_type # 如user_input, agent_thought, tool_call, tool_result, system self.timestamp timestamp # 创建时间戳 self.parent_id parent_id # 指向依赖的父单元ID如tool_result指向tool_call self.token_count len(encode(content)) # 令牌占用数 self.current_score 0.0 # 当前价值评分 self.metadata {} # 其他扩展信息如工具名称、参数等3.2 策略仓库几种实用的驱逐算法策略仓库里存放着不同的“驱逐算法”可以根据场景切换使用。最低价值优先Lowest-Score-First这是最直接的策略。当需要腾出空间时持续移除当前评分最低的单元直到满足预算要求。风险在于可能移除一个当前分数低但未来至关重要的单元例如一个早期设定的关键约束。时间片滑动窗口Time-Slice Sliding Window保留最近N个时间单位内产生的所有信息单元。这种方法简单能保证信息新鲜度但可能无法保留关键的早期信息。分层配额管理Hierarchical Quota Management这是我更推荐的方式。它将上下文划分为几个逻辑层并为每层分配令牌配额核心层存放最初始的任务描述、核心约束和全局目标。除非任务变更否则永不驱逐。工作层存放当前子任务相关的思考、工具调用和结果。配额最大是活跃工作区。摘要层存放对已完成子任务或低优先级详细信息的摘要。当工作层单元“老化”或价值降低时将其内容用LLM生成一个简短摘要移入摘要层释放原始文本占用的空间。当总令牌数接近上限时优先在摘要层进行压缩进一步缩短摘要或在工作层驱逐低分单元。核心层动不得。3.3 动作执行不只是删除驱逐不等于删除。我们有几个动作选项保留单元留在活跃上下文不做任何处理。压缩使用LLM生成该单元的简短摘要。这里有一个技巧摘要的提示词Prompt需要精心设计以保留对后续任务最关键的信息。例如对于工具调用结果摘要应聚焦于“成功/失败”状态和关键输出数据忽略冗长的日志文本。外化将整个单元或它的一个丰富表示存入一个外部的、可检索的记忆存储如向量数据库。同时在活跃上下文中插入一个“占位符”如[关于X会议的参会者时间信息已存储ID: mem_123]。当智能体后续可能需要这些信息时可以通过检索Retrieval将其重新引入上下文。移除当确认信息完全无关或已被充分替代时直接删除。4. 实操实现基于现有框架的改造我们以构建一个基于LangChain的智能体为例演示如何集成一个简单的结构化上下文管理模块。这里不追求全自动的完美系统而是先实现一个可工作、可观察的原型。4.1 步骤一封装自定义的上下文管理器首先我们创建一个管理器它包裹了LangChain的对话历史ChatMessageHistory并增加了我们的管理逻辑。import uuid from typing import List, Dict, Any, Optional from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage from langchain.memory import ChatMessageHistory from some_tokenizer import encode # 你需要一个分词器例如tiktoken class StructuredContextManager: def __init__(self, token_limit: int 8000): self.token_limit token_limit self.active_context: List[InformationUnit] [] # 活跃信息单元列表 self.external_memory [] # 简化的外部记忆存储实际可用向量数据库 self.core_units: List[str] [] # 核心单元ID列表 def add_message(self, message: BaseMessage): 将LangChain消息转换为信息单元并添加 unit_type self._classify_message(message) unit InformationUnit( contentmessage.content, unit_typeunit_type, timestamptime.time(), parent_idself._get_parent_id() # 需要根据逻辑实现 ) unit.token_count self._count_tokens(unit.content) self.active_context.append(unit) self._evaluate_and_evict(unit) # 添加后触发评估和驱逐 def _classify_message(self, message: BaseMessage) - str: 根据消息类型和内容进一步分类 if isinstance(message, SystemMessage): return system elif isinstance(message, HumanMessage): # 可以进一步解析内容判断是否是工具输出 if Observation: in message.content: return tool_result return user_input elif isinstance(message, AIMessage): if Action: in message.content: return tool_call return agent_thought return other def _evaluate_and_evict(self, new_unit: InformationUnit): 评估新单元并执行驱逐策略以维持令牌预算 # 1. 计算当前总令牌数 total_tokens sum(unit.token_count for unit in self.active_context) # 2. 如果未超限仅更新评分 if total_tokens self.token_limit: self._update_scores() return # 3. 超限执行驱逐策略这里实现最低价值优先 self._update_scores() # 按分数升序排序分数低的前面 self.active_context.sort(keylambda x: x.current_score) while total_tokens self.token_limit and len(self.active_context) 0: # 跳过核心单元 if self.active_context[0].id in self.core_units: # 将核心单元移到列表末尾避免被移除 core_unit self.active_context.pop(0) self.active_context.append(core_unit) continue # 移除分数最低的非核心单元 removed_unit self.active_context.pop(0) total_tokens - removed_unit.token_count print(fEvicted unit (Score: {removed_unit.current_score:.2f}): {removed_unit.content[:50]}...) # 可选将移除的单元转移到外部记忆 # self._externalize_unit(removed_unit) # 4. 按时间顺序重新排序以保持上下文连贯性 self.active_context.sort(keylambda x: x.timestamp) def _update_scores(self): 更新所有活跃单元的价值评分简化版规则 now time.time() for unit in self.active_context: recency max(0, 1 - (now - unit.timestamp) / 300) # 5分钟衰减因子 relevance 1.0 if unit.type in [user_input, system] else 0.7 density min(1.0, 100 / unit.token_count) if unit.token_count 0 else 1.0 # 假设信息密度与长度成反比 unit.current_score 0.5 * relevance 0.3 * recency 0.2 * density def get_active_context_text(self) - str: 将活跃上下文转换为LLM可接受的文本格式 context_lines [] for unit in self.active_context: prefix { user_input: User: , agent_thought: Thought: , tool_call: Action: , tool_result: Observation: , system: System: , other: }.get(unit.type, ) context_lines.append(f{prefix}{unit.content}) return \n.join(context_lines)4.2 步骤二与LangChain Agent集成接下来我们需要在自定义的Agent执行循环中用我们的管理器替代默认的记忆处理。from langchain.agents import AgentExecutor, Tool from langchain.llms import OpenAI # 假设你已经定义好了tools和llm tools [...] llm OpenAI(temperature0) # 创建自定义的上下文管理器 context_manager StructuredContextManager(token_limit4000) # 设置一个较小的预算便于测试 # 定义Agent的提示词模板其中包含一个 {context} 占位符 agent_prompt_template 你是一个智能助手。请根据以下上下文信息来回答问题或执行任务。 如果任务需要调用工具请严格按照格式输出。 当前上下文 {context} 现在开始处理。 # 在Agent的执行步骤中 def custom_agent_step(user_input: str): # 1. 将用户输入作为新单元加入管理器 context_manager.add_message(HumanMessage(contentuser_input)) # 2. 从管理器获取格式化后的当前活跃上下文 formatted_context context_manager.get_active_context_text() # 3. 填充提示词并调用LLM prompt agent_prompt_template.format(contextformatted_context) llm_response llm(prompt) # 这里简化了实际应使用完整的Agent推理逻辑 # 4. 解析LLM响应如果是工具调用执行工具... # ... (这里省略具体的工具调用和结果获取逻辑) # 5. 将LLM的“思考”和“工具调用”动作作为AIMessage加入管理器 context_manager.add_message(AIMessage(contentllm_response)) # 6. 如果有工具结果将其作为HumanMessageObservation加入管理器 # tool_result ... # context_manager.add_message(HumanMessage(contentfObservation: {tool_result})) # 返回最终结果 return parse_final_answer(llm_response)4.3 关键参数调优与监控实现基础功能后调优至关重要令牌预算token_limit不要设置为模型的最大上下文窗口如16384要留出安全余量例如80%以防单次生成过长。评分权重relevance,recency,density需要通过实验调整。一个实用的方法是记录智能体在长任务中因信息丢失导致的错误然后反推哪些信息被误驱逐了从而调整权重。例如如果智能体总是忘记用户早期提出的预算限制就需要提高system和早期user_input类型单元的relevance权重。驱逐阈值可以设置一个“软上限”和一个“硬上限”。当令牌数超过软上限如预算的90%时开始执行低强度驱逐如只压缩摘要层超过硬上限如预算的100%时执行高强度驱逐移除工作层低分单元。监控与日志务必详细记录每次驱逐动作驱逐了哪个单元、为什么、其评分如何。这不仅是调试的依据也是优化评估算法的重要数据来源。5. 常见问题与实战避坑指南在实际部署结构化上下文驱逐系统时我踩过不少坑这里分享几个典型的案例和解决方案。5.1 问题一上下文连贯性断裂现象智能体在长对话中突然“失忆”引用了一个已经被驱逐的信息或者提出的问题与之前逻辑衔接不上。根因驱逐策略过于激进或者评估算法未能正确识别信息单元之间的依赖关系。例如驱逐了一个工具调用请求但保留了它的结果导致结果失去意义。解决方案强化依赖链在InformationUnit中显式建模parent_id和children_ids。驱逐一个单元时进行“级联检查”。如果它是一个父单元考虑将其所有子单元一并驱逐或外化如果它是一个孤立的子单元父单元已被驱逐则应提高其优先级或为其生成一个自包含的摘要。引入“上下文锚点”在生成摘要或进行外化时强制在摘要文本或占位符中包含足够的关键词如实体名、任务ID以便未来需要时能通过检索准确找回。设置保护名单对于标识当前子任务目标的信息单元例如“当前步骤预订会议室”即使分数不高也在一段时间内免于驱逐。5.2 问题二摘要导致的信息失真现象为了节省空间将一个详细的工具结果如包含多行数据的JSON压缩成一句摘要如“成功查询到3条会议室信息”。后续步骤需要具体的会议室ID时智能体无法从摘要中获取导致任务失败。根因摘要的提示词Prompt设计不佳丢失了关键的结构化数据。解决方案面向任务的摘要设计不同的摘要模板。对于数据查询结果摘要应保留“实体类型-数量-关键属性”的模板例如“[查询结果] 会议室3间 (ID: A101, B202, C303 容量10人)”。这比自然语言摘要保留了更多可解析的信息。混合存储对于包含关键参数的结果采用“摘要关键数据提取”的方式。将生成的纯文本摘要存入上下文同时将提取出的结构化数据如[A101, B202, C303]以注释或隐藏字段的形式附着在单元上。虽然这些数据可能不直接输入LLM但可以被智能体的逻辑代码访问和使用。延迟压缩不要立即压缩刚产生的结果。让它在“工作层”保留一段时间例如完成与之相关的下一个子任务后再压缩确保其详细信息被充分使用。5.3 问题三评估开销影响性能现象智能体的响应速度变慢尤其是当上下文很长时每次评估所有单元分数耗时明显。根因评估函数过于复杂或者频繁地对整个上下文进行全量重评估。解决方案增量评估新单元加入时只计算新单元的分数。只有当需要执行驱逐时才对候选单元通常是分数可能较低的老旧单元进行重新评估而非全部。简化评分模型初期完全可以使用基于规则的线性评分。只有在规则无法处理复杂情况时才考虑引入微小的预测模型如预测某个信息单元在未来N步内被引用的概率。异步执行将评估和驱逐操作放在一个独立的、低优先级的线程中进行不阻塞主推理循环。当然这需要处理好数据同步的问题。5.4 问题四与外部记忆检索的协同失效现象将信息外化到向量数据库后当智能体需要历史信息时检索回来的内容不相关或无法有效融入当前上下文。根因外化时的“占位符”描述不准确或者检索查询query的生成策略不佳。解决方案优化占位符生成占位符不应只是[信息已存储]而应是一个高度浓缩的“索引”例如[关于项目‘Alpha’的2023年Q3销售数据包含区域拆分和同比已存储]。这个索引本身就可以作为未来检索的查询基础。动态生成检索查询当智能体的推理过程表明需要某类历史信息时例如它生成了“我需要回顾一下之前讨论过的预算限制”用这个意图陈述作为查询去检索比用当前整个对话作为上下文去检索要精准得多。检索结果的重集成将检索回来的信息重新插入上下文时不要简单拼接。最好以“回忆”的形式呈现例如“[根据之前存储的记录关于预算的讨论如下...]”帮助LLM理解这是召回的信息而非当前对话的新内容。实施结构化上下文驱逐不是一个一劳永逸的开关而是一个需要持续观察和调优的子系统。它迫使开发者更深入地思考智能体如何“思考”和“记忆”从而设计出更鲁棒、更高效的智能体系统。对于长程任务Long-Horizon Tasks而言良好的上下文管理能力往往是区分一个玩具原型和一个可用系统的关键所在。 LLM智能体上下文管理：结构化驱逐策略与工程实践
1. 项目概述当长程智能体遭遇“上下文之困”最近在折腾LLM驱动的自主智能体LLM-powered Autonomous Agents时一个绕不开的瓶颈越来越清晰地摆在面前上下文窗口。无论是构建一个能处理复杂工作流的自动化助手还是一个需要长期记忆和规划的探索型智能体我们都在追求更长的任务视野Long-Horizon。然而随着对话轮次和工具调用次数的增加上下文长度会像滚雪球一样膨胀最终触及模型的上限。传统的解决方案简单粗暴——压缩Compaction或直接截断Truncation。但这无异于给智能体做了“脑前额叶切除手术”丢失的关键信息可能导致后续决策完全跑偏陷入逻辑混乱或重复循环。这就引出了我们这次要深入探讨的核心“超越压缩的结构化上下文驱逐”Structured Context Eviction。这不仅仅是一个技术策略更是一种设计哲学。它的目标不是简单地腾出空间而是像一位经验丰富的图书管理员在有限的馆藏空间Token预算内决定哪些书籍记忆片段应该保留在触手可及的开架区活跃上下文哪些可以归档到密集书库外部记忆而哪些则可以安全地剔除。Lilian Weng等研究者对智能体系统的思考也指向了记忆与上下文管理的重要性。本文将从一个实践者的角度拆解这一策略背后的逻辑、实现的关键技术点并分享一套可落地的实操方案。2. 核心思路从“无差别丢弃”到“有策略保留”为什么传统的压缩或截断不够用因为它们是“无差别”的。假设智能体正在执行一个“策划并执行一场线上会议”的长程任务上下文里混杂了最初的用户指令、搜索到的参会者时间信息、生成的会议议程草稿、与日历API交互的日志、以及中途用户提出的细节修改。如果单纯从尾部截断可能把用户最新的修改要求给丢了如果进行通用文本压缩可能会模糊化那些对后续工具调用至关重要的结构化参数。结构化上下文驱逐的核心思路在于引入一个评估框架对上下文中的每一个信息片段进行价值评估和分类然后基于当前任务状态执行有策略的保留、转移或移除。这背后是三个关键转变从“文本流”到“信息单元”不再将上下文视为一个连续的文本字符串而是将其解析为结构化的信息单元Information Units。这些单元可以是一条用户消息、一个工具调用及其结果、一段系统提示、或智能体自身的推理过程。从“静态预算”到“动态配额”Token预算的分配不再是固定的。根据任务阶段我们可以为“核心指令”、“近期工具结果”、“历史摘要”等不同类别的信息单元设置动态配额。在规划阶段可能保留更多历史决策逻辑在执行阶段则优先保留最新的工具I/O。从“被动清理”到“主动管理”驱逐动作不是等到令牌超限时才触发而是作为一个持续的后台进程。每当新的信息单元产生时管理系统就会评估其重要性并可能触发对旧单元的降级或移除。2.1 信息单元的价值评估维度要实现有策略的保留首先得会“估价”。我们可以从以下几个维度对每个信息单元进行打分任务相关性该信息与当前最高层级任务目标的直接关联程度。例如在会议策划任务中“会议主题”的相关性始终很高而某次失败的“发送测试邮件”的详细错误日志在问题解决后相关性会急剧下降。时效性信息的新旧程度。通常越新的信息越可能影响下一步操作。但要注意一些早期设定的“约束条件”如“预算不超过1000元”虽旧却至关重要。信息密度该单元所承载的不可替代信息的浓度。一段冗长的、重复性的API响应可以被摘要替代低密度而一个包含具体日期、时间、链接的会议邀请结果则是高密度的。结构性依赖该信息是否是其他信息单元理解或产生的前提。例如一个工具调用的结果严重依赖于之前那条工具调用请求的参数。驱逐了请求结果就变得无法理解。在实际评分时我会采用一个加权公式例如Score w1 * Relevance w2 * Recency w3 * Density。权重w1, w2, w3可以根据智能体的类型进行调整。一个偏重规划的智能体Relevance权重更高一个偏重实时交互的智能体Recency权重更高。注意评估模型本身不宜过于复杂否则其计算开销会抵消上下文管理带来的收益。初期可以从基于规则的简单评分开始例如标记所有“用户输入”和“关键工具结果”为高优先级。3. 系统架构与核心组件设计一个完整的结构化上下文驱逐系统可以看作是在智能体主循环旁增加的一个“记忆管理”子系统。其核心组件与数据流如下图所示概念描述[新信息单元产生] ↓ [解析与标注组件] - 将原始文本解析为单元并打上类型、时间戳、依赖关系等元数据标签。 ↓ [价值评估组件] - 根据预设规则或轻量模型计算该单元的当前价值分数。 ↓ [上下文状态监控器] - 持续计算当前总令牌数及各分类配额使用情况。 ↓ [驱逐策略执行器] - [策略仓库] (包含多种驱逐算法) ↓ [动作执行] - 1. 保留至活跃上下文。 2. 摘要后保留摘要。 3. 转移至外部向量数据库。 4. 直接移除。 ↓ [更新后的活跃上下文] - 送入LLM进行下一轮推理。3.1 解析与标注组件这是所有工作的基础。我们需要从LLM的对话历史中准确地切分出信息单元。对于遵循标准框架如LangChain的AgentExecutor、AutoGPT的架构的智能体这相对容易因为工具调用、结果、思考过程通常有明确的格式如Thought:,Action:,Observation:。关键是要设计一套统一的元数据schemaclass InformationUnit: def __init__(self, content: str, unit_type: str, timestamp: float, parent_id: Optional[str] None): self.id str(uuid.uuid4()) # 唯一标识 self.content content # 原始文本内容 self.type unit_type # 如user_input, agent_thought, tool_call, tool_result, system self.timestamp timestamp # 创建时间戳 self.parent_id parent_id # 指向依赖的父单元ID如tool_result指向tool_call self.token_count len(encode(content)) # 令牌占用数 self.current_score 0.0 # 当前价值评分 self.metadata {} # 其他扩展信息如工具名称、参数等3.2 策略仓库几种实用的驱逐算法策略仓库里存放着不同的“驱逐算法”可以根据场景切换使用。最低价值优先Lowest-Score-First这是最直接的策略。当需要腾出空间时持续移除当前评分最低的单元直到满足预算要求。风险在于可能移除一个当前分数低但未来至关重要的单元例如一个早期设定的关键约束。时间片滑动窗口Time-Slice Sliding Window保留最近N个时间单位内产生的所有信息单元。这种方法简单能保证信息新鲜度但可能无法保留关键的早期信息。分层配额管理Hierarchical Quota Management这是我更推荐的方式。它将上下文划分为几个逻辑层并为每层分配令牌配额核心层存放最初始的任务描述、核心约束和全局目标。除非任务变更否则永不驱逐。工作层存放当前子任务相关的思考、工具调用和结果。配额最大是活跃工作区。摘要层存放对已完成子任务或低优先级详细信息的摘要。当工作层单元“老化”或价值降低时将其内容用LLM生成一个简短摘要移入摘要层释放原始文本占用的空间。当总令牌数接近上限时优先在摘要层进行压缩进一步缩短摘要或在工作层驱逐低分单元。核心层动不得。3.3 动作执行不只是删除驱逐不等于删除。我们有几个动作选项保留单元留在活跃上下文不做任何处理。压缩使用LLM生成该单元的简短摘要。这里有一个技巧摘要的提示词Prompt需要精心设计以保留对后续任务最关键的信息。例如对于工具调用结果摘要应聚焦于“成功/失败”状态和关键输出数据忽略冗长的日志文本。外化将整个单元或它的一个丰富表示存入一个外部的、可检索的记忆存储如向量数据库。同时在活跃上下文中插入一个“占位符”如[关于X会议的参会者时间信息已存储ID: mem_123]。当智能体后续可能需要这些信息时可以通过检索Retrieval将其重新引入上下文。移除当确认信息完全无关或已被充分替代时直接删除。4. 实操实现基于现有框架的改造我们以构建一个基于LangChain的智能体为例演示如何集成一个简单的结构化上下文管理模块。这里不追求全自动的完美系统而是先实现一个可工作、可观察的原型。4.1 步骤一封装自定义的上下文管理器首先我们创建一个管理器它包裹了LangChain的对话历史ChatMessageHistory并增加了我们的管理逻辑。import uuid from typing import List, Dict, Any, Optional from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage from langchain.memory import ChatMessageHistory from some_tokenizer import encode # 你需要一个分词器例如tiktoken class StructuredContextManager: def __init__(self, token_limit: int 8000): self.token_limit token_limit self.active_context: List[InformationUnit] [] # 活跃信息单元列表 self.external_memory [] # 简化的外部记忆存储实际可用向量数据库 self.core_units: List[str] [] # 核心单元ID列表 def add_message(self, message: BaseMessage): 将LangChain消息转换为信息单元并添加 unit_type self._classify_message(message) unit InformationUnit( contentmessage.content, unit_typeunit_type, timestamptime.time(), parent_idself._get_parent_id() # 需要根据逻辑实现 ) unit.token_count self._count_tokens(unit.content) self.active_context.append(unit) self._evaluate_and_evict(unit) # 添加后触发评估和驱逐 def _classify_message(self, message: BaseMessage) - str: 根据消息类型和内容进一步分类 if isinstance(message, SystemMessage): return system elif isinstance(message, HumanMessage): # 可以进一步解析内容判断是否是工具输出 if Observation: in message.content: return tool_result return user_input elif isinstance(message, AIMessage): if Action: in message.content: return tool_call return agent_thought return other def _evaluate_and_evict(self, new_unit: InformationUnit): 评估新单元并执行驱逐策略以维持令牌预算 # 1. 计算当前总令牌数 total_tokens sum(unit.token_count for unit in self.active_context) # 2. 如果未超限仅更新评分 if total_tokens self.token_limit: self._update_scores() return # 3. 超限执行驱逐策略这里实现最低价值优先 self._update_scores() # 按分数升序排序分数低的前面 self.active_context.sort(keylambda x: x.current_score) while total_tokens self.token_limit and len(self.active_context) 0: # 跳过核心单元 if self.active_context[0].id in self.core_units: # 将核心单元移到列表末尾避免被移除 core_unit self.active_context.pop(0) self.active_context.append(core_unit) continue # 移除分数最低的非核心单元 removed_unit self.active_context.pop(0) total_tokens - removed_unit.token_count print(fEvicted unit (Score: {removed_unit.current_score:.2f}): {removed_unit.content[:50]}...) # 可选将移除的单元转移到外部记忆 # self._externalize_unit(removed_unit) # 4. 按时间顺序重新排序以保持上下文连贯性 self.active_context.sort(keylambda x: x.timestamp) def _update_scores(self): 更新所有活跃单元的价值评分简化版规则 now time.time() for unit in self.active_context: recency max(0, 1 - (now - unit.timestamp) / 300) # 5分钟衰减因子 relevance 1.0 if unit.type in [user_input, system] else 0.7 density min(1.0, 100 / unit.token_count) if unit.token_count 0 else 1.0 # 假设信息密度与长度成反比 unit.current_score 0.5 * relevance 0.3 * recency 0.2 * density def get_active_context_text(self) - str: 将活跃上下文转换为LLM可接受的文本格式 context_lines [] for unit in self.active_context: prefix { user_input: User: , agent_thought: Thought: , tool_call: Action: , tool_result: Observation: , system: System: , other: }.get(unit.type, ) context_lines.append(f{prefix}{unit.content}) return \n.join(context_lines)4.2 步骤二与LangChain Agent集成接下来我们需要在自定义的Agent执行循环中用我们的管理器替代默认的记忆处理。from langchain.agents import AgentExecutor, Tool from langchain.llms import OpenAI # 假设你已经定义好了tools和llm tools [...] llm OpenAI(temperature0) # 创建自定义的上下文管理器 context_manager StructuredContextManager(token_limit4000) # 设置一个较小的预算便于测试 # 定义Agent的提示词模板其中包含一个 {context} 占位符 agent_prompt_template 你是一个智能助手。请根据以下上下文信息来回答问题或执行任务。 如果任务需要调用工具请严格按照格式输出。 当前上下文 {context} 现在开始处理。 # 在Agent的执行步骤中 def custom_agent_step(user_input: str): # 1. 将用户输入作为新单元加入管理器 context_manager.add_message(HumanMessage(contentuser_input)) # 2. 从管理器获取格式化后的当前活跃上下文 formatted_context context_manager.get_active_context_text() # 3. 填充提示词并调用LLM prompt agent_prompt_template.format(contextformatted_context) llm_response llm(prompt) # 这里简化了实际应使用完整的Agent推理逻辑 # 4. 解析LLM响应如果是工具调用执行工具... # ... (这里省略具体的工具调用和结果获取逻辑) # 5. 将LLM的“思考”和“工具调用”动作作为AIMessage加入管理器 context_manager.add_message(AIMessage(contentllm_response)) # 6. 如果有工具结果将其作为HumanMessageObservation加入管理器 # tool_result ... # context_manager.add_message(HumanMessage(contentfObservation: {tool_result})) # 返回最终结果 return parse_final_answer(llm_response)4.3 关键参数调优与监控实现基础功能后调优至关重要令牌预算token_limit不要设置为模型的最大上下文窗口如16384要留出安全余量例如80%以防单次生成过长。评分权重relevance,recency,density需要通过实验调整。一个实用的方法是记录智能体在长任务中因信息丢失导致的错误然后反推哪些信息被误驱逐了从而调整权重。例如如果智能体总是忘记用户早期提出的预算限制就需要提高system和早期user_input类型单元的relevance权重。驱逐阈值可以设置一个“软上限”和一个“硬上限”。当令牌数超过软上限如预算的90%时开始执行低强度驱逐如只压缩摘要层超过硬上限如预算的100%时执行高强度驱逐移除工作层低分单元。监控与日志务必详细记录每次驱逐动作驱逐了哪个单元、为什么、其评分如何。这不仅是调试的依据也是优化评估算法的重要数据来源。5. 常见问题与实战避坑指南在实际部署结构化上下文驱逐系统时我踩过不少坑这里分享几个典型的案例和解决方案。5.1 问题一上下文连贯性断裂现象智能体在长对话中突然“失忆”引用了一个已经被驱逐的信息或者提出的问题与之前逻辑衔接不上。根因驱逐策略过于激进或者评估算法未能正确识别信息单元之间的依赖关系。例如驱逐了一个工具调用请求但保留了它的结果导致结果失去意义。解决方案强化依赖链在InformationUnit中显式建模parent_id和children_ids。驱逐一个单元时进行“级联检查”。如果它是一个父单元考虑将其所有子单元一并驱逐或外化如果它是一个孤立的子单元父单元已被驱逐则应提高其优先级或为其生成一个自包含的摘要。引入“上下文锚点”在生成摘要或进行外化时强制在摘要文本或占位符中包含足够的关键词如实体名、任务ID以便未来需要时能通过检索准确找回。设置保护名单对于标识当前子任务目标的信息单元例如“当前步骤预订会议室”即使分数不高也在一段时间内免于驱逐。5.2 问题二摘要导致的信息失真现象为了节省空间将一个详细的工具结果如包含多行数据的JSON压缩成一句摘要如“成功查询到3条会议室信息”。后续步骤需要具体的会议室ID时智能体无法从摘要中获取导致任务失败。根因摘要的提示词Prompt设计不佳丢失了关键的结构化数据。解决方案面向任务的摘要设计不同的摘要模板。对于数据查询结果摘要应保留“实体类型-数量-关键属性”的模板例如“[查询结果] 会议室3间 (ID: A101, B202, C303 容量10人)”。这比自然语言摘要保留了更多可解析的信息。混合存储对于包含关键参数的结果采用“摘要关键数据提取”的方式。将生成的纯文本摘要存入上下文同时将提取出的结构化数据如[A101, B202, C303]以注释或隐藏字段的形式附着在单元上。虽然这些数据可能不直接输入LLM但可以被智能体的逻辑代码访问和使用。延迟压缩不要立即压缩刚产生的结果。让它在“工作层”保留一段时间例如完成与之相关的下一个子任务后再压缩确保其详细信息被充分使用。5.3 问题三评估开销影响性能现象智能体的响应速度变慢尤其是当上下文很长时每次评估所有单元分数耗时明显。根因评估函数过于复杂或者频繁地对整个上下文进行全量重评估。解决方案增量评估新单元加入时只计算新单元的分数。只有当需要执行驱逐时才对候选单元通常是分数可能较低的老旧单元进行重新评估而非全部。简化评分模型初期完全可以使用基于规则的线性评分。只有在规则无法处理复杂情况时才考虑引入微小的预测模型如预测某个信息单元在未来N步内被引用的概率。异步执行将评估和驱逐操作放在一个独立的、低优先级的线程中进行不阻塞主推理循环。当然这需要处理好数据同步的问题。5.4 问题四与外部记忆检索的协同失效现象将信息外化到向量数据库后当智能体需要历史信息时检索回来的内容不相关或无法有效融入当前上下文。根因外化时的“占位符”描述不准确或者检索查询query的生成策略不佳。解决方案优化占位符生成占位符不应只是[信息已存储]而应是一个高度浓缩的“索引”例如[关于项目‘Alpha’的2023年Q3销售数据包含区域拆分和同比已存储]。这个索引本身就可以作为未来检索的查询基础。动态生成检索查询当智能体的推理过程表明需要某类历史信息时例如它生成了“我需要回顾一下之前讨论过的预算限制”用这个意图陈述作为查询去检索比用当前整个对话作为上下文去检索要精准得多。检索结果的重集成将检索回来的信息重新插入上下文时不要简单拼接。最好以“回忆”的形式呈现例如“[根据之前存储的记录关于预算的讨论如下...]”帮助LLM理解这是召回的信息而非当前对话的新内容。实施结构化上下文驱逐不是一个一劳永逸的开关而是一个需要持续观察和调优的子系统。它迫使开发者更深入地思考智能体如何“思考”和“记忆”从而设计出更鲁棒、更高效的智能体系统。对于长程任务Long-Horizon Tasks而言良好的上下文管理能力往往是区分一个玩具原型和一个可用系统的关键所在。