优化 RAG 的高级检索器技术

---

目录

· 简介 · 向量库创建 · 方法：朴素检索器 · 方法：父文档检索器 · 方法：自查询检索器 ∘ 查询构造器 ∘ 查询翻译器· 方法：上下文压缩检索器（重排序） · 结论

---

导言

让我们简要回顾一下组成 RAG 这个词的 3 个首字母缩写词的含义

• Retrieval（检索）：RAG 的主要目标是针对查询收集最相关的文档/分块。
• Augmented（增强）：创建一个结构良好的提示词（Prompt），以便在调用 LLM 时，它能清楚地了解其目的、背景以及应该如何响应。
• Generation（生成）：这是 LLM 发挥作用的地方。当模型获得良好的上下文（由“检索”步骤提供）并有明确的指令（由“增强”步骤提供）时，它将为用户生成高质量的响应。

正如我们所见，用户查询响应的生成（如果我们应用 RAG 进行问答）直接取决于我们构建“增强”以及特别是“检索”部分的质量。

在本文中，我们将专门关注“检索”部分。在这个返回最相关文档的重要过程中，向量库（vector store）的概念出现了。

为了创建这些检索，我们将使用 Langchain 库。

向量库本质上就是一个向量数据库，它以向量格式存储文档。这种向量表示法源于 Transformer 的使用。目前来说，这并不是什么新鲜事。

很明显，向量库越稳健和完善，我们运行的检索器就越好。我们已经知道，数据库的创建本身就是一门艺术。根据我们使用的分块大小或嵌入模型，我们的 RAG 效果会有所不同。

在此做一个说明

在本文中，我们不会讨论如何创建向量库。我们将讨论用于检索相关文档的一些技术。

由于一图胜千言，建议您查看以下内容

因此，我重申，在本文中我们将深入研究创建优质 RAG 工具的众多重要步骤之一。“检索”步骤是关键，因为它直接改善了 LLM 在生成响应时所拥有的上下文。

我们将研究的方法有

• 朴素检索器（Naive Retriever）
• 父文档检索器（Parent Document Retriever）
• 自查询检索器（Self-Query Retriever）
• 上下文压缩检索器（Contextual Compression Retriever，即重排序）

您可以在此处找到包含笔记本的项目。您也可以看看我的 GitHub

damiangilgonzalez1995 – 概览

---

向量库创建

为了阐述这些方法，我们将进行一个实际案例来辅助说明。因此，我们将创建一个关于《疾速追杀》（John Wick）系列电影评论的 RAG。

为了让读者能够跟随本文的每一步，您可以访问我创建的代码库。在其中您将找到每种方法的代码，以及用于创建向量库的文档。负责此任务的 Jupyter 笔记本可以在 git 仓库中找到，文件名为 "0__create_vectore_db.ipynb_"。

关于我们 RAG 的数据源，共有 4 个 CSV 文件，分别对应《疾速追杀》系列每部电影获得的评论。文件包含以下信息

如您所见，“Review”（评论）字段将是我们要检索的目标。其他字段作为元数据存储也很重要

• Movie_Title（电影标题）
• Review_Date（评论日期）
• Review_Title（评论标题）
• Review_Url（评论链接）
• Author（作者）
• Rating（评分）

要读取文件并将每一行转换为“Document”格式，我们执行以下代码

from langchain_community.document_loaders.csv_loader import CSVLoader
from datetime import datetime, timedelta

documents = []

for i in range(1, 4):
  loader = CSVLoader(
    encoding="utf8",
    file_path=f"data/john_wick_{i}.csv",
    metadata_columns=["Review_Date", "Review_Title", "Review_Url", "Author", "Rating"]
  )

  movie_docs = loader.load()
  for doc in movie_docs:

    # We add metadate about the number of the movi
    doc.metadata["Movie_Title"] = f"John Wick {i}"

    # convert "Rating" to an `int`, if no rating is provided - None
    doc.metadata["Rating"] = int(doc.metadata["Rating"]) if doc.metadata["Rating"] else 5

  documents.extend(movie_docs)

我们已经有了“Document”格式的文档

print(documents[0])

Document(page_content=": 0nReview: The best way I can describe John Wick is to picture Taken but instead of Liam Neeson it's Keanu Reeves and instead of his daughter it's his dog. That's essentially the plot of the movie. John Wick (Reeves) is out to seek revenge on the people who took something he loved from him. It's a beautifully simple premise for an action movie - when action movies get convoluted, they get bad i.e. A Good Day to Die Hard. John Wick gives the viewers what they want: Awesome action, stylish stunts, kinetic chaos, and a relatable hero to tie it all together. John Wick succeeds in its simplicity.", metadata={'source': 'data/john_wick_1.csv', 'row': 0, 'Review_Date': '6 May 2015', 'Review_Title': ' Kinetic, concise, and stylish; John Wick kicks ass.n', 'Review_Url': '/review/rw3233896/?ref_=tt_urv', 'Author': 'lnvicta', 'Rating': 8, 'Movie_Title': 'John Wick 1', 'last_accessed_at': datetime.datetime(2024, 4, 8, 11, 49, 47, 92560)})

我们只需在本地创建一个向量数据库（向量库）。为此，我使用了 Chroma。还要记住，必须使用一个嵌入模型，它会将文档转换为向量格式以进行存储。上述所有内容都可以在以下代码片段中看到

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
import os
from dotenv import load_dotenv

load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_KEY')

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

db = Chroma.from_documents(documents=documents, embedding=embeddings, collection_name="doc_jonhWick", persist_directory="./jonhWick_db")

这将在我们的本地创建一个名为 "_JonhWick_db_" 的数据库。这将是我们的 RAG 将使用的数据库，也是我们的检索器获取关于用户查询的最相关文档的地方。

现在是展示创建检索器的不同方法的时候了。

---

方法：朴素检索器

代码位于 1_naive_retriever.ipynb 文件中。

这种方法最简单，事实上它的名字就说明了这一点。我们使用这个形容词来标识该方法的原因很简单：当把查询输入到我们的数据库时，我们（天真地）希望它能返回最相关的文档/分块。

基本过程是我们使用创建向量库时所用的同一个 Transformer 对用户查询进行编码。获得其向量表示后，我们通过计算余弦相似度、距离等来计算相似度。

然后我们收集与查询最接近/相似的前 K 个文档。

这种检索器的工作流程可以从下图中看出

记住这个方案，让我们看看这一切在代码中是如何体现的。我们读取数据库

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
import os
from dotenv import load_dotenv

load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_KEY')

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectordb= Chroma(persist_directory="./jonhWick_db", 
                  embedding_function=embeddings, 
                  collection_name="doc_jonhWick")pyth

然后我们创建我们的 检索器。除了其他参数外，我们还可以配置相似度计算方法。

检索器

# Specifying top k
naive_retriever = vectordb.as_retriever(search_kwargs={ "k" : 10})

# Similarity score threshold retrieval
# naive_retriever = db.as_retriever(search_kwargs={"score_threshold": 0.8}, search_type="similarity_score_threshold")

# Maximum marginal relevance retrieval
# naive_retriever = db.as_retriever(search_type="mmr")

事实上，我们已经创建了我们的“朴素检索器”，但为了看看它是如何工作的，我们将创建完整的 RAG，我们记得它由以下组件组成

• R (Retrieval): 已完成
• A (Augmented): 尚未完成
• G (Generation): 尚未完成

增强与生成

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Augmented
TEMPLATE = """
You are happy assistant. Use the context provided below to answer the question.

If you do not know the answer, or are unsure, say you don't know.

Query:
{question}

Context:
{context}
"""

rag_prompt = ChatPromptTemplate.from_template(TEMPLATE)

# Generation
chat_model = ChatOpenAI()

我们已经有了 RAG 的 3 个组件。剩下的就是把它们组合起来，为此我们将使用 Langchain 链（chains）来创建 RAG。

我不知道您是否了解 Langchain 为更高效地创建链而创建的语言。这种语言被称为 LCEL (LangChain Expression Language)。如果您对这种在 Langchain 中创建链的方式不熟悉，我在这里为您留下了一个非常好的教程

最后，我们使用 Langchain 自己的链创建语言 (LCEL) 创建我们的 RAG

from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser

setup_and_retrieval = RunnableParallel({"question": RunnablePassthrough(), "context": naive_retriever })
output_parser = StrOutputParser()

naive_retrieval_chain = setup_and_retrieval 
                        | rag_prompt 
                        | chat_model 
                        | output_parser

naive_retrieval_chain.invoke( "Did people generally like John Wick?")

# response: 'Yes, people generally liked John Wick.'

这是为 RAG 创建链的最简单方法。在 Jupyter 笔记本中，您可以找到同样但更健壮的链。因为我不想让我们现在迷失在这个话题中，所以我只展示了最简单的形式。此外，为了让我们理解上面的代码中发生了什么，我创建了这个非常清晰的图表

太好了，我们已经完成了 朴素 RAG 的创建。让我们进入下一种方法。

---

方法：父文档检索器

代码位于 2_parent_document_retriever.ipynb 文件中。

想象一下，我们创建了一个 RAG，通过输入一些症状来识别可能的疾病。如果我们有一个朴素 RAG，我们可能会收集到一系列仅在这一两个症状上重合的可能疾病，这会让我们的工具显得有点糟糕。

这是使用父文档检索器（Parent Doc Retriever）的理想场景。这种技术的类型在于将大块（父块）切成更小的片（子块）。通过拥有小的分块，它们包含的信息更集中，因此其信息价值不会在文本段落中被稀释。

这一切有一个小问题

• 如果我们想在搜索最相关文档时精确，我们需要将文档拆分成 小块。
• 但为 LLM 提供良好的上下文也非常重要，这需要通过提供 更大的块 来实现。

以上内容可以从下图中看出

看起来这个问题没有解决办法，因为当我们提高精度时，上下文就会减少，反之亦然。这时，这种可以拯救我们生命的方法就出现了。

其主要思想是将大块（父块/父文档）进一步切分成更小的块（子块/子文档）。完成后，使用子块执行最相关前 K 个文档的搜索，并返回该前 K 子文档所属的父块。

我们已经有了主要思路，现在让我们脚踏实地地执行它。最好的解释方式是按步骤进行

1. 获取文档并创建大块（父块）
2. 对每个父块进行拆分，以生成 子块。
3. 将子块（向量表示）保存在 向量库 中。
4. 将 父块保存在内存中（我们不需要创建它们的向量表示）。

以上内容可以从下图中看出

这看起来可能非常复杂，因为我们必须用小块创建一个新数据库，并将父块保存在内存中。此外，还要知道每个子块属于哪个父块。幸好有 Langchain，构建它的方式非常简单。

您肯定得出了结论，即这种方法需要创建一个新的向量库。此外，对于《疾速追杀》电影评论的情况，由于数据源是 CSV 文件，不需要执行第一次拆分（父块）。这是因为我们可以认为 CSV 文件的每一行本身就是一个块。

总之，让我们直观地看看反映这种方法如何工作的下图

在代码中表示如下

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# documents = Read csv files. Check jupyter notebook for more details

parent_docs = documents

# Embedding Model
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Splitters
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
# We don't need a parent splitter because the data cames from CSV file, and each row is a parent doc.
# parent_splitter = RecursiveCharacterTextSplitter(chunk_size=800)

# Stores
store = InMemoryStore()
vectorstore = Chroma(embedding_function=embeddings, collection_name="fullDoc", persist_directory="./JohnWick_db_parentsRD")

parent_document_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    # parent_splitter =parent_splitter
)

这里的一个直观事实是，向量库中的块数（子块数）应该远大于存储在内存中的文档数（父块数）。通过以下代码，我们可以验证这一点

print(f"Number of parent chunks  is: {len(list(store.yield_keys()))}")

print(f"Number of child chunks is: {len(parent_document_retriever.vectorstore.get()['ids'])}")

'''
Number of parent chunks  is: 75
Number of child chunks is: 3701
'''

太好了，我们已经有了 父文档检索器，我们只需要基于这个检索器创建我们的 RAG 就行了。它的做法与前一种方法完全相同。我附上了在 Langchain 中创建链的代码。要查看更多详细信息，请查看 Jupyter 笔记本。

setup_and_retrieval = RunnableParallel({"question": RunnablePassthrough(), "context": parent_document_retriever })
output_parser = StrOutputParser()

parent_retrieval_chain = setup_and_retrieval | rag_prompt | chat_model | output_parser

请注意，它与前一种情况完全相同，唯一的细微差别是我们在 _"setup_and_retrieval" 变量中配置我们想要使用我们的 "parent_document_retriever"，而不是 "naive_retriever"_。

---

方法：自查询检索器

代码位于 3_self_query_retriever.ipynb 文件中。

这可能是提高我们检索器效率的最优方法之一。

它的主要特点是能够执行向量库搜索，并根据元数据应用过滤器。

我们知道，当我们应用“朴素检索”时，我们是在计算向量数据库的所有块与查询的相似度。向量库中的块越多，需要进行的相似度计算就越多。现在，想象一下能够预先根据元数据进行过滤，在选择了符合元数据相关条件的块之后，再计算相似度。这可以极大地降低计算和时间成本。

让我们看一个用例，以充分理解何时应用这种类型的检索。

想象一下，我们在向量数据库中存储了大量的体验和休闲优惠（例如：冲浪课程、高空滑索、美食路线等）。体验的描述是我们使用嵌入模型编码的内容。此外，每个优惠都有 3 个关键值或元数据：日期、价格和地点。

让我们想象一下，用户正在寻找这种风格的体验：一种在大自然中的体验，适合全家且安全。此外，价格必须低于 50 美元，地点是加利福尼亚。

这里有一点很清楚

我们不想让系统返回不符合用户要求的价格或地点的活动/体验。

因此，计算与不符合元数据过滤器的块/体验的相似度是没有意义的。

这种情况非常适合应用 自查询检索器。这种类型的检索器允许我们先通过元数据进行过滤，然后在满足元数据要求的块与用户输入之间进行相似度计算。

该技术可以总结为两个非常具体的步骤

• 查询构造器（Query Constructor）
• 查询翻译器（Query Translater）

---

查询构造器（Query Constructor）

名为“查询构造器”的步骤的目标是 根据用户输入创建合适的查询和过滤器。

谁负责应用相应的过滤器以及如何知道它们是什么？

为此，我们将使用一个 LLM。这个 LLM 必须能够决定应用哪些过滤器以及何时应用。我们还必须事先解释什么是元数据以及它们中的每一个意味着什么。简而言之，提示词必须包含 3 个关键点

• 上下文：个性、你应该如何行动、输出格式等。
• 元数据：有关可用元数据的信息。
• 查询：用户的查询/输入/问题。

LLM 生成的输出不能直接输入到数据库中。因此，需要所谓的“查询翻译器”。

---

查询翻译器（Query Translater）

这是一个负责 将 LLM (查询构造器) 的输出翻译成执行查询所需适当格式的模块。 根据您使用的向量数据库，您将不得不使用不同的翻译器。就我而言，我使用了 Chroma db，因此，我需要一个专注于该数据库的翻译器。幸运的是，Langchain 几乎为所有数据库都提供了专门的数据库翻译器。

---

正如您可能已经注意到的，我是图表的忠实粉丝。让我们看看下面这个图表，它为这个问题提供了相当多的清晰度

关于前一张图片，我们看到一切都始于用户的查询。我们创建包含 3 个关键字段的提示词，并将其输入到 LLM 中，LLM 生成一个带有两个关键字段的响应：“查询”和“过滤器”。这被输入到查询翻译器中，查询翻译器将这两个字段翻译成 Chroma DB 所需的正确格式。执行查询并根据用户的初始问题返回最相关的文档。

需要强调的是，用户输入的查询不必与输入到数据库的查询相同。在显示的图表中，可以看出 LLM 在考虑到 可用元数据和用户问题的情况下，检测到它可以创建一个带有“评分”元数据的过滤器。它还根据用户的查询创建了一个新查询。

让我们在代码中看看这一切。正如我所解释的，为 LLM 提供向量库中可用元数据的详细描述非常重要。这体现为以下代码片段

from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain_openai import ChatOpenAI
from langchain.retrievers.self_query.chroma import ChromaTranslator

metadata_field_info = [
    AttributeInfo(
        name="Movie_Title",
        description="The title of the movie",
        type="string",
    ),
    AttributeInfo(
        name="Review_Date",
        description="The date of the review",
        type="string",
    ),
    AttributeInfo(
        name="Review_Title",
        description="The title of the review",
        type="string",
    ),
    AttributeInfo(
        name="Review_Url",
        description="The URL of the review",
        type="string",
    ),
    AttributeInfo(
        name="Author",
        description="The author of the review",
        type="string",
    ),
    AttributeInfo(
        name="Rating",
        description="A 1 to 10 rating for the movie",
        type="integer",
    )
]

要定义我们的检索，我们必须定义以下几点

• 要使用的 LLM
• 要使用的嵌入模型
• 所访问的向量基础

• 关于在此向量库的文档中可以找到什么信息的描述。

• 元数据描述
• 您想使用的查询翻译器

让我们看看它在代码中是什么样子的

document_content_desription = "A review of the Jonh Wick movie."
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
chat_model = ChatOpenAI()

self_query_retriever = SelfQueryRetriever.from_llm(
    llm=ChatOpenAI(temperature=0),
    vectorstore =vectordb,
    document_contents = document_content_desription,
    metadata_field_info =metadata_field_info,
    verbose = True,
    structured_query_translator = ChromaTranslator()
)

让我们通过一个非常清晰的例子来看看我们通过使用这种类型的检索器极大地改善了我们的 RAG。 首先我们使用朴素检索器，然后使用自查询检索器。

Question = "Make a summary of the reviews that talk about John Wick 3 and have a score higher than 7"
response = naive_retrieval_chain.invoke(Question)
print(response)

'''
I don't know the answer.
'''
------------------------------------------------------------------------

response = self_retrieval_chain.invoke(Question)
print(response)

'''
John Wick: Chapter 3 - Parabellum is quite literally 
about consequences, dealing with the fallout of John's...
'''

正如我们所见，有显著的改善。

---

方法：上下文压缩检索器（重排序）

代码位于 4_contextual_compression_retriever(reranking).ipynb 文件中。

• 上下文窗口：从向量库获得的文档越多，LLM 给出的答案就越好。
• 召回率：从向量库检索的文档越多，获得 不相关分块的概率就越大，因此，召回率增加（这不是一件好事）。

这个问题似乎没有解决办法。当我们在一项指标中获得更好的性能时，另一项似乎注定会变差。 我们确定是这样吗？

这时，这种技术——压缩检索器——被提出，专注于重排序（Reranking）技术。可以说这项技术由两个非常不同的步骤组成

• 第 1 步：根据输入/问题获取大量相关文档。通常我们设置最相关的 K 个。
• 第 2 步：重新计算这些文档中哪些是真正相关的。丢弃其他真正没用的文档（压缩）。

对于第一步，使用的是所谓的 双编码器（Bi-Encoder），这与我们通常用来做基本 RAG 的方法没什么两样。向量化我们的文档，向量化查询并使用我们选择的任何指标计算相似度。

第二步与我们通常看到的有所不同。这种重新计算/重排序由 重排序模型 或 交叉编码器（Cross-Encoder） 执行。

这些模型期望输入两个文档/文本，返回一对之间的相似度得分。

如果这两个输入中的一个是 查询，另一个是 分块，我们可以计算两者之间的相似度。

这两种方法可以显示如下

您会意识到这两种方法最终提供相同的结果，即反映两个文本间相似度的指标。这完全正确，但有一个关键特征

交叉编码器返回的结果比双编码器可靠得多

好吧，它工作得更好，那么，为什么我们不直接对所有分块使用它，而只对前 K 个分块使用呢？因为这将是 时间、金钱/计算方面的极其昂贵。因此，我们对与查询相似度最接近的分块进行 第一次过滤，将重排序模型的使用减少到仅 K 次。

一个好的问题是去哪里寻找交叉编码器模型？我们很幸运，可以在 HuggingFace 上找到开源模型，但对于本文的实践案例，我们将使用 Cohere 公司提供的模型。

Cohere | 企业领先的 AI 平台

为了更好地理解这种方法的架构，让我们看一个直观的例子。

图片显示了以下步骤

• 1º) 我们获取查询，用 Transformer 将其编码为向量形式，并将其输入到向量库中。
• 2º) 从我们的数据库中收集 与查询最相似的文档。我们可以使用任何检索器方法。
• 3º) 接下来，我们使用 Cohere 交叉编码器模型。在图片示例中，此模型将被使用总共 4 次。记住，此模型的输入将是查询和一个文档/分块，以收集这两个文本的相似度。
• 4º) 在上一步中已对该模型进行了 4 次调用，并获得了查询与每个文档之间 4 个新的相似度值（0 到 1 之间）。可以看出，在前面的步骤中获得的分块 1，在重排序后，现在处于第 4 位。
• 5º) 我们将前 3 个最相关的分块添加到上下文中。

再次回到计算成本和时间问题，如果直接应用交叉编码器，请想一想，对于每个 新查询，都必须计算查询与每个文档的相似度。这根本不是最优的。

另一方面，使用 双编码器，文档的向量表示对于每个新查询都是相同的。

因此，我们有一种执行昂贵的优越方法，以及另一种工作良好但在每次新查询时没有巨大计算成本的方法。这一切都以统一这两种方法以获得更好的 RAG 告终。这就是所谓的 带重排序的上下文压缩方法。

让我们转到代码部分。让我们记住，此方法使用了一个检索器，在我们的例子中将是一个朴素检索器

naive_retriever = vectordb.as_retriever(search_kwargs={ "k" : 10})

多亏了 Langchain 及其与 Cohere 的集成，我们只需导入将执行对 Cohere 交叉编码器模型调用的模块

from langchain_cohere import CohereRerank

os.environ["COHERE_API_KEY"] = "YOUR API KEY FROM COHERE"

compressor = CohereRerank(top_n=3)

最后，我们使用 Langchain 创建我们的 上下文压缩检索器

from langchain.retrievers.contextual_compression import ContextualCompressionRetriever

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor, 
    base_retriever=naive_retriever
)

就是这么简单。让我们看看 朴素检索器和重排序检索器 之间的比较

正如我们所见，朴素检索器返回前 10 个分块/文档。在执行重排序并获得 3 个最相关的文档/分块后，有显著的变化。请注意 编号 16 的文档，它在第一个检索器的相关性排名中处于 第三位，在执行重排序后 变成了第一位。

---

结论

我们已经看到，根据我们想要应用 RAG 的案例的特点，我们将想要使用一种或另一种方法。此外，可能存在不知道该使用哪种检索器方法的情况。为此，有不同的库来评估您的 RAG。

有几种工具可用于此目的。我个人推荐的一些选择是 RAGAS 和 LangSmith 的组合。

使用 Ragas + LangSmith 评估 RAG 流水线

我强烈建议关注、学习和观看这些人的视频，他们确实是激励我创作这篇文章的人。

AI Makerspace

---

感谢阅读！

如果您发现我的工作有用，您可以订阅以 在我每次发布新文章时收到电子邮件。

如果您愿意，请在 Linkedin 上 关注 我！