RAG学习-从200 行代码最佳实践看企业级RAG实现

手写一个 200 行 RAG：从零理解检索增强生成

RAG（Retrieval-Augmented Generation）概念简单，但实现细节容易被各种框架遮蔽。这篇文章逐行拆解一个最小 RAG 实现，只依赖 Ollama 和 NumPy，200 行代码讲透四个核心环节。

---

一、手写需几步

RAG 的宏观流程只有四步：

文档 → [切片] → [向量化] → [检索] → [生成]

但一接触实现层，Chunking Strategy、Embedding Model、Vector Store、Reranker、Hybrid Search、Query Rewriting 这些术语会迅速淹没你。

更高效的路径：先手写一个最简实现，把每个环节的输入输出跑通，完整代码在文末。

---

二、完整代码结构

项目结构非常简单——一个 Python 脚本 + 一个数据目录：

vault-rag/
├── demo-rag.py          ← 223 行，含注释
└── demo-data/           ← 知识库（5 个 Markdown 文件）
    ├── 股票短线交易策略.md
    ├── AI学习路线.md
    ├── 安全测试方法论.md
    ├── macOS开发环境.md
    └── RAG原理简述.md

demo-rag.py 的数据流：

main()
  ├─ demo_fingerprint()    → 检测文件是否变化
  ├─ load_cache()          → 有缓存直接复用
  │   └─ (无缓存或文件变了则走下面)
  ├─ load_notes()          → 步骤1: 读文件 + 切片
  ├─ embed_chunks()        → 步骤2: 文本 → 向量
  ├─ save_cache()          → 保存索引到磁盘
  ├─ search()              → 步骤3: 语义检索
  └─ generate()            → 步骤4: LLM 生成回答

下面逐个函数拆解。

---

三、函数详解

3.1 demo_fingerprint() — 文件指纹

import hashlib

def demo_fingerprint(demo_path):
    hasher = hashlib.md5()                                    # MD5 哈希计算器
    for f in sorted(demo_path.glob("*.md")):                  # 遍历所有 md 文件
        hasher.update(str(f.relative_to(demo_path)).encode()) # 喂入文件路径
        hasher.update(str(f.stat().st_mtime).encode())        # 喂入修改时间
    return hasher.hexdigest()                                 # 输出 32 位十六进制指纹

作用：判断知识库文件有没有增删改。指纹变了 → 重建索引，没变 → 直接用缓存。

逐行解读：

行	代码	解释
hashlib.md5()	创建 MD5 对象	MD5 能把任意长度的数据压缩成固定 32 位十六进制字符串
sorted(...)	排序遍历	保证每次遍历顺序一致，否则同一个目录算出不同哈希
hasher.update(路径)	喂入文件名	文件改名、增删都会改变哈希
hasher.update(mtime)	喂入修改时间	文件内容改了但名字没变，也能检测
hexdigest()	输出指纹	如 "a1b2c3d4e5f67890..."，32 字符

为什么用哈希而不是逐文件比较内容？ 只需读文件名和 mtime，不用打开文件，快得多。

---

3.2 load_cache() / save_cache() — 索引缓存

import pickle
import numpy as np

def save_cache(chunks, fingerprint):
    data = {
        "fingerprint": fingerprint,
        "chunks": [{
            "text": c["text"],
            "source": c["source"],
            "embedding": c["embedding"].tolist()   # numpy → list
        } for c in chunks]
    }
    with open(".demo-cache/index.pkl", "wb") as f:
        pickle.dump(data, f)

def load_cache():
    cache_file = Path(".demo-cache/index.pkl")
    if not cache_file.exists():
        return None                                  # 无缓存，触发重建
    with open(cache_file, "rb") as f:
        data = pickle.load(f)
    for c in data["chunks"]:
        c["embedding"] = np.array(c["embedding"])    # list → numpy
    return data["chunks"], data["fingerprint"]

pickle 是什么？

pickle 是 Python 内置的序列化库——把内存中的 Python 对象「冻干」成二进制文件存磁盘，需要时再「还原」回来。类比 RPG 游戏存档/读档：

# 存档
pickle.dump(data, open("save.pkl", "wb"))   # wb = write binary

# 程序退出，内存清空...

# 读档
data = pickle.load(open("save.pkl", "rb"))  # rb = read binary
# data 完整恢复！

关键细节：c["embedding"].tolist()

• embedding 是 numpy 数组，底层是 C 结构
• pickle 不认 numpy 的二进制格式 → 必须先转成 Python 原生 list
• 读回来时 np.array(...) 转回去，后续算余弦相似度用

pickle vs JSON 怎么选？

	pickle	JSON
支持类型	几乎所有 Python 对象	dict/list/str/int/float/bool/null
可读性	二进制，人不可读	纯文本，人可读
文件大小	更小	更大
适用场景	本地缓存	API 通信、配置文件

缓存场景选 pickle：本地用、存嵌套 dict + 向量、文件尽量小。

---

3.3 load_notes() — 步骤1：加载 + 切片

import re
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_notes(demo_path):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1500,                                  # 每块最多 1500 字符
        chunk_overlap=80,                                 # 相邻块重叠 80 字符
        separators=["\n## ", "\n### ", "\n#### ", "\n", "。", ". ", " "]
    )
    chunks = []
    for md_file in sorted(demo_path.glob("*.md")):
        text = md_file.read_text(encoding="utf-8")
        if len(text.strip()) < 50:                        # 太短没价值，跳过
            continue
        text = re.sub(r"^---\n.*?\n---\n", "", text, flags=re.DOTALL)  # 去掉 YAML 头
        for piece in splitter.split_text(text):
            chunks.append({
                "text": piece.strip(),
                "source": md_file.name,
                "embedding": None                         # 占位，后面向量化填充
            })
    return chunks

分隔符优先级：为什么是 ## → ### → #### → \n → 。 → 空格？

RecursiveCharacterTextSplitter 按 separators 从高到低逐级尝试切分：

输入文本（5000 字符）
  ↓ 第 1 轮：按 \n##  （二级标题）切
  → 如果某段仍 > 1500 字符
    ↓ 第 2 轮：按 \n### （三级标题）切
    → 如果某段仍 > 1500 字符
      ↓ 第 3 轮：按 \n####（四级标题）切
      → 如果还超
        ↓ 第 4 轮：按 \n   （换行/段落）切
        → 如果还超
          ↓ 第 5 轮：按 。（中文句号）切
          → 如果还超
            ↓ 第 6 轮：按 . （英文句号）切
            → 最后兜底：按空格硬切

为什么这样设计？ 优先在 Markdown 标题边界处切，其次段落边界，最后句子边界——每一步都在保证语义完整性。一个 Chunk 大概率是一个完整的小节，而不是在句子中间被拦腰斩断。

对比如果不设优先级、直接按固定长度切：

✗ [Chunk A] ...买入信号出现后，应当在次日开盘价附  ← 句子断了
✗ [Chunk B] 近建仓，止损位设在 10 日均线下方。     ← 语义丢失

✓ [Chunk A] ## 买入策略\n买入信号出现后，应当在次日开...  ← 完整小节

chunk_overlap = 80 不是随便填的

考虑这段文本：

…应当在次日开盘价附近建仓，止损位设在 10 日均线下方。

如果 chunk_size=500 恰好在这里断开，overlap=80 保证「止损位设在 10 日均线下方」至少完整出现在一个块里。没有 overlap，检索时关键信息直接丢失。

re.sub：去掉 YAML frontmatter

Markdown 文件开头常见元数据头：

---
title: 我的笔记
date: 2026-07-08
tags: [RAG, AI]
---

# 正文开始

这段 YAML 对语义检索没有价值，用正则去掉：

正则	含义
^---\n	匹配开头的 --- + 换行
.*?\n	匹配任意内容（? = 非贪婪，遇第一个换行就停）
---\n	匹配结尾的 --- + 换行
re.DOTALL	让 . 也能匹配换行符，实现跨行匹配

---

3.4 embed_chunks() — 步骤2：向量化

import ollama

def embed_chunks(chunks, batch_size=20):
    texts = [c["text"] for c in chunks]                   # 提取所有文本
    embeddings = []
    for i in range(0, len(texts), batch_size):             # 分批处理
        batch = texts[i : i + batch_size]
        resp = ollama.embed(model="qwen3-embedding:0.6b", input=batch)
        embeddings.extend(resp["embeddings"])              # 收集向量
    for chunk, emb in zip(chunks, embeddings):             # 一对一填回
        chunk["embedding"] = np.array(emb)

Embedding 是什么？ 把文本映射到高维空间的一个点。语义相近的文本，向量距离近。

"止损第一"     → [0.12, -0.34, 0.56, ..., 0.08]  (768 维)
"控制亏损"     → [0.10, -0.31, 0.53, ..., 0.06]  ← 距离近（都是风控话题）
"天气预报"     → [-0.67, 0.81, 0.03, ..., 0.42]  ← 距离远（无关话题）

为什么批量（batch）发送？ 逐条调用 = 发 N 次 HTTP 请求，批量 = 发 1 次。实测批量比逐条快 20 倍——所有推理服务（OpenAI、vLLM、Ollama）都是批量吞吐远高于逐条。

zip(chunks, embeddings)：两个列表一一配对。

chunks     = [chunk_A, chunk_B, chunk_C]
embeddings = [emb_A,   emb_B,   emb_C  ]
zip(...)   → [(chunk_A, emb_A), (chunk_B, emb_B), (chunk_C, emb_C)]

---

3.5 search() — 步骤3：检索

def search(query, chunks, top_k=5):
    resp = ollama.embed(model="qwen3-embedding:0.6b", input=[query])  # 问题 → 向量
    query_vec = np.array(resp["embeddings"][0])

    emb_matrix = np.stack([c["embedding"] for c in chunks])           # N×768 矩阵

    # 余弦相似度 = 点积 / (模长 × 模长)
    similarities = np.dot(emb_matrix, query_vec) / (
        np.linalg.norm(emb_matrix, axis=1) * np.linalg.norm(query_vec)
    )

    top_indices = np.argsort(similarities)[-top_k:][::-1]             # Top-K
    return [chunks[i] for i in top_indices]

余弦相似度

公式：cos(θ) = (A · B) / (|A| × |B|)

• A · B（点积）：np.dot(emb_matrix, query_vec) — 两个向量方向越一致，值越大
• |A|（模长）：np.linalg.norm(emb_matrix, axis=1) — 向量的长度，axis=1 表示沿行计算
• 结果范围 [-1, 1]，越接近 1 越相似

np.argsort 怎么取 Top-K？

import numpy as np
similarities = np.array([0.3, 0.9, 0.5, 0.1, 0.7])   # 5 个 chunk 的相似度

np.argsort(similarities)             # → [3, 0, 2, 4, 1]    从小到大排序的索引
np.argsort(similarities)[-3:]        # → [2, 4, 1]           取最后 3 个（最大的）
np.argsort(similarities)[-3:][::-1]  # → [1, 4, 2]           反转 → 从大到小

np.stack 是什么？

vecs = [np.array([1,2]), np.array([3,4]), np.array([5,6])]
np.stack(vecs)
# → array([[1, 2],
#          [3, 4],
#          [5, 6]])    # shape: (3, 2)，3 个向量，每个 2 维

把 N 个独立向量堆叠成 N×768 矩阵，后续 np.dot 一次性算完所有相似度。

这个阶段不需要向量数据库。 NumPy 精确余弦相似度在万级数据下完全够用。向量数据库（Milvus、Pinecone）解决的是另一个量级的问题：

	NumPy（本方案）	向量数据库
数据规模	< 10 万条	百万 ~ 亿级
检索方式	精确余弦相似度	ANN 近似搜索
增量更新	全量重建	实时写入
部署复杂度	零依赖	独立服务

---

3.6 generate() — 步骤4：生成

PROMPT = """基于以下参考资料回答问题。如果资料中没有相关信息，请明确说明。

## 参考资料
{context}

## 问题
{question}

请仅基于以上资料回答："""

def generate(query, retrieved):
    context = "\n\n---\n\n".join(                                  # 分隔线拼接
        f"[来源: {c['source']}]\n{c['text']}" for c in retrieved  # 每条格式
    )
    prompt = PROMPT.format(context=context, question=query)
    resp = ollama.chat(
        model="qwen3:0.6b",
        messages=[{"role": "user", "content": prompt}],
        options={"temperature": 0.3},
    )
    return resp["message"]["content"]

str.join — 字符串拼接

"、".join(["苹果", "香蕉", "橘子"])   # → "苹果、香蕉、橘子"
"---".join(["苹果", "香蕉"])         # → "苹果---香蕉"

这里用 "\n\n---\n\n" 连接多个检索结果：

[来源: 股票短线交易策略.md]
止损第一：单笔亏损不超过总资金2%...

---

[来源: RAG原理简述.md]
RAG = Retrieval-Augmented Generation...

str.format — 字符串模板

"你好，{name}，你的成绩是{score}分".format(name="张三", score=95)
# → "你好，张三，你的成绩是95分"

{context} 和 {question} 是占位符，.format() 时替换为实际内容。

temperature 参数

temperature	效果	适用场景
0.1~0.3	稳定、确定性高，几乎不编造	知识问答、RAG
0.7~1.0	多样、有创意	写作、头脑风暴
>1.0	随机性强，可能胡言乱语	极少使用

RAG 用 0.3：我们要事实，不要创意。

Prompt 设计三要素

1. 标注来源 — [来源: xxx] 让 LLM 知道每段出自哪里
2. 划定边界 — 「仅基于以上资料」是防幻觉最有效的约束
3. 诚实兜底 — 「如果没有请说明」防止 LLM 被迫编造

---

3.7 main() — 主流程

import sys

def main():
    if len(sys.argv) < 2:                                     # 没传问题
        print("用法: python demo-rag.py <问题>")
        sys.exit(1)

    query = " ".join(sys.argv[1:])                            # 拼成完整问题
    fp = demo_fingerprint(DEMO_PATH)                          # 计算文件指纹

    cached = load_cache()                                     # 尝试加载缓存
    if cached and cached[1] == fp:                            # 指纹匹配 → 用缓存
        chunks, _ = cached
    else:                                                     # 重建索引
        chunks = load_notes(DEMO_PATH)                        #   步骤1
        embed_chunks(chunks)                                  #   步骤2
        save_cache(chunks, fp)                                #   保存

    results = search(query, chunks)                           # 步骤3
    answer = generate(query, results)                         # 步骤4
    print(answer)

if __name__ == "__main__":
    main()

sys.argv 怎么处理命令行参数？

python demo-rag.py 股票 交易 策略

sys.argv        # → ["demo-rag.py", "股票", "交易", "策略"]
sys.argv[1:]    # → ["股票", "交易", "策略"]
" ".join(...)   # → "股票 交易 策略"

if __name__ == "__main__" 是什么？

• 直接运行 python demo-rag.py → __name__ = "__main__" → 执行 main()
• 被别的脚本 import demo-rag → __name__ = "demo-rag" → 不执行 main()

既能当脚本跑，也能被 import 复用函数。

---

四、完整代码

依赖只有三个库：

pip install ollama numpy langchain-text-splitters

项目代码如下

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Date          : 2026/7/8
@File          : 1.py
@Author        : liwei68
@Description   : 
"""
"""
demo-rag.py — 最小 RAG Demo，5 个文件秒级跑完
用法: python demo-rag.py <问题>
示例: python demo-rag.py 股票交易策略是什么
"""
import hashlib  # 哈希库，用于生成文件指纹（检测文件变化）
import pickle   # 序列化库，用于将索引缓存到磁盘
import re       # 正则表达式，用于去除 YAML frontmatter
import sys      # 系统库，读取命令行参数
import time     # 时间库，计时各步骤耗时
from pathlib import Path                        # 路径库，跨平台处理文件路径
from typing import Optional, Tuple              # 类型标注，让函数签名更清晰

import numpy as np                              # 数值计算库，用于向量相似度计算
import ollama                                   # Ollama 客户端，调用本地大模型
from langchain_text_splitters import RecursiveCharacterTextSplitter  # LangChain 文本切片器

# ═══════════════════════════════════════════
# 配置 — 只加载 demo-data/ 下的 5 个文件
# ═══════════════════════════════════════════

DEMO_PATH = Path(__file__).parent / "demo-data"      # 知识库路径：脚本同级目录下的 demo-data/
CACHE_DIR = Path(__file__).parent / ".demo-cache"    # 缓存目录：脚本同级目录下的 .demo-cache/
EMBED_MODEL = "qwen3-embedding:0.6b"                 # 嵌入模型：将文本转为向量（语义表示）
CHAT_MODEL = "qwen3:0.6b"                            # 对话模型：基于检索结果生成回答
CHUNK_SIZE = 1500                                    # 切片大小：每个文本块最多 1500 字符
CHUNK_OVERLAP = 80                                   # 切片重叠：相邻两块共享 80 字符（避免切断上下文）
TOP_K = 5                                            # 检索数量：取相似度最高的 5 个文本块

# ═══════════════════════════════════════════
# 缓存 — 避免每次运行都重新向量化（首次 ~5s，之后秒级）
# ═══════════════════════════════════════════

def demo_fingerprint(demo_path: Path) -> str:
    """生成文件指纹：遍历所有 md 文件，对「文件名+修改时间」做 MD5 哈希。
       文件有增删改时指纹变化 → 触发重建索引。"""
    hasher = hashlib.md5()                                    # 创建 MD5 哈希对象
    for f in sorted(demo_path.glob("*.md")):                  # 遍历所有 md 文件（按文件名排序保证一致性）
        hasher.update(str(f.relative_to(demo_path)).encode()) # 加入文件相对路径
        hasher.update(str(f.stat().st_mtime).encode())        # 加入文件最后修改时间
    return hasher.hexdigest()                                 # 返回 32 位十六进制指纹


def save_cache(chunks: list, fingerprint: str):
    """将文本块和向量保存到磁盘，下次直接加载跳过向量化。"""
    CACHE_DIR.mkdir(exist_ok=True)                            # 创建缓存目录（已存在则不报错）
    data = {
        "fingerprint": fingerprint,                           # 保存指纹，加载时比对
        "chunks": [{                                          # 列表推导式：把每个 chunk 的 numpy 数组转成 list
            "text": c["text"],                                #   文本内容
            "source": c["source"],                            #   来源文件名
            "embedding": c["embedding"].tolist()              #   向量 → 普通 list（pickle 不能直接序列化 numpy 数组）
        } for c in chunks],
    }
    with open(CACHE_DIR / "index.pkl", "wb") as f:            # 以二进制写入模式打开文件
        pickle.dump(data, f)                                  # 序列化写入


def load_cache() -> Optional[Tuple[list, str]]:
    """从磁盘加载缓存。返回 None 表示缓存不存在。"""
    cache_file = CACHE_DIR / "index.pkl"                      # 缓存文件路径
    if not cache_file.exists():                               # 文件不存在
        return None                                           #   返回 None，触发重建
    with open(cache_file, "rb") as f:                         # 以二进制读取模式打开
        data = pickle.load(f)                                 # 反序列化读出
    for c in data["chunks"]:                                  # 遍历每个 chunk
        c["embedding"] = np.array(c["embedding"])             #   把 list 转回 numpy 数组（推理时用 numpy 算相似度）
    return data["chunks"], data["fingerprint"]                # 返回（文本块列表, 指纹）


# ═══════════════════════════════════════════
# 步骤 1：加载 + 切片 — 读文件 → 去格式 → 切小块
# ═══════════════════════════════════════════

def load_notes(demo_path: Path) -> list[dict]:
    """加载所有 md 文件，去除 frontmatter，切成带重叠的文本块。"""
    splitter = RecursiveCharacterTextSplitter(                # 递归字符切片器：按优先级逐级切分
        chunk_size=CHUNK_SIZE,                                #   每个块不超过 1500 字符
        chunk_overlap=CHUNK_OVERLAP,                          #   相邻块重叠 80 字符
        separators=["\n## ", "\n### ", "\n#### ", "\n", "。", ". ", " "],  # 分隔符优先级：二级标题→三级→四级→换行→中文句号→英文句号→空格
    )
    chunks: list[dict] = []                                   # 结果列表，每个元素 {text, source, embedding}
    files = sorted(demo_path.glob("*.md"))                    # 获取所有 md 文件（排序保证顺序一致）
    print(f"📂 加载 {len(files)} 个文件: {[f.name for f in files]}")  # 打印加载的文件名列表

    for md_file in files:                                     # 逐个处理文件
        try:
            text = md_file.read_text(encoding="utf-8")        #   读取文件内容（UTF-8 编码）
        except Exception:                                     #   编码错误等异常
            continue                                          #     跳过这个文件
        if len(text.strip()) < 50:                            #   内容太短（<50 字符）没检索价值
            continue                                          #     跳过
        text = re.sub(                                        #   正则替换：去掉 Obsidian 笔记的 YAML frontmatter
            r"^---\n.*?\n---\n", "", text,                    #     匹配开头的 ---...--- 块
            flags=re.DOTALL                                   #     DOTALL 让 . 也能匹配换行符
        )
        for piece in splitter.split_text(text):               #   对清理后的文本切片
            chunks.append({                                   #     每个切片存为一个 chunk
                "text": piece.strip(),                        #       文本内容（去除首尾空白）
                "source": md_file.name,                       #       来源文件名
                "embedding": None                             #       向量占位，后面 embed_chunks 填充
            })
    return chunks                                             # 返回所有文本块


# ═══════════════════════════════════════════
# 步骤 2：向量化 — 把文本块变成语义向量
# ═══════════════════════════════════════════

def embed_chunks(chunks: list[dict], batch_size: int = 20):
    """批量调用 Ollama 嵌入模型，将每个文本块转为向量。
       批量发送比逐个发快很多（减少网络往返）。"""
    texts = [c["text"] for c in chunks]                       # 提取所有文本内容
    embeddings = []                                           # 存放返回的向量列表
    for i in range(0, len(texts), batch_size):                # 按 batch_size 分批处理
        batch = texts[i:i + batch_size]                       #   取当前批次
        resp = ollama.embed(model=EMBED_MODEL, input=batch)   #   调用 Ollama 嵌入接口
        embeddings.extend(resp["embeddings"])                 #   把返回的向量追加到结果
    for chunk, emb in zip(chunks, embeddings):                # 把向量一对一填回 chunk
        chunk["embedding"] = np.array(emb)                    #   转成 numpy 数组（方便后面矩阵运算）


# ═══════════════════════════════════════════
# 步骤 3：检索 — 把问题向量化，找最相似的文本块
# ═══════════════════════════════════════════

def search(query: str, chunks: list[dict], top_k: int = TOP_K) -> list[dict]:
    """余弦相似度检索：问题 → 向量 → 和所有 chunk 算相似度 → 取 Top-K。"""
    resp = ollama.embed(model=EMBED_MODEL, input=[query])     # 将用户问题也向量化
    query_vec = np.array(resp["embeddings"][0])               # 取第一个（也是唯一一个）向量 → numpy 数组
    emb_matrix = np.stack([c["embedding"] for c in chunks])   # 把所有 chunk 向量堆叠成矩阵（N×D）
    similarities = np.dot(emb_matrix, query_vec) / (          # 余弦相似度 = 向量点积 / (模长乘积)
        np.linalg.norm(emb_matrix, axis=1)                    #   emb_matrix 每行的模长（axis=1 沿行计算）
        * np.linalg.norm(query_vec)                           #   查询向量的模长
    )
    top_indices = np.argsort(similarities)[-top_k:][::-1]     # argsort 返回从小到大排序的索引 → 取最后 top_k 个 → 反转得从大到小
    return [chunks[i] for i in top_indices]                   # 按索引取出最相似的文本块


# ═══════════════════════════════════════════
# 步骤 4：生成 — 把检索结果拼接进 prompt，让 LLM 回答
# ═══════════════════════════════════════════

PROMPT = """你是用户的第二大脑，基于用户的笔记内容回答问题。

## 相关笔记
{context}

## 问题
{question}

基于以上笔记回答。如果答案不在笔记中，诚实说明。"""


def generate(query: str, retrieved: list[dict]) -> str:
    """将检索到的文本块拼成上下文，发给 LLM 生成最终回答。"""
    context = "\n\n---\n\n".join(                              # 用分隔线拼接多个文本块
        f"[{c['source']}]\n{c['text']}" for c in retrieved    #   每条格式: [文件名]\n内容
    )
    prompt = PROMPT.format(context=context, question=query)   # 把上下文和问题填入 prompt 模板
    resp = ollama.chat(                                       # 调用 Ollama 对话接口
        model=CHAT_MODEL,                                     #   使用对话模型
        messages=[{"role": "user", "content": prompt}],        #   消息列表：只有一条用户消息
        options={"temperature": 0.3},                         #   temperature=0.3 让回答更稳定、少胡编
    )
    return resp["message"]["content"]                         # 提取 LLM 回复内容


# ═══════════════════════════════════════════
# CLI — 命令行入口，串联四步流程
# ═══════════════════════════════════════════

def main():
    """RAG 主流程：检查缓存 → 加载/构建索引 → 检索 → 生成。"""
    if len(sys.argv) < 2:                                     # 没有传问题
        print("用法: python demo-rag.py <问题>")               #   打印用法
        print("示例: python demo-rag.py 如何做安全测试")
        print("      python demo-rag.py AI学习路线是什么")
        sys.exit(1)                                           #   退出

    query = " ".join(sys.argv[1:])                            # 把命令行参数拼成完整问题（支持空格）
    fp = demo_fingerprint(DEMO_PATH)                          # 计算当前文件指纹

    # 尝试加载缓存
    cached = load_cache()                                     # 尝试从磁盘加载
    if cached and cached[1] == fp:                            # 缓存存在且指纹匹配（文件没变）
        chunks, _ = cached                                    #   直接使用缓存
        print(f"📦 从缓存加载索引 ({len(chunks)} 个文本块)\n")
    else:                                                     # 缓存不存在或文件变了 → 重建索引
        print(f"📂 {DEMO_PATH.name}/ ({len(list(DEMO_PATH.glob('*.md')))} 个文件)")
        t0 = time.time()                                      #   开始计时
        chunks = load_notes(DEMO_PATH)                        #   步骤1: 加载 + 切片
        print(f"  ✅ 切片 → {len(chunks)} 个文本块 ({time.time()-t0:.1f}s)")

        t0 = time.time()                                      #   重新计时
        embed_chunks(chunks)                                  #   步骤2: 向量化
        print(f"  🧮 向量化完成 ({time.time()-t0:.1f}s)")

        save_cache(chunks, fp)                                #   步骤2.5: 保存缓存（下次秒开）
        print(f"  💾 索引已缓存\n")

    # 检索
    print(f"🔍 问题: {query}")
    t0 = time.time()                                          # 计时开始
    results = search(query, chunks)                           # 步骤3: 语义检索
    print(f"  ✅ 检索 {len(results)} 个片段 ({time.time()-t0:.1f}s)")
    for r in results:                                         # 打印每个检索结果
        preview = r['text'][:80].replace('\n', ' ')           #   取前 80 字符作为预览（换行符替换为空格）
        print(f"    ─ {r['source']}: {preview}...")

    # 生成
    print(f"\n🤖 生成回答 ({CHAT_MODEL})")
    t0 = time.time()                                          # 计时开始
    answer = generate(query, results)                         # 步骤4: LLM 生成
    print(f"\n{'='*60}")                                      # 分割线
    print(answer)                                             # 打印回答
    print(f"{'='*60}")
    print(f"\n⏱ 总耗时 {time.time()-t0:.1f}s")                # 打印生成耗时


if __name__ == "__main__":                                    # 直接运行时（非 import）才执行 main
    main()

---

使用 Ollama 本地模型（qwen3-embedding:0.6b + qwen3:0.6b），所有计算在本地完成，无需 API Key。

---

总结：200 行代码 → 企业级 RAG 的映射表

代码模块	200 行教你的原则	企业级落地（MaxKB）
配置 (L19-29)	集中管理可变参数	数据库存配置，多知识库独立策略，Web 界面修改
缓存 (L31-69)	文件指纹避免重复计算	文档级增量更新，改一个文档只重建一个
切片 (L72-104)	按文档结构切，分隔符有优先级	多格式解析（PDF/Word/HTML），按文档类型配策略，保留元数据
向量化 (L107-121)	批处理减少网络往返	异步流水线，多模型支持，pgvector 存储
检索 (L124-138)	余弦相似度语义搜索	混合检索 + Reranker + 多轮对话改写 + 元数据过滤
生成 (L141-167)	三段式 Prompt + 来源标注	硬约束防幻觉 + 置信度阈值 + 结构化输出 + 引用溯源
编排 (L170-223)	分步计时，可观测可调试	全链路追踪 + 对话日志 + REST API + Web UI + 多租户

核心方案点：
200 行代码证明了 RAG 的可行性，但每一条原则在企业级系统中都被放大了 10 倍。

缓存从"文件指纹 + pickle"变成了"文档级增量 + 数据库"；检索从"numpy 余弦相似度"变成了"混合检索 + Reranker + 元数据过滤"；生成从"三段式 Prompt"变成了"硬约束 + 置信度阈值 + 结构化输出"。这不是代码行数的堆叠，而是从"能跑"到"可靠"的工程化。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/weixin_42439274/article/details/162706859