高级检索技术：多路召回与重排序（Rerank）

核心知识点讲解

为什么需要高级检索技术？

在传统的RAG系统中，单一的检索策略可能存在以下局限性：

1. 召回率不足：单一检索策略可能无法覆盖所有相关文档
2. 精度不高：初始检索结果可能包含较多无关内容
3. 语义理解有限：简单的向量相似度可能无法捕捉复杂的语义关系
4. 上下文依赖：缺乏对查询上下文的深度理解
5. 领域适应性差：在特定领域可能表现不佳

高级检索技术通过以下方式解决这些问题：

• 多路召回：结合多种检索策略，提高召回率
• 重排序：对初始检索结果进行精细排序，提高精度
• 语义增强：利用更高级的语义理解模型
• 上下文感知：考虑查询的上下文信息
• 领域适应：针对特定领域优化检索策略

多路召回策略

多路召回是指同时使用多种不同的检索策略，然后将结果合并的技术。常见的多路召回策略包括：

1. 基于向量的召回

• 语义召回：使用文本嵌入模型，基于语义相似度检索
• 多嵌入召回：使用多个不同的嵌入模型，综合不同角度的语义理解
• 层次化召回：先粗粒度召回，再细粒度筛选

2. 基于词法的召回

• 关键词召回：基于关键词匹配的传统检索
• 短语召回：考虑短语级别的匹配
• 布尔召回：使用布尔逻辑组合多个检索条件

3. 基于结构的召回

• 元数据召回：基于文档的元数据进行筛选
• 结构化召回：利用文档的结构化信息
• 图结构召回：基于知识图谱的关联检索

4. 混合召回策略

• 线性融合：对不同召回结果进行线性加权融合
• 轮次融合：先使用一种策略，再使用另一种策略优化
• 投票机制：基于多个策略的共识进行排序

重排序（Rerank）技术

重排序是指对初始检索结果进行再次排序，以提高结果质量的技术。常见的重排序技术包括：

1. 基于机器学习的重排序

• 点排序模型：对单个文档进行评分
• ** pairwise排序模型**：比较文档对的相对顺序
• listwise排序模型：直接优化整个结果列表的质量

2. 基于预训练模型的重排序

• Cross-Encoder：同时编码查询和文档，捕捉它们之间的交互
• ColBERT：上下文 Late Interaction 模型，高效捕捉细粒度匹配
• ERNIE-Search：百度开发的语义检索模型

3. 基于规则的重排序

• 新鲜度排序：优先考虑时间较新的文档
• 权威性排序：优先考虑权威来源的文档
• 相关性排序：基于多种相关性指标的综合评分

4. 混合重排序策略

• 特征融合：结合多种特征进行重排序
• 级联排序：使用多个排序模型级联处理
• 动态排序：根据查询类型动态调整排序策略

实用案例分析

案例1：多路召回实现

# 安装必要的库
# pip install langchain chromadb sentence-transformers

from langchain.vectorstores import Chroma, FAISS
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.docstore.document import Document
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.chat_models import ChatOpenAI
import os
import numpy as np

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 1. 准备测试文档
documents = [
    Document(
        page_content="人工智能（AI）是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。",
        metadata={"category": "ai", "type": "definition", "source": "book", "date": "2023-01-01"}
    ),
    Document(
        page_content="机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习并改进，而无需明确编程。",
        metadata={"category": "ai", "type": "branch", "source": "book", "date": "2023-01-01"}
    ),
    Document(
        page_content="深度学习是机器学习的一个分支，它使用多层神经网络来模拟人脑的学习过程。",
        metadata={"category": "ai", "type": "branch", "source": "book", "date": "2023-01-01"}
    ),
    Document(
        page_content="自然语言处理是人工智能的一个领域，它使计算机能够理解、解释和生成人类语言。",
        metadata={"category": "ai", "type": "field", "source": "book", "date": "2023-01-01"}
    ),
    Document(
        page_content="计算机视觉是人工智能的一个领域，它使计算机能够理解和解释图像和视频。",
        metadata={"category": "ai", "type": "field", "source": "book", "date": "2023-01-01"}
    ),
    Document(
        page_content="人工智能在医疗领域的应用包括疾病诊断、药物发现、个性化治疗等。",
        metadata={"category": "ai", "type": "application", "source": "article", "date": "2023-06-15"}
    ),
    Document(
        page_content="人工智能在金融领域的应用包括风险评估、 fraud detection、算法交易等。",
        metadata={"category": "ai", "type": "application", "source": "article", "date": "2023-07-20"}
    ),
    Document(
        page_content="大型语言模型（LLM）是人工智能的重要进展，能够生成连贯、自然的文本。",
        metadata={"category": "ai", "type": "model", "source": "article", "date": "2023-12-01"}
    )
]

# 2. 创建多个向量库（不同嵌入模型）
print("=== 创建多个向量库 ===")

# 嵌入模型1：OpenAI
try:
    openai_embeddings = OpenAIEmbeddings()
    vectorstore_openai = Chroma.from_documents(
        documents=documents,
        embedding=openai_embeddings,
        persist_directory="./chroma_openai"
    )
    vectorstore_openai.persist()
    print("OpenAI向量库创建成功")
except Exception as e:
    print(f"OpenAI向量库创建失败: {str(e)}")
    vectorstore_openai = None

# 嵌入模型2：HuggingFace (all-MiniLM-L6-v2)
hf_embeddings_mini = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vectorstore_hf_mini = Chroma.from_documents(
    documents=documents,
    embedding=hf_embeddings_mini,
    persist_directory="./chroma_hf_mini"
)
vectorstore_hf_mini.persist()
print("HuggingFace (MiniLM)向量库创建成功")

# 嵌入模型3：HuggingFace (all-mpnet-base-v2)
hf_embeddings_mpnet = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2"
)
vectorstore_hf_mpnet = Chroma.from_documents(
    documents=documents,
    embedding=hf_embeddings_mpnet,
    persist_directory="./chroma_hf_mpnet"
)
vectorstore_hf_mpnet.persist()
print("HuggingFace (MPNet)向量库创建成功")

# 3. 创建多路召回检索器
print("\n=== 创建多路召回检索器 ===")

# 创建多个基础检索器
retrievers = []

# OpenAI检索器
if vectorstore_openai:
    retriever_openai = vectorstore_openai.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3}
    )
    retrievers.append(retriever_openai)

# HuggingFace (MiniLM)检索器
retriever_hf_mini = vectorstore_hf_mini.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)
retrievers.append(retriever_hf_mini)

# HuggingFace (MPNet)检索器
retriever_hf_mpnet = vectorstore_hf_mpnet.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)
retrievers.append(retriever_hf_mpnet)

# 创建多查询检索器（基于同一向量库的不同查询表述）
llm = ChatOpenAI(temperature=0)
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever_hf_mpnet,
    llm=llm
)
retrievers.append(multi_query_retriever)

# 4. 集成多路召回
print("\n=== 集成多路召回 ===")

# 方法1：使用EnsembleRetriever
ensemble_retriever = EnsembleRetriever(
    retrievers=retrievers,
    weights=[0.3, 0.2, 0.3, 0.2]  # 权重可以根据实际效果调整
)

# 方法2：自定义多路召回
class CustomMultiRetriever:
    def __init__(self, retrievers, weights=None):
        self.retrievers = retrievers
        self.weights = weights if weights else [1.0/len(retrievers)] * len(retrievers)
    
    def get_relevant_documents(self, query):
        """获取相关文档"""
        # 从每个检索器获取结果
        all_results = []
        for i, retriever in enumerate(self.retrievers):
            results = retriever.get_relevant_documents(query)
            # 为每个结果添加来源和权重
            for doc in results:
                if not hasattr(doc, "metadata"):
                    doc.metadata = {}
                doc.metadata["retriever_index"] = i
                doc.metadata["retriever_weight"] = self.weights[i]
                all_results.append(doc)
        
        # 去重
        unique_docs = {}
        for doc in all_results:
            content = doc.page_content
            if content not in unique_docs:
                unique_docs[content] = doc
        
        # 合并权重
        for content, doc in unique_docs.items():
            # 计算文档的总权重
            total_weight = sum(
                self.weights[r_idx] for r_idx in 
                {d.metadata.get("retriever_index") for d in all_results if d.page_content == content}
            )
            doc.metadata["total_weight"] = total_weight
        
        # 按总权重排序
        sorted_docs = sorted(unique_docs.values(), 
                           key=lambda x: x.metadata.get("total_weight", 0), 
                           reverse=True)
        
        return sorted_docs[:5]  # 返回前5个结果

# 创建自定义多路召回检索器
custom_multi_retriever = CustomMultiRetriever(
    retrievers=retrievers,
    weights=[0.3, 0.2, 0.3, 0.2]
)

# 5. 测试多路召回
print("\n=== 测试多路召回 ===")
query = "人工智能在医疗和金融领域的应用"

# 测试单一检索器
print("\n=== 单一检索器结果 ===")
if vectorstore_openai:
    openai_results = retriever_openai.get_relevant_documents(query)
    print("OpenAI检索器结果:")
    for i, doc in enumerate(openai_results):
        print(f"{i+1}. {doc.page_content}")

# 测试集成检索器
print("\n=== 集成检索器结果 ===")
ensemble_results = ensemble_retriever.get_relevant_documents(query)
print("EnsembleRetriever结果:")
for i, doc in enumerate(ensemble_results):
    print(f"{i+1}. {doc.page_content}")

# 测试自定义多路召回
print("\n=== 自定义多路召回结果 ===")
custom_results = custom_multi_retriever.get_relevant_documents(query)
print("CustomMultiRetriever结果:")
for i, doc in enumerate(custom_results):
    print(f"{i+1}. {doc.page_content} (权重: {doc.metadata.get('total_weight', 0):.2f})")

print("\n多路召回测试完成！")

案例2：重排序（Rerank）技术实现

# 安装必要的库
# pip install langchain chromadb sentence-transformers rank_bm25

from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.docstore.document import Document
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from rank_bm25 import BM25Okapi
import nltk
from nltk.tokenize import word_tokenize
import numpy as np

# 下载nltk数据
nltk.download('punkt', quiet=True)

# 1. 准备测试文档
documents = [
    Document(
        page_content="人工智能（AI）是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。",
        metadata={"category": "ai", "type": "definition"}
    ),
    Document(
        page_content="机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习并改进，而无需明确编程。",
        metadata={"category": "ai", "type": "branch"}
    ),
    Document(
        page_content="深度学习是机器学习的一个分支，它使用多层神经网络来模拟人脑的学习过程。",
        metadata={"category": "ai", "type": "branch"}
    ),
    Document(
        page_content="自然语言处理是人工智能的一个领域，它使计算机能够理解、解释和生成人类语言。",
        metadata={"category": "ai", "type": "field"}
    ),
    Document(
        page_content="计算机视觉是人工智能的一个领域，它使计算机能够理解和解释图像和视频。",
        metadata={"category": "ai", "type": "field"}
    ),
    Document(
        page_content="人工智能在医疗领域的应用包括疾病诊断、药物发现、个性化治疗等。",
        metadata={"category": "ai", "type": "application"}
    ),
    Document(
        page_content="人工智能在金融领域的应用包括风险评估、fraud detection、算法交易等。",
        metadata={"category": "ai", "type": "application"}
    ),
    Document(
        page_content="大型语言模型（LLM）是人工智能的重要进展，能够生成连贯、自然的文本。",
        metadata={"category": "ai", "type": "model"}
    )
]

# 2. 创建向量库
print("=== 创建向量库 ===")
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_rerank"
)
vectorstore.persist()

# 3. 创建基础检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}  # 召回更多结果用于重排序
)

# 4. 实现不同的重排序方法
print("\n=== 实现重排序方法 ===")

# 方法1：基于Cross-Encoder的重排序
print("\n方法1：基于Cross-Encoder的重排序")
try:
    # 创建Cross-Encoder重排序器
    cross_encoder_reranker = CrossEncoderReranker(
        model="cross-encoder/ms-marco-MiniLM-L-6-v2",
        top_n=3
    )
    
    # 创建压缩检索器
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=cross_encoder_reranker,
        base_retriever=retriever
    )
    
    print("Cross-Encoder重排序器创建成功")
except Exception as e:
    print(f"Cross-Encoder重排序器创建失败: {str(e)}")
    compression_retriever = None

# 方法2：基于BM25的重排序
print("\n方法2：基于BM25的重排序")
class BM25Reranker:
    def __init__(self, documents):
        # 预处理文档
        self.documents = documents
        self.corpus = [doc.page_content for doc in documents]
        # 分词
        self.tokenized_corpus = [word_tokenize(doc.lower()) for doc in self.corpus]
        # 初始化BM25
        self.bm25 = BM25Okapi(self.tokenized_corpus)
    
    def rerank(self, query, docs, top_n=3):
        """重排序文档"""
        # 分词查询
        tokenized_query = word_tokenize(query.lower())
        # 计算BM25分数
        scores = self.bm25.get_scores(tokenized_query)
        # 按分数排序
        sorted_indices = np.argsort(scores)[::-1]
        # 返回排序后的文档
        sorted_docs = [self.documents[i] for i in sorted_indices if self.documents[i] in docs]
        return sorted_docs[:top_n]

# 创建BM25重排序器
bm25_reranker = BM25Reranker(documents)
print("BM25重排序器创建成功")

# 方法3：基于规则的重排序
print("\n方法3：基于规则的重排序")
class RuleBasedReranker:
    def __init__(self):
        pass
    
    def rerank(self, query, docs, top_n=3):
        """基于规则重排序文档"""
        # 规则1：包含查询关键词的文档优先
        query_keywords = set(word_tokenize(query.lower()))
        
        def score(doc):
            doc_text = doc.page_content.lower()
            doc_keywords = set(word_tokenize(doc_text))
            # 计算关键词匹配数
            keyword_match = len(query_keywords.intersection(doc_keywords))
            # 计算文档长度（较短的文档优先）
            length_score = 1.0 / (1.0 + len(doc_text) / 100)
            # 综合评分
            return keyword_match * 0.7 + length_score * 0.3
        
        # 按规则评分排序
        sorted_docs = sorted(docs, key=score, reverse=True)
        return sorted_docs[:top_n]

# 创建规则重排序器
rule_reranker = RuleBasedReranker()
print("规则重排序器创建成功")

# 5. 测试重排序
print("\n=== 测试重排序 ===")
query = "人工智能在医疗领域的应用"

# 获取初始检索结果
initial_results = retriever.get_relevant_documents(query)
print("\n初始检索结果:")
for i, doc in enumerate(initial_results):
    print(f"{i+1}. {doc.page_content}")

# 测试Cross-Encoder重排序
if compression_retriever:
    print("\nCross-Encoder重排序结果:")
    cross_encoder_results = compression_retriever.get_relevant_documents(query)
    for i, doc in enumerate(cross_encoder_results):
        print(f"{i+1}. {doc.page_content}")

# 测试BM25重排序
print("\nBM25重排序结果:")
bm25_results = bm25_reranker.rerank(query, initial_results)
for i, doc in enumerate(bm25_results):
    print(f"{i+1}. {doc.page_content}")

# 测试规则重排序
print("\n规则重排序结果:")
rule_results = rule_reranker.rerank(query, initial_results)
for i, doc in enumerate(rule_results):
    print(f"{i+1}. {doc.page_content}")

# 6. 集成多路召回和重排序
print("\n=== 集成多路召回和重排序 ===")

class MultiRetrieverWithRerank:
    def __init__(self, retrievers, reranker):
        self.retrievers = retrievers
        self.reranker = reranker
    
    def get_relevant_documents(self, query):
        """获取相关文档"""
        # 多路召回
        all_docs = []
        for retriever in self.retrievers:
            docs = retriever.get_relevant_documents(query)
            all_docs.extend(docs)
        
        # 去重
        unique_docs = {}
        for doc in all_docs:
            content = doc.page_content
            if content not in unique_docs:
                unique_docs[content] = doc
        
        # 重排序
        reranked_docs = self.reranker.rerank(query, list(unique_docs.values()))
        
        return reranked_docs

# 创建多路召回+重排序检索器
if compression_retriever:
    multi_rerank_retriever = MultiRetrieverWithRerank(
        retrievers=[retriever],  # 这里可以添加多个检索器
        reranker=cross_encoder_reranker
    )
    
    # 测试集成效果
    print("\n多路召回+重排序结果:")
    final_results = multi_rerank_retriever.get_relevant_documents(query)
    for i, doc in enumerate(final_results):
        print(f"{i+1}. {doc.page_content}")

print("\n重排序测试完成！")

案例3：高级检索在RAG系统中的应用

from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.chat_models import ChatOpenAI
import os

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 1. 准备文档
from langchain.docstore.document import Document

documents = [
    Document(
        page_content="人工智能（AI）是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。人工智能的发展可以分为符号主义、连接主义和行为主义三个主要流派。",
        metadata={"category": "ai", "type": "definition"}
    ),
    Document(
        page_content="机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习并改进，而无需明确编程。机器学习主要包括监督学习、无监督学习、半监督学习和强化学习等方法。",
        metadata={"category": "ai", "type": "branch"}
    ),
    Document(
        page_content="深度学习是机器学习的一个分支，它使用多层神经网络来模拟人脑的学习过程。深度学习在计算机视觉、自然语言处理、语音识别等领域取得了重大突破。",
        metadata={"category": "ai", "type": "branch"}
    ),
    Document(
        page_content="自然语言处理（NLP）是人工智能的一个领域，它使计算机能够理解、解释和生成人类语言。NLP的主要任务包括分词、词性标注、命名实体识别、情感分析、机器翻译等。",
        metadata={"category": "ai", "type": "field"}
    ),
    Document(
        page_content="计算机视觉是人工智能的一个领域，它使计算机能够理解和解释图像和视频。计算机视觉的主要任务包括图像分类、目标检测、语义分割、人脸识别等。",
        metadata={"category": "ai", "type": "field"}
    ),
    Document(
        page_content="人工智能在医疗领域的应用包括疾病诊断、药物发现、个性化治疗、医疗影像分析等。AI辅助诊断系统可以帮助医生提高诊断准确率和效率。",
        metadata={"category": "ai", "type": "application"}
    ),
    Document(
        page_content="人工智能在金融领域的应用包括风险评估、欺诈检测、算法交易、客户服务等。AI系统可以分析大量金融数据，识别潜在风险和机会。",
        metadata={"category": "ai", "type": "application"}
    ),
    Document(
        page_content="大型语言模型（LLM）是人工智能的重要进展，能够生成连贯、自然的文本。代表性的LLM包括GPT系列、BERT、LLaMA等。这些模型在问答、文本生成、摘要等任务中表现出色。",
        metadata={"category": "ai", "type": "model"}
    ),
    Document(
        page_content="人工智能的伦理问题包括隐私保护、算法偏见、就业影响、安全风险等。随着AI技术的发展，如何确保AI的负责任使用成为重要议题。",
        metadata={"category": "ai", "type": "ethics"}
    ),
    Document(
        page_content="人工智能的未来发展趋势包括多模态学习、自主智能体、联邦学习、小样本学习等。这些技术将推动AI向更智能、更安全、更可靠的方向发展。",
        metadata={"category": "ai", "type": "future"}
    )
]

# 2. 创建向量库
print("=== 创建向量库 ===")
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_rag"
)
vectorstore.persist()

# 3. 创建高级检索器
print("\n=== 创建高级检索器 ===")

# 基础检索器
base_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# 选项1：多查询检索器
print("\n选项1：多查询检索器")
llm = ChatOpenAI(temperature=0)
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=llm,
    num_queries=3  # 生成3个不同的查询表述
)

# 选项2：重排序检索器
print("\n选项2：重排序检索器")
try:
    cross_encoder_reranker = CrossEncoderReranker(
        model="cross-encoder/ms-marco-MiniLM-L-6-v2",
        top_n=3
    )
    
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=cross_encoder_reranker,
        base_retriever=base_retriever
    )
    print("重排序检索器创建成功")
except Exception as e:
    print(f"重排序检索器创建失败: {str(e)}")
    compression_retriever = None

# 选项3：多查询+重排序检索器
print("\n选项3：多查询+重排序检索器")
try:
    multi_query_compression_retriever = ContextualCompressionRetriever(
        base_compressor=cross_encoder_reranker,
        base_retriever=multi_query_retriever
    )
    print("多查询+重排序检索器创建成功")
except Exception as e:
    print(f"多查询+重排序检索器创建失败: {str(e)}")
    multi_query_compression_retriever = None

# 4. 构建RAG链
print("\n=== 构建RAG链 ===")
llm_qa = OpenAI(temperature=0.7)

# 基础RAG链
base_qa_chain = RetrievalQA.from_chain_type(
    llm=llm_qa,
    chain_type="stuff",
    retriever=base_retriever,
    return_source_documents=True,
    verbose=True
)

# 多查询RAG链
multi_query_qa_chain = RetrievalQA.from_chain_type(
    llm=llm_qa,
    chain_type="stuff",
    retriever=multi_query_retriever,
    return_source_documents=True,
    verbose=True
)

# 重排序RAG链
if compression_retriever:
    compression_qa_chain = RetrievalQA.from_chain_type(
        llm=llm_qa,
        chain_type="stuff",
        retriever=compression_retriever,
        return_source_documents=True,
        verbose=True
    )

# 多查询+重排序RAG链
if multi_query_compression_retriever:
    multi_query_compression_qa_chain = RetrievalQA.from_chain_type(
        llm=llm_qa,
        chain_type="stuff",
        retriever=multi_query_compression_retriever,
        return_source_documents=True,
        verbose=True
    )

# 5. 测试不同RAG链
print("\n=== 测试不同RAG链 ===")
queries = [
    "人工智能在医疗和金融领域有哪些应用？",
    "大型语言模型的特点和应用场景是什么？",
    "人工智能的伦理问题和未来发展趋势？"
]

for query in queries:
    print(f"\n\n=== 测试查询: {query} ===")
    
    # 测试基础RAG链
    print("\n1. 基础RAG链结果:")
    base_result = base_qa_chain(query)
    print(f"回答: {base_result['result']}")
    print(f"源文档数量: {len(base_result['source_documents'])}")
    
    # 测试多查询RAG链
    print("\n2. 多查询RAG链结果:")
    multi_query_result = multi_query_qa_chain(query)
    print(f"回答: {multi_query_result['result']}")
    print(f"源文档数量: {len(multi_query_result['source_documents'])}")
    
    # 测试重排序RAG链
    if compression_retriever:
        print("\n3. 重排序RAG链结果:")
        compression_result = compression_qa_chain(query)
        print(f"回答: {compression_result['result']}")
        print(f"源文档数量: {len(compression_result['source_documents'])}")
    
    # 测试多查询+重排序RAG链
    if multi_query_compression_retriever:
        print("\n4. 多查询+重排序RAG链结果:")
        multi_query_compression_result = multi_query_compression_qa_chain(query)
        print(f"回答: {multi_query_compression_result['result']}")
        print(f"源文档数量: {len(multi_query_compression_result['source_documents'])}")

print("\n高级检索在RAG系统中的应用测试完成！")

代码解析

多路召回实现

1. 多向量库创建：

   • 使用不同的嵌入模型创建多个向量库
   • 包括OpenAI、HuggingFace MiniLM和MPNet
   • 每个向量库从不同角度捕捉文本的语义信息

2. 多路召回集成：

   • 使用EnsembleRetriever进行线性加权融合
   • 实现自定义多路召回，包括去重和权重合并
   • 测试不同召回策略的效果

3. 结果分析：

   • 比较单一检索器和多路召回的结果
   • 分析不同权重配置的影响
   • 验证多路召回的优势

重排序技术实现

1. 多种重排序方法：

   • Cross-Encoder重排序：利用预训练的交叉编码器模型
   • BM25重排序：基于词频的传统信息检索模型
   • 规则重排序：基于自定义规则的排序策略

2. 重排序流程：

   • 首先进行初始召回，获取足够多的候选文档
   • 然后使用重排序模型对候选文档进行精细排序
   • 最后返回排序后的Top-N结果

3. 效果评估：

   • 比较初始检索和重排序后的结果
   • 分析不同重排序方法的优缺点
   • 验证重排序对检索精度的提升

高级检索在RAG系统中的应用

1. 高级检索器集成：

   • 多查询检索器：生成多个查询表述，提高召回率
   • 重排序检索器：对初始结果进行精细排序，提高精度
   • 多查询+重排序检索器：结合两者的优势

2. RAG链构建：

   • 基于不同检索器构建多个RAG链
   • 比较不同RAG链的回答质量
   • 分析高级检索对RAG系统的提升

3. 实际应用测试：

   • 使用复杂查询测试不同RAG链
   • 评估回答的准确性、完整性和相关性
   • 验证高级检索技术在实际应用中的价值

高级技巧

1. 检索策略自适应

根据查询类型和上下文，自动选择合适的检索策略：

class AdaptiveRetriever:
    def __init__(self, retrievers):
        self.retrievers = retrievers
    
    def classify_query(self, query):
        """分类查询类型"""
        # 简单的查询分类逻辑
        query_lower = query.lower()
        
        if any(keyword in query_lower for keyword in ["什么是", "定义", "概念"]):
            return "definition"  # 定义类查询
        elif any(keyword in query_lower for keyword in ["如何", "怎样", "方法"]):
            return "how_to"  # 方法类查询
        elif any(keyword in query_lower for keyword in ["应用", "使用", "案例"]):
            return "application"  # 应用类查询
        elif any(keyword in query_lower for keyword in ["比较", "区别", "对比"]):
            return "comparison"  # 比较类查询
        else:
            return "general"  # 通用查询
    
    def get_relevant_documents(self, query):
        """根据查询类型选择合适的检索器"""
        query_type = self.classify_query(query)
        print(f"查询类型: {query_type}")
        
        # 根据查询类型选择检索器
        if query_type == "definition":
            # 定义类查询，使用语义理解能力强的检索器
            return self.retrievers[0].get_relevant_documents(query)
        elif query_type == "how_to":
            # 方法类查询，使用多查询检索器
            return self.retrievers[1].get_relevant_documents(query)
        elif query_type == "application":
            # 应用类查询，使用重排序检索器
            return self.retrievers[2].get_relevant_documents(query)
        elif query_type == "comparison":
            # 比较类查询，使用多路召回
            return self.retrievers[3].get_relevant_documents(query)
        else:
            # 通用查询，使用默认检索器
            return self.retrievers[0].get_relevant_documents(query)

# 使用自适应检索器
adaptive_retriever = AdaptiveRetriever([
    base_retriever,
    multi_query_retriever,
    compression_retriever,
    ensemble_retriever
])

# 测试自适应检索
print("\n=== 测试自适应检索 ===")
test_queries = [
    "什么是人工智能？",
    "如何使用人工智能进行图像识别？",
    "人工智能在医疗领域的应用有哪些？",
    "深度学习和机器学习的区别是什么？"
]

for query in test_queries:
    print(f"\n查询: {query}")
    results = adaptive_retriever.get_relevant_documents(query)
    for i, doc in enumerate(results[:2]):
        print(f"结果 {i+1}: {doc.page_content}")

2. 动态权重调整

根据检索结果的质量，动态调整不同检索策略的权重：

class DynamicWeightRetriever:
    def __init__(self, retrievers, initial_weights=None):
        self.retrievers = retrievers
        self.weights = initial_weights if initial_weights else [1.0/len(retrievers)] * len(retrievers)
    
    def evaluate_retriever_performance(self, query, results, gold_documents):
        """评估检索器性能"""
        # 简单的性能评估：计算返回结果中相关文档的比例
        relevant_count = 0
        gold_contents = [doc.page_content for doc in gold_documents]
        
        for result in results:
            if result.page_content in gold_contents:
                relevant_count += 1
        
        precision = relevant_count / len(results) if results else 0
        return precision
    
    def update_weights(self, query, all_results, gold_documents):
        """根据性能更新权重"""
        # 评估每个检索器的性能
        performances = []
        for i, retriever in enumerate(self.retrievers):
            # 获取该检索器的结果
            retriever_results = [r for r in all_results if r.metadata.get("retriever_index") == i]
            # 评估性能
            performance = self.evaluate_retriever_performance(query, retriever_results, gold_documents)
            performances.append(performance)
        
        # 计算新权重
        total_performance = sum(performances)
        if total_performance > 0:
            new_weights = [p / total_performance for p in performances]
            self.weights = new_weights
            print(f"更新权重: {new_weights}")
        
        return self.weights
    
    def get_relevant_documents(self, query, gold_documents=None):
        """获取相关文档"""
        # 多路召回
        all_results = []
        for i, retriever in enumerate(self.retrievers):
            results = retriever.get_relevant_documents(query)
            for doc in results:
                doc.metadata["retriever_index"] = i
                doc.metadata["weight"] = self.weights[i]
                all_results.append(doc)
        
        # 去重
        unique_docs = {}
        for doc in all_results:
            content = doc.page_content
            if content not in unique_docs:
                unique_docs[content] = doc
        
        # 计算综合得分
        for content, doc in unique_docs.items():
            # 找到所有包含该文档的检索结果
            doc_occurrences = [r for r in all_results if r.page_content == content]
            # 计算加权得分
            weighted_score = sum(r.metadata["weight"] for r in doc_occurrences)
            doc.metadata["weighted_score"] = weighted_score
        
        # 排序
        sorted_docs = sorted(
            unique_docs.values(),
            key=lambda x: x.metadata.get("weighted_score", 0),
            reverse=True
        )
        
        # 如果有 gold documents，更新权重
        if gold_documents:
            self.update_weights(query, all_results, gold_documents)
        
        return sorted_docs[:5]

# 使用动态权重检索器
dynamic_retriever = DynamicWeightRetriever(
    retrievers=retrievers,
    initial_weights=[0.25, 0.25, 0.25, 0.25]
)

# 测试动态权重调整
print("\n=== 测试动态权重调整 ===")
query = "人工智能在医疗领域的应用"
# 假设的相关文档
gold_docs = [documents[5]]  # 医疗应用文档

results = dynamic_retriever.get_relevant_documents(query, gold_documents=gold_docs)
print("\n动态权重调整结果:")
for i, doc in enumerate(results):
    print(f"{i+1}. {doc.page_content} (得分: {doc.metadata.get('weighted_score', 0):.2f})")

3. 上下文感知检索

考虑查询的上下文信息，提高检索的相关性：

class ContextAwareRetriever:
    def __init__(self, base_retriever, max_context_size=3):
        self.base_retriever = base_retriever
        self.context = []
        self.max_context_size = max_context_size
    
    def add_to_context(self, query, response):
        """添加到上下文"""
        self.context.append((query, response))
        # 保持上下文大小
        if len(self.context) > self.max_context_size:
            self.context = self.context[-self.max_context_size:]
    
    def get_contextual_query(self, query):
        """构建上下文感知的查询"""
        if not self.context:
            return query
        
        # 构建上下文
        context_str = "\n".join([f"Q: {q}\nA: {a}" for q, a in self.context])
        
        # 构建增强查询
        contextual_query = f"基于以下对话历史，回答问题:\n\n{context_str}\n\n当前问题: {query}"
        
        return contextual_query
    
    def get_relevant_documents(self, query):
        """获取相关文档"""
        # 构建上下文感知查询
        contextual_query = self.get_contextual_query(query)
        print(f"上下文感知查询: {contextual_query[:100]}...")
        
        # 检索
        results = self.base_retriever.get_relevant_documents(contextual_query)
        
        return results

# 使用上下文感知检索器
context_aware_retriever = ContextAwareRetriever(base_retriever)

# 测试上下文感知检索
print("\n=== 测试上下文感知检索 ===")

# 第一轮对话
query1 = "什么是人工智能？"
print(f"\n第一轮查询: {query1}")
results1 = context_aware_retriever.get_relevant_documents(query1)
response1 = "人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。"
context_aware_retriever.add_to_context(query1, response1)

# 第二轮对话
query2 = "它有哪些应用领域？"
print(f"\n第二轮查询: {query2}")
results2 = context_aware_retriever.get_relevant_documents(query2)
response2 = "人工智能的应用领域包括医疗、金融、教育、交通、制造等。"
context_aware_retriever.add_to_context(query2, response2)

# 第三轮对话
query3 = "在医疗领域具体有哪些应用？"
print(f"\n第三轮查询: {query3}")
results3 = context_aware_retriever.get_relevant_documents(query3)

print("\n上下文感知检索结果:")
for i, doc in enumerate(results3):
    print(f"{i+1}. {doc.page_content}")

4. 领域特定检索优化

针对特定领域优化检索策略：

class DomainSpecificRetriever:
    def __init__(self, domain, base_retriever):
        self.domain = domain
        self.base_retriever = base_retriever
        # 领域特定的关键词和权重
        self.domain_keywords = {
            "medical": ["医疗", "疾病", "诊断", "药物", "治疗", "健康"],
            "finance": ["金融", "投资", "风险", "交易", "银行", "保险"],
            "technology": ["技术", "算法", "系统", "软件", "硬件", "网络"],
            "education": ["教育", "学习", "教学", "课程", "培训", "知识"]
        }
    
    def enhance_query(self, query):
        """增强查询"""
        # 添加领域特定的关键词
        if self.domain in self.domain_keywords:
            keywords = self.domain_keywords[self.domain]
            # 检查查询是否包含领域关键词
            has_domain_keyword = any(keyword in query for keyword in keywords)
            if not has_domain_keyword:
                # 添加领域关键词到查询
                enhanced_query = f"{query} {self.domain}"
                print(f"增强查询: {enhanced_query}")
                return enhanced_query
        return query
    
    def filter_results(self, results):
        """过滤结果"""
        if self.domain not in self.domain_keywords:
            return results
        
        keywords = self.domain_keywords[self.domain]
        # 过滤出包含领域关键词的结果
        filtered_results = []
        for doc in results:
            content = doc.page_content
            if any(keyword in content for keyword in keywords):
                filtered_results.append(doc)
        
        # 如果过滤后结果太少，返回原始结果
        return filtered_results if len(filtered_results) >= 2 else results
    
    def get_relevant_documents(self, query):
        """获取相关文档"""
        # 增强查询
        enhanced_query = self.enhance_query(query)
        # 检索
        results = self.base_retriever.get_relevant_documents(enhanced_query)
        # 过滤结果
        filtered_results = self.filter_results(results)
        return filtered_results

# 使用领域特定检索器
medical_retriever = DomainSpecificRetriever(
    domain="medical",
    base_retriever=base_retriever
)

# 测试领域特定检索
print("\n=== 测试领域特定检索 ===")
query = "人工智能的应用"
print(f"原始查询: {query}")

# 普通检索
normal_results = base_retriever.get_relevant_documents(query)
print("\n普通检索结果:")
for i, doc in enumerate(normal_results[:3]):
    print(f"{i+1}. {doc.page_content}")

# 医疗领域检索
medical_results = medical_retriever.get_relevant_documents(query)
print("\n医疗领域检索结果:")
for i, doc in enumerate(medical_results[:3]):
    print(f"{i+1}. {doc.page_content}")

最佳实践

1. 检索策略选择

2. 性能优化

• 召回与排序分离：先快速召回足够多的结果，再精细排序
• 批量处理：批量执行嵌入和检索操作
• 缓存机制：缓存频繁查询的结果
• 异步处理：使用异步方式执行检索操作
• 硬件加速：使用GPU加速嵌入和重排序

3. 常见问题与解决方案

4. 部署建议

• 开发环境：

  • 使用轻量级模型进行快速测试
  • 启用详细日志便于调试
  • 尝试多种检索策略，找到最适合的方案

• 生产环境：

  • 根据实际效果选择最佳检索策略
  • 实现缓存机制，提高响应速度
  • 监控检索性能，及时调整策略
  • 考虑使用专业的检索服务或API

总结与展望

本集要点总结

1. 高级检索技术基础：

   • 多路召回的概念和优势
   • 重排序技术的原理和方法
   • 高级检索在RAG系统中的重要性

2. 多路召回实现：

   • 基于不同嵌入模型的多路召回
   • 基于不同检索策略的多路召回
   • 多路召回的结果融合和去重

3. 重排序技术：

   • 基于Cross-Encoder的重排序
   • 基于BM25的传统重排序
   • 基于规则的自定义重排序

4. 高级检索在RAG中的应用：

   • 多查询检索器提高召回率
   • 重排序检索器提高精度
   • 多种高级检索技术的组合应用

5. 高级技巧：

   • 检索策略自适应
   • 动态权重调整
   • 上下文感知检索
   • 领域特定检索优化

未来发展方向

1. 检索技术演进：

   • 更先进的语义理解模型
   • 端到端的检索-重排序联合优化
   • 自适应检索策略的自动化学习

2. 多模态检索：

   • 融合文本、图像、音频等多种模态
   • 跨模态检索和重排序
   • 多模态RAG系统

3. 个性化检索：

   • 基于用户偏好的个性化检索
   • 长期用户行为建模
   • 个性化重排序策略

4. 可解释性增强：

   • 检索结果的可解释性
   • 重排序决策的透明度
   • 检索过程的可视化

5. 大规模部署：

   • 分布式检索系统
   • 实时流处理
   • 边缘设备上的高效检索

通过本集的学习，我们了解了高级检索技术的核心概念和实现方法，掌握了多路召回和重排序等关键技术，以及如何将这些技术应用到RAG系统中。高级检索技术是构建高性能RAG系统的关键，它能够显著提高检索精度和召回率，从而提升整个智能体系统的性能和用户体验。

在后续的课程中，我们将深入探讨RAG的更多高级应用，以及如何构建更复杂、更强大的智能体记忆系统。