深入 ‘Prompt Compression’：利用长文本压缩技术将 10k Token 的上下文精炼到 2k 且不丢失信息

各位同仁、各位技术爱好者，大家好！

今天，我们齐聚一堂，共同探讨一个在大型语言模型（LLM）时代日益凸显的关键议题——Prompt Compression，即提示词压缩。具体来说，我们将深入研究如何将一个长达 10,000 token 的上下文，在几乎不损失核心信息的前提下，精炼到 2,000 token 的长度。这不仅仅是一个工程挑战，更是一门艺术，它关乎我们如何高效、经济且精准地与最先进的AI模型交互。

作为一名编程专家，我深知在实际应用中，LLM 的上下文窗口限制、API 调用成本以及处理效率是制约其广泛部署的重要因素。当我们的输入文本远远超出模型的能力范围时，或者当我们希望在有限的预算内最大化信息利用率时，Prompt Compression 就显得尤为重要。我们将从理论基础出发，结合具体的代码实践，逐步剖析实现这一目标的各种策略和技术。

一、 Prompt Compression 的核心驱动力与挑战

在深入技术细节之前，我们首先要明确为何 Prompt Compression 如此重要，以及它所面临的根本挑战。

1.1 上下文窗口限制 (Context Window Limitations)

当前最先进的 LLM，如 GPT-4, Claude 3 等，虽然已经拥有了数万乃至数十万 token 的上下文窗口，但这并不意味着我们可以无限制地输入信息。一方面，超长上下文的模型往往调用成本更高；另一方面，即使模型支持，过长的输入也可能导致模型在提取关键信息时效率下降，甚至出现“迷失在中间”的现象（lost in the middle），即模型对位于输入中间部分的信息处理能力减弱。将 10k token 压缩到 2k token，能够有效适配更多主流模型的上下文窗口，提升兼容性。

1.2 成本与效率考量 (Cost and Efficiency)

LLM 的 API 调用通常是按 token 计费的。这意味着，每一次成功的压缩都直接转化为成本的节省。从 10,000 token 减少到 2,000 token，理论上可以带来高达 80% 的输入 token 成本节约。同时，处理更短的输入通常意味着更快的推理速度和更低的计算资源消耗，这对于实时应用或大规模部署至关重要。

1.3 信息过载与模型注意力 (Information Overload and Model Attention)

即使模型能够处理所有 10k token，过多的冗余信息也可能分散模型的注意力，使其难以聚焦于真正重要的部分。人类在阅读长文时也会主动提炼要点，LLM 同样需要这种“预处理”来帮助它更好地理解和响应。

1.4 “不丢失信息”的挑战 (The Challenge of "No Information Loss")

这是本次讲座的核心，也是最困难的部分。“不丢失信息”是一个极高的要求，它意味着我们不能简单地截断或粗暴概括。我们需要的是一种智能的、上下文感知的压缩，确保所有对最终任务至关重要的事实、论点、实体关系和语义完整性都能得到保留。这通常会将我们推向“近乎无损”或“任务相关无损”的范畴。

二、 Prompt Compression 的技术分类与策略概览

为了实现从 10k 到 2k 的精炼目标，我们将探索多种技术路径。这些方法可以大致分为以下几类：

分类	特点	适用场景	信息损失程度
1. 启发式/规则基方法	基于统计、关键词或预设规则进行裁剪和筛选。	文本结构清晰、冗余度高、对信息完整度要求不是极致的场景。	中到高，可能丢失上下文或细微语义。
2. 嵌入式语义压缩	利用文本嵌入（embeddings）识别语义相似度，进行冗余去除和关键信息检索。	需要保留语义相关性、识别重要主题或事实的场景。	低到中，能够较好地保留语义核心。
3. 模型基抽象/提取	利用更小的 LLM 或专门的 summarization 模型进行文本摘要或关键信息提取。	对信息密度要求高、需要生成精炼新文本的场景。	低到中，旨在保留核心意义，但表达方式可能改变。
4. 混合与迭代方法	结合上述多种技术，通过多阶段处理达到最佳效果。	对信息损失要求极低、复杂文本、追求极致压缩率和效果的场景。	最低，通过层层筛选和重构，最大程度保留信息。

我们将围绕这四类方法展开详细讨论，并辅以代码示例。

三、 深入技术实践：从 10k 到 2k 的精炼之路

为了模拟 10k token 的长文本，我们可以假设有一个包含多篇文章、聊天记录或详细文档的文本集合。我们的目标是从中提取出最核心的 2k token。

3.1 预处理与分块 (Preprocessing and Chunking)

在进行任何压缩之前，对原始文本进行基本的预处理和分块是必要的。这有助于后续的分析和选择。

import tiktoken
import re

# 假设的原始长文本 (这里用占位符，实际文本会很长)
# 为了演示，我们先构建一个模拟的10k token文本
def generate_long_text(num_tokens=10000):
    example_sentence = "Large language models (LLMs) are powerful AI systems that can understand and generate human-like text. They are trained on vast amounts of text data and can perform various natural language processing tasks, such as translation, summarization, and question answering. The development of LLMs has accelerated rapidly in recent years, leading to significant advancements in artificial intelligence. However, their computational requirements and context window limitations remain challenges."

    # 估算每个句子的token数
    encoding = tiktoken.get_encoding("cl100k_base") # 适用于GPT系列模型
    sentence_tokens = len(encoding.encode(example_sentence))

    # 计算需要重复多少次才能达到目标token数
    repetitions = (num_tokens // sentence_tokens) + 1

    long_text = (example_sentence + " ") * repetitions

    # 截断到准确的10k token，或者接近10k
    tokens = encoding.encode(long_text)
    if len(tokens) > num_tokens:
        long_text = encoding.decode(tokens[:num_tokens])

    return long_text

raw_long_text = generate_long_text(10000)

print(f"原始文本长度 (characters): {len(raw_long_text)}")

# 使用 tiktoken 估算 token 数量
encoding = tiktoken.get_encoding("cl100k_base")
initial_tokens = encoding.encode(raw_long_text)
print(f"原始文本估算 token 数量: {len(initial_tokens)}")

# 文本分句函数
def split_text_into_sentences(text):
    # 使用正则表达式进行分句，考虑常见的标点符号
    sentences = re.split(r'(?<=[.!?])s+', text)
    return [s.strip() for s in sentences if s.strip()]

sentences = split_text_into_sentences(raw_long_text)
print(f"原始文本分句数量: {len(sentences)}")

# 为了后续处理，将句子分块，每个块不超过一定token数
def chunk_sentences_by_tokens(sentences, max_chunk_tokens=500, encoding_model="cl100k_base"):
    encoding = tiktoken.get_encoding(encoding_model)
    chunks = []
    current_chunk_sentences = []
    current_chunk_tokens = 0

    for sentence in sentences:
        sentence_tokens = len(encoding.encode(sentence))
        if current_chunk_tokens + sentence_tokens <= max_chunk_tokens:
            current_chunk_sentences.append(sentence)
            current_chunk_tokens += sentence_tokens
        else:
            if current_chunk_sentences:
                chunks.append(" ".join(current_chunk_sentences))
            current_chunk_sentences = [sentence]
            current_chunk_tokens = sentence_tokens

    if current_chunk_sentences:
        chunks.append(" ".join(current_chunk_sentences))

    return chunks

text_chunks = chunk_sentences_by_tokens(sentences, max_chunk_tokens=500)
print(f"分块数量 (每个块最大500 token): {len(text_chunks)}")
print(f"第一个分块示例: {text_chunks[0][:150]}...") # 打印前150个字符

代码解析：

• generate_long_text：生成一个模拟 10k token 的长文本，用于演示。实际中，这会是你需要处理的真实文档。
• tiktoken：OpenAI 提供的 token 计数工具，用于准确估算文本的 token 数量。这对于控制压缩目标至关重要。
• split_text_into_sentences：一个简单的分句函数，将长文本拆分为独立的句子。句子是进行语义分析和选择的基本单位。
• chunk_sentences_by_tokens：将句子进一步组合成块。这样做有两个好处：
  • 效率： 处理句子列表可能非常庞大，分块可以减少后续模型处理的次数。
  • 上下文： 确保每个块内部仍保留一定的局部上下文，避免过度碎片化。

3.2 启发式与统计学方法 (Heuristic and Statistical Methods)

这类方法通常是第一步，能够快速去除显式冗余。

3.2.1 关键词提取与句子重要性排序 (Keyword Extraction & Sentence Importance Ranking)

利用 TF-IDF 或 TextRank 等算法识别文本中的重要词汇和句子。然后，优先选择包含这些重要词汇的句子，直到达到目标 token 数量。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

# 假设我们已经有了分好的句子列表 `sentences`
# 如果句子列表过长，可以先对text_chunks进行处理

def get_sentence_importance(sentences, top_n_sentences=None):
    if not sentences:
        return []

    # 使用TF-IDF计算词语重要性
    vectorizer = TfidfVectorizer(stop_words='english') # 可以自定义停用词
    tfidf_matrix = vectorizer.fit_transform(sentences)

    # 计算每个句子的TF-IDF平均值作为重要性分数
    # 更高级的方法会构建句子图并使用TextRank
    sentence_scores = tfidf_matrix.mean(axis=1).A.flatten()

    # 根据分数降序排序句子索引
    sorted_sentence_indices = np.argsort(sentence_scores)[::-1]

    if top_n_sentences is not None:
        selected_indices = sorted_sentence_indices[:top_n_sentences]
    else:
        selected_indices = sorted_sentence_indices

    return [sentences[i] for i in selected_indices], [sentence_scores[i] for i in selected_indices]

# 示例：从原始句子中选择一部分
# 注意：直接从10k token的原始句子中选择可能会非常慢，
# 实际操作中，可能先对`text_chunks`进行初步筛选，再对筛选出的块内部进行句子选择。
# 这里为了演示，我们假设原始句子数量适中。
# 我们先对模拟的10k文本进行分句，然后进行重要性排序
all_sentences_from_10k = split_text_into_sentences(raw_long_text)
print(f"n开始进行TF-IDF句子重要性排序 (共 {len(all_sentences_from_10k)} 句)...")

# 这是一个耗时操作，如果句子非常多，可能需要优化或采样
# 这里我们选择前500个句子进行演示，以控制计算量
# selected_sentences_by_tfidf, scores = get_sentence_importance(all_sentences_from_10k[:500])
# print(f"通过TF-IDF选出的部分重要句子数量: {len(selected_sentences_by_tfidf)}")
# print(f"第一个重要句子示例: {selected_sentences_by_tfidf[0][:150]}...")

# 实际应用中，我们需要循环选择，直到达到目标token数
def select_sentences_by_importance_until_target(sentences_with_scores, target_tokens=2000, encoding_model="cl100k_base"):
    encoding = tiktoken.get_encoding(encoding_model)
    selected_sentences = []
    current_tokens = 0

    # 句子及其分数的列表，按分数降序排列
    sorted_sentences = sorted(sentences_with_scores, key=lambda x: x[1], reverse=True)

    for sentence, _ in sorted_sentences:
        sentence_tokens = len(encoding.encode(sentence))
        if current_tokens + sentence_tokens <= target_tokens:
            selected_sentences.append(sentence)
            current_tokens += sentence_tokens
        else:
            # 如果加上当前句子会超过目标，则尝试添加下一个更短的句子
            # 简化处理：直接停止，或者可以尝试寻找更小的替代句子
            continue # 继续检查其他句子，看是否有更小的能塞进去

    # 最后再检查一次，确保总token数没超，并尽可能接近目标
    final_sentences = []
    final_tokens = 0
    # 对选中的句子进行排序，可以按原文顺序或重要性顺序
    # 这里我们按重要性排序，但更好的做法是按原文顺序保留连贯性
    for sentence in selected_sentences:
        sentence_tokens = len(encoding.encode(sentence))
        if final_tokens + sentence_tokens <= target_tokens:
            final_sentences.append(sentence)
            final_tokens += sentence_tokens

    return " ".join(final_sentences), final_tokens

# 假设我们已经计算了所有句子的TF-IDF分数
# 为了实际操作，我们可能会对分块进行TF-IDF，而不是所有句子
# 这里为了演示，我们直接从`all_sentences_from_10k`中计算
# 这是一个计算密集型步骤，实际部署时需要考虑性能
sentences_with_scores = []
if all_sentences_from_10k:
    vectorizer = TfidfVectorizer(stop_words='english')
    tfidf_matrix = vectorizer.fit_transform(all_sentences_from_10k)
    sentence_scores = tfidf_matrix.mean(axis=1).A.flatten()
    sentences_with_scores = list(zip(all_sentences_from_10k, sentence_scores))

compressed_text_tfidf, final_tokens_tfidf = select_sentences_by_importance_until_target(
    sentences_with_scores, target_tokens=2000
)
print(f"nTF-IDF 压缩后的文本 token 数量: {final_tokens_tfidf}")
# print(f"TF-IDF 压缩后的文本 (前200字): {compressed_text_tfidf[:200]}...")

代码解析：

• TfidfVectorizer：用于计算每个词在文档中的重要性（Term Frequency-Inverse Document Frequency）。
• get_sentence_importance：通过计算句子中词语的 TF-IDF 平均值，来衡量句子的重要性。这是一种简化的方法，TextRank 会更鲁棒。
• select_sentences_by_importance_until_target：迭代地选择重要句子，直到接近 2000 token 的目标。
• 局限性： 这种方法虽然简单快速，但可能导致选出的句子之间缺乏连贯性，丢失上下文，尤其是在“不丢失信息”的高要求下。它更适合从大量独立的事实性陈述中进行筛选。

3.2.2 冗余消除 (Redundancy Elimination)

长文本中常常存在重复或高度相似的句子，尤其是在聊天记录、会议纪要等场景。通过识别并去除这些冗余信息，可以有效压缩文本。

from sentence_transformers import SentenceTransformer
import torch

# 加载 SentenceTransformer 模型，用于生成句子嵌入
# 'all-MiniLM-L6-v2' 是一个轻量级且性能不错的模型
model = SentenceTransformer('all-MiniLM-L6-v2')

def remove_redundant_sentences(sentences, similarity_threshold=0.9):
    if not sentences:
        return []

    # 生成所有句子的嵌入
    print(f"正在为 {len(sentences)} 个句子生成嵌入...")
    sentence_embeddings = model.encode(sentences, convert_to_tensor=True, show_progress_bar=True)
    print("嵌入生成完毕。")

    # 用于存储非冗余句子的索引
    unique_sentence_indices = []
    # 用于存储已经处理过的句子的嵌入
    unique_embeddings = []

    for i, current_embedding in enumerate(sentence_embeddings):
        is_redundant = False
        if not unique_embeddings:
            # 如果是第一个句子，总是保留
            unique_embeddings.append(current_embedding)
            unique_sentence_indices.append(i)
            continue

        # 计算当前句子嵌入与所有已保留句子嵌入的余弦相似度
        similarities = cosine_similarity(current_embedding.unsqueeze(0), torch.stack(unique_embeddings))

        # 如果与任何一个已保留的句子相似度过高，则认为是冗余
        if torch.any(similarities > similarity_threshold):
            is_redundant = True

        if not is_redundant:
            unique_embeddings.append(current_embedding)
            unique_sentence_indices.append(i)

    return [sentences[i] for i in unique_sentence_indices]

# 示例：对原始分句进行冗余消除
# 同样，对于10k token的文本，句子数量会非常多，这里可能需要分批处理或对采样进行。
# 实际场景中，可以先对TF-IDF筛选出的句子进行冗余消除，或者反之。
print(f"n开始进行语义冗余消除 (共 {len(all_sentences_from_10k)} 句)...")
# 选取前2000个句子进行演示，以控制计算量
sentences_for_dedup = all_sentences_from_10k[:2000] 
unique_sentences = remove_redundant_sentences(sentences_for_dedup, similarity_threshold=0.9)
print(f"冗余消除后句子数量: {len(unique_sentences)}")
# print(f"冗余消除后的文本 (前200字): {' '.join(unique_sentences)[:200]}...")

# 将去重后的句子再次组合，并检查token数
dedup_text = " ".join(unique_sentences)
dedup_tokens = len(encoding.encode(dedup_text))
print(f"冗余消除后文本 token 数量: {dedup_tokens}")

代码解析：

• SentenceTransformer：一个强大的库，用于生成句子级别的语义嵌入。这些嵌入能够捕捉句子的深层含义。
• model.encode：将句子转换为高维向量。
• cosine_similarity：计算两个向量之间的余弦相似度，衡量它们的语义相似程度。
• remove_redundant_sentences：遍历所有句子，如果一个句子与之前已保留的任何句子语义相似度高于阈值，则将其视为冗余并丢弃。
• 优点： 能够有效去除语义重复的信息，这对于提高信息密度非常有用。
• 局限性： 阈值设置很关键，过高可能保留过多冗余，过低可能误删重要但表述略有差异的信息。计算所有句子两两相似度在句子数量巨大时计算量大。

3.3 嵌入式语义压缩 (Embedding-based Semantic Compression)

这类方法侧重于保留与某个特定查询或整体任务目标最相关的语义信息。

3.3.1 基于查询的语义检索 (Query-based Semantic Retrieval)

如果用户在长文本的上下文中有一个明确的“查询”或“目标”，我们可以利用语义检索技术，从 10k token 中挑选出与这个查询语义最相关的 2k token。

# 假设我们已经有了分好的 `text_chunks`
# 并且加载了 SentenceTransformer 模型 `model`

def retrieve_relevant_chunks(query_text, text_chunks, top_k=5, encoding_model="cl100k_base"):
    if not text_chunks:
        return []

    print(f"n正在为 {len(text_chunks)} 个文本块生成嵌入...")
    chunk_embeddings = model.encode(text_chunks, convert_to_tensor=True, show_progress_bar=True)
    print("文本块嵌入生成完毕。")

    # 生成查询的嵌入
    query_embedding = model.encode(query_text, convert_to_tensor=True)

    # 计算查询与所有文本块的余弦相似度
    similarities = cosine_similarity(query_embedding.unsqueeze(0), chunk_embeddings)[0]

    # 获取相似度最高的 top_k 个块的索引
    top_k_indices = torch.topk(similarities, k=min(top_k, len(text_chunks))).indices.tolist()

    # 根据原始顺序返回这些块
    # 确保返回的文本块按它们在原始文档中的顺序排列，以保持连贯性
    sorted_top_k_chunks = sorted([(text_chunks[i], i) for i in top_k_indices], key=lambda x: x[1])
    return [chunk for chunk, _ in sorted_top_k_chunks]

# 假设我们的目标是关于 "LLM的发展和挑战"
query = "What are the key developments and challenges regarding large language models?"

# 尝试检索最相关的块，直到达到目标 token 数
def select_chunks_by_relevance_until_target(query_text, text_chunks, target_tokens=2000, encoding_model="cl100k_base"):
    encoding = tiktoken.get_encoding(encoding_model)

    # 生成所有文本块的嵌入
    chunk_embeddings = model.encode(text_chunks, convert_to_tensor=True)
    query_embedding = model.encode(query_text, convert_to_tensor=True)

    # 计算相似度
    similarities = cosine_similarity(query_embedding.unsqueeze(0), chunk_embeddings)[0]

    # 获取按相似度降序排序的块索引
    sorted_chunk_indices = torch.argsort(similarities, descending=True).tolist()

    selected_chunks = []
    current_tokens = 0

    # 按照相似度从高到低选择块
    for idx in sorted_chunk_indices:
        chunk = text_chunks[idx]
        chunk_tokens = len(encoding.encode(chunk))

        if current_tokens + chunk_tokens <= target_tokens:
            selected_chunks.append((chunk, idx)) # 同时保留原始索引
            current_tokens += chunk_tokens
        else:
            # 如果当前块太大，跳过，尝试下一个
            continue

    # 将选中的块按原始文档顺序重新排序，以保持上下文连贯性
    selected_chunks_sorted_by_original_order = sorted(selected_chunks, key=lambda x: x[1])

    return " ".join([chunk for chunk, _ in selected_chunks_sorted_by_original_order]), current_tokens

retrieved_text_semantic, final_tokens_semantic = select_chunks_by_relevance_until_target(
    query, text_chunks, target_tokens=2000
)
print(f"n语义检索压缩后的文本 token 数量: {final_tokens_semantic}")
# print(f"语义检索压缩后的文本 (前200字): {retrieved_text_semantic[:200]}...")

代码解析：

• 此方法与前述冗余消除类似，都利用了句子嵌入。
• 核心区别在于，这里引入了一个 query_text，所有块都与这个查询进行相似度比较。
• select_chunks_by_relevance_until_target：按与查询的相似度降序选择文本块，直到达到目标 token 数。关键在于，为了保持文本的连贯性，最终选出的块会按照它们在原始文档中的顺序重新排序。
• 优点： 能够高度聚焦于与用户意图或任务目标相关的信息，是“任务相关无损”压缩的强大手段。
• 局限性： 依赖于一个明确的查询。如果用户意图不明确，或者文本中有多个不相关的子主题都重要，则难以有效选择。

3.3.2 最大边际相关性 (Maximal Marginal Relevance – MMR)

MMR 是一种更高级的语义检索策略，它不仅考虑相关性，还考虑多样性。在选择下一个文本块时，MMR 会选择与查询相关性高，同时与已选文本块相似度低的块。这有助于防止选出的文本过于重复，确保覆盖更广泛的关键信息。

def select_chunks_by_mmr_until_target(query_text, text_chunks, target_tokens=2000, lambda_param=0.5, encoding_model="cl100k_base"):
    encoding = tiktoken.get_encoding(encoding_model)

    if not text_chunks:
        return "", 0

    print(f"n正在为 {len(text_chunks)} 个文本块生成嵌入 (MMR)...")
    chunk_embeddings = model.encode(text_chunks, convert_to_tensor=True, show_progress_bar=True)
    query_embedding = model.encode(query_text, convert_to_tensor=True)
    print("嵌入生成完毕。")

    # 计算所有块与查询的相似度
    query_similarities = cosine_similarity(query_embedding.unsqueeze(0), chunk_embeddings)[0]

    selected_chunks = []
    selected_indices = []
    current_tokens = 0

    candidate_indices = list(range(len(text_chunks)))

    while current_tokens < target_tokens and candidate_indices:
        best_mmr_score = -1
        best_chunk_idx = -1

        for i in candidate_indices:
            # Relevancy term (similarity to query)
            relevance = query_similarities[i].item()

            # Diversity term (max similarity to already selected chunks)
            diversity = 0
            if selected_indices:
                # Need to compute pairwise similarity between current candidate and all selected chunks
                selected_embeddings = torch.stack([chunk_embeddings[j] for j in selected_indices])
                candidate_to_selected_similarities = cosine_similarity(
                    chunk_embeddings[i].unsqueeze(0), selected_embeddings
                )[0]
                diversity = torch.max(candidate_to_selected_similarities).item()

            # MMR score: lambda * relevance - (1 - lambda) * diversity
            # Higher lambda_param means more emphasis on relevance, lower means more emphasis on diversity
            mmr_score = lambda_param * relevance - (1 - lambda_param) * diversity

            if mmr_score > best_mmr_score:
                best_mmr_score = mmr_score
                best_chunk_idx = i

        if best_chunk_idx != -1:
            chunk = text_chunks[best_chunk_idx]
            chunk_tokens = len(encoding.encode(chunk))

            if current_tokens + chunk_tokens <= target_tokens:
                selected_chunks.append((chunk, best_chunk_idx))
                selected_indices.append(best_chunk_idx)
                current_tokens += chunk_tokens
                candidate_indices.remove(best_chunk_idx) # 从候选集中移除
            else:
                # 如果当前选出的最佳块太大，无法加入，则尝试下一个
                # 简单处理：将当前块从候选集中移除，继续寻找下一个最佳
                candidate_indices.remove(best_chunk_idx)
                # 如果没有其他块可以添加，则退出循环
                if not candidate_indices:
                    break
        else:
            # 如果没有找到任何合适的块，退出
            break

    # 将选中的块按原始文档顺序重新排序
    selected_chunks_sorted_by_original_order = sorted(selected_chunks, key=lambda x: x[1])

    return " ".join([chunk for chunk, _ in selected_chunks_sorted_by_original_order]), current_tokens

# 使用 MMR 进行压缩
compressed_text_mmr, final_tokens_mmr = select_chunks_by_mmr_until_target(
    query, text_chunks, target_tokens=2000, lambda_param=0.7 # 0.7表示更侧重相关性
)
print(f"nMMR 压缩后的文本 token 数量: {final_tokens_mmr}")
# print(f"MMR 压缩后的文本 (前200字): {compressed_text_mmr[:200]}...")

代码解析：

• lambda_param：平衡相关性（与查询的相似度）和多样性（与已选文本的相似度）的参数。lambda_param 越大，越侧重相关性；越小，越侧重多样性。
• 循环迭代地选择块，每次选择都计算 MMR 分数。
• mmr_score = lambda * relevance - (1 - lambda) * diversity：MMR 的核心公式。
• 优点： 解决了纯语义检索可能导致信息重复的问题，确保选出的信息既相关又全面。对于“不丢失信息”的要求，MMR 能更好地保留广度。
• 局限性： 计算复杂度高于纯语义检索，尤其是当已选文本块数量增加时。需要仔细调整 lambda_param。

3.4 模型基抽象/提取 (Model-based Abstractive/Extractive Summarization)

当上述方法仍不足以达到目标 token 数，或者需要更高级的语义理解和重构时，可以引入小型 LLM 或专门的摘要模型。

3.4.1 使用小型 LLM 进行分块摘要 (Chunk Summarization with Smaller LLMs)

策略是：将 10k token 文本分割成若干个小块（例如，每个块 2k token），然后用一个较小的、成本较低的 LLM 对每个小块进行摘要，将每个 2k token 块压缩到 500 token 甚至更少。最后，将所有摘要合并起来，形成一个更精炼的文本。如果合并后的文本仍然过长，可以再进行一次摘要。

from transformers import pipeline

# 加载一个轻量级的摘要模型
# 例如：'sshleifer/distilbart-cnn-12-6' 或 'facebook/bart-large-cnn'
# 对于中文可以考虑其他模型，如'csebuetnlp/mT5_mSum_XLSum'
# 这里为了演示，我们使用一个英文模型
try:
    summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6", device=0 if torch.cuda.is_available() else -1)
    print("摘要模型加载成功。")
except Exception as e:
    print(f"摘要模型加载失败，请检查模型名称或网络连接: {e}")
    print("将使用CPU进行处理，可能较慢。")
    summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6", device=-1) # Fallback to CPU

def summarize_chunks(text_chunks, max_summary_tokens=200):
    summaries = []
    print(f"n开始对 {len(text_chunks)} 个文本块进行摘要...")
    for i, chunk in enumerate(text_chunks):
        # 确保输入摘要模型的文本不超过其最大输入限制 (通常是 1024 token)
        # 我们可以用tiktoken再次检查并截断
        chunk_tokens_encoded = encoding.encode(chunk)
        if len(chunk_tokens_encoded) > 1000: # 假设摘要模型最大输入1000 token
            chunk = encoding.decode(chunk_tokens_encoded[:1000])

        # 摘要模型通常有自己的token计数器，这里使用 max_length 和 min_length 控制输出
        # max_length 对应的是模型自身的 token 计数
        summary = summarizer(chunk, max_length=max_summary_tokens, min_length=max_summary_tokens // 4, do_sample=False)[0]['summary_text']
        summaries.append(summary)
        print(f"已摘要第 {i+1}/{len(text_chunks)} 块。")
    return summaries

# 对原始的 text_chunks 进行摘要
# 假设每个原始块最大500 token，我们希望将其摘要到平均100 token
summarized_chunks = summarize_chunks(text_chunks, max_summary_tokens=100)

# 合并所有摘要
combined_summaries = " ".join(summarized_chunks)
combined_summaries_tokens = len(encoding.encode(combined_summaries))
print(f"所有摘要合并后的 token 数量: {combined_summaries_tokens}")

# 如果合并后的文本仍然超过目标 2k token，可以进行二次摘要
if combined_summaries_tokens > 2000:
    print(f"合并摘要仍过长 ({combined_summaries_tokens} token)，进行二次摘要...")
    final_summary_tokens_target = 2000 # 假设最终目标是2000 token
    # 计算每个 chunk 平均需要压缩到多少 token
    avg_tokens_per_chunk = final_summary_tokens_target / len(summarized_chunks)

    # 简单的二次摘要策略：对合并后的文本进行一次整体摘要
    # 注意：如果合并后的文本仍然很长，可能需要分块再摘要
    final_compressed_text_model = summarizer(combined_summaries, max_length=final_summary_tokens_target, min_length=final_summary_tokens_target // 4, do_sample=False)[0]['summary_text']
    final_compressed_tokens_model = len(encoding.encode(final_compressed_text_model))
    print(f"二次摘要后的文本 token 数量: {final_compressed_tokens_model}")
else:
    final_compressed_text_model = combined_summaries
    final_compressed_tokens_model = combined_summaries_tokens

# print(f"模型摘要后的文本 (前200字): {final_compressed_text_model[:200]}...")

代码解析：

• transformers.pipeline("summarization", ...)：Hugging Face transformers 库提供了一个方便的接口来使用预训练的摘要模型。
• summarizer(...)：对输入文本进行摘要。max_length 和 min_length 参数用于控制摘要的长度。
• 优点： 能够生成高度凝练、语义完整的新文本，有效去除冗余和非核心信息。对于“不丢失信息”的要求，这种方法旨在保留核心的“意义”而非字面表达。
• 局限性： 引入了额外的模型推理成本和延迟。摘要本身是“有损”的，它重写了文本，可能会丢失原文中的特定细节或风格。选择合适的摘要模型和调整参数至关重要。

3.5 混合与迭代策略 (Hybrid and Iterative Strategies)

实现“不丢失信息”的 10k 到 2k 压缩，通常需要结合多种方法的优势，形成一个多阶段的、迭代的流程。

3.5.1 推荐的综合工作流 (Recommended Integrated Workflow)

1. 初始清洗与分块 (Initial Cleaning & Chunking):

   • 去除无关字符、HTML 标签等。
   • 将 10k token 的原始文本分割成若干个大小适中的文本块（例如，每个块 500-1000 token）。
   • text_chunks

2. 冗余与低信息密度去除 (Redundancy & Low-Density Information Removal):

   • 对每个文本块或所有句子进行语义去重 (remove_redundant_sentences)。
   • 可以结合 TF-IDF 或其他启发式方法，初步筛选掉明显不重要的句子或段落。

3. 任务相关信息筛选 (Task-Specific Information Filtering – MMR/Semantic Retrieval):

   • 根据最终 LLM 任务的描述（即你的“终极提示词”）构建一个查询。
   • 利用 MMR (select_chunks_by_mmr_until_target) 或纯语义检索，从去重后的文本块中筛选出与查询最相关且具有多样性的块，将 token 数进一步压缩到 3k-4k 左右。

4. 分块抽象摘要 (Chunk Abstractive Summarization):

   • 对 MMR 筛选出的块进行进一步的抽象摘要 (summarize_chunks)。将每个块压缩到更小的 token 数。
   • 目标是使总 token 数接近 2k。

5. 最终检查与微调 (Final Check & Refinement):

   • 合并所有摘要，计算总 token 数。
   • 如果仍略微超出 2k，可以进行简单的截断（从不重要的末尾截断），或使用一个更强的摘要模型进行一次最终的整体摘要。
   • 确保文本的连贯性和可读性。

# 综合压缩流程示例 (伪代码，整合上述函数)

def comprehensive_prompt_compression(
    raw_long_text, 
    target_tokens=2000, 
    query_for_mmr="What are the main discussion points and conclusions?",
    encoding_model="cl100k_base"
):
    encoding = tiktoken.get_encoding(encoding_model)

    print("--- 阶段1: 预处理与分块 ---")
    sentences = split_text_into_sentences(raw_long_text)
    # 确保每个 chunk 不会过大，以适应后续模型的输入限制
    initial_chunks = chunk_sentences_by_tokens(sentences, max_chunk_tokens=500)
    print(f"初始分块数量: {len(initial_chunks)}")

    print("n--- 阶段2: 冗余消除 (在句子级别) ---")
    # 对所有句子进行去重，减少计算量
    # 注意：如果句子数量过多，这里可能仍需优化
    unique_sentences_global = remove_redundant_sentences(sentences, similarity_threshold=0.9)
    print(f"全局去重后句子数量: {len(unique_sentences_global)}")

    # 将去重后的句子重新分块，用于后续MMR
    dedup_chunks = chunk_sentences_by_tokens(unique_sentences_global, max_chunk_tokens=500)
    print(f"去重后重新分块数量: {len(dedup_chunks)}")

    print("n--- 阶段3: 任务相关信息筛选 (MMR) ---")
    # 假设MMR将文本压缩到3000-4000 token，为摘要留出空间
    intermediate_mmr_target = int(target_tokens * 2) # 比如先压缩到4k token
    mmr_compressed_text, mmr_tokens = select_chunks_by_mmr_until_target(
        query_for_mmr, dedup_chunks, target_tokens=intermediate_mmr_target, lambda_param=0.6
    )
    print(f"MMR 筛选后 token 数量: {mmr_tokens}")

    if mmr_tokens == 0:
        print("MMR 筛选未找到相关内容，返回空文本。")
        return "", 0

    # 将MMR选出的文本重新分块，用于摘要
    mmr_sentences = split_text_into_sentences(mmr_compressed_text)
    mmr_sub_chunks = chunk_sentences_by_tokens(mmr_sentences, max_chunk_tokens=500)
    print(f"MMR 文本重新分块数量: {len(mmr_sub_chunks)}")

    print("n--- 阶段4: 分块抽象摘要 ---")
    # 计算每个子块需要压缩到的目标 token 数
    if len(mmr_sub_chunks) > 0:
        avg_target_tokens_per_chunk = max(50, target_tokens // len(mmr_sub_chunks)) # 每个chunk至少50 token
    else:
        avg_target_tokens_per_chunk = target_tokens # 如果只有一个chunk，直接摘要到目标

    summarized_sub_chunks = summarize_chunks(mmr_sub_chunks, max_summary_tokens=avg_target_tokens_per_chunk)

    final_compressed_text = " ".join(summarized_sub_chunks)
    final_tokens = len(encoding.encode(final_compressed_text))
    print(f"抽象摘要后 token 数量 (初次合并): {final_tokens}")

    print("n--- 阶段5: 最终检查与微调 ---")
    if final_tokens > target_tokens:
        print(f"最终文本仍超出目标 ({final_tokens} > {target_tokens})，进行最终精简...")
        # 再次进行一次整体摘要，或智能截断
        # 这里的 summarizer 应该是之前加载好的
        final_compressed_text_refined = summarizer(final_compressed_text, max_length=target_tokens, min_length=target_tokens // 4, do_sample=False)[0]['summary_text']
        final_tokens_refined = len(encoding.encode(final_compressed_text_refined))
        print(f"最终精简后 token 数量: {final_tokens_refined}")
        return final_compressed_text_refined, final_tokens_refined

    return final_compressed_text, final_tokens

# 运行综合压缩流程
final_compressed_prompt, final_token_count = comprehensive_prompt_compression(
    raw_long_text, 
    target_tokens=2000,
    query_for_mmr="Summarize the key findings and recommendations from the document regarding LLM advancements and their implications.",
)

print(f"n--- 压缩结果 ---")
print(f"最终压缩后的 token 数量: {final_token_count}")
# print(f"最终压缩后的提示词 (前500字): {final_compressed_prompt[:500]}...")

代码解析：
这个综合流程将前面讨论的所有技术串联起来，形成一个多阶段的、有损但语义“近乎无损”的压缩管道。

• 阶段1： 标准化输入。
• 阶段2： 采用语义去重，这是无损压缩的第一步。
• 阶段3： 引入 query_for_mmr，通过 MMR 算法，在保证相关性和多样性的前提下，将文本大幅度削减。
• 阶段4： 对 MMR 筛选后的文本块进行抽象摘要，这是主要的有损压缩步骤，但目标是保留核心语义。
• 阶段5： 最后的微调确保不超出目标 token 数。

3.5.2 任务导向的动态压缩 (Task-Oriented Dynamic Compression)

“不丢失信息”的定义往往取决于最终的 LLM 任务。

• 问答系统 (Q&A): 重点是保留实体、事实和关系。
• 内容创作 (Content Generation): 重点是保留风格、语气、主题和关键创意元素。
• 决策支持 (Decision Support): 重点是保留论点、证据、利弊分析。

在实际应用中，query_for_mmr 和摘要模型的配置都应根据具体任务进行调整。例如，如果任务是问答，lambda_param 可以设得更高以侧重相关性；如果任务是生成创意文本，可以允许摘要模型更自由地重构。

四、 评估“不丢失信息”的有效性

如何判断我们是否真的“不丢失信息”了？这是一个复杂的问题，没有完美的量化指标。但我们可以从多个维度进行评估：

4.1 自动评估指标 (Automatic Evaluation Metrics):

• ROUGE scores (Recall-Oriented Understudy for Gisting Evaluation): 主要用于评估摘要质量。ROUGE-L 衡量最长公共子序列，适用于评估信息召回率。
• Embedding Similarity: 将原始 10k token 文本和 2k token 压缩文本分别生成一个整体嵌入（例如，通过平均所有句子嵌入或使用专用的文档嵌入模型），然后计算它们之间的余弦相似度。相似度越高，表明语义信息保留越好。
• Perplexity (困惑度): 如果有原始文本的后续部分，可以检查压缩后的文本作为上下文时，LLM 对后续文本的困惑度是否增加。

4.2 任务性能评估 (Task Performance Evaluation):

这是最直接也最可靠的方法。

• 设计一个基准任务： 使用原始 10k token 作为 LLM 的输入，得到一个基线输出。
• 使用压缩后的 2k token： 将压缩后的文本作为 LLM 的输入，得到另一个输出。
• 比较输出质量：
  • 人工评估： 让人类专家判断两个输出的质量、完整性、准确性和相关性。这是黄金标准。
  • LLM 辅助评估： 可以使用一个强大的 LLM（如 GPT-4）来评估两个输出，要求它判断“哪个输出在信息完整度、准确性和满足任务要求方面更好”。

4.3 挑战：

• “不丢失信息”往往是“不丢失对特定任务重要的信息”。因此，评估必须紧密围绕任务展开。
• 人类评估成本高昂，LLM 辅助评估可能存在模型偏见。
• 不同类型的“信息”有不同的重要性，例如事实性信息、情感信息、风格信息等，需要针对性地评估。

五、 结语

Prompt Compression 并非一劳永逸的解决方案，而是一个需要持续迭代和优化的过程。通过结合启发式规则、语义嵌入技术、模型抽象能力，并以任务性能为导向进行评估，我们能够有效地将长达 10,000 token 的上下文精炼到 2,000 token，同时最大程度地保留关键信息。这不仅能显著降低 LLM 的使用成本，提升处理效率，更能帮助模型在复杂、冗长的输入中更好地聚焦，从而解锁更多 LLM 在实际应用中的潜力。我们今天所探讨的技术，为构建更智能、更高效的 AI 驱动系统奠定了坚实的基础。