系列导读：本系列共 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 维

def embed(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)
    return vec

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

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

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

class VectorStore:
    def __init__(self):
        self.texts = []    # 原始文本列表
        self.vectors = []  # 对应的向量列表

    def add(self, text: str):
        self.texts.append(text)
        self.vectors.append(embed(text))

    def search(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])) for i in top_indices]

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

$$\text{cosine\_similarity}(A,B)=\frac{A\cdot B}{|A||B|}$$

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

组件 3：生成（Generator）

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

def generate(query: str, contexts: list) -> str:
    context_text = "\n".join(f"- {c}" for c in contexts)

    # 这就是发给大模型的完整 Prompt，结构非常重要
    prompt = f"""你是一个企业内部知识库助手。请根据以下资料回答用户问题。
如果资料中没有相关信息，请说"知识库中没有找到相关信息"。

【参考资料】
{context_text}

【用户问题】
{query}

【回答】"""
    # 本文模拟返回，第 2 篇替换为真实 Ollama 调用
    return f"[模拟回答] 根据知识库，关于「{query}」：{contexts[0]}"

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

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

---

四、完整代码

import numpy as np

DOCUMENTS = [
    "年假政策：员工入职满一年后享有15天年假，满三年后20天。",
    "病假政策：病假需提供医院证明，当天请假需电话通知直属上级。",
    "请假流程：登录OA系统，填写请假申请单，提前3个工作日提交。",
    "加班规定：工作日加班按1.5倍计算，周末加班按2倍计算。",
    "报销流程：填写费用报销单，附发票原件，提交财务部审核。",
    "入职流程：携带身份证、学历证明、离职证明，到HR办理手续。",
    "绩效考核：每季度考核一次，分A/B/C/D四个等级，影响年终奖。",
    "办公时间：上午9:00-12:00，下午13:30-18:00，弹性工作时间±1小时。",
]

EMBED_DIM = 16

def embed(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)
    return vec

class VectorStore:
    def __init__(self):
        self.texts = []
        self.vectors = []

    def add(self, text: str):
        self.texts.append(text)
        self.vectors.append(embed(text))

    def search(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])) for i in top_indices]

def generate(query: str, contexts: list) -> str:
    context_text = "\n".join(f"- {c}" for c in contexts)
    prompt = f"""你是一个企业内部知识库助手。请根据以下资料回答用户问题。
如果资料中没有相关信息，请说"知识库中没有找到相关信息"。

【参考资料】
{context_text}

【用户问题】
{query}

【回答】"""
    print(f"\n{'='*50}\n{prompt}\n{'='*50}")
    return f"[模拟回答] {contexts[0]}"

class SimpleRAG:
    def __init__(self):
        self.store = VectorStore()

    def build_index(self, documents: list):
        print(f"📚 建立索引，共 {len(documents)} 个文档片段...")
        for doc in documents:
            self.store.add(doc)
        print("✅ 索引建立完成\n")

    def query(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} 个相关片段：")
        for i, (text, score) in enumerate(results, 1):
            print(f"  [{i}] 相似度={score:.3f}  {text}")
        contexts = [text for text, _ in results]
        answer = generate(question, contexts)
        print(f"\n💬 最终回答：{answer}\n")
        return answer

if __name__ == "__main__":
    rag = SimpleRAG()
    rag.build_index(DOCUMENTS)
    rag.query("我想请假，怎么申请？")
    rag.query("年假有多少天？")

运行：

python3 step1_minimal_rag.py

---

五、运行结果分析

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

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

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

---

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

注意代码的分层设计：

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

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

---

七、面试常问

Q: RAG 和 Fine-tuning 的区别？

	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 本地模型，让检索结果真正有语义。