（0）从零手写 RAG：不依赖任何框架，彻底搞懂检索增强生成原理

系列导读：本系列共 6 篇，带你从零到一构建一个完整的 RAG + LangGraph + MCP 项目。每一步都是可独立运行的完整程序，理解原理后再进下一步。

• 第 1 篇（本文）：最小 RAG 实现，纯 numpy，无任何 AI 框架
• 第 2 篇：接入 Ollama 本地大模型
• 第 3 篇：接入 ChromaDB 持久化向量数据库
• 第 4 篇：用 LangChain 重构 + 多轮对话
• 第 5 篇：LangGraph 多步推理工作流
• 第 6 篇：MCP 工具调用协议集成

一、RAG 是什么，为什么需要它

大模型有两个核心问题：

1. 知识截止：训练数据有截止日期，不知道最新信息
2. 私域盲区：不知道你公司的内部文档、业务数据

RAG（Retrieval-Augmented Generation，检索增强生成）的解决思路很直接：

问大模型之前，先去知识库里找相关资料，一起塞给大模型参考。

用户问题 ↓ 【检索】去知识库找最相关的几段文字 ↓ 【增强】把问题 + 找到的资料拼在一起 ↓ 【生成】大模型基于这些资料回答

本文用最少的代码（只依赖numpy）实现完整的 RAG 流程，把每一行的原理讲清楚。

二、RAG 的两个阶段

阶段一：离线建库（只做一次）

原始文档 → 切片 → Embedding（向量化）→ 存入向量数据库

阶段二：在线查询（每次问答）

用户问题 → Embedding → 向量相似度检索 → 拼 Prompt → 大模型生成回答

三、核心组件详解

组件 1：Embedding（向量化）

把文字变成一串数字（向量），语义相似的文字，向量距离近。

EMBED_DIM=16# 真实模型一般是 768 或 1536 维defembed(text:str)->np.ndarray:# 用文本的 hash 作为随机种子，保证同一文本每次结果相同seed=hash(text)%(2**31)rng=np.random.RandomState(seed)vec=rng.randn(EMBED_DIM).astype(np.float32)# 归一化，方便后续用点积代替余弦相似度vec=vec/np.linalg.norm(vec)returnvec

注意：本文用随机向量模拟，不具备真实语义，但流程完全一致。第 2 篇会换成真实的 Ollama embedding 模型。

组件 2：向量数据库（内存版）

存储所有文档的向量，支持相似度检索：

classVectorStore:def__init__(self):self.texts=[]# 原始文本列表self.vectors=[]# 对应的向量列表defadd(self,text:str):self.texts.append(text)self.vectors.append(embed(text))defsearch(self,query:str,top_k:int=3)->list:query_vec=embed(query)matrix=np.array(self.vectors)# shape: (N, EMBED_DIM)# 余弦相似度 = 两个归一化向量的点积，值域 [-1, 1]，越大越相似scores=matrix @ query_vec# shape: (N,)# 取 top_k 个最高分的下标top_indices=np.argsort(scores)[::-1][:top_k]return[(self.texts[i],float(scores[i]))foriintop_indices]

余弦相似度是 RAG 检索的核心数学工具：

cosine_similarity(A,B)=A⋅B∣A∣∣B∣\text{cosine\_similarity}(A, B) = \frac{A \cdot B}{|A||B|}cosine_similarity(A,B)=∣A∣∣B∣A⋅B​

向量归一化后（vec / norm(vec)），点积就等于余弦相似度，所以matrix @ query_vec一行搞定。

组件 3：生成（Generator）

把问题和检索到的资料拼成 Prompt，发给大模型：

defgenerate(query:str,contexts:list)->str:context_text="\n".join(f"-{c}"forcincontexts)# 这就是发给大模型的完整 Prompt，结构非常重要prompt=f"""你是一个企业内部知识库助手。请根据以下资料回答用户问题。 如果资料中没有相关信息，请说"知识库中没有找到相关信息"。 【参考资料】{context_text}【用户问题】{query}【回答】"""# 本文模拟返回，第 2 篇替换为真实 Ollama 调用returnf"[模拟回答] 根据知识库，关于「{query}」：{contexts[0]}"

Prompt 结构是 RAG 效果的关键，标准三段式：

1. 角色设定（告诉模型它是什么）
2. 参考资料（检索到的内容）
3. 用户问题

四、完整代码

importnumpyasnp DOCUMENTS=["年假政策：员工入职满一年后享有15天年假，满三年后20天。","病假政策：病假需提供医院证明，当天请假需电话通知直属上级。","请假流程：登录OA系统，填写请假申请单，提前3个工作日提交。","加班规定：工作日加班按1.5倍计算，周末加班按2倍计算。","报销流程：填写费用报销单，附发票原件，提交财务部审核。","入职流程：携带身份证、学历证明、离职证明，到HR办理手续。","绩效考核：每季度考核一次，分A/B/C/D四个等级，影响年终奖。","办公时间：上午9:00-12:00，下午13:30-18:00，弹性工作时间±1小时。",]EMBED_DIM=16defembed(text:str)->np.ndarray:seed=hash(text)%(2**31)rng=np.random.RandomState(seed)vec=rng.randn(EMBED_DIM).astype(np.float32)vec=vec/np.linalg.norm(vec)returnvecclassVectorStore:def__init__(self):self.texts=[]self.vectors=[]defadd(self,text:str):self.texts.append(text)self.vectors.append(embed(text))defsearch(self,query:str,top_k:int=3)->list:query_vec=embed(query)matrix=np.array(self.vectors)scores=matrix @ query_vec top_indices=np.argsort(scores)[::-1][:top_k]return[(self.texts[i],float(scores[i]))foriintop_indices]defgenerate(query:str,contexts:list)->str:context_text="\n".join(f"-{c}"forcincontexts)prompt=f"""你是一个企业内部知识库助手。请根据以下资料回答用户问题。 如果资料中没有相关信息，请说"知识库中没有找到相关信息"。 【参考资料】{context_text}【用户问题】{query}【回答】"""print(f"\n{'='*50}\n{prompt}\n{'='*50}")returnf"[模拟回答]{contexts[0]}"classSimpleRAG:def__init__(self):self.store=VectorStore()defbuild_index(self,documents:list):print(f"📚 建立索引，共{len(documents)}个文档片段...")fordocindocuments:self.store.add(doc)print("✅ 索引建立完成\n")defquery(self,question:str,top_k:int=2)->str:print(f"\n🔍 问题：{question}")results=self.store.search(question,top_k=top_k)print(f"\n📋 检索到{top_k}个相关片段：")fori,(text,score)inenumerate(results,1):print(f" [{i}] 相似度={score:.3f}{text}")contexts=[textfortext,_inresults]answer=generate(question,contexts)print(f"\n💬 最终回答：{answer}\n")returnanswerif__name__=="__main__":rag=SimpleRAG()rag.build_index(DOCUMENTS)rag.query("我想请假，怎么申请？")rag.query("年假有多少天？")

运行：

python3 step1_minimal_rag.py

五、运行结果分析

你会发现检索结果是错误的——“年假有多少天"没有检索到"年假政策”，而是检索到了不相关的内容。

这是正常的！因为随机向量没有语义，无法判断"年假"和"年假政策"相似。

这就是第 2 篇要解决的问题：换成真实的 Embedding 模型（Ollamanomic-embed-text）。

六、设计亮点：只需替换一个函数

注意代码的分层设计：

embed() ← 第 2 篇替换这里（随机向量 → 语义向量） VectorStore ← 第 3 篇替换这里（内存 → ChromaDB） generate() ← 第 2 篇替换这里（模板 → Ollama） SimpleRAG ← 第 4 篇替换这里（手写 → LangChain）

每步只替换一个组件，其余代码完全不变。这是学习新技术的最好方式：控制变量，每次只改一件事。

七、面试常问

Q: RAG 和 Fine-tuning 的区别？

Q: Chunk 大小怎么选？

• 太小：语义不完整，缺少上下文
• 太大：噪声多，关键信息被稀释
• 经验值：512~1024 token，相邻 chunk 重叠 10%~20%

Q: 为什么用余弦相似度而不是欧氏距离？

• 余弦相似度只关注方向（语义），不受向量长度影响
• 短文本和长文本的向量长度差异大，欧氏距离会偏向长文本

总结

本文用~80 行纯 Python实现了完整的 RAG 流程，核心只有三个组件：

1. embed()：文本 → 向量
2. VectorStore：存储 + 检索向量
3. generate()：Prompt 构造 + 大模型调用

下一篇将把embed()和generate()替换为真实的 Ollama 本地模型，让检索结果真正有语义。