第 6 章：文档解析 —— 让 PDF/Word/Markdown 进得来
6.1 三种常见格式的读取代码
# pip install pypdf python-docx
from pypdf import PdfReader
from docx import Document
from pathlib import Path
def load_pdf(path: str) -> str:
reader = PdfReader(path)
return "\\n".join(page.extract_text() or "" for page in reader.pages)
def load_docx(path: str) -> str:
doc = Document(path)
return "\\n".join(p.text for p in doc.paragraphs if p.text.strip())
def load_text(path: str) -> str: # .md / .txt 直接读
return Path(path).read_text(encoding="utf-8")
def load_any(path: str) -> str:
suffix = Path(path).suffix.lower()
loaders = {".pdf": load_pdf, ".docx": load_docx, ".md": load_text, ".txt": load_text}
if suffix not in loaders:
raise ValueError(f"暂不支持的格式: {suffix}")
return loaders[suffix](path) # 注册表模式再登场（03批12.2）
6.2 残酷现实：解析是 RAG 的隐形天花板
pypdf 这类"提取文本"的库对付规整的文字型 PDF 没问题，但真实企业文档充满地雷：
|
地雷 |
现象 |
对策 |
|
扫描件（图片型 PDF） |
提取出来是空的 |
需 OCR（如 PaddleOCR） |
|
表格 |
行列关系被打散成一锅粥 |
版面解析工具（见下），或转成 Markdown 表格再入库 |
|
双栏/复杂版式 |
阅读顺序错乱、页眉页脚混入 |
版面解析工具 |
|
图表里的信息 |
纯文字提取完全丢失 |
多模态方案（第 15 章一提） |
行业主流对策：上专业的文档版面解析方案——开源的 MinerU、PaddleOCR 的 PP-Structure 系、以及各云厂商的文档智能 API，它们能把 PDF 还原成带结构的 Markdown（标题层级、表格、公式各归各位）。本教程用 Markdown/规整文档教学以聚焦主线，但面试聊到"你们文档解析怎么做的"，能说出上面这张地雷表 + MinerU 这类方案名，就是有实战认知的信号。
第 7 章：文档切分 Chunking —— RAG 效果第一杠杆
7.1 为什么要切？切错会怎样？
不能整篇文档做一个向量：①一篇文档讲十个主题，向量成了"平均脸"，谁搜都似像非像；②检索命中后要塞进上下文，整篇太长。所以切成块（chunk）——每块最好只讲一件事。
切分的两难（面试必考的 trade-off）：
- 块太大 → 向量语义被稀释、检索不准；命中后夹带大量无关内容浪费上下文。
- 块太小 → 语义残缺（"需在30天内提交"——什么东西30天内？主语在上一块！）；模型拿到碎片拼不出完整答案。
7.2 策略一：固定长度 + 重叠（基线方案）
read_in_chunks正式转正，加上 overlap（重叠）：
def split_fixed(text: str, chunk_size: int = 400, overlap: int = 60) -> list[str]:
"""固定长度切分，相邻块重叠 overlap 个字符"""
if overlap >= chunk_size:
raise ValueError("overlap 必须小于 chunk_size")
chunks, start = [], 0
while start < len(text):
chunks.append(text[start: start + chunk_size])
start += chunk_size – overlap # 每次前进"块长-重叠"
return chunks
overlap 存在的意义：一句关键话恰好被切口斩断时，重叠区保证它在相邻块里至少完整出现一次。经验值：overlap = chunk_size 的 10%~20%。
chunk_size 经验值（中文，起点值，最终靠第 14 章评测定）：制度/FAQ 类 200400 字；叙述性长文 400600 字。换算成 token 约乘 0.6~1。
7.3 策略二：递归切分（尊重自然边界）
固定切分会把句子拦腰砍断。递归切分按分隔符优先级"先段落、再句子、最后才硬切"：
def split_recursive(text: str, chunk_size: int = 400,
seps: list[str] = ["\\n\\n", "\\n", "。", "，"]) -> list[str]:
"""按分隔符优先级递归切分：能按段切不按句切，能按句切不硬切"""
if len(text) <= chunk_size:
return [text]
if not seps: # 分隔符用尽，只能硬切
return split_fixed(text, chunk_size, overlap=0)
sep, rest = seps[0], seps[1:]
pieces = [p for p in text.split(sep) if p.strip()]
chunks, buf = [], ""
for p in pieces:
candidate = (buf + sep + p) if buf else p
if len(candidate) <= chunk_size:
buf = candidate # 还装得下，继续攒
else:
if buf:
chunks.append(buf)
# 单段自身超长 → 用更细的分隔符递归收拾它
buf = ""
if len(p) > chunk_size:
chunks.extend(split_recursive(p, chunk_size, rest))
else:
buf = p
if buf:
chunks.append(buf)
return chunks
这就是 LangChain 里大名鼎鼎的 RecursiveCharacterTextSplitter 的简化原理版
7.4 策略三：结构感知切分（Markdown 按标题切）★综合项目主力
对有标题层级的文档（Markdown/解析后的 PDF），按章节切天然语义完整，还白送元数据（标题）：
def split_by_headers(md_text: str) -> list[dict]:
"""按 Markdown 标题切分，每块携带标题作为元数据"""
sections, title, buf = [], "文档开头", []
for line in md_text.splitlines():
if line.lstrip().startswith("#"): # 遇到新标题
if buf and "".join(buf).strip():
sections.append({"title": title, "text": "\\n".join(buf).strip()})
title, buf = line.lstrip("#").strip(), []
else:
buf.append(line)
if buf and "".join(buf).strip():
sections.append({"title": title, "text": "\\n".join(buf).strip()})
return sections
生产组合拳：先按标题切大节 → 超长的节内部再递归切 → 每个小块的元数据记下它所属的标题路径。检索时标题还能拼进块文本一起向量化（"考勤制度 > 迟到处理：……"），显著提升召回——这个技巧综合项目直接用。
7.5 策略四速览：语义切分与"一页纸决策表"
语义切分：逐句算 Embedding，相邻句相似度骤降处（话题切换点）下刀。效果好但建库成本高（每句都要向量化），大规模场景酌情使用——知道原理即可。
|
文档类型 |
首选策略 |
|
Markdown/有标题结构 |
标题切分 + 节内递归（★默认答案） |
|
无结构长文 |
递归切分 |
|
FAQ/条款（天然一条一意） |
按条切，一条一块 |
|
代码 |
按函数/类切（语法感知） |
|
什么都不确定时 |
递归切分 400 字 + 15% overlap 起步，评测调优 |
第 8 章：向量数据库入门 —— Chroma
8.1 为什么需要向量数据库
第 4 章的"for 循环逐条算相似度"是暴力检索：5 条文档没问题，500 万条时每次提问算 500 万次余弦=灾难。向量数据库解决三件事：①近似最近邻索引（ANN，如 HNSW 算法）把检索从"遍历"变"跳查"，百万级毫秒响应；②向量+原文+元数据一起持久化管理；③增删改查、过滤、多租户等数据库该有的一切。
HNSW 一句话（面试够用）：把向量组织成多层"高速公路网"，查询时从高层粗跳到低层精找，以极小的精度损失换取数量级的速度提升——"近似"二字的由来。
8.2 Chroma 五分钟上手（学习与原型首选）
# pip install chromadb
import chromadb
chroma = chromadb.PersistentClient(path="./chroma_db") # 数据落盘到本目录
col = chroma.get_or_create_collection(
name="handbook",
metadata={"hnsw:space": "cosine"}, # 指定用余弦距离（默认是L2，中文检索务必改）
)
chunks = [
{"id": "c1", "text": "员工每年享有5天带薪年假，入职满3年增加至10天。", "title": "休假制度"},
{"id": "c2", "text": "报销需在费用发生后30天内提交，超期不予受理。", "title": "报销制度"},
{"id": "c3", "text": "试用期为3个月，考核合格后转正。", "title": "入职转正"},
]
col.add( # 入库：四个平行列表，一一对应
ids=[c["id"] for c in chunks],
documents=[c["text"] for c in chunks],
embeddings=embed([c["text"] for c in chunks]), # 用第4章的embed函数
metadatas=[{"title": c["title"]} for c in chunks],
)
res = col.query(
query_embeddings=embed(["休假有什么规定"]),
n_results=2,
)
print(res["documents"][0]) # 命中的文本列表（注意是列表套列表：外层对应多个查询）
print(res["distances"][0]) # 距离列表：cosine空间下 距离=1-相似度，越小越相似！
print(res["metadatas"][0]) # 元数据跟着回来 —— 引用溯源的原料
三个必踩坑提前排雷：①返回结构是"列表套列表"（支持一次多查询），取结果先 [0]；②cosine 空间下 Chroma 返回的是距离（1−相似度），别把 0.3 当成"不相似"——它其实=相似度 0.7；③自己传 embeddings（如上）而不是让 Chroma 用内置默认模型——默认模型偏英文，中文效果差，且自己传才能保证"查询和文档用同一个模型"（第 5 章细节 2）。
8.3 增删改与"重建库"
col.update(ids=["c1"], documents=["新版：年假改为10天起步。"],
embeddings=embed(["新版：年假改为10天起步。"])) # 改内容必须重算向量！
col.delete(ids=["c3"])
print(col.count())
工程习惯：文档源变更后，简单粗暴但可靠的方案是按文档为单位删旧增新（元数据里记 source 文件名，col.delete(where={"source": "员工手册.md"}) 后重灌）——增量同步的复杂度综合项目里体会。
第 9 章：FAISS 与 Milvus —— 从单机到大厂主流
9.1 三兄弟怎么分工（选型必考）
|
|
Chroma |
FAISS |
Milvus |
|
定位 |
轻量向量数据库 |
Meta 开源的向量索引库（不是数据库：无持久化管理/无过滤器/纯算法引擎） |
分布式向量数据库（Zilliz 出品，国内大厂事实主流） |
|
规模 |
单机百万级 |
单机千万级、算法性能天花板 |
集群十亿级、高可用 |
|
场景 |
学习/原型/小项目 |
嵌进自己代码做极致性能检索、学术 |
企业生产（JD 里点名率最高） |
其它常见选项一句话备着：Elasticsearch（8.x 起支持向量，老 ES 团队顺手升级的选择，天然强于混合检索）、pgvector（PostgreSQL 插件，"不想多养一个数据库"的团队最爱）、Qdrant/Weaviate（海外流行）、腾讯云 VectorDB 等云托管。
9.2 FAISS 十行体验（理解"索引库"是什么）
# pip install faiss-cpu numpy
import faiss, numpy as np
vecs = np.array(embed([c["text"] for c in chunks]), dtype="float32")
faiss.normalize_L2(vecs) # 归一化后，内积 == 余弦相似度
index = faiss.IndexFlatIP(vecs.shape[1]) # IP=内积；Flat=暴力精确检索
index.add(vecs)
q = np.array(embed(["休假规定"]), dtype="float32")
faiss.normalize_L2(q)
scores, ids = index.search(q, k=2)
print(scores, ids) # 相似度 与 向量的序号 —— 原文要靠你自己按序号对回去
看到了吗——FAISS 只管向量进向量出，原文、元数据、持久化全要你自己管。这就是"库(library)"与"数据库(database)"的区别，面试一句话讲清。
9.3 Milvus：大厂里的真主角（Lite 版本地练）
Milvus 完整版是分布式集群；练手用官方 Milvus Lite——pip 装完一个文件就是库：
# pip install pymilvus
from pymilvus import MilvusClient
mc = MilvusClient("milvus_demo.db") # 本地文件即数据库（Lite模式）
mc.create_collection(collection_name="handbook", dimension=1024,
metric_type="COSINE")
mc.insert(collection_name="handbook", data=[
{"id": i, "vector": v, "text": c["text"], "title": c["title"]}
for i, (v, c) in enumerate(zip(embed([c["text"] for c in chunks]), chunks))
])
res = mc.search(collection_name="handbook",
data=embed(["休假规定"]), limit=2,
filter='title == "休假制度"', # 标量过滤（下一章主角）
output_fields=["text", "title"])
for hit in res[0]:
print(hit["distance"], hit["entity"]["text"])
概念对齐（Milvus 黑话 → 人话）：Collection=表；Schema=字段定义（生产上会显式定义而非用上面的快捷模式）；Index=为向量字段建的 HNSW/IVF 索引；标量过滤=按普通字段筛。面试聊到 Milvus，能说出"生产上我们会显式定义 Schema、为向量字段建 HNSW 索引、用分区（Partition）做多租户隔离"，就是用过的样子。
