LangChain RAG 应用开发组件深入学习

章节介绍

1. LangChain文档组件与文档加载器：学习Document组件作用，各类文档加载器的配置、使用及自定义封装。
2. LangChain 文档转换器与分割器：掌握文档转换器功能与场景，各类文本分割器、语义分割器的使用及自定义技巧，不同模式下分割器选择思路。
3. VectorStore组件与检索器的使用：深入理解VectorStore组件，相似性搜索的几何意义与选择策略，掌握检索器的配置、使用及自定义方法。

Document组件与文档加载器的使用

01. Document与文档加载器

Document类核心作用

LangChain的核心组件，定义文档对象结构，由page_content（页面内容）+ metadata（元数据） 组成，是文档加载器、分割器、向量数据库、检索器之间交互的核心数据载体；新版本仅保留基础记录功能，移除了旧版本的lookup检索功能。

文档加载器的价值

RAG开发中需从本地文件、网页、PDF等多来源读取数据，而非手动录入。文档加载器的核心作用是从各类数据源提取信息并转换为标准Document组件，屏蔽不同文件类型的读取差异。

数据处理流程扩展

RAG应用外部需单独处理读取数据-切割数据-存储数据流程，该流程耗时，通常通过队列/异步实现，整体架构中加载器是外部数据进入RAG流程的入口。

文档加载器基类BaseLoader

所有LangChain文档加载器的统一基类，封装5个核心方法：

1. load()/aload()：同步/异步加载文档，返回文档列表；
2. load_and_split()：传入分割器，加载并切割大文档，返回分割后的文档列表；
3. lazy_load()/alazy_load()：同步/异步懒加载文档，返回迭代器，适用于多文档数据源，无需等待所有文档加载完成，可逐次获取最新文档。

LangChain内置加载器

封装了数十种加载器，支持绝大多数文件类型，核心包括：

• 文本类：TextLoader（纯文本）、UnstructuredMarkdownLoader（Markdown）
• 表格类：CSVLoader、UnstructuredCSVLoader、UnstructuredExcelLoader
• 网页类：BSHTMLLoader（基于BeautifulSoup的HTML加载）
• 通用类：UnstructuredFileLoader（非结构化文件）

02. TextLoader使用技巧与源码解析

核心功能

最简单的加载器，支持加载纯文本结构文件（.txt、.md等，不支持.doc），将文件内容整体读入一个Document对象，并为metadata添加source字段记录文件路径。

使用示例

from langchain_community.document_loaders import TextLoader
# 实例化加载器，指定文件路径和编码
loader = TextLoader("./电商产品数据.txt", encoding="utf-8")
documents = loader.load()
# 输出结果
print(documents)
print(len(documents))
print(documents[0].metadata)

输出：

[Document(page_content='xxx',metadata={'source':'./电商产品数据.txt'})]
1
{'source':'./电商产品数据.txt'}

核心源码逻辑（lazy_load方法）

1. 通过open()函数按指定编码读取文件内容，若遇编码错误，开启自动检测编码并重试；
2. 构造metadata，核心包含source（文件路径）；
3. 通过yield返回Document对象，实现懒加载。

通用使用技巧

所有内置加载器使用逻辑一致：实例化时传入数据源信息（路径/网址/目录）→ 调用load()方法加载文档。

自定义LangChain文档加载器使用技巧

01. 自定义加载器的适用场景

企业内部定制化数据源（数据库、API接口），或通用加载器提取的数据格式/样式不符合需求（如加载网页出现大量空白数据），会降低RAG检索和生成效率，需自定义加载器。

实现要求

继承LangChain核心基类BaseLoader，必须实现lazy_load()方法（同步懒加载，返回迭代器）；若需异步使用，需额外实现alazy_load()方法。

实现示例：逐行读取文件为独立Document

from typing import Iterator, AsyncIterator
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document

class CustomDocumentLoader(BaseLoader):
    """自定义文档加载器：逐行读取文件，每行生成一个Document"""
    def __init__(self, file_path: str) -> None:
        self.file_path = file_path

    def lazy_load(self) -> Iterator[Document]:
        """同步懒加载：逐行读取并yield Document"""
        with open(self.file_path, encoding="utf-8") as f:
            line_number = 0
            for line in f:
                yield Document(
                    page_content=line,
                    metadata={"line_number": line_number, "source": self.file_path}
                )
                line_number += 1

    async def alazy_load(self) -> AsyncIterator[Document]:
        """异步懒加载：基于aiofiles实现"""
        import aiofiles
        async with aiofiles.open(self.file_path, encoding="utf-8") as f:
            line_number = 0
            async for line in f:
                yield Document(
                    page_content=line,
                    metadata={"line_number": line_number, "source": self.file_path}
                )
                line_number += 1

# 使用示例
loader = CustomDocumentLoader("./喵喵.txt")
documents = loader.load()
print(documents)
print(len(documents))
print(documents[0].metadata)

输出：

[Document(page_content='喵\n',metadata={'line_number':0,'source':'./喵喵.txt'}), Document(page_content='喵喵 \n',metadata={'line_number':1,'source':'./喵喵.txt'}), Document(page_content='',metadata={'line_number':2,'source':'./喵喵.txt'})]
3
{'line_number':0,'source':'./喵喵.txt'}

02. 文档加载器扩展思考

核心拆解逻辑

所有文档加载器的核心均包含两个步骤：二进制数据读取 + 解析逻辑，其中数据读取步骤高度通用（如.md/.txt/.pdf均可读取为二进制内容），差异仅在于解析逻辑。

优化思路：加载与解析分离

将二进制数据读取和解析逻辑解耦，让所有加载器共用数据读取逻辑，仅实现各自的解析逻辑，便于维护和逻辑复用。

• 这是LangChain正在设计的新方案，核心基于BaseBlobParser实现；
• 流程：存储的原始数据 → Blob（二进制数据对象） → BaseBlobParser（解析） → Document列表。

Blob与BlobParser代替文档加载器

LangChain推出的数据加载+解析解耦方案，灵感源于前端Web API的Blob规范，解决传统文档加载器加载与解析耦合的问题。

01. LangChain中的Blob核心类

包含Blob、BlobLoader、BaseBlobParser三个核心类，分工明确：

1. Blob：封装的二进制数据对象，表示原始数据，提供统一的二进制数据操作接口，实现加载与解析解耦；
2. BlobLoader：Blob数据加载器，类似DocumentLoader，设计目标为支持加载任意类型数据；
3. BaseBlobParser：Blob数据解析器，核心负责将Blob数据转换为Document列表，需自定义实现解析逻辑。

实现示例：自定义BlobParser逐行解析文本

from typing import Iterator
from langchain_core.document_loaders import Blob
from langchain_core.document_loaders.base import BaseBlobParser
from langchain_core.documents import Document

class CustomParser(BaseBlobParser):
    """自定义解析器：将文本文件的每一行解析为一个Document"""
    def lazy_parse(self, blob: Blob) -> Iterator[Document]:
        line_number = 0
        with blob.as_bytes_io() as f:
            for line in f:
                yield Document(
                    page_content=line,
                    metadata={"source": blob.source, "line_number": line_number}
                )
                line_number += 1

# 使用示例
blob = Blob.from_path("./喵喵.txt")
parser = CustomParser()
documents = list(parser.lazy_parse(blob))
print(documents[0].metadata)
print(documents)
print(len(documents))

输出：

{'source':'./喵喵.txt','line_number':0}
[Document(page_content='喵喵\r\n',metadata={'source':'./喵喵.txt','line_number':0}), Document(page_content='\r\n',metadata={'source':'./喵喵.txt','line_number':1}), Document(page_content='',metadata={'source':'./喵喵.txt','line_number':2})]
3

Blob的两种加载方式

1. 从文件加载：Blob.from_path(文件路径)；
2. 从内存加载：直接传入data参数，Blob(data="喵\r\n喵喵")。

02. Blob数据对象的方法与属性

与Web API的Blob规范高度接近，核心属性/方法如下：

类型	名称	功能
属性	data	存储原始数据，支持字节/字符串类型
属性	mimetype	文件的MIME类型
属性	encoding	文件编码，默认utf-8
属性	path	文件原始路径，支持字符串/Path类
属性	metadata	元数据，默认包含source字段
方法	source()	只读，返回数据来源
方法	as_string()	将Blob数据转换为字符串
方法	as_bytes()	将Blob数据转换为字节数据
方法	as_bytes_io()	将Blob数据转换为缓冲流字节数据
类方法	from_path()	从文件路径加载Blob数据
类方法	from_data()	从内存原始数据加载Blob数据

03. Blob加载器

封装从指定存储位置加载Blob数据的逻辑，目前LangChain仅集成FilesystemBlobLoader（文件系统Blob加载器），支持加载指定文件夹下的文件。

FilesystemBlobLoader使用示例

from langchain_community.document_loaders.blob_loaders import FilesystemBlobLoader
# 加载当前目录下所有文件，显示进度
loader = FilesystemBlobLoader(".", show_progress=True)
for blob in loader.yield_blobs():
    print(blob.source)

输出：

100%|3/3[00:00<00:00,58.15it/s]
Blob解析器示例.py
FilesystemBlobLoader示例.py
喵喵.txt

自定义BlobLoader要求

继承BlobLoader抽象基类，必须实现yield_blobs()方法，返回Blob对象的迭代器。

from langchain_core.document_loaders.blob_loaders import BlobLoader
from abc import ABC, abstractmethod

class BlobLoader(ABC):
    """Blob加载器抽象基类"""
    @abstractmethod
    def yield_blobs(self) -> Iterable[Blob]:
        """懒加载Blob数据，返回Blob迭代器"""
        pass

04. Blob通用加载器GenericLoader

LangChain封装的BlobLoader + BaseBlobParser组合类，提供标准化的Blob加载+解析流程，目前仅支持FilesystemBlobLoader，简化Blob的使用流程。

使用示例

from langchain_community.document_loaders.generic import GenericLoader
# 加载当前目录下所有.txt文件，显示进度
loader = GenericLoader.from_filesystem(".", glob="*.txt", show_progress=True)
for idx, doc in enumerate(loader.lazy_load()):
    print(f"当前加载第{idx+1}个文件，文件信息:{doc.metadata}")

输出：

100%|1/1[00:00<00:00,19.76it/s]
当前加载第1个文件，文件信息:{'source':'喵喵.txt'}

Blob方案现状

目前LangChain对Blob的封装和集成较少，使用时需大量自定义加载/解析逻辑，效率较低；未来会逐步完善，最终替代传统的DocumentLoader方案，现阶段仅需了解基础概念和使用方式。

文档转换器与字符分割器组件的使用

01. DocumentTransformer 组件

核心作用

解决文档加载后的数据问题：原始文档过大、格式不符、信息未提炼等，若直接将此类数据存入向量数据库，会大幅提升RAG的错误率。对加载后的文档列表进行转换，得到符合RAG需求的文档数据。

支持的转换操作

文档切割、属性提取、翻译、HTML转文本、重排、元数据标记等。

新增转换步骤的RAG架构

外部数据 → 文档加载器 → 文档转换器 → 分割器 → 向量数据库 → 检索器 → LLM生成 → 格式化输出

核心基类BaseDocumentTransformer

所有文档转换组件的统一基类，封装两个核心方法：

1. transform_documents()：抽象方法，传入文档列表，返回转换后的文档列表；
2. atransform_documents()：异步转换方法，若未实现，将委托同步方法执行。

文档转换组件分类

1. 文档分割器：使用频率最高，已单独拆分为独立Python包langchain-text-splitters；
2. 文档处理转换器：使用率低，为旧版本写法。

文本分割器独立包安装

pip install -q langchain-text-splitters

文本分割器的基类体系

• 基础基类：BaseDocumentTransformer（所有转换组件的父类）；
• 专属基类：TextSplitter（为文本分割器设计，实现更丰富的分割功能）；
• 所有文本分割器均继承TextSplitter，同时实现BaseDocumentTransformer的接口。

02. 字符分割器基础使用技巧

核心组件：CharacterTextSplitter

最简单的文本分割器，基于指定字符串分割文档（默认\n\n），分割时尽可能保证数据连续性，通过字符数衡量分割块大小，是最常用的分割器之一。

核心实例化参数

参数	类型	默认值	功能
separator	str	\n\n	分割符
is_separator_regex	bool	False	分割符是否为正则表达式
chunk_size	int	4000	每个分割块的最大字符数
chunk_overlap	int	200	块与块之间的重叠字符数（保证上下文连续性）
length_function	function	len	计算文本长度的函数
keep_separator	bool	False	是否将分割符保留在分割块中
add_start_index	bool	False	是否在metadata中添加分割块的起始索引
strip_whitespace	bool	True	是否删除文档头尾的空白字符

使用示例：分割Markdown文档

from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter

# 1. 加载Markdown文档
loader = UnstructuredMarkdownLoader("./项目API文档.md")
documents = loader.load()

# 2. 实例化分割器：最大500字符/块，重叠50字符，添加起始索引
text_splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=500,
    chunk_overlap=50,
    add_start_index=True,
)

# 3. 分割文档
chunks = text_splitter.split_documents(documents)

# 4. 输出分割结果
for chunk in chunks:
    print(f"块内容大小:{len(chunk.page_content)}, 元数据:{chunk.metadata}")

典型输出

Created a chunk of size 771,which is longer than the specified 500
Created a chunk of size 980,which is longer than the specified 500
块内容大小:251,元数据:{'source':'./项目API文档.md','start_index':0}
块内容大小:451,元数据:{'source':'./项目API文档.md','start_index':246}
块内容大小:435,元数据:{'source':'./项目API文档.md','start_index':1472}

核心分割逻辑与注意事项

1. 分割流程：先按指定分割符拆分文档为小片段 → 遍历片段并逐次合并，直到接近chunk_size → 生成一个Document块；
2. 关键注意点：若单个分割符拆分的片段本身超过chunk_size，分割器会抛出警告，并将该片段直接作为一个独立块，无法保证块大小严格小于chunk_size；
3. 核心源码方法：split_text()（拆分文本）、_split_text_with_regex()（基于正则分割）、_merge_splits()（合并片段为指定大小）。

核心分割源码片段

def split_text(self, text: str) -> List[str]:
    """拆分文本为指定大小的块"""
    separator = self._separator if self._is_separator_regex else re.escape(self._separator)
    # 基于正则拆分文本
    splits = _split_text_with_regex(text, separator, self._keep_separator)
    _separator = "" if self._keep_separator else self._separator
    # 合并拆分后的片段为指定chunk_size
    return self._merge_splits(splits, _separator)