上篇文章《从零学大模型》我们聊了入门路线图,其中反复提到一个观点:RAG 解决 90% 的问题,微调只解决剩下的 10%。今天就来兑现这个承诺——从零搭一个能用的 RAG 系统,全程不超过 200 行 Python。
RAG 到底是什么?一句话说清楚
大模型的知识截止于训练数据那一天。你问它”我们公司最新的产品价格”,它答不上来,因为它的训练数据里没有。RAG 的思路很简单:
- 你把公司文档切碎、转成向量存起来
- 用户提问时,先从文档库里搜出相关内容
- 把搜到的内容 + 用户问题一起丢给大模型
- 大模型基于搜到的内容回答——相当于让它”开卷考试”
整个过程不需要训练模型,不需要显卡,一个普通服务器甚至笔记本就能跑。
环境准备
你需要:
- Python 3.10+
- 一个 LLM API key(OpenAI、DeepSeek 都行)
- 一个文本嵌入模型(我们这里用免费开源的,不用花钱)
安装依赖:
pip install sentence-transformers chromadb langchain-core
三个库各自分工:
sentence-transformers→ 把文本变成向量chromadb→ 向量数据库,存和搜langchain-core→ 组装 RAG pipeline 的工具
第一步:准备文档
假设你有一堆 Markdown 格式的文档(产品手册、API 文档、FAQ 等),先按段落切分。太长会超模型上下文窗口,太短会丢失语义:
import os
def load_documents(folder_path):
"""读取文件夹下所有 .md 文件,按段落切分"""
docs = []
for fname in os.listdir(folder_path):
if not fname.endswith('.md'):
continue
with open(os.path.join(folder_path, fname), 'r') as f:
content = f.read()
paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]
for i, para in enumerate(paragraphs):
docs.append({
'id': f'{fname}#p{i}',
'text': para,
'source': fname
})
return docs
切分时有个经验值:每段 200-500 字效果最好。太短搜出来缺上下文,太长又容易混入无关信息。如果你的文档本来就是短条目(比如 FAQ 一问一答),可以按条目切。
第二步:转向量并存入数据库
用 sentence-transformers 里的 BAAI/bge-base-zh-v1.5 这个中文嵌入模型——它专门针对中文优化,而且完全免费:
from sentence_transformers import SentenceTransformer
import chromadb
model = SentenceTransformer('BAAI/bge-base-zh-v1.5')
client = chromadb.PersistentClient(path='./kb_db')
collection = client.get_or_create_collection(name='knowledge_base')
def index_documents(docs):
"""将文档批量转向量并存入 ChromaDB"""
texts = [d['text'] for d in docs]
ids = [d['id'] for d in docs]
metadatas = [{'source': d['source']} for d in docs]
embeddings = model.encode(texts, show_progress_bar=True)
collection.add(
documents=texts,
embeddings=embeddings.tolist(),
ids=ids,
metadatas=metadatas
)
print(f'✅ 已索引 {len(docs)} 个段落')
PersistentClient 会把数据存到你指定的目录,下次启动不需要重新索引。bge-base-zh-v1.5 生成的向量是 768 维,占用不大——一万段文档大约占 50MB 磁盘。
第三步:检索
用户提问时,先把问题转成向量,然后在数据库里搜相似度最高的段落:
def search(query, top_k=5):
"""搜索与 query 最相似的 k 个段落"""
query_vec = model.encode([query])
results = collection.query(
query_embeddings=query_vec.tolist(),
n_results=top_k
)
return results['documents'][0]
这里返回的是文本片段。注意 n_results 不要设太大——给大模型塞太多上下文反而会稀释注意力,3-5 段是最佳实践。
第四步:组装回答
把搜到的内容和用户问题拼到一起,交给大模型:
import requests
LLM_API_KEY = '***' # 替换为你的 API key
LLM_URL = 'https://api.deepseek.com/chat/completions'
def ask_with_context(query, context_chunks):
"""基于检索到的上下文,让大模型回答问题"""
context = '\n\n---\n\n'.join(context_chunks)
messages = [
{
'role': 'system',
'content': '你是一个知识库助手。基于以下文档内容回答用户问题。'
'如果文档中没有相关信息,直接说"文档中没有找到相关内容",不要编造。'
'引用时标注来源文件名。'
},
{
'role': 'user',
'content': f'### 文档内容\n\n{context}\n\n### 问题\n\n{query}'
}
]
resp = requests.post(LLM_URL, json={
'model': 'deepseek-chat',
'messages': messages,
'temperature': 0.3
}, headers={'Authorization': f'Bearer {LLM_API_KEY}'})
return resp.json()['choices'][0]['message']['content']
temperature=0.3 让模型尽量严谨,不要自由发挥。如果你的场景需要创造性(比如写文案),可以调到 0.7。
第五步:串起来
def rag(query):
"""完整 RAG 流程:检索 + 生成"""
chunks = search(query, top_k=5)
answer = ask_with_context(query, chunks)
return answer
# 跑起来试试
print(rag('产品的退款政策是什么?'))
全流程不到 80 行代码,但已经是一个生产可用的 RAG 系统骨架。把它包装成 Flask/FastAPI 接口,就是一个完整的知识库问答 API。
进阶优化
上面的基础版能用,但要上线还有几个可以提升的点:
| 优化项 | 怎么做 | 效果 |
|---|---|---|
| 🔍 混合检索 | 向量检索 + BM25 关键词检索取交集 | 专有名词搜索更准(如产品型号 “ABC-123″) |
| 📐 重排序 | 用 Cross-Encoder 模型对初筛结果重新打分 | 召回 Top 20 → 重排取 Top 3,精度提升 10-20% |
| 🧩 文档解析 | 用 Unstructured.io 或 MarkItDown 解析 PDF/Word | 不再只限 Markdown,什么格式都能吃 |
| 🔄 增量更新 | ChromaDB 支持按 ID 增删改 | 文档更新了不用全量重建 |
| ⚡ 缓存热点 | 高频问题结果缓存到 Redis | 命中缓存时零延迟 |
什么时候别用 RAG?
RAG 不是万能的。如果你遇到以下情况,可能需要考虑微调:
- ✅ 模型需要学会某种固定输出格式(比如必须输出 JSON+签名)
- ✅ 你要让模型模仿某种风格(比如模仿某个作家的口吻)
- ❌ 你的知识库很大(几万篇以上),但命中率要求极高
至于怎么微调,那是下篇文章的内容。
下期预告
下一篇我们聊 LoRA 微调实战——用 LLaMA-Factory 在自己的数据上微调模型,让模型学会你想要的”说话方式”。不需要 4 块 3090,一张消费级显卡就能跑。
参考来源:
- Sentence-Transformers 官方文档
- ChromaDB 官方文档
- LangChain RAG 教程
- BAAI/bge-base-zh-v1.5 – Hugging Face
- DeepSeek API 文档
转载请注明:Falost的小窝 » 手写 RAG:200 行代码搭一个本地知识库问答系统


