第3章 RAG下: 与自有数据聊天 | Alan Hou的个人博客

上一章中，我们学习了如何处理数据并在向量数据库中创建和存储嵌入。在本章中，我们将学习如何根据用户的查询高效地检索最相关的嵌入和文档块。这样就能够构建一个包含相关文档作为上下文的提示，从而提高 LLM 最终输出的准确性。

这个过程——包括嵌入用户查询、从数据源检索相似文档，然后将它们作为上下文传递给发送到 LLM 的提示——正式名称为检索增强生成 (RAG)。

RAG 是构建准确、高效且保持最新状态的聊天型 LLM 应用的重要组成部分。在本章中，我们将从基础知识逐步学习到高级策略，为各种数据源（如向量存储和数据库）和数据结构（结构化和非结构化）构建一个有效的 RAG 系统。

但首先，让我们来定义 RAG 并讨论它的优点。

检索增强生成入门

RAG是一种通过提供来自外部数据源的上下文来增强 LLM 生成输出准确性的技术。这个术语最初由 Meta AI 的研究人员在一篇论文中创造，他们发现支持 RAG 的模型比非 RAG 模型更具事实性和具体性。

没有 RAG，LLM 仅依赖其预训练数据，这些数据可能已经过时。例如，比如我们问 ChatGPT 一个关于时事的问题，看看它的回答：

输入

哪个国家是最新一届的男足世界杯冠军？

1	哪个国家是最新一届的男足世界杯冠军？

输出

最近一届的世界杯冠军是法国队，他们在2018年赢得了该项赛事。

1	最近一届的世界杯冠军是法国队，他们在2018年赢得了该项赛事。

LLM 的回答在事实上是不正确且过时的。在写本书时，最新的冠军是阿根廷队，他们在 2022 年赢得了世界杯。虽然这个示例问题可能微不足道，但如果依赖其答案进行事实核查或重要决策，LLM 的幻觉可能会带来灾难性的后果。

为了解决这个问题，我们需要为 LLM 提供事实准确、最新的信息，以便它能够形成一个准确的回答。继续前面的例子，让我们访问维基百科上关于FIFA 世界杯的页面，复制引言段落，然后将其作为上下文附加到我们给 ChatGPT 的提示中：

哪个国家是最新一届的男足世界杯冠军？

请参考下面的上下文。

国际足联世界杯（FIFA World Cup），通常简称为世界杯，是由国际足球联合会（FIFA）成员国的成年男子国家队参与的国际足球比赛，FIFA 是这项运动的全球管理机构。该赛事自 1930 年首届比赛以来每四年举办一次，除了因第二次世界大战而在 1942 年和 1946 年停办。卫冕冠军是阿根廷队，他们在 2022 年的比赛中赢得了他们的第三个冠军头衔。

哪个国家是最新一届的男足世界杯冠军？

请参考下面的上下文。

国际足联世界杯（FIFA World Cup），通常简称为世界杯，是由国际足球联合会（FIFA）成员国的成年男子国家队参与的国际足球比赛，FIFA 是这项运动的全球管理机构。该赛事自 1930 年首届比赛以来每四年举办一次，除了因第二次世界大战而在 1942 年和 1946 年停办。卫冕冠军是阿根廷队，他们在 2022 年的比赛中赢得了他们的第三个冠军头衔。

请注意，最后一句包含了 LLM 可以用来提供准确答案的必要上下文。这是 LLM 的回答：

最新一届的男足世界杯冠军是阿根廷队，他们在2022年的比赛中赢得了他们的第三个冠军头衔。

1	最新一届的男足世界杯冠军是阿根廷队，他们在2022年的比赛中赢得了他们的第三个冠军头衔。

由于提供了最新的附加上下文，LLM 能够对提示生成一个准确的回答。但是，对于一个生产环境的 AI 应用来说，复制和粘贴相关信息作为上下文既不实际也不可扩展。我们需要一个自动化系统，根据用户的查询获取相关信息，将其作为上下文附加到提示中，然后向 LLM 执行生成请求。

检索相关文档

用于 AI 应用的 RAG 系统通常有三个核心阶段：

索引（Indexing）: 这一阶段涉及预处理外部数据源，并将代表数据的嵌入存储在向量数据库中，以便于检索。
检索（Retrieval）: 这一阶段涉及根据用户的查询从向量数据库中检索相关的嵌入和数据。
生成（Generation）: 这一阶段涉及将原始提示与检索到的相关文档合成为一个最终提示，发送给模型进行预测。

这三个基本阶段如图 3-1 所示。

第3章 RAG下: 与自有数据聊天

图 3-1. RAG 的主要阶段

其中索引阶段已在第 2 章中详细介绍，我们学习了如何使用文档加载器、文本分割器、嵌入和向量存储。

让我们从头从索引阶段开始运行一个例子：

Python

from langchain_community.document_loaders import TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_postgres.vectorstores import PGVector

# Load the document, split it into chunks
raw_documents = TextLoader('./test.txt', encoding='utf-8').load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, 
    chunk_overlap=200)
documents = text_splitter.split_documents(raw_documents)

# embed each chunk and insert it into the vector store
model = OpenAIEmbeddings()
connection = 'postgresql+psycopg://langchain:langchain@localhost:6024/langchain'
db = PGVector.from_documents(documents, model, connection=connection)

from langchain_community.document_loaders import TextLoader

from langchain_openai import OpenAIEmbeddings

from langchain_text_splitters import RecursiveCharacterTextSplitter

from langchain_postgres.vectorstores import PGVector

# Load the document, split it into chunks

raw_documents = TextLoader('./test.txt', encoding='utf-8').load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000,

chunk_overlap=200)

documents = text_splitter.split_documents(raw_documents)

# embed each chunk and insert it into the vector store

model = OpenAIEmbeddings()

connection = 'postgresql+psycopg://langchain:langchain@localhost:6024/langchain'

db = PGVector.from_documents(documents, model, connection=connection)

JavaScript

import { TextLoader } from "langchain/document_loaders/fs/text";
import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters";
import { OpenAIEmbeddings } from "@langchain/openai";
import { PGVectorStore } from "@langchain/community/vectorstores/pgvector";

// Load the document, split it into chunks
const loader = new TextLoader("./test.txt");
const raw_docs = await loader.load();
const splitter = new RecursiveCharacterTextSplitter({
  chunkSize: 1000,
  chunkOverlap: 200,
});
const docs = await splitter.splitDocuments(docs)

// embed each chunk and insert it into the vector store
const model = new OpenAIEmbeddings();
const db = await PGVectorStore.fromDocuments(docs, model, {
  postgresConnectionOptions: {
    connectionString: 'postgresql://langchain:langchain@localhost:6024/langchain'
  }
})

import { TextLoader } from "langchain/document_loaders/fs/text";

import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters";

import { OpenAIEmbeddings } from "@langchain/openai";

import { PGVectorStore } from "@langchain/community/vectorstores/pgvector";

// Load the document, split it into chunks

const loader = new TextLoader("./test.txt");

const raw_docs = await loader.load();

const splitter = new RecursiveCharacterTextSplitter({

chunkSize: 1000,

chunkOverlap: 200,

});

const docs = await splitter.splitDocuments(docs)

// embed each chunk and insert it into the vector store

const model = new OpenAIEmbeddings();

const db = await PGVectorStore.fromDocuments(docs, model, {

postgresConnectionOptions: {

connectionString: 'postgresql://langchain:langchain@localhost:6024/langchain'

}

})

第 2 章中对索引阶段有更详细的介绍。

索引阶段现已完成。为了执行检索阶段，我们需要在用户的查询和我们存储的嵌入之间执行相似性搜索计算——例如余弦相似度——这样我们索引文档的相关块就会被检索出来（参见图 3-2）。

第3章 RAG下: 与自有数据聊天

图 3-2. 索引文档以及从向量存储中检索相关文档的流程示例；Hierarchical Navigable Small World (HNSW) 框描绘了计算文档与用户查询的相似度

图 3-2阐释了检索过程的步骤：

将用户查询转换为嵌入。
计算向量存储中与用户查询最相似的嵌入。
检索相关的文档嵌入及其对应的文本块。

我们可以使用 LangChain 以编程方式表示这些步骤，如下所示：

Python

# 创建检索器
retriever = db.as_retriever()

# 获取相关文档
docs = retriever.invoke("""古希腊哲学史上的关键人物是谁？""")

# 创建检索器

retriever = db.as_retriever()

# 获取相关文档

docs = retriever.invoke("""古希腊哲学史上的关键人物是谁？""")

JavaScript

// 创建检索器
const retriever = db.asRetriever()

// 获取相关文档
const docs = await retriever.invoke(`古希腊哲学史上的关键人物是谁？`)

// 创建检索器

const retriever = db.asRetriever()

// 获取相关文档

const docs = await retriever.invoke(`古希腊哲学史上的关键人物是谁？`)

我们使用了一个此前用过的向量存储方法：as_retriever。这个函数抽象了嵌入用户查询的逻辑以及向量存储为检索相关文档所执行的底层相似性搜索计算。

还有一个参数 k，用于指定从向量存储中获取相关文档的数量。例如：

Python

# 创建 k=2 的检索器
retriever = db.as_retriever(search_kwargs={"k": 2})

# 获取 2 个最相关的文档
docs = retriever.invoke("""古希腊哲学史上的关键人物是谁？""")

# 创建 k=2 的检索器

retriever = db.as_retriever(search_kwargs={"k": 2})

# 获取 2 个最相关的文档

docs = retriever.invoke("""古希腊哲学史上的关键人物是谁？""")

JavaScript

# 创建 k=2 的检索器
const retriever = db.asRetriever({k: 2})

# 获取 2 个最相关的文档
const docs = await retriever.invoke(`古希腊哲学史上的关键人物是谁？`)

# 创建 k=2 的检索器

const retriever = db.asRetriever({k: 2})

# 获取 2 个最相关的文档

const docs = await retriever.invoke(`古希腊哲学史上的关键人物是谁？`)

上例中，参数 k 被指定为 2。这样向量数据库根据用户的查询会返回两个最相关的文档。

使用较低的 k 值可能看起来不符合直觉，但检索更多文档并不一定更好。检索的文档越多，应用程序性能就会越慢，提示（以及相关的生成成本）就会越大，并且检索到包含不相关信息的文本块的可能性也会增加，这会导致 LLM 产生幻觉。

我们已经完成了 RAG 系统的检索阶段，下面进入最后的生成阶段。

使用相关文档生成 LLM 预测

我们根据用户的查询检索到相关文档后，最后一步是将其作为上下文添加到原始提示中，然后调用模型生成最终输出（图 3-3）。

第3章 RAG下: 与自有数据聊天

图 3-3. 一个流程示例，演示了索引文档、从向量存储中检索相关文档，并将检索到的文档作为上下文包含在 LLM 提示中

我们在之前示例代码的基础上做修改：

Python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

retriever = db.as_retriever()

prompt = ChatPromptTemplate.from_template("""Answer the question based only on 
    the following context:
{context}

Question: {question}
""")

llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)

chain = prompt | llm

# fetch relevant documents 
docs = retriever.get_relevant_documents("""Who are the key figures in the 
    ancient greek history of philosophy?""")

# run
chain.invoke({"context": docs,"question": """Who are the key figures in the 
    ancient greek history of philosophy?"""})

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate

retriever = db.as_retriever()

prompt = ChatPromptTemplate.from_template("""Answer the question based only on

the following context:

{context}

Question: {question}

""")

llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)

chain = prompt | llm

# fetch relevant documents

docs = retriever.get_relevant_documents("""Who are the key figures in the

ancient greek history of philosophy?""")

# run

chain.invoke({"context": docs,"question": """Who are the key figures in the

ancient greek history of philosophy?"""})

JavaScript

import {ChatOpenAI} from '@langchain/openai'
import {ChatPromptTemplate} from '@langchain/core/prompts'

const retriever = db.asRetriever()

const prompt = ChatPromptTemplate.fromTemplate(`Answer the question based only 
  on the following context:
{context}

Question: {question}
`)

const llm = new ChatOpenAI({temperature: 0, modelName: 'gpt-4o-mini'})

const chain = prompt.pipe(llm)

// fetch relevant documents
const docs = await retriever.invoke(`Who are the key figures in the ancient 
  greek history of philosophy?`)

await chain.invoke({context: docs, question: `Who are the key figures in the 
  ancient greek history of philosophy?`})

import {ChatOpenAI} from '@langchain/openai'

import {ChatPromptTemplate} from '@langchain/core/prompts'

const retriever = db.asRetriever()

const prompt = ChatPromptTemplate.fromTemplate(`Answer the question based only

on the following context:

{context}

Question: {question}

const llm = new ChatOpenAI({temperature: 0, modelName: 'gpt-4o-mini'})

const chain = prompt.pipe(llm)

// fetch relevant documents

const docs = await retriever.invoke(`Who are the key figures in the ancient

greek history of philosophy?`)

await chain.invoke({context: docs, question: `Who are the key figures in the

ancient greek history of philosophy?`})

注意以下的修改：

在提示中实现了动态的context和question变量，通过它们定义了一个ChatPromptTemplate，模型可以用它来生成响应。
定义一个ChatOpenAI接口作为我们的LLM。温度设置为0，以消除模型输出中的创造性。
创建一个链来组合提示和LLM。提示一下：|运算符（或JS中的pipe方法）将prompt的输出用作llm的输入。
调用(invoke)链，传入context变量（我们所检索到的相关文档）和用户的问题，生成最终输出。

我们可以将这个检索逻辑封装在一个函数中：

Python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain

retriever = db.as_retriever()

prompt = ChatPromptTemplate.from_template("""Answer the question based only on 
    the following context:
{context}

Question: {question}
""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

@chain
def qa(input):
    # fetch relevant documents 
    docs = retriever.get_relevant_documents(input)
    # format prompt
    formatted = prompt.invoke({"context": docs, "question": input})
    # generate answer
    answer = llm.invoke(formatted)
    return answer

# run
qa.invoke("Who are the key figures in the ancient greek history of philosophy?")

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.runnables import chain

retriever = db.as_retriever()

prompt = ChatPromptTemplate.from_template("""Answer the question based only on

the following context:

{context}

Question: {question}

""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

@chain

def qa(input):

# fetch relevant documents

docs = retriever.get_relevant_documents(input)

# format prompt

formatted = prompt.invoke({"context": docs, "question": input})

# generate answer

answer = llm.invoke(formatted)

return answer

# run

qa.invoke("Who are the key figures in the ancient greek history of philosophy?")

JavaScript

import {ChatOpenAI} from '@langchain/openai'
import {ChatPromptTemplate} from '@langchain/core/prompts'
import {RunnableLambda} from '@langchain/core/runnables'

const retriever = db.asRetriever()

const prompt = ChatPromptTemplate.fromTemplate(`Answer the question based only 
  on the following context:
{context}

Question: {question}
`)

const llm = new ChatOpenAI({temperature: 0, modelName: 'gpt-4o-mini'})

const qa = RunnableLambda.from(async input => {
  // fetch relevant documents
  const docs = await retriever.invoke(input)
  // format prompt
  const formatted = await prompt.invoke({context: docs, question: input})
  // generate answer
  const answer = await llm.invoke(formatted)
  return answer
})

await qa.invoke(`Who are the key figures in the ancient greek history of philosophy?`)

import {ChatOpenAI} from '@langchain/openai'

import {ChatPromptTemplate} from '@langchain/core/prompts'

import {RunnableLambda} from '@langchain/core/runnables'

const retriever = db.asRetriever()

const prompt = ChatPromptTemplate.fromTemplate(`Answer the question based only

on the following context:

{context}

Question: {question}

const llm = new ChatOpenAI({temperature: 0, modelName: 'gpt-4o-mini'})

const qa = RunnableLambda.from(async input => {

// fetch relevant documents

const docs = await retriever.invoke(input)

// format prompt

const formatted = await prompt.invoke({context: docs, question: input})

// generate answer

const answer = await llm.invoke(formatted)

return answer

})

await qa.invoke(`Who are the key figures in the ancient greek history of philosophy?`)

注意，我们现在有一个新的可运行函数（runnable）qa，它只需要一个问题就可以调用，并负责先获取相关的上下文文档，将它们格式化到提示中，最后生成答案。在 Python 代码中，@chain 装饰器将函数变成一个可运行的链。这种将多个步骤封装成一个函数的概念，对于用 LLM 构建有趣的应用至关重要。

还可以返回检索到的文档以供进一步处理：

Python

@chain
def qa(input):
    # 获取相关文档
    docs = retriever.get_relevant_documents(input)
    # 格式化提示
    formatted = prompt.invoke({"context": docs, "question": input})
    # 生成回答
    answer = llm.invoke(formatted)
    return {"answer": answer, "docs": docs}

@chain

def qa(input):

# 获取相关文档

docs = retriever.get_relevant_documents(input)

# 格式化提示

formatted = prompt.invoke({"context": docs, "question": input})

# 生成回答

answer = llm.invoke(formatted)

return {"answer": answer, "docs": docs}

JavaScript

const qa = RunnableLambda.from(async input => {
  # 获取相关文档
  const docs = await retriever.invoke(input)
  # 格式化提示
  const formatted = await prompt.invoke({context: docs, question: input})
  # 生成回答
  const answer = await llm.invoke(formatted)
  return {answer, docs}
})

const qa = RunnableLambda.from(async input => {

# 获取相关文档

const docs = await retriever.invoke(input)

# 格式化提示

const formatted = await prompt.invoke({context: docs, question: input})

# 生成回答

const answer = await llm.invoke(formatted)

return {answer, docs}

})

恭喜！现在已经构建了一个基本的 RAG 系统，可以为个人使用的 AI 应用提供支持。

但供多个用户使用的用于生产的 AI 应用需要更先进的 RAG 系统。为构建一套稳健的 RAG 系统，我们需要有效地回答以下问题：

如何处理用户输入质量的变化？
如何路由查询以从各种数据源中检索相关数据？
如何将自然语言转换为目标数据源的查询语言？
如何优化我们的索引过程，即嵌入、文本分割？

接下来，我们将讨论最新的、有研究支撑的策略来回答这些问题，并构建一个可用于生产的 RAG 系统。这些策略可以总结在图 3-4 中。

第3章 RAG下: 与自有数据聊天

图 3-4. 优化 RAG 系统准确性的有效策略

注：本章后续代码块都使用我们在本章开头设置的向量数据库。

查询转换

基础 RAG 系统的主要问题之一是它过于依赖用户查询的质量来生成准确的输出。在生产环境中，用户很可能以不完整、模棱两可或措辞不当的方式构建查询，这会导致模型产生幻觉。

查询转换是一系列用于修改用户输入的策略，以回答第一个 RAG 问题：我们如何处理用户输入质量的变化？图 3-5说明了查询转换策略的范围，修改用户输入的抽象程度，以生成准确的 LLM 输出。下一节先从一个中间策略开始。

第3章 RAG下: 与自有数据聊天

图 3-5. 基于抽象级别转换用户查询的各种方法

重写-检索-读取

微软研究院团队提出的“重写-检索-读取”（Rewrite-Retrieve-Read）策略，简单地提示 LLM 在执行检索前重写用户的查询。为了说明这一点，让我们回到上一节中构建的链，这次使用一个措辞不佳的用户查询来调用它：

Python

@chain
def qa(input):
    # 获取相关文档
    docs = retriever.get_relevant_documents(input)
    # 格式化提示
    formatted = prompt.invoke({"context": docs, "question": input})
    # 生成回答
    answer = llm.invoke(formatted)
    return answer

qa.invoke("""Today I woke up and brushed my teeth, then I sat down to read the 
    news. But then I forgot the food on the cooker. Who are some key figures in 
    the ancient greek history of philosophy?""")

@chain

def qa(input):

# 获取相关文档

docs = retriever.get_relevant_documents(input)

# 格式化提示

formatted = prompt.invoke({"context": docs, "question": input})

# 生成回答

answer = llm.invoke(formatted)

return answer

qa.invoke("""Today I woke up and brushed my teeth, then I sat down to read the

news. But then I forgot the food on the cooker. Who are some key figures in

the ancient greek history of philosophy?""")

JavaScript

const qa = RunnableLambda.from(async input => {
  // 获取相关文档
  const docs = await retriever.invoke(input)
  // 格式化提示
  const formatted = await prompt.invoke({context: docs, question: input})
  // 生成回答
  const answer = await llm.invoke(formatted)
  return answer
})

await qa.invoke(`Today I woke up and brushed my teeth, then I sat down to read 
  the news. But then I forgot the food on the cooker. Who are some key figures 
  in the ancient greek history of philosophy?`)

const qa = RunnableLambda.from(async input => {

// 获取相关文档

const docs = await retriever.invoke(input)

// 格式化提示

const formatted = await prompt.invoke({context: docs, question: input})

// 生成回答

const answer = await llm.invoke(formatted)

return answer

})

await qa.invoke(`Today I woke up and brushed my teeth, then I sat down to read

the news. But then I forgot the food on the cooker. Who are some key figures

in the ancient greek history of philosophy?`)

输出（如果重新运行，输出可能与此不同）：

Based on the given context, there is no information provided.

1	Based on the given context, there is no information provided.

模型未能回答问题，因为它被用户查询中提供的不相关信息分散了注意力。

现在让我们实现“重写-检索-读取”提示：

Python

rewrite_prompt = ChatPromptTemplate.from_template("""Provide a better search 
    query for web search engine to answer the given question, end the queries 
    with ’**’. Question: {x} Answer:""")

def parse_rewriter_output(message):
    return message.content.strip('"').strip("**")

rewriter = rewrite_prompt | llm | parse_rewriter_output

@chain
def qa_rrr(input):
    # rewrite the query
    new_query = rewriter.invoke(input)
    # fetch relevant documents 
    docs = retriever.get_relevant_documents(new_query)
    # format prompt
    formatted = prompt.invoke({"context": docs, "question": input})
    # generate answer
    answer = llm.invoke(formatted)
    return answer

# run
qa_rrr.invoke("""Today I woke up and brushed my teeth, then I sat down to read 
    the news. But then I forgot the food on the cooker. Who are some key 
    figures in the ancient greek history of philosophy?""")

rewrite_prompt = ChatPromptTemplate.from_template("""Provide a better search

query for web search engine to answer the given question, end the queries

with ’**’. Question: {x} Answer:""")

def parse_rewriter_output(message):

return message.content.strip('"').strip("**")

rewriter = rewrite_prompt | llm | parse_rewriter_output

@chain

def qa_rrr(input):

# rewrite the query

new_query = rewriter.invoke(input)

# fetch relevant documents

docs = retriever.get_relevant_documents(new_query)

# format prompt

formatted = prompt.invoke({"context": docs, "question": input})

# generate answer

answer = llm.invoke(formatted)

return answer

# run

qa_rrr.invoke("""Today I woke up and brushed my teeth, then I sat down to read

the news. But then I forgot the food on the cooker. Who are some key

figures in the ancient greek history of philosophy?""")

JavaScript

const rewritePrompt = ChatPromptTemplate.fromTemplate(`Provide a better search 
  query for web search engine to answer the given question, end the queries 
  with ’**’. Question: {question} Answer:`)

const rewriter = rewritePrompt.pipe(llm).pipe(message => {
  return message.content.replaceAll('"', '').replaceAll('**')
})

const qa = RunnableLambda.from(async input => {
  const newQuery = await rewriter.invoke({question: input});
  // fetch relevant documents
  const docs = await retriever.invoke(newQuery)
  // format prompt
  const formatted = await prompt.invoke({context: docs, question: input})
  // generate answer
  const answer = await llm.invoke(formatted)
  return answer
})

await qa.invoke(`Today I woke up and brushed my teeth, then I sat down to read 
  the news. But then I forgot the food on the cooker. Who are some key 
  figures in the ancient greek history of philosophy?`)

const rewritePrompt = ChatPromptTemplate.fromTemplate(`Provide a better search

query for web search engine to answer the given question, end the queries

with ’**’. Question: {question} Answer:`)

const rewriter = rewritePrompt.pipe(llm).pipe(message => {

return message.content.replaceAll('"', '').replaceAll('**')

})

const qa = RunnableLambda.from(async input => {

const newQuery = await rewriter.invoke({question: input});

// fetch relevant documents

const docs = await retriever.invoke(newQuery)

// format prompt

const formatted = await prompt.invoke({context: docs, question: input})

// generate answer

const answer = await llm.invoke(formatted)

return answer

})

await qa.invoke(`Today I woke up and brushed my teeth, then I sat down to read

the news. But then I forgot the food on the cooker. Who are some key

figures in the ancient greek history of philosophy?`)

输出：

Based on the given context, some key figures in the ancient greek history of 
philosophy include: Themistocles (an Athenian statesman), Pythagoras, and Plato.

1 2	Based on the given context, some key figures in the ancient greek history of philosophy include: Themistocles (an Athenian statesman), Pythagoras, and Plato.

注意，我们让一个 LLM 将用户最初分心的查询重写成一个更清晰的查询，然后将这个查询传递给检索器获取最相关的文档。注：此技术可与任何检索方法一起使用，无论是像我们这里的向量存储，还是网络搜索工具等。这种方法的缺点是它会在链中引入额外的延迟，因为现在我们需要按顺序执行两次 LLM 调用。

多查询检索

用户的单个查询可能不足以捕捉到全面回答查询所需的全部信息。多查询检索策略通过指示 LLM 根据用户的初始查询生成多个查询，并行执行每个查询对数据源的检索，然后将检索到的结果作为提示上下文插入以生成最终模型输出来解决此问题。如图 3-6所示。

第3章 RAG下: 与自有数据聊天

图 3-6. 多查询检索策略演示

此策略特别适用于依赖多个视角来提供全面回答的情况。

以下是多查询检索的实际代码示例：

Python

from langchain.prompts import ChatPromptTemplate

perspectives_prompt = ChatPromptTemplate.from_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}""")

def parse_queries_output(message):
    return message.content.split('\n')

query_gen = perspectives_prompt | llm | parse_queries_output

from langchain.prompts import ChatPromptTemplate

perspectives_prompt = ChatPromptTemplate.from_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}""")

def parse_queries_output(message):

return message.content.split('\n')

query_gen = perspectives_prompt | llm | parse_queries_output

JavaScript

const perspectivesPrompt = ChatPromptTemplate.fromTemplate(`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}`)

const queryGen = perspectivesPrompt.pipe(llm).pipe(message => {
  return message.content.split('\n')
})

const perspectivesPrompt = ChatPromptTemplate.fromTemplate(`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}`)

const queryGen = perspectivesPrompt.pipe(llm).pipe(message => {

return message.content.split('\n')

})

因为我们使用多个（相关的）查询从同一个检索器中检索文档，所以很可能有一些文档是重复的。在将它们用作上下文来回答问题之前，我们需要对它们进行去重，以得到每个文档的单个实例。在这里，我们通过使用文档的内容（一个字符串）作为字典（或 JS 中的对象）的键来去重，因为一个字典对于每个键只能包含一个条目。在遍历了所有文档之后，我们只需获取所有字典的值，这样就得到了没有重复的文档列表。

还要注意我们使用了.batch，它并行运行所有生成的查询并返回一个结果列表——本例中为一个文档列表的列表，然后我们如前所述将其精简和去重。

最后一步是构建一个提示，包含用户的问题和组合检索到的相关文档，以及一个模型接口来生成预测：

Python

def get_unique_union(document_lists):
    # Flatten list of lists, and dedupe them
    deduped_docs = {
        doc.page_content: doc
        for sublist in document_lists for doc in sublist
    }
    # return a flat list of unique docs
    return list(deduped_docs.values())

retrieval_chain = query_gen | retriever.batch | get_unique_union

def get_unique_union(document_lists):

# Flatten list of lists, and dedupe them

deduped_docs = {

doc.page_content: doc

for sublist in document_lists for doc in sublist

}

# return a flat list of unique docs

return list(deduped_docs.values())

retrieval_chain = query_gen | retriever.batch | get_unique_union

JavaScript

const retrievalChain = queryGen
  .pipe(retriever.batch.bind(retriever))
  .pipe(documentLists => {
    const dedupedDocs = {}
    documentLists.flat().forEach(doc => {
      dedupedDocs[doc.pageContent] = doc
    })
    return Object.values(dedupedDocs)
  })

const retrievalChain = queryGen

.pipe(retriever.batch.bind(retriever))

.pipe(documentLists => {

const dedupedDocs = {}

documentLists.flat().forEach(doc => {

dedupedDocs[doc.pageContent] = doc

})

return Object.values(dedupedDocs)

})

这与我们之前的 QA 链并没有太大区别，因为多查询检索的所有新逻辑都在 retrieval_chain 中。这是充分利用这些技术的关键——将每种技术实现为一个独立的链（本例中为 retrieval_chain），这使得采用它们甚或组合更容易。

RAG-Fusion

RAG-Fusion 策略与多查询检索策略有相似之处，不同之处在于我们会对所有检索到的文档应用一个最终的重排序步骤。这个重排序步骤利用了倒排序融合 (RRF) 算法，该算法涉及将不同搜索结果的排名结合起来，产生一个单一的、统一的排名。通过结合来自不同查询的排名，我们将最相关的文档拉到最终列表的顶部。RRF 非常适合于合并可能具有不同量级或分布的查询结果。

我们用代码演示 RAG-Fusion。首先，制作一个类似于多查询检索策略的提示，根据用户查询生成一个查询列表：

Python

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt_rag_fusion = ChatPromptTemplate.from_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):""")

def parse_queries_output(message):
    return message.content.split('\n')

llm = ChatOpenAI(temperature=0)

query_gen = prompt_rag_fusion | llm | parse_queries_output

from langchain.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

prompt_rag_fusion = ChatPromptTemplate.from_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):""")

def parse_queries_output(message):

return message.content.split('\n')

llm = ChatOpenAI(temperature=0)

query_gen = prompt_rag_fusion | llm | parse_queries_output

JavaScript

import {ChatPromptTemplate} from '@langchain/core/prompts';
import {ChatOpenAI} from '@langchain/openai';
import {RunnableLambda} from '@langchain/core/runnables';

const perspectivesPrompt = ChatPromptTemplate.fromTemplate(`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):`)

const queryGen = perspectivesPrompt.pipe(llm).pipe(message => {
  return message.content.split('\n')
})

import {ChatPromptTemplate} from '@langchain/core/prompts';

import {ChatOpenAI} from '@langchain/openai';

import {RunnableLambda} from '@langchain/core/runnables';

const perspectivesPrompt = ChatPromptTemplate.fromTemplate(`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):`)

const queryGen = perspectivesPrompt.pipe(llm).pipe(message => {

return message.content.split('\n')

})

生成查询后，我们就可以对每个查询获取相关文档，并将它们传递给一个函数，重排（即根据相关性重排序）最终的相关文档列表。

函数 reciprocal_rank_fusion 接收包含各个查询搜索结果的一个列表，即一个文档列表的列表，其中每个内部文档列表都按其与该查询的相关性排序。然后，RRF 算法根据每个文档在不同列表中的排名（或位置）计算一个新的分数，并对它们进行排序以创建一个最终的重排列表。

在计算了融合分数后，函数按这些分数的降序对文档进行排序，获取最终的重排列表，然后返回该列表：

Python

def reciprocal_rank_fusion(results: list[list], k=60):
    """reciprocal rank fusion on multiple lists of ranked documents 
       and an optional parameter k used in the RRF formula
    """
    
    # Initialize a dictionary to hold fused scores for each document
    # Documents will be keyed by their contents to ensure uniqueness
    fused_scores = {}
    documents = {}

    # 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):
            # Use the document contents as the key for uniqueness
            doc_str = doc.page_content
            # If the document hasn't been seen yet,
            # - initialize score to 0
            # - save it for later
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
                documents[doc_str] = doc
            # 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
    reranked_doc_strs = sorted(
        fused_scores, key=lambda d: fused_scores[d], reverse=True
    )
    # retrieve the corresponding doc for each doc_str
    return [
        documents[doc_str]
        for doc_str in reranked_doc_strs
    ]

retrieval_chain = generate_queries | retriever.batch | reciprocal_rank_fusion

def reciprocal_rank_fusion(results: list[list], k=60):

"""reciprocal rank fusion on multiple lists of ranked documents

and an optional parameter k used in the RRF formula

"""

# Initialize a dictionary to hold fused scores for each document

# Documents will be keyed by their contents to ensure uniqueness

fused_scores = {}

documents = {}

# 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):

# Use the document contents as the key for uniqueness

doc_str = doc.page_content

# If the document hasn't been seen yet,

# - initialize score to 0

# - save it for later

if doc_str not in fused_scores:

fused_scores[doc_str] = 0

documents[doc_str] = doc

# 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

reranked_doc_strs = sorted(

fused_scores, key=lambda d: fused_scores[d], reverse=True

)

# retrieve the corresponding doc for each doc_str

return [

documents[doc_str]

for doc_str in reranked_doc_strs

]

retrieval_chain = generate_queries | retriever.batch | reciprocal_rank_fusion

JavaScript

function reciprocalRankFusion(results, k = 60) {
  // Initialize a dictionary to hold fused scores for each document
  // Documents will be keyed by their contents to ensure uniqueness
  const fusedScores = {}
  const documents = {}

  results.forEach(docs => {
    docs.forEach((doc, rank) => {
      // Use the document contents as the key for uniqueness
      const key = doc.pageContent
      // If the document hasn't been seen yet,
      // - initialize score to 0
      // - save it for later
      if (!(key in fusedScores)) {
        fusedScores[key] = 0
        documents[key] = 0
      }
      // Update the score of the document using the RRF formula:
      // 1 / (rank + k)
      fusedScores[key] += 1 / (rank + k)
    })
  })

  // Sort the documents based on their fused scores in descending order 
  // to get the final reranked results
  const sorted = Object.entries(fusedScores).sort((a, b) => b[1] - a[1])
  // retrieve the corresponding doc for each key
  return sorted.map(([key]) => documents[key])
}

const retrievalChain = queryGen
  .pipe(retriever.batch.bind(retriever))
  .pipe(reciprocalRankFusion)

function reciprocalRankFusion(results, k = 60) {

// Initialize a dictionary to hold fused scores for each document

// Documents will be keyed by their contents to ensure uniqueness

const fusedScores = {}

const documents = {}

results.forEach(docs => {

docs.forEach((doc, rank) => {

// Use the document contents as the key for uniqueness

const key = doc.pageContent

// If the document hasn't been seen yet,

// - initialize score to 0

// - save it for later

if (!(key in fusedScores)) {

fusedScores[key] = 0

documents[key] = 0

}

// Update the score of the document using the RRF formula:

// 1 / (rank + k)

fusedScores[key] += 1 / (rank + k)

})

// Sort the documents based on their fused scores in descending order

// to get the final reranked results

const sorted = Object.entries(fusedScores).sort((a, b) => b[1] - a[1])

// retrieve the corresponding doc for each key

return sorted.map(([key]) => documents[key])

}

const retrievalChain = queryGen

.pipe(retriever.batch.bind(retriever))

.pipe(reciprocalRankFusion)

该函数还接受一个参数k，它决定了每个查询结果集中的文档对最终文档列表的影响程度。较高的值表示排名较低的文档具备更大的影响。

最后，我们将新的检索链（当前使用 RRF）与我们之前的完整链结合起来：

Python

prompt = ChatPromptTemplate.from_template("""Answer the following question based 
    on this context:

{context}

Question: {question}
""")

llm = ChatOpenAI(temperature=0)

@chain
def multi_query_qa(input):
    # fetch relevant documents 
    docs = retrieval_chain.invoke(input)
    # format prompt
    formatted = prompt.invoke({"context": docs, "question": input})
    # generate answer
    answer = llm.invoke(formatted)
    return answer

multi_query_qa.invoke("""Who are some key figures in the ancient greek history 
    of philosophy?""")

prompt = ChatPromptTemplate.from_template("""Answer the following question based

on this context:

{context}

Question: {question}

""")

llm = ChatOpenAI(temperature=0)

@chain

def multi_query_qa(input):

# fetch relevant documents

docs = retrieval_chain.invoke(input)

# format prompt

formatted = prompt.invoke({"context": docs, "question": input})

# generate answer

answer = llm.invoke(formatted)

return answer

multi_query_qa.invoke("""Who are some key figures in the ancient greek history

of philosophy?""")

JavaScript

const rewritePrompt = ChatPromptTemplate.fromTemplate(`Answer the following 
  question based on this context:

{context}

Question: {question}
`)

const llm = new ChatOpenAI({temperature: 0})

const multiQueryQa = RunnableLambda.from(async input => {
  // fetch relevant documents
  const docs = await retrievalChain.invoke(input)
  // format prompt
  const formatted = await prompt.invoke({context: docs, question: input})
  // generate answer
  const answer = await llm.invoke(formatted)
  return answer
})

await multiQueryQa.invoke(`Who are some key figures in the ancient greek 
  history of philosophy?`)

const rewritePrompt = ChatPromptTemplate.fromTemplate(`Answer the following

question based on this context:

{context}

Question: {question}

const llm = new ChatOpenAI({temperature: 0})

const multiQueryQa = RunnableLambda.from(async input => {

// fetch relevant documents

const docs = await retrievalChain.invoke(input)

// format prompt

const formatted = await prompt.invoke({context: docs, question: input})

// generate answer

const answer = await llm.invoke(formatted)

return answer

})

await multiQueryQa.invoke(`Who are some key figures in the ancient greek

history of philosophy?`)

RAG-Fusion 的优势在于它能够捕捉用户的意图表达，处理复杂查询，并拓宽检索文档的范围，从而实现意外发现。

假设性文档嵌入

假设性文档嵌入 (HyDE) 是一种策略，它涉及根据用户的查询创建一个假设性文档，嵌入该文档，并基于向量相似性检索相关文档。HyDE 背后的启发是，由 LLM 生成的假设性文档将比原始查询更接近最相关的文档，如图 3-7 所示。
第3章 RAG下: 与自有数据聊天

图 3-7. HyDE 在向量空间中比普通查询嵌入更接近文档嵌入的示意图

首先，定义一个提示来生成一个假设性文档：

Python

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

prompt_hyde = ChatPromptTemplate.from_template("""Please write a passage to 
   answer the question.\n Question: {question} \n Passage:""")

generate_doc = (
    prompt_hyde | ChatOpenAI(temperature=0) | StrOutputParser() 
)

from langchain.prompts import ChatPromptTemplate

from langchain_core.output_parsers import StrOutputParser

from langchain_openai import ChatOpenAI

prompt_hyde = ChatPromptTemplate.from_template("""Please write a passage to

answer the question.\n Question: {question} \n Passage:""")

generate_doc = (

prompt_hyde | ChatOpenAI(temperature=0) | StrOutputParser()

)

JavaScript

import {ChatOpenAI} from '@langchain/openai'
import {ChatPromptTemplate} from '@langchain/core/prompts'
import {RunnableLambda} from '@langchain/core/runnables';

const prompt = ChatPromptTemplate.fromTemplate(`Please write a passage to 
  answer the question
Question: {question}
Passage:`)

const llm = new ChatOpenAI({temperature: 0})

const generateDoc = prompt.pipe(llm).pipe(msg => msg.content)

import {ChatOpenAI} from '@langchain/openai'

import {ChatPromptTemplate} from '@langchain/core/prompts'

import {RunnableLambda} from '@langchain/core/runnables';

const prompt = ChatPromptTemplate.fromTemplate(`Please write a passage to

answer the question

Question: {question}

Passage:`)

const llm = new ChatOpenAI({temperature: 0})

const generateDoc = prompt.pipe(llm).pipe(msg => msg.content)

接下来，我们取这个假设性文档，并将其用作retriever的输入，retriever将生成其嵌入并在向量存储中搜索相似的文档：

Python

retrieval_chain = generate_doc | retriever

1	retrieval_chain = generate_doc \| retriever

JavaScript

const retrievalChain = generateDoc.pipe(retriever)

1	const retrievalChain = generateDoc.pipe(retriever)

最后，我们取检索到的文档，将它们作为上下文传递给最终的提示，并指示模型生成一个输出：

Python

prompt = ChatPromptTemplate.from_template("""Answer the following question based 
    on this context:

{context}

Question: {question}
""")

llm = ChatOpenAI(temperature=0)

@chain
def qa(input):
  # fetch relevant documents from the hyde retrieval chain defined earlier
  docs = retrieval_chain.invoke(input)
  # format prompt
  formatted = prompt.invoke({"context": docs, "question": input})
  # generate answer
  answer = llm.invoke(formatted)
  return answer

qa.invoke("""Who are some key figures in the ancient greek history of 
    philosophy?""")

prompt = ChatPromptTemplate.from_template("""Answer the following question based

on this context:

{context}

Question: {question}

""")

llm = ChatOpenAI(temperature=0)

@chain

def qa(input):

# fetch relevant documents from the hyde retrieval chain defined earlier

docs = retrieval_chain.invoke(input)

# format prompt

formatted = prompt.invoke({"context": docs, "question": input})

# generate answer

answer = llm.invoke(formatted)

return answer

qa.invoke("""Who are some key figures in the ancient greek history of

philosophy?""")

JavaScript

const prompt = ChatPromptTemplate.fromTemplate(`Answer the following 
  question based on this context:

{context}

Question: {question}
`)

const llm = new ChatOpenAI({temperature: 0})

const qa = RunnableLambda.from(async input => {
  // fetch relevant documents from the hyde retrieval chain defined earlier
  const docs = await retrievalChain.invoke(input)
  // format prompt
  const formatted = await prompt.invoke({context: docs, question: input})
  // generate answer
  const answer = await llm.invoke(formatted)
  return answer
})

await qa.invoke(`Who are some key figures in the ancient greek history of 
  philosophy?`)

const prompt = ChatPromptTemplate.fromTemplate(`Answer the following

question based on this context:

{context}

Question: {question}

const llm = new ChatOpenAI({temperature: 0})

const qa = RunnableLambda.from(async input => {

// fetch relevant documents from the hyde retrieval chain defined earlier

const docs = await retrievalChain.invoke(input)

// format prompt

const formatted = await prompt.invoke({context: docs, question: input})

// generate answer

const answer = await llm.invoke(formatted)

return answer

})

await qa.invoke(`Who are some key figures in the ancient greek history of

philosophy?`)

总结一下我们在本节中介绍的内容，查询转换包括获取用户的原始查询并执行以下操作：

重写为一个或多个查询
将这些查询的结果合并成一个最相关结果的单一集合

重写查询可以有多种形式，但通常以类似的方式完成：获取用户的原始查询——你写的提示——然后要求 LLM 编写一个新的查询或多个查询。一些典型的更改示例包括：

从查询中删除不相关/无关的文本。
用过去的对话历史来为查询提供依据。例如，为了理解像 那洛杉矶呢 这样的查询，我们需要将它与一个关于旧金山天气的假设性过去问题结合起来，以得到一个有用的查询，比如 洛杉矶的天气。
通过同时为相关查询获取文档，为相关文档撒下更广的网。
将一个复杂的问题分解成多个更简单的问题，然后在最终的提示中包含所有这些问题的结果来生成答案。

使用哪种重写策略将取决于实际用例。

现在我们已经介绍了主要的查询转换策略，下面来讨论构建一个稳健的 RAG 系统需要回答的第二个主要问题：我们如何路由查询以从多个数据源检索相关数据？

查询路由

尽管使用单个向量数据库很有用，但所需的数据可能存在于各种数据源中，包括关系数据库或其他向量存储。

例如，您可能有两个向量存储：一个用于 LangChain Python 文档，另一个用于 LangChain JS 文档。给定一个用户的问题，我们希望将查询路由到合理推断的数据源以检索相关文档。查询路由是一种用于将用户的查询转发到相关数据源的策略。

第3章 RAG下: 与自有数据聊天

图 3-8. 将查询路由到相关数据源

为了实现这一点，我们利用像 GPT-4o mini 这样的函数调用模型来帮助将每个查询分类到可用的某个路由。函数调用涉及定义模式，模型可以根据查询使用该模式来生成函数的参数。这样我们能够生成结构化输出，用于运行其他函数。以下 Python 代码根据三种不同语言的文档定义了我们路由器的模式：
Python

from typing import Literal

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# Data model
class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""

    datasource: Literal["python_docs", "js_docs"] = Field(
        ...,
        description="""Given a user question, choose which datasource would be 
            most relevant for answering their question""",
    )

# LLM with function call 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(RouteQuery)

# Prompt 
system = """You are an expert at routing a user question to the appropriate data 
    source.

Based on the programming language the question is referring to, route it to the 
    relevant data source."""

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}"),
    ]
)

# Define router 
router = prompt | structured_llm

from typing import Literal

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import ChatOpenAI

# Data model

class RouteQuery(BaseModel):

"""Route a user query to the most relevant datasource."""

datasource: Literal["python_docs", "js_docs"] = Field(

...,

description="""Given a user question, choose which datasource would be

most relevant for answering their question""",

)

# LLM with function call

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

structured_llm = llm.with_structured_output(RouteQuery)

# Prompt

system = """You are an expert at routing a user question to the appropriate data

source.

Based on the programming language the question is referring to, route it to the

relevant data source."""

prompt = ChatPromptTemplate.from_messages(

[

("system", system),

("human", "{question}"),

]

)

# Define router

router = prompt | structured_llm

JavaScript

import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";

const routeQuery = z.object({
  datasource: z.enum(["python_docs", "js_docs"]).describe(`Given a user 
    question, choose which datasource would be most relevant for answering 
    their question`),
}).describe("Route a user query to the most relevant datasource.")

const llm = new ChatOpenAI({model: "gpt-4o-mini", temperature: 0})
const structuredLlm = llm.withStructuredOutput(routeQuery, {name: "RouteQuery"})

const prompt = ChatPromptTemplate.fromMessages([
  ['system', `You are an expert at routing a user question to the appropriate 
      data source.

Based on the programming language the question is referring to, route it to 
  the relevant data source.`],
  ['human', '{question}']
])

const router = prompt.pipe(structuredLlm)

import { ChatOpenAI } from "@langchain/openai";

import { z } from "zod";

const routeQuery = z.object({

datasource: z.enum(["python_docs", "js_docs"]).describe(`Given a user

question, choose which datasource would be most relevant for answering

their question`),

}).describe("Route a user query to the most relevant datasource.")

const llm = new ChatOpenAI({model: "gpt-4o-mini", temperature: 0})

const structuredLlm = llm.withStructuredOutput(routeQuery, {name: "RouteQuery"})

const prompt = ChatPromptTemplate.fromMessages([

['system', `You are an expert at routing a user question to the appropriate

data source.

Based on the programming language the question is referring to, route it to

the relevant data source.`],

['human', '{question}']

])

const router = prompt.pipe(structuredLlm)

接下来我们调用 LLM，根据预定义的模式提取数据源：

Python

question = """Why doesn't the following code work:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(["human", "speak in {language}"])
prompt.invoke("french")
"""

result = router.invoke({"question": question})

result.datasource
# "python_docs"

question = """Why doesn't the following code work:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(["human", "speak in {language}"])

prompt.invoke("french")

"""

result = router.invoke({"question": question})

result.datasource

# "python_docs"

JavaScript

const question = `Why doesn't the following code work:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(["human", "speak in {language}"])
prompt.invoke("french")
`

await router.invoke({ question })

const question = `Why doesn't the following code work:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(["human", "speak in {language}"])

prompt.invoke("french")

await router.invoke({ question })

输出：

{
    datasource: "python_docs"
}

{

datasource: "python_docs"

}

请注意 LLM 是如何产生符合我们所定义模式的 JSON 输出的。这在许多其他任务中都很有用。

提取到了相关的数据源，我们就可以将该值传递给另一个函数以根据需要执行其它的逻辑：

Python

def choose_route(result):
    if "python_docs" in result.datasource.lower():
        ### Logic here 
        return "chain for python_docs"
    else:
        ### Logic here 
        return "chain for js_docs"

full_chain = router | RunnableLambda(choose_route)

def choose_route(result):

if "python_docs" in result.datasource.lower():

### Logic here

return "chain for python_docs"

else:

### Logic here

return "chain for js_docs"

full_chain = router | RunnableLambda(choose_route)

JavaScript

function chooseRoute(result) {
  if (result.datasource.toLowerCase().includes('python_docs')) {
    return 'chain for python_docs';
  } else {
    return 'chain for js_docs';
  }
} 

const fullChain = router.pipe(chooseRoute)

function chooseRoute(result) {

if (result.datasource.toLowerCase().includes('python_docs')) {

return 'chain for python_docs';

} else {

return 'chain for js_docs';

}

const fullChain = router.pipe(chooseRoute)

注意我们没有进行精确的字符串比较，而是先将生成的输出转换为小写，然后再进行子字符串匹配。这样我们的链更能适应 LLM 的意外行为，即产生不完全符合我们要求的模式的输出。

小贴士：对 LLM 输出的随机性做弹性支持是构建 LLM 应用程序时需要记住的一个重要主题。

逻辑路由最适用于有了一个明确的数据源列表，可以从中检索相关数据并由 LLM 用于生成准确输出的场景。这些数据源可以是向量存储、数据库，甚至是API。

语义路由

与逻辑路由不同，语义路由涉及将代表各种数据源的多个提示与用户的查询一起嵌入，然后执行向量相似性搜索以检索最相似的提示。如图 3-9 所示。

第3章 RAG下: 与自有数据聊天

图 3-9. 通过语义路由提高检索文档的准确性

以下是语义路由的例子：

Python

from langchain.utils.math import cosine_similarity
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import chain
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Two prompts
physics_template = """You are a very smart physics professor. You are great at 
    answering questions about physics in a concise and easy-to-understand manner. 
    When you don't know the answer to a question, you admit that you don't know.

Here is a question:
{query}"""

math_template = """You are a very good mathematician. You are great at answering 
    math questions. You are so good because you are able to break down hard 
    problems into their component parts, answer the component parts, and then 
    put them together to answer the broader question.

Here is a question:
{query}"""

# Embed prompts
embeddings = OpenAIEmbeddings()
prompt_templates = [physics_template, math_template]
prompt_embeddings = embeddings.embed_documents(prompt_templates)

# Route question to prompt
@chain
def prompt_router(query):
    # Embed question
    query_embedding = embeddings.embed_query(query)
    # Compute similarity
    similarity = cosine_similarity([query_embedding], prompt_embeddings)[0]
    # Pick the prompt most similar to the input question
    most_similar = prompt_templates[similarity.argmax()]
    return PromptTemplate.from_template(most_similar)

semantic_router = (
    prompt_router
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

print(semantic_router.invoke("What's a black hole"))

from langchain.utils.math import cosine_similarity

from langchain_core.output_parsers import StrOutputParser

from langchain_core.prompts import PromptTemplate

from langchain_core.runnables import chain

from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Two prompts

physics_template = """You are a very smart physics professor. You are great at

answering questions about physics in a concise and easy-to-understand manner.

When you don't know the answer to a question, you admit that you don't know.

Here is a question:

{query}"""

math_template = """You are a very good mathematician. You are great at answering

math questions. You are so good because you are able to break down hard

problems into their component parts, answer the component parts, and then

put them together to answer the broader question.

Here is a question:

{query}"""

# Embed prompts

embeddings = OpenAIEmbeddings()

prompt_templates = [physics_template, math_template]

prompt_embeddings = embeddings.embed_documents(prompt_templates)

# Route question to prompt

@chain

def prompt_router(query):

# Embed question

query_embedding = embeddings.embed_query(query)

# Compute similarity

similarity = cosine_similarity([query_embedding], prompt_embeddings)[0]

# Pick the prompt most similar to the input question

most_similar = prompt_templates[similarity.argmax()]

return PromptTemplate.from_template(most_similar)

semantic_router = (

prompt_router

| ChatOpenAI(model="gpt-4o-mini")

| StrOutputParser()

)

print(semantic_router.invoke("What's a black hole"))

JavaScript

import {cosineSimilarity} from '@langchain/core/utils/math'
import {ChatOpenAI, OpenAIEmbeddings} from '@langchain/openai'
import {PromptTemplate} from '@langchain/core/prompts'
import {RunnableLambda} from '@langchain/core/runnables';

const physicsTemplate = `You are a very smart physics professor. You are great 
  at answering questions about physics in a concise and easy-to-understand 
  manner. When you don't know the answer to a question, you admit that you 
  don't know.

Here is a question:
{query}`

const mathTemplate = `You are a very good mathematician. You are great at 
  answering math questions. You are so good because you are able to break down 
  hard problems into their component parts, answer the component parts, and 
  then put them together to answer the broader question.

Here is a question:
{query}`

const embeddings = new OpenAIEmbeddings()

const promptTemplates = [physicsTemplate, mathTemplate]
const promptEmbeddings = await embeddings.embedDocuments(promptTemplates)

const promptRouter = RunnableLambda.from(query => {
  // Embed question
  const queryEmbedding = await embeddings.embedQuery(query)
  // Compute similarity
  const similarities = cosineSimilarity([queryEmbedding], promptEmbeddings)[0]
  // Pick the prompt most similar to the input question
  const mostSimilar = similarities[0] > similarities[1] 
    ? promptTemplates[0] 
    : promptTemplates[1]
  return PromptTemplate.fromTemplate(mostSimilar)
})


const semanticRouter = promptRouter.pipe(
  new ChatOpenAI({ modelName: 'gpt-4o-mini', temperature: 0 })
);

await semanticRouter.invoke("What's a black hole")

import {cosineSimilarity} from '@langchain/core/utils/math'

import {ChatOpenAI, OpenAIEmbeddings} from '@langchain/openai'

import {PromptTemplate} from '@langchain/core/prompts'

import {RunnableLambda} from '@langchain/core/runnables';

const physicsTemplate = `You are a very smart physics professor. You are great

at answering questions about physics in a concise and easy-to-understand

manner. When you don't know the answer to a question, you admit that you

don't know.

Here is a question:

{query}`

const mathTemplate = `You are a very good mathematician. You are great at

answering math questions. You are so good because you are able to break down

hard problems into their component parts, answer the component parts, and

then put them together to answer the broader question.

Here is a question:

{query}`

const embeddings = new OpenAIEmbeddings()

const promptTemplates = [physicsTemplate, mathTemplate]

const promptEmbeddings = await embeddings.embedDocuments(promptTemplates)

const promptRouter = RunnableLambda.from(query => {

// Embed question

const queryEmbedding = await embeddings.embedQuery(query)

// Compute similarity

const similarities = cosineSimilarity([queryEmbedding], promptEmbeddings)[0]

// Pick the prompt most similar to the input question

const mostSimilar = similarities[0] > similarities[1]

? promptTemplates[0]

: promptTemplates[1]

return PromptTemplate.fromTemplate(mostSimilar)

})

const semanticRouter = promptRouter.pipe(

new ChatOpenAI({ modelName: 'gpt-4o-mini', temperature: 0 })

);

await semanticRouter.invoke("What's a black hole")

现在你已经了解了如何将用户的查询路由到相关的数据源，下面讨论在构建稳健的 RAG 系统时的第三个主要问题：“我们如何将自然语言转换为目标数据源的查询语言？”

查询构建

如前所述，RAG 是一种基于查询嵌入和检索向量数据库中相关非结构化数据的有效策略。但生产应用中使用的大多数数据是结构化的，通常存储在关系数据库中。此外，向量存储中嵌入的非结构化数据也包含具有重要信息的结构化元数据。

查询构建是将自然语言查询转换为我们正在交互的数据库或数据源的查询语言的过程。参见图 3-10。

第3章 RAG下: 与自有数据聊天

图 3-10. 各种数据源的查询语言示意图

例如，考虑查询：1980年有哪些关于外星人的电影？这个问题包含一个可以通过嵌入检索的非结构化主题（外星人），但它也包含潜在的结构化组件（年份 == 1980）。

以下各节将深入探讨各种形式的查询构建。

文本转元数据过滤器

大多数向量存储提供了基于元数据条件的向量搜索能力。在嵌入过程中，我们可以将元数据键值对附加到索引中的向量上，然后在查询索引时指定过滤表达式。

LangChain 提供了一个 SelfQueryRetriever（自查询检索器），它抽象了这一逻辑，并使得将自然语言查询转换为各种数据源的结构化查询变得更容易。自查询利用 LLM 根据用户的查询和预定义的元数据模式提取并执行相关的元数据过滤器：

Python

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

fields = [
    AttributeInfo(
        name="genre",
        description="The genre of the movie",
        type="string or list[string]",
    ),
    AttributeInfo(
        name="year",
        description="The year the movie was released",
        type="integer",
    ),
    AttributeInfo(
        name="director",
        description="The name of the movie director",
        type="string",
    ),
    AttributeInfo(
        name="rating", description="A 1-10 rating for the movie", type="float"
    ),
]
description = "Brief summary of a movie"

llm = ChatOpenAI(temperature=0)

retriever = SelfQueryRetriever.from_llm(
    llm, db, description, fields,
)

print(retriever.invoke(
    "What's a highly rated (above 8.5) science fiction film?"))

from langchain.chains.query_constructor.base import AttributeInfo

from langchain.retrievers.self_query.base import SelfQueryRetriever

from langchain_openai import ChatOpenAI

fields = [

AttributeInfo(

name="genre",

description="The genre of the movie",

type="string or list[string]",

AttributeInfo(

name="year",

description="The year the movie was released",

type="integer",

AttributeInfo(

name="director",

description="The name of the movie director",

type="string",

AttributeInfo(

name="rating", description="A 1-10 rating for the movie", type="float"

]

description = "Brief summary of a movie"

llm = ChatOpenAI(temperature=0)

retriever = SelfQueryRetriever.from_llm(

llm, db, description, fields,

)

print(retriever.invoke(

"What's a highly rated (above 8.5) science fiction film?"))

JavaScript

import { ChatOpenAI } from "@langchain/openai";
import { SelfQueryRetriever } from "langchain/retrievers/self_query";
import { FunctionalTranslator } from "@langchain/core/structured_query";

/**
 * First, we define the attributes we want to be able to query on.
 * in this case, we want to be able to query on the genre, year, director, 
 * rating, and length of the movie.
 * We also provide a description of each attribute and the type of the attribute.
 * This is used to generate the query prompts.
 */
const fields = [
  {
    name: "genre",
    description: "The genre of the movie",
    type: "string or array of strings",
  },
  {
    name: "year",
    description: "The year the movie was released",
    type: "number",
  },
  {
    name: "director",
    description: "The director of the movie",
    type: "string",
  },
  {
    name: "rating",
    description: "The rating of the movie (1-10)",
    type: "number",
  },
  {
    name: "length",
    description: "The length of the movie in minutes",
    type: "number",
  },
];
const description = "Brief summary of a movie";

const llm = new ChatOpenAI();
const attributeInfos = fields.map((field) => new AttributeInfo(field.name,  
  field.description, field.type));
  
const selfQueryRetriever = SelfQueryRetriever.fromLLM({
  llm,
  db,
  description,
  attributeInfo: attributeInfos,
  /**
   * We need to use a translator that translates the queries into a
   * filter format that the vector store can understand. LangChain provides one 
   * here.
   */
  structuredQueryTranslator: new FunctionalTranslator(),
});

await selfQueryRetriever.invoke(
  "What's a highly rated (above 8.5) science fiction film?"
);

import { ChatOpenAI } from "@langchain/openai";

import { SelfQueryRetriever } from "langchain/retrievers/self_query";

import { FunctionalTranslator } from "@langchain/core/structured_query";

/**

* First, we define the attributes we want to be able to query on.

* in this case, we want to be able to query on the genre, year, director,

* rating, and length of the movie.

* We also provide a description of each attribute and the type of the attribute.

* This is used to generate the query prompts.

const fields = [

{

description: "The genre of the movie",

type: "string or array of strings",

{

description: "The year the movie was released",

type: "number",

{

description: "The director of the movie",

type: "string",

{

description: "The rating of the movie (1-10)",

type: "number",

{

description: "The length of the movie in minutes",

type: "number",

];

const description = "Brief summary of a movie";

const llm = new ChatOpenAI();

const attributeInfos = fields.map((field) => new AttributeInfo(field.name,

field.description, field.type));

const selfQueryRetriever = SelfQueryRetriever.fromLLM({

llm,

db,

description,

attributeInfo: attributeInfos,

/**

* We need to use a translator that translates the queries into a

* filter format that the vector store can understand. LangChain provides one

* here.

structuredQueryTranslator: new FunctionalTranslator(),

});

await selfQueryRetriever.invoke(

"What's a highly rated (above 8.5) science fiction film?"

);

这将产生一个检索器，它会接收用户查询，并将其拆分为：

用于每个文档元数据的过滤器
用于对文档进行语义搜索的查询

为此，我们必须描述文档元数据包含哪些字段；该描述将包含在提示中。然后，检索器将执行以下操作：

将查询生成提示发送给 LLM。
从 LLM 输出中解析元数据过滤器和重写的搜索查询。
将 LLM 生成的元数据过滤器转换为适合我们向量存储的格式。
对向量存储进行相似性搜索，过滤仅匹配其元数据通过所生成过滤器的文档。

文本转SQL

SQL 和关系数据库是结构化数据的重要来源，但它们不直接与自然语言交互。尽管我们可以简单地使用 LLM 将用户的查询转换为 SQL 查询，但容错余地很小。

以下是一些有效的文本到 SQL 转换的有用策略：

数据库描述: 为了使 SQL 查询有据可依，必须为 LLM 提供数据库的准确描述。一种常见的文本到 SQL 提示采用了这篇论文及其他论文中报告的一个思想：向 LLM 提供每个表的 CREATE TABLE 描述，包括列名和类型。我们还可以提供表中几行（例如，三行）示例数据。
小样本示例: 通过在提示中提供一些问题-查询匹配的小样本示例，可以提高查询生成的准确性。这可以通过在提示中简单地附加标准的静态示例来实现，以指导agent如何根据问题构建查询。

参见图 3-11，了解该过程的图示。

第3章 RAG下: 与自有数据聊天

图 3-11. 用户查询被转换为 SQL 查询

以下是一个完整的代码示例：

Python

from langchain_community.tools import QuerySQLDatabaseTool
from langchain_community.utilities import SQLDatabase
from langchain.chains import create_sql_query_chain
from langchain_openai import ChatOpenAI

# replace this with the connection details of your db
db = SQLDatabase.from_uri("sqlite:///Chinook.db")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# convert question to sql query
write_query = create_sql_query_chain(llm, db)

# Execute SQL query
execute_query = QuerySQLDatabaseTool(db=db)

# combined
chain = write_query | execute_query

# invoke the chain
chain.invoke('How many employees are there?');

from langchain_community.tools import QuerySQLDatabaseTool

from langchain_community.utilities import SQLDatabase

from langchain.chains import create_sql_query_chain

from langchain_openai import ChatOpenAI

# replace this with the connection details of your db

db = SQLDatabase.from_uri("sqlite:///Chinook.db")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# convert question to sql query

write_query = create_sql_query_chain(llm, db)

# Execute SQL query

execute_query = QuerySQLDatabaseTool(db=db)

# combined

chain = write_query | execute_query

# invoke the chain

chain.invoke('How many employees are there?');

JavaScript

import { ChatOpenAI } from "@langchain/openai";
import { createSqlQueryChain } from "langchain/chains/sql_db";
import { SqlDatabase } from "langchain/sql_db";
import { DataSource } from "typeorm";
import { QuerySqlTool } from "langchain/tools/sql";

const datasource = new DataSource({
  type: "sqlite",
  database: "./Chinook.db", // here should be the details of your database
});
const db = await SqlDatabase.fromDataSourceParams({
  appDataSource: datasource,
});
const llm = new ChatOpenAI({ model: "gpt-4o-mini", temperature: 0 });

// convert question to sql query
const writeQuery = await createSqlQueryChain({ llm, db, dialect: "sqlite" });

// execute query
const executeQuery = new QuerySqlTool(db);

// combined
const chain = writeQuery.pipe(executeQuery);

// invoke the chain
await chain.invoke('How many employees are there?');

import { ChatOpenAI } from "@langchain/openai";

import { createSqlQueryChain } from "langchain/chains/sql_db";

import { SqlDatabase } from "langchain/sql_db";

import { DataSource } from "typeorm";

import { QuerySqlTool } from "langchain/tools/sql";

const datasource = new DataSource({

type: "sqlite",

database: "./Chinook.db", // here should be the details of your database

});

const db = await SqlDatabase.fromDataSourceParams({

appDataSource: datasource,

});

const llm = new ChatOpenAI({ model: "gpt-4o-mini", temperature: 0 });

// convert question to sql query

const writeQuery = await createSqlQueryChain({ llm, db, dialect: "sqlite" });

// execute query

const executeQuery = new QuerySqlTool(db);

// combined

const chain = writeQuery.pipe(executeQuery);

// invoke the chain

await chain.invoke('How many employees are there?');

我们首先将用户的查询转换为适合我们数据库方言的 SQL 查询。然后在数据库上执行该查询。请注意，在生产应用程序中，由用户输入给 LLM 生成的 SQL 在数据库上执行查询是危险的。要在生产中使用这些想法，需要考虑许多安全措施，以减少在数据库中运行意外查询的风险。以下是一些示例：

使用只读权限的用户在数据库上运行查询。
运行查询的数据库用户应仅有权访问所希望提供查询的表。
为此应用程序运行的查询添加一个超时；这将确保即使生成了一个重度资源消耗的查询，它也会在占用过多数据库资源之前被取消。

这个安全清单并不全面。LLM 应用程序的安全性是一个目前正在发展的领域，随着新漏洞的发现，我们需要增加更多的安全措施。

总结

本章讨论了各种流行的策略，以根据用户的查询高效地检索最相关的文档，并将其与我们的提示合成，帮助 LLM 生成准确、最新的输出。

如我们所讨论的，一个稳健、生产就绪的 RAG 系统需要广泛的有效策略，这些策略可以执行查询转换、查询构建、路由和索引优化。

查询转换使得 AI 应用能够将一个模棱两可或格式不正确的用户查询转换为一个具有代表性的查询，该查询最适合检索。查询构建使得 AI 应用能够将用户的查询转换为结构化数据所在的数据库或数据源的查询语言的语法。路由使得 AI 应用能够动态地将用户的查询路由到相关的数据源检索相关信息。

在第 4 章中，我们将在此知识的基础上为我们的 AI 聊天机器人添加记忆，使其能够记住并从每次交互中学习。这将使用户能够像与 ChatGPT 一样，在多轮对话中与应用程序“聊天”。

检索增强生成入门

检索相关文档

使用相关文档生成 LLM 预测

查询转换

重写-检索-读取

多查询检索

RAG-Fusion

假设性文档嵌入

查询路由

语义路由

查询构建

文本转元数据过滤器

文本转SQL

总结

Hi，您需要填写昵称和邮箱！