1. 项目概述为什么长会话是Agent的“阿喀琉斯之踵”如果你正在开发或使用一个AI Agent无论是处理客户服务的聊天机器人、自动编写代码的编程助手还是分析文档的智能体你一定遇到过这个令人头疼的场景对话进行得越久Agent的反应就越迟钝回答开始偏离主题甚至完全忘记了几分钟前你交代的关键指令。这不是你的错觉也不是模型变笨了其核心症结在于“上下文窗口”的限制。简单来说模型能同时“记住”并处理的对话文本长度是有限的。当对话轮次增多历史消息不断累积最终会超出这个限制导致最早、但可能最关键的信息被“挤出”模型的记忆区。这就是“Agent的上下文压缩”要解决的核心问题。它不是一个简单的文本摘要而是一套让Agent在长生命周期会话中持续保持高性能、高准确度的系统工程方法。想象一下你有一个不知疲倦、记忆力超群的数字员工但随着工作时间拉长它的办公桌上下文窗口堆满了杂乱的纸张历史消息效率自然暴跌。上下文压缩就是教会这位员工一套高效的文件归档、摘要和索引系统确保它手边永远是最相关、最重要的信息从而让会话能够持续、稳定地运行下去而不是在几十轮对话后就变得“老年痴呆”。对于工具型Agent即能够调用外部API、执行具体任务的智能体而言这个问题尤为尖锐。因为工具调用的结果、用户的复杂指令、系统的状态变更都需要被精确地记录和回溯。一个没有良好上下文管理能力的Agent就像一辆没有后视镜和导航记录的赛车在长途奔袭中极易迷失方向。因此理解并实施上下文压缩是从“玩具级”Demo迈向“生产级”可用Agent的关键一步。2. 核心原理消息结构的“瘦身”与“健身”哲学上下文压缩的本质是对Agent与模型交互的“消息结构”进行优化。我们通常会把整个对话历史包括用户消息、助手回复、系统指令、工具调用及结果一股脑地塞给模型作为上下文。这种“堆砌式”的消息结构在长会话中效率极低。压缩的核心思想是从“堆砌”转向“结构化精炼”。2.1 无损压缩与有损压缩的权衡在数据压缩领域有无损压缩如ZIP和有损压缩如JPEG。在Agent上下文中我们也面临类似的权衡无损压缩保留所有信息但改变形式关键信息提取与结构化不再保存完整的、冗长的工具调用结果JSON而是从中提取出状态、关键数据、错误码等核心字段以更紧凑的键值对形式存储。对话回合合并将多轮简单的“问-答”合并为一轮“多轮交互摘要”。例如用户连续追问几个概念Agent逐一回答可以合并为“用户咨询了A、B、C三个概念的定义Agent已分别解释”。指令归一化将分散在各处的、重复或相似的系统指令进行合并和去重。有损压缩舍弃部分信息保留精华摘要Summarization这是最直接的方法。当历史记录达到一定长度时触发一个摘要动作用模型将之前的对话浓缩成一段精炼的摘要。后续对话只携带这个摘要和最近的若干轮原始记录。选择性遗忘Selective Forgetting并非所有历史都同等重要。可以设计规则主动丢弃已被成功处理、不再相关的任务片段或者时间久远的寒暄内容。重要性评分Importance Scoring为每一条消息或信息片段赋予一个重要性权重。在上下文窗口紧张时优先保留权重高的内容。权重可以根据信息来源用户核心指令权重高、是否包含工具调用结果、是否被后续对话引用等因素动态计算。在实际应用中一个健壮的压缩策略往往是“组合拳”。例如对工具调用结果进行无损的结构化提取同时对长篇的对话历史进行有损的摘要并对所有信息进行重要性标记。2.2 工具型Agent消息结构的特殊挑战工具型Agent的消息结构比纯聊天机器人复杂得多通常遵循类似OpenAI的Function Calling或ReAct的格式。一个典型的回合可能包含用户: “查询北京明天天气然后根据天气建议我是否要带伞。” Agent思考: 我需要先调用天气查询工具。 [工具调用] function: get_weather, arguments: {“city”: “beijing”, “date”: “tomorrow”} [工具结果] {“city”: “北京”, “date”: “2023-10-27”, “weather”: “小雨”, “temp_range”: “15-20°C”} Agent回复: “北京明天有小雨气温15-20°C。建议您携带雨伞。”在长会话中这种用户-思考-工具调用-工具结果-回复的循环会不断重复上下文膨胀速度极快。其中工具调用结果尤其是包含大量数据的JSON是主要的“体积杀手”。因此针对工具型Agent的压缩必须重点处理工具交互记录。一个有效的策略是引入“工具结果摘要”层。原始的工具结果Raw Tool Output进入上下文前先经过一个轻量级处理模块。这个模块可能是一个简单的规则引擎提取status和data主字段也可能是一个小模型用于总结工具执行结果的核心结论。处理后的“精炼结果”才放入主对话上下文。原始结果可以存入一个独立的、可被查询的外部存储如向量数据库仅在需要溯源时才按需检索。3. 实操架构构建一个可管理的上下文系统纸上谈兵终觉浅我们来设计一个可落地的上下文管理架构。这个架构不依赖于某个特定框架而是提供一种通用的设计模式。3.1 分层上下文存储设计不要试图把所有东西都塞进模型的上下文窗口。我们应该建立一个分层存储系统活跃上下文Active Context直接提供给模型的最新、最相关的消息列表。其长度严格受限于模型上下文窗口如128K。这是模型的“工作记忆”。会话记忆Session Memory存储在外部数据库如Redis、PostgreSQL中的完整或压缩后的对话历史。这是Agent的“长期记忆”。知识库Knowledge Base与会话相关的、但非对话产生的静态或动态知识如产品文档、用户资料、工具手册等。通常存储在向量数据库如Chroma、Weaviate中支持语义检索。我们的压缩系统主要管理的是“活跃上下文”与“会话记忆”之间的数据流动。3.2 压缩触发与执行策略压缩不是定时任务而应该是基于策略的触发式操作。触发条件长度阈值当活跃上下文中的Token数达到模型窗口上限的某个比例如70%时触发。这是最直接的防御性策略。回合数阈值每完成N轮对话如10轮后触发一次压缩。这提供了规律的“整理”节奏。事件驱动在完成一个复杂任务如成功调用一系列工具解决了用户问题、或话题发生明显切换时触发。这更符合认知逻辑。压缩执行器 这是一个独立的模块其输入是待压缩的历史消息列表输出是压缩后的新上下文。它内部可能包含多种压缩器摘要压缩器调用一个文本摘要模型可以是主模型本身也可以是一个更小、更快的专用摘要模型生成历史摘要。结构化提取器基于预定义的规则或Schema从工具结果等结构化数据中提取关键字段。重要性过滤器根据预设规则如“包含工具调用的消息权重10”“用户第一句指令权重5”对消息打分保留高分消息。元数据压缩器将详细的系统提示如冗长的工具描述替换为简短的引用ID。3.3 消息结构的重构实践压缩不仅仅是删减文本更是重构消息结构使其信息密度更高。以下是一个重构示例压缩前原始堆砌:[系统指令] 你是一个天气助手可以调用get_weather工具...200字 [用户1] 今天上海天气如何 [助手1] 我来帮你查一下。 [工具调用1] get_weather(上海, today) [工具结果1] {city:上海,date:2023-10-26,weather:晴,temp:22,humidity:65%, wind:东风3级, ...更多字段} [助手1回复] 上海今天晴天气温22度。 [用户2] 那北京呢 [助手2] 再查一下北京。 [工具调用2] get_weather(北京, today) [工具结果2] {city:北京,date:2023-10-26,weather:多云,temp:15,humidity:30%, wind:北风2级, ...} [助手2回复] 北京今天多云15度。 [用户3] 两个城市温差挺大我该去哪 此时上下文已很长模型可能已忘记用户最初对比的意图压缩后结构化精炼:[系统指令] 天气助手可用get_weather工具。 [历史摘要] 用户曾询问上海26日晴22°C和北京26日多云15°C的今日天气意在对比。 [工具结果缓存引用] weather_cache: {sh_1026: {weather:晴, temp:22}, bj_1026: {weather:多云, temp:15}} [最近对话] [用户3] 两个城市温差挺大我该去哪在这个重构中冗长的系统指令被简化。前两轮对话被合并成一个“历史摘要”抓住了核心事实城市、日期、天气、温度和用户意图对比。详细但占空间的工具结果被移出主上下文替换为一个简单的“缓存引用”。当模型需要具体数据时可以通过这个引用键去外部缓存查询。主上下文中只保留了最相关的摘要和最新的用户问题信息密度极大提升。4. 实现细节与代码示例让我们以一个基于Python和类似LangChain理念的简易框架为例展示核心组件的实现。这里我们假设使用OpenAI的ChatCompletion模型。4.1 定义消息与上下文管理器首先我们需要一个比简单列表更智能的消息容器。from typing import List, Dict, Any, Optional from dataclasses import dataclass, field import tiktoken # 用于计算Token dataclass class Message: role: str # system, user, assistant, tool content: str # 新增字段用于压缩 importance: float 1.0 # 默认重要性权重 metadata: Dict[str, Any] field(default_factorydict) # 存储如工具名、结果摘要等 class ContextManager: def __init__(self, model_name: str gpt-4, max_tokens: int 128000): self.model_name model_name self.max_context_tokens max_tokens self.encoder tiktoken.encoding_for_model(model_name) self.active_messages: List[Message] [] # 活跃上下文 self.long_term_memory: List[Message] [] # 长期记忆简化示例 self.compression_threshold_ratio 0.7 def add_message(self, message: Message): 添加消息到活跃上下文 self.active_messages.append(message) self._check_and_compress() def get_messages_for_model(self) - List[Dict[str, str]]: 将活跃消息转换为模型API所需的格式 # 这里可以进行最后的格式转换比如将工具消息转换成模型认识的格式 formatted [] for msg in self.active_messages: if msg.role tool: # 对于工具消息我们可能只传递摘要内容而非完整结果 formatted.append({role: tool, content: msg.content}) else: formatted.append({role: msg.role, content: msg.content}) return formatted def _calculate_tokens(self, messages: List[Message]) - int: 粗略计算消息列表的token数 total 0 for msg in messages: total len(self.encoder.encode(msg.content)) 5 # 为role等元数据留余量 return total def _check_and_compress(self): 检查并触发压缩 current_tokens self._calculate_tokens(self.active_messages) if current_tokens self.max_context_tokens * self.compression_threshold_ratio: self._perform_compression() def _perform_compression(self): 执行压缩策略 print(上下文过长触发压缩...) # 策略1: 将重要性低的消息移到长期记忆 important_msgs [] for msg in self.active_messages: if msg.importance 0.5: # 重要性阈值 important_msgs.append(msg) else: self.long_term_memory.append(msg) # 移入长期记忆 # 策略2: 如果仍然过长对最早的消息进行摘要 if self._calculate_tokens(important_msgs) self.max_context_tokens * 0.6: # 假设我们有一个摘要函数 summarize_early_messages summary_msg self._summarize_early_messages(important_msgs[:4]) # 摘要前4条 # 用摘要替换被摘要的原始消息 compressed_messages [summary_msg] important_msgs[4:] self.active_messages compressed_messages else: self.active_messages important_msgs def _summarize_early_messages(self, messages: List[Message]) - Message: 模拟摘要过程。实际中需要调用摘要模型或算法。 # 这里简化处理拼接内容并标记为摘要 summary_content [历史摘要] | .join([f{m.role}:{m.content[:50]}... for m in messages]) return Message(rolesystem, contentsummary_content, importance0.8) # 使用示例 manager ContextManager(max_tokens4096) # 用小窗口演示 manager.add_message(Message(rolesystem, content你是助手..., importance1.0)) manager.add_message(Message(roleuser, content你好, importance0.2)) # ... 模拟添加很多消息后 model_ready_messages manager.get_messages_for_model()4.2 工具结果处理器的实现对于工具型Agent一个专门的结果处理器至关重要。class ToolResultProcessor: 处理并压缩工具调用结果 staticmethod def compress_raw_result(tool_name: str, raw_result: Dict[str, Any]) - Dict[str, Any]: 根据工具类型压缩原始结果 compression_strategies { get_weather: ToolResultProcessor._compress_weather, search_web: ToolResultProcessor._compress_search, execute_calculation: ToolResultProcessor._compress_calculation, # ... 其他工具 } strategy compression_strategies.get(tool_name, ToolResultProcessor._compress_generic) return strategy(raw_result) staticmethod def _compress_weather(raw_result: Dict) - Dict: 压缩天气结果只保留核心字段 return { compressed: True, essential_data: { weather: raw_result.get(weather), temperature: raw_result.get(temp), city: raw_result.get(city), date: raw_result.get(date) }, raw_reference: fweather_{raw_result.get(city)}_{raw_result.get(date)} # 原始数据引用键 } staticmethod def _compress_search(raw_result: Dict) - Dict: 压缩搜索结果提取前N条结果的标题和摘要 items raw_result.get(results, [])[:3] # 只取前3条 compressed_items [{title: i.get(title), snippet: i.get(snippet)[:100]} for i in items] return { compressed: True, count: len(raw_result.get(results, [])), top_results: compressed_items, raw_reference: fsearch_{hash(str(raw_result))[:8]} } staticmethod def _compress_generic(raw_result: Dict) - Dict: 通用压缩尝试转换为字符串并截断 result_str str(raw_result) if len(result_str) 200: return { compressed: True, preview: result_str[:200] ..., raw_reference: fgeneric_{hash(result_str)[:8]} } return raw_result # 短结果不压缩 # 在Agent调用工具后使用 raw_weather_data {city: 北京, date: 2023-10-27, weather: 小雨, temp: 15, humidity: 80%, wind: 东风2级, aqi: 45, ...} compressed ToolResultProcessor.compress_raw_result(get_weather, raw_weather_data) print(compressed) # 输出: {compressed: True, essential_data: {weather: 小雨, temperature: 15, city: 北京, date: 2023-10-27}, raw_reference: weather_北京_2023-10-27} # 然后将 compressed[essential_data] 的内容放入活跃上下文将 raw_weather_data 存入数据库键为 raw_reference。4.3 集成到Agent循环中最后我们将这些组件集成到Agent的主循环逻辑中。class CompressedContextAgent: def __init__(self, context_manager: ContextManager, result_processor: ToolResultProcessor): self.context context_manager self.processor result_processor self.raw_data_store {} # 简易内存存储用于存放原始数据 def run_tool(self, tool_name: str, tool_args: dict) - dict: 模拟调用工具并处理结果 # 1. 实际调用外部工具API (此处模拟) raw_result self._call_external_tool(tool_name, tool_args) # 2. 压缩工具结果 compressed_result self.processor.compress_raw_result(tool_name, raw_result) # 3. 存储原始结果以备后续详细查询 if compressed_result.get(compressed) and raw_reference in compressed_result: ref compressed_result[raw_reference] self.raw_data_store[ref] raw_result # 准备放入上下文的消息内容使用压缩后的精华数据 message_content f工具 {tool_name} 调用成功。结果概要: {compressed_result.get(essential_data, compressed_result.get(preview, N/A))} [详情ID: {ref}] else: message_content f工具 {tool_name} 调用成功。结果: {raw_result} # 4. 将工具结果作为一条消息添加到上下文 tool_msg Message(roletool, contentmessage_content, importance0.7) self.context.add_message(tool_msg) return raw_result # 返回原始结果给Agent逻辑 def _call_external_tool(self, name: str, args: dict) - dict: 模拟工具调用 # 这里是模拟实现 if name get_weather: return {city: args[city], date: args.get(date, today), weather: 晴, temp: 22, humidity: 65%, wind: 微风, aqi: 50} return {status: success, data: 模拟结果} def chat_cycle(self, user_input: str): 处理一轮用户输入 # 1. 将用户输入加入上下文 user_msg Message(roleuser, contentuser_input, importance0.9) self.context.add_message(user_msg) # 2. 获取当前压缩后的上下文发送给模型 messages_for_model self.context.get_messages_for_model() # 这里应调用实际的LLM API例如 openai.ChatCompletion.create # assistant_response call_llm_api(messages_for_model) # 为演示我们模拟一个回复 assistant_response 这是模型的回复。 # 3. 将助手回复加入上下文 assistant_msg Message(roleassistant, contentassistant_response, importance0.5) self.context.add_message(assistant_msg) print(f当前活跃消息数: {len(self.context.active_messages)}) return assistant_response # 模拟运行 agent CompressedContextAgent(ContextManager(max_tokens1000), ToolResultProcessor()) agent.chat_cycle(北京天气怎么样) # 模拟工具调用和结果压缩 agent.run_tool(get_weather, {city: 北京}) agent.chat_cycle(谢谢。那上海呢) # ... 持续对话ContextManager会在内部自动触发压缩5. 高级策略与优化方向基础的压缩策略能解决大部分问题但要打造真正鲁棒的生产级系统还需要考虑以下高级策略。5.1 基于向量检索的动态上下文重构这是更智能的压缩方式。我们不定期地删除或摘要旧消息而是将所有历史消息包括被压缩的都存入向量数据库。当需要构造当前对话的上下文时不按时间顺序堆砌而是用当前最新的用户消息作为查询向量去历史记忆库中检索最相关的N条历史记录。这样无论对话多长模型得到的上下文始终是“与当前问题最相关”的信息片段而不是时间最近但可能无关的片段。这种方法特别适合话题跳跃的对话。其实现步骤为每轮对话后将消息或其摘要向量化并存入向量库同时关联原始文本和元数据如时间、角色。收到新用户消息时将其向量化。从向量库中检索相似度最高的K条历史消息。将这些检索到的消息连同最新的系统指令和当前用户消息一起组装成模型的上下文。可以结合时间衰减因子让近期消息在相似度评分上有一定加成以保持对话的连贯性。5.2 重要性权重的动态计算消息的重要性不应是静态的。一个动态权重系统可以显著提升压缩质量。权重计算可以考虑以下因素消息类型系统指令 用户核心指令 工具调用与结果 助手确认回复 寒暄。信息新鲜度近期消息通常比远古消息更重要可以施加一个随时间指数衰减的权重。被引用次数如果一条历史消息如一个工具结果被后续对话多次提及或追问其权重应增加。包含关键实体消息中是否包含用户明确指定的关键实体如人名、产品名、任务ID。情感强度用户表达强烈情绪如不满、紧急的消息权重应提高。实现时可以为每条消息维护一个动态权重分数在每次压缩决策时根据最新对话状态重新计算或调整。5.3 分层摘要与主题分割对于超长对话单一摘要可能丢失过多细节。可以采用分层摘要回合级摘要每5-10轮对话生成一个局部摘要。主题级摘要当检测到对话主题切换时可通过文本聚类或意图识别对上一个主题的所有内容包括多个回合级摘要生成一个更高级别的主题摘要。全局摘要整个会话的顶层摘要。在构造上下文时可以呈现“全局摘要 当前主题摘要 最近若干轮原始对话”的结构。这样既保持了宏观背景又不丢失当前话题的细节。5.4 压缩效果的评估指标如何判断你的压缩策略是有效的需要定义一些评估指标任务完成率在长对话测试中Agent能否持续正确地完成用户请求的核心任务这是终极指标。上下文长度压缩后活跃上下文的平均Token数是否稳定在安全阈值以下信息保留度人工或通过辅助模型评估压缩后的上下文是否保留了完成当前任务所必需的关键信息如用户约束、工具结果关键数据。模型性能压缩前后模型生成回复的延迟Latency和Token消耗是否有显著优化用户满意度在真实或模拟测试中用户对长对话连贯性和准确性的主观评分。建立一套包含这些指标的测试集用于持续迭代和优化你的压缩策略。6. 常见陷阱与实战避坑指南在实际开发和运维中上下文压缩会引入新的复杂性以下是一些常见的坑和应对经验。6.1 过度压缩导致信息丢失这是最危险的问题。过于激进的摘要或过滤可能会丢掉用户早前设定的一个微小但关键的约束条件例如“请忽略上一条指令中的A选项”或者丢失工具结果中的一个关键错误码。避坑策略白名单保护定义绝对不能丢弃的信息类型。例如标记为用户核心指令的消息、包含错误码的工具结果必须无条件保留或在其生命周期内保持高权重。压缩前校验在应用压缩前用一个简单的规则或轻量模型快速检查即将被移除或摘要的消息中是否包含未完成的待办项todo、明确的否定指令不要、忽略或未解决的错误。可逆压缩尽量采用“引用”而非“删除”。将详细信息移出主上下文但保留一个可查询的ID。当模型后续表现异常时可以设计一个自省机制让Agent主动查询这些被“折叠”的原始信息。6.2 压缩引入的额外延迟和成本摘要模型调用、向量检索、动态权重计算都需要额外的计算资源可能会增加单轮对话的延迟和API成本。优化经验异步压缩压缩操作不必阻塞主对话流。可以在后台线程或任务队列中执行摘要和存储操作。主流程只进行轻量的、基于规则的必要压缩如工具结果字段提取。缓存摘要对于常见的对话模式或任务类型可以预生成或缓存摘要模板。例如“用户进行了多轮商品属性咨询”这类模式可以用一个模板来概括而不是每次都调用模型摘要。分级模型使用小模型如gpt-3.5-turbo或专用、高效的本地模型来处理摘要任务而不是每次都使用昂贵的主模型如gpt-4。惰性计算重要性权重不需要每轮都全量重算。可以设计增量更新算法只在相关事件如消息被引用发生时更新权重。6.3 上下文不一致与幻觉压缩后的上下文是原始历史的一种“失真”表示。模型基于这个失真版本进行推理可能会产生与完整历史不一致的回复甚至“幻觉”出不存在的内容。应对方法保留原始锚点即使在摘要中也要保留关键事实的原始出处标识。例如在摘要中写明“根据第3轮对话中工具X的结果”或者在消息元数据中链接回原始记录。一致性检查在Agent生成关键行动如调用工具、做出最终结论前可以设计一个验证步骤。例如让模型简要复述它所依据的关键历史事实系统可以将其与原始存储进行快速核对。明确压缩标记在给模型的上下文中明确告知哪些内容是摘要。可以使用特殊的标记如[系统摘要]、[历史回顾]让模型知道这部分信息是经过加工的二手信息可能需要谨慎对待。6.4 工具调用链的断裂在长任务中用户可能要求Agent执行一系列依赖前序结果的操作。如果中间某步的工具结果被过度压缩后续步骤可能因缺乏必要输入而失败。解决方案维护任务状态机为复杂的多步任务显式地维护一个状态机或任务图谱。上下文压缩不压缩这个状态机它独立于对话文本清晰记录每一步的输入、输出和状态。参数传递的显式化当模型决定调用工具时强制它从上下文中明确引用所需参数的来源。例如要求模型生成如调用工具Y参数data来自工具X结果中的字段。系统可以解析这种引用并从外部存储中获取原始值确保数据的完整性。压缩感知的工具设计在设计工具时就考虑压缩场景。让工具API的返回结构本身就包含“核心结果”和“完整详情”两个部分便于系统直接提取。上下文压缩不是一项一劳永逸的工作而是一个需要与你的Agent具体任务、用户交互模式以及所使用的模型特性深度结合并进行持续调优的工程领域。它没有银弹最好的策略往往是多种简单策略的组合并在真实流量中通过A/B测试不断迭代。记住目标是让Agent变得更聪明、更健谈而不是让它失忆。