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:)。关键是要设计一套统一的元数据schema:
class 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( content=message.content, unit_type=unit_type, timestamp=time.time(), parent_id=self._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(key=lambda 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(f"Evicted unit (Score: {removed_unit.current_score:.2f}): {removed_unit.content[:50]}...") # 可选:将移除的单元转移到外部记忆 # self._externalize_unit(removed_unit) # 4. 按时间顺序重新排序,以保持上下文连贯性 self.active_context.sort(key=lambda 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(temperature=0) # 创建自定义的上下文管理器 context_manager = StructuredContextManager(token_limit=4000) # 设置一个较小的预算便于测试 # 定义Agent的提示词模板,其中包含一个 {context} 占位符 agent_prompt_template = """ 你是一个智能助手。请根据以下上下文信息来回答问题或执行任务。 如果任务需要调用工具,请严格按照格式输出。 当前上下文: {context} 现在,开始处理。 """ # 在Agent的执行步骤中 def custom_agent_step(user_input: str): # 1. 将用户输入作为新单元加入管理器 context_manager.add_message(HumanMessage(content=user_input)) # 2. 从管理器获取格式化后的当前活跃上下文 formatted_context = context_manager.get_active_context_text() # 3. 填充提示词并调用LLM prompt = agent_prompt_template.format(context=formatted_context) llm_response = llm(prompt) # 这里简化了,实际应使用完整的Agent推理逻辑 # 4. 解析LLM响应,如果是工具调用,执行工具... # ... (这里省略具体的工具调用和结果获取逻辑) # 5. 将LLM的“思考”和“工具调用”动作作为AIMessage加入管理器 context_manager.add_message(AIMessage(content=llm_response)) # 6. 如果有工具结果,将其作为HumanMessage(Observation)加入管理器 # tool_result = ... # context_manager.add_message(HumanMessage(content=f"Observation: {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)而言,良好的上下文管理能力,往往是区分一个玩具原型和一个可用系统的关键所在。