---

前言

"检索增强生成”（RAG）系列教程2：该教程介绍了如何对RAG的查询部分进行优化。

一、概述

1-1、RAG概念

概念：目前的LLM通常是用很多已经存在的文字数据训练出来的。这就导致一个问题：LLM对最新的信息或者个人隐私信息不太了解，因为这些内容在训练时没有被包括进去。虽然可以通过“微调”（也就是针对特定任务再训练一下LLM）来解决这个问题，但微调成本很高，技术相对比较复杂，现在出现了一种新的方法，叫“检索增强生成”（RAG）。这个方法的思路是：从外部的数据源（比如数据库或者网页）中找到相关的资料，然后把这些资料“喂”给聊天机器人，帮助它更好地回答问题。这种方法就像是给聊天机器人提供了一个“外挂”，让它能够接触到更多的知识。

1-2、前置知识

1-2-1、ModelScopeEmbeddings 词嵌入模型

ModelScope Embeddings 是阿里巴巴达摩院推出的嵌入模型，旨在将文本、图像等数据转换为高维向量，便于机器学习模型处理。这些嵌入向量能够捕捉数据的语义信息，广泛应用于自然语言处理（NLP）、计算机视觉（CV）等领域。

安装库：

pip install modelscope

Demo:

from langchain.embeddings import ModelScopeEmbeddings

model_id = "damo/nlp_corom_sentence-embedding_english-base"
embeddings = ModelScopeEmbeddings(model_id=model_id)
text = "This is a test document."
query_result = embeddings.embed_query(text)
doc_results = embeddings.embed_documents(["foo"])

输出：

1-2-2、FAISS介绍&安装 (向量相似性搜索)

FAISS（Facebook AI Similarity Search）是由 Meta（前 Facebook）开发的一个高效相似性搜索和密集向量聚类库。它主要用于在大规模数据集中进行向量相似性搜索，特别适用于机器学习和自然语言处理中的向量检索任务。FAISS 提供了多种索引类型和算法，可以在 CPU 和 GPU 上运行，以实现高效的向量搜索。

FAISS 的主要特性

• 高效的相似性搜索：支持大规模数据集的高效相似性搜索，包括精确搜索和近似搜索。
• 多种索引类型：支持多种索引类型，如扁平索引（Flat Index）、倒排文件索引（IVF）、产品量化（PQ）等。
• GPU 加速：支持在 GPU 上运行，以加速搜索过程。
• 批量处理：支持批量处理多个查询向量，提高搜索效率。
• 灵活性：支持多种距离度量，如欧氏距离（L2）、内积（Inner Product）等。

安装：

# cpu或者是GPU版本
pip install faiss-cpu
# 或者
pip install faiss-gpu

Demo分析： 使用 LangChain 库来处理一个长文本文件，将其分割成小块，然后使用 Hugging Face 嵌入和 FAISS 向量存储来执行相似性搜索。

• CharacterTextSplitter：用于将长文本分割成小块。
• FAISS：用于创建向量数据库。
• TextLoader：用于加载文本文件。
• HuggingFaceEmbeddings：另一个用于生成文本嵌入向量的类。

from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.document_loaders import TextLoader
from langchain.embeddings import HuggingFaceEmbeddings

# This is a long document we can split up.
with open('./index.txt', encoding='utf-8') as f:
    state_of_the_union = f.read()

text_splitter = CharacterTextSplitter(        
    chunk_size = 100,
    chunk_overlap  = 0,
)
docs = text_splitter.create_documents([state_of_the_union])

embeddings = HuggingFaceEmbeddings()
db = FAISS.from_documents(docs, embeddings)

query = "学生的表现怎么样？"
docs = db.similarity_search(query)
print(docs[0].page_content)

输出：

Notice： 查询分数，这里的分数为L2距离，因此越低越好

1-2-3、Tiktoken 分词工具

Tiktoken 是 OpenAI 开发的一个高效的分词工具，专门用于处理 GPT 系列模型（如 GPT-3、GPT-4）的文本输入和输出。它能够将自然语言文本转换为模型可以理解的 token 序列，同时支持从 token 序列还原为文本。Tiktoken 的设计目标是高效、灵活且易于集成到各种自然语言处理（NLP）任务中。

安装：

pip install tiktoken

使用：

import tiktoken
# 编码器的加载
encoder = tiktoken.get_encoding("cl100k_base")
text = "这是一个示例文本。"

# 对文本进行编码
tokens = encoder.encode(text)
print(tokens)

# 对文本进行解码
decoded_text = encoder.decode(tokens)
print(decoded_text)

二、Rag From Scratch：Query Transformations

Rag From Scratch：Query Transformations 查询转换： 即侧重于重写和/或修改检索问题的方法。

2-1、前置环境安装

使用到的包：

pip install langchain_community 
pip install tiktoken 
pip install langchain-openai 
pip install langchainhub 
pip install chromadb 
pip install langchain
pip install modelscope

# 这里我使用的向量检索工具是FAISS
# cpu或者是GPU版本
pip install faiss-cpu
# 或者
pip install faiss-gpu

2-2、多查询检索器

多查询检索器： 通过使用大语言模型（LLM），我们可以从不同的角度为用户的查询生成多个相关的问题。这样做的目的是自动优化搜索过程。对于每一个生成的问题，系统会找到一组相关的文档。然后，系统会把这些文档合并起来，形成一个更大的、可能更相关的文档集合。通过从多个角度生成问题，我们可以克服单纯依靠距离匹配检索的局限性，从而得到更丰富、更全面的搜索结果。如下图所示：

2-2-1、加载网页内容

• WebBaseLoader：从指定 URL 加载网页内容。
• bs4.SoupStrainer：只解析特定类名的 HTML 元素（如 post-content、post-title、post-header），以减少解析时间。
• blog_docs：加载后的文档对象。

import bs4
from langchain_community.document_loaders import WebBaseLoader
loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
blog_docs = loader.load()

2-2-2、分割文档

• RecursiveCharacterTextSplitter：将文档递归分割成小块。
• chunk_size=300：每个块的最大 token 数量。
• chunk_overlap=50：块之间的重叠 token 数量，用于保持上下文连贯。
• splits：分割后的文档块。

# Split
from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=300,
    chunk_overlap=50)

# Make splits
splits = text_splitter.split_documents(blog_docs)

输出：

2-2-3、向量化文档并创建向量存储

• ModelScopeEmbeddings：使用 ModelScope 嵌入模型将文档块转换为向量。
• FAISS：将嵌入向量存储到 FAISS 向量数据库中。
• retriever：创建一个检索器，用于查询相关文档。

from langchain_community.embeddings import ModelScopeEmbeddings
from langchain_openai import ChatOpenAI
import os

from langchain_community.vectorstores import FAISS
vectorstore = FAISS.from_documents(
    documents=splits,
    embedding=ModelScopeEmbeddings(),
)
retriever = vectorstore.as_retriever()

2-2-4、初始化LLM

• ChatOpenAI：初始化一个 LLM 实例，使用 qwen-max 模型。
• temperature=0： 控制生成文本的随机性，值为 0 时生成确定性结果。
• max_tokens=1024：限制生成文本的最大长度。
• base_url：指定 API 的基础 URL。

from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="qwen-max",
    temperature=0,
    max_tokens=1024,
    timeout=None,
    max_retries=2,
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

2-2-5、多维度查询的生成

• ChatPromptTemplate：定义一个提示模板，用于生成多个查询。
• generate_queries：将用户问题输入 LLM，生成多个相关查询。

from langchain.prompts import ChatPromptTemplate

# Multi Query: Different Perspectives
template = """You are an AI language model assistant. Your task is to generate five 
different versions of the given user question to retrieve relevant documents from a vector 
database. By generating multiple perspectives on the user question, your goal is to help
the user overcome some of the limitations of the distance-based similarity search. 
Provide these alternative questions separated by newlines. Original question: {question}"""
prompt_perspectives = ChatPromptTemplate.from_template(template)

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

generate_queries = (
    prompt_perspectives
    | llm
    | StrOutputParser()
    | (lambda x: x.split("\n"))
)

# 这里我们小试牛刀，可以看得出来输出非常的人性化。
generate_queries.invoke({"question": '论如何解决提出无理要求的员工？'})

输出：

[‘如何妥善处理员工提出的不合理要求？’,
‘面对员工的无理要求时，有哪些有效的解决策略？’,
‘当员工提出过分的要求时，管理者应该如何应对？’,
‘对于那些经常提出不切实际要求的员工，公司应该采取什么措施？’,
‘在工作中遇到员工提出难以接受的要求时，怎样做才能既解决问题又保持良好的工作关系？’]

2-2-6、检索相关文档

• get_unique_union：将检索到的文档合并并去重。
• retrieval_chain：生成多个查询，检索相关文档，并返回唯一的文档集合。
• retriever.map()： 用于检索和查询相关文档
• docs：最终检索到的文档。

from langchain.load import dumps, loads

def get_unique_union(documents: list[list]):
    """ Unique union of retrieved docs """
    # Flatten list of lists, and convert each Document to string
    flattened_docs = [dumps(doc) for sublist in documents for doc in sublist]
    # Get unique documents
    unique_docs = list(set(flattened_docs))
    # Return
    return [loads(doc) for doc in unique_docs]

# Retrieve
question = "What is task decomposition for LLM agents?"
retrieval_chain = generate_queries | retriever.map() | get_unique_union
docs = retrieval_chain.invoke({"question":question})
len(docs)

2-2-7、构建RAG链，生成最终答案

• ChatPromptTemplate：定义一个提示模板，用于生成最终答案。
• final_rag_chain：结合检索到的文档和用户问题，生成最终答案。
• invoke：执行 RAG 链并输出结果。

from operator import itemgetter
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough

# RAG
template = """Answer the following question based on this context:

{context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

# 输入数据有两个
# 1、context：通过 retrieval_chain 检索到的相关文档。（检索链检索到的文档）
# 2、itemgetter("question") 是 Python 标准库 operator 模块中的一个函数，它的作用是从一个字典或对象中提取指定的键或属性。在这里，它的作用是从输入的字典中提取键为 "question" 的值。
final_rag_chain = (
    {"context": retrieval_chain,
     "question": itemgetter("question")}
    | prompt
    | llm
    | StrOutputParser()
)

print(final_rag_chain.invoke({"question":question}))

输出：

Task decomposition for LLM (Large Language Model) agents is a process where complex tasks are broken down into smaller, more manageable subtasks. This approach helps the agent to handle and execute complex tasks more efficiently. The idea is to make the problem-solving process more structured and step-by-step, which can improve the overall performance and reliability of the agent.

In the context provided, task decomposition can be achieved in several ways:

1. Chain of Thought (CoT): This technique involves instructing the model to “think step by step” to break down a large task into smaller, simpler steps. This not only makes the task more manageable but also provides insight into the model’s reasoning process.

2. Tree of Thoughts (ToT): This extends the CoT approach by exploring multiple reasoning possibilities at each step, creating a tree structure. The search process can use breadth-first search (BFS) or depth-first search (DFS), with each state evaluated by a classifier or through majority voting.

3. Simple Prompting: Using straightforward prompts like “Steps for XYZ. 1.” or “What are the subgoals for achieving XYZ?” to guide the LLM in breaking down the task.

4. Task-Specific Instructions: Providing specific instructions tailored to the task, such as “Write a story outline” for writing a novel.

5. Human Inputs: Incorporating human input to help define and refine the subtasks.

By decomposing tasks, LLM agents can better manage complexity, improve their planning and execution, and ultimately enhance their problem-solving capabilities.

2-3、RAG-Fusion（加重排的多查询检索）

概述： RAG-Fusion与多查询检索大部分来看比较相似，区别在于。

• 相关问题生成：RAG-Fusion直接要求生成与输入问题相关的查询，而多查询检索更侧重于生成多个不同视角的查询，以克服基于距离的相似性搜索的局限性。
• 文档检索与融合部分：多查询检索在进行这部分内容时，直接对检索到的文档进行去重处理，而RAG-Fusion，则引入了倒数排名融合（Reciprocal Rank Fusion, RRF） 算法，对多个检索器返回的文档进行融合和重排。

2-3-1、倒数排名融合（Reciprocal Rank Fusion, RRF）

倒数排名融合（Reciprocal Rank Fusion, RRF）是一种用于融合多个排序列表的算法，常用于信息检索和推荐系统。它通过将不同排序列表中的排名进行加权融合，生成一个综合排序列表。

算法原理

RRF 的核心思想是将每个排序列表中的排名转换为倒数，然后进行加权求和，最终根据总和重新排序。

公式

对于每个项目 d ，其在融合后的得分S(d) 计算公式为：

其中：

• n 是排序列表的数量。
• $r_i(d)$ 是项目 d 在第 i 个排序列表中的排名（从 1 开始）。
• k 是一个常数，通常取 60，用于平滑排名差异。

步骤

1. 输入多个排序列表：每个列表包含一组项目及其排名。
2. 计算每个项目的 RRF 得分：根据公式计算每个项目在所有列表中的 RRF 得分。
3. 按得分排序：根据 RRF 得分对所有项目进行降序排列，生成最终的融合排序列表。

优点

• 简单易实现：算法逻辑简单，易于实现和调试。
• 无需归一化：RRF 不需要对原始排名进行归一化处理。
• 鲁棒性强：对个别列表的噪声和异常值具有较强的鲁棒性。

应用场景

• 信息检索：融合多个搜索引擎的搜索结果。
• 推荐系统：融合多个推荐算法的输出。
• 数据融合：融合来自不同数据源的排序数据。

示例

假设有两个排序列表：

• 列表 A: [A, B, C]
• 列表 B: [B, A, C]

取 ( k = 60 )，计算 RRF 得分：

• 项目 A:

  • 列表 A 排名 1: $\frac{1}{60+1}=\frac{1}{61}$
  • 列表 B 排名 2: $\frac{1}{60+2}=\frac{1}{62}$
  • 总得分: $\frac{1}{61}+\frac{1}{62}\approx 0.0326$

• 项目 B:

  • 列表 A 排名 2: $\frac{1}{60+2}=\frac{1}{62}$
  • 列表 B 排名 1: $\frac{1}{60+1}=\frac{1}{61}$
  • 总得分: $\frac{1}{62}+\frac{1}{61}\approx 0.0326$

• 项目 C:

  • 列表 A 排名 3: $\frac{1}{60+3}=\frac{1}{63}$
  • 列表 B 排名 3: $\frac{1}{60+3}=\frac{1}{63}$
  • 总得分: $\frac{1}{63}+\frac{1}{63}\approx 0.0317$

最终排序为 [A, B, C] 或 [B, A, C]，取决于具体实现, RRF 是一种简单有效的排序融合算法，适用于多种场景，能够有效提升排序结果的准确性和鲁棒性。

2-3-2、相关查询问题的生成

概述： 加载网页内容、分割文档、向量化文档并创建向量存储、初始化LLM与之前部分相同。相关查询问题的提示词模板稍有不同，如下所示。

# RAG-Fusion: Related
template = """You are a helpful assistant that generates multiple search queries based on a single input query. \n
Generate multiple search queries related to: {question} \n
Output (4 queries):"""
prompt_rag_fusion = ChatPromptTemplate.from_template(template)

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

generate_queries = (
    prompt_rag_fusion
    | llm
    | StrOutputParser()
    | (lambda x: x.split("\n"))
)

generate_queries.invoke({"question": '论如何解决提出无理要求的员工？'})

输出：
[‘1. 如何有效沟通处理员工的不合理要求’, ‘2. 企业应对员工无理请求的最佳实践’, ‘3. 解决工作场所中不切实际的员工诉求的方法’, ‘4. 管理技巧：面对员工过分要求时的策略与解决方案’]

2-3-3、检索相关文档

reciprocal_rank_fusion： 倒数排名融合（RRF）函数（1、融合分数计算 2、重新排序文档，即按照融合分数对文档进行降序排序。）

• results：一个包含多个排序列表的列表，每个子列表包含一组排序后的文档。
• k：RRF 公式中的平滑常数（默认值为 60）。主要是为了防止除0错误，并且平衡低排名文档的影响。

from langchain.load import dumps, loads
from operator import itemgetter

def reciprocal_rank_fusion(results: list[list], k=60):
    """ Reciprocal_rank_fusion that takes multiple lists of ranked documents
        and an optional parameter k used in the RRF formula """

    # Initialize a dictionary to hold fused scores for each unique document
    fused_scores = {}

    # Iterate through each list of ranked documents
    for docs in results:
        # Iterate through each document in the list, with its rank (position in the list)
        for rank, doc in enumerate(docs):
            # Convert the document to a string format to use as a key (assumes documents can be serialized to JSON)
            # 使用 dumps 将文档序列化为字符串，以便用作字典的键。
            doc_str = dumps(doc)
            # If the document is not yet in the fused_scores dictionary, add it with an initial score of 0
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            # Retrieve the current score of the document, if any
            previous_score = fused_scores[doc_str]
            # Update the score of the document using the RRF formula: 1 / (rank + k)
            fused_scores[doc_str] += 1 / (rank + k)

    # Sort the documents based on their fused scores in descending order to get the final reranked results
    # 重新排序后，使用 loads 将文档字符串反序列化为原始格式。
    reranked_results = [
        (loads(doc), score)
        for doc, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    ]

    # Return the reranked results as a list of tuples, each containing the document and its fused score
    return reranked_results


question = "What is task decomposition for LLM agents?"
retrieval_chain_rag_fusion = generate_queries | retriever.map() | reciprocal_rank_fusion
docs = retrieval_chain_rag_fusion.invoke({"question": question})
print(docs)

输出：

2-3-4、构建RAG链

# RAG
template = """Answer the following question based on this context:

{context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

final_rag_chain = (
    {"context": retrieval_chain_rag_fusion,
     "question": itemgetter("question")}
    | prompt
    | llm
    | StrOutputParser()
)
print(final_rag_chain.invoke({"question":question}))

输出：

2-4、Decomposition（将复杂问题进行拆分）（子问题之间关联）

概述： Decomposition的目标是将一个复杂问题分解为多个子问题，后续的子问题会用到前边的子问题的结果，适用于需要分步骤解决复杂问题的场景。

流程：

• 使用 LLM 生成多个子问题（generate_queries_decomposition）。
• 对每个子问题分别检索和生成答案，除了第一个子问题之外，其他子问题都会参考上一步的子问题+答案。
• 最后一个子问题的答案为最终答案。

2-4-1、子问题生成

概述： 加载网页内容、分割文档、向量化文档并创建向量存储、初始化LLM与之前部分相同。分解任务的提示词模板不同，如下所示。

from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Decomposition
template = """You are a helpful assistant that generates multiple sub-questions related to an input question. \n
The goal is to break down the input into a set of sub-problems / sub-questions that can be answers in isolation. \n
Generate multiple search queries related to: {question} \n
Output (3 queries):"""
prompt_decomposition = ChatPromptTemplate.from_template(template)

# Chain
generate_queries_decomposition = ( prompt_decomposition | llm | StrOutputParser() | (lambda x: x.split("\n")))

# Run
question = "如何学习好人工智能?"
questions = generate_queries_decomposition.invoke({"question":question})

print(questions)

输出：

[‘1. 人工智能的基础知识有哪些？’, ‘2. 学习人工智能需要掌握哪些编程语言和技术？’, ‘3. 有哪些优质的人工智能在线课程和资源？’]

2-4-2、RAG Chain的构建

方法优势：

• 对每个子问题分别进行检索，获取到RAG中相关的上下文信息
• 在回答每个子问题时，会动态的整合之前生成的子问题及其答案作为背景知识，帮助模型更好的理解问题的全貌。

template = """Here is the question you need to answer:
\n --- \n {question} \n --- \n
Here is any available background question + answer pairs:
\n --- \n {q_a_pairs} \n --- \n
Here is additional context relevant to the question: 
\n --- \n {context} \n --- \n
Use the above context and any background question + answer pairs to answer the question: \n {question}
"""

decomposition_prompt = ChatPromptTemplate.from_template(template)

from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser


# 构建问题和答案对
def format_qa_pair(question, answer):
    """Format Q and A pair"""

    formatted_string = ""
    formatted_string += f"Question: {question}\nAnswer: {answer}\n\n"
    return formatted_string.strip()

q_a_pairs = ""
for q in questions:
    rag_chain = (
            {"context": itemgetter("question") | retriever,
             "question": itemgetter("question"),
             "q_a_pairs": itemgetter("q_a_pairs")}
            | decomposition_prompt
            | llm
            | StrOutputParser())

    answer = rag_chain.invoke({"question": q, "q_a_pairs": q_a_pairs})
    q_a_pair = format_qa_pair(q, answer)
    q_a_pairs = q_a_pairs + "\n---\n" + q_a_pair

print(answer)

2-5、Decomposition（将复杂问题进行拆分）（子问题之间无关）

概述： Decomposition的目标是将一个复杂问题分解为多个子问题，然后分别回答这些子问题，最后整合答案。适用于需要分步骤解决复杂问题的场景。

流程：

• 使用 LLM 生成多个子问题（generate_queries_decomposition）。
• 对每个子问题分别检索和生成答案。
• 将子问题的答案整合为最终答案。

参考文章：
rag-from-scratch 官方GitHub仓库.

---

总结

不顺路的话，那就祝你以后一路顺风。☀