1. 从“玩具”到“工程”为什么我们需要手写一个RAG最近和几个做AI应用的朋友聊天发现一个挺有意思的现象大家聊起RAG检索增强生成都头头是道知道它能解决大模型“幻觉”、知识更新慢的问题。但真到了要落地一个严肃的、对答案准确性有要求的业务场景时比如内部知识库问答、专利检索分析或者客服系统很多人第一反应还是去翻LangChain、LlamaIndex这类框架的文档试图用“搭积木”的方式快速拼出一个系统。结果往往是Demo跑得飞快一上真实数据就各种“翻车”——检索不准、回答冗长、关键信息丢失最要命的是当用户问“这个结论是哪里来的”时系统根本给不出一个清晰、可追溯的答案。这其实就是“玩具级”RAG和“工程级”RAG的核心区别。框架降低了入门门槛但也隐藏了太多细节。它像一辆组装好的汽车你踩油门就能走但你不清楚发动机的工况、变速箱的换挡逻辑一旦在复杂路况下抛锚你连从哪里开始排查都不知道。而“手写一个RAG”并不意味着我们要从零发明轮子而是像资深机械师一样亲手拆解、组装、调试每一个核心部件真正理解数据从输入到输出究竟经历了什么。所以这篇内容我想和你一起抛开那些厚重的框架用最直接的代码和设计思路搭建一个可溯源、高可控的RAG系统。我们将重点关注“为什么”要这么做而不仅仅是“怎么做”。你会看到一个可靠的RAG系统远不止是“向量检索LLM”那么简单它涉及到知识切片的艺术、多路召回的策略、重排序的智慧以及最终让答案“有据可查”的溯源机制。这不仅是技术实现更是一种工程思维的训练。2. 核心组件拆解一个可溯源RAG的四大支柱在开始写代码之前我们必须先画好蓝图。一个完整的、可溯源的RAG系统可以抽象为四个紧密耦合的核心阶段我把它称为“四大支柱”。理解每一根支柱的职责和它们之间的数据流是后续一切工作的基础。2.1 知识切片从“文档”到“知识片段”的炼金术这是整个流程的起点也是最容易被低估的环节。很多人以为切片就是把文档按固定长度比如512个token切碎然后扔进向量数据库。这会导致灾难性的后果检索出来的片段可能是一个不完整的句子、一个没有上下文的表格或者一个被腰斩的核心概念。切片的核心目标是创造出既能被独立理解又包含足够上下文信息的“知识单元”。这需要根据文档类型进行策略设计基于语义的切片这是最理想的方式。对于格式良好的Markdown、HTML或结构化的技术文档我们可以利用其标题层级H1, H2, H3进行切片。一个H2标题下的所有内容直到下一个H2出现可以作为一个知识单元。这保证了语义的完整性。基于固定长度的滑动窗口切片对于纯文本或无结构文档这是退而求其次的选择。关键技巧在于使用重叠窗口。例如设置片段长度为1000字符重叠度为200字符。这样能确保一个概念如果恰好落在两个片段的边界依然有很高的概率被完整检索到避免信息割裂。混合切片策略实战中我们常常需要混合使用。例如先按标题进行粗切对于过长的章节再使用滑动窗口进行细切。一个必须考虑的细节元数据附着。在切片时我们必须为每一个片段chunk记录丰富的元数据metadata这是未来实现可溯源的基石。元数据至少应包括doc_id: 原始文档的唯一标识。chunk_id: 当前片段的唯一标识。source: 原始文档的路径或URL。title: 所属章节的标题。start_index/end_index: 该片段在原文中的起止字符位置。这样当我们最终给出答案时才能精确地告诉用户“这个信息来源于《XX项目设计文档V2.1》的‘第三章 系统架构’部分具体位置在第2050到第2180字符之间。”2.2 向量化与检索让机器“理解”问题切片完成后我们需要将这些文本片段转换为机器可以“理解”和“比较”的形式——向量或称嵌入Embedding。这个过程的核心是选择一个合适的嵌入模型。模型选型考量过去我们可能默认使用OpenAI的text-embedding-ada-002但现在有了更多优秀的选择。例如SigLIP这类开源模型在图文多模态和纯文本任务上表现都相当出色且可以本地部署避免了网络延迟和API费用。选择时我们需要在MTEB等基准测试中关注模型在“检索”任务上的表现而不仅仅是通用语义相似度。检索的“多路召回”策略这是提升召回率Recall的关键。单一向量检索语义检索可能漏掉那些表述不同但核心关键词相同的文档。因此成熟的系统会采用混合检索语义检索Dense Retrieval使用向量数据库如Milvus, Pinecone, PGVector进行近似最近邻搜索。它擅长理解“意图”比如把“如何开车”和“驾驶教程”关联起来。关键词检索Sparse Retrieval使用BM25等算法。它擅长精确匹配“关键词”对于术语、代码、产品型号等精确信息的召回无可替代。例如查询“Qwen2-7B-Instruct模型的上下文长度”BM25能精准命中包含这些确切词汇的片段。我们的系统会并行执行这两路检索各自返回Top K个候选片段比如向量检索返回10个BM25返回10个形成一个更大的候选池20个。这大大增加了找到正确答案的几率。2.3 重排序从“相关”到“最相关”的精选多路召回给我们带来了数量可观的候选片段但它们的质量参差不齐顺序也不一定最优。直接把这些片段全部塞给大模型不仅会消耗大量token还可能让模型被无关信息干扰产生“幻觉”。重排序Re-ranking的作用就是充当一个“精炼官”。它使用一个更精细、但通常也更耗资源的模型专门训练用于判断“query-document”相关性的交叉编码器模型如bge-reranker对这20个候选片段进行重新打分和排序。这个过程可以理解为向量检索和BM25做了粗筛找到了“可能相关”的文档而重排序模型则进行精读判断哪一个片段“最直接、最有用”于回答当前问题。最终我们只选取重排序后得分最高的前N个比如3-5个片段作为上下文送给大模型。这显著提升了上下文的信噪比和答案的质量。2.4 生成与溯源给出“有据可依”的答案这是最后一步也是直接面向用户的一步。我们将精心筛选出的3-5个知识片段连同用户的问题一起构造提示词Prompt发送给大模型如Qwen、GPT等要求其生成答案。可溯源性的实现就体现在这里。我们的Prompt需要明确指令模型严格基于提供的上下文生成答案。如果上下文信息不足请坦诚回答“不知道”切勿杜撰。在答案中以引用的形式注明信息来源。例如模型生成的答案可能是“Qwen2-7B-Instruct模型的上下文长度为128K tokens。[来源模型卡文档 章节‘关键参数’ 片段ID: doc_001_chunk_005]”在后台我们需要建立一个从片段ID到元数据的映射。当呈现答案给用户时系统可以将[来源...]这部分渲染成一个可点击的链接或悬浮提示直接展示原文片段、所属文档和位置。这就是一个完整的、可信的答案生成与溯源流程。3. 实战构建用Python一步步实现核心流水线理论清晰了我们开始动手。这里我会用Python展示最核心的代码逻辑并解释关键设计决策。我们假设使用PGVectorPostgreSQL的向量扩展作为向量数据库因为它结合了关系数据库的成熟生态和向量检索能力非常适合需要复杂元数据过滤的场景。3.1 环境准备与依赖安装首先确保你的环境已经准备好。我们需要以下核心库pip install langchain # 我们只使用其文本分割器等基础工具不依赖其完整框架 pip install sentence-transformers # 用于本地嵌入模型如all-MiniLM-L6-v2 # 或者使用更先进的模型如 # pip install transformers pip install rank-bm25 # BM25算法实现 pip install psycopg2-binary pgvector # PostgreSQL连接和向量支持 pip install openai # 如果需要使用OpenAI的嵌入或生成模型 pip install tiktoken # 用于精确的token计数切片时很重要数据库方面你需要一个运行中的PostgreSQL建议12以上版本并安装pgvector扩展。3.2 知识切片与向量化入库我们来实现一个兼顾语义和重叠的切片器并完成向量化存储。import os from typing import List, Dict, Any from sentence_transformers import SentenceTransformer import psycopg2 from psycopg2.extras import execute_values import tiktoken class KnowledgeChunker: def __init__(self, chunk_size1000, chunk_overlap200): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.tokenizer tiktoken.get_encoding(cl100k_base) # 用于准确计算token长度 def chunk_document(self, text: str, doc_metadata: Dict) - List[Dict[str, Any]]: 将一篇文档切分成片段并附加元数据。 chunks [] # 简化版这里使用简单的滑动窗口。实际应优先按标题切分。 words text.split() start 0 while start len(words): end start self.chunk_size chunk_text .join(words[start:end]) # 计算token长度确保不超过LLM上下文限制为后续的上下文预留空间 token_len len(self.tokenizer.encode(chunk_text)) if token_len 800: # 预留buffer # 可以在这里实现更精细的动态调整 pass chunk_id f{doc_metadata[doc_id]}_chunk_{len(chunks)} chunk_meta { **doc_metadata, chunk_id: chunk_id, start_word_idx: start, end_word_idx: end, token_length: token_len, } chunks.append({ text: chunk_text, metadata: chunk_meta }) start (self.chunk_size - self.chunk_overlap) # 滑动窗口实现重叠 return chunks class VectorStoreManager: def __init__(self, db_conn_str, embed_model_nameall-MiniLM-L6-v2): self.conn psycopg2.connect(db_conn_str) self.embed_model SentenceTransformer(embed_model_name) self._init_db() def _init_db(self): cur self.conn.cursor() # 启用pgvector扩展 cur.execute(CREATE EXTENSION IF NOT EXISTS vector;) # 创建存储知识片段的表包含向量列和丰富的元数据列 cur.execute( CREATE TABLE IF NOT EXISTS knowledge_chunks ( id BIGSERIAL PRIMARY KEY, chunk_id TEXT UNIQUE NOT NULL, text TEXT NOT NULL, embedding vector(384), -- 维度需与模型匹配 doc_id TEXT NOT NULL, source TEXT, title TEXT, start_idx INTEGER, end_idx INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_embedding ON knowledge_chunks USING ivfflat (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS idx_doc_id ON knowledge_chunks(doc_id); ) self.conn.commit() cur.close() def store_chunks(self, chunks: List[Dict]): 存储切片及其向量。 cur self.conn.cursor() texts [c[text] for c in chunks] embeddings self.embed_model.encode(texts).tolist() data_to_insert [] for chunk, emb in zip(chunks, embeddings): meta chunk[metadata] data_to_insert.append(( meta[chunk_id], chunk[text], emb, meta[doc_id], meta.get(source, ), meta.get(title, ), meta.get(start_word_idx, 0), meta.get(end_word_idx, 0), )) execute_values(cur, INSERT INTO knowledge_chunks (chunk_id, text, embedding, doc_id, source, title, start_idx, end_idx) VALUES %s ON CONFLICT (chunk_id) DO NOTHING; , data_to_insert) self.conn.commit() cur.close() print(f成功存储 {len(chunks)} 个知识片段。) # 使用示例 if __name__ __main__: db_conn_str dbnameragdb userpostgres passwordyour_password hostlocalhost chunker KnowledgeChunker(chunk_size500, chunk_overlap50) vs_manager VectorStoreManager(db_conn_str) sample_doc_text open(sample_tech_doc.md).read() doc_meta {doc_id: tech_doc_001, source: docs/sample_tech_doc.md, title: 系统设计指南} chunks chunker.chunk_document(sample_doc_text, doc_meta) vs_manager.store_chunks(chunks)注意这里为了清晰使用了简单的空格分词和滑动窗口。在生产环境中你需要集成更强大的文本分割器比如利用langchain.text_splitter中的RecursiveCharacterTextSplitter并优先尝试按Markdown标题进行分割。3.3 实现混合检索与重排序接下来我们实现查询端的核心混合检索器与重排序器。from rank_bm25 import BM25Okapi import numpy as np class HybridRetriever: def __init__(self, vector_store_manager, bm25_k10, vector_k10): self.vs_manager vector_store_manager self.bm25_k bm25_k self.vector_k vector_k self.bm25_index None self.chunk_texts_for_bm25 [] self.chunk_metadatas_for_bm25 [] def build_bm25_index(self): 从数据库加载所有文本构建BM25索引。适用于数据量不大或可定期更新的场景。 cur self.vs_manager.conn.cursor() cur.execute(SELECT chunk_id, text, doc_id, source, title FROM knowledge_chunks;) rows cur.fetchall() cur.close() self.chunk_texts_for_bm25 [] self.chunk_metadatas_for_bm25 [] for row in rows: chunk_id, text, doc_id, source, title row self.chunk_texts_for_bm25.append(text) self.chunk_metadatas_for_bm25.append({ chunk_id: chunk_id, doc_id: doc_id, source: source, title: title, text: text # 保留原文用于后续展示 }) # 使用简单的分词生产环境建议使用更好的分词器 tokenized_corpus [doc.split() for doc in self.chunk_texts_for_bm25] self.bm25_index BM25Okapi(tokenized_corpus) print(fBM25索引构建完成共 {len(self.chunk_texts_for_bm25)} 个文档。) def retrieve(self, query: str, top_n: int 5) - List[Dict]: 执行混合检索返回初步的候选片段列表。 candidates [] # 1. 向量检索 query_embedding self.vs_manager.embed_model.encode([query])[0] cur self.vs_manager.conn.cursor() cur.execute( SELECT chunk_id, text, doc_id, source, title, 1 - (embedding %s) as cosine_sim FROM knowledge_chunks ORDER BY embedding %s LIMIT %s; , (query_embedding, query_embedding, self.vector_k)) for row in cur.fetchall(): chunk_id, text, doc_id, source, title, score row candidates.append({ chunk_id: chunk_id, text: text, metadata: {doc_id: doc_id, source: source, title: title}, score: float(score), retriever: vector }) cur.close() # 2. BM25检索 if self.bm25_index: tokenized_query query.split() bm25_scores self.bm25_index.get_scores(tokenized_query) top_bm25_indices np.argsort(bm25_scores)[::-1][:self.bm25_k] for idx in top_bm25_indices: # 避免重复添加根据chunk_id去重 existing_ids {c[chunk_id] for c in candidates} meta self.chunk_metadatas_for_bm25[idx] if meta[chunk_id] not in existing_ids: candidates.append({ chunk_id: meta[chunk_id], text: meta[text], metadata: {doc_id: meta[doc_id], source: meta[source], title: meta[title]}, score: float(bm25_scores[idx]), retriever: bm25 }) # 3. 按原始分数简单合并后续由重排序优化 # 这里可以先按分数排序但更优的做法是交给重排序模型 return candidates[:top_n*2] # 返回较多候选供重排序筛选 class Reranker: def __init__(self, model_nameBAAI/bge-reranker-base): # 这里使用一个轻量级的交叉编码器模型进行重排序 # 实际部署可能需要加载本地模型 from transformers import AutoModelForSequenceClassification, AutoTokenizer self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval() import torch self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.model.to(self.device) def rerank(self, query: str, candidates: List[Dict], top_n: int 3) - List[Dict]: 对候选片段进行重排序。 if not candidates: return [] pairs [[query, cand[text]] for cand in candidates] import torch with torch.no_grad(): inputs self.tokenizer(pairs, paddingTrue, truncationTrue, return_tensorspt, max_length512).to(self.device) scores self.model(**inputs).logits.squeeze(dim-1).cpu().numpy() for cand, score in zip(candidates, scores): cand[rerank_score] float(score) # 按重排序分数降序排列 candidates.sort(keylambda x: x[rerank_score], reverseTrue) return candidates[:top_n]这段代码实现了检索的核心逻辑。HybridRetriever并行执行向量检索和BM25检索合并结果。Reranker则使用一个预训练的交叉编码器模型对合并后的候选列表进行精细打分筛选出最相关的几个片段。3.4 集成大模型生成与溯源最后我们将检索到的精华上下文发送给大模型并设计Prompt使其生成带引用的答案。import openai # 示例使用OpenAI API可替换为其他本地模型调用 class AnswerGenerator: def __init__(self, llm_api_key, llm_modelgpt-3.5-turbo): openai.api_key llm_api_key self.llm_model llm_model def generate_answer(self, query: str, top_chunks: List[Dict]) - Dict[str, Any]: 根据检索到的片段生成答案并附带溯源信息。 if not top_chunks: return { answer: 根据现有知识库我无法回答这个问题。, sources: [] } # 1. 构建上下文和溯源映射 context_parts [] source_mapping {} # chunk_id - metadata for i, chunk in enumerate(top_chunks): chunk_id chunk[chunk_id] source_mapping[chunk_id] chunk[metadata] # 在上下文中加入片段标识便于模型引用 context_parts.append(f[片段{i1}: {chunk[text]}]) context \n\n.join(context_parts) # 2. 设计系统Prompt明确要求引用和诚实 system_prompt 你是一个专业的问答助手将严格根据用户提供的上下文信息来回答问题。 请遵循以下规则 1. 答案必须完全基于提供的上下文。如果上下文没有足够信息请直接说“根据提供的资料我无法回答此问题”。 2. 在答案中对于来自上下文的具体信息请使用方括号注明来源格式为[来源片段X]其中X是上下文中的片段编号。 3. 保持答案简洁、准确。 上下文如下 user_prompt f问题{query} # 3. 调用LLM try: response openai.ChatCompletion.create( modelself.llm_model, messages[ {role: system, content: system_prompt context}, {role: user, content: user_prompt} ], temperature0.1 # 低温度使输出更确定、更忠于上下文 ) answer response.choices[0].message.content except Exception as e: answer f生成答案时出错{e} # 4. 解析答案中的引用并关联到具体的元数据 # 这里简化处理实际可以写更复杂的正则表达式来提取 [来源片段X] import re source_refs re.findall(r\[来源片段(\d)\], answer) used_sources [] for ref in source_refs: idx int(ref) - 1 if 0 idx len(top_chunks): chunk_id top_chunks[idx][chunk_id] used_sources.append(source_mapping.get(chunk_id, {})) return { answer: answer, sources: used_sources, # 实际使用的来源元数据 all_retrieved_chunks: top_chunks # 返回所有检索到的片段用于调试 } # 完整的查询流程 def query_pipeline(query: str, retriever: HybridRetriever, reranker: Reranker, generator: AnswerGenerator): print(f用户查询: {query}) # 1. 混合检索 candidates retriever.retrieve(query, top_n10) print(f混合检索到 {len(candidates)} 个候选片段。) # 2. 重排序 top_chunks reranker.rerank(query, candidates, top_n3) print(f重排序后选取 top {len(top_chunks)} 个片段。) # 3. 生成答案 result generator.generate_answer(query, top_chunks) # 4. 输出结果 print(f\n--- 答案 ---\n{result[answer]}\n) if result[sources]: print(--- 溯源信息 ---) for src in result[sources]: print(f 文档: {src.get(title, N/A)}, 来源: {src.get(source, N/A)}) return result这个AnswerGenerator的关键在于Prompt工程。我们通过系统指令明确要求模型基于上下文、诚实回答并注明引用。返回的结果中包含了答案和用到的源数据前端可以据此渲染出可点击的引用链接。4. 超越基础工程化落地的关键考量与优化一个能跑通的Demo只是起点。要让这个RAG系统在生产环境中可靠运行我们还需要考虑很多工程细节。这部分才是区分“玩具”和“工具”的关键。4.1 知识切片的质量控制与迭代切片策略不是一劳永逸的。你需要建立一套评估和迭代机制。人工抽样检查定期从不同文档类型中随机抽样切片检查其语义完整性和独立性。一个片段是否是一个完整的“问答对”检索效果反馈分析历史查询日志哪些问题没找到答案是不是因为相关知识点被切碎了据此调整chunk_size和overlap甚至为不同文档类型如API文档、论文、会议纪要定制不同的切片策略。元数据增强除了基础元数据可以考虑自动提取片段的关键实体人名、地名、技术术语、摘要或问题这个片段可能回答什么问题并将其作为元数据存储。这可以为后续的检索提供更丰富的过滤和排序维度。4.2 检索阶段的性能与精度权衡索引更新策略知识库不是静态的。如何增量更新对于PGVector你可以直接插入新向量。但BM25索引需要重建。对于百万级以下文档可以定期如每天全量重建BM25索引。对于更大规模需要考虑增量更新算法或切换到支持增量更新的检索引擎如Elasticsearch。多路召回融合策略我们之前简单合并了向量和BM25的结果。更高级的做法是加权融合。例如给向量检索和BM25检索的结果分别赋予权重然后按加权总分排序。权重可以通过一个小的验证集进行调优。元数据过滤在检索前或检索后利用元数据进行过滤至关重要。例如用户可能指定“只在去年的项目报告里搜索”那么就需要在SQL查询中增加WHERE doc_source LIKE %2023_report%这样的过滤条件。PGVector支持在向量检索的同时进行复杂的元数据过滤这是它的巨大优势。4.3 重排序模型的选择与成本模型选型bge-reranker是一个很好的起点。但对于中文场景可能需要选择bge-reranker-zh。对于延迟极其敏感的场景可能需要在效果和速度之间权衡甚至使用更轻量的模型或基于传统特征如词频、共现的排序方法。成本与缓存重排序模型调用是计算密集型的。对于热门或重复查询可以实现一个缓存层将(query, top_chunk_ids)映射到重排序后的结果避免重复计算。两阶段重排序为了平衡精度和延迟可以采用两阶段策略第一阶段用一个轻快模型对大量候选如50个进行粗排第二阶段再用强大模型对粗排后的Top候选如10个进行精排。4.4 可溯源性的用户体验设计溯源不能停留在后台数据。在前端呈现上需要精心设计高亮显示在答案中将被引用的原文部分高亮显示。侧边栏或弹窗点击引用标记时在侧边栏或弹窗中展示完整的源文本片段并指示在原文中的位置。置信度评分除了引用还可以展示系统对这个答案的置信度例如基于重排序分数的归一化值让用户对答案的可靠性有直观感受。上下文展示不仅展示被引用的片段也可以选择性地展示其他高相关但未被引用的片段让用户了解检索的全貌增加系统透明度。4.5 系统的监控与评估没有度量就无法改进。必须为你的RAG系统建立监控指标检索指标召回率RecallK、平均精度MAP。需要一个小型的标注测试集。生成指标答案的忠实度是否歪曲原文、信息完整性、引用准确率。可以通过人工评估或利用更强大的LLM作为裁判进行自动评估。性能指标端到端延迟、各阶段检索、重排序、生成耗时、Token消耗。业务指标用户满意度评分、问题解决率、人工接管率。定期分析这些指标才能发现瓶颈持续优化切片策略、检索模型和Prompt设计。手写一个RAG系统的过程是一个深度理解信息检索、表示学习和提示工程如何协同工作的绝佳机会。它迫使你思考每一个环节的取舍而不仅仅是调用一个from langchain.vectorstores import Chroma。当你亲手构建了这条流水线并看着它从混乱的文档中精准地找出答案并标明出处时那种对系统的掌控感和对问题本质的理解是使用任何高级框架都无法替代的。这个系统可能一开始不如框架功能花哨但它的每一行代码你都了如指掌每一个环节都可以按需定制和优化这才是工程实践中最宝贵的资产。