RAG（Retrieval-Augmented Generation，检索增强生成）是给大模型接入外部知识的标准做法。大模型虽有海量参数，却存在知识截止、回答幻觉、无法访问你私有文档三个硬伤。本文用可运行代码，从零搭一个 RAG 私域知识问答系统，让你真正理解每一步在做什么、为什么这么做。
一、为什么需要 RAG：大模型的三个硬伤
直接用大模型问答，常常遇到三类问题。一是知识截止：模型训练数据有时间点，之后的新政策、新版本、内部资料它一概不知；二是幻觉：对不确定的事实，模型会”自信地编”，给出看似通顺却错误的答案；三是数据隔离：公司的产品文档、工单、Wiki 永远不会进公开训练集。RAG 的思路很简单——先把相关资料检索出来，再让模型”带着资料答题”，把生成建立在可引用的证据之上。
有人会问：直接微调模型不行吗？微调适合”改变模型能力风格”，而 RAG 适合”补充模型不知道的事实”。前者成本高、周期长、难更新；后者只改检索语料就能即时生效，且每次回答都能溯源到原文，对合规和可信度更友好。所以绝大多数”接私有知识”的场景，RAG 都是第一选择。
想深入理解向量检索的底层原理，可以参考我们之前写的 Embeddings 向量嵌入实战 与 向量数据库实战；想用平台快速搭，可看 Dify+Ollama 私有化部署。本文则聚焦”自己写一遍”的掌控感。
二、RAG 整体架构与组件选型
一个最小可用的 RAG 系统由五段流水线组成，离线建库与在线问答各占一半：
| 阶段 | 职责 | 常见选型 |
|---|---|---|
| 文档加载 | 读取 PDF / Markdown / 网页 | LangChain Loaders、Unstructured |
| 文本切块 | 按语义切分为 chunk | 递归字符切分、按标题切分 |
| 向量化 | 文本 → 向量 | OpenAI / bge / bge-m3 嵌入模型 |
| 向量存储 | 存向量并支持检索 | FAISS、Chroma、pgvector |
| 检索 + 生成 | 取 top-k 拼 Prompt 调 LLM | GPT / DeepSeek / Qwen |
三、环境准备
我们用 OpenAI 兼容接口做嵌入与生成（可一键换成本地模型，参见 DeepSeek 本地部署），向量检索用轻量的 FAISS。
pip install openai faiss-cpu tiktoken
# 配置（本地部署时把 base_url 指向 Ollama / vLLM 即可）
export OPENAI_API_KEY="sk-xxxx"
export OPENAI_BASE_URL="https://api.openai.com/v1"四、第一步：文档加载与清洗
真实文档常有噪声（页眉页脚、多余空行）。先做一次最小清洗，避免脏数据进入索引。
import re
def load_markdown(path: str) -> str:
with open(path, "r", encoding="utf-8") as f:
text = f.read()
# 去掉代码块以外的多余空行，保留结构
text = re.sub(r"
{3,}", "
", text)
text = re.sub(r"[ ]+", " ", text)
return text.strip()
docs = [load_markdown(f) for f in ["product.md", "faq.md", "wiki.md"]]五、第二步：文本切分（Chunking）
切块直接决定检索质量：太大则噪声多、命中不精准；太小则语义被切碎。经验值中文 300–500 字、重叠 10%–20% 比较稳。
切分策略也有讲究。最朴素的是”按固定字符数递归切”（遇到段落、句子边界再断），实现简单、通用；更进阶的是”按标题/章节结构切”，能保住文档的层级语义；若文档结构松散，还可以用小模型做”语义切分”，把意思完整的一句群归为一块。对中文来说，还要注意按标点而非按空格断句——英文的空格分词在中文里完全失效，所以本文示例直接按字符数切并保留重叠，是最稳的起步方案。
def split_text(text: str, size: int = 400, overlap: int = 60):
step = size - overlap
chunks = []
for i in range(0, len(text), step):
chunk = text[i:i + size]
if len(chunk) >= 50: # 丢弃过短碎片
chunks.append(chunk)
return chunks
all_chunks = []
for d in docs:
all_chunks.extend(split_text(d)) # 全局合并成一个 chunk 池
print(f"共切出 {len(all_chunks)} 个 chunk")| 参数 | 含义 | 建议值（中文） |
|---|---|---|
| size | 单块最大字符数 | 300–500 |
| overlap | 相邻块重叠字符 | size 的 10%–20% |
| 最小长度 | 丢弃过短碎片 | ≥ 50 |
六、第三步：向量化（Embedding）
把每段文本映射成一个高维向量。同一个嵌入模型必须同时用于建库和查询，否则向量空间不一致，检索全错。
嵌入模型怎么选？英文场景 OpenAI text-embedding-3 系列性价比高；中文或需要本地化部署时，bge-large-zh、bge-m3 这类开源模型表现更好，且能跑在本地显卡上，数据不出内网。向量维度通常在 768–3072 之间，维度越高表达力越强但存储与检索开销也越大。无论选哪个，关键是”建库和查询用同一个模型、同一套预处理”，这是新手最常踩的坑。
from openai import OpenAI
client = OpenAI()
def embed(texts: list[str]) -> list[list[float]]:
resp = client.embeddings.create(
model="text-embedding-3-small",
input=texts,
)
return [d.embedding for d in resp.data]
# 分批嵌入，避免单次请求过长
batch, vectors = 16, []
for i in range(0, len(all_chunks), batch):
vectors.extend(embed(all_chunks[i:i + batch]))七、第四步：向量存储与相似检索
FAISS 用内积 / 余弦检索最近邻。建库只需把向量矩阵喂进去，查询时把问题向量丢进去取 top-k。
import numpy as np, faiss
dim = len(vectors[0])
index = faiss.IndexFlatIP(dim) # 内积 ≈ 余弦（已归一化时）
faiss.normalize_L2(np.array(vectors, dtype="float32"))
index.add(np.array(vectors, dtype="float32"))
def retrieve(query: str, k: int = 4):
q = np.array(embed([query]), dtype="float32")
faiss.normalize_L2(q)
scores, idx = index.search(q, k)
return [all_chunks[i] for i in idx[0]]
hits = retrieve("如何重置账户密码？")
print("检索到的相关片段：")
for h in hits:
print("-", h[:80])八、第五步：拼装 Prompt 并生成回答
把检索到的片段拼进系统提示，要求模型”只依据资料回答、无依据就如实说不知道”，并带上引用来源。
def ask(query: str) -> str:
ctx = "
".join(f"[资料{i+1}] {c}" for i, c in enumerate(retrieve(query)))
prompt = f'''你是企业知识助手。只依据下面的资料回答，
若资料中没有答案，请明确说「资料未覆盖」。不要编造。
{ctx}
问题：{query}'''
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
)
return resp.choices[0].message.content
print(ask("忘记密码后多久能找回账户？"))九、有无 RAG 的效果对比
| 维度 | 无 RAG（裸模型） | 有 RAG |
|---|---|---|
| 私有数据问答 | 答不出 / 瞎编 | 基于真实文档作答 |
| 时效性 | 停留在训练截止 | 随文档更新而更新 |
| 可溯源性 | 无引用 | 可标注资料来源 |
| 幻觉率 | 高 | 显著降低 |
十、进阶优化：混合检索、重排与溯源
朴素的向量检索对”精确关键词”（如报错码、型号）不敏感，可叠加 BM25 做混合检索；检索后接一个轻量重排模型（rerank）把真正相关的片段顶到前面；生成时让模型输出引用的资料编号，实现溯源。关于如何量化检索质量，可看 RAG 评估实战（RAGAS），它能量化”检索准不准、回答贴不贴”。
十一、常见坑与排障
| 现象 | 原因 | 解法 |
|---|---|---|
| 检索到的内容不相关 | chunk 太大 / 切块无重叠 | 调小 size、加大 overlap |
| 问答”答非所问” | 建库与查询用了不同嵌入模型 | 统一 model 与维度 |
| 回答仍幻觉 | Prompt 没限制”无依据就说不知道” | 收紧系统提示 + 降低 temperature |
| 长文档越界 | top-k 拼接超上下文 | 限制 token 预算 / 先重排再截断 |
十二、知识库持久化与增量更新
上面示例在内存里建索引，进程一关就没了。生产环境要把索引落盘，并支持”加新文档不重建”。FAISS 可以直接序列化，新文档只需嵌入后 index.add 增量追加；同时要另存一份 chunk 文本列表，保证索引下标与原文一一对应。如果规模再大，建议换成 Chroma、pgvector（见 pgvector 实战）这类带元数据的向量库，按文档 ID 做增删更省心。
import faiss
faiss.write_index(index, "kb.index") # 落盘
loaded = faiss.read_index("kb.index") # 重载
# 新文档：嵌入后追加，并更新 all_chunks 列表
new_vecs = np.array(embed(new_chunks), dtype="float32")
faiss.normalize_L2(new_vecs)
loaded.add(new_vecs)
all_chunks.extend(new_chunks)十三、RAG 与 Agent 的边界
RAG 解决”模型不知道的事实”，Agent 解决”模型不会主动执行的动作”。两者并不冲突：更成熟的形态是让 Agent 拥有一个”知识检索”工具，遇到需要内部资料的问题时自动调用检索、再基于结果作答。想体验这种组合，可以从 从零手写 AI Agent 入手，把本文的 retrieve 函数包装成一个工具函数即可。也别忘了用 RAGAS 评估 持续量化检索与回答质量，避免”看起来能用、实际在瞎答”。
十四、落地前检查清单
上线前对照四件事：① 嵌入模型建库与查询是否一致；② chunk 大小与重叠是否经过小样本验证；③ Prompt 是否强制”无依据就说不知道”以防幻觉；④ 是否保留了原文索引以支持溯源。把这几条守住，一个朴素 RAG 的可用性就已经超过大多数”裸模型 + 复制粘贴文档”的临时方案。
十五、小结
RAG 并不神秘：加载 → 切块 → 向量化 → 存储检索 → 拼 Prompt 生成，每一步都可独立替换与调优。把它跑通一次，你对”大模型如何用好私有知识”的理解会扎实很多。想进一步做成自主问答的智能体，可延伸阅读 从零手写 AI Agent，把检索能力嵌进 Agent 的工具箱里。
