【AI大模型应用开发】【LangChain系列】2. 一文全览LangChain数据连接模块：从文档加载到向量检索RAG，理论+实战+细节

04-02 1868阅读 0评论

大家好，我是【同学小张】。持续学习，持续干货输出，关注我，跟我一起学AI大模型技能。

本文学习 LangChain 中的数据连接（Retrieval）模块。该模块提供文档加载、切分，向量存储、检索等操作的封装。最后，结合RAG基本流程、LangChain Prompt模板和输入输出模块，我们将利用LangChain实现RAG的基本流程。

文章目录

0. 模块介绍
1. Document loaders 文档加载模块

1.1 加载本地文件
1.2 加载在线PDF文件

1.2.1 可能需要的环境配置
1.2.2 示例代码
2. Text Splitting 文档切分模块
3. Text embedding models 文本向量化模型封装
4. Vector stores 向量存储（数据库）
5. Retrievers 检索器
6. Indexing

6.1 概念和用途
6.2 工作原理
6.3 Deletion modes

7. 总结，用LangChain实现RAG流程

0. 模块介绍

在前面文章中我们已经讲了大模型存在的缺陷：数据不实时，缺少垂直领域数据和私域数据等。解决这些缺陷的主要方法是通过检索增强生成（RAG）。首先检索外部数据，然后在执行生成步骤时将其传递给LLM。

LangChain为RAG应用程序提供了从简单到复杂的所有构建块，本文要学习的数据连接（Retrieval）模块包括与检索步骤相关的所有内容，例如数据的获取、切分、向量化、向量存储、向量检索等模块（见下图）。

1. Document loaders 文档加载模块

LangChain封装了一系列类型的文档加载模块，例如PDF、CSV、HTML、JSON、Markdown、File Directory等。下面以PDF文件夹在为例看一下用法，其它类型的文档加载的用法都类似。

1.1 加载本地文件

LangChain加载PDF文件使用的是pypdf，先安装：

pip install pypdf

加载代码示例：

from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("D:\GitHub\LEARN_LLM\RAG\如何向 ChatGPT 提问以获得高质量答案：提示技巧工程完全指南.pdf")
pages = loader.load_and_split()
print(f"第0页：\n{pages[0]}") ## 也可通过 pages[0].page_content只获取本页内容

看下运行结果：pypdf将PDF分成了一个数组，数组中的每个元素包含本页内容、文件路径和名称以及所在页码。

1.2 加载在线PDF文件

LangChain竟然也能加载在线的PDF文件。

1.2.1 可能需要的环境配置

在开始之前，你可能需要安装以下的Python包：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple unstructured
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pdf2image
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple opencv-python
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple unstructured-inference
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pikepdf

除了Python包，还需要下载 nltk_data，这东西非常大，下载起来非常慢。所以们可以事先下好，放到固定的位置。

下载地址：https://github.com/nltk/nltk_data/tree/gh-pages

下载完后，将其中的packages文件夹内的全部内容拷贝到固定位置，例如上面的 C:\Users\xxx\AppData\Roaming\nltk_data

1.2.2 示例代码

from langchain_community.document_loaders import OnlinePDFLoader
loader = OnlinePDFLoader("https://arxiv.org/pdf/2302.03803.pdf")
data = loader.load()
print(data)

运行结果：

2. Text Splitting 文档切分模块

LangChain提供了许多不同类型的文本切分器，具体见下表：

名称	分割依据	添加元数据	描述
Recursive	用户定义字符列表		递归地分割文本。递归地分割文本的目的是试图保持相关的文本片段相邻。这是开始分割文本的推荐方法。
HTML	HTML特定字符	✅	基于HTML特定字符分割文本。特别地，这会添加关于该片段来自何处的相关信息（基于HTML）。
Markdown	Markdown特定字符	✅	基于Markdown特定字符分割文本。特别地，这会添加关于该片段来自何处的相关信息（基于Markdown）。
Code	代码（Python、JS）特定字符		基于编程语言特定字符分割文本。可选择15种不同的语言。
Token	标记		基于标记分割文本。有几种不同的标记计量方法。
Character	用户定义字符		基于用户定义字符分割文本。这是比较简单的方法之一。
[Experimental] Semantic Chunker	句子		首先基于句子分割文本。然后，如果相邻的句子在语义上足够相似，则合并它们。

这里以Recursive为例展示用法。RecursiveCharacterTextSplitter是LangChain对这种文档切分方式的封装，里面的几个重点参数：

chunk_size：每个切块的token数量

chunk_overlap：相邻两个切块之间重复的token数量

from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("D:\GitHub\LEARN_LLM\RAG\如何向 ChatGPT 提问以获得高质量答案：提示技巧工程完全指南.pdf")
pages = loader.load_and_split()
print(f"第0页：\n{pages[0].page_content}")
from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=100,
    length_function=len,
    add_start_index=True,
)
paragraphs = text_splitter.create_documents([pages[0].page_content])
for para in paragraphs:
    print(para.page_content)
    print('-------')

以上示例程序将chunk_overlap设置为100，看下运行效果，可以看到上一个chunk和下一个chunk会有一部分的信息重合，这样做的原因是尽可能地保证两个chunk之间的上下文关系：

LangChain虽然提供了文本加载和分割模块的封装，但很明显，想要用在实际的项目中，还远远不够，需要在此基础上做非常细致的策略兜底或干脆自己实现一套符合自己实际需求的文档加载和分割器。

这里提供了一个可视化展示文本如何分割的工具，感兴趣的可以看下。

工具网址：https://chunkviz.up.railway.app/

3. Text embedding models 文本向量化模型封装

LangChain对一些文本向量化模型的接口做了封装，例如OpenAI, Cohere, Hugging Face等。

向量化模型的封装提供了两种接口，一种针对文档的向量化embed_documents，一种针对句子的向量化embed_query。

示例代码：

文档的向量化embed_documents，接收的参数是字符串数组

from langchain_openai import OpenAIEmbeddings
embeddings_model = OpenAIEmbeddings()  ## OpenAI文本向量化模型接口的封装
embeddings = embeddings_model.embed_documents(
    [
        "Hi there!",
        "Oh, hello!",
        "What's your name?",
        "My friends call me World",
        "Hello World!"
    ]
)
len(embeddings), len(embeddings[0])
##运行结果 (5, 1536)

句子的向量化embed_query，接收的参数是字符串

embedded_query = embeddings_model.embed_query("What was the name mentioned in the conversation?")
embedded_query[:5]
## 运行结果：
## [0.0053587136790156364,
##  -0.0004999046213924885,
##  0.038883671164512634,
##  -0.003001077566295862,
##  -0.00900818221271038]

4. Vector stores 向量存储（数据库）

将文本向量化之后，下一步就是进行向量的存储。

这部分包含两块：一是向量的存储。二是向量的查询。

官方提供了三种开源、免费的可用于本地机器的向量数据库示例（chroma、FAISS、 Lance）。因为我在之前RAG的文章中用的chroma数据库，所以这里还是以这个数据库为例。

安装：

pip install chromadb

创建向量数据库，灌入数据from_documents：

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
db = Chroma.from_documents(paragraphs, OpenAIEmbeddings()) ## 一行代码搞定

查询similarity_search：

query = "什么是角色提示？"
docs = db.similarity_search(query) ## 一行代码搞定
for doc in docs:
    print(f"{doc.page_content}\n-------\n")

运行结果：

它也接受传入一个向量来进行向量检索similarity_search_by_vector。以下是代码示例。在以下代码中可能体现的价值不是很大，但是在实际项目中，如果出现只知道向量值，不知道具体文字的情况，这个接口就有用了。

embedding_vector = OpenAIEmbeddings().embed_query(query)
docs = db.similarity_search_by_vector(embedding_vector)
print(docs[0].page_content)

5. Retrievers 检索器

检索器是在给定非结构化查询的情况下返回相关文本的接口。它比Vector stores更通用。检索器不需要能够存储文档，只需要返回（或检索）文档即可。Vector stores可以用作检索器的主干，但也有其他类型的检索器。检索器接受字符串查询作为输入，并返回文档列表作为输出。

LangChain检索器提供的检索类型如下：

名称	索引类型	使用LLM	何时使用	描述
Vectorstore	Vectorstore	否	刚开始接触，想快速简单的入门	这是最简单的方法，也是最容易入门的方法。它涉及为每个文本片段创建向量。
ParentDocument	Vectorstore + 文档存储	否	如果您的页面有许多较小的独立信息片段，最好单独索引，但最好一起检索。	这涉及为每个文档索引多个片段。然后找到在嵌入空间中最相似的片段，但检索整个父文档并返回（而不是单个片段）。
Multi Vector	Vectorstore + 文档存储	有时在索引过程中	如果您能够从文档中提取比文本本身更相关的信息。	这涉及为每个文档创建多个向量。每个向量可以以多种方式创建 - 例如，文本摘要和假设性问题。
Self Query	Vectorstore	是	如果用户提出的问题更适合根据元数据而不是文本相似性来获取文档回答。	这使用LLM将用户输入转换为两件事：（1）语义查找的字符串，（2）与之配套的元数据过滤器。这很有用，因为通常问题是关于文档的元数据（而不是内容本身）。
Contextual Compression	任意	有时	如果您发现检索到的文档包含太多不相关信息，并且干扰了LLM。	这在另一个检索器之上放置了后处理步骤，并仅从检索到的文档中提取最相关的信息。这可以使用嵌入或LLM完成。
Time-Weighted Vectorstore	Vectorstore	否	如果您的文档有时间戳，并且您想检索最近的文档。	这根据语义相似性（与常规向量检索相同）和时间权重（查看索引文档的时间戳）检索文档。
Multi-Query Retriever	任意	是	如果用户提出的问题复杂，并且需要多个独立信息片段来回应。	这使用LLM从原始查询生成多个查询。当原始查询需要有关多个主题的信息片段才能正确回答时，这很有用。通过生成多个查询，我们可以为每个查询获取文档。
Ensemble	任意	否	如果您有多个检索方法并希望尝试将它们组合起来。	这从多个检索器中获取文档，然后将它们组合在一起。
Long-Context Reorder	任意	否	如果您使用长上下文模型，并且注意到它没有关注检索到的文档中的中间信息。	这从基础检索器中获取文档，然后重新排序文档，使最相似的文档靠近开头和结尾。这很有用，因为长上下文模型有时不关注上下文窗口中间的信息。

表中的每个检索类型都可以单独拿出来学习和研究。本文先大体了解都有哪些类型，以Vectorstore类型的检索为例简单使用（功能与Vectorstore中的查询一样）：

retriever = db.as_retriever()
docs = retriever.get_relevant_documents("什么是角色提示？")
for doc in docs:
    print(f"{doc.page_content}\n-------\n")

as_retriever：生成检索器实例

get_relevant_documents获取相关文本

as_retriever可指定使用的检索类型，同时也可指定一些其它参数，例如：

指定一个相似度阈值为0.5，只有相似度超过这个值才会召回

retriever = db.as_retriever(
    search_type="similarity_score_threshold", search_kwargs={"score_threshold": 0.5}
)

指定检索几个文本片段：topK

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

6. Indexing

这块还没用到，先看下概念和作用。总的来说，这块是帮助用户省钱的，帮助用户减少重复的文档向量化和灌入，帮助用户定制化的删除一些冲突内容或无用内容。

6.1 概念和用途

索引API允许您将任何来源的文档加载到向量存储中并保持同步。具体来说，它有助于：

避免将重复的内容写入向量数据库
避免重写未更改的内容

避免在未更改的内容上重新计算向量

6.2 工作原理

LangChain索引使用记录管理器（RecordManager）来跟踪文档写入向量数据库的情况。为内容编制索引时，会为每个文档计算哈希值，并将以下信息存储在记录管理器中：

文档哈希（页面内容和元数据的哈希）
写入时间

源id – 每个文档都应该在其元数据中包含信息，以便我们确定该文档的最终来源

6.3 Deletion modes

该模块还提供了 Deletion modes。它的应用场景是：将文档索引到向量数据库时，可能会删除数据库中的一些现有文档。在某些情况下，您可能希望删除与正在编制索引的新文档来源相同的现有文档。在其他情况下，您可能希望批量删除所有现有文档。索引API删除模式帮助你进行自定义删除。

7. 总结，用LangChain实现RAG流程

请先阅读前置文章，了解RAG基本流程和LangChain的Prompt模板：

【AI大模型应用开发】3. RAG初探 - 动手实现一个最简单的RAG应用

【AI大模型应用开发】【LangChain系列】2. 一文全览LangChain数据连接模块：从文档加载到向量检索，理论+实战+细节

以上，我们学习了LangChain数据连接模块的所有内容。下面，我们将所有流程串起来，用LangChain实现一下RAG流程。

    ## 1. 文档加载
    from langchain.document_loaders import PyPDFLoader
    loader = PyPDFLoader("D:\GitHub\LEARN_LLM\RAG\如何向 ChatGPT 提问以获得高质量答案：提示技巧工程完全指南.pdf")
    pages = loader.load_and_split()
    
    ## 2. 文档切分
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=200,
        chunk_overlap=100,
        length_function=len,
        add_start_index=True,
    )
    paragraphs = []
    for page in pages:
        paragraphs.extend(text_splitter.create_documents([page.page_content]))
    
    ## 3. 文档向量化，向量数据库存储
    from langchain_openai import OpenAIEmbeddings
    from langchain_community.vectorstores import Chroma
    db = Chroma.from_documents(paragraphs, OpenAIEmbeddings())
    
    ## 4. 向量检索
    # query = "什么是角色提示？"
    # docs = db.similarity_search(query)
    # for doc in docs:
    #     print(f"{doc.page_content}\n-------\n")
    
    retriever = db.as_retriever()
    docs = retriever.get_relevant_documents("什么是角色提示？")
    for doc in docs:
        print(f"{doc.page_content}\n-------\n")
    
    ## 5. 组装Prompt模板
    import os
    # 加载 .env 到环境变量
    from dotenv import load_dotenv, find_dotenv
    _ = load_dotenv(find_dotenv())
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI() # 默认是gpt-3.5-turbo
    prompt_template = """
    你是一个问答机器人。
    你的任务是根据下述给定的已知信息回答用户问题。
    确保你的回复完全依据下述已知信息。不要编造答案。
    如果下述已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。
    已知信息:
    {info}
    用户问：
    {question}
    请用中文回答用户问题。
    """
    from langchain.prompts import PromptTemplate
    template = PromptTemplate.from_template(prompt_template)
    prompt = template.format(info=docs[0].page_content, question='什么是角色提示？')
    ## 6. 调用LLM
    response = llm.invoke(prompt)
    print(response.content)

运行结果：