网页加载：WebBaseLoader#

WebBaseLoader 是课程列出的常用加载器之一，专门吃 HTTP(S) 链接：

1
from langchain_community.document_loaders import WebBaseLoader
2

3
loader = WebBaseLoader("https://example.com")
4
docs = loader.load()
5
print(docs[0].metadata)      # {'source': 'https://example.com', 'title': 'Example Domain', 'language': 'en'}
6
print(docs[0].page_content)  # 已经去掉 HTML 标签的正文

PDF 加载（方式 1）：PyPDFLoader#

PDF 的来源格式有扫描版（图片 PDF）、电子文本版、混合版，布局又有单栏、双栏甚至竖排，还可能包含段落、标题、页眉页脚、表格、数学公式、化学式、图片——所以 PDF 解析本身挑战很大，复杂 PDF 需要文本提取、布局检测、表格解析、公式识别等一整套处理。课程给了两种方式。

PyPDFLoader 是轻量方案：每一页变成一个 Document，还支持直接传在线链接。

1
from langchain_community.document_loaders import PyPDFLoader
2

3
loader = PyPDFLoader(
4
    # 文件路径，支持本地文件和在线文件链接
5
    # file_path="../asset/load/04-sample.pdf",
6
    file_path="https://arxiv.org/pdf/alg-geom/9202012",
7
    # 提取模式：控制如何从 PDF 中解析和提取文本结构
8
    #   plain  提取文本，默认值
9
    #   layout 布局感知提取模式，通常会通过插入大量的空格、换行符，
10
    #          来模拟原文档中的多栏、缩进和间距
11
    #          （适用场景：学术论文（如 arXiv 论文）、多栏报刊杂志、带左右分栏的合同）
12
    extraction_mode="plain",
13
)
14
docs = loader.load()
15
print(len(docs))          # 每个页面一个 Document

PDF 加载（方式 2）：MinerU#

MinerU 提供了 PDF、Word、PPT、图片等文件的解析，支持图像提取、OCR、公式、表格解析等功能；调用在线服务（https://mineru.net/apiManage/docs ）可以从本地批量上传文件进行解析，并接收解析结果。

需要在 .env 里提供：

1
# MinerU的API_TOKEN
2
MINERU_API_TOKEN=<你的API TOKEN>

完整脚本分三步：申请上传链接并逐个上传 → 轮询批量任务状态 → 把解析结果的 zip 下载到本地：

1
import os
2
import time
3
import requests
4
from dotenv import load_dotenv
5

6
load_dotenv(override=True)
7

8

9
def upload_files(file_paths: list[str]) -> str:
10
    """批量上传文件"""
11
    url = "https://mineru.net/api/v4/file-urls/batch"
12
    api_token = os.getenv("MINERU_API_TOKEN")
13
    header = {
14
        "Content-Type": "application/json",
15
        "Authorization": f"Bearer {api_token}",
16
    }
17

18
    files_info = [
19
        {
20
            "name": os.path.basename(file_path),
21
            "is_ocr": True,            # 扫描版 PDF 要开 OCR
22
            "data_id": f"file_{i}",    # 自定义任务标识，用于后续对账
23
        }
24
        for i, file_path in enumerate(file_paths)
25
    ]
26

27
    data = {
28
        "enable_formula": True,        # 解析公式
29
        "enable_table": True,          # 解析表格
30
        "language": "ch",              # 文档语言
31
        "files": files_info,
32
    }
33

34
    try:
35
        response = requests.post(url, headers=header, json=data)
36
        if response.status_code == 200:
37
            result = response.json()
38
            print("response success. result:{}".format(result))
39

40
            if result["code"] == 0:
41
                batch_id = result["data"]["batch_id"]
42
                urls = result["data"]["file_urls"]
43
                print("batch_id:{}\nurls:{}".format(batch_id, urls))
44

45
                # 拿到预签名 URL 后，用 PUT 把文件本体传上去
46
                for i in range(0, len(urls)):
47
                    with open(file_paths[i], "rb") as f:
48
                        res_upload = requests.put(urls[i], data=f)
49
                        if res_upload.status_code == 200:
50
                            print(f"{urls[i]} upload success")
51
                        else:
52
                            print(f"{urls[i]} upload failed")
53
                            return None
54

55
                return batch_id
56
            else:
57
                print("apply upload url failed, reason:{}".format(result.get("msg")))
58
                return None
59
        else:
60
            print(
61
                "response not success. status:{} ,result:{}".format(
62
                    response.status_code, response.text
63
                )
64
            )
65
            return None
66

67
    except Exception as err:
68
        print(err)
69
        return None
70

71

72
def download_files(batch_id):
73
    """批量获取任务结果"""
74
    if not batch_id:
75
        print("batch_id为空，跳过下载")
76
        return
77

78
    os.makedirs("parsed_files", exist_ok=True)
79

80
    url = f"https://mineru.net/api/v4/extract-results/batch/{batch_id}"
81
    api_token = os.getenv("MINERU_API_TOKEN")
82
    header = {
83
        "Content-Type": "application/json",
84
        "Authorization": f"Bearer {api_token}",
85
    }
86

87
    failed_files = set()
88
    done_files = set()
89

90
    while True:
91
        res = requests.get(url, headers=header)
92
        result_json = res.json()
93

94
        if res.status_code != 200 or result_json.get("code") != 0:
95
            print("get result failed:", result_json)
96
            break
97

98
        extract_results = result_json["data"]["extract_result"]
99

100
        for result in extract_results:
101
            data_id = result["data_id"]
102

103
            if result["state"] == "failed":
104
                failed_files.add(data_id)
105

106
            elif result["state"] == "done" and data_id not in done_files:
107
                done_files.add(data_id)
108

109
                full_zip_url = result["full_zip_url"]
110
                res_download = requests.get(full_zip_url, stream=True)
111

112
                with open(
113
                    f"parsed_files/{result['file_name']}_{result['data_id']}.zip", "wb"
114
                ) as f:
115
                    for chunk in res_download.iter_content(chunk_size=1024):
116
                        if chunk:
117
                            f.write(chunk)
118

119
        # 全部任务都有结果（成功或失败）才退出轮询
120
        if len(failed_files) + len(done_files) == len(extract_results):
121
            break
122

123
        time.sleep(5)      # 每 5 秒查一次
124

125
    for i in failed_files:
126
        print("failed:", i)
127

128
    for i in done_files:
129
        print("done:", i)
130

131

132
file_paths = ["../asset/load/04-sample.pdf"]
133
batch_id = upload_files(file_paths)
134

135
if batch_id:
136
    download_files(batch_id)

Word 加载：UnstructuredWordDocumentLoader#

1
from langchain_community.document_loaders import UnstructuredWordDocumentLoader
2

3
loader = UnstructuredWordDocumentLoader(
4
    file_path="../asset/load/05-sgg_chat.docx",
5
    # 加载模式：
6
    #   single   返回单个 Document 对象
7
    #   elements 按标题等元素切分文档
8
    mode="single",
9
)
10
docs = loader.load()
11
print(len(docs))
12
print(docs)

Markdown 加载：UnstructuredMarkdownLoader#

举例1：整篇加载（mode="single"）

1
from langchain_community.document_loaders import UnstructuredMarkdownLoader
2
from pprint import pprint
3

4
loader = UnstructuredMarkdownLoader(
5
    file_path="../asset/load/06-load.md",
6
    # 加载模式:
7
    #   single 返回单个Document对象
8
    #   elements 按标题等元素切分文档
9
    mode="single",
10
    # 解析策略：
11
    #   "fast"（快速模式），它会以最快的速度提取文本，不进行复杂的版面分析
12
    #   "hi_res" 高分辨率模式
13
    strategy="fast",
14
)
15
docs = loader.load()
16
print(len(docs))
17
pprint(docs)

举例2：精细分割文档，保留结构信息（mode="elements"）

把 Markdown 按语义元素（标题、段落、列表、表格等）拆分成多个独立的小文档（Element 对象），而不是返回单个大文档——指定 mode="elements" 就能保持这种分离：

1
md_loader = UnstructuredMarkdownLoader(
2
    file_path="../asset/load/06-load.md",
3
    mode="elements",
4
    strategy="fast",
5
)
6
docs = md_loader.load()
7
print(len(docs))          # 19：一篇文档被拆成 19 个小块
8
for doc in docs:
9
    pprint(doc.page_content)

课程示例的拆解结果是标题、段落、列表项各自成块：

1
'自然语言处理技术文档'
2
'本文档用于测试UnstructuredMarkdownLoader的中文处理能力。'
3
'第一章：简介'
4
'自然语言处理(NLP)是人工智能的重要分支，主要技术包括：'
5
'文本分类'
6
...
7
'2.2 代码示例'
8
'```python from transformers import pipeline'

HTML 加载：UnstructuredHTMLLoader#

1
from langchain_community.document_loaders import UnstructuredHTMLLoader
2

3
# strategy:
4
#   "fast" 解析加载html文件速度比较快（但可能丢失部分结构或元数据）
5
#   "hi_res" (高分辨率解析) 解析精准（速度慢一些）
6
#   "ocr_only" 强制使用ocr提取文本，仅仅适用于图像（对HTML无效）
7
#
8
# mode ：one of `{'paged', 'elements', 'single'}`
9
#   "elements" 按语义元素（标题、段落、列表、表格等）拆分成多个独立的小文档
10
loader = UnstructuredHTMLLoader(
11
    file_path="../asset/load/07-load.html",
12
    mode="elements",
13
    strategy="fast",
14
)
15
docs = loader.load()
16
print(len(docs))     # 16  ← 一个 HTML 文件拆出 16 个 Document
17
for doc in docs:
18
    pprint(doc)

目录批量加载：DirectoryLoader#

除了单个文件，也可以批量加载一个文件夹内的所有文件：

1
from langchain_community.document_loaders import DirectoryLoader
2
from langchain_community.document_loaders import PythonLoader
3

4
directory_loader = DirectoryLoader(
5
    path="../asset/load",
6
    glob="*.py",             # 文件匹配模式（过滤器）。使用标准的 Unix 路径通配符
7
    use_multithreading=True, # 是否启用多线程。True 表示 LangChain 会并发读取多个文件
8
    show_progress=True,      # 是否显示进度条
9
    loader_cls=PythonLoader, # 指定底层核心加载器
10
)
11
docs = directory_loader.load()
12
print(len(docs))

本机实测（同一目录下的 4 个 .py）：控制台先打印进度条 100%|██████████| 4/4 [00:00<00:00, 2251.67it/s]，随后 len(docs) == 4，每个 Document 的 metadata 是 {'source': '..\\asset\\load\\08-fun.py'} 这种形式。

了解：BaseLoader 与 Document 类#

一方面，LangChain 在设计时要保证 Source 中各种不同的数据源，接下来的流程可以用统一的形式读取、调用；另一方面，为什么 PDFLoader、TextLoader 等 Document Loader 都用 load() 加载、都用 .page_content 和 .metadata 读取数据？

【解答】每一个在 LangChain 中集成的文档加载器，都要继承自 BaseLoader（文档加载器），BaseLoader 提供了一个名为 load 的公开方法，用于从配置的不同数据源加载数据，全部作为 Document 对象。实现逻辑如下（本机实测源码）：

1
class BaseLoader(ABC):
2
    """Interface for loading documents.
3

4
    Implementations should implement the lazy-loading method using generators
5
    to avoid loading all documents into memory at once.
6

7
    The `load` method will remain as is for backwards compatibility, but its
8
    implementation should be just `list(self.lazy_load())`.
9
    """
10

11
    # Sub-classes should not implement this method directly. Instead, they
12
    # should implement the lazy load method.
13
    def load(self) -> list[Document]:
14
        """Load data into `Document` objects."""
15
        return list(self.lazy_load())
16

17
    async def aload(self) -> list[Document]:
18
        """Load data into `Document` objects."""
19
        return [document async for document in self.alazy_load()]

这解释了三条约定：

1. 任何具体实现的 loader，最少要实现 load()；实际更推荐实现生成器版的 lazy_load()——“延迟加载缓解大文件内存压力”这句话就是这么来的；
2. load() 本身不应被重写，它干的事就是 list(self.lazy_load())；异步接口则是 aload() / alazy_load()；
3. load_and_split() 也定义在这个基类里，并且已标注为弃用（见上文）。

Document 类允许用户与文档内容交互，其继承体系如下（本机实测 Document.__mro__）：

1
Serializable
2
   ↑
3
BaseMedia
4
   ├── id: str | None        # 可选的文档标识符（理想是 UUID，但不强制）
5
   ├── metadata: dict        # 与内容关联的任意元数据
6
   ↑
7
Document
8
   ├── page_content: str     # 真正的文档内容
9
   └── type: Literal["Document"] = "Document"

Document 源码（本机实测，注释原意保留）：

1
class Document(BaseMedia):
2
    """Class for storing a piece of text and associated metadata.
3

4
    !!! note
5

6
        `Document` is for **retrieval workflows**, not chat I/O. For sending text
7
        to an LLM in a conversation, use message types from `langchain.messages`.
8
    """
9

10
    page_content: str
11
    """String text."""
12

13
    type: Literal["Document"] = "Document"
14

15
    def __init__(self, page_content: str, **kwargs: Any) -> None:
16
        """Pass page_content in as positional or named arg."""
17
        # my-py is complaining that page_content is not defined on the base class.
18
        # Here, we're relying on pydantic base class to handle the validation.
19
        super().__init__(page_content=page_content, **kwargs)

为什么必须切#

切块之后才能向量化并存入数据库；同时切块也决定了每次送给模型的上下文片段有多长（要适配模型的上下文窗口限制）。

获取 Document 对象后之所以必须把它切成一个个小块（Chunk），课程给了三条理由：

无论是在存储还是检索过程中，都以这些块（chunk）为基本单位——这样才能有效避免内容噪声干扰和超出最大 Token 的问题。

五种切分策略#

方法 2、3 只看字符、不考虑语义，容易造成主题断裂；方法 4 结合了固定长度与语义分析，能更好保证每段落含完整主题；方法 5 精度高但慢。

TextSplitter 源码分析#

所有切分器都继承自 TextSplitter（本机实测源码）：

1
class TextSplitter(BaseDocumentTransformer, ABC):
2
    """用于将文本切分为多个块的接口。"""
3

4
    def __init__(
5
        self,
6
        chunk_size: int = 4000,
7
        chunk_overlap: int = 200,
8
        length_function: Callable[[str], int] = len,
9
        keep_separator: bool | Literal["start", "end"] = False,
10
        add_start_index: bool = False,
11
        strip_whitespace: bool = True,
12
    ) -> None:
13
        ...

（前三个参数在下面的”核心参数”里会以 CharacterTextSplitter 为例再展开；keep_separator / add_start_index / strip_whitespace 是基类提供的通用能力，本节后面也会给实测例子。）

三条初始化校验（本机实测，都会抛 ValueError）：

1
CharacterTextSplitter(chunk_size=0)                     -> ValueError: chunk_size must be > 0, got 0
2
CharacterTextSplitter(chunk_size=10, chunk_overlap=-1)  -> ValueError: chunk_overlap must be >= 0, got -1
3
CharacterTextSplitter(chunk_size=10, chunk_overlap=20)  -> ValueError: Got a larger chunk overlap (20) than chunk size (10), should be smaller.

一眼就能看出：chunk_overlap > chunk_size 是直接报错的，不是警告。

三种调用方式与调用链：

1
split_documents(documents)
2
    -> create_documents(texts, metadatas=metadatas)
3
        -> split_text(text)

另外两个接口：

• transform_documents(documents)：实现的是 BaseDocumentTransformer 的接口，内部就一行 return self.split_documents(list(documents))——它的意义是让切分器能塞进”文档转换器链”里统一调用。
• from_huggingface_tokenizer(tokenizer, **kwargs) / from_tiktoken_encoder(encoding_name="gpt2", model_name=None, ...)：两个 @classmethod 构造器，用来把 length_function 换成 token 计数器（详见下文”Token 与按 Token 切分”）。

1
@classmethod
2
def from_tiktoken_encoder(
3
    cls,
4
    encoding_name: str = "gpt2",
5
    model_name: str | None = None,     # 提供该参数时会覆盖 encoding_name
6
    allowed_special: Literal["all"] | AbstractSet[str] = set(),
7
    disallowed_special: Literal["all"] | Collection[str] = "all",
8
    **kwargs: Any,
9
) -> Self:
10
    ...

核心参数#

以 CharacterTextSplitter 为例：

图：chunk_size 是每个切块的最大长度，chunk_overlap 是相邻两个切块之间的重叠部分（不重叠就不会切出灰块那一段）

1
cs = CharacterTextSplitter(chunk_size=50, chunk_overlap=5, separator="")
2
# 结果：[50 字符的块, 20 字符的块]  ← 第二块只有 20 字

四种具体实现（实测）#

图：RecursiveCharacterTextSplitter 的”先下探、再回溯”——① 按 separators 从 \n\n、\n、 到 "" 逐级递归切分，超长块继续往下探，合格的块进 good_splits；② 再按顺序把小块合并回 chunk，超过 chunk_size 就输出并保留 chunk_overlap

1
块1: ❌ '退换货政策：客户在收到�'
2
块2: ❌ '�品之后 7 个自然日内，如果�'
3
...
4
块11: ❌ '�。'

1
from langchain_text_splitters import RecursiveCharacterTextSplitter
2

3
splitter = RecursiveCharacterTextSplitter(
4
    chunk_size=200,
5
    chunk_overlap=80,
6
    separators=["\n\n", "\n", "。", " ", ""],   # 切分策略：优先按段落，其次句子
7
)
8
chunks = splitter.split_documents(documents)     # 直接吃 Document 列表
9
print(f"文档共切分为 {len(chunks)} 个 chunk")

1
keep_separator=True（默认）→ 3 块
2
  块1(48): '退换货政策：客户在收到商品之后 7 个自然日内可以申请无理由退货，15 个自然日内可以申请换货。'
3
  块2(30): '=============================='        ← 只装分隔符的块
4
  块3(36): '发货时间：订单支付成功后 24 小时内发货，遇到大促或法定节假日会顺延。'
5

6
keep_separator=False → 2 块
7
  块1(48): '退换货政策：…'
8
  块2(36): '发货时间：…'

separator 优先原则（四条）#

给 CharacterTextSplitter 指定 separator（比如 "。"）之后，切分器会先尝试在分隔符处分割，然后再考虑 chunk_size——这是为了避免在句子中间硬性切断。这套设计带来四条必须记住的结论：

1. 优先保持语义完整性（不切断句子）
2. 避免产生无意义的碎片（如半个单词、不完整句子）
3. 如果 chunk_size 比片段还小，就无法拆分片段，导致 overlap 失效
4. chunk_overlap 仅在合并后的片段之间生效（前提是 chunk_size 足够大）；如果没有片段被合并，overlap 同样失效

实测（text = "这是第一段文本。这是第二段内容。最后一段结束。"）：

1
# ① chunk_size=6 比"这是第一段文本"（7 字）还小 → 块长超标，块与块之间没有任何重叠
2
CharacterTextSplitter(separator="。", chunk_size=6, chunk_overlap=3)
3
#   Created a chunk of size 7, which is longer than the specified 6     ← 控制台会打印这句
4
#   块1(7): '这是第一段文本' / 块2(7): '这是第二段内容' / 块3(6): '最后一段结束'
5

6
# ② chunk_size=20 足够大 → 相邻块之间出现重叠（"这是第二段内容"被前后两块共享）
7
CharacterTextSplitter(separator="。", chunk_size=20, chunk_overlap=8)
8
#   块1(15): '这是第一段文本。这是第二段内容'
9
#   块2(14): '这是第二段内容。最后一段结束'

1
Created a chunk of size 33, which is longer than the specified 30

keep_separator 的三种取值#

keep_separator 控制分隔符本身要不要留在块里。实测对比（同一段文本，chunk_size=20, chunk_overlap=8）：

**什么时候要在意它？**处理中文时，句尾标点承载着语气和逻辑，课程自定义中文标点切分就配了 keep_separator=True；反过来，用业务分隔线（"\n=====…=====\n"）切分时就要设 False，免得分隔线自己变成一个块（见上文那招）。

中文文档：自定义分隔符#

有些书写系统没有单词边界，例如中文、日文和泰文。使用默认分隔符列表 ["\n\n", "\n", " ", ""] 分割这类文本，可能导致句子被错误地切断。为了尽量让句子待在一起、保持语义完整，可以自定义分割字符，覆盖分隔符列表以包含额外的标点符号：

1
from langchain_text_splitters import RecursiveCharacterTextSplitter
2

3
text_splitter = RecursiveCharacterTextSplitter(
4
    chunk_size=200,
5
    chunk_overlap=20,      # 增加重叠字符
6
    separators=["\n\n", "\n", "。", "！", "？", "……", "，", ""],  # 添加中文标点
7
    length_function=len,
8
    keep_separator=True,   # 保留句尾标点（如 ……），避免切割后丢失语气和逻辑
9
    add_start_index=True,  # 在 metadata 里记下每块在原文中的起始位置
10
)

效果：算法会优先在句号、感叹号、问号、省略号、逗号处切割，尽量保持句子完整。add_start_index=True 则让每个 Document 的 metadata 里多出一个 start_index（本机实测形如 {'start_index': 0}）——排查”这句话到底来自原文哪个位置”时特别有用。

递归切分的底层：先拆分、后合并#

RecursiveCharacterTextSplitter 的底层处理逻辑分两步（了解即可）：

① 先拆分（下探）

1. 底层的 self._split_text() 按照分隔符列表的顺序，用当前递归层可用的第一个分隔符把文档切成若干块；
2. 如果切分后的块大小仍 > chunk_size，就调用 self._split_text() 用下一个分隔符递归处理大块；
3. 直到所有块大小都不超过 chunk_size，停止递归。

② 后合并（回溯）

合并过程不是一次完成的，为便于理解可以抽象成”一次合并”，最终得到完整块列表 final_chunks：

1. 遍历切分后的 chunk 列表：对每个当前 chunk，先判断把它加入候选窗口后，加上必要的合并分隔符（默认为 ""），候选窗口长度会不会超过 chunk_size；若超过，则先把历史块合并为整体，添加到 final_chunks；
2. 然后从候选列表左侧逐个弹出 chunk，直到：① 剩余 chunk 拼接后的累计长度不大于 chunk_overlap；② 并且”剩余部分 + 下一个 chunk + 合并分隔符长度”之和不超过 chunk_size；
3. 这样可以让拆分过细的小块同时出现在前后两个相邻的块中——这就是chunk_overlap 重叠的来源。

处理完成后，就得到了既满足 chunk_size 约束、又按 chunk_overlap 保留重叠区域的 chunk 列表。

Token 与按 Token 切分#

**什么是 Token？**对模型而言，Token 是文本的最小处理单位：

为什么按 Token 分割？

1. 语言模型对输入长度的限制是基于 Token 数（如 GPT-4 的 8k / 32k Token 上限），直接按字符或单词分割可能导致实际 Token 数超限；
2. 大语言模型通常以 token 数量作为计量（或收费）的依据，所以按 token 切分也有助于控制成本。

TokenTextSplitter 的核心依据是 Token 数量 + 自然边界：严格按 token 数量切分，但会优先在自然边界（如句尾）处切断，以尽量保证语义完整。

它会用到 token 编码器：把文本切分为 token 序列，再把 token 序列映射为 ID 序列——本质上是一个 tokenizer。可选编码器位于 openai_public.py 的全局变量中：

1
ENCODING_CONSTRUCTORS = {
2
    "gpt2": gpt2,
3
    "r50k_base": r50k_base,
4
    "p50k_base": p50k_base,
5
    "p50k_edit": p50k_edit,
6
    "cl100k_base": cl100k_base,
7
    "o200k_base": o200k_base,
8
    "o200k_harmony": o200k_harmony,
9
}

1
from langchain_text_splitters import TokenTextSplitter
2

3
text_splitter = TokenTextSplitter(
4
    chunk_size=33,                 # 最大 token 数为 33
5
    chunk_overlap=0,               # 重叠 token 数为 0
6
    # model_name="gpt-4",          # 选择 GPT-4 模型的编码器（会覆盖 encoding_name）
7
    encoding_name="cl100k_base",   # 使用 OpenAI 的编码器，将文本转换为 token 序列
8
)

也可以把任意切分器的 length_function 换成 token 计数器——这样 chunk_size 数的就是 token 而不是字符：

1
from langchain_text_splitters import CharacterTextSplitter
2
import tiktoken  # 用于计算Token数量
3

4
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
5
    encoding_name="cl100k_base",  # 使用 OpenAI 的编码器
6
    chunk_size=18,                 # 注意：这里的 18 是 18 个 token
7
    chunk_overlap=0,
8
    separator="。",                 # 指定中文句号为分隔符
9
    keep_separator=False,
10
)

1
cl100k_base：  hello=1   ChatGPT=3    人工智能=5     LangChain=2
2
o200k_base ：  hello=1   ChatGPT=2    人工智能=2     LangChain=2
3
gpt2       ：  hello=1   ChatGPT=3    人工智能=8     LangChain=3

语义分块：SemanticChunker#

Semantic Chunking（语义分块）是 LangChain 中一种更高级的文本分割方法：它超越传统的”按字符数或固定大小”的分块方式，而是根据文本的语义结构进行智能分块，使每个分块保持语义完整性，从而提高 RAG 的效果。

做法是：把文本转成向量（Embedding），计算前后句子的语义差异；当发现前后两句的语义变化很大（超过设定的阈值）时，就在这里一刀切断。这样能保证切分出来的每个 Chunk 在含义上是完整、连贯的。

1
# pip install langchain_experimental
2
from langchain_experimental.text_splitter import SemanticChunker
3

4
# embedding_model 就是嵌入模型（下一章 25 详讲：init_embeddings(...) 或 OpenAIEmbeddings(...)）
5
text_splitter = SemanticChunker(
6
    embeddings=embedding_model,
7
    breakpoint_threshold_type="percentile",      # 断点阈值类型
8
    breakpoint_threshold_amount=65.0,            # 断点阈值数量（阈值越低 → 分割越敏感）
9
    sentence_split_regex=r"(?<=[。？！])\s+",     # 句子切分正则
10
)
11
docs = text_splitter.create_documents(texts=[state_of_the_union])

参数说明：

底层逻辑：先按正则把文本切成 chunk 列表 → 计算相邻 chunk 之间的距离 → 按 breakpoint_threshold_type / breakpoint_threshold_amount 的规则确定切分位置 → 按切分位置合并相邻块。

其它切分器#

① HTMLHeaderTextSplitter：按 HTML 标题切

它根据 HTML 的标题标签（<h1>、<h2> 等）把文档划分为逻辑分块，同时保留标题的层级结构信息：

1
from langchain_text_splitters import HTMLHeaderTextSplitter
2

3
headers_to_split_on = [("h1", "标题1"), ("h2", "标题2"), ("h3", "标题3")]
4
html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
5
html_header_splits = html_splitter.split_text(html_string)

本机实测输出（标题自己也会变成一个块，每个块的 metadata 自动继承各级标题）：

1
metadata={'标题1': '欢迎来到尚硅谷！'}                                  content='欢迎来到尚硅谷！'
2
metadata={'标题1': '欢迎来到尚硅谷！'}                                  content='尚硅谷是专门培训IT技术方向'
3
metadata={'标题1': '欢迎来到尚硅谷！', '标题2': '尚硅谷老师简介'}           content='尚硅谷老师简介'
4
metadata={'标题1': '欢迎来到尚硅谷！', '标题2': '尚硅谷老师简介'}           content='尚硅谷老师拥有多年教学经验，都是从一线互联网下来'
5
metadata={'标题1': '欢迎来到尚硅谷！', '标题2': ..., '标题3': '尚硅谷北京校区'}  content='尚硅谷北京校区'

② CodeTextSplitter：按代码语法切

专为代码文件设计。本机实测支持的语言（28 种）：

1
['cpp', 'go', 'java', 'kotlin', 'js', 'ts', 'php', 'proto', 'python', 'r', 'rst', 'ruby',
2
 'rust', 'scala', 'swift', 'markdown', 'latex', 'html', 'sol', 'csharp', 'cobol', 'c',
3
 'lua', 'perl', 'haskell', 'elixir', 'powershell', 'visualbasic6']

用法是 RecursiveCharacterTextSplitter.from_language(...)——它根据编程语言的语法结构（函数、类、代码块等）智能拆分代码，不会在函数或类的中间截断：

1
from langchain_text_splitters import Language, RecursiveCharacterTextSplitter
2

3
python_splitter = RecursiveCharacterTextSplitter.from_language(
4
    language=Language.PYTHON,
5
    chunk_size=50,
6
    chunk_overlap=0,
7
)
8
python_docs = python_splitter.create_documents(texts=[PYTHON_CODE])

本机实测（两个函数、chunk_size=50）：两个函数各自完整成一个块，没有被切一半——

1
'def hello_world():\n    print("Hello, World!")'
2
'def hello_world1():\n    print("Hello, World1!")'

③ MarkdownTextSplitter：按 Markdown 语法切

因为 Markdown 由 # / ## 等多级标题组织内容，它的切分策略就是根据标题来分割：

1
from langchain_text_splitters import MarkdownTextSplitter
2

3
splitter = MarkdownTextSplitter(chunk_size=30, chunk_overlap=0)
4
splitter._is_separator_regex = True    # 关键步骤：直接修改实例属性，强制把分隔符视为正则表达式
5
docs = splitter.create_documents(texts=[markdown_text])

它的默认分隔符本身就是一组正则（本机实测）：

1
['\n#{1,6} ', '```\n', '\n\\*\\*\\*+\n', '\n---+\n', '\n___+\n', '\n\n', '\n', ' ', '']

_is_separator_regex = True 就是让”按标题切”这条规则真正生效——本机实测切出的两块正好是 '# 一级标题\n这是一级标题下的内容' 和 '## 二级标题\n- 二级下列表项1\n- 二级下列表项2'，列表项没有和自己的标题被拆散。

一、回忆填空（写完再展开对答案）#

1. 大模型的三个局限：知识____、知识____、____
2. RAG 的全称是____，它结合了信息____与文本____
3. RAG 相比提示词工程的优点是上下文更____；相比模型微调的优点是能提升____和可靠性；缺点是响应____较高、消耗大量____
4. RAG 六个环节：Source → ____ → Transform → ____ → Store → ____
5. 加载器把数据加载成 ____ 对象，它由 ____ 和 metadata 两部分组成
6. 加载器用 ____() 加载、用 ____() 一步完成加载+切分
7. 五种切分策略中最常用的是____字符切分；它通过____方式动态确定切分点
8. CharacterTextSplitter 的两个核心参数：____（每块最大字符数，默认 4000）和 ____（块间重叠字符数，默认 200）
9. 按 Markdown 标题切分的切分器是 ____TextSplitter，它会把标题写进____
10. split_documents() 吃的是 Document 列表，返回的也是____列表（保留 metadata）；split_text() 吃的是____

二、裸写题#

• 2-1 加载一个 txt 并看清 Document 对象 自己造一个 test.txt（写两三句中文），用 TextLoader 加载，打印：文档数量、page_content、metadata。

  提示（先自己想，实在想不出再点开）

  一级 · 思路：加载器干的事就是”读文件 → 包成 Document” 二级 · 方法：TextLoader("test.txt", encoding="utf-8").load() 三级 · 骨架：中文文件一定要显式传 encoding="utf-8"，否则 Windows 下可能乱码

• 2-2 三种切分器对比 用同一段 100 字左右的中文，分别用 CharacterTextSplitter(chunk_size=50)、RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=10)、TokenTextSplitter(chunk_size=33) 切分，打印每块的长度和内容，比较差异。

  提示

  一级 · 思路：都是”切”，差别在”从哪里下刀” 二级 · 方法：splitter.split_text(text) 三级 · 骨架：观察 Recursive 是否更倾向在换行/句号处断开

• 2-3 用 Markdown 切分器保住标题结构 造一个带 # 一级标题、## 二级标题 的 md 文本，用 MarkdownHeaderTextSplitter(headers_to_split_on=[('#', 'h1'), ('##', 'h2')]) 切分，打印每块的 content 和 metadata。

  提示

  一级 · 思路：标题变成了 metadata，正文才是 content 二级 · 方法：headers_to_split_on 是 (标题符号, 元数据键名) 的元组列表 三级 · 骨架：想一想：检索时”标题进 metadata”有什么用？（提示：可以按章节过滤）

三、综合题#

• 3-1 给一份知识文档设计切分方案 自造一份”客服知识库”文本：3 条较长的问答（每条 100 字以上），条目之间用 \n==============================\n 分隔。然后依次做四件事：

  1. 用 TextLoader 加载成 Document
  2. 用 RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=0, separators=[分隔线, "\n\n", "\n", "。", " "]) 切分，打印每块的字符数和内容
  3. 观察两个现象：长条目会被切成两块、那行分隔线自己成了一个块；然后加上 keep_separator=False 再切一次，看分隔线块是否消失
  4. 再用 TokenTextSplitter(chunk_size=30) 切同一段中文，检查有没有出现 � 乱码，并解释原因
  提示（先自己想，实在想不出再点开）

  一级 · 思路：切分的目标不是”块数少”，而是”每块自成一个小主题、且不含垃圾内容” 二级 · 方法：separators=[sep, "\n\n", "\n", "。", " "] + keep_separator=False 三级 · 骨架：判断”垃圾块”的简单标准——chunk.page_content 里有没有实际信息（比如只剩分隔线就得处理掉）

1
import os
2

3
from langchain_community.document_loaders import TextLoader
4
from langchain_text_splitters import (
5
    CharacterTextSplitter,
6
    MarkdownHeaderTextSplitter,
7
    RecursiveCharacterTextSplitter,
8
    TokenTextSplitter,
9
)
10

11
BASE = os.path.dirname(os.path.abspath(__file__))
12

13
# ---------- 2-1 加载 txt ----------
14
txt_path = os.path.join(BASE, "test.txt")
15
with open(txt_path, "w", encoding="utf-8") as f:
16
    f.write("LangChain 是一个用于开发语言模型应用的框架。\n它提供了一套工具和抽象。\n")
17

18
docs = TextLoader(txt_path, encoding="utf-8").load()
19
print("文档数量:", len(docs))
20
print("内容:", docs[0].page_content)
21
print("元数据:", docs[0].metadata)
22

23
# ---------- 2-2 三种切分器对比 ----------
24
text = ("LangChain 是一个用于开发由语言模型驱动的应用程序的框架，"
25
        "它提供了一套工具和抽象，使开发者能够更容易地构建复杂的应用程序。"
26
        "它让模型能够结合检索、记忆与工具调用完成任务。")
27

28
cs = CharacterTextSplitter(chunk_size=50, chunk_overlap=0, separator="")
29
rs = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=10)
30
ts = TokenTextSplitter(chunk_size=33, chunk_overlap=5)
31

32
for name, splitter in [("Character", cs), ("Recursive", rs)]:
33
    chunks = splitter.split_text(text)
34
    print(f"\n【{name}】共 {len(chunks)} 块")
35
    for i, c in enumerate(chunks, 1):
36
        print(f"  块{i} ({len(c)} 字): {c}")
37

38
# ---------- 2-3 Markdown 切分 ----------
39
md = "# 一级标题\n\n正文一\n\n## 二级标题\n\n正文二\n"
40
md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[("#", "h1"), ("##", "h2")])
41
for d in md_splitter.split_text(md):
42
    print("\ncontent:", repr(d.page_content), "| metadata:", d.metadata)
43

44
# ---------- 3-1 知识库切分方案 ----------
45
sep = "\n==============================\n"
46
entries = [
47
    "退换货政策：客户在收到商品之后 7 个自然日内，如果商品不影响二次销售，可以申请无理由退货；"
48
    "15 个自然日内，如果商品存在质量问题，可以申请换货。退货运费由谁承担要看具体原因：质量问题由平台承担，"
49
    "非质量问题由客户承担。特殊商品（如定制商品、生鲜）不支持无理由退货。",
50
    "发货时间：正常情况下，订单支付成功后 24 小时内由仓库安排发货，遇到大促或法定节假日会顺延。"
51
    "发货后会推送物流单号，客户可以在订单详情页实时查看物流轨迹。若超过 48 小时仍未发货，可以联系客服加急处理。",
52
    "运费说明：单笔订单金额满 99 元包邮；不满 99 元收取 8 元基础运费。偏远地区（新疆、西藏等）单独计费，"
53
    "具体金额以下单页面显示为准。会员用户每月享有 3 次免运费权益，权益当月有效，不累计到下个月。",
54
]
55

56
kb_path = os.path.join(BASE, "knowledge.txt")
57
with open(kb_path, "w", encoding="utf-8") as f:
58
    f.write(sep.join(entries))
59
documents = TextLoader(kb_path, encoding="utf-8").load()
60

61
def show(label, splitter, docs_in):
62
    chunks = splitter.split_documents(docs_in)
63
    print(f"\n【{label}】{len(chunks)} 块")
64
    for i, c in enumerate(chunks, 1):
65
        body = c.page_content.replace("\n", "⏎")
66
        print(f"  块{i}({len(c.page_content)}字): {body[:60]}{'…' if len(body) > 60 else ''}")
67
    return chunks
68

69
# ① 默认 keep_separator=True：分隔线会自己变成一个块
70
show(
71
    "业务分隔线优先（默认 keep_separator=True）",
72
    RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=0,
73
                                   separators=[sep, "\n\n", "\n", "。", " "]),
74
    documents,
75
)
76

77
# ② keep_separator=False：分隔线被丢掉，块更干净
78
show(
79
    "业务分隔线优先（keep_separator=False）",
80
    RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=0, keep_separator=False,
81
                                   separators=[sep, "\n\n", "\n", "。", " "]),
82
    documents,
83
)
84

85
# ③ TokenTextSplitter 切中文：会出现 � 乱码（按字节级 token 边界切开汉字）
86
print("\n【TokenTextSplitter 切中文】")
87
tt = TokenTextSplitter(chunk_size=30, chunk_overlap=0)
88
for i, c in enumerate(tt.split_text(entries[0]), 1):
89
    print(f"  块{i}: {'❌乱码' if '�' in c else '✅正常'} {c[:36]!r}")
90

91
# 结论：
92
# 1) 中文用 RecursiveCharacterTextSplitter，别用 TokenTextSplitter
93
# 2) 用业务分隔符切分时设 keep_separator=False，避免"只有一行等号"的垃圾块进向量库