第13章：检索压缩优化¶

检索结果太长？上下文窗口不够用？检索压缩技术帮你在保留关键信息的同时大幅减少token消耗！

📚 学习目标¶

学完本章后，你将能够：

预计学习时间：4小时 难度等级：⭐⭐⭐⭐☆

前置知识¶

完成第9章：混合检索与重排序
理解向量检索原理
熟悉LLM上下文限制
了解RAG工作流程

13.1 为什么需要检索压缩？¶

13.1.1 问题分析¶

场景1：长文档检索

问题: "Python的异步编程如何使用？"

传统检索:
├─ 检索到3个相关文档
├─ 每个文档2000 tokens
├─ 总共6000 tokens
└─ 上下文窗口紧张!

结果:
✗ 可能超出模型限制
✗ 大量无关信息
✗ 响应速度慢
✗ API成本高

场景2：冗余信息

检索结果:
文档1: "Python异步编程使用async/await语法..."
文档2: "Python的asyncio库提供了异步编程支持..."
文档3: "异步编程在Python中通过async def实现..."

问题:
- 三个文档都在说同一件事
- 大量重复信息
- 关键点被稀释

13.1.2 压缩的价值¶

压缩前:
┌────────────────────────────────┐
│ 文档1 (2000 tokens)            │
│ - 引言: 300 tokens             │
│ - 基础概念: 500 tokens          │
│ - 核心内容: 800 tokens          │
│ - 示例代码: 400 tokens          │
└────────────────────────────────┘
┌────────────────────────────────┐
│ 文档2 (2000 tokens)            │
│ - 类似结构...                  │
└────────────────────────────────┘
总计: 4000+ tokens

压缩后:
┌────────────────────────────────┐
│ 压缩摘要 (500 tokens)           │
│ - 合并核心概念                 │
│ - 去除冗余信息                 │
│ - 保留关键示例                 │
│ - 突出重点内容                 │
└────────────────────────────────┘
总计: 500 tokens (节省87.5%)

核心优势:

优势	说明
节省成本	减少token消耗 → 降低API调用成本
提升速度	更短的上下文 → 更快的生成速度
提高质量	去除噪声 → 聚焦关键信息
扩展容量	相同窗口 → 检索更多文档

13.2 压缩策略分类¶

13.2.1 四大压缩策略¶

检索压缩策略层次:

Level 1: 简单截断
├─ 固定长度截断
├─ 保留开头/结尾
└─ 成本: 低, 效果: 差

Level 2: 规则压缩
├─ 去除停用词
├─ 去除格式标记
├─ 去除冗余句子
└─ 成本: 中, 效果: 中

Level 3: 语义压缩 ⭐
├─ 保留关键信息
├─ 重述简化表达
├─ 多文档合并
└─ 成本: 中, 效果: 好

Level 4: 智能压缩 ⭐⭐
├─ LLM重写
├─ 段落级压缩
├─ 查询感知压缩
└─ 成本: 高, 效果: 最好

13.2.2 选择指南¶

场景	推荐策略	理由
实时问答	Level 2	速度快，质量可接受
技术文档	Level 3	保留技术细节
复杂推理	Level 4	需要完整语义
成本敏感	Level 2-3	平衡效果和成本

13.3 简单压缩技术¶

13.3.1 固定长度截断¶

# 文件名：simple_compression.py
"""
简单压缩方法
"""

def truncate_by_length(text: str, max_length: int) -> str:
    """
    按固定长度截断

    Args:
        text: 原始文本
        max_length: 最大长度(字符数)

    Returns:
        截断后的文本
    """
    if len(text) <= max_length:
        return text

    # 保留开头80%，结尾20%
    head_length = int(max_length * 0.8)
    tail_length = max_length - head_length

    head = text[:head_length]
    tail = text[-tail_length:] if tail_length > 0 else ""

    return f"{head}\n...\n{tail}"


# 示例
long_text = """
Python是一门高级编程语言，由Guido van Rossum于1991年创建。
Python以其简洁、易读的语法而闻名，被广泛应用于各种领域。

[此处省略2000字]

Python拥有丰富的生态系统，包括NumPy、Pandas、Django等流行库。
这使得Python成为数据科学、机器学习、Web开发等领域的首选语言。
"""

compressed = truncate_by_length(long_text, max_length=200)

print("压缩前:", len(long_text), "字符")
print("压缩后:", len(compressed), "字符")
print(f"压缩率: {len(compressed)/len(long_text)*100:.1f}%")

13.3.2 去除格式和冗余¶

import re

def clean_text(text: str) -> str:
    """
    清理文本：去除格式标记和冗余内容
    """
    # 去除多余空白
    text = re.sub(r'\s+', ' ', text)

    # 去除Markdown格式
    text = re.sub(r'#{1,6}\s', '', text)  # 标题
    text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)  # 粗体
    text = re.sub(r'\*(.*?)\*', r'\1', text)  # 斜体
    text = re.sub(r'`(.*?)`', r'\1', text)  # 代码

    # 去除HTML标签
    text = re.sub(r'<[^>]+>', '', text)

    # 去除特殊符号
    text = re.sub(r'[^\w\s\u4e00-\u9fff.,!?;:()\-\"\']', '', text)

    return text.strip()


def remove_redundant_sentences(text: str, similarity_threshold: float = 0.9):
    """
    去除冗余句子

    Args:
        text: 输入文本
        similarity_threshold: 相似度阈值
    """
    sentences = text.split('。')
    unique_sentences = []

    for sent in sentences:
        if not sent.strip():
            continue

        # 检查是否与已有句子过于相似
        is_redundant = False
        for existing in unique_sentences:
            # 简单的相似度检查（实际应该用更复杂的算法）
            if (sent.strip() in existing) or (existing in sent.strip()):
                is_redundant = True
                break

        if not is_redundant:
            unique_sentences.append(sent.strip())

    return '。'.join(unique_sentences) + '。'


# 示例
text_with_format = """
## Python简介

**Python**是一门*高级*编程语言，由Guido van Rossum创建。

Python以其简洁、易读的语法而闻名。
Python的语法非常简洁。
Python的代码很容易阅读。

Python被广泛应用于各种领域。
"""

cleaned = clean_text(text_with_format)
deduped = remove_redundant_sentences(cleaned)

print("原始文本:")
print(text_with_format)
print("\n清理去重后:")
print(deduped)

13.4 语义保留压缩 ⭐¶

13.4.1 原理¶

目标: 在大幅压缩的同时保留关键语义

语义压缩流程:

原文档 (2000 tokens)
   ↓
[提取关键信息]
   ├─ 识别核心概念
   ├─ 提取关键论点
   ├─ 保留重要示例
   └─ 去除修饰内容
   ↓
[重述简化]
   ├─ 保持原意
   ├─ 简化表达
   └─ 合并重复
   ↓
压缩文档 (500 tokens)
   └─ 保留80%+的关键信息

13.4.2 完整实现¶

# 文件名：semantic_compression.py
"""
语义保留压缩实现
"""

from typing import List, Dict
import json


class SemanticCompressor:
    """
    语义压缩器

    特点:
    1. 保留关键信息
    2. 简化表达
    3. 去除冗余
    4. 保持可读性
    """

    def __init__(self, llm):
        self.llm = llm

    def compress_single_document(self,
                                  document: str,
                                  compression_ratio: float = 0.25) -> str:
        """
        压缩单个文档

        Args:
            document: 原始文档
            compression_ratio: 压缩比例 (0-1), 默认0.25表示压缩到25%

        Returns:
            压缩后的文档
        """
        # 计算目标长度
        target_length = int(len(document) * compression_ratio)

        prompt = f"""请将以下文档压缩到{target_length}字符左右。

要求:
1. 保留所有关键信息
2. 去除冗余和修饰内容
3. 简化表达但保持原意
4. 保持逻辑结构清晰
5. 保留重要示例

原文档:
{document}

请输出压缩后的文档:"""

        compressed = self.llm.predict(prompt)
        return compressed

    def compress_multiple_documents(self,
                                     documents: List[str],
                                     query: str,
                                     max_total_length: int = 1000) -> str:
        """
        压缩并合并多个文档

        Args:
            documents: 文档列表
            query: 原始查询（用于相关性判断）
            max_total_length: 最大总长度

        Returns:
            合并压缩后的文档
        """
        print(f"压缩{len(documents)}个文档，目标长度: {max_total_length}")

        # 步骤1: 逐个压缩文档
        compressed_docs = []
        for i, doc in enumerate(documents, 1):
            print(f"  压缩文档 {i}/{len(documents)}...")

            # 为每个文档分配长度预算
            budget = max_total_length // len(documents)

            compressed = self._compress_with_context(
                document=doc,
                query=query,
                max_length=budget
            )
            compressed_docs.append(compressed)

        # 步骤2: 合并压缩后的文档
        merged = self._merge_compressed_documents(
            compressed_docs,
            query=query,
            max_length=max_total_length
        )

        return merged

    def _compress_with_context(self,
                                document: str,
                                query: str,
                                max_length: int) -> str:
        """
        基于查询上下文压缩文档
        """
        prompt = f"""用户查询: {query}

请将以下文档压缩到{max_length}字符以内，要求:
1. 优先保留与查询相关的内容
2. 保留关键细节和数据
3. 去除与查询无关的内容
4. 简化表达

文档:
{document}

压缩结果:"""

        return self.llm.predict(prompt)

    def _merge_compressed_documents(self,
                                     compressed_docs: List[str],
                                     query: str,
                                     max_length: int) -> str:
        """
        合并多个压缩文档
        """
        # 计算每个文档的平均长度
        avg_length = max_length // len(compressed_docs)

        # 再次压缩以确保总长度不超限
        final_docs = []
        for doc in compressed_docs:
            if len(doc) > avg_length:
                # 进一步压缩
                prompt = f"""请将以下内容进一步压缩到{avg_length}字符：

{doc}

要求: 保留最核心的信息"""
                doc = self.llm.predict(prompt)
            final_docs.append(doc)

        # 合并
        merged = "\n\n".join(final_docs)

        # 如果还是太长，整体压缩
        if len(merged) > max_length:
            prompt = f"""请将以下内容整体压缩到{max_length}字符：

{merged}

要求: 去除冗余，合并重复内容"""
            merged = self.llm.predict(prompt)

        return merged


# 使用示例
if __name__ == "__main__":
    # 模拟LLM
    class MockLLM:
        def predict(self, prompt):
            if "压缩到" in prompt:
                # 简单模拟压缩
                return "这是压缩后的内容。包含了原文档的关键信息，去除了冗余内容。"
            return "模拟响应"

    # 创建压缩器
    llm = MockLLM()
    compressor = SemanticCompressor(llm)

    # 示例文档
    docs = [
        "Python异步编程使用async/await语法。asyncio是Python的标准库...",
        "异步编程可以提高程序的并发性能。通过事件循环机制...",
        "Python的asyncio模块提供了丰富的事件循环API。包括Task、Future..."
    ]

    # 压缩并合并
    query = "Python异步编程如何使用？"
    result = compressor.compress_multiple_documents(
        documents=docs,
        query=query,
        max_total_length=300
    )

    print(f"压缩结果:\n{result}")

13.5 段落级压缩 ⭐⭐¶

13.5.1 什么是段落级压缩？¶

传统压缩 vs 段落级压缩:

传统压缩:
文档1 (2000 tokens) → LLM → 压缩后 (500 tokens)
文档2 (2000 tokens) → LLM → 压缩后 (500 tokens)
文档3 (2000 tokens) → LLM → 压缩后 (500 tokens)

问题:
- 每个文档独立压缩，可能丢失关联信息
- 需要多次LLM调用，成本高

段落级压缩:
文档1-3 (6000 tokens)
   ↓
[拆分段落]
   ├─ 段落1-1 (500 tokens)
   ├─ 段落1-2 (500 tokens)
   ├─ 段落2-1 (500 tokens)
   └─ ...
   ↓
[逐段压缩]
   ├─ 段落1-1' (200 tokens) ⭐ 保留
   ├─ 段落1-2' (200 tokens) ⭐ 保留
   ├─ 段落2-1' (200 tokens) ⊗ 丢弃(相关性低)
   └─ ...
   ↓
[智能合并]
最终文档 (1000 tokens)

13.5.2 完整实现¶

# 文件名：paragraph_compression.py
"""
段落级压缩实现
"""

from typing import List, Tuple
import re


class ParagraphCompressor:
    """
    段落级压缩器

    特点:
    1. 段落级别的细粒度控制
    2. 相关性筛选
    3. 智能合并
    4. 更好的可解释性
    """

    def __init__(self, llm):
        self.llm = llm

    def compress(self,
                 documents: List[str],
                 query: str,
                 target_length: int) -> Tuple[str, Dict]:
        """
        段落级压缩主方法

        Returns:
            (压缩后的文档, 统计信息)
        """
        stats = {
            "original_length": sum(len(d) for d in documents),
            "num_paragraphs": 0,
            "num_kept": 0,
            "num_discarded": 0,
            "compression_ratio": 0.0
        }

        # 步骤1: 拆分所有文档为段落
        paragraphs = self._split_into_paragraphs(documents)
        stats["num_paragraphs"] = len(paragraphs)
        print(f"拆分得到 {len(paragraphs)} 个段落")

        # 步骤2: 评估段落相关性
        scored_paragraphs = self._score_relevance(paragraphs, query)

        # 步骤3: 选择高相关性段落
        selected = self._select_paragraphs(
            scored_paragraphs,
            target_length=target_length
        )
        stats["num_kept"] = len(selected)
        stats["num_discarded"] = len(paragraphs) - len(selected)

        # 步骤4: 压缩选中的段落
        compressed_paragraphs = []
        for para, score in selected:
            compressed = self._compress_paragraph(para, query)
            compressed_paragraphs.append(compressed)

        # 步骤5: 合并
        final_doc = "\n\n".join(compressed_paragraphs)

        stats["final_length"] = len(final_doc)
        stats["compression_ratio"] = stats["final_length"] / stats["original_length"]

        return final_doc, stats

    def _split_into_paragraphs(self, documents: List[str]) -> List[str]:
        """
        将文档拆分为段落
        """
        paragraphs = []

        for doc in documents:
            # 按双换行符拆分
            doc_paragraphs = doc.split('\n\n')
            paragraphs.extend([p.strip() for p in doc_paragraphs if p.strip()])

        return paragraphs

    def _score_relevance(self,
                         paragraphs: List[str],
                         query: str) -> List[Tuple[str, float]]:
        """
        评估段落相关性

        Returns:
            [(段落, 相关性分数), ...]
        """
        scored = []

        for para in paragraphs:
            # 简单的相关性评估（实际应该用embedding或更复杂的方法）
            score = self._calculate_relevance(para, query)
            scored.append((para, score))

        # 按相关性排序
        scored.sort(key=lambda x: x[1], reverse=True)

        return scored

    def _calculate_relevance(self, paragraph: str, query: str) -> float:
        """
        计算段落与查询的相关性
        """
        # 简化版：基于关键词重叠
        query_words = set(query.lower().split())
        para_words = set(paragraph.lower().split())

        overlap = len(query_words & para_words)
        return min(overlap / len(query_words), 1.0) if query_words else 0.0

    def _select_paragraphs(self,
                            scored_paragraphs: List[Tuple[str, float]],
                            target_length: int) -> List[Tuple[str, float]]:
        """
        选择段落，确保总长度不超过目标
        """
        selected = []
        current_length = 0

        for para, score in scored_paragraphs:
            # 预估压缩后的长度（假设压缩到40%）
            estimated_length = len(para) * 0.4

            if current_length + estimated_length <= target_length:
                selected.append((para, score))
                current_length += estimated_length
            else:
                # 尝试截断当前段落
                remaining = target_length - current_length
                if remaining > 100:  # 至少保留100字符
                    truncated = para[:int(remaining * 2.5)]  # 反推原始长度
                    selected.append((truncated, score))
                break

        return selected

    def _compress_paragraph(self, paragraph: str, query: str) -> str:
        """
        压缩单个段落
        """
        # 目标：压缩到原始长度的40%
        target_length = max(len(paragraph) * 0.4, 50)

        prompt = f"""用户查询: {query}

请将以下段落压缩到{int(target_length)}字符左右。

要求:
1. 保留与查询相关的关键信息
2. 去除冗余和修饰
3. 保持语义完整

段落:
{paragraph}

压缩结果:"""

        return self.llm.predict(prompt)


# 使用示例
if __name__ == "__main__":
    class MockLLM:
        def predict(self, prompt):
            return "这是压缩后的段落，保留了关键信息。"

    llm = MockLLM()
    compressor = ParagraphCompressor(llm)

    docs = [
        """
        Python的异步编程是一个重要的特性。它允许程序在等待IO操作时执行其他任务。

        async/await语法是Python 3.5引入的，用于简化异步代码的编写。

        asyncio是Python的标准库，提供了事件循环、协程、Future等组件。
        """,

        """
        异步编程可以显著提高IO密集型应用的性能。

        在Web开发中，异步框架如FastAPI、aiohttp可以利用异步特性处理大量并发请求。

        异步编程的挑战包括：调试困难、回调地狱、并发控制等。
        """,

        """
        Python的异步生态系统包括很多第三方库。

        aiohttp用于异步HTTP请求，asyncpg用于异步数据库操作。

        选择合适的异步库对于构建高性能应用很重要。
        """
    ]

    query = "Python异步编程有哪些优势和挑战？"
    result, stats = compressor.compress(docs, query, target_length=500)

    print("压缩结果:")
    print(result)
    print("\n统计信息:")
    print(f"  原始长度: {stats['original_length']}")
    print(f"  最终长度: {stats['final_length']}")
    print(f"  段落总数: {stats['num_paragraphs']}")
    print(f"  保留: {stats['num_kept']}")
    print(f"  丢弃: {stats['num_discarded']}")
    print(f"  压缩率: {stats['compression_ratio']:.1%}")

13.6 压缩效果评估¶

13.6.1 评估指标¶

# 文件名：compression_evaluation.py
"""
压缩效果评估
"""

from typing import Dict, List
import math


class CompressionEvaluator:
    """
    压缩效果评估器
    """

    def evaluate(self,
                 original: str,
                 compressed: str,
                 query: str) -> Dict:
        """
        评估压缩效果

        Returns:
            评估结果字典
        """
        return {
            "compression_ratio": self._compression_ratio(original, compressed),
            "token_savings": self._token_savings(original, compressed),
            "semantic_similarity": self._semantic_similarity(original, compressed),
            "query_relevance": self._query_relevance(compressed, query),
            "readability": self._readability(compressed)
        }

    def _compression_ratio(self, original: str, compressed: str) -> float:
        """压缩比例"""
        return len(compressed) / len(original)

    def _token_savings(self, original: str, compressed: str) -> Dict:
        """Token节省"""
        # 简化估算：中文约1.5字符=1token，英文约4字符=1token
        original_tokens = len(original) / 2
        compressed_tokens = len(compressed) / 2

        return {
            "original_tokens": int(original_tokens),
            "compressed_tokens": int(compressed_tokens),
            "saved_tokens": int(original_tokens - compressed_tokens),
            "savings_ratio": (original_tokens - compressed_tokens) / original_tokens
        }

    def _semantic_similarity(self, original: str, compressed: str) -> float:
        """
        语义相似度（简化版）

        实际应该使用embedding计算余弦相似度
        """
        # 简化版：基于关键词重叠
        orig_words = set(original.lower().split())
        comp_words = set(compressed.lower().split())

        overlap = len(orig_words & comp_words)
        union = len(orig_words | comp_words)

        return overlap / union if union > 0 else 0.0

    def _query_relevance(self, compressed: str, query: str) -> float:
        """查询相关性"""
        query_words = set(query.lower().split())
        comp_words = set(compressed.lower().split())

        overlap = len(query_words & comp_words)
        return overlap / len(query_words) if query_words else 0.0

    def _readability(self, text: str) -> Dict:
        """可读性分析"""
        sentences = text.split('。')
        words = text.split()

        return {
            "avg_sentence_length": len(words) / len(sentences) if sentences else 0,
            "num_sentences": len(sentences),
            "num_words": len(words)
        }


# 使用示例
if __name__ == "__main__":
    original = "Python异步编程使用async/await语法。asyncio是Python的标准库，提供了事件循环机制。异步编程可以提高程序的并发性能，特别适合IO密集型任务。"
    compressed = "Python异步编程用async/await语法，通过asyncio的事件循环机制提升IO密集型任务的并发性能。"

    evaluator = CompressionEvaluator()
    results = evaluator.evaluate(original, compressed, "Python异步编程")

    print("压缩效果评估:")
    for metric, value in results.items():
        if isinstance(value, dict):
            print(f"\n{metric}:")
            for k, v in value.items():
                print(f"  {k}: {v}")
        else:
            print(f"{metric}: {value}")

13.7 实战案例¶

案例1：智能客服RAG系统¶

# 场景：用户问题检索到多个长文档
query = "如何退款？"
retrieved_docs = [
    "用户可以在订单详情页点击退款按钮...（1500字）",
    "退款流程包括提交申请、商家审核、平台处理...（1200字）",
    "退款规则：7天无理由退款、质量问题退款...（1000字）"
]

# 使用检索压缩
compressor = SemanticCompressor(llm)
compressed = compressor.compress_multiple_documents(
    documents=retrieved_docs,
    query=query,
    max_total_length=500
)

# 结果：从3700字压缩到500字，保留所有退款关键信息

案例2：技术文档问答¶

# 场景：开发者查询API用法
query = "FastAPI的Dependency Injection如何使用？"
retrieved_docs = [
    # 从FastAPI官方文档检索到的多个长章节
    "Dependencies Intro ...",
    "Dependencies Advanced ...",
    "Dependencies Security ..."
]

# 使用段落级压缩，保留代码示例
compressor = ParagraphCompressor(llm)
compressed, stats = compressor.compress(
    documents=retrieved_docs,
    query=query,
    target_length=800
)

# 结果：保留所有代码示例，压缩解释性文字

练习题¶

练习1：实现基础压缩器¶

题目：实现一个简单的文本压缩器

要求： 1. 支持固定长度截断 2. 去除格式标记 3. 去除冗余句子 4. 计算压缩比例

练习2：语义压缩实现¶

题目：实现语义保留压缩

要求： 1. 集成LLM进行智能压缩 2. 支持单文档压缩 3. 支持多文档合并压缩 4. 保留查询相关内容

练习3：压缩效果评估¶

题目：构建完整的评估系统

要求： 1. 实现多个评估指标 2. 对比不同压缩策略 3. 生成可视化报告 4. 提供优化建议

总结¶

本章要点¶

压缩价值
节省成本
提升速度
提高质量
扩展容量
压缩策略
简单截断
规则压缩
语义压缩
段落级压缩
实现方法
SemanticCompressor
ParagraphCompressor
效果评估

学习检查清单¶

下一步学习¶

恭喜完成第13章！ 🎉

掌握检索压缩，让RAG系统更高效！ → 第14章