Claude第三方模型上下文压缩实战：从原理到工程化落地

1次阅读

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

在处理长文本对话场景时，Claude 模型的上下文窗口限制确实是个头疼的问题。今天我就来分享下自己折腾第三方压缩技术的实战经验，希望能帮到有同样困扰的朋友们。

刚开始用 Claude 做对话系统时，经常遇到这些糟心情况：

截断失真：当用户输入超过 8K tokens 时，Claude 会自动截断，导致后半部分关键信息丢失。有次用户发了篇 2000 字的反馈，最后几句重要建议直接被砍掉了
API 成本飙升：按 token 计费的模式下，传大量上下文就像在烧钱。我们的客服机器人每月因此多花 $300+ 的冤枉钱
记忆断裂：在多轮对话中，早期的重要约定会被新对话挤掉。测试时发现 38% 的会话在第 5 轮后就开始答非所问

试过市面上主流的几种方案后，我整理了个对比表格：

方案	Rouge- L 得分	平均延迟	适用场景
LLMLingua	0.82	120ms	通用文本
GIST	0.76	85ms	技术文档
本文方案	0.85	150ms	对话系统

最终选择自研混合方案的原因是：

对话场景需要保持意图连贯性（Rouge- L 更重要）
能接受稍高的延迟换取更好的语义保持
需要灵活调整压缩粒度

我们的混合压缩方案分为三个关键模块：

def preprocess_text(text: str) -> tuple[list[str], list[float]]:
    """
    文本预处理：分句 + 初步清洗
    :param text: 原始文本
    :return: (句子列表, 初始权重列表)
    """
    try:
        sentences = sent_tokenize(text)
        weights = [1.0] * len(sentences)  # 初始权重

        # 过滤空句子和超短句
        valid_sentences = []
        valid_weights = []
        for sent, w in zip(sentences, weights):
            if len(sent.strip()) > 5:  
                valid_sentences.append(sent)
                valid_weights.append(w)

        return valid_sentences, valid_weights
    except Exception as e:
        logging.error(f"Preprocessing failed: {str(e)}")
        raise

这里融合了两种技术：

TF-IDF：快速识别高频关键词
Sentence-BERT：计算语义相似度

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('paraphrase-MiniLM-L6-v2')

def extract_key_sentences(sentences: list[str], 
    weights: list[float],
    target_ratio: float = 0.6
) -> list[str]:
    """
    混合算法抽取关键句
    :param target_ratio: 保留比例
    """
    # TF-IDF 计算
    vectorizer = TfidfVectorizer()
    tfidf_matrix = vectorizer.fit_transform(sentences)

    # SBERT 语义嵌入
    embeddings = model.encode(sentences)

    # 综合得分计算
    combined_scores = []
    for i in range(len(sentences)):
        tfidf_score = np.mean(tfidf_matrix[i].data)
        sbert_score = np.max([cosine_similarity([embeddings[i]], [embeddings[j]])[0][0]
            for j in range(len(sentences)) if j != i
        ])
        combined = 0.4 * tfidf_score + 0.6 * sbert_score
        combined_scores.append(combined * weights[i])

    # 按得分排序选取
    sorted_idx = np.argsort(combined_scores)[::-1]
    keep_num = int(len(sentences) * target_ratio)
    return [sentences[i] for i in sorted_idx[:keep_num]]

from langchain.chains import ConversationChain
from langchain_community.chat_models import ChatAnthropic

class CompressedConversationChain(ConversationChain):
    def __init__(self, *args, compress_ratio=0.6, **kwargs):
        super().__init__(*args, **kwargs)
        self.compress_ratio = compress_ratio

    def _call(self, inputs):
        # 原始上下文
        raw_context = inputs['history']

        # 压缩处理
        sentences, weights = preprocess_text(raw_context)
        compressed = extract_key_sentences(sentences, weights, self.compress_ratio)

        # 更新输入
        inputs['history'] = ' '.join(compressed)
        return super()._call(inputs)

# 使用示例
chain = CompressedConversationChain(llm=ChatAnthropic(),
    compress_ratio=0.5
)

在客服对话数据集上的测试结果：