上下文压缩策略：突破LLM窗口限制的实战方法

LLM 的上下文窗口是有限的——即使是最新的 200K token 模型，在处理长对话、大文档和复杂任务时也会遇到瓶颈。上下文压缩不仅是节省 Token 成本的手段，更是突破 LLM 能力天花板的关键技术。

一、为什么要压缩上下文？

1.1 上下文膨胀的代价

一次性对话（不考虑压缩）：
第1轮: 500 (sys) + 100 (user) + 200 (assistant) = 800 tokens
第5轮: 500 + 500×4 + 100 + 200 = 2,800 tokens
第10轮: 500 + 500×10 + 100 + 200 = 5,800 tokens
第50轮: 500 + 500×50 + 100 + 200 = 25,800 tokens

使用压缩后：
第50轮: 500 (sys) + 300 (压缩历史) + 100 (user) + 200 (output) = 1,100 tokens
节省: 95.7% 的 Token 消耗

1.2 压缩的三重收益

二、五大压缩策略

2.1 摘要压缩

class SummaryCompression:
    """
    将历史对话压缩为简洁摘要
    
    原理：定期让模型对之前的对话进行总结，
    用摘要替代完整历史
    """
    
    def __init__(self, llm, compress_every=5):
        self.llm = llm
        self.compress_every = compress_every
        self.summary = ""
        self.conversation_count = 0
    
    def compress_history(self, messages):
        """压缩对话历史"""
        # 提取最近的几轮对话
        recent = messages[-self.compress_every:]
        
        prompt = f"""
        当前对话摘要：
        {self.summary}
        
        以下是最新的对话轮次：
        {self._format_messages(recent)}
        
        请生成一份更新的对话摘要，包含：
        1. 用户的核心需求和关注点
        2. 已经提供的关键信息
        3. 尚未解决的问题
        4. 用户的偏好（如果有）
        
        摘要（200字以内）：
        """
        
        self.summary = self.llm(prompt)
        self.conversation_count += len(recent)
        
        return self.summary
    
    def get_context(self, current_message):
        """获取压缩后的上下文"""
        return f"""
        [对话历史摘要]
        {self.summary}
        
        [当前消息]
        {current_message}
        """

2.2 结构化压缩

class StructuredCompression:
    """
    将非结构化的对话历史转化为结构化数据
    
    格式：关键信息键值对 + 状态标记
    """
    
    def compress(self, messages):
        """结构化压缩"""
        prompt = """
        从以下对话中提取关键信息，以结构化格式输出：
        
        {messages}
        
        输出JSON格式：
        {
          "user_info": {
            "name": "如果用户提到过",
            "preferences": ["偏好列表"],
            "pain_points": ["痛点列表"]
          },
          "task_progress": {
            "completed": ["已完成事项"],
            "pending": ["待办事项"],
            "blockers": ["阻塞项"]
          },
          "key_decisions": [
            {"topic": "决策主题", "decision": "结果"}
          ],
          "context_tags": ["关键词标签"]
        }
        """
        
        return self.llm(prompt)

2.3 滑动窗口

class SlidingWindow:
    """
    只保留最近 N 轮对话，丢弃早期内容
    适用于：短期记忆即可的任务
    """
    
    def __init__(self, window_size=10):
        self.window_size = window_size
        self.history = []
    
    def add_turn(self, user_msg, assistant_msg):
        self.history.append(("user", user_msg))
        self.history.append(("assistant", assistant_msg))
        
        # 保留最近 window_size 轮
        if len(self.history) > self.window_size * 2:
            self.history = self.history[-(self.window_size * 2):]
    
    def get_context(self):
        return self.history

2.4 重要性裁剪

class ImportancePruning:
    """
    根据信息重要性选择性保留
    
    保留策略：
    - 高重要性：用户明确确认的信息、关键决策
    - 中重要性：一般讨论、问题
    - 低重要性：寒暄、确认性对话
    """
    
    IMPORTANCE_KEYWORDS = {
        'high': ['确定', '选择', '决定', '确认', '重要', 
                 '关键', '必须', '记住', '信息'],
        'low': ['嗯', '好的', '明白', '谢谢', '哦', 
                '对的', '是的', '没错']
    }
    
    def prune(self, messages):
        """根据重要性裁剪历史"""
        pruned = []
        
        for role, msg in messages:
            importance = self._score_importance(msg)
            
            if importance == 'high':
                pruned.append((role, msg))
            elif importance == 'low':
                continue  # 丢弃低重要性内容
            else:
                # 中等重要性：压缩后保留
                compressed = self._compress_message(msg)
                pruned.append((role, compressed))
        
        return pruned

2.5 检索增强压缩

class RetrievalAugmentedCompression:
    """
    结合向量检索，只保留与当前问题相关的历史
    """
    
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.vector_store = []
        self.messages = []
    
    def add_message(self, role, content):
        embedding = self.embedding_model(content)
        self.vector_store.append(embedding)
        self.messages.append((role, content))
    
    def get_relevant_context(self, query, top_k=5):
        """检索与当前查询最相关的内容"""
        query_emb = self.embedding_model(query)
        
        similarities = [
            cosine_similarity(query_emb, stored_emb)
            for stored_emb in self.vector_store
        ]
        
        top_indices = np.argsort(similarities)[-top_k:]
        
        context = []
        for idx in top_indices:
            if similarities[idx] > 0.5:  # 阈值过滤
                context.append(self.messages[idx])
        
        return context

三、策略对比

四、组合策略

class HybridCompression:
    """
    多层组合压缩策略
    
    L1: 滑动窗口（去掉太久远的内容）
    L2: 重要性裁剪（去掉低价值内容）
    L3: 摘要压缩（保留关键信息的摘要）
    """
    
    def compress(self, messages, query):
        # L1: 滑动窗口
        recent = messages[-20:]
        
        # L2: 检索增强
        relevant = self.retrieval.get_relevant_context(query, top_k=10)
        
        # L3: 合并去重
        merged = self._merge_and_dedup(recent, relevant)
        
        # L4: 摘要压缩长内容
        if self._total_tokens(merged) > 3000:
            merged = self.summary.compress(merged)
        
        return merged

五、最佳实践

5.1 场景推荐

5.2 注意事项

1. 信息丢失是必然的：压缩必然有信息损失，需根据场景权衡
2. 压缩频率：不是每轮都需要压缩，设置合理的压缩间隔
3. 可回溯性：对重要对话，保留原始日志以便回溯
4. 阈值调优：根据实际效果调整相似度阈值和裁剪强度

总结

上下文压缩是 LLM 生产环境中的必备技术。没有放之四海而皆准的策略，需要根据具体场景选择合适的压缩方式，或组合多种策略以达到最佳效果。核心原则：在 Token 成本和信息完整性之间找到最适合当前场景的平衡点。