构建 RAG 系统的高级技术

作者： Muhammad Asad Iqbal Khan 发布于 2025年5月15日归类于 Hugging Face Transformers 2条评论

在上一篇文章中，您学习了如何构建一个简单的检索增强生成（RAG）系统。RAG是一种强大的方法，可以利用外部知识增强大型语言模型，并且有许多变体可以使其更好地工作。接下来，您将看到一些高级特性和技术，以提高RAG系统的性能。特别是，您将学习到：

如何改进RAG中使用的提示词
如何使用混合检索来提高检索文档的质量
如何使用带有重排序的多阶段检索来提高生成响应的质量

通过我的书籍《Hugging Face Transformers中的NLP》，快速启动您的项目。它提供了带有工作代码的自学教程。

让我们开始吧。

构建 RAG 系统的高级技术
图片来源：Limonovich。保留部分权利。

概述

这篇博文分为三部分；它们是：

查询扩展和重构
混合检索：密集和稀疏方法
带有重排序的多阶段检索

查询扩展和重构

RAG系统面临的挑战之一是用户的查询可能与知识库中使用的术语不匹配。如果使用一个好的模型来生成嵌入，这不是问题，因为查询的上下文很重要。但是，您永远不知道特定查询是否如此。

查询扩展和重构可以通过生成查询的多个版本来弥补这一差距。其假设是，通过同一查询的几种变体，至少其中之一可以帮助检索RAG最相关的文档。

要进行查询扩展，您需要一个能够生成输入变体的模型。BART就是一个例子。让我们看看如何使用它进行查询扩展。

from transformers import BartForConditionalGeneration, BartTokenizer

# Load BART model and tokenizer
tokenizer = BartTokenizer.from_pretrained("facebook/bart-large")
model = BartForConditionalGeneration.from_pretrained("facebook/bart-large")

def reformulate_query(query, n=2):
    inputs = tokenizer(query, return_tensors="pt")
    outputs = model.generate(
        **inputs,
        max_length=64,
        num_beams=10,
        num_return_sequences=n,
        temperature=1.5,  # High temperature for diversity
        top_k=50,
        do_sample=True
    )
    # Decode the outputs one by one
    reformulations = [tokenizer.decode(output, skip_special_tokens=True) for output in outputs]
    all_queries = [query] + reformulations
    return all_queries

# Generate reformulations from an example query
query = "How do transformer-based systems process natural language?"
reformulated_queries = reformulate_query(query)
print(f"Original Query: {query}")
print("Reformulated Queries:")
for i, q in enumerate(reformulated_queries[1:], 1):
    print(f"{i}. {q}")

from transformers import BartForConditionalGeneration, BartTokenizer

# 加载BART模型和分词器

tokenizer = BartTokenizer.from_pretrained("facebook/bart-large")

model = BartForConditionalGeneration.from_pretrained("facebook/bart-large")

def reformulate_query(query, n=2):

inputs = tokenizer(query, return_tensors="pt")

outputs = model.generate(

**inputs,

max_length=64,

num_beams=10,

num_return_sequences=n,

temperature=1.5, # 高温以增加多样性

top_k=50,

do_sample=True

)

# 逐个解码输出

reformulations = [tokenizer.decode(output, skip_special_tokens=True) for output in outputs]

all_queries = [query] + reformulations

return all_queries

# 从示例查询生成重构

query = "How do transformer-based systems process natural language?"

reformulated_queries = reformulate_query(query)

print(f"Original Query: {query}")

print("Reformulated Queries:")

for i, q in enumerate(reformulated_queries[1:], 1):

print(f"{i}. {q}")

在此代码中，您加载了一个预训练的BART模型和分词器。它被创建为一个`BartForConditionalGeneration`对象，这是一个用于文本生成的序列到序列模型。与您在Hugging Face transformers库中使用模型的方式相同，您将输入进行分词并将其传递给`reformulate_query()`函数中的模型。您要求模型对一个输入生成`n`个输出。

为了创建更多变体，您将温度略高于1，您甚至可以尝试更高的值。使用BART生成实际上是要求模型读取您的输入并记住其在“隐藏状态”中的含义，然后将隐藏状态解码回文本，并可能出现变体。使用束搜索创建多个变体，如果您愿意，可以添加更多生成参数。

使用分词器将多个输出逐个解码成文本。然后您可以在代码的末尾打印出来。如果您运行此代码，您可能会看到：

Original Query: How do transformer-based systems process natural language?
Reformulated Queries:
1. How do transformer-based systems process natural language?
2. How do transformer-based systems work in natural language?

原始查询：基于Transformer的系统如何处理自然语言？

重构查询

1. 基于Transformer的系统如何处理自然语言？

2. 基于Transformer的系统在自然语言中如何工作？

原始查询的歧义越大，您将获得的变体就越多。

混合检索：密集和稀疏方法

RAG的理念是用知识库中最相关的文档来补充查询的上下文。这些额外的信息可以帮助模型生成更好的响应。您可以使用不同的方法来查找相关文档。

密集向量检索意味着将知识库中的文档表示为高维向量。此向量中的所有维度都很重要，但没有具体原因来确定每个维度代表什么。通常，密集向量是一个看起来随机的浮点数向量。

然而，稀疏向量有许多零。它通常具有更高的维度，并且是整数向量。一个例子是独热向量，其中每个位置代表词汇表中的一个单词，并且只有当该单词存在于文档中时，其值才为1。

密集向量和稀疏向量都没有绝对的优劣。如果您使用嵌入模型生成密集向量，它擅长捕捉语义相似性。另一方面，稀疏向量通常擅长捕捉关键词。对稀疏向量进行操作可能会消耗大量内存，但您可以通过使用Okapi BM25等技术来减少工作量。

在下面的代码中，您需要安装库来计算BM25分数。您可以使用pip完成此操作

pip install rank-bm25

1	pip install rank-bm25

让我们看看如何结合稀疏向量和密集向量来构建检索系统

from rank_bm25 import BM25Okapi
from transformers import AutoTokenizer, AutoModel
import faiss
import numpy as np
import torch

dense_tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")
dense_model = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")

def generate_embedding(text):
    """Generate dense vector using mean pooling"""
    inputs = dense_tokenizer(text, padding=True, truncation=True, return_tensors="pt", max_length=512)
    with torch.no_grad():
        outputs = dense_model(**inputs)

    attention_mask = inputs['attention_mask']
    embeddings = outputs.last_hidden_state

    expanded_mask = attention_mask.unsqueeze(-1).expand(embeddings.shape).float()
    sum_embeddings = torch.sum(embeddings * expanded_mask, axis=1)
    sum_mask = torch.clamp(expanded_mask.sum(axis=1), min=1e-9)
    mean_embeddings = sum_embeddings / sum_mask
    return mean_embeddings.cpu().numpy()

# Sample document collection
documents = [
    "Transformers use self-attention mechanisms to process input sequences in "
        "parallel, making them efficient for long sequences.",
    "The attention mechanism in transformers allows the model to focus on different "
        "parts of the input sequence when generating each output element.",
    "Transformer models have a fixed context length determined by the positional "
        "encoding and self-attention mechanisms.",
    "To handle sequences longer than the context length, transformers can use "
        "techniques like sliding windows or hierarchical processing.",
    "Recurrent Neural Networks (RNNs) process sequences sequentially, which can be "
        "inefficient for long sequences.",
    "Long Short-Term Memory (LSTM) networks are a type of RNN designed to handle "
        "long-term dependencies in sequences.",
    "The Transformer architecture was introduced in the paper 'Attention Is All "
        "You Need' by Vaswani et al.",
    "BERT (Bidirectional Encoder Representations from Transformers) is a "
        "transformer-based model designed for understanding the context of words.",
    "GPT (Generative Pre-trained Transformer) is a transformer-based model designed "
        "for natural language generation.",
    "Transformer-XL extends the context length of transformers by using a "
        "segment-level recurrence mechanism."
]

# Prepare for sparse retrieval (BM25)
tokenized_corpus = [doc.lower().split() for doc in documents]
bm25 = BM25Okapi(tokenized_corpus)

# Prepare for dense retrieval (FAISS)
document_embeddings = generate_embedding(documents)
dimension = document_embeddings.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(document_embeddings)

def hybrid_retrieval(query, k=3, alpha=0.5):
    """Hybrid retrieval: Use both the BM25 and L2 index on FAISS"""
    # Sparse score of each document with BM25
    tokenized_query = query.lower().split()
    bm25_scores = bm25.get_scores(tokenized_query)

    # Normalize BM25 scores to [0,1] unless all elements are zero
    if max(bm25_scores) > 0:
        bm25_scores = bm25_scores / max(bm25_scores)

    # Sort all documents according to L2 distance to query
    query_embedding = generate_embedding(query)
    distances, indices = index.search(query_embedding, len(documents))

    # Dense score: 1/distance as similarity metric, then normalize to [0,1]
    eps = 1e-5  # a small value to prevent division by zero
    dense_scores = 1 / (eps + np.array(distances[0]))
    dense_scores = dense_scores / max(dense_scores)

    # Combine scores = affine combination of sparse and dense scores
    combined_scores = alpha * dense_scores + (1 - alpha) * bm25_scores

    # Get top-k documents
    top_indices = np.argsort(combined_scores)[::-1][:k]
    results = [(documents[idx], combined_scores[idx]) for idx in top_indices]
    return results

# Retrieve documents using hybrid retrieval
query = "How do transformers handle long sequences?"
results = hybrid_retrieval(query)
print(f"Query: {query}")
for i, (doc, score) in enumerate(results):
    print(f"Document {i+1} (Score: {score:.4f}):")
    print(doc)
    print()

from rank_bm25 import BM25Okapi

from transformers import AutoTokenizer, AutoModel

import faiss

import numpy as np

import torch

dense_tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")

dense_model = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")

def generate_embedding(text):

"""使用均值池化生成密集向量"""

inputs = dense_tokenizer(text, padding=True, truncation=True, return_tensors="pt", max_length=512)

with torch.no_grad():

outputs = dense_model(**inputs)

attention_mask = inputs['attention_mask']

embeddings = outputs.last_hidden_state

expanded_mask = attention_mask.unsqueeze(-1).expand(embeddings.shape).float()

sum_embeddings = torch.sum(embeddings * expanded_mask, axis=1)

sum_mask = torch.clamp(expanded_mask.sum(axis=1), min=1e-9)

mean_embeddings = sum_embeddings / sum_mask

return mean_embeddings.cpu().numpy()

# 示例文档集合

documents = [

"Transformers use self-attention mechanisms to process input sequences in "

"parallel, making them efficient for long sequences.",

"The attention mechanism in transformers allows the model to focus on different "

"parts of the input sequence when generating each output element.",

"Transformer models have a fixed context length determined by the positional "

"encoding and self-attention mechanisms.",

"To handle sequences longer than the context length, transformers can use "

"techniques like sliding windows or hierarchical processing.",

"Recurrent Neural Networks (RNNs) process sequences sequentially, which can be "

"inefficient for long sequences.",

"Long Short-Term Memory (LSTM) networks are a type of RNN designed to handle "

"long-term dependencies in sequences.",

"The Transformer architecture was introduced in the paper 'Attention Is All "

"You Need' by Vaswani et al.",

"BERT (Bidirectional Encoder Representations from Transformers) is a "

"transformer-based model designed for understanding the context of words.",

"GPT (Generative Pre-trained Transformer) is a transformer-based model designed "

"for natural language generation.",

"Transformer-XL extends the context length of transformers by using a "

"segment-level recurrence mechanism."

]

# 准备稀疏检索 (BM25)

tokenized_corpus = [doc.lower().split() for doc in documents]

bm25 = BM25Okapi(tokenized_corpus)

# 准备密集检索 (FAISS)

document_embeddings = generate_embedding(documents)

dimension = document_embeddings.shape[1]

index = faiss.IndexFlatL2(dimension)

index.add(document_embeddings)

def hybrid_retrieval(query, k=3, alpha=0.5):

"""混合检索：在FAISS上同时使用BM25和L2索引"""

# 使用BM25计算每个文档的稀疏分数

tokenized_query = query.lower().split()

bm25_scores = bm25.get_scores(tokenized_query)

# 将BM25分数归一化到[0,1]之间，除非所有元素都为零

if max(bm25_scores) > 0:

bm25_scores = bm25_scores / max(bm25_scores)

# 根据与查询的L2距离对所有文档进行排序

query_embedding = generate_embedding(query)

distances, indices = index.search(query_embedding, len(documents))

# 密集分数：1/距离作为相似性度量，然后归一化到[0,1]

eps = 1e-5 # 一个小值以防止除以零

dense_scores = 1 / (eps + np.array(distances[0]))

dense_scores = dense_scores / max(dense_scores)

# 结合分数 = 稀疏和密集分数的仿射组合

combined_scores = alpha * dense_scores + (1 - alpha) * bm25_scores

# 获取top-k文档

top_indices = np.argsort(combined_scores)[::-1][:k]

results = [(documents[idx], combined_scores[idx]) for idx in top_indices]

return results

# 使用混合检索检索文档

query = "How do transformers handle long sequences?"

results = hybrid_retrieval(query)

print(f"Query: {query}")

for i, (doc, score) in enumerate(results):

print(f"Document {i+1} (Score: {score:.4f}):")

print(doc)

print()

当你运行这段代码时，你会看到：

Query: How do transformers handle long sequences?

Document 1 (Score: 0.7924):
Transformers use self-attention mechanisms to process input sequences in parallel, making them efficient for long sequences.

Document 2 (Score: 0.7458):
Long Short-Term Memory (LSTM) networks are a type of RNN designed to handle long-term dependencies in sequences.

Document 3 (Score: 0.7131):
To handle sequences longer than the context length, transformers can use techniques like sliding windows or hierarchical processing.

查询：Transformers如何处理长序列？

文档1（得分：0.7924）

Transformers使用自注意力机制并行处理输入序列，使其对长序列高效。

文档2（得分：0.7458）

长短期记忆（LSTM）网络是一种RNN，旨在处理序列中的长期依赖性。

文档3（得分：0.7131）

为了处理比上下文长度更长的序列，transformers可以使用滑动窗口或分层处理等技术。

首先，您为集合中的所有文档创建了一个Okapi BM25索引。Okapi BM25是一种基于TF-IDF的评分方法，这意味着它通过检查精确单词的交集来比较两个文本。从这个意义上说，大小写不重要。因此，在使用BM25时，您将文档转换为小写。

然后，您使用预训练的句子Transformer模型为文档集合生成密集向量。您将这些密集向量存储在FAISS索引中，以便使用L2距离进行高效的相似性搜索。

此代码的关键部分在`hybrid_retrieval()`函数中。在准备好Okapi BM25和FAISS索引后，您将查找最适合您的查询字符串的文档。获得的BM25分数是与每个文档对应的TF-IDF分数。您还计算了每个文档的FAISS的L2距离度量。然后将此距离转换为分数，以匹配BM25的分数：更高的分数应该意味着更好的匹配。为了确保两种方法可比较，您将分数归一化到[0, 1]范围。

根据您的选择，您可以通过更改参数`alpha`来更强调密集检索或稀疏检索。然后使用组合分数来查找要返回的top-k文档。正如您从上面的输出中看到的那样。

这种混合方法通常优于任何单一方法，特别是对于同时需要语义理解和特定术语的复杂查询。

带有重排序的多阶段检索

如果您有一个完美的模型来评估文档与您的查询的相关性，那么一个简单的检索系统就足够了。然而，没有模型是完美的。实际上，通常质量越高的模型计算强度也越大。这就是多阶段检索的用武之地。

混合检索擅长快速挑选文档。特别是如果您使用快速模型，您可以轻松计算大量文档的分数。然而，挑选并不总是好的。但是您可以使用更慢但更准确的模型重新计算分数。这次，并非所有文档都考虑在内，而只考虑混合检索挑选的文档。只要第一阶段使用的模型大致正确，第二阶段计算密集度更高的模型将为您提供准确的选择。

这就是多阶段检索技术的作用。让我们看看如何实现它。

...

# Load pre-trained model and tokenizer for re-ranking
reranker_tokenizer = AutoTokenizer.from_pretrained("cross-encoder/ms-marco-MiniLM-L-6-v2")
reranker_model = AutoModelForSequenceClassification.from_pretrained("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query, documents, top_k=3):
    """Sort documents by the reranker model and select top-k"""
    # Prepare inputs for the re-ranker
    pairs = [[query, doc] for doc in documents]
    features = reranker_tokenizer(pairs, padding=True, truncation=True, return_tensors="pt")
    # Get re-ranking scores
    with torch.no_grad():
        scores = reranker_model(**features).logits.squeeze(-1).cpu().numpy()
    # Sort documents by score, then pick top-k
    ranked_indices = np.argsort(scores)[::-1][:top_k]
    reranked_docs = [(documents[idx], float(scores[idx])) for idx in ranked_indices]
    return reranked_docs

def multi_stage_retrieval(query, documents, initial_k=5, final_k=3):
    """Multi-stage retrieval: Hybrid retrievel to shortlist documents, then pick with a reranker"""
    # Stage 1: Initial retrieval using hybrid method
    initial_results = hybrid_retrieval(query, k=initial_k)
    initial_docs = [doc for doc, _ in initial_results]
    # Stage 2: Re-ranking
    reranked_results = rerank(query, initial_docs, top_k=final_k)
    return reranked_results

# Example query
query = "How do transformers handle long sequences?"
results = multi_stage_retrieval(query, documents)
print(f"Query: {query}")
print("Re-ranked Results:")
for i, (doc, score) in enumerate(results):
    print(f"Document {i+1} (Score: {score:.4f}):")
    print(doc)
    print()

...

# 加载用于重排序的预训练模型和分词器

reranker_tokenizer = AutoTokenizer.from_pretrained("cross-encoder/ms-marco-MiniLM-L-6-v2")

reranker_model = AutoModelForSequenceClassification.from_pretrained("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query, documents, top_k=3):

"""通过重排序模型对文档进行排序并选择top-k"""

# 为重排序器准备输入

pairs = [[query, doc] for doc in documents]

features = reranker_tokenizer(pairs, padding=True, truncation=True, return_tensors="pt")

# 获取重排序分数

with torch.no_grad():

scores = reranker_model(**features).logits.squeeze(-1).cpu().numpy()

# 按分数排序文档，然后选择top-k

ranked_indices = np.argsort(scores)[::-1][:top_k]

reranked_docs = [(documents[idx], float(scores[idx])) for idx in ranked_indices]

return reranked_docs

def multi_stage_retrieval(query, documents, initial_k=5, final_k=3):

"""多阶段检索：混合检索筛选文档，然后用重排序器挑选"""

# 阶段1：使用混合方法进行初步检索

initial_results = hybrid_retrieval(query, k=initial_k)

initial_docs = [doc for doc, _ in initial_results]

# 阶段2：重新排序

reranked_results = rerank(query, initial_docs, top_k=final_k)

return reranked_results

# 示例查询

query = "How do transformers handle long sequences?"

results = multi_stage_retrieval(query, documents)

print(f"Query: {query}")

print("Re-ranked Results:")

for i, (doc, score) in enumerate(results):

print(f"Document {i+1} (Score: {score:.4f}):")

print(doc)

print()

此代码是在上一节的基础上构建的。它使用了与之前相同的`hybrid_retrieval()`函数。

在`multi_stage_retrieval()`函数中，您首先使用混合检索获取文档列表。然后，您使用重排序模型对这些文档进行重排序。

重排序模型是一个交叉编码器模型，一种可用于排序任务的Transformer模型。简而言之，它接受两个序列作为输入，以`[CLS] query [SEP] document [SEP]`的格式连接。模型的输出是一个分数，显示文档与查询的相关性。这是一个慢模型，但比L2距离或BM25更准确。

在`rerank()`函数中，您在混合检索筛选出的查询和文档上运行重排序模型。然后，您根据重排序模型提供的分数选择top-k文档。`multi_stage_retrieval()`函数中的参数`initial_k`和`final_k`允许您控制召回率（检索所有相关文档）和准确率（确保检索到的文档相关）之间的权衡。较大的`initial_k`会增加召回率，但需要更多的重排序计算，而较小的`final_k`则侧重于最相关的文档。