手写一个 RAG：从零搭建可溯源的检索增强问答系统
1. 引言：为什么需要手写 RAG
大语言模型虽然强大，但存在知识截止、幻觉和无法引用来源等问题。检索增强生成（Retrieval-Augmented Generation，RAG）通过先检索再生成的方式，把外部知识注入模型，从而提升回答的准确性和可溯源性。本文不依赖 LangChain 等框架，从零手写一个可运行的 RAG 系统，覆盖文档加载、切分、向量化、检索、生成和溯源全流程。
手写 RAG 的好处在于：你能清楚看到每个环节的输入输出，便于调试、定制和教学。下面我们逐步搭建。
2. 系统架构与整体流程
一个最小可用的 RAG 系统包含以下模块：
- 文档加载器：读取本地文本或 Markdown 文件。
- 文本切分器：把长文档切成适合检索的块（Chunk）。
- 向量化器：把文本块编码为向量。
- 向量存储与检索：保存向量并支持相似度检索。
- 提示词组装：把检索结果和用户问题拼成 Prompt。
- 生成器：调用大模型生成回答。
- 溯源模块：把回答关联到原始文档块。
整体流程如下：
flowchart TD A[加载文档] --> B[文本切分] B --> C[向量化] C --> D[向量存储] E[用户提问] --> F[问题向量化] F --> G[相似度检索] D --> G G --> H[组装 Prompt] H --> I[大模型生成] I --> J[返回回答与溯源]3. 环境准备与依赖安装
本文使用 Python 3.10+，核心依赖如下：
pip install openai numpy说明：
- openai：用于调用 Embedding 模型和对话模型，兼容 OpenAI 接口的服务均可。
- numpy：用于向量相似度计算。
如果你希望完全本地运行，可以把 Embedding 和生成模型替换为本地模型（如 BGE、Qwen），本文以 OpenAI 兼容接口为例，便于快速跑通。
4. 文档加载与文本切分
首先实现文档加载器，支持读取纯文本和 Markdown 文件：
from pathlib import Path def load_document(file_path: str) -> str: """读取本地文本或 Markdown 文件内容。""" path = Path(file_path) if not path.exists(): raise FileNotFoundError(f"文件不存在: {file_path}") return path.read_text(encoding="utf-8")接下来实现文本切分器。切分策略直接影响检索质量，这里采用「按段落优先、按长度兜底」的策略：
import re def split_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]: """ 把长文本切成多个块。 - chunk_size: 每个块的最大字符数 - overlap: 相邻块之间的重叠字符数，用于保持上下文连贯 """ # 先按空行切分为段落 paragraphs = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] chunks = [] current = "" for para in paragraphs: # 如果当前块加上新段落会超长，先保存当前块 if len(current) + len(para) + 1 > chunk_size and current: chunks.append(current) # 保留尾部 overlap 字符作为下一块的开头，保持上下文 current = current[-overlap:] if overlap > 0 else "" current = (current + "\n" + para).strip() if current: chunks.append(current) return chunks切分时保留重叠区域，可以避免关键信息恰好落在块边界而被截断。
5. 向量化与向量存储
向量化负责把文本块转换为向量。这里封装一个 Embedding 客户端：
from openai import OpenAI class EmbeddingClient: def init(self, api_key: str, base_url: str, model: str = "text-embedding-3-small"): self.client = OpenAI(api_key=api_key, base_url=base_url) self.model = model def embed(self, texts: list[str]) -> list[list[float]]: """批量把文本编码为向量。""" resp = self.client.embeddings.create(model=self.model, input=texts) return [item.embedding for item in resp.data]向量存储使用简单的内存实现，支持添加和检索：
import numpy as np class VectorStore: def init(self): self.vectors = [] # 向量列表 self.metadata = [] # 每个向量对应的文本块和来源信息 def add(self, vectors: list[list[float]], metadatas: list[dict]): """批量添加向量及其元数据。""" self.vectors.extend(vectors) self.metadata.extend(metadatas) def search(self, query_vector: list[float], top_k: int = 3) -> list[dict]: """ 余弦相似度检索，返回最相似的 top_k 个结果。 每个结果包含 text、score 和 source 字段。 """ if not self.vectors: return [] q = np.array(query_vector) mat = np.array(self.vectors) # 余弦相似度 = 点积 / (模长乘积) scores = (mat @ q) / (np.linalg.norm(mat, axis=1) * np.linalg.norm(q) + 1e-9) top_indices = np.argsort(scores)[::-1][:top_k] results = [] for idx in top_indices: results.append({ "text": self.metadata[idx]["text"], "score": float(scores[idx]), "source": self.metadata[idx].get("source", ""), "chunk_index": self.metadata[idx].get("chunk_index", 0), }) return results这里使用余弦相似度衡量相关性，分数越高表示越相关。
6. 索引构建：把文档写入向量库
把加载、切分、向量化、存储串起来，构建索引：
def build_index(file_path: str, embed_client: EmbeddingClient, store: VectorStore): """读取文档、切分、向量化并写入向量存储。""" text = load_document(file_path) chunks = split_text(text) vectors = embed_client.embed(chunks) metadatas = [ { "text": chunk, "source": file_path, "chunk_index": i, } for i, chunk in enumerate(chunks) ] store.add(vectors, metadatas) print(f"已索引 {len(chunks)} 个文本块")这一步完成后，向量库就具备了检索能力。
7. 检索与提示词组装
用户提问时，先把问题向量化，再从向量库检索相关块：
def retrieve(query: str, embed_client: EmbeddingClient, store: VectorStore, top_k: int = 3) -> list[dict]: """检索与问题最相关的文本块。""" query_vector = embed_client.embed([query])[0] return store.search(query_vector, top_k=top_k)把检索结果组装成带上下文的 Prompt：
def build_prompt(query: str, retrieved: list[dict]) -> str: """把检索结果和用户问题组装成 Prompt。""" context = "\n\n".join( f"[来源 {i+1}] {item['text']}" for i, item in enumerate(retrieved) ) prompt = f"""请根据以下参考资料回答用户问题。如果资料中没有相关信息，请明确说明“资料中未找到相关内容”，不要编造。 参考资料： {context} 用户问题：{query} 请给出准确、简洁的回答，并在回答末尾列出引用的来源编号。""" return prompt提示词中明确要求模型引用来源编号，这是实现可溯源回答的关键。
8. 生成回答与溯源输出
调用对话模型生成回答，并把来源信息一并返回：
class ChatClient: def __init__(self, api_key: str, base_url: str, model: str = "gpt-4o-mini"): self.client = OpenAI(api_key=api_key, base_url=base_url) self.model = model def generate(self, prompt: str) -> str: resp = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.2, ) return resp.choices[0].message.content def ask(query: str, embed_client: EmbeddingClient, store: VectorStore, chat_client: ChatClient, top_k: int = 3) -> dict: """完整的 RAG 问答入口，返回回答和溯源信息。""" retrieved = retrieve(query, embed_client, store, top_k) prompt = build_prompt(query, retrieved) answer = chat_client.generate(prompt) return { "answer": answer, "sources": [ { "text": item["text"], "score": item["score"], "source": item["source"], "chunk_index": item["chunk_index"], } for item in retrieved ], }返回结果中同时包含回答和来源块，前端可以据此展示引用来源，实现可溯源。
9. 完整实战：跑通一个问答示例
下面把以上模块串起来，用一个示例文档跑通全流程。假设有一份产品说明文档product.md：
# 智能音箱 X1 使用说明 基本功能 智能音箱 X1 支持语音播放音乐、设置闹钟、查询天气和控制智能家居设备。 连接方式 首次使用请下载 App，并按照提示连接 Wi-Fi。设备支持 2.4G 和 5G 双频段。 常见问题 如果音箱无法唤醒，请检查电源和网络连接，并确认麦克风未被静音。主程序如下：
def main(): api_key = "your-api-key" base_url = "https://api.openai.com/v1" embed_client = EmbeddingClient(api_key, base_url) chat_client = ChatClient(api_key, base_url) store = VectorStore() 1. 构建索引 build_index("product.md", embed_client, store) 2. 提问 query = "音箱无法唤醒怎么办？" result = ask(query, embed_client, store, chat_client, top_k=2) 3. 输出回答 print("回答：", result["answer"]) print("\n溯源信息：") for i, src in enumerate(result["sources"], 1): print(f" [{i}] 来源: {src['source']} 第{src['chunk_index']}块 相似度: {src['score']:.4f}") print(f" 内容: {src['text'][:80]}...") if name == "main": main()运行后，系统会先索引文档，再根据问题检索相关块，最终生成带来源编号的回答。
10. 可溯源机制的设计要点
可溯源是本文 RAG 的核心目标，设计上包含三个层面：
- 块级溯源：每个文本块都记录来源文件和块序号，检索结果天然携带来源。
- 提示词约束：在 Prompt 中要求模型引用来源编号，从生成侧约束回答可追溯。
- 结果透出：问答接口同时返回回答和来源块，前端可展示引用卡片。
此外，相似度分数可以作为溯源可信度的参考：分数越高，说明该块与问题越相关，回答依据越可靠。
11. 优化方向与进阶建议
手写版本跑通后，可以从以下方向继续优化：
- 切分策略：引入语义切分或按标题结构切分，提升块质量。
- 混合检索：结合关键词检索（BM25）和向量检索，兼顾精确匹配和语义匹配。
- 重排序：在检索后增加重排序模型，进一步提升 top_k 结果的相关性。
- 持久化存储：把向量库落盘（如使用 SQLite 或 FAISS），避免每次重启重建索引。
- 多文档支持：扩展加载器，支持 PDF、Word、网页等格式。
这些优化都可以在本文的模块化结构上逐步叠加，而无需推翻重写。
12. 总结
本文从零手写了一个可运行的 RAG 系统，覆盖文档加载、文本切分、向量化、向量存储、检索、提示词组装、生成和溯源全流程。核心收获有三点：
- 理解 RAG 的完整数据流：文档到块、块到向量、向量到检索、检索到生成。
- 掌握可溯源的实现方法：块级元数据 + 提示词约束 + 结果透出。
- 获得一个可扩展的模块化代码骨架，便于后续接入更复杂的优化策略。
你可以把这份代码作为基础，结合自己的业务文档和模型服务，快速搭建一套属于自己的检索增强问答系统。