从数据流动理解每一行代码
这页不只解释”怎么写”，更是把”文本怎么流到向量、chunk 怎么影响召回、Prompt 为什么这样组装”拆成可跟踪的数据流。
标签：V1 最小闭环 | V2 分块策略 | 数据流 | 实现理解
这页负责解释数据流和实现，不负责统一定义概念。 如果你还没建立 embedding / 余弦的直觉，先回到概念手册。
Page Summary
- V1 的核心：把”文本 → 向量 → 排序 → Prompt 注入”这条最小闭环看清楚。
- V2 的核心：理解 chunk_size 和 overlap 如何影响检索命中的完整性，而不是只看代码写法。
- 必须记住的结论：RAG 质量的上限先由检索决定，而检索上限先由分块质量决定。
对应代码：code/01_v1_最小RAG循环.py 与 code/02_v2_文档分块策略.py
V1 · 最小 RAG 循环
“三步走完，RAG 的本质就在这里”
涉及概念：Embedding · cosine_similarity · Top-K 检索 · Prompt 注入上下文 · 有/无 RAG 对比
STEP 01 · 文档 → Embedding 向量
把文本变成可以计算”距离”的数字
LLM 不能直接比较两段文字”意思有多像”，但计算机可以计算两个向量的夹角。Embedding 就是把文本压缩成一个高维向量的过程——语义相近的文本，向量的方向会相近。text-embedding-3-small 输出 1536 维向量，每个维度捕捉语言的某种特征。
# 知识库：5 段文本（真实场景来自 PDF/Wiki/手册）
DOCUMENTS = [
"RAG 在生成前检索外部文档，让回答有据可查，有效减少幻觉。",
"向量数据库通过余弦相似度找到语义最接近的文档块。",
"LangChain 封装了文档加载、分块、检索、生成的完整流程。",
"Transformer 的核心是注意力机制，让模型动态关注输入。",
"RAG 的瓶颈在检索而非生成：垃圾进，垃圾出。",
]
def embed(text: str) -> np.ndarray:
"""
调用 OpenAI Embedding API。
为什么用 API 而不是本地模型？
→ 快速验证概念。v4 会讨论模型选型和本地部署。
"""
response = client.embeddings.create(
input=text,
model="text-embedding-3-small" # 1536 维，中英文均可
)
return np.array(response.data[0].embedding) # shape: (1536,)
# 为所有文档建立索引
doc_embeddings = np.array([embed(doc) for doc in DOCUMENTS])
# doc_embeddings.shape → (5, 1536) [文档数 × 向量维度]预期输出（节选）
STEP 1 ／ 为知识库建立 Embedding 索引
模型: text-embedding-3-small | 文档数: 5
Doc[0] RAG 在生成前检索外部文档，让回...
维度=1536 示例(前4维)=[ 0.02341 -0.01823 0.04512 -0.00834]
Doc[1] 向量数据库通过余弦相似度找到语...
维度=1536 示例(前4维)=[ 0.01902 -0.02341 0.03819 0.01023]
...
→ 知识库矩阵形状: (5, 1536) [文档数 × 向量维度]
认知对齐： 5 段文字现在变成了一个 5×1536 的数字矩阵。每行是一篇文档的”语义指纹”。这个矩阵就是索引——真实系统中它会被存进向量数据库（v3 实现持久化）。
STEP 02 · 余弦相似度 + Top-K 检索
找到语义最相近的文档，不是字面匹配
余弦相似度公式（衡量两向量的语义距离）：
cos(θ) = (a · b) / (|a| × |b|)
值域：1.0 = 完全相同方向（语义极相似）| 0.0 = 垂直（语义无关）| -1.0 = 相反方向
RAG 实践中，得分 > 0.75 通常才算真正相关。
def cosine_sim(a: np.ndarray, b: np.ndarray) -> float:
return float(
np.dot(a, b) # 点积
/ (np.linalg.norm(a) * np.linalg.norm(b)) # 除以两向量的模
)
def retrieve(query: str, doc_embeddings: np.ndarray, top_k=2):
q_emb = embed(query) # 问题也转成向量
scores = [cosine_sim(q_emb, de) for de in doc_embeddings]
ranked = sorted(
enumerate(scores),
key=lambda x: x[1], reverse=True
)[:top_k] # 取最高的 K 个
return [{"text": DOCUMENTS[i], "score": s} for i, s in ranked]预期输出
STEP 2 ／ 语义检索
用户问题: RAG 系统里最关键的环节是什么？
全库相似度得分（热力条）：
[0] 0.8234 ████████████████████████████████
RAG 在生成前检索外部文档，让回答有据可查...
[1] 0.7103 ████████████████████████
向量数据库通过余弦相似度找到语义最接近的...
[2] 0.6281 ████████████████████
LangChain 封装了文档加载、分块、检索...
[3] 0.5410 █████████████████
Transformer 的核心是注意力机制...
[4] 0.8567 ██████████████████████████████████
RAG 的瓶颈在检索而非生成：垃圾进，垃圾出...
→ Top-2 命中文档：
✦ [0.8567] RAG 的瓶颈在检索而非生成：垃圾进，垃圾出。
✦ [0.8234] RAG 在生成前检索外部文档，让回答有据可查...
认知对齐： 注意 Doc[4]（垃圾进垃圾出）得分最高，因为它和”最关键的环节”语义最接近。Doc[3]（Transformer）得分最低。这是语义检索，不是关键词匹配——问题里没有”瓶颈”这个词，但语义一致。
STEP 03 · Prompt 注入上下文（RAG 的”增强”）
把检索结果塞进 Prompt，扩展 LLM 的视野
这里是 RAG 中 “A”（Augmented 增强）的字面含义：把检索到的真实资料拼进 Prompt，让 LLM 不需要从训练数据”回忆”，而是从当前上下文”阅读后回答”。
关键提示词设计：明确要求 LLM 只用参考资料，不要猜测。这不是银弹（LLM 仍可能忽视指令），但是控制幻觉的标准做法。
def build_prompt(query: str, contexts: list) -> str:
ctx = "\n".join(
f" [{i+1}] {c['text']}"
for i, c in enumerate(contexts)
)
return f"""你是一个严谨的知识助手。请仅根据以下参考资料回答问题。
若参考资料中没有相关信息，请明确说"资料中未提及"，不要猜测。
参考资料：
{ctx}
问题：{query}
回答："""
# 构建后的 Prompt 长这样：
# ┌──────────────────────────────────────────────┐
# │ 你是一个严谨的知识助手...（系统指令） │
# │ 参考资料： │
# │ [1] RAG 的瓶颈在检索而非生成... │ ← 检索结果注入
# │ [2] RAG 在生成前检索外部文档... │ ← 检索结果注入
# │ 问题：RAG 系统里最关键的环节是什么？ │ ← 用户原问题
# └──────────────────────────────────────────────┘STEP 04 · 对比实验：有 RAG vs 无 RAG
这是最重要的一步——看见差异
# temperature=0：让输出稳定，便于对比，不是随机创作
def chat(prompt: str) -> str:
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
)
return resp.choices[0].message.content.strip()
# 对比实验
ans_raw = chat(QUERY) # ❌ 无 RAG，直接问
ans_rag = chat(prompt_rag) # ✅ 有 RAG，带上下文预期输出对比
❌ 无 RAG（直接问 LLM）：
RAG 系统中最关键的环节因不同视角而异。从数据质量来看，
知识库的构建质量至关重要；从系统设计来看，检索策略和
生成模型的协调同样关键；从用户体验来看...
（泛泛而谈，没有明确结论）
✅ 有 RAG（基于检索内容）：
根据参考资料，RAG 系统中最关键的环节是检索而非生成。
资料明确指出"垃圾进，垃圾出"——即使生成能力再强，
如果召回了错误的文档，最终回答也无法保证质量。
认知对齐： 没有 RAG 时，LLM 用训练时学到的”泛知识”回答，倾向于面面俱到却不具体。有 RAG 后，LLM 被约束在检索到的资料里回答，答案更具体、有出处、可溯源。这就是 RAG 的核心价值。
V1 完整数据流
知识库文档 × 5
↓ embed() × 5（API 调用）
向量矩阵 [5 × 1536] ← 这是"索引"
↑
用户问题: "RAG 最关键的环节是什么？"
↓ embed()（API 调用）
query 向量 [1536]
↓ cosine_sim() × 5 → 排序
Top-2 文档 ✓
↓ build_prompt() 注入上下文
增强 Prompt（问题 + 参考资料）
↓ chat()（API 调用）
最终回答 ✓
V2 · 文档加载与分块策略
“分块决定上限，垃圾进垃圾出”
涉及概念：chunk_size · chunk_overlap · 固定大小分块 · 句子感知分块 · 边界信息保护 · 分块对比实验
V1 的知识库只有 5 句话。真实场景：一份产品手册 100 页，一个 Wiki 几千篇，一个 PDF 几万字。你不能把整篇文档塞进 Prompt（Token 限制 + 注意力稀释）。
解决方案：把长文档切成小块（Chunk），每块独立存入向量库，检索时只取相关的几块。
分块是 RAG 的第一道工程决策，直接决定检索质量的上限。
策略 A · 固定大小分块（无 overlap）
最简单，但边界信息会丢失
def chunk_fixed_no_overlap(text: str, chunk_size: int = 150) -> list[str]:
"""
按字符数切割，走到哪切到哪。
问题：可能在句子中间切断。
例："推荐分块大小在 200 到 500 个字符之间，并设置重叠"
→ 块 A：...推荐分块大小在 200 到 500
→ 块 B：个字符之间，并设置重叠...
两个块都是残缺信息。
"""
text = text.strip()
chunks, start = [], 0
while start < len(text):
chunk = text[start : start + chunk_size].strip()
if chunk: chunks.append(chunk)
start += chunk_size # 每次直接跳 chunk_size，无回退
return chunks⚠ 当 chunk_size=120，文档里”推荐分块大小在200到500个字符，设置10%~20%重叠”这句关键信息极有可能被切成两半，任何单块都拿不到完整的工程建议。
策略 B · 固定大小 + Overlap（推荐生产基线）
overlap 是解决边界问题的最简单方案
Overlap 的核心机制： 下一个块的起点 = 上一个块的终点 - overlap。即相邻两块有 overlap 个字符是重叠的。这保证了边界附近的信息一定会完整出现在某个块里。
def chunk_fixed_with_overlap(text: str, chunk_size=200, overlap=40):
text = text.strip()
chunks, start = [], 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end].strip())
next_start = end - overlap # ← 核心：回退 overlap 个字符再开始下一块
if next_start <= start: next_start = start + 1 # 防死循环
start = next_start
if start >= len(text): break
return [c for c in chunks if len(c) > 20]
# 直觉理解 overlap=40 的效果：
#
# 位置: 0 200 360 520 ...
# ├────────┤ ├────────┤
# 块 A: [0 :200]
# 块 B: [160 :360] ← 从 200-40=160 开始
# 块 C: [320 :520] ← 从 360-40=320 开始
#
# 160~200 这 40 个字符，同时出现在块 A 和块 B 里。
# 任何跨越 200 这个边界的句子，都会完整保存在某个块中。overlap 效果验证
STEP 2 ／ Overlap 的作用：保护边界信息
目标关键句：「推荐分块大小在200到500个字符之间，并设置约10%到20%的重叠区域」
策略A（120字无overlap）：关键句完整存在？ ❌ 被切断了
策略B（200字有overlap）：关键句完整存在？ ✅ 是
策略C（句子感知）： 关键句完整存在？ ✅ 是
策略A中被切断的情况：
Chunk[X]: "...分块太大（如2000个字符）：召回的块包含太多无关内容，LLM在
长上下文中丢失焦点，答案精度降低。工程实践建议：推荐分块大小在200到500"
Chunk[X+1]: "个字符之间，并设置约10%到20%的重叠区域（overlap）防止边界信息..."
→ 看到了吗？数字建议被切成两截，两块都是残缺信息。
策略 C · 句子感知分块
先分句，再合并，不在句子中间切断
def chunk_by_sentence(text: str, max_chars: int = 300) -> list[str]:
# 按中文标点分句，保留标点（lookahead 不消耗字符）
sentences = re.split(r'(?<=[。！？；\n])', text)
sentences = [s.strip() for s in sentences if s.strip() and len(s) > 5]
chunks, current = [], ""
for sent in sentences:
if len(current) + len(sent) <= max_chars:
current += sent # 累积句子，不超限就继续加
else:
if current: chunks.append(current.strip())
current = sent # 超限，保存旧块，开新块
if current.strip():
chunks.append(current.strip()) # 最后一块别忘了
return [c for c in chunks if len(c) > 20]关键实验 · 分块策略 → 检索质量 → 回答质量因果链
这是 v2 最重要的实验
问题: RAG 的分块策略有哪些工程建议？推荐用多大的分块和 overlap？
── 策略 A｜固定 120 字，无 overlap（块太小）──
[0.7821] "...推荐分块大小在200到500" ← 句子被切断了
[0.7640] "个字符之间，并设置约10%到20%..." ← 上一句的后半段
回答: 根据资料，分块大小建议在200到500...
（注意：两个残缺的块凑在一起，LLM 可能拼出答案，
但也可能混乱，且没有"overlap"建议的完整上下文）
── 策略 B｜固定 200 字 + 40 字 overlap──
[0.8234] "...推荐分块大小在200到500个字符之间，
并设置约10%到20%的重叠区域（overlap）防止边界信息丢失。"
回答: 根据参考资料，RAG 的工程建议：分块大小推荐 200~500 字符，
并设置 chunk_size 10%~20% 的 overlap。
── 策略 C｜句子感知，max=300──
[0.8401] 完整的"分块策略"段落（句子完整，语义最好）
回答: 根据资料，分块建议：①大小 200-500 字符；
②overlap 比例 10%-20%；③不同文档类型有不同最优策略。
认知对齐： 分块太小，关键信息被切断，检索命中的是残缺内容；合理分块，关键信息完整保留，回答更准确。这就是”分块决定上限”——不管你的检索算法多好、Reranker 多精准，如果 chunk 里根本没有完整的答案，LLM 就算再聪明也无法凭空生成正确答案。
三种策略对比
| 策略 | chunk_size | overlap | 语义完整性 | 块数量 | 推荐场景 |
|---|---|---|---|---|---|
| A 固定无重叠 | 任意 | 0 | 低（边界易断） | 最少 | 快速原型 |
| B 固定+overlap | 200~500 | 10%~20% | 中（推荐基线） | 略多 | 通用文档 |
| C 句子感知 | max_chars | — | 高（不切句子） | 不固定 | 技术规范/手册 |
V2 的边界在哪里 → V3 要解决什么
现在的问题（V2 的局限）
chunks 只存在内存里
↓ 程序退出 → 全部消失
每次运行都要重新计算所有 chunk 的 embedding
↓ 100 块文档 = 100 次 API 调用（慢 + 贵）
没有持久化，不支持增量更新
↓ 添加一篇新文档 = 重算所有文档（不可接受）
V3 要解决：把 chunks + embeddings 存入 ChromaDB
新文档只算新文档的 embedding → 写入 → 不影响旧数据
程序重启 → 从磁盘加载 → 毫秒级恢复
支持 metadata 过滤：按来源、日期、部门筛选文档
下一步与回查
- 回到概念：如果你还在问”为什么文本能变成向量”，回到概念手册。
- 进入评估：如果你已经能比较分块策略，下一步去工程手册 / 实验记录，把观察变成指标。
- 快速回查：文档索引 · 知识地图