三种主流压缩方式

在 my_ai_platform 中集成 EmbeddingsRedundantFilter 和 DocumentCompressorPipeline，是两种完全不依赖大模型、极速且零额外成本的上下文压缩方案。它们与 BGE Reranker 互补，你可以根据场景选择或组合使用。

🧩 一、EmbeddingsRedundantFilter（语义冗余过滤器）

📌 原理

它利用向量相似度，剔除文档中与其他句子高度重复的句子，保留信息密度更高的独特内容。这样可以减少冗余，压缩上下文长度。

📁 新增文件

创建 src/infrastructure/compressor/embeddings_filter.py：

# src/infrastructure/compressor/embeddings_filter.py
from typing import List, Optional
from langchain_core.documents import Document
from langchain.retrievers.document_compressors import EmbeddingsRedundantFilter
from langchain_core.embeddings import Embeddings

class EmbeddingsRedundantCompressor:
    """
    基于向量相似度的冗余过滤器：剔除文档中重复的句子。
    """
    def __init__(self, embeddings: Embeddings, similarity_threshold: float = 0.95, k: int = 20):
        """
        :param embeddings: 嵌入模型实例（如你的 SiliconFlowEmbeddings）
        :param similarity_threshold: 相似度阈值，高于此值的句子视为重复，默认0.95
        :param k: 每个文档最多保留的句子数，防止截断过少
        """
        self.filter = EmbeddingsRedundantFilter(
            embeddings=embeddings,
            similarity_threshold=similarity_threshold,
            k=k
        )

    def compress(self, documents: List[Document]) -> List[Document]:
        """
        对文档列表进行冗余过滤，返回压缩后的文档列表。
        """
        if not documents:
            return []
        # 该过滤器内部会拆分句子、计算向量、去重
        compressed = self.filter.compress_documents(documents)
        print(f"🧹 冗余过滤：原始 {len(documents)} 个文档，压缩后 {len(compressed)} 个文档")
        return compressed

🔗 二、DocumentCompressorPipeline（管道组合压缩器）

📌 原理

它是一个容器，可以按顺序组合多个压缩步骤（如截断、过滤、重排序等）。你可以将 EmbeddingsRedundantFilter 与其他压缩器（如长度截断）串联成一个流水线。

📁 新增文件

创建 src/infrastructure/compressor/pipeline.py：

# src/infrastructure/compressor/pipeline.py
from typing import List, Optional
from langchain_core.documents import Document
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain.retrievers.document_compressors import (
    LLMChainExtractor,      # 我们不推荐，但留作示例
    LLMChainFilter,
)
from langchain.text_splitter import CharacterTextSplitter
from langchain_core.callbacks import Callbacks

class PipelineCompressor:
    """
    组合多个压缩器，按顺序执行。
    """
    def __init__(self, compressors: List):
        """
        :param compressors: 压缩器列表，顺序执行
        """
        self.pipeline = DocumentCompressorPipeline(transformers=compressors)

    def compress(self, documents: List[Document], callbacks: Optional[Callbacks] = None) -> List[Document]:
        if not documents:
            return []
        compressed = self.pipeline.compress_documents(documents, callbacks=callbacks)
        print(f"🔧 管道压缩：原始 {len(documents)} 个文档，压缩后 {len(compressed)} 个文档")
        return compressed

🧪 如何在 main.py 中集成？

你可以选择以下两种方式之一，将压缩器注入到 RAGService 中。

方案 A：直接替换 ensemble_retriever 为压缩检索器（推荐）

在 main.py 中，用 ContextualCompressionRetriever 包装你的 ensemble_retriever，并传入压缩器。

# main.py（片段）
from langchain.retrievers import ContextualCompressionRetriever
from src.infrastructure.compressor.embeddings_filter import EmbeddingsRedundantCompressor
from src.infrastructure.compressor.pipeline import PipelineCompressor
from langchain.text_splitter import CharacterTextSplitter

# ... 已经创建好 embeddings, llm, ensemble_retriever ...

# 1. 创建压缩器实例（使用你的 embeddings 对象）
redundant_compressor = EmbeddingsRedundantCompressor(
    embeddings=embeddings,
    similarity_threshold=0.95,
    k=10
)

# 2. （可选）添加长度截断压缩器，控制总字符数
from langchain.retrievers.document_compressors import (
    LLMChainExtractor,   # 我们不用，用下面的方法
)

# 我们可以用文本分割器作为截断器
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
splitter = CharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separator="。"
)

# 组合管道：先冗余过滤，再截断
pipeline = PipelineCompressor(compressors=[
    redundant_compressor.filter,      # 注意这里传入的是内部的 .filter，或者我们直接构造两个压缩器
    # 但 DocumentCompressorPipeline 需要每个压缩器有 compress_documents 方法
])

# 更简单的方法：直接用 ContextualCompressionRetriever 包装一个压缩器
# 如果只想用冗余过滤：
compression_retriever = ContextualCompressionRetriever(
    base_compressor=redundant_compressor.filter,  # 直接传入 EmbeddingsRedundantFilter 实例
    base_retriever=ensemble_retriever
)

# 如果想组合多个压缩器，可以用 DocumentCompressorPipeline
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain.retrievers.document_compressors import (
    EmbeddingsRedundantFilter,
)

# 实例化两个压缩器
filter1 = EmbeddingsRedundantFilter(embeddings=embeddings, similarity_threshold=0.95, k=10)
# 第二个压缩器可以是自定义函数，但需有 compress_documents 方法
# 这里我们可以用字符截断，但 DocumentCompressorPipeline 需要 compress_documents 方法
# LangChain 没有内置字符截断压缩器，我们可以自己实现一个简单的
class TruncationCompressor:
    def __init__(self, max_chars=2000):
        self.max_chars = max_chars
    def compress_documents(self, documents, **kwargs):
        result = []
        total_chars = 0
        for doc in documents:
            remaining = self.max_chars - total_chars
            if remaining <= 0:
                break
            truncated = doc.page_content[:remaining]
            new_doc = Document(
                page_content=truncated,
                metadata=doc.metadata
            )
            result.append(new_doc)
            total_chars += len(truncated)
        return result

trunc_compressor = TruncationCompressor(max_chars=2000)

# 构建管道
pipeline = DocumentCompressorPipeline(
    transformers=[filter1, trunc_compressor]
)

# 最终压缩检索器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=pipeline,
    base_retriever=ensemble_retriever
)

# 然后把 compression_retriever 注入 RAGService
rag_service = RAGService(
    llm_client=llm,
    retriever=compression_retriever
)

为了保持简洁，我们可以把管道构建移到单独的函数中。

方案 B：在 RAGService.ask 内部手动调用压缩

如果你不想用 ContextualCompressionRetriever，也可以修改 RAGService，在检索后、生成前手动调用压缩器。

修改 rag_service.py：

def ask(self, query: str):
    # 检索
    docs = self.retriever.invoke(query)
    # 压缩（传入压缩器实例）
    if self.compressor:
        docs = self.compressor.compress(docs)
    # 生成...

然后在 main.py 中实例化压缩器并传入 RAGService：

compressor = EmbeddingsRedundantCompressor(embeddings=embeddings)
rag_service = RAGService(llm_client=llm, retriever=ensemble_retriever, compressor=compressor)

这种方式更加灵活，你可以随时切换压缩策略。

📦 最终建议：组合使用 重排序 + 冗余过滤 + 截断

在实际生产环境中，最佳实践是：

1. 检索（混合检索）→ 召回较多文档（如 10 个）
2. 重排序（BGE Reranker）→ 选出最相关的 Top 5
3. 冗余过滤（EmbeddingsRedundantFilter）→ 去除重复句子
4. 截断（TruncationCompressor）→ 控制总输入长度
5. 生成（大模型）

你可以用 DocumentCompressorPipeline 将步骤 3 和 4 组合成一个管道，再用 ContextualCompressionRetriever 包装。这样，你的 RAGService 无需改动，只需在 main.py 中配置即可。

现在，你拥有了完整的 “检索 → 改写 → 重排序 → 压缩 → 生成” 工业级 RAG 链路！🚀

本作品采用《CC 协议》，转载必须注明作者和本文链接