Agent上下文管理:从Token限制到压缩策略的工程实践 面试时被问到 Agent 的上下文管理很多人能说出“有 Token 限制”、“需要压缩”这些概念但一旦追问“具体怎么压缩”、“压缩后信息怎么用”、“不同压缩策略的 trade-off 是什么”超过 80% 的候选人就开始卡壳回答停留在表面。这恰恰是区分“会用框架”和“理解原理”的关键分水岭。上下文管理不是简单地截断或总结而是一套权衡信息完整性、计算成本与任务目标的工程逻辑。尤其在 RAG、长对话、多轮工具调用等复杂场景下糟糕的上下文管理会直接导致 Agent “失忆”、逻辑混乱或成本失控。本文将从面试高频问题切入拆解 Agent 上下文管理的核心挑战并重点剖析那套让大多数人栽跟头的压缩逻辑。我们会用具体的代码示例展示从基础截断到高级压缩策略的实现路径并讨论在实际项目中如何根据场景选择策略。无论你是正在准备 Agent 相关面试还是在实际开发中遇到了上下文瓶颈这篇文章都将提供可直接落地的解决方案和深度思考。1. 为什么上下文管理是 Agent 的“阿克琉斯之踵”在讨论“怎么做”之前必须先理解“为什么难”。Agent 的上下文管理之所以复杂是因为它同时面临三个几乎相互矛盾的目标信息完整性Agent 需要记住完整的对话历史、工具调用结果、系统指令和用户偏好以做出连贯、准确的决策。计算成本可控大语言模型LLM按 Token 计费且长上下文会显著增加推理延迟和内存占用。无限制地增长上下文在经济和技术上都不可行。任务目标对齐并非所有历史信息都同等重要。冗余、过时或无关的信息会干扰模型判断降低任务完成质量。传统的“滑动窗口”或简单截断如只保留最近 N 轮对话虽然简单但粗暴地丢弃了可能关键的历史信息容易导致 Agent 在长任务中“失忆”。例如在一个多步骤的代码调试对话中早期用户指出的核心问题可能在第十轮对话时被截断导致 Agent 给出的解决方案完全偏离方向。因此上下文管理的本质是在有限的计算预算Token 数内最大化保留对当前决策最有价值的信息。而“压缩逻辑”就是实现这一目标的核心算法。2. 核心概念Token、上下文窗口与压缩在深入压缩逻辑前我们需要统一几个基础概念这是后续所有讨论的基石。2.1 Token 与上下文窗口TokenLLM 处理文本的基本单位。在英文中一个 Token 大约相当于 4 个字符或 0.75 个单词中文中一个汉字通常对应 1-2 个 Token。模型的输入和输出都受 Token 数量限制。上下文窗口Context Window模型单次处理所能接受的最大 Token 数量总和包括系统提示词、用户输入、历史对话和模型自身的输出。例如GPT-4 Turbo 拥有 128K 的上下文窗口。关键点上下文窗口是“硬约束”。当对话总长度超过这个限制你必须通过某种方式截断、压缩来缩减内容否则请求会失败。2.2 上下文压缩 vs. 上下文截断这是最容易混淆的一对概念也是面试中的高频考点。特性上下文截断上下文压缩核心逻辑直接丢弃超出窗口的部分如丢弃最早的对话。对原始信息进行提炼、摘要或表征保留其语义核心。信息保留永久丢失被丢弃部分的所有信息。试图保留原始信息的核心语义但存在信息损耗。计算开销几乎为零。需要额外的计算调用模型进行摘要、计算嵌入等。适用场景对话主题频繁切换早期历史完全无关或对成本极度敏感可接受信息丢失。长文档问答、多轮复杂任务、需要长期记忆的对话。举例只保留最近10轮对话。将前20轮对话总结成一段200字的摘要。简单来说截断是“物理删除”压缩是“化学提纯”。一个成熟的 Agent 系统通常会结合使用两者。3. 环境准备与前置工具为了演示后续的压缩逻辑我们需要一个基础的开发环境。本文以 Python 为例使用langchain和openai这两个流行的库。# 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langchain langchain-openai tiktokenlangchain: 提供了构建 Agent 和链的高级框架包含丰富的上下文处理工具。langchain-openai: LangChain 的 OpenAI 集成包。tiktoken: OpenAI 开源的 Token 计数工具准确且高效。确保你已设置好 OpenAI API 密钥export OPENAI_API_KEYyour-api-key-here # 或者在代码中设置# 文件config.py import os from langchain_openai import ChatOpenAI os.environ[OPENAI_API_KEY] your-api-key-here # 初始化一个 LLM这里使用 gpt-3.5-turbo 以控制成本 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0)4. 基础策略滑动窗口与简单截断我们先从最简单的策略开始实现理解其局限性。4.1 实现一个滑动窗口管理器# 文件simple_window_manager.py from typing import List, Dict, Any class SimpleWindowContextManager: 一个简单的基于轮次的滑动窗口上下文管理器。 def __init__(self, max_turns: int 10): 初始化管理器。 Args: max_turns: 保留的最大对话轮次数。 self.max_turns max_turns self.conversation_history: List[Dict[str, str]] [] # 格式: [{role: user, content: ...}, ...] def add_interaction(self, role: str, content: str): 添加一轮交互用户输入或AI回复。 self.conversation_history.append({role: role, content: content}) # 如果超出最大轮次从头部最旧开始删除 while len(self.conversation_history) self.max_turns * 2: # 每轮包含 user 和 assistant 两条 self.conversation_history.pop(0) def get_context_for_next_request(self) - List[Dict[str, str]]: 获取用于下一次LLM请求的上下文历史。 return self.conversation_history.copy() def get_current_turns(self) - int: 获取当前存储的对话轮次userassistant 算一轮。 return len(self.conversation_history) // 2 # 使用示例 if __name__ __main__: manager SimpleWindowContextManager(max_turns5) # 模拟一个长对话 for i in range(1, 8): manager.add_interaction(user, f用户第{i}轮提问) manager.add_interaction(assistant, fAI第{i}轮回答) print(f第{i}轮后历史记录数{len(manager.conversation_history)}) print(f当前最早的消息{manager.conversation_history[0] if manager.conversation_history else 无})运行结果分析 当对话进行到第6轮以后最早的第1轮对话会被自动移除。这就是最基础的滑动窗口。它的致命缺陷是如果第1轮对话包含了核心任务目标如“请帮我写一个Python爬虫目标是某网站”那么在后续的调试对话中Agent 将完全忘记这个初始目标。4.2 基于 Token 数的精确截断更专业的做法是基于 Token 数而非轮次进行截断。这需要用到tiktoken。# 文件token_based_truncator.py import tiktoken from typing import List, Dict, Any class TokenBasedContextTruncator: 基于Token数量进行精确截断的上下文管理器。 def __init__(self, model_name: str gpt-3.5-turbo, max_context_tokens: int 4096, reserve_for_output: int 1024): 初始化截断器。 Args: model_name: 使用的模型名称用于选择正确的编码器。 max_context_tokens: 模型总上下文Token限制。 reserve_for_output: 为模型输出预留的Token数。 self.encoding tiktoken.encoding_for_model(model_name) self.max_tokens max_context_tokens self.reserve_for_output reserve_for_output self.conversation_history: List[Dict[str, str]] [] def _count_tokens(self, messages: List[Dict[str, str]]) - int: 计算一组消息的Token总数。 # 根据OpenAI API的格式计算Token这是一个简化版 # 实际生产环境应使用更精确的方法如 tiktoken 的官方示例 total_tokens 0 for message in messages: total_tokens len(self.encoding.encode(message[content])) total_tokens 4 # 每个消息的格式开销近似值 total_tokens 2 # 每次请求的额外开销 return total_tokens def add_interaction(self, role: str, content: str): 添加交互并在必要时截断最旧的历史。 self.conversation_history.append({role: role, content: content}) # 计算当前总Token数包括预留的输出空间 while True: current_tokens self._count_tokens(self.conversation_history) estimated_total current_tokens self.reserve_for_output if estimated_total self.max_tokens: break # 如果超出移除最旧的一条消息 if len(self.conversation_history) 1: # 至少保留最新的一条 self.conversation_history.pop(0) else: # 即使单条消息也超限只能进行更激进的截断如截断内容本身 # 这里简化处理直接清空实际项目需更精细处理 self.conversation_history.clear() break def get_context_for_next_request(self) - List[Dict[str, str]]: 获取截断后的上下文。 return self.conversation_history.copy() # 使用示例 if __name__ __main__: truncator TokenBasedContextTruncator(max_context_tokens2000, reserve_for_output500) long_text 这是一段非常长的文本。 * 100 # 模拟长内容 truncator.add_interaction(user, long_text) print(f添加长文本后历史记录数{len(truncator.conversation_history)}) print(f第一条内容已截断{truncator.conversation_history[0][content][:50]}...)关键点基于 Token 的截断更精确能最大化利用上下文窗口。但它依然没有解决信息重要性判别的问题——最早被丢弃的可能恰恰是最重要的信息。5. 进阶核心上下文压缩逻辑详解现在我们进入文章的核心——压缩逻辑。压缩的目标是用更少的 Token 承载尽可能多的语义信息。5.1 压缩的两种基本范式提取式压缩Extractive Compression从原文中挑选出“关键”的句子、短语或片段直接拼接。类似于高亮。优点保留原文措辞无事实扭曲风险。缺点可能不连贯且压缩率有限不能比原文最短的句子更短。抽象式压缩Abstractive Compression模型理解原文后用自己的话重新表述核心内容。类似于写摘要。优点压缩率高输出连贯。缺点可能引入模型幻觉改变细微含义且计算成本高。在实际的 Agent 系统中两者常结合使用。5.2 实现一个基于 LLM 的对话摘要器抽象式压缩这是面试中最常被问到的“压缩逻辑”的具体实现之一。# 文件conversation_summarizer.py from langchain.prompts import ChatPromptTemplate from langchain.schema import StrOutputParser from langchain_openai import ChatOpenAI from typing import List, Dict, Any class ConversationSummarizer: 使用LLM对历史对话进行增量式摘要。 def __init__(self, llm): self.llm llm self.summary_prompt ChatPromptTemplate.from_messages([ (system, 你是一个高效的对话摘要助手。你的任务是将一段对话历史浓缩成一个简洁、连贯的段落保留所有关键事实、决策和用户意图。不要添加任何原文中没有的信息。), (user, 对话历史\n{history}\n\n请生成摘要) ]) self.chain self.summary_prompt | self.llm | StrOutputParser() self.current_summary # 存储当前的摘要 self.raw_buffer [] # 存储尚未被摘要的近期对话 def add_interaction(self, role: str, content: str): 添加交互到缓冲区。 self.raw_buffer.append(f{role}: {content}) def summarize_if_needed(self, buffer_size_threshold: int 5) - str: 如果缓冲区达到阈值则触发摘要并将摘要与当前摘要合并。 Args: buffer_size_threshold: 触发摘要的缓冲区消息条数。 Returns: 更新后的完整摘要。 if len(self.raw_buffer) buffer_size_threshold: # 准备要摘要的文本 to_summarize \n.join(self.raw_buffer) # 生成本轮摘要 new_summary self.chain.invoke({history: to_summarize}) # 合并摘要将本轮摘要与历史摘要结合再生成一个更精炼的总结 if self.current_summary: combined f历史摘要{self.current_summary}\n\n近期对话{new_summary} # 对合并后的内容进行再摘要防止摘要无限膨胀 final_summary self.chain.invoke({history: combined}) self.current_summary final_summary else: self.current_summary new_summary # 清空缓冲区 self.raw_buffer.clear() print(f[Summarizer] 已生成摘要。当前摘要长度{len(self.current_summary)} 字符) return self.current_summary def get_context_for_next_request(self) - List[Dict[str, str]]: 构建用于下一次请求的上下文。 messages [] # 1. 如果有摘要先加入系统消息或单独的消息段告知模型这是历史摘要 if self.current_summary: # 方式一作为系统消息的一部分更常见 # 方式二作为一条独立的历史用户消息 messages.append({role: user, content: f【历史对话摘要】{self.current_summary}}) # 2. 加入缓冲区中尚未摘要的近期原始对话 for line in self.raw_buffer: if line.startswith(user:): messages.append({role: user, content: line[5:]}) elif line.startswith(assistant:): messages.append({role: assistant, content: line[11:]}) return messages # 使用示例 if __name__ __main__: from config import llm # 导入之前配置的llm summarizer ConversationSummarizer(llm) # 模拟一个长对话 demo_dialogue [ (user, 我想开发一个个人博客系统使用Python的Django框架。), (assistant, 好的Django是个不错的选择。你需要有文章发布、分类和评论功能吗), (user, 是的还需要用户注册登录和简单的后台管理。), (assistant, 明白。我们可以使用Django内置的认证系统。你对前端有要求吗), (user, 前端希望用Bootstrap快速搭个样子后期再优化。), (assistant, 可以。我们先从创建项目和核心的Post模型开始吧。), (user, Post模型应该包含标题、内容、作者、创建时间这些字段吗), ] for role, content in demo_dialogue: summarizer.add_interaction(role, content) # 每3轮交互后检查并触发摘要 if len(summarizer.raw_buffer) 6: # 3轮 userassistant summarizer.summarize_if_needed() # 手动触发一次最终摘要 final_summary summarizer.summarize_if_needed(buffer_size_threshold1) print(\n 最终生成的对话摘要 ) print(final_summary) print(\n 用于下一次请求的上下文 ) context summarizer.get_context_for_next_request() for msg in context: print(f{msg[role]}: {msg[content][:100]}...)这段代码揭示了压缩逻辑的核心增量摘要不是每次请求都摘要全部历史而是定期将近期原始对话压缩成摘要。摘要合并将新摘要与旧摘要再次压缩防止“摘要的摘要”无限膨胀。混合上下文最终的提示词包含“精炼的摘要” “未压缩的近期原始对话”。这保证了长期记忆的保留和短期对话的精确性。5.3 基于嵌入向量的关键信息检索提取式压缩另一种思路是将历史对话片段向量化存储。当上下文窗口不足时根据当前问题从向量库中检索最相关的历史片段加入上下文。这就是 RAG 思想在上下文管理中的应用。# 文件vector_based_compressor.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from typing import List, Dict, Any, Tuple class VectorBasedContextCompressor: 使用向量检索来动态选择最相关的历史上下文。 def __init__(self, embedding_modeltext-embedding-3-small): self.embeddings OpenAIEmbeddings(modelembedding_model) self.vectorstore None self.text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段的Token数 chunk_overlap50, separators[\n\n, \n, 。, , , , , ] ) self.all_chunks [] # 存储所有文本块 self.all_metadatas [] # 存储对应的元数据如轮次、角色 def add_interaction(self, role: str, content: str, turn_id: int): 将交互内容分割并存入向量库。 # 为内容添加元数据 metadata {role: role, turn_id: turn_id, content_preview: content[:50]} chunks self.text_splitter.split_text(content) for chunk in chunks: self.all_chunks.append(chunk) self.all_metadatas.append(metadata.copy()) # 注意复制metadata # 每次添加后重建向量库简单实现生产环境应增量更新 if self.all_chunks: self.vectorstore Chroma.from_texts( textsself.all_chunks, metadatasself.all_metadatas, embeddingself.embeddings ) def get_relevant_context(self, query: str, k: int 3) - List[Tuple[str, Dict]]: 根据当前查询检索最相关的历史片段。 Args: query: 当前的查询或问题。 k: 返回的片段数量。 Returns: 一个列表包含文本片段 元数据的元组。 if not self.vectorstore: return [] docs self.vectorstore.similarity_search_with_score(query, kk) # 按相关性排序分数越低越相关 docs.sort(keylambda x: x[1]) return [(doc.page_content, doc.metadata) for doc, _ in docs] def get_context_for_next_request(self, current_query: str, max_tokens: int 2000) - List[Dict[str, str]]: 构建上下文系统指令 相关历史 当前查询。 messages [] # 1. 系统指令固定 system_msg 你是一个有帮助的助手。以下是一些可能相关的历史对话片段请参考它们来回答当前问题。 messages.append({role: system, content: system_msg}) # 2. 检索相关历史片段 relevant_chunks self.get_relevant_context(current_query, k5) history_context for chunk, meta in relevant_chunks: history_context f[第{meta[turn_id]}轮 {meta[role]}] {chunk}\n\n if history_context: messages.append({role: user, content: f相关历史\n{history_context}}) # 3. 加入当前查询 messages.append({role: user, content: current_query}) # 此处可添加基于Token数的截断逻辑确保总长度不超过max_tokens return messages # 使用示例 if __name__ __main__: compressor VectorBasedContextCompressor() # 模拟历史对话 history [ (1, user, Python里怎么读取一个CSV文件), (1, assistant, 可以使用内置的csv模块或者pandas库。pandas的read_csv更强大。), (2, user, 那用pandas读取后怎么查看前5行数据), (2, assistant, 用df.head()方法。), (3, user, 如果我想筛选出某列大于10的行呢), (3, assistant, 可以使用布尔索引比如df[df[column] 10]。), ] for turn_id, role, content in history: compressor.add_interaction(role, content, turn_id) # 模拟当前新问题 new_query 我怎么把筛选后的结果保存成新的CSV context_messages compressor.get_context_for_next_request(new_query) print( 为当前查询构建的上下文 ) for msg in context_messages: print(f{msg[role].upper()}: {msg[content][:150]}...)这种方法的优势上下文是动态的、与当前问题最相关的。它避免了固定窗口的盲目性也避免了摘要可能带来的信息损耗。劣势需要维护向量库引入额外的复杂性和延迟并且严重依赖嵌入模型的质量。6. 工程实践混合策略与 LangChain 集成在实际生产中单一的压缩策略往往不够。我们需要一个混合策略。幸运的是LangChain 提供了强大的ContextualCompressionRetriever等高级抽象。但理解其底层原理至关重要。一个典型的混合策略流水线可能是固定保留永远保留系统指令和最近 1-2 轮原始对话保证短期连贯性。摘要压缩对更早的对话进行定期摘要形成“长期记忆”。向量检索当上下文窗口仍紧张时用当前问题从“长期记忆”和更早的原始片段中检索最关键的部分。优先级排序与截断将上述所有内容按优先级如系统指令 当前问题 近期原始对话 检索到的相关历史 长期摘要排序然后进行基于 Token 的精确截断。# 文件hybrid_context_manager.py (概念性代码展示逻辑) class HybridContextManager: def __init__(self, llm, embedding_model): self.llm llm self.summarizer ConversationSummarizer(llm) self.vector_compressor VectorBasedContextCompressor(embedding_model) self.raw_recent [] # 存储最近N轮原始对话 self.max_raw_turns 3 def add_interaction(self, role, content, turn_id): # 1. 添加到原始近期缓冲区 self.raw_recent.append({role: role, content: content, turn_id: turn_id}) if len(self.raw_recent) self.max_raw_turns * 2: # 2. 缓冲区满了将最旧的内容移出并交给摘要器和向量库 oldest self.raw_recent.pop(0) self.summarizer.add_interaction(oldest[role], oldest[content]) self.vector_compressor.add_interaction(oldest[role], oldest[content], oldest[turn_id]) # 定期触发摘要 self.summarizer.summarize_if_needed() def build_context(self, current_query, max_tokens4000): messages [] # 策略1: 系统指令 (高优先级) messages.append({role: system, content: 你是一个助手。}) # 策略2: 长期摘要 (中优先级) summary self.summarizer.current_summary if summary: messages.append({role: user, content: f历史背景{summary}}) # 策略3: 向量检索的相关片段 (中高优先级因为最相关) relevant self.vector_compressor.get_relevant_context(current_query, k3) if relevant: rel_text \n.join([f[相关记录]{c} for c,_ in relevant]) messages.append({role: user, content: f相关参考{rel_text}}) # 策略4: 近期原始对话 (最高优先级之一) for msg in self.raw_recent[-4:]: # 保留最近2轮 messages.append({role: msg[role], content: msg[content]}) # 策略5: 当前问题 (最高优先级) messages.append({role: user, content: current_query}) # 最终基于Token的截断 return self._truncate_messages(messages, max_tokens) def _truncate_messages(self, messages, max_tokens): # 实现基于Token的截断逻辑从优先级最低的消息开始移除 # 此处省略具体实现 pass7. 常见问题与排查思路在实现和应用上下文压缩时你会遇到一些典型问题。问题现象可能原因排查方式解决方案Agent 忘记很早之前设定的核心目标。1. 使用了简单的滑动窗口截断。2. 摘要过于激进丢失了关键信息。3. 向量检索未命中关键片段。1. 检查上下文管理策略。2. 打印出实际发送给模型的完整上下文历史。3. 检查摘要内容是否包含目标关键词。1. 将核心目标固化在系统提示词中。2. 调整摘要提示词强调保留目标和约束。3. 优化检索查询或对关键信息进行手动标记和加权。上下文 Token 数计算不准确导致请求失败。1. Token 计数逻辑错误未考虑消息格式开销。2. 不同模型的 Token 计数方式不同。1. 使用官方库如tiktoken精确计算。2. 对比实际 API 调用返回的usage.prompt_tokens与自己计算的结果。1. 采用经过验证的 Token 计数函数。2. 在总限制中预留安全余量如 10%。摘要引入错误信息幻觉。摘要模型通常是通用 LLM在总结时扭曲了事实。人工抽样检查摘要内容与原始对话的一致性。1. 在摘要提示词中强调“严格基于原文”。2. 采用提取式摘要选择关键句作为补充或替代。3. 对摘要结果进行事实性验证可用另一个LLM。向量检索返回不相关的历史片段。1. 嵌入模型不适合该领域。2. 文本分块策略不合理如割裂了完整句子。3. 检索查询当前问题表述太模糊。1. 检查检索片段的相似度分数。2. 人工评估检索结果的相关性。3. 尝试不同的分块大小和重叠。1. 尝试领域微调的嵌入模型。2. 优化分块策略尝试按语义句子、段落分块。3. 对当前查询进行重写或扩展后再检索。系统延迟明显增加。压缩逻辑尤其是LLM摘要和向量检索引入额外计算。1. 性能剖析定位耗时环节。2. 监控摘要和检索的调用延迟。1. 降低摘要频率或使用更小的摘要模型。2. 对向量检索进行缓存。3. 采用异步处理压缩任务。8. 最佳实践与面试要点8.1 最佳实践分层管理将上下文视为短期原始、中期摘要、长期向量存储/知识库三层针对不同层级采用不同策略。重要性标记允许用户或系统对关键信息如任务目标、约束条件进行手动标记确保其不被压缩或丢弃。可配置策略提供配置选项让开发者根据应用场景如客服对话 vs. 代码生成调整压缩策略、窗口大小和摘要频率。监控与评估持续监控上下文使用情况平均Token数、压缩率和任务完成质量用数据驱动策略优化。备选方案当主要压缩策略如摘要失败时应有降级方案如回退到关键词提取或更激进的截断。8.2 面试要点梳理如果面试官问到“Agent的上下文管理”你可以按以下结构回答展现深度指出核心矛盾首先点明上下文管理是平衡信息完整性、成本与性能的挑战。对比基础方案简述简单截断的优缺点说明其无法满足长上下文需求。详解压缩逻辑这是重点。分点阐述提取式 vs 抽象式说明两者的区别和适用场景。摘要压缩的实现描述增量摘要、摘要合并、混合上下文摘要原始的完整流程。向量检索压缩说明如何将RAG思想用于上下文管理动态选取相关历史。混合策略提出在实际系统中如何结合固定窗口、摘要和检索并设定优先级。提及工程细节Token 的精确计算。压缩策略的可配置性。错误处理如摘要幻觉。性能考量。结合场景最后说明你会根据具体 Agent 的应用场景如文档分析、多轮对话、编程助手来选择和调整策略。上下文管理没有银弹。最有效的策略源于对具体业务场景、成本约束和技术栈的深刻理解。通过本文的拆解希望你能不仅记住“压缩”这个概念更能掌握设计一套健壮、高效上下文管理系统的逻辑和工具在面试和实战中都能从容应对。