第四篇章：RAG 知识检索增强生成——让 AI 拥有你的数据

覆盖内容：RAG 原理 → 文档加载 → 文本分块 → Embedding 向量化 → ChromaDB 向量数据库 → 检索 → 生成 → 召回评估 → 对话式 RAG → 引用溯源 → FastAPI 服务 → 实战 最终产出：完整的知识问答系统（上传文档 → 提问 → 回答 + 溯源） 前置要求：完成第二篇 API 调用入门 核心技能：大模型应用开发最值钱的技术，没有之一

一、RAG 是什么？为什么它是核心？

1.1 大模型的致命缺陷

# 问大模型一个内部问题
response = client.chat.completions.create(
model=\”deepseek-chat\”,
messages=[{

\”role\”: \”user\”, \”content\”: \”公司今年的 OKR 是什么？\”}]
)
# 答案：对不起，我不知道贵公司的内部信息。

大模型只知道训练数据里的东西。你的个人文档、公司知识库、产品手册——它一概不知。

1.2 RAG 怎么解决

RAG = Retrieval（检索） + Augmented（增强） + Generation（生成）

一句话：把相关文档片段塞进 prompt，让大模型「开卷考试」。

二、装好工具

pip install chromadb openai pypdf2 python-docx tqdm

三、文档加载——把各种格式读进 Python

3.1 加载 TXT

def load_txt(filepath):
with open(filepath, \”r\”, encoding=\”utf-8\”) as f:
return f.read()

text = load_txt(\”docs/readme.txt\”)
print(f\”加载了 {

len(text)} 个字符\”)

3.2 加载 PDF

from PyPDF2 import PdfReader

def load_pdf(filepath):
reader = PdfReader(filepath)
text = \”\”
for page in reader.pages:
text += page.extract_text() + \”\\n\”
return text

# text = load_pdf(\”docs/report.pdf\”)

3.3 加载 Word

from docx import Document

def load_docx(filepath):
doc = Document(filepath)
text = \”\\n\”.join([p.text for p in doc.paragraphs if p.text.strip()])
return text

# text = load_docx(\”docs/manual.docx\”)

3.4 统一加载器

import os

def load_document(filepath):
\”\”\”自动识别格式并加载\”\”\”
ext = os.path.splitext(filepath)[1].lower()

if ext == \”.txt\”:
with open(filepath, \”r\”, encoding=\”utf-8\”) as f:
return f.read()
elif ext == \”.pdf\”:
from PyPDF2 import PdfReader
reader = PdfReader(filepath)
return \”\\n\”.join([page.extract_text() for page in reader.pages])
elif ext in [\”.docx\”, \”.doc\”]:
from docx import Document
doc = Document(filepath)
return \”\\n\”.join([p.text for p in doc.paragraphs if p.text.strip()])
elif ext == \”.md\”:
with open(filepath, \”r\”, encoding=\”utf-8\”) as f:
return f.read()
else:
raise ValueError(f\”不支持的文件格式：{

ext}\”)

四、文本分块（Chunking）——把长文档切成小段

4.1 为什么要分块

• 向量检索是按「片段」匹配的，不是按整篇文档
• 每个 chunk 是检索的最小单位
• 太大：语义稀释，检索不准
• 太小：信息碎片，AI 看不懂上下文

4.2 手写分块函数

def chunk_text(text, chunk_size=500, chunk_overlap=100):
\”\”\”把长文本切成带重叠的片段

chunk_size: 每个片段的目标字符数
chunk_overlap: 相邻片段重叠的字符数
\”\”\”
chunks = []
start = 0
text_length = len(text)

while start < text_length:
end = start + chunk_size

# 尽量在句子边界断开（找最后一个句号/换行）
if end < text_length:
# 从 end 往前找最近的句子结束符
for sep in [\”\\n\\n\”, \”\\n\”, \”。\”, \”. \”, \”?\”, \”!\”, \”；\”]:
pos = text.rfind(sep, start, end)
if pos > start + chunk_size // 2: # 不能太短
end = pos + 1
break

chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)

start = end – chunk_overlap # 下一段从重叠位置开始

return chunks

# 测试
text = \”第一段内容很长的文字。\” * 50 + \”\\n\\n\” + \”第二段内容很长的文字。\” * 50
chunks = chunk_text(text, chunk_size=300, chunk_overlap=50)
print(f\”分成了 {

len(chunks)} 个片段\”)
for i, c in enumerate(chunks):
print(f\”片段 {

i}: {

len(c)} 字符 | 开头: {

c[:50]}…\”)

4.3 分块策略

# 策略 1：按段落分（适合格式规范的文档）
def chunk_by_paragraph(text):
paragraphs = text.split(\”\\n\\n\”)
return [p.strip() for p in paragraphs if p.strip()]

# 策略 2：按固定大小（适合纯文本）
chunks = chunk_text(text, chunk_size=500, chunk_overlap=100)

# 策略 3：按句子，且合并短句（适合对话记录）
def chunk_by_sentence(text, min_chunk_size=200):
sentences = text.replace(\”\\n\”, \” \”).split(\”。\”)
chunks = []
buffer = \”\”
for s in sentences:
buffer += s + \”。\”
if len(buffer) >= min_chunk_size:
chunks.append(buffer.strip())
buffer = \”\”
if buffer.strip():
chunks.append