大模型上下文压缩工程2026：让100K Token的信息塞进4K窗口

超长上下文固然好，但它带来高成本、高延迟和注意力稀释问题。本文深入探讨如何通过智能压缩技术，在有限上下文窗口内保留最大信息量，实现质量与效率的最优平衡。
—## 上下文窗口的本质矛盾表面上看，模型支持的上下文窗口越来越大——Gemini 2.5 Pro支持100万Token，Claude 3.7支持20万Token。但工程实践中，“能用"和"好用"是两回事：问题一：成本爆炸GPT-4o每百万Token输入成本$5，一次包含10万Token的请求就要$0.5。系统每天处理1万次请求，单日费用就是$5000。问题二：Lost in the Middle现象研究表明，LLM对超长上下文中间位置的信息注意力显著下降。即使你把所有内容都放进去，模型可能"漏看"关键信息。问题三：延迟变长上下文越长，首Token延迟（TTFT）越高。100K Token的请求可能需要等待5-10秒才开始输出，用户体验极差。解决方案：上下文压缩——在不丢失关键信息的前提下，减少输入Token数量。—## 五大压缩技术详解### 1. 摘要压缩（Summarization Compression）最直观的方法：把长对话历史压缩为摘要。pythonfrom openai import AsyncOpenAIfrom typing import listclass ConversationCompressor: """对话历史压缩器""" def __init__(self, max_history_tokens: int = 3000): self.openai = AsyncOpenAI() self.max_history_tokens = max_history_tokens self.summary = "" # 压缩后的历史摘要 self.recent_messages = [] # 保留最近的原始消息 self.recent_turns_to_keep = 3 # 保留最近3轮不压缩 def _estimate_tokens(self, text: str) -> int: """估算Token数（粗略，实际应用tiktoken）""" return len(text) // 4 async def _summarize(self, messages: list[dict]) -> str: """将消息列表压缩为摘要""" messages_text = "\n".join([ f"{m['role'].upper()}: {m['content']}" for m in messages ]) response = await self.openai.chat.completions.create( model="gpt-4o-mini", # 用便宜的模型做摘要 messages=[{ "role": "user", "content": f"""将以下对话历史压缩为简洁的摘要，保留所有关键信息、决策和上下文。用第三人称客观描述，不要遗漏重要细节。对话历史：{messages_text}请生成300字以内的摘要：""" }], max_tokens=400, temperature=0 ) return response.choices[0].message.content async def add_message(self, role: str, content: str): """添加新消息，必要时触发压缩""" self.recent_messages.append({"role": role, "content": content}) # 计算当前历史总Token total_tokens = sum( self._estimate_tokens(m["content"]) for m in self.recent_messages ) if self.summary: total_tokens += self._estimate_tokens(self.summary) # 超过阈值时压缩 if total_tokens > self.max_history_tokens: await self._compress() async def _compress(self): """压缩旧消息""" # 保留最近N轮，压缩其余部分 keep_count = self.recent_turns_to_keep * 2 # 每轮包含user+assistant to_compress = self.recent_messages[:-keep_count] if len(self.recent_messages) > keep_count else [] if not to_compress: return # 生成新摘要（包含旧摘要） new_summary_content = "" if self.summary: new_summary_content = f"历史摘要：{self.summary}\n\n" new_summary_content += "以上摘要基础上的新对话：" new_summary = await self._summarize(to_compress) if self.summary: # 合并摘要 merged = await self._merge_summaries(self.summary, new_summary) self.summary = merged else: self.summary = new_summary # 只保留最近的消息 self.recent_messages = self.recent_messages[-keep_count:] print(f"✂️ 压缩完成，从{len(to_compress)}条消息提取摘要") async def _merge_summaries(self, old: str, new: str) -> str: """合并两段摘要""" response = await self.openai.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"将以下两段对话摘要合并为一段，不超过500字：\n\n旧摘要：{old}\n\n新摘要：{new}" }], temperature=0 ) return response.choices[0].message.content def get_context(self) -> list[dict]: """获取当前压缩后的上下文""" messages = [] # 将摘要作为系统消息 if self.summary: messages.append({ "role": "system", "content": f"以下是对话历史摘要，请记住这些上下文：\n{self.summary}" }) # 追加最近的原始消息 messages.extend(self.recent_messages) return messages### 2. 选择性保留（Selective Retention）不是所有内容都同等重要，关键是找出"必须保留的”：pythonclass SelectiveContextManager: """选择性上下文管理器""" IMPORTANCE_PROMPT = """评估以下对话片段对当前任务的重要性，返回0-10的整数分数：对话片段：{conversation}当前任务：{current_task}重要性标准：10分：包含直接影响当前任务的关键信息、约束或决策7-9分：包含相关背景信息或可能有用的细节4-6分：相关但非必要的信息1-3分：边缘相关或已被后续对话更新的信息0分：完全无关的闲聊或已无效信息只返回数字：""" async def score_and_filter( self, messages: list[dict], current_task: str, keep_top_n: int = 10 ) -> list[dict]: """对消息评分并保留最重要的""" if len(messages) <= keep_top_n: return messages scored_messages = [] # 批量评分（用更便宜的模型） for i, msg in enumerate(messages): response = await self.openai.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": self.IMPORTANCE_PROMPT.format( conversation=f"{msg['role']}: {msg['content'][:500]}", current_task=current_task ) }], max_tokens=5, temperature=0 ) try: score = int(response.choices[0].message.content.strip()) except ValueError: score = 5 scored_messages.append((i, msg, score)) # 按分数排序，保留top N（同时保留消息的时间顺序） top_indices = sorted( [s[0] for s in sorted(scored_messages, key=lambda x: x[2], reverse=True)[:keep_top_n]] ) return [messages[i] for i in top_indices]### 3. 提取式压缩（Extractive Compression）从长文档中直接提取关键句子，无需重新生成：pythonfrom sklearn.feature_extraction.text import TfidfVectorizerfrom sklearn.metrics.pairwise import cosine_similarityimport numpy as npclass ExtractiveSummarizer: """提取式文档压缩（无需LLM API）""" def __init__(self, compression_ratio: float = 0.3): self.ratio = compression_ratio # 保留30%的句子 def compress(self, text: str, query: str = None) -> str: """提取最重要的句子""" import re # 分句 sentences = re.split(r'[。！？\n]+', text) sentences = [s.strip() for s in sentences if len(s.strip()) > 10] if not sentences: return text n_keep = max(1, int(len(sentences) * self.ratio)) if query: # 基于查询相关性选择句子 vectorizer = TfidfVectorizer() all_texts = [query] + sentences tfidf_matrix = vectorizer.fit_transform(all_texts) # 计算每句话与查询的相似度 query_vec = tfidf_matrix[0] sentence_vecs = tfidf_matrix[1:] scores = cosine_similarity(query_vec, sentence_vecs)[0] else: # 基于句子中心性选择 vectorizer = TfidfVectorizer() tfidf_matrix = vectorizer.fit_transform(sentences) similarity_matrix = cosine_similarity(tfidf_matrix) scores = similarity_matrix.sum(axis=1) # 选择得分最高的句子 top_indices = sorted(np.argsort(scores)[-n_keep:]) return "。".join([sentences[i] for i in top_indices])### 4. 分层压缩（Hierarchical Compression）对不同类型的内容采用不同压缩策略：pythonclass HierarchicalContextCompressor: """分层上下文压缩器""" async def compress_rag_context( self, retrieved_chunks: list[str], query: str, target_tokens: int = 2000 ) -> str: """压缩RAG检索到的文档块""" # 第一步：提取式过滤（快速去噪，无需LLM） extractor = ExtractiveSummarizer(compression_ratio=0.5) filtered_chunks = [ extractor.compress(chunk, query=query) for chunk in retrieved_chunks ] combined = "\n\n".join(filtered_chunks) # 如果还是太长，用LLM进行生成式压缩 current_tokens = len(combined) // 4 if current_tokens > target_tokens: response = await self.openai.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"""基于以下查询，从提供的文档中提取并压缩最相关的信息。目标：{target_tokens * 4}字以内。查询：{query}文档内容：{combined}请提取最关键的信息，保持准确性：""" }], max_tokens=target_tokens ) return response.choices[0].message.content return combined—## 各方案对比与选型建议| 方案 | Token减少率 | 信息保留质量 | 计算成本 | 适用场景 ||------|------------|-------------|---------|---------|| 摘要压缩 | 70-85% | 高 | 中（需LLM） | 长对话历史 || 选择性保留 | 40-60% | 很高 | 高（需逐条评分） | 文档检索 || 提取式压缩 | 50-70% | 中 | 低（无需API） | 快速预处理 || 分层压缩 | 60-80% | 高 | 中 | RAG系统 |推荐组合策略：- RAG系统：提取式（快速去噪）+ 选择性保留（精准筛选）- 长对话：摘要压缩（定期归档）+ 保留最近N轮原文- 实时系统：提取式（零成本）作为第一层过滤—## 生产指标监控上线后需要持续监控以下指标，确保压缩没有影响质量：pythonclass CompressionMetrics: def track(self, original: str, compressed: str, response_quality: float): return { "compression_ratio": len(compressed) / len(original), "token_reduction": (len(original) - len(compressed)) // 4, "response_quality_score": response_quality, # 人工或LLM评分 "cost_saved_usd": (len(original) - len(compressed)) / 4 / 1_000_000 * 5 }上下文压缩是LLM工程化中被严重低估的优化手段。在追求更大上下文窗口的同时，学会用好压缩技术，才能构建真正高效的生产系统。