登录RAG-如何对文档分块
上文我们讲了RAG是如何进行数据加载的，那么文档加载完数据就能直接喂给大模型进行问答吗，答案是否定的。因为把所有的文档都一并喂给大模型，那么大模型接受的上下文是非常巨大的，这会超出大模型所支持的最大token，而且每次会话，都要把上下文喂给大模型才能回答我们问的问题，这使得大模型的响应速度会变得很慢，如果是调用在线的大模型API的话，一次问答会消耗很多的token，钱包顶不住啊。所以要将文档数据加载后，进行数据分块、向量嵌入、存入向量数据库，通过向量检索将有用的数据喂给大模型，最后生成结果返回。这一篇我们着重说明数据分块是怎么做的。
在展示文本分块前说明下什么是token：
- 在英文里，一个单词可能是一个
token,也可能被拆成多个。例如：playing可能拆成play+ing - 在中文里，通常一个汉字常常接近1个
token, 但也不绝对 - 标点、空格、换行也可能占
token
文档分块方法
字符分块
用单一分隔符进行文档分块，代码如下：
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import CharacterTextSplitter
# ---------- 1. 加载 PDF ----------
loader = PyMuPDFLoader("../document/企业财务报表分析-图表.pdf")
documents = loader.load()
# ---------- 2. 文档分块 ----------
# chunk_size: 每块最大字符数；chunk_overlap: 块与块之间的重叠字符数，避免语义被截断
text_splitter = CharacterTextSplitter(chunk_size=200, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)
# ---------- 3. 打印分块内容 ----------
print("=== 文档分块结果 ===\n")
print(f" CharacterTextSplitter -> {len(chunks)} 块")
char_lens = [len(c.page_content) for c in chunks]
print(f" Character: 最小={min(char_lens)}, 最大={max(char_lens)}, 平均={sum(char_lens)//len(char_lens)}")
for i, chunk in enumerate(chunks, 1):
print(f"--- 第 {i} 块 (共 {len(chunks)} 块) ---")
print(chunk.page_content.strip() or "(空)")
if chunk.metadata:
print(f"[元数据] {chunk.metadata}")
print("-" * 50)
print()
返回的部分结果：CharacterTextSplitter切分的文本
(C:\Users\yd-19\AppData\Roaming\Typora\typora-user-images\1773908015462.png)
文中的CharacterTextSplitter是按照字符长度切分文本，其配置是：
-
chunk_size=500: 每块最多约为200字符 -
chunk_overlap=50: 相邻块重叠50字符，减少语言被截断 - 不考虑语义，只看长度
这里有个问题就是虽然我们配置的文本块约为200个字符，但看返回的结果最大的文本块是1266个字符，远超200字符。这是为什么呢。因为CharacterTextSplitter的工作方式是：
- 先用
sparator把文本切开（默认是“\n\n”） - 然后把切出来的小段尝试合并，合并到接近
chunk_size为止 - 但如果某一段本身就超过了
chunk_size,它就不会再进一步切割
因为样例PDF里“\n\n”很少，CharacterTextSplitter按照“\n\n”切完块后，每段本身就很长，也不会对超长段再做二次切分。所以分块出来的结果最大文本块超过了200字符，并且切割出来的字符很不均匀。接下来我们介绍另一种分块方法。
递归分块
多级，按优先级递归分隔符，代码如下：
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
PDF_PATH = "../document/企业财务报表分析-图表.pdf"
loader = PyMuPDFLoader(PDF_PATH)
documents = loader.load()
recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=50,
)
chunks = recursive_splitter.split_documents(documents)
# ---------- 3. 打印分块内容 ----------
print("=== 文档分块结果 ===\n")
print(f" RecursiveCharacterTextSplitter -> {len(chunks)} 块")
char_lens = [len(c.page_content) for c in chunks]
print(f" Character: 最小={min(char_lens)}, 最大={max(char_lens)}, 平均={sum(char_lens)//len(char_lens)}")
for i, chunk in enumerate(chunks, 1):
print(f"--- 第 {i} 块 (共 {len(chunks)} 块) ---")
print(chunk.page_content.strip() or "(空)")
if chunk.metadata:
print(f"[元数据] {chunk.metadata}")
print("-" * 50)
print()
返回的部分结果：RecursiveCharacterTextSplitter切出的文本
通过结果我们可以看出，
RecursiveCharacterTextSplitter切出来的文本更多，更加均匀，更接近我们设置的字符数。RecursiveCharacterTextSplitter切割分隔符是通过递归：\n\n → \n → 空格 → 字符，对于超长块的处理，会自动降级到更细的分隔符继续切。
我们继续观察结果得知，切出来的内容语义并不完整，一段完整的话被切成两个分块，所以也要根据文中的内容进行策略分块。
分块思想
分层分块
按照文档的章节结构、句子边界进行分块，优先保留完整的句子，在元数据中加入页码、章节、分块数量。代码如下：
import re
from copy import deepcopy
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
PDF_PATH = "../document/企业财务报表分析-图表.pdf"
MAX_CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
CHAPTER_RE = re.compile(r"(?=(?:^|\n)[一二三四五六七八九十]+、)")
SECTION_RE = re.compile(r"(?=(?:^|\n)（[一二三四五六七八九十]+）)")
fallback_splitter = RecursiveCharacterTextSplitter(
chunk_size=MAX_CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=["。", "；", "\n", "，", " ", ""],
keep_separator=True,
)
def extract_heading(text: str, pattern: re.Pattern) -> str:
"""从块开头提取标题行。"""
first_line = text.strip().split("\n")[0].strip()
if pattern.search("\n" + first_line):
return first_line
return ""
def split_by_regex(text: str, pattern: re.Pattern) -> list[str]:
"""按正则切分，保留分隔符在各段开头。"""
parts = pattern.split(text)
result = []
for p in parts:
stripped = p.strip()
if stripped:
result.append(stripped)
return result if result else [text]
def hierarchical_chunk(docs: list[Document]) -> list[Document]:
full_text = "\n\n".join(doc.page_content for doc in docs)
base_meta = docs[0].metadata if docs else {}
chapters = split_by_regex(full_text, CHAPTER_RE)
chunks: list[Document] = []
for chapter_text in chapters:
chapter_heading = extract_heading(chapter_text, CHAPTER_RE)
sections = split_by_regex(chapter_text, SECTION_RE)
for section_text in sections:
section_heading = extract_heading(section_text, SECTION_RE)
meta = deepcopy(base_meta)
meta["chapter"] = chapter_heading
meta["section"] = section_heading
if len(section_text) <= MAX_CHUNK_SIZE:
chunks.append(Document(page_content=section_text.strip(), metadata=meta))
else:
sub_chunks = fallback_splitter.split_text(section_text)
for idx, sub in enumerate(sub_chunks):
sub_meta = deepcopy(meta)
sub_meta["sub_chunk"] = f"{idx + 1}/{len(sub_chunks)}"
chunks.append(Document(page_content=sub.strip(), metadata=sub_meta))
# 过小的块（如纯章节标题）合并到下一块，避免碎片
MIN_CHUNK_SIZE = 50
merged: list[Document] = []
carry = ""
for chunk in chunks:
if len(chunk.page_content) < MIN_CHUNK_SIZE:
carry += chunk.page_content + "\n"
else:
if carry:
chunk.page_content = carry + chunk.page_content
carry = ""
merged.append(chunk)
if carry and merged:
merged[-1].page_content += "\n" + carry.strip()
return merged
# ---------- 执行 ----------
loader = PyMuPDFLoader(PDF_PATH)
documents = loader.load()
chunks = hierarchical_chunk(documents)
# ---------- 打印结果 ----------
print(f"=== 分层分块结果（共 {len(chunks)} 块）===\n")
char_lens = [len(c.page_content) for c in chunks]
print(f" 长度统计: 最小={min(char_lens)}, 最大={max(char_lens)}, 平均={sum(char_lens) // len(char_lens)}\n")
for i, chunk in enumerate(chunks, 1):
ch = chunk.metadata.get("chapter", "")
sec = chunk.metadata.get("section", "")
sub = chunk.metadata.get("sub_chunk", "")
label = f"[{ch}]" if ch else ""
if sec:
label += f" [{sec}]"
if sub:
label += f" (子块 {sub})"
content = chunk.page_content
preview = content[:200] + "..." if len(content) > 200 else content
print(f"--- 第 {i}/{len(chunks)} 块 {label} (长度: {len(content)}) ---")
print(preview)
print("-" * 80)
print()
返回的部分结果：
这种分块的方法能保留语义的完整性，切出来的块自带章节的标签，定位精准
滑动窗口分块
滑动窗口分块不看标点、不看换行、不看章节，纯按字符位置滑动。
- 优点：块大小完全均匀，覆盖无死角（每个字符至少出现在 1~2 个块里）
- 缺点：会从句子/词中间切断，语义完整性最差
代码如下：
from copy import deepcopy
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_core.documents import Document
PDF_PATH = "../document/企业财务报表分析-图表.pdf"
WINDOW_SIZE = 300
STEP_SIZE = 200
def sliding_window_chunk(docs: list[Document], window: int, step: int) -> list[Document]:
"""
滑动窗口分块：固定窗口大小，按步长向前滑动。
window - step = 重叠字符数（本例 300 - 200 = 100 字符重叠）
"""
chunks: list[Document] = []
for doc in docs:
text = doc.page_content
if not text.strip():
continue
start = 0
chunk_idx = 0
while start < len(text):
end = start + window
segment = text[start:end].strip()
if segment:
meta = deepcopy(doc.metadata)
meta["chunk_index"] = chunk_idx
meta["char_start"] = start
meta["char_end"] = min(end, len(text))
chunks.append(Document(page_content=segment, metadata=meta))
chunk_idx += 1
start += step
return chunks
# ---------- 执行 ----------
loader = PyMuPDFLoader(PDF_PATH)
documents = loader.load()
chunks = sliding_window_chunk(documents, window=WINDOW_SIZE, step=STEP_SIZE)
# ---------- 打印结果 ----------
print(f"=== 滑动窗口分块结果（window={WINDOW_SIZE}, step={STEP_SIZE}, overlap={WINDOW_SIZE - STEP_SIZE}）===\n")
print(f" 共 {len(chunks)} 块")
char_lens = [len(c.page_content) for c in chunks]
print(f" 长度统计: 最小={min(char_lens)}, 最大={max(char_lens)}, 平均={sum(char_lens)//len(char_lens)}\n")
for i, chunk in enumerate(chunks, 1):
content = chunk.page_content
preview = content[:200] + "..." if len(content) > 200 else content
start = chunk.metadata["char_start"]
end = chunk.metadata["char_end"]
print(f"--- 第 {i}/{len(chunks)} 块 [字符 {start}~{end}] (长度: {len(content)}) ---")
print(preview)
print("-" * 80)
print()
返回部分结果：
句子边界优先分块
按照标点符号将整段文本拆成一句一句的，再把句子一句一句的往块里放，快满了就输出一块。输出一块后，不是从零开始。而是从前一块末尾回带几句（总字符数 ≤ chunk_overlap=50）作为新块的开头。回带也是以整句为单位，不会把句子劈开。
- 优点：每个块里的句子都是完整的，
embedding质量好，检索到的上下文读起来通顺。 - 缺点：不感知文档结构（章节/标题），可能把不同章节的内容拼到同一个块里。
import re
from copy import deepcopy
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
PDF_PATH = "../document/企业财务报表分析-图表.pdf"
CHUNK_SIZE = 300
CHUNK_OVERLAP = 50
def split_sentences_zh(text: str) -> list[str]:
"""按中文句号/问号/感叹号/分号切句，尽量保留句子语义完整。"""
text = text.strip()
if not text:
return []
parts = re.split(r"(?<=[。！？；!?;])\s*", text)
return [p.strip() for p in parts if p.strip()]
def sentence_aware_chunk_documents(
docs: list[Document],
chunk_size: int,
chunk_overlap: int,
) -> list[Document]:
"""先按句切，再按句合并；超长句再兜底按字符切分。"""
fallback_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
keep_separator=True,
)
chunks: list[Document] = []
overlap_chars = max(0, chunk_overlap)
for doc in docs:
sentences = split_sentences_zh(doc.page_content)
if not sentences:
continue
current_sentences: list[str] = []
current_len = 0
for sentence in sentences:
sent_len = len(sentence)
# 单句本身超长，先把当前块落盘，再对超长句做兜底切分
if sent_len > chunk_size:
if current_sentences:
content = "".join(current_sentences).strip()
if content:
chunks.append(Document(page_content=content, metadata=deepcopy(doc.metadata)))
current_sentences = []
current_len = 0
for sub in fallback_splitter.split_text(sentence):
sub = sub.strip()
if sub:
chunks.append(Document(page_content=sub, metadata=deepcopy(doc.metadata)))
continue
# 如果加上当前句会超长，则先输出当前块，再按 overlap 回带末尾句子
if current_sentences and (current_len + sent_len > chunk_size):
content = "".join(current_sentences).strip()
if content:
chunks.append(Document(page_content=content, metadata=deepcopy(doc.metadata)))
# 按字符数控制 overlap（以句子为单位回带，避免把句子切开）
overlap_buf: list[str] = []
overlap_len = 0
for prev in reversed(current_sentences):
if overlap_len >= overlap_chars:
break
overlap_buf.insert(0, prev)
overlap_len += len(prev)
current_sentences = overlap_buf
current_len = sum(len(s) for s in current_sentences)
current_sentences.append(sentence)
current_len += sent_len
if current_sentences:
content = "".join(current_sentences).strip()
if content:
chunks.append(Document(page_content=content, metadata=deepcopy(doc.metadata)))
return chunks
loader = PyMuPDFLoader(PDF_PATH)
documents = loader.load()
chunks = sentence_aware_chunk_documents(
docs=documents,
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
)
# ---------- 3. 打印分块内容 ----------
print("=== 文档分块结果（句子边界优先）===\n")
print(f" Sentence-aware splitter -> {len(chunks)} 块")
char_lens = [len(c.page_content) for c in chunks]
print(
f" 长度统计: 最小={min(char_lens)}, 最大={max(char_lens)}, 平均={sum(char_lens)//len(char_lens)}"
)
for i, chunk in enumerate(chunks, 1):
print(f"--- 第 {i} 块 (共 {len(chunks)} 块) ---")
print(chunk.page_content.strip() or "(空)")
if chunk.metadata:
print(f"[元数据] {chunk.metadata}")
print("-" * 50)
print()
返回的部分结果：
通过返回结果看，分块内的句子是完整的。这个方法与分层分块结合效果更好
父子文本分块
将文本切成子块和父块，其检索流程是，用子块向量搜索，命中子块后回溯拿到它对应的父块，把父块拼成上下文喂给LLM。
- 子块：切的更小，用来做向量检索（更容易精准命中）。
- 父块：比子块更大，用来给LLM作为更完整的上下文（避免只拿到碎片）。
代码如下：
import uuid
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
PDF_PATH = "../document/企业财务报表分析-图表.pdf"
loader = PyMuPDFLoader(PDF_PATH)
documents = loader.load()
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
parent_chunks = parent_splitter.split_documents(documents)
all_children = []
for parent in parent_chunks:
parent_id = str(uuid.uuid4())[:8]
parent.metadata["parent_id"] = parent_id
children = child_splitter.split_documents([parent])
for child in children:
child.metadata["parent_id"] = parent_id
all_children.extend(children)
# ---------- 打印父块 ----------
print(f"=== 父块（共 {len(parent_chunks)} 块，chunk_size=800）===\n")
for i, p in enumerate(parent_chunks, 1):
pid = p.metadata["parent_id"]
preview = p.page_content[:150] + "..." if len(p.page_content) > 150 else p.page_content
print(f"[父块 {i}] id={pid} 长度={len(p.page_content)}")
print(f" {preview}")
print()
# ---------- 打印子块（只展示前 3 个父块对应的子块）----------
print("=" * 80)
print(f"=== 子块（共 {len(all_children)} 块，chunk_size=200）===\n")
shown_parents = set()
for child in all_children:
pid = child.metadata["parent_id"]
if pid not in shown_parents:
shown_parents.add(pid)
if len(shown_parents) > 3:
break
print(f" ┌─ 父块 id={pid}")
siblings = [c for c in all_children if c.metadata["parent_id"] == pid]
for j, sib in enumerate(siblings, 1):
preview = sib.page_content[:100] + "..." if len(sib.page_content) > 100 else sib.page_content
print(f" │ 子块 {j}/{len(siblings)} 长度={len(sib.page_content)}")
print(f" │ {preview}")
print(f" └─ 共 {len(siblings)} 个子块")
print()
返回的部分结果：
检索时拿小块的 parent_id 回溯到父块，把父块的完整内容交给 LLM。
实现文本分块后的问答
说完分块思想，接下来让我们通过分块后的文本做个简单的RAG系统。实现流程如下：
在做RAG之前，有必要说明下嵌入模型和向量库。
嵌入模型
嵌入模型是把文本变成一组数字（向量）的模型，让计算机能“理解”文本的语义。
如人看到"营业收入增长"和"营收提升"会知道意思差不多，但计算机只认数字。嵌入模型的作用就是：
"营业收入增长" → [0.12, -0.33, 0.87, ..., 0.07] (一个 1024 维的向量)
"营收提升" → [0.11, -0.31, 0.85, ..., 0.08] (和上面很接近)
"今天天气不错" → [0.78, 0.42, -0.15, ..., 0.63] (和上面离得远)
- 语义相近->向量距离近
- 语义无关->向量距离远
嵌入模型VS大语言模型（LLM）
| 嵌入模型 | 大语言模型（LLM） | |
|---|---|---|
| 输入 | 一段文本 | 一段文本（提示/对话） |
| 输出 | 一个向量（一组数字） | 文本（回答/续写） |
| 用途 | 计算文本相似度、检索 | 理解问题、生成回答 |
| RAG 中的角色 | 负责找到相关文档片段 | 负责根据片段回答问题 |
我用的线上嵌入模型是BAAI/bge-large-zh-v1.5，支持最大512个的token输入长度。
向量数据库
专门用来存储向量，按相似度搜索向量的数据库。文本切成块之后就会被嵌入模型转成向量，存入向量数据库。
| 传统数据库 | 向量数据库 | |
|---|---|---|
| 存什么 | 行、列、文本、数字 | 向量（一组浮点数） |
| 怎么查 |
WHERE name = '张三'（精确匹配） |
"找最像这个向量的 Top-K"（相似度匹配） |
| 核心算法 | B-tree 索引 | ANN（近似最近邻）索引 |
实现代码
"""
基于 PDF 的 RAG 问答脚本：
加载 PDF → 分块 → 将分块内容作为上下文 → 使用 LLM 回答用户问题。
"""
import os
from dotenv import load_dotenv
load_dotenv()
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
# ---------- 1. 加载 PDF ----------
loader = PyMuPDFLoader("../document/企业财务报表分析-图表.pdf")
documents = loader.load()
# ---------- 2. 文档分块 ----------
# chunk_size: 每块最大字符数；chunk_overlap: 块与块之间的重叠字符数，避免语义被截断
text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)
# ---------- 3. 打印分块内容 ----------
print("=== 文档分块结果 ===\n")
for i, chunk in enumerate(chunks, 1):
print(f"--- 第 {i} 块 (共 {len(chunks)} 块) ---")
print(chunk.page_content.strip() or "(空)")
if chunk.metadata:
print(f"[元数据] {chunk.metadata}")
print("-" * 50)
print()
# ---------- 4. 配置 LLM（代理地址与 API Key 从 .env 读取） ----------
llm = ChatOpenAI(
model=os.getenv("PROXY_AI_MODEL", "gemini-2.5-flash"),
base_url=os.getenv("PROXY_AI_BASE_URL"),
api_key=os.getenv("PROXY_AI_API_KEY"),
temperature=0.3,
max_tokens=1024,
)
embeddings = OpenAIEmbeddings(
model="BAAI/bge-large-zh-v1.5",
api_key=os.getenv("SILICONFLOW_API_KEY"),
base_url="https://api.siliconflow.cn/v1",
chunk_size=32,
)
vector_store = InMemoryVectorStore.from_documents(chunks, embeddings)
# ---------- 5. 构建提示与调用链 ----------
# 系统消息中注入 PDF 上下文，用户消息为问题
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"你是一个助手。请仅根据下面「PDF 内容」回答用户问题，不要编造。回答简洁。\n\nPDF 内容：\n{context}",
),
("human", "{question}"),
]
)
chain = prompt | llm
# ---------- 6. 交互式问答 ----------
print("基于 PDF 的问答（输入空行回车退出）\n")
while True:
question = input("你的问题: ").strip()
if not question:
break
# 把问题做成向量检索
retrieved = vector_store.similarity_search(question, k=8)
context = "\n\n".join(doc.page_content for doc in retrieved)
answer = chain.invoke({"context": context, "question": question})
print(f"回答: {answer.content}\n")
返回部分结果：
回答的结果对比文档出处：
总结
- 字符分块：按一个分隔符切一次，超长也不管。
- 递归分块：多级分隔符递归切，尽量控制块大小。
- 句子边界：以句子为最小单位，不在句中截断。
- 层级分块：先按章节结构切，再对超长段做二次切。
- 滑动窗口：按固定字符数滑窗，重叠一段，块大小均匀。
- 父子分块：小块检索、大块回答，检索细、回答有上下文。
结尾
文本分块的目的，是让每块内容更聚焦、语义更完整，从而提升RAG系统的检索准确度。好了，文档分块的内容就分享到这儿。在座的彦祖、亦菲们有什么好的文档分块方法，也欢迎到评论区讨论哦！
