引言
假设你的公司有一万份内部文档——产品手册、技术规范、会议纪要、项目报告。你想做一个 AI 助手，能回答关于这些文档的问题。
最直觉的想法：把所有文档塞进上下文窗口。但一万份文档可能有几千万 token，远远超出任何模型的上下文窗口。
RAG（Retrieval-Augmented Generation，检索增强生成） 给出了一个优雅的解决方案：不把所有文档都给模型，而是先找到最相关的几份，只把这些放进上下文。
就像开卷考试——你不需要把整本教材背下来，只需要知道去哪页找答案。
RAG 的核心流程
RAG = Retrieval + Augmented + Generation
┌──────────────┐
│ 用户提问 │
└──────┬───────┘
│
▼
┌──────────────┐
│ 检索（R） │ ← 从知识库中找到最相关的文档片段
│ Retrieval │
└──────┬───────┘
│
▼
┌──────────────┐
│ 增强（A） │ ← 把检索到的内容放入 prompt
│ Augmented │
└──────┬───────┘
│
▼
┌──────────────┐
│ 生成（G） │ ← 模型基于检索内容生成回答
│ Generation │
└──────┬───────┘
│
▼
┌──────────────┐
│ 最终回答 │
└──────────────┘第一步：Embedding——把文本变成向量
什么是 Embedding？
Embedding：把文本映射为高维空间中的向量（一组数字）
"人工智能改变世界" → [0.123, -0.456, 0.789, ..., 0.234]
通常是 768 或 1536 维
关键特性：语义相近的文本 → 向量也相近
"人工智能" → [0.12, -0.45, 0.79, ...]
"AI" → [0.11, -0.44, 0.80, ...] ← 很接近
"苹果" → [0.78, 0.23, -0.56, ...] ← 很远Embedding 代码示例
python
from openai import OpenAI
client = OpenAI()
def get_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
# 语义相近的文本，向量距离更近
v1 = get_embedding("机器学习")
v2 = get_embedding("深度学习")
v3 = get_embedding("红烧肉")
import numpy as np
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
print(f"机器学习 vs 深度学习: {cosine_similarity(v1, v2):.3f}") # ~0.85
print(f"机器学习 vs 红烧肉: {cosine_similarity(v1, v3):.3f}") # ~0.20第二步：向量数据库
为什么需要专门的数据库？
传统数据库：精确匹配
SELECT * FROM docs WHERE title = "AI 入门"
→ 只能找到标题完全匹配的文档
向量数据库：语义搜索
查询 "人工智能入门"
→ 能找到标题为 "AI 导论"、"机器学习基础" 等语义相近的文档主流向量数据库对比
| 数据库 | 类型 | 特点 |
|---|---|---|
| Chroma | 嵌入式 | 最简单，适合原型和小项目 |
| FAISS | 库 | Meta 开源，性能极高，不支持持久化 |
| Pinecone | 云服务 | 全托管，无需运维 |
| Milvus | 分布式 | 企业级，支持大规模部署 |
| Qdrant | 独立服务 | Rust 实现，性能好 |
| pgvector | PostgreSQL 扩展 | 和现有 PG 数据库一起用 |
Chroma 快速上手
python
import chromadb
# 初始化客户端
client = chromadb.PersistentClient(path="./chroma_db")
# 创建集合
collection = client.get_or_create_collection(
name="company_docs",
metadata={"hnsw:space": "cosine"}
)
# 添加文档
docs = [
{"id": "doc1", "text": "公司年假政策：入职满一年可享受 10 天年假。", "metadata": {"source": "hr handbook"}},
{"id": "doc2", "text": "报销流程：填写报销单 → 直属领导审批 → 财务审核。", "metadata": {"source": "finance"}},
{"id": "doc3", "text": "请假需提前三天通过 OA 系统提交申请。", "metadata": {"source": "hr handbook"}},
]
collection.add(
ids=[d["id"] for d in docs],
documents=[d["text"] for d in docs],
metadatas=[d["metadata"] for d in docs]
)
# 查询
results = collection.query(
query_texts=["我想请假，怎么操作？"],
n_results=2 # 返回最相关的 2 条
)
for doc, dist in zip(results["documents"][0], results["distances"][0]):
print(f"[距离: {dist:.3f}] {doc}")第三步：Chunking——文档切分
为什么需要切分？
问题：
一份文档可能有 10000 个 token
你不可能把整篇文档放进上下文（浪费 token）
也不可能把每个字都单独检索（丢失上下文）
解决方案：把文档切成适当大小的片段（chunk）
完整文档：
┌──────────────────────────────────────────┐
│ 第一章：概述...（2000 token） │
│ 第二章：技术方案...（3000 token） │
│ 第三章：实施计划...（2500 token） │
│ 第四章：预算...（1500 token） │
└──────────────────────────────────────────┘
切分后：
┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐
│ Chunk 1 │ │ Chunk 2 │ │ Chunk 3 │ │ Chunk 4 │
│ 第一章概述 │ │ 第二章技术 │ │ 第三章实施 │ │ 第四章预算 │
│ ~500 token │ │ ~500 token │ │ ~500 token │ │ ~500 token │
└────────────┘ └────────────┘ └────────────┘ └────────────┘
检索时只返回最相关的 2-3 个 chunk切分策略
python
# 策略 1：固定大小切分
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每个 chunk 约 500 字符
chunk_overlap=50, # 相邻 chunk 重叠 50 字符
separators=["\n\n", "\n", "。", "！", "？", " "] # 优先在段落/句子边界切
)
chunks = splitter.split_text(long_document)| 策略 | 方法 | 优点 | 缺点 |
|---|---|---|---|
| 固定大小 | 按 token/字符数切 | 简单 | 可能切断语义 |
| 递归切分 | 按分隔符层级切 | 保留段落/句子结构 | 长短不一 |
| 语义切分 | 按 embedding 相似度切 | 语义完整 | 计算成本高 |
| 文档结构 | 按标题/章节切 | 最符合文档逻辑 | 需要解析文档结构 |
第四步：完整 RAG 管线
python
from openai import OpenAI
import chromadb
import json
openai_client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./company_docs_db")
collection = chroma_client.get_or_create_collection("docs")
def rag_query(question: str, n_results: int = 3) -> str:
"""完整的 RAG 查询流程"""
# Step 1: 检索（Retrieval）
results = collection.query(
query_texts=[question],
n_results=n_results
)
# Step 2: 增强（Augmented）— 构造增强 prompt
context = "\n\n".join(results["documents"][0])
messages = [
{"role": "system", "content": f"""你是公司内部知识库助手。
基于以下参考文档回答用户问题。
如果文档中没有相关信息，说「根据现有文档无法回答」。
不要编造信息。
参考文档：
{context}"""},
{"role": "user", "content": question}
]
# Step 3: 生成（Generation）
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0 # RAG 场景用低温度，减少编造
)
return response.choices[0].message.content
# 使用
answer = rag_query("公司年假怎么算？")
print(answer)
# → "根据文档，入职满一年可享受 10 天年假。"RAG 的 Prompt 设计
python
RAG_SYSTEM_PROMPT = """你是公司内部知识库助手。
规则：
1. 只基于提供的参考文档回答
2. 如果文档中没有相关信息，说「现有文档中没有相关信息」
3. 引用具体的文档来源
4. 不要使用模型自身知识补充回答
5. 如果多个文档有冲突信息，全部列出并标注来源
参考文档：
{context}"""本节小结
| 概念 | 要点 |
|---|---|
| RAG 定义 | Retrieval-Augmented Generation，检索增强生成 |
| Embedding | 把文本映射为向量，语义相近的文本向量也相近 |
| 向量数据库 | 存储和搜索 embedding，实现语义检索 |
| Chunking | 文档切分：固定大小、递归、语义、结构化 |
| RAG 管线 | 查询 → Embedding → 向量搜索 → 构造增强 prompt → 生成 |
| Prompt 设计 | 明确指示模型只基于文档回答，不要编造 |
思考题
- 为什么 RAG 能减少幻觉？它的信息来源和纯模型回答有什么区别？
- 如果两个 chunk 的内容互相矛盾（比如两份文档有不同的政策），模型应该怎么处理？
- Chunk 的大小会影响 RAG 的效果。太大的 chunk 和太小的 chunk 分别有什么问题？