ClaudeCode 上下文压缩实战：如何高效管理当前窗口的对话历史

1次阅读

没有评论

共计 2580 个字符，预计需要花费 7 分钟才能阅读完成。

在使用 Claude 等大语言模型进行长对话时，我们经常会遇到上下文窗口的限制。例如，Claude 的上下文窗口通常为 4K 或 8K token，当对话历史超过这个限制时，最早的信息会被丢弃。这会导致几个典型问题：

关键信息丢失：早期对话中的重要细节可能被遗忘
重复提问：用户不得不反复提供相同信息
连贯性下降：模型无法维持长期对话的上下文一致性

目前主流的上下文压缩方法主要有三种，各有优缺点：

正则匹配提取关键实体
优点：实现简单，计算成本低
缺点：依赖预定义规则，无法处理复杂语义
Sentence-BERT 语义聚类
优点：能捕捉语义相似性，聚类效果较好
缺点：需要额外模型加载，有一定计算开销
调用 Claude 自身生成摘要
优点：摘要质量高，保持原始语义
缺点：API 调用成本较高，延迟明显

下面是一个结合多种策略的动态压缩实现，包含重要性打分和滑动窗口逻辑：

from typing import List, Dict
import re
from sentence_transformers import SentenceTransformer
import numpy as np

class ContextCompressor:
    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
        self.sbert_model = SentenceTransformer(model_name)

    def score_importance(self, text: str) -> float:
        """计算文本片段的重要性得分"""
        # 规则 1：包含特定关键词
        key_terms = ['重要', '关键', '务必', '总结']
        term_score = sum(1 for term in key_terms if term in text) * 0.2

        # 规则 2：问题语句加分
        qmark_score = 0.3 if '?' in text else 0

        # 规则 3：长度适中加分(50-200 字符)
        length = len(text)
        len_score = 0.1 if 50 <= length <= 200 else 0

        return term_score + qmark_score + len_score

    def semantic_cluster(self, texts: List[str], threshold: float = 0.85) -> List[int]:
        """基于语义相似度聚类"""
        embeddings = self.sbert_model.encode(texts)
        sim_matrix = np.inner(embeddings, embeddings)

        clusters = []
        visited = set()

        for i in range(len(texts)):
            if i not in visited:
                cluster = [i]
                for j in range(i+1, len(texts)):
                    if sim_matrix[i][j] > threshold:
                        cluster.append(j)
                        visited.add(j)
                clusters.append(cluster)
                visited.add(i)

        return clusters

    def compress_context(self, 
                        history: List[Dict[str, str]], 
                        max_tokens: int = 3000) -> List[Dict[str, str]]:
        """执行上下文压缩"""
        try:
            # 第一步：重要性打分
            scored = [(msg, self.score_importance(msg['content'])) 
                     for msg in history]

            # 第二步：语义聚类
            texts = [msg['content'] for msg in history]
            clusters = self.semantic_cluster(texts)

            # 第三步：选择每个簇中得分最高的
            compressed = []
            for cluster in clusters:
                cluster_msgs = [scored[i] for i in cluster]
                best_msg = max(cluster_msgs, key=lambda x: x[1])[0]
                compressed.append(best_msg)

            # 第四步：token 限制
            current_tokens = sum(len(msg['content'].split()) for msg in compressed)
            while current_tokens > max_tokens and len(compressed) > 1:
                # 移除得分最低的
                min_score_idx = min(range(len(compressed)), 
                                  key=lambda i: self.score_importance(compressed[i]['content']))
                removed = compressed.pop(min_score_idx)
                current_tokens -= len(removed['content'].split())

            return compressed

        except Exception as e:
            print(f"压缩出错: {str(e)}")
            return history[:]  # 失败时返回原始历史

我们设计了三组对比测试：