从零实现 RAG：拆解知识库问答的 10 个核心函数

大模型能够流畅作答，却不代表答案一定可靠。面对专业资料时，仅依赖模型自身知识容易产生缺少依据的内容，而 RAG 可以在生成前先检索相关文档，再将可靠片段交给模型。下面从一条完整的知识库问答流程入手，拆解文档处理、向量检索与回答生成背后的十个核心函数。

从 0 手敲 RAG：我把这 10 个函数全部搞懂了

学习日记 · RAG 基础篇 ｜ M3E + FAISS + LLM · 10 个核心函数全拆解

关于代码：掘金编辑器是所见即所得的，从 .md 文件整篇复制时不会自动识别 ``` 代码块。本文采用 「正文讲清思路 + 关键 API 调用示例 + 完整代码集中文末附录」 的写法；如要按段复制代码，请直接跳到文末的「附录：完整代码片段」或打开同目录下的 从0手敲RAG-我把10个函数全搞懂了-代码片段.md 文件。

本文是 FastAPI + RAG 学习日记系列的基础篇。前几篇我们把整个 RAG 系统接进了 FastAPI、做了聊天客户端、加了闲聊与联网模式——但一直没说底层那条流水线到底是怎么跑起来的。这一篇回到原点，把 10 个核心函数从头到尾拆一遍。

之前我对 AI 最大的吐槽，就是它一本正经地胡说八道。

问它专业问题，回答得头头是道，语气坚定得像背过答案，等你拿着去核对，好家伙，一半是它编的。

直到这周我把 RAG 的整个流程手敲了一遍，才反应过来：AI 会乱编，不是它笨，是它在**「闭卷考试」**。

没看过课本的人闭卷答题，只能瞎编。RAG 干的事，就是给它一张**「开卷考试」的卷子**——先把资料检索好，再让它照着资料回答。

我把自己整理的这条 RAG 流水线拆成了 10 个函数，从加载模型到最终回答，每一个细节都踩过、也都弄明白了。今天全部分享给你。

---

一、RAG 到底是什么？

一句话：RAG = Retrieval-Augmented Generation，检索增强生成。

模型在回答之前，先去知识库里检索相关资料，再把这些资料一起塞给大模型，让它照着资料回答。带来的好处是：

• 减少幻觉：资料里没写的，模型不会硬编。

• 可追溯：每条回答都能列出引用了哪几个片段。

• 知识可更新：换一份文档，回答立刻跟得上，不用重训模型。

---

二、整条流水线长什么样

把 10 个函数按调用顺序串起来，就是一条完整的 RAG 链路：

① load_m3e_model 加载文本转向量的模型

② load_document 把 pdf/docx/txt/md 统一读成纯文本

③ chunk_text 按空行 / 滑动窗口切成块

④ encode_texts 把每块文本转成向量

⑤ build_faiss_index 向量灌进 FAISS 建索引

────────────── 索引构建完成 ──────────────

⑥ search 用户问题 → 向量 → 检索 Top-K

⑦ get_openai_client 拿到大模型客户端

⑧ _build_messages 把检索片段编进 Chat Completions 消息

⑨ generate_answer 一次性生成完整回答

⑩ stream_answer 流式回答（边生成边 yield）

上半部分是离线索引：把文档准备好，向量化存好；下半部分是在线问答：拿问题去检索，召回片段，喂给大模型。

下面逐个拆。每个函数先讲「目的 + 关键调用」，完整代码全部放在文末附录。

---

三、10 个函数逐个拆解

1. load_m3e_model：给文字装上一把"尺子"

目的：m3e 是一个把文本转向量的模型。RAG 本质是个向量库，要先把文档转成向量存进去——这是大模型解析之前的必要环节。

关键调用：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("m3e模型路径")

SentenceTransformer 会优先读本地缓存好的 m3e 模型，本地没有就自动下载并缓存，后面再加载就不用重复下。

划重点：RAG 的第一步不是写业务逻辑，而是「把文字量出坐标」。后面所有检索都建立在坐标距离上——意思越相近的文字，坐标离得越近。

---

2. load_document：不管什么格式，先变成纯文本

目的：文档格式五花八门（txt、md、docx、pdf），必须统一转换成可读的纯文本，后面才能处理。

关键调用：

from pypdf import PdfReader # pip install pypdf

from docx import Document # pip install python-docx

from pathlib import Path

text = "nn".join(page.extract_text() or "" for page in PdfReader("a.pdf").pages)

text = "nn".join(p.text for p in Document("a.docx").paragraphs)

text = Path("a.txt").read_text(encoding="utf-8")

按后缀各找各的解析方式：

• pdf → 引入 PdfReader 解析

• docx → 用 Document 解析

• txt / md → 直接 path.read_text() 读

为什么要统一：后续 chunk_text 只认字符串。多走这一层，后面的逻辑就再也不用关心格式。

---

3. chunk_text：把长文本切成块

目的：一整本资料没法直接喂给模型。切块后每一块的「边界」更清晰，单独编码成向量后，检索时定位更精准。

切块策略：

• 先按空行切成段落；

• 短段落先合并，攒到接近 chunk_size 才切一刀；

• 太长的段落按字符数硬切，并留出一段 overlap（重叠）。

关键代码（长段落的滑动窗口切分）：

step = chunk_size - overlap

for start in range(0, len(para), step):

piece = para[start : start + chunk_size]

chunks.append(piece)

if start + chunk_size >= len(para):

break

为什么要有 overlap：硬切的边界会切断语义。留一段重叠，相当于把前后两块「粘」起来，让语义连续——这是切块质量的关键。

---

4. encode_texts：把文字块变成坐标

目的：模型看不懂文字，把切好的块都转成向量，后面才能做相似度查询。

关键调用：

embeddings = model.encode(

texts,

normalize_embeddings=True, # L2 长度归一化成 1

show_progress_bar=False,

)

# shape = (条数, 向量维度)，例如 (10, 768)

划重点：向量就是把文字「量出坐标」。意思越相近的文字，坐标离得越近。

为什么开 normalize_embeddings=True：归一化之后，后面用内积计算就等价于余弦相似度——又快又准。

---

5. build_faiss_index：给向量建个"索引本"

目的：把向量放进 FAISS 索引，之后才能查询。

关键调用：

import faiss

import numpy as np

dim = embeddings.shape[1] # 第二维 = 向量维度，例如 768

index = faiss.IndexFlatIP(dim) # Flat=不压缩，IP=内积

index.add(embeddings.astype(np.float32)) # FAISS 要求 float32

这里用的是 IndexFlatIP，意思是暴力内积检索（Flat = 不压缩，精确计算）。前面已经把向量归一化，所以这里算出的内积分数就约等于余弦相似度，越接近 1 越相似。

一个容易踩的坑：index.add() 之后，每条向量在索引里会有个下标 0..n-1，这个下标必须和 chunks / sources 列表严格对齐——对齐不上，检索到了也找不到原文。

---

6. search：拿问题去"捞"最近的资料

目的：把用户问题转成向量，检索出最接近的文本片段，返回给大模型解析。

实现分五步：

# ① 问题也要向量化

query_vec = encode_texts(model, [query])

# ② 多取一些再按阈值过滤，避免刚好过线的结果被硬截断

fetch_k = min(max(top_k * 3, top_k), len(chunks))

# ③ FAISS 搜索：scores / indices 的 shape 都是 (查询条数, fetch_k)

scores, indices = index.search(query_vec.astype(np.float32), k=fetch_k)

# ④ 只取第 0 条查询的结果，拼成可读结构

for score, idx in zip(scores[0], indices[0]):

if idx < 0: continue # 无效命中

if float(score) < min_score: continue # 相似度太低当噪音

results.append({"score": float(score), "source": sources[idx], "text": chunks[idx]})

if len(results) >= top_k: break

两个关键设计：

• fetch_k = top_k * 3：先多捞三倍回来再用阈值筛，防止刚好过线的结果被截断；

• 取 [0]：目前只支持单问题查询；多条问题会生成二维数组——这点后面还要改。

---

7. get_openai_client：把大模型"请"出来

目的：检索完的资料最终要交给大模型解析、生成回答——这一步负责创建大模型客户端。

.env 配置：

OPENAI_API_KEY=你的密钥 # 必填

OPENAI_BASE_URL=https://... # 可选，兼容服务（智谱 / DeepSeek）

OPENAI_MODEL=glm-4-flash # generate / stream 里读取

智谱示例：

OPENAI_BASE_URL=https://open.bigmodel.cn/api/paas/v4/

关键调用：

import os

from openai import OpenAI

api_key = os.getenv("OPENAI_API_KEY")

if not api_key:

raise RuntimeError("未设置 OPENAI_API_KEY。请复制 .env.example 为 .env 并填写密钥。")

return OpenAI(

api_key=api_key,

base_url=os.getenv("OPENAI_BASE_URL") or None, # 空字符串也走默认官方

)

小细节：.env 没设 OPENAI_BASE_URL 时，os.getenv 返回 None；但用户可能写空字符串——用 or None 把空字符串也归一成「默认官方」。

---

8. _build_messages：RAG 真正"合体"的地方

目的：把 search() 返回的片段编进对话消息，让大模型根据资料精确作答——这是 RAG 减幻觉的核心。

关键调用：

# ① 把多条检索片段格式化成一段可读上下文

context_text = "nn".join(

f"[{i}] 来源: {c['source']}n{c['text']}"

for i, c in enumerate(contexts, start=1)

)

# ② 系统提示词：给模型定规矩（只根据上下文答、别瞎编）

system_prompt = (

"你是一个知识库助手。请只根据提供的上下文回答问题；"

"如果上下文不足以回答，请明确说明「根据现有资料无法确定」。"

"回答简洁、准确，使用中文。"

)

# ③ 用户提示词：问题 + 上下文（大模型主要读这里）

user_prompt = f"问题：{query}nn上下文：n{context_text}"

return [

{"role": "system", "content": system_prompt},

{"role": "user", "content": user_prompt},

]

Chat Completions 一般有 system / user 两种角色：

• system 负责定规矩；

• user 负责放本次问题和检索材料。

划重点：提示词里那句「只根据上下文回答；上下文不足就明确说不知道」是 RAG 减幻觉的关键——把模型的嘴管住，不许它靠参数记忆乱编。

---

9. generate_answer：一次性生成完整回答

目的：调大模型，等模型整段生成完，一次性返回。

关键调用：

client = get_openai_client()

model_name = model or os.getenv("OPENAI_MODEL", "gpt-4o-mini")

response = client.chat.completions.create(

model=model_name,

messages=_build_messages(query, contexts),

temperature=0.2, # 越低越稳、越「照着材料说」

)

return (response.choices[0].message.content or "").strip()

关于 temperature，我花了不少时间才理解：

• 数值越低，模型越稳、越照着材料说——知识库问答通常给 0.2；

• 数值越高越有创意、越容易跑偏——常见 0.9，更适合闲聊场景。

两个收尾细节：

• or ""：防止 content 为 None；

• .strip()：去掉首尾空白。

---

10. stream_answer：流式回答，打字机效果

目的：边生成边返回，前端体验像打字机一样一个字一个字蹦出来。

逻辑和第 9 步完全相同，只多一个 stream=True：

stream = client.chat.completions.create(

model=model_name,

messages=_build_messages(query, contexts),

temperature=0.2,

stream=True,

)

for chunk in stream:

delta = chunk.choices[0].delta.content

if delta: # 有的 chunk 内容为空（角色信息、结束标记）

yield delta

两种模式的区别：

• generate_answer：等模型全部生成完，一次性返回整段文字；

• stream_answer：模型每产出一小段就 yield 出来——前端 /ask/stream 接口会把这些 token 包装成 SSE（Server-Sent Events）推给浏览器。

踩坑提醒：delta 可能为 None（比如只包含角色信息或结束标记的 chunk），必须判空再 yield，不然前端会收到 undefined。

---

四、我踩过的两个最大的坑

跑完之后回头看，链路虽然通，但过程里有两个坑几乎反复绊倒我。

坑 1：FAISS 下标必须和 chunks / sources 严格对齐

index.add(embeddings) 之后，每条向量在索引里都对应一个下标 0..n-1。search() 返回的 indices 就是这些下标——必须保证它们和 chunks[i]、sources[i] 是同一份数据按同一顺序加进去的。

build_faiss_index(chunks) 时

↓

embeddings[i] 是 chunks[i] 的向量

↓

search() 返回的 indices[j] = i

↓

chunks[i] / sources[i] 就是这次命中的原文

对不齐就完蛋：检索命中了，但读出来的原文牛头不对马嘴——而且这种 bug 不会报错，只会「答案越看越怪」。

坑 2：fetch_k 和 min_score 必须配合着用

• 只设 top_k=3：如果第 4、5 条是「勉强过线但有用」，直接被截掉；

• 只设 min_score=0.3：噪音进不来，但有效召回也变少了。

正确做法：先多捞几倍（fetch_k = top_k * 3），再用阈值（min_score）筛。两者配合才稳——这个组合我调了一个晚上才想明白。

如果你发现 AI 答非所问：

先别骂模型，回去检查检索回来的那几段资料对不对——

资料没捞对，模型再强也白搭。

---

五、写在最后

跑通最后一步那天晚上，我把这 10 个函数从头到尾又看了一遍，突然有种「整条链路在脑子里亮起来」的感觉：

加载模型 → 读文档 → 切块 → 向量化 → 建索引

↓

用户问题 → 检索 → 拼 Prompt → 一次性回答 / 流式回答

每一环单拿出来都是常识，连在一起，就是让 AI **「开卷考试」**的完整闭环。

下一步会继续更新 FastAPI + RAG 系列：

• 索引缓存：重复启动不用每次重建，毫秒级拉起；

• 文档管理上传：让 chat.html 里就能上传新文件、自动重建索引；

• MCP 方向拓展：把这条流水线暴露成 Model Context Protocol 工具，让外部 Agent 直接调用。

---

附录：完整代码片段

10 个函数的完整实现，按「装环境 → 离线索引 → 在线问答 → 一个最小 demo」四段组织。复制时请用掘金编辑器的「代码块」按钮，或直接打开同目录下的 从0手敲RAG-我把10个函数全搞懂了-代码片段.md。

A. 通用依赖

pip install sentence-transformers faiss-cpu pypdf python-docx openai numpy python-dotenv

B. 离线索引：模型 + 文档 + 切块 + 向量化 + 建索引

# rag_index.py —— 离线索引：把文档变成可检索的 FAISS 索引

from __future__ import annotations

from pathlib import Path

import faiss

import numpy as np

from docx import Document

from pypdf import PdfReader

from sentence_transformers import SentenceTransformer

# 1. load_m3e_model：加载文本转向量的模型

def load_m3e_model(model_path: str = "m3e模型路径") -> SentenceTransformer:

return SentenceTransformer(model_path)

# 2. load_document：把 pdf/docx/txt/md 统一读成纯文本

def load_document(path: str | Path) -> str:

p = Path(path)

suffix = p.suffix.lower()

if suffix == ".pdf":

reader = PdfReader(str(p))

return "nn".join(page.extract_text() or "" for page in reader.pages)

if suffix == ".docx":

doc = Document(str(p))

return "nn".join(para.text for para in doc.paragraphs)

return p.read_text(encoding="utf-8")

# 3. chunk_text：按空行/滑动窗口切块

def chunk_text(

text: str,

chunk_size: int = 500,

overlap: int = 50,

) -> list[str]:

paragraphs = [p.strip() for p in text.split("nn") if p.strip()]

chunks: list[str] = []

for para in paragraphs:

if len(para) <= chunk_size:

chunks.append(para)

continue

step = chunk_size - overlap

for start in range(0, len(para), step):

chunks.append(para[start : start + chunk_size])

if start + chunk_size >= len(para):

break

return chunks

# 4. encode_texts：把文字块变成向量

def encode_texts(model: SentenceTransformer, texts: list[str]) -> np.ndarray:

return model.encode(

texts,

normalize_embeddings=True,

show_progress_bar=False,

)

# 5. build_faiss_index：把向量灌进 FAISS

def build_faiss_index(embeddings: np.ndarray) -> faiss.Index:

dim = embeddings.shape[1]

index = faiss.IndexFlatIP(dim)

index.add(embeddings.astype(np.float32))

return index

C. 在线问答：检索 + Prompt + 一次性 / 流式回答

# rag_qa.py —— 在线问答：检索 + 调大模型

from __future__ import annotations

import os

from collections.abc import Iterator

import faiss

import numpy as np

from openai import OpenAI

from sentence_transformers import SentenceTransformer

# 6. search：拿问题去"捞"最近的资料

def search(

query: str,

model: SentenceTransformer,

index: faiss.Index,

chunks: list[str],

sources: list[str],

top_k: int = 3,

min_score: float = 0.3,

) -> list[dict]:

query_vec = encode_texts(model, [query])

fetch_k = min(max(top_k * 3, top_k), len(chunks))

scores, indices = index.search(query_vec.astype(np.float32), k=fetch_k)

results: list[dict] = []

for score, idx in zip(scores[0], indices[0]):

if idx < 0:

continue

score_f = float(score)

if score_f < min_score:

continue

results.append({

"score": score_f,

"source": sources[idx],

"text": chunks[idx],

})

if len(results) >= top_k:

break

return results

# 7. get_openai_client：拿到大模型客户端

def get_openai_client() -> OpenAI:

api_key = os.getenv("OPENAI_API_KEY")

if not api_key:

raise RuntimeError(

"未设置 OPENAI_API_KEY。请复制 .env.example 为 .env 并填写密钥。"

)

return OpenAI(

api_key=api_key,

base_url=os.getenv("OPENAI_BASE_URL") or None,

)

# 8. _build_messages：把检索片段编进 Chat Completions 消息

def _build_messages(query: str, contexts: list[dict]) -> list[dict]:

context_text = "nn".join(

f"[{i}] 来源: {c['source']}n{c['text']}"

for i, c in enumerate(contexts, start=1)

)

system_prompt = (

"你是一个知识库助手。请只根据提供的上下文回答问题；"

"如果上下文不足以回答，请明确说明「根据现有资料无法确定」。"

"回答简洁、准确，使用中文。"

)

user_prompt = f"问题：{query}nn上下文：n{context_text}"

return [

{"role": "system", "content": system_prompt},

{"role": "user", "content": user_prompt},

]

# 9. generate_answer：一次性生成完整回答

def generate_answer(

query: str,

contexts: list[dict],

model: str | None = None,

) -> str:

client = get_openai_client()

model_name = model or os.getenv("OPENAI_MODEL", "gpt-4o-mini")

response = client.chat.completions.create(

model=model_name,

messages=_build_messages(query, contexts),

temperature=0.2,

)

return (response.choices[0].message.content or "").strip()

# 10. stream_answer：流式回答，打字机效果

def stream_answer(

query: str,

contexts: list[dict],

model: str | None = None,

) -> Iterator[str]:

client = get_openai_client()

model_name = model or os.getenv("OPENAI_MODEL", "gpt-4o-mini")

stream = client.chat.completions.create(

model=model_name,

messages=_build_messages(query, contexts),

temperature=0.2,

stream=True,

)

for chunk in stream:

delta = chunk.choices[0].delta.content

if delta:

yield delta

D. 把整条流水线串起来（一个最小 demo）

# demo.py —— 一条命令跑通整条流水线

from rag_index import (

load_m3e_model, load_document, chunk_text,

encode_texts, build_faiss_index,

)

from rag_qa import search, generate_answer, stream_answer

# ① 加载模型

model = load_m3e_model("m3e模型路径")

# ② 读文档 + 切块

text = load_document("test/data/intro.md")

chunks = chunk_text(text)

sources = ["intro.md"] * len(chunks)

# ③ 向量化 + 建索引

embeddings = encode_texts(model, chunks)

index = build_faiss_index(embeddings)

# ④ 检索

query = "什么是 RAG？"

hits = search(query, model, index, chunks, sources, top_k=3, min_score=0.3)

for h in hits:

print(f"[{h['score']:.3f}] {h['source']}: {h['text'][:80]}...")

# ⑤ 一次性回答

print("n=== 一次性回答 ===")

print(generate_answer(query, hits))

# ⑥ 流式回答

print("n=== 流式回答 ===")

for token in stream_answer(query, hits):

print(token, end="", flush=True)

print()