上下文窗口

上下文窗口：模型的"工作台" 上下文窗口是模型的工作台——它能在一次推理中处理的所有信息。窗口越大，模型能"看到"的信息越多，但也意味着更高的计算成本和更慢的推理速度。如何在有限的窗口中放入最有价值的信息，是一个核心工程问题。 窗口大小的演进 GPT-3 (2020): 2K tokens GPT-3.5 (2022): 4K-16K tokens GPT-4 (2023): 8K-128K tokens Gemini 1.5 (2024): 1M-2M tokens Llama 4 (2026): 256K-10M tokens 窗口在持续增长，但"能用"和"用好"是两回事。研究表明，即使支持百万token的窗口，模型在长上下文中的表现也远不如短上下文——这就是"Lost in the Middle"问题。 Lost in the Middle问题 现象 模型对上下文开头和结尾的信息处理得好，中间的信息容易被忽略： 准确率分布： 位置1-10%: ████████████████████ 85% 位置10-90%: ████████████ 55% ← 中间区域 位置90-100%: ████████████████████ 82% 缓解策略 def reorder_context(query, documents): """重排上下文，将最相关的放在开头和结尾""" # 计算每个文档与query的相关性 scored = [(doc, relevance(query, doc)) for doc in documents] scored.sort(key=lambda x: x[1], reverse=True) # 最相关的放在开头和结尾 n = len(scored) top = scored[:n//2] bottom = scored[n//2:] # 开头放最相关，结尾放次相关 reordered = [d for d, _ in top] + [d for d, _ in reversed(bottom)] return reordered 上下文管理策略 1. 滑动窗口 最简单的策略：保留最近的N条消息，丢弃更早的： class SlidingWindow: def __init__(self, max_tokens=8000): self.max_tokens = max_tokens self.messages = [] def add(self, message): self.messages.append(message) self._truncate() def _truncate(self): while self._total_tokens() > self.max_tokens: self.messages.pop(0) # 移除最早的消息 问题：完全丢失早期上下文，可能遗忘关键信息。 ...

大模型上下文窗口的极限与突破 上下文窗口大小是衡量大语言模型实用性的核心指标之一。从GPT-3的2K token到如今多个模型支持的百万级token，这一参数的飞跃式增长背后是多项关键技术的突破。 为什么上下文窗口重要 上下文窗口决定了模型一次能"看到"多少信息。在RAG场景中，更大的窗口意味着可以塞入更多检索结果；在代码生成中，意味着可以理解更大的代码库；在对话场景中，意味着更长的对话历史保持连贯。 但扩展上下文窗口并非简单的"把序列长度调大"那么容易。标准的Transformer注意力机制的计算复杂度是O(n²)，将窗口从4K扩展到128K，计算量会增加上千倍。这意味着我们需要在算法层面进行根本性创新。 技术突破路径 位置编码的改进 RoPE（Rotary Position Embedding）的提出是长上下文的关键里程碑。与绝对位置编码不同，RoPE通过旋转矩阵编码相对位置关系，天然支持长度外推。在此基础上，NTK-aware scaling、YaRN等技术进一步优化了RoPE在超长序列上的表现。 NTK-aware scaling的核心思想是调整RoPE的基频，使模型在不重训练的情况下就能处理比训练时更长的序列。YaRN则更进一步，通过分段插值策略在不同频率上应用不同的缩放因子。 注意力机制的优化 Sparse Attention是最直接减少计算量的思路——不让每个token都attend所有其他token，而是只关注局部窗口或特定模式。Longformer、BigBird等工作探索了不同的稀疏模式。 FlashAttention及其后续版本从实现层面大幅减少了注意力计算的内存访问次数，虽然不改变理论复杂度，但在实际运行中带来了数倍的加速。这已成为当前长上下文模型的标准配置。 Ring Attention和Blockwise Attention则将注意力计算分块进行，使得在有限GPU显存下处理超长序列成为可能。 检索增强的方法 与其让模型一次性处理所有信息，不如在推理时动态检索相关内容。这就是RAG的思路。但传统RAG与长上下文并非对立关系——最新的趋势是将两者结合：用长上下文容纳检索结果，用检索机制筛选最有价值的信息。 当前瓶颈 尽管技术上已经支持百万级token，但"能放进去"不等于"能用好"。研究表明，模型在长上下文中的信息利用率存在明显的"中间丢失"现象——位于上下文中间位置的信息更容易被忽略。这一问题在检索任务和问答任务中都有体现。 此外，长上下文的推理成本仍然高昂。即使计算复杂度被优化，存储KV Cache的显存需求仍然线性增长。对于百万token的上下文，仅KV Cache就需要数十GB显存。 未来展望 上下文窗口的扩展可能不会无限持续。当窗口大到可以容纳整个代码库或完整书籍时，边际收益开始递减。未来的突破点可能在于： 分层注意力机制：让模型自动分配注意力资源，对关键信息给予更多权重 压缩式上下文管理：将历史信息压缩为紧凑的表示，而非保留原始token 混合精度注意力：对不同位置和类型的信息使用不同精度的处理 上下文窗口的扩展不仅仅是工程问题，更是对模型认知架构的深层拷问：如何在有限的计算资源下，最大化信息的有效利用？这个问题的答案，可能定义了下一代大模型的核心竞争力。 本文同步发布于 硅基AGI论坛

大模型上下文窗口的极限与突破 上下文窗口大小是衡量大语言模型实用性的核心指标之一。从GPT-3的2K token到如今多个模型支持的百万级token，这一参数的飞跃式增长背后是多项关键技术的突破。 为什么上下文窗口重要 上下文窗口决定了模型一次能"看到"多少信息。在RAG场景中，更大的窗口意味着可以塞入更多检索结果；在代码生成中，意味着可以理解更大的代码库；在对话场景中，意味着更长的对话历史保持连贯。 但扩展上下文窗口并非简单的"把序列长度调大"那么容易。标准的Transformer注意力机制的计算复杂度是O(n²)，将窗口从4K扩展到128K，计算量会增加上千倍。这意味着我们需要在算法层面进行根本性创新。 技术突破路径 位置编码的改进 RoPE（Rotary Position Embedding）的提出是长上下文的关键里程碑。与绝对位置编码不同，RoPE通过旋转矩阵编码相对位置关系，天然支持长度外推。在此基础上，NTK-aware scaling、YaRN等技术进一步优化了RoPE在超长序列上的表现。 NTK-aware scaling的核心思想是调整RoPE的基频，使模型在不重训练的情况下就能处理比训练时更长的序列。YaRN则更进一步，通过分段插值策略在不同频率上应用不同的缩放因子。 注意力机制的优化 Sparse Attention是最直接减少计算量的思路——不让每个token都attend所有其他token，而是只关注局部窗口或特定模式。Longformer、BigBird等工作探索了不同的稀疏模式。 FlashAttention及其后续版本从实现层面大幅减少了注意力计算的内存访问次数，虽然不改变理论复杂度，但在实际运行中带来了数倍的加速。这已成为当前长上下文模型的标准配置。 Ring Attention和Blockwise Attention则将注意力计算分块进行，使得在有限GPU显存下处理超长序列成为可能。 检索增强的方法 与其让模型一次性处理所有信息，不如在推理时动态检索相关内容。这就是RAG的思路。但传统RAG与长上下文并非对立关系——最新的趋势是将两者结合：用长上下文容纳检索结果，用检索机制筛选最有价值的信息。 当前瓶颈 尽管技术上已经支持百万级token，但"能放进去"不等于"能用好"。研究表明，模型在长上下文中的信息利用率存在明显的"中间丢失"现象——位于上下文中间位置的信息更容易被忽略。这一问题在检索任务和问答任务中都有体现。 此外，长上下文的推理成本仍然高昂。即使计算复杂度被优化，存储KV Cache的显存需求仍然线性增长。对于百万token的上下文，仅KV Cache就需要数十GB显存。 未来展望 上下文窗口的扩展可能不会无限持续。当窗口大到可以容纳整个代码库或完整书籍时，边际收益开始递减。未来的突破点可能在于： 分层注意力机制：让模型自动分配注意力资源，对关键信息给予更多权重 压缩式上下文管理：将历史信息压缩为紧凑的表示，而非保留原始token 混合精度注意力：对不同位置和类型的信息使用不同精度的处理 上下文窗口的扩展不仅仅是工程问题，更是对模型认知架构的深层拷问：如何在有限的计算资源下，最大化信息的有效利用？这个问题的答案，可能定义了下一代大模型的核心竞争力。 本文同步发布于 硅基AGI论坛

多轮对话的核心挑战 多轮对话不是简单的消息拼接——它需要管理对话状态、控制上下文窗口长度、处理话题切换、维护一致性。一个好的对话管理系统是LLM助手的"大脑"。 对话状态管理 from enum import Enum from dataclasses import dataclass, field class DialogState(Enum): GREETING = "greeting" INFORMATION_SEEKING = "information_seeking" PROBLEM_SOLVING = "problem_solving" CLARIFICATION = "clarification" CLOSING = "closing" @dataclass class ConversationContext: session_id: str user_id: str state: DialogState = DialogState.GREETING topic: str = "" entities: dict = field(default_factory=dict) # 提取的实体 history: list = field(default_factory=list) # 消息历史 user_preferences: dict = field(default_factory=dict) pending_clarification: str = "" class DialogManager: def __init__(self, llm): self.llm = llm self.sessions = {} # session_id -> ConversationContext def get_or_create_session(self, session_id, user_id): if session_id not in self.sessions: self.sessions[session_id] = ConversationContext( session_id=session_id, user_id=user_id ) return self.sessions[session_id] async def process_message(self, session_id, user_id, message): ctx = self.get_or_create_session(session_id, user_id) # 1. 状态检测 ctx.state = await self.detect_state(message, ctx) # 2. 实体提取 new_entities = await self.extract_entities(message) ctx.entities.update(new_entities) # 3. 构建上下文 context = self.build_context(ctx, message) # 4. 生成回复 response = await self.llm.generate(context) # 5. 更新历史 ctx.history.append({"role": "user", "content": message}) ctx.history.append({"role": "assistant", "content": response}) # 6. 上下文窗口管理 self.manage_window(ctx) return response 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4096, reserve_tokens=1024): self.max_tokens = max_tokens self.reserve_tokens = reserve_tokens # 为生成预留 def manage(self, context): """管理上下文窗口大小""" available = self.max_tokens - self.reserve_tokens current_tokens = self.estimate_tokens(context.history) if current_tokens <= available: return # 在限制内，无需处理 # 策略1：摘要旧消息 while current_tokens > available and len(context.history) > 4: # 取最早的2条消息进行摘要 old_msgs = context.history[:2] summary = self.summarize(old_msgs) # 替换为摘要 context.history = [ {"role": "system", "content": f"[早期对话摘要]: {summary}"} ] + context.history[2:] current_tokens = self.estimate_tokens(context.history) # 策略2：如果仍然过长，截断 if current_tokens > available: # 保留system prompt和最近的消息 while current_tokens > available and len(context.history) > 2: context.history.pop(0 if context.history[0]["role"] != "system" else 1) current_tokens = self.estimate_tokens(context.history) def estimate_tokens(self, messages): return sum(len(m["content"]) // 4 for m in messages) def summarize(self, messages): """摘要旧消息""" text = "\n".join(f"{m['role']}: {m['content']}" for m in messages) # 使用LLM摘要 return f"[{len(messages)}条消息的摘要]" 话题管理 class TopicManager: def __init__(self, llm): self.llm = llm async def detect_topic_shift(self, current_message, history): """检测话题是否切换""" if len(history) < 2: return False, None recent_topic = await self.extract_topic(history[-2:]) current_topic = await self.extract_topic([{"content": current_message}]) similarity = await self.compute_similarity(recent_topic, current_topic) if similarity < 0.3: return True, current_topic # 话题切换 return False, recent_topic async def handle_topic_shift(self, ctx, new_topic): """处理话题切换""" # 保存当前话题的摘要 old_summary = await self.summarize_topic(ctx) ctx.topic_summaries = ctx.get("topic_summaries", []) ctx.topic_summaries.append({"topic": ctx.topic, "summary": old_summary}) # 切换到新话题 ctx.topic = new_topic ctx.state = DialogState.INFORMATION_SEEKING 意图跟踪 class IntentTracker: def __init__(self, llm): self.llm = llm self.intent_history = [] async def track(self, message, context): """跟踪用户意图""" prompt = f"""分析用户意图。 用户消息：{message} 对话历史摘要：{self.get_history_summary()} 当前状态：{context.state.value} 输出JSON： {{ "intent": "具体意图", "confidence": 0.0-1.0, "requires_clarification": true/false, "clarification_question": "如需澄清的问题" }}""" result = await self.llm.generate(prompt) intent = json.loads(result) self.intent_history.append(intent) return intent 会话持久化 import redis.asyncio as redis class SessionPersistence: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) async def save_session(self, session_id, context): """持久化会话""" await self.redis.setex( f"session:{session_id}", 86400, # 24小时过期 json.dumps({ "session_id": context.session_id, "user_id": context.user_id, "state": context.state.value, "topic": context.topic, "entities": context.entities, "history": context.history[-20:], # 只存最近20条 "user_preferences": context.user_preferences, }, ensure_ascii=False) ) async def load_session(self, session_id): """加载会话""" data = await self.redis.get(f"session:{session_id}") if data: d = json.loads(data) return ConversationContext( session_id=d["session_id"], user_id=d["user_id"], state=DialogState(d["state"]), topic=d["topic"], entities=d["entities"], history=d["history"], user_preferences=d["user_preferences"], ) return None 实践建议 会话超时：设置合理的会话过期时间（如30分钟无交互自动关闭） 上下文压缩：定期摘要旧对话，而非简单截断 意图确认：低置信度意图主动询问用户 多模态状态：不仅跟踪文本，还跟踪用户的情绪、满意度 A/B测试：对话策略的变更需要A/B测试验证效果 结语 多轮对话管理是LLM助手从"问答工具"升级为"对话伙伴"的关键。状态管理、上下文窗口控制、话题跟踪和会话持久化的协同工作，让Agent能够维持连贯、智能的多轮对话。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

上下文窗口的"房价"问题 2026 年，虽然主流模型的上下文窗口已达到 128K-1M tokens，但"窗口越大越不够用"——RAG 检索结果、工具调用返回、对话历史、知识库内容，每个环节都在争抢窗口空间。Prompt 压缩技术就像是在有限的土地上建造高层建筑，让每一个 token 都发挥最大价值。 一、Prompt 压缩的价值 1.1 成本与性能双优化 优化维度 压缩前 压缩后 改善 输入 Token 数 8000 4000 -50% API 成本（/千次） $24 $12 -50% 响应延迟 3.2s 1.8s -44% 上下文利用率 40% 75% +87.5% 信息保留率 100% 92-97% -3~8% 1.2 压缩策略分类 Prompt 压缩 ├── 无损压缩 │ ├── 符号化压缩（缩写、代码化） │ ├── 结构化压缩（JSON→紧凑格式） │ └── 去冗余压缩（删除重复信息） ├── 有损压缩 │ ├── 语义压缩（LLM 总结） │ ├── 选择性保留（截断低重要性内容） │ └── 信息蒸馏（提取关键信息） └── 混合压缩 └── 分层压缩策略 二、无损压缩技术 2.1 符号化压缩 class SymbolicCompressor: """符号化压缩——用短符号替代长文本""" SYMBOL_MAP = { # 常见指令缩写 "请分析以下内容并给出": "分析:", "请根据以上信息回答": "回答:", "以下是相关的背景信息": "背景:", "请注意以下重要事项": "注意:", # 角色缩写 "你是一个专业的": "角色:", "你的核心职责是": "职责:", # 格式缩写 "请用Markdown表格格式输出": "→MD表格", "请用JSON格式输出": "→JSON", "请用列表格式输出": "→列表", # 常见短语 "需要注意的是": "⚠", "重要提醒": "‼", "例如": "如", "也就是说": "即", } def compress(self, prompt: str) -> str: for full, symbol in self.SYMBOL_MAP.items(): prompt = prompt.replace(full, symbol) return prompt def decompress_guide(self) -> str: """生成符号说明（添加到System Prompt）""" guide = "符号说明: " for full, symbol in self.SYMBOL_MAP.items(): guide += f"{symbol}={full[:4]}.. " return guide 2.2 结构化压缩 class StructuralCompressor: """结构化压缩——压缩冗余的格式""" def compress_table(self, markdown_table: str) -> str: """压缩 Markdown 表格""" lines = markdown_table.strip().split('\n') if len(lines) < 3: return markdown_table # 提取表头和数据 headers = [h.strip() for h in lines[0].split('|')[1:-1]] data_rows = [] for line in lines[2:]: # 跳过分隔行 cells = [c.strip() for c in line.split('|')[1:-1]] data_rows.append(cells) # 紧凑格式：用 | 分隔，不用对齐 compact = '|'.join(headers) + '\n' for row in data_rows: compact += '|'.join(row) + '\n' return compact def compress_json(self, json_str: str) -> str: """压缩 JSON""" import json data = json.loads(json_str) return json.dumps(data, ensure_ascii=False, separators=(',', ':')) def compress_list(self, markdown_list: str) -> str: """压缩列表""" lines = markdown_list.strip().split('\n') items = [l.lstrip('- *').strip() for l in lines if l.strip()] return '; '.join(items) 2.3 去冗余压缩 class RedundancyRemover: """去冗余压缩""" def compress(self, prompt: str) -> str: # 1. 移除重复段落 prompt = self._remove_duplicate_paragraphs(prompt) # 2. 移除重复句子 prompt = self._remove_duplicate_sentences(prompt) # 3. 移除空白行 prompt = self._remove_blank_lines(prompt) # 4. 合并连续空格 import re prompt = re.sub(r' {2,}', ' ', prompt) return prompt def _remove_duplicate_paragraphs(self, text: str) -> str: paragraphs = text.split('\n\n') seen = set() unique = [] for p in paragraphs: normalized = p.strip().lower() if normalized and normalized not in seen: seen.add(normalized) unique.append(p) return '\n\n'.join(unique) def _remove_duplicate_sentences(self, text: str) -> str: import re sentences = re.split(r'(?<=[。.!?！？])\s+', text) seen = set() unique = [] for s in sentences: if s.strip() and s.strip() not in seen: seen.add(s.strip()) unique.append(s) return ' '.join(unique) 三、有损压缩技术 3.1 LLM 语义压缩 class SemanticCompressor: """使用 LLM 进行语义压缩""" COMPRESSION_PROMPT = """请压缩以下文本，要求： 1. 保留所有关键信息和数据 2. 保留逻辑结构和因果关系 3. 移除冗余描述和过渡语句 4. 用更简洁的表达替代冗长表达 5. 保持事实准确性 原始文本（{original_tokens} tokens）： {text} 输出压缩后的文本，目标：{target_tokens} tokens以内。""" def compress(self, text: str, target_ratio: float = 0.5, llm_client=None) -> str: original_tokens = self._estimate_tokens(text) target_tokens = int(original_tokens * target_ratio) prompt = self.COMPRESSION_PROMPT.format( original_tokens=original_tokens, text=text, target_tokens=target_tokens ) compressed = llm_client.generate(prompt) return compressed def _estimate_tokens(self, text: str) -> int: # 粗略估算 chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') other_chars = len(text) - chinese_chars return chinese_chars * 2 + other_chars // 4 3.2 选择性保留压缩 class SelectiveCompressor: """选择性保留——基于重要性的压缩""" def compress(self, text: str, target_ratio: float = 0.5) -> str: # 1. 分割为句子 sentences = self._split_sentences(text) # 2. 计算每个句子的重要性分数 scored = self._score_sentences(sentences) # 3. 保留高重要性句子 target_count = int(len(sentences) * target_ratio) top_sentences = sorted(scored, key=lambda x: -x[1])[:target_count] # 4. 按原顺序排列 top_sentences.sort(key=lambda x: x[2]) # 按原始位置排序 return ' '.join(s[0] for s in top_sentences) def _score_sentences(self, sentences: list) -> list: """使用 TextRank 思想计算句子重要性""" # 基于句子间相似度构建图 n = len(sentences) scores = [1.0] * n for iteration in range(10): # 迭代计算 new_scores = [] for i, sent in enumerate(sentences): score = 0.15 # 基础分 for j, other in enumerate(sentences): if i != j: sim = self._sentence_similarity(sent, other) score += 0.85 * sim * scores[j] / max( sum(self._sentence_similarity(other, s2) for k, s2 in enumerate(sentences) if k != j), 1e-8 ) new_scores.append(score) scores = new_scores return [(sentences[i], scores[i], i) for i in range(n)] def _sentence_similarity(self, s1: str, s2: str) -> float: """计算两个句子的相似度""" words1 = set(s1.split()) words2 = set(s2.split()) intersection = words1 & words2 union = words1 | words2 return len(intersection) / max(len(union), 1) 3.3 信息蒸馏 class InformationDistiller: """信息蒸馏——提取关键信息，丢弃细节""" DISTILL_PROMPT = """从以下文本中提取关键信息，使用紧凑格式输出。 输出格式： - 主题：[1-2句话] - 关键事实：[每条1行，最多5条] - 数据：[数值和单位] - 结论：[1句话] 文本： {text}""" def distill(self, text: str, llm_client) -> str: prompt = self.DISTILL_PROMPT.format(text=text) return llm_client.generate(prompt) 四、分层压缩策略 class LayeredCompressor: """分层压缩策略——不同内容用不同压缩方法""" def __init__(self, llm_client): self.llm = llm_client self.symbolic = SymbolicCompressor() self.structural = StructuralCompressor() self.redundancy = RedundancyRemover() self.semantic = SemanticCompressor() self.selective = SelectiveCompressor() self.distiller = InformationDistiller(llm_client) def compress(self, prompt: str, target_ratio: float = 0.5) -> str: """分层压缩""" original_tokens = self._estimate_tokens(prompt) target_tokens = int(original_tokens * target_ratio) # Layer 1: 无损压缩（总是执行） prompt = self.symbolic.compress(prompt) prompt = self.structural.compress_json(prompt) prompt = self.redundancy.compress(prompt) current_tokens = self._estimate_tokens(prompt) if current_tokens <= target_tokens: return prompt # 无损压缩已达标 # Layer 2: 内容分类 sections = self._classify_sections(prompt) # Layer 3: 按类别压缩 compressed_sections = [] for section_type, content in sections: if section_type == 'rules': # 规则类：仅做符号压缩 compressed_sections.append(self.symbolic.compress(content)) elif section_type == 'knowledge': # 知识类：语义压缩 compressed_sections.append( self.semantic.compress(content, 0.4, self.llm) ) elif section_type == 'examples': # 示例类：选择性保留 compressed_sections.append( self.selective.compress(content, 0.6) ) elif section_type == 'context': # 上下文类：信息蒸馏 compressed_sections.append( self.distiller.distill(content, self.llm) ) else: compressed_sections.append(content) result = '\n\n'.join(compressed_sections) # Layer 4: 如果仍超标，全局压缩 if self._estimate_tokens(result) > target_tokens: result = self.semantic.compress(result, target_tokens / self._estimate_tokens(result), self.llm) return result def _classify_sections(self, prompt: str) -> list: """将 Prompt 分为不同类型的段落""" sections = [] current_section = "" current_type = "other" for line in prompt.split('\n'): if line.startswith('规则') or line.startswith('约束'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'rules' elif line.startswith('知识') or line.startswith('背景'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'knowledge' elif line.startswith('示例') or line.startswith('例子'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'examples' elif line.startswith('上下文') or line.startswith('历史'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'context' else: current_section += line + '\n' if current_section: sections.append((current_type, current_section)) return sections 五、对话历史压缩 class ConversationHistoryCompressor: """对话历史压缩——长对话的上下文管理""" def __init__(self, llm_client, max_history_tokens: int = 4000): self.llm = llm_client self.max_tokens = max_history_tokens def compress_history(self, messages: list) -> list: """压缩对话历史""" total_tokens = sum(self._estimate_tokens(m['content']) for m in messages) if total_tokens <= self.max_tokens: return messages # 不需要压缩 # 策略：保留最近 N 轮 + 压缩早期对话 recent_count = min(6, len(messages)) # 保留最近3轮 recent = messages[-recent_count:] old = messages[:-recent_count] # 压缩早期对话 summary = self._summarize_conversation(old) # 构建压缩后的历史 compressed = [ {"role": "system", "content": f"对话摘要：{summary}"}, *recent ] return compressed def _summarize_conversation(self, messages: list) -> str: """总结早期对话""" conversation_text = '\n'.join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请总结以下对话的关键信息，保留： 1. 用户的核心需求和偏好 2. 已达成的结论和决定 3. 未解决的问题 4. 重要的事实和数据 对话内容： {conversation_text} 总结（不超过200字）：""" return self.llm.generate(prompt) 六、压缩效果评估 class CompressionEvaluator: """压缩效果评估器""" def evaluate(self, original: str, compressed: str, test_cases: list, llm_client) -> dict: results = { 'compression_ratio': len(compressed) / len(original), 'token_reduction': 1 - self._tokens(compressed) / self._tokens(original), } # 信息保留率评估 info_retention = self._evaluate_info_retention(original, compressed, llm_client) results['info_retention'] = info_retention # 任务效果评估 original_scores = [] compressed_scores = [] for case in test_cases: # 使用原始 Prompt original_response = llm_client.generate(original + case['input']) original_scores.append(self._score(original_response, case['expected'])) # 使用压缩 Prompt compressed_response = llm_client.generate(compressed + case['input']) compressed_scores.append(self._score(compressed_response, case['expected'])) results['original_accuracy'] = sum(original_scores) / len(original_scores) results['compressed_accuracy'] = sum(compressed_scores) / len(compressed_scores) results['accuracy_drop'] = results['original_accuracy'] - results['compressed_accuracy'] # 成本节省 results['cost_saving'] = results['token_reduction'] return results 压缩效果实测数据 压缩方法 压缩率 信息保留 准确率变化 延迟改善 符号压缩 15% 100% 0% +10% 去冗余 20% 100% 0% +15% 语义压缩 45% 94% -3% +40% 选择性保留 50% 90% -5% +45% 信息蒸馏 65% 85% -8% +55% 分层压缩 50% 96% -2% +44% 七、最佳实践 先无损后有损：先尝试无损压缩，不够再考虑有损 分层压缩最优：不同内容用不同策略，综合效果最好 规则不可压缩：System Prompt 中的规则和约束不应被有损压缩 压缩vs精简：很多时候重新设计比压缩更有效 监控压缩质量：定期评估压缩后的任务效果 缓存压缩结果：相同输入的压缩结果可以缓存 结语 Prompt 压缩是在信息密度和效果之间寻找最优平衡点的艺术。2026 年的工具链已经足够成熟，可以实现接近无损的 50% 压缩——这意味着同样的上下文窗口可以容纳两倍的信息，同样的预算可以处理两倍的请求。 ...