1. 项目缘起为什么从零开始做RAG最近几年AI应用开发的热度居高不下尤其是RAG检索增强生成技术几乎成了大模型落地的“标配”。网上教程很多框架也层出不穷像LangChain、LlamaIndex、Dify这些号称能让你“三行代码”搞定一个智能问答。但说实话跟着这些“快餐式”教程走一遍你大概率还是懵的向量化到底在干什么为什么我的答案总是不准所谓的“工程化”到底包含了哪些脏活累活这正是我决定抛开所有现成框架从零手搓一个RAG问答应用的原因。我想弄明白在那些高级抽象的API和封装良好的Pipeline背后每一个环节到底是如何运作的又会遇到哪些“坑”。这个过程远比调用pip install然后复制粘贴代码要痛苦但也远比那样做要收获巨大。它让我对RAG的认知从一个模糊的“检索生成”概念变成了一个由数据准备、向量化、检索、重排序、生成等多个精密齿轮咬合而成的系统工程。如果你也厌倦了当“调包侠”想真正理解RAG的筋骨或者你是一名开发者无论是前端、Java还是其他背景正考虑转向AI应用开发觉得前景迷茫不知从何下手那么这篇从零开始的实践记录或许能给你一些实实在在的参考。我们不止要做出一个能跑的Demo更要搞清楚每一个选择背后的“为什么”以及那些只有踩过坑才知道的“怎么办”。2. 核心架构拆解RAG不是“检索生成”那么简单很多人把RAG简单理解为“先用向量数据库搜一下再把结果喂给大模型”。这个理解没错但过于粗糙无法指导工程实践。一个健壮、可用的RAG系统其内部是一个分层级的架构。结合业界共识和我的实践可以将其梳理为以下几个核心层级2.1 数据层一切始于“知识”的预处理这是整个RAG系统的基石也是最容易被轻视的环节。你的原始数据PDF、Word、网页、数据库不是直接就能用的。知识切片Chunking这是第一个技术难点。怎么切按固定长度按段落按语义这里没有银弹。固定长度切片最简单比如每256个token切一段。但问题很明显很可能把一个完整的句子或概念从中间切断导致检索到的片段信息不完整。按段落/标题切片利用文档的天然结构如Markdown的# PDF的段落。这比固定长度好但依赖文档本身格式良好。语义切片更高级的方法目标是让每一个切片在语义上尽可能独立、完整。这通常需要先用一个轻量模型如BERT计算句子的嵌入向量然后根据向量间的相似度或变化点来划分边界。实操心得对于技术文档、知识库我推荐“重叠式滑动窗口切片”。比如设置切片大小为500字符重叠部分为100字符。这样既能保证每个片段有足够上下文又能避免因切在关键位置而丢失信息是效果和复杂度的一个很好平衡。元数据附加切片时不要光存文本内容。一定要把来源文件名、章节标题、在原文中的位置页码、行号等信息作为元数据Metadata附加到每个切片上。这在后续的引用溯源和混合检索中至关重要。2.2 索引层让知识“可被检索”处理好的文本切片需要转换成一种便于快速查找的形式这就是索引。向量化Embedding这是核心中的核心。我们使用一个嵌入模型Embedding Model将文本切片转换为高维空间中的向量一组浮点数。语义相似的文本其向量在空间中的距离如余弦相似度也更近。模型选型选对模型事半功倍。对于中文text2vec、BGE系列是很好的开源选择。对于这个实践项目我选择了BGE-large-zh-v1.5它在中文语义相似度任务上表现稳健且社区活跃。为什么不直接用OpenAI的Embedding成本、延迟和隐私。自托管开源模型一次部署无限次使用数据不出私域。向量维度比如BGE模型输出1024维的向量。维度越高表征能力越强但存储和计算成本也越高。需要权衡。向量数据库Vector Database用于存储和高效检索这些向量。市面上有Pinecone、Weaviate等云服务也有Chroma、Qdrant、Milvus等可以自部署的开源方案。我的选择为了彻底掌控我选择了本地部署的ChromaDB。它轻量、简单Python集成友好对于中小规模知识库完全够用。你当然可以选更强大的Milvus但Chroma能让我们的注意力更集中在RAG流程本身而非数据库的运维上。索引算法Chroma默认使用HNSW近似最近邻搜索算法来构建索引。你不需要手动实现但要知道它的存在它通过构建一个图结构来加速检索在精度和速度之间取得了很好的平衡是当前向量检索的主流算法。2.3 检索与重排层找到“最相关”的知识当用户提问时系统需要从海量切片中找到最相关的几个。检索Retrieval / Recall将用户问题也向量化然后在向量数据库中搜索与之最相似的N个文本切片比如Top 10。这就是经典的“向量检索”或“语义检索”。多路召回Hybrid Search单一向量检索可能不够。例如用户问“2023年公司的营收是多少”其中“2023”、“营收”是关键词。这时可以结合关键词检索如BM25算法。向量检索负责语义匹配“营收”可能匹配“收入”、“销售额”关键词检索负责精确匹配。将两路召回的结果合并能显著提升召回率。这就是“混合检索”的工程价值。重排序Re-ranking从向量库召回的可能有10个片段但它们与问题的相关度排序不一定准确。重排序就是一个“精排”过程使用一个更精细的通常是交叉编码器模型对“问题-候选片段”对进行打分重新排序。为什么需要向量检索基于“向量相似度”而重排序模型能更好地理解问题和答案之间的“相关性”尤其是对于需要复杂推理或存在语义鸿沟的情况。例如问题“如何解决内存泄漏”向量检索可能返回一堆讲“内存管理”的片段而重排序模型能识别出其中真正在讲“泄漏检测工具使用”的片段更相关。模型选择可以使用专门的重排序模型如BGE-Reranker。也可以直接用一个大语言模型LLM来做让它给相关性打分但成本较高。在本项目中为了流程完整我引入了BGE-Reranker-large模型对Top 10的召回结果进行重排选出最终的Top 3作为上下文。2.4 生成层基于知识的“安全”作答这是最后一步也是用户直接感知的一步。提示工程Prompt Engineering将重排序得到的最相关文本片段上下文连同用户的问题按照一定的模板组织起来形成给大模型的“提示词”Prompt。 一个经典的模板如下请基于以下上下文信息回答问题。如果上下文信息不足以回答问题请直接回答“根据已知信息无法回答该问题”不要编造信息。 上下文 {context} 问题{question} 请给出答案设计要点这个模板明确了角色、任务边界防止幻觉、输入格式。“不要编造信息”这个指令对于RAG的可靠性至关重要。大语言模型LLM调用将组装好的Prompt发送给LLM得到最终答案。这里可以选择云端API如GPT-4、文心一言、通义千问或本地部署的模型如Qwen、ChatGLM。我的选择为了项目的完整性和可控性我选择了在本地部署Qwen2.5-7B-Instruct模型。它性能足够对中文友好并且完全免费、隐私安全。使用vLLM或llama.cpp这样的推理框架可以高效地部署和服务化这个模型。事实性GroundingRAG的核心优势就在于模型的回答被“锚定”在你提供的上下文里。你需要确保模型严格基于上下文生成并在答案中引用片段的来源元数据例如“根据《2023年度报告》第5页…”这能极大提升可信度。3. 从零到一的实战手把手搭建核心流水线理论讲完了我们进入实战环节。我会用代码片段展示核心步骤并解释关键参数和设计理由。3.1 环境准备与依赖安装首先创建一个干净的Python环境推荐3.9并安装核心库。# 创建虚拟环境 python -m venv rag_env source rag_env/bin/activate # Linux/Mac # rag_env\Scripts\activate # Windows # 安装核心依赖 pip install chromadb # 向量数据库 pip install sentence-transformers # 用于BGE嵌入模型 pip install pypdf # 用于解析PDF比PyPDF2更活跃 pip install langdetect # 用于语言检测可选 pip install torch # 深度学习框架sentence-transformers依赖 # 如果需要本地LLM例如用ollama pip install ollama # 或者用vLLM部署Qwen # pip install vllm3.2 第一步文档加载与智能切片假设我们有一个knowledge_base.pdf文件。我们来实现加载和切片。import PyPDF2 from typing import List, Dict import re class DocumentProcessor: def __init__(self, chunk_size: int 500, chunk_overlap: int 100): self.chunk_size chunk_size # 每个切片的目标字符数 self.chunk_overlap chunk_overlap # 切片间的重叠字符数 def load_pdf(self, file_path: str) - str: 加载PDF文件提取所有文本 text with open(file_path, rb) as file: reader PyPDF2.PdfReader(file) for page_num, page in enumerate(reader.pages): page_text page.extract_text() if page_text: text f[Page {page_num1}] {page_text}\n # 附加页码元数据 return text def smart_chunking(self, text: str, source: str) - List[Dict]: 使用重叠滑动窗口进行切片。 返回一个字典列表每个字典包含‘text‘和‘metadata‘。 # 基础清洗去除多余空白字符 text re.sub(r\s, , text).strip() chunks [] start 0 text_length len(text) while start text_length: # 计算切片结束位置 end start self.chunk_size # 如果没到文本末尾尝试在句末、分号或逗号处截断避免切碎句子 if end text_length: # 查找最近的句子边界 for break_point in range(end, start, -1): if text[break_point] in .。!?;: end break_point 1 # 包含标点 break # 如果没找到标点就在空格处截断 else: for break_point in range(end, start, -1): if text[break_point] : end break_point break else: end text_length chunk_text text[start:end].strip() if chunk_text: # 忽略空切片 metadata { source: source, start_char: start, end_char: end, } # 尝试提取切片内的标题作为更细粒度的元数据 title_match re.search(r^(#{1,6}\s*.)$, chunk_text, re.MULTILINE) if title_match: metadata[section] title_match.group(1) chunks.append({ text: chunk_text, metadata: metadata }) # 移动窗口考虑重叠 start end - self.chunk_overlap return chunks # 使用示例 processor DocumentProcessor(chunk_size500, chunk_overlap50) full_text processor.load_pdf(knowledge_base.pdf) doc_chunks processor.smart_chunking(full_text, sourceknowledge_base.pdf) print(f共切分出 {len(doc_chunks)} 个文本片段。)关键点解析chunk_size500, overlap50这是一个经验值。对于技术文档500字符能容纳一个中等长度的概念说明。50字符的重叠确保了上下文连贯。智能断句代码中尝试在句末标点处截断这是为了保持语义完整性比粗暴地按固定字符数切割效果好得多。元数据丰富化除了来源和位置我们还尝试提取Markdown风格的标题作为section元数据。这为后续的混合检索提供了可能你不仅可以按语义搜还可以按“章节标题”过滤。3.3 第二步向量化与索引构建接下来我们将文本片段转换为向量并存入ChromaDB。from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings class VectorIndexer: def __init__(self, embedding_model_name: str BAAI/bge-large-zh-v1.5): # 加载嵌入模型 self.embedding_model SentenceTransformer(embedding_model_name) # 初始化Chroma客户端持久化到本地目录 self.client chromadb.PersistentClient(path./chroma_db) # 获取或创建集合类似数据库的表 self.collection self.client.get_or_create_collection( nameknowledge_base, metadata{hnsw:space: cosine} # 使用余弦相似度 ) def create_index(self, chunks: List[Dict]): 将文本切片向量化并存入数据库 texts [chunk[text] for chunk in chunks] metadatas [chunk[metadata] for chunk in chunks] ids [fchunk_{i} for i in range(len(texts))] # 为每个切片生成唯一ID # 批量生成向量嵌入 print(正在生成向量嵌入...) embeddings self.embedding_model.encode(texts, normalize_embeddingsTrue).tolist() print(f已生成 {len(embeddings)} 个向量维度为 {len(embeddings[0])}。) # 批量添加到集合 self.collection.add( embeddingsembeddings, documentstexts, metadatasmetadatas, idsids ) print(向量索引构建完成。) # 使用示例 indexer VectorIndexer() indexer.create_index(doc_chunks)关键点解析normalize_embeddingsTrue将向量归一化为单位长度。这样向量间的点积就等于余弦相似度是语义相似度计算的常用方法。hnsw:space: “cosine”指定Chroma使用余弦相似度作为距离度量。对于归一化后的向量余弦相似度范围是[-1,1]值越大越相似。批处理encode方法支持批量文本输入一次性生成所有向量效率远高于循环单条处理。持久化PersistentClient将数据和索引保存在本地./chroma_db目录下次启动无需重新索引。3.4 第三步实现混合检索与重排序当用户提问时我们需要执行检索流程。# 首先需要安装重排序模型库如果使用BGE Reranker # pip install FlagEmbedding from FlagEmbedding import FlagReranker class Retriever: def __init__(self, indexer: VectorIndexer, reranker_model_name: str BAAI/bge-reranker-large): self.collection indexer.collection self.embedding_model indexer.embedding_model # 初始化重排序模型 self.reranker FlagReranker(reranker_model_name, use_fp16True) # 使用半精度加速 def hybrid_retrieve(self, query: str, top_k: int 10, keyword_weight: float 0.3): 混合检索结合向量检索和简单关键词匹配。 keyword_weight: 关键词检索结果的权重用于融合分数。 # 1. 向量检索 query_embedding self.embedding_model.encode([query], normalize_embeddingsTrue).tolist()[0] vector_results self.collection.query( query_embeddings[query_embedding], n_resultstop_k * 2, # 多取一些供后续融合和重排序筛选 ) # 简单模拟关键词检索实际可用whoosh、elasticsearch或BM25库 # 这里仅作演示在实际项目中应替换为真正的关键词检索引擎 all_docs self.collection.get()[documents] all_metadatas self.collection.get()[metadatas] keyword_scores [] for i, doc in enumerate(all_docs): # 简单计算查询词在文档中出现的频率作为分数 score sum([doc.count(word) for word in query.split() if len(word) 1]) keyword_scores.append((i, score)) keyword_scores.sort(keylambda x: x[1], reverseTrue) keyword_top_indices [idx for idx, _ in keyword_scores[:top_k]] # 2. 结果融合简化版 # 为向量检索结果赋予初始分数这里用1 - 距离因为chroma返回的是距离 fused_results {} for i, (doc, meta, dist) in enumerate(zip(vector_results[documents][0], vector_results[metadatas][0], vector_results[distances][0])): vector_score 1 - dist # 将距离转换为相似度分数 # 检查该文档是否也在关键词检索结果中 keyword_boost 1.0 if i in keyword_top_indices else 0.0 combined_score vector_score keyword_weight * keyword_boost fused_results[i] { document: doc, metadata: meta, score: combined_score, from_vector: True } # 按融合分数排序 sorted_fused sorted(fused_results.items(), keylambda x: x[1][score], reverseTrue) candidates [item[1] for item in sorted_fused[:top_k * 2]] # 取前 top_k*2 作为重排序候选 return candidates def rerank(self, query: str, candidates: List[Dict], top_n: int 3): 使用重排序模型对候选文档进行精排 if not candidates: return [] # 准备重排序模型需要的输入对 (query, document) pairs [[query, cand[document]] for cand in candidates] # 批量计算相关性分数 scores self.reranker.compute_score(pairs) # 将分数附加到候选文档上 for cand, score in zip(candidates, scores): cand[rerank_score] score # 按重排序分数降序排列 candidates.sort(keylambda x: x[rerank_score], reverseTrue) return candidates[:top_n] def retrieve(self, query: str): 完整的检索流程 print(f用户查询: {query}) # 1. 混合检索获取较多候选 candidates self.hybrid_retrieve(query, top_k10) print(f混合检索得到 {len(candidates)} 个候选片段。) # 2. 重排序获取最相关的少数几个 final_contexts self.rerank(query, candidates, top_n3) print(f重排序后选取 top {len(final_contexts)} 个上下文。) return final_contexts # 使用示例 retriever Retriever(indexer) contexts retriever.retrieve(公司2023年的主要营收来源是什么) for i, ctx in enumerate(contexts): print(f\n--- 上下文 {i1} (得分: {ctx[rerank_score]:.4f}) ---) print(f来源: {ctx[metadata][source]}, 位置: {ctx[metadata].get(start_char, N/A)}) print(f内容摘要: {ctx[document][:200]}...)关键点解析混合检索模拟上述代码中的关键词检索是极简模拟。在生产环境中你需要集成如rank_bm25这样的库或者使用Elasticsearch。核心思想是并行执行向量检索和关键词检索然后按一定策略如加权分数、RRF融合结果。重排序的价值注意看hybrid_retrieve返回的candidates是按初步融合分数排序的。但经过rerank模型后顺序很可能发生变化。重排序模型交叉编码器会同时编码问题和文档进行深度的交互匹配其相关性判断通常比单纯的向量相似度更精准。Top-K策略通常第一轮检索召回会取一个较大的K如50保证召回率重排序后再取一个较小的N如3-5作为最终上下文保证精确率。这被称为“召回后重排序”范式。3.5 第四步提示词构建与大模型生成最后我们将精选的上下文送给大模型让它生成答案。# 假设我们使用本地部署的Ollama服务运行了Qwen2.5模型 # 启动命令ollama run qwen2.5:7b import requests import json class AnswerGenerator: def __init__(self, llm_api_url: str http://localhost:11434/api/generate): self.llm_api_url llm_api_url def build_prompt(self, query: str, contexts: List[Dict]) - str: 构建包含上下文和指令的Prompt context_str for i, ctx in enumerate(contexts): source ctx[metadata][source] # 可以加入更多元数据如章节 context_str f[上下文片段 {i1}, 来源: {source}]:\n{ctx[document]}\n\n prompt_template 你是一个专业的问答助手。请严格根据以下提供的上下文信息来回答问题。如果上下文信息中没有包含问题答案所需的信息或者信息不充分请明确回答“根据已知信息无法回答该问题”不要编造任何信息。 上下文信息 {context} 用户问题{question} 请基于上下文信息给出准确、简洁的答案。如果答案涉及具体数据或观点请注明其来源上下文片段的编号。 return prompt_template.format(contextcontext_str.strip(), questionquery) def generate_answer(self, prompt: str) - str: 调用LLM API生成答案 payload { model: qwen2.5:7b, # 指定模型 prompt: prompt, stream: False, options: { temperature: 0.1, # 低温度减少随机性让答案更确定 top_p: 0.9, } } try: response requests.post(self.llm_api_url, jsonpayload, timeout60) response.raise_for_status() result response.json() return result.get(response, LLM返回结果为空。) except requests.exceptions.RequestException as e: return f调用LLM API时出错: {e} def answer(self, query: str, contexts: List[Dict]) - str: 完整的答案生成流程 prompt self.build_prompt(query, contexts) print(--- 构建的Prompt (前500字符) ---) print(prompt[:500] ...) print(--- 正在生成答案 ---) answer self.generate_answer(prompt) return answer # 使用示例 generator AnswerGenerator() final_answer generator.answer(公司2023年的主要营收来源是什么, contexts) print(\n 最终答案 ) print(final_answer)关键点解析Prompt设计这是控制LLM行为的关键。我们的模板强调了三点角色定义专业助手、指令约束严格基于上下文不胡编乱造、输出要求准确简洁注明来源。temperature0.1使得模型输出更确定、更忠实于上下文。上下文格式化将每个上下文片段与其元数据如来源一起格式化不仅有助于模型理解也方便在最终答案里做引用。错误处理网络请求总是可能失败必须有基本的try-except来处理超时或连接错误给用户友好的反馈。本地LLM替代方案除了Ollama你也可以使用vLLM部署并调用或者使用transformers库直接加载模型。Ollama的优势是简单易用适合快速原型验证。4. 超越DemoRAG工程化的核心挑战与优化一个能跑的Demo和一个能在生产环境服务的RAG系统之间隔着巨大的鸿沟这就是“工程化”要解决的问题。以下是几个你必须面对的挑战和优化方向。4.1 知识切片的质量最大的变量切片策略直接决定检索的上限。不好的切片会导致“信息碎片化”或“信息冗余”。动态切片不要对所有文档都用一种切片方式。对于API文档可以按函数/方法切对于长篇文章可以按章节切对于QA对可以保持一对一切片。可以训练一个分类器先判断文档类型再选择切片策略。小颗粒度大上下文一种先进实践是“小切片存储大上下文检索”。即存储时用较小的颗粒度如200字但在检索时如果相邻切片来自同一来源或章节则将它们合并后作为上下文送给LLM。这平衡了检索精度和上下文完整性。结构化信息提取在切片前先用模型如UIE、LayoutLM从文档中提取出表格、键值对、标题层级等结构化信息并将其作为元数据存储。检索时这些元数据可以作为强大的过滤条件。4.2 检索效果的评估与迭代没有度量就没有优化你怎么知道你的RAG系统变好了还是变坏了你需要一套评估体系。构建测试集收集一批真实用户可能问的问题Q并人工标注每个问题对应的标准答案A以及支撑答案的文档片段C即Ground Truth Context。这就是你的“黄金测试集”。核心评估指标检索召回率Retrieval Recall系统检索到的Top K个片段中是否包含了人工标注的标准上下文C这是衡量检索模块能力的核心。答案相关性Answer RelevanceLLM生成的答案是否直接回答了问题可以用另一个LLM如GPT-4来打分或者用基于嵌入的相似度计算如答案与标准答案的向量相似度。答案事实性/忠实度Answer Faithfulness/Groundedness答案中的每一个陈述是否都能在提供的上下文中找到依据这是防止“幻觉”的关键。同样可以用LLM判断或规则匹配。持续迭代当你更换嵌入模型、调整切片大小、启用重排序后跑一遍测试集看这些指标的变化。只有数据才能告诉你什么优化是真正有效的。4.3 处理“未知问题”与拒答一个成熟的RAG系统必须知道什么时候该说“我不知道”。置信度阈值可以为检索和生成环节设置阈值。检索置信度如果重排序后Top 1片段的相关性分数低于某个阈值例如rerank_score 0.5可能意味着知识库中没有相关信息。生成置信度在Prompt中明确要求模型在无法回答时输出特定语句如“根据已知信息无法回答”并在后端解析答案。如果模型输出了这句话就触发拒答。一致性检查让LLM对自己生成的答案进行评估判断其是否严格基于上下文。这虽然增加了开销但能进一步降低幻觉率。4.4 性能、成本与扩展性缓存对常见问题FAQ的问答对进行缓存可以极大降低检索和LLM调用的开销。异步与流式对于耗时的重排序或LLM生成采用异步处理。对于长答案支持流式输出Server-Sent Events提升用户体验。多模态RAG如果知识库包含图片、图表需要多模态嵌入模型如CLIP来为图像生成向量实现“以图搜图”或“图文联合检索”。图增强RAGGraph RAG对于知识之间存在复杂关系如人物关系、事件脉络的场景可以先将知识抽取成图结构知识图谱。检索时先在图上游走找到相关实体和关系再获取对应的文本片段。这能极大提升对复杂、关联性问题的回答能力。从零实现一个RAG应用就像亲手组装一台精密仪器。你不仅知道了每个零件的名字更清楚了它们如何咬合哪里容易出故障以及如何调试。这个过程赋予你的是对AI应用开发生态更深的理解和更强的掌控力。当你再看到“Agentic RAG”、“Graph RAG”这些新概念时你看到的将不再是一个黑盒术语而是一系列可拆解、可实现的工程组件的有机组合。这才是从“使用工具”到“创造工具”的关键一步。