1. 项目概述从“大海捞针”到“按图索骥”如果你最近在折腾大模型应用尤其是想让AI能回答你公司内部文档、产品手册或者个人知识库里的问题那你大概率已经听过“RAG”这个词了。RAG检索增强生成听起来挺高大上但它的核心思想其实很朴素当AI自己不知道答案时让它学会去“查资料”。这就像我们人类遇到不熟悉的问题第一反应也是去翻书、搜网页。RAG技术就是给大模型装上一个“外部记忆库”和一套高效的“检索系统”。这个“007”项目就是我们踏入RAG世界的第一步实战。它不追求构建一个庞大复杂的系统而是聚焦于RAG最核心、也最基础的两个环节向量嵌入与检索。你可以把它理解为RAG的“心脏”和“搜索引擎”。没有高质量的心脏向量嵌入知识就无法被AI理解没有高效的搜索引擎检索即使有答案也找不到。很多朋友在搭建RAG系统时效果不理想回答不准确问题往往就出在这两个基础环节没做好。所以这个入门项目的目标非常明确我们亲手走一遍从原始文本到向量再到根据问题精准召回相关片段的全流程。我们会使用当前社区里备受好评的BGE-M3模型来做嵌入并深入对比两种经典的检索方法基于关键词的BM25和基于语义的向量检索。最终我们会实现一个简单的“混合检索”雏形体验如何结合两者的优势。无论你是想为自己的项目搭建一个知识库问答机器人还是单纯想理解RAG背后的技术原理这个从“嵌入”到“检索”的完整闭环实验都将为你打下最坚实的实践基础。2. 核心思路解析为什么是“嵌入”与“检索”在开始动手之前我们得先想明白为什么RAG非得从这两个环节入手这背后是一连串的工程现实和技术权衡。2.1 大模型的“幻觉”与“知识截止”问题当前的大语言模型本质上是基于海量文本训练出的“概率大师”。它们擅长生成流畅、合乎语法的文本但存在两个致命弱点幻觉对于训练数据中不存在或模糊的信息模型会“自信地”编造答案。知识截止模型的知识停留在训练数据的时间点无法获取最新信息或私有数据。RAG的提出正是为了根治这两个问题。其核心逻辑是将生成答案的责任从“完全依赖模型参数记忆”转变为“模型参数理解能力 外部检索到的精准信息”。模型不再需要记住所有知识只需要学会如何理解问题并从提供的资料中找到答案并组织语言。2.2 向量嵌入将文字转化为AI的“数学语言”文本是人类的语言而计算机和神经网络最擅长处理的是数字。如何让AI“理解”一段文字的含义向量嵌入技术就是答案。你可以把“嵌入”想象成一个“语义翻译机”。它把一段文本无论是一个词、一句话还是一整段转换成一个固定长度的数字列表也就是“向量”。这个向量的神奇之处在于语义相似的文本其对应的向量在数学空间里的距离比如余弦相似度也会很近。例如“狗”和“犬”的向量会很接近“编程”和“代码”的向量也会很接近而“苹果”水果和“苹果”公司的向量则会相对较远。我们项目选用的BGE-M3模型正是这类“翻译机”中的佼佼者。它由北京智源研究院发布不仅支持多语言更重要的是它产生的向量在各类语义相似度评测基准上都名列前茅而且它原生支持一种叫“ColBERT”的后期交互机制能实现更精细的匹配这是我们后续做高效检索的保障。注意选择嵌入模型是RAG效果的“天花板”。一个差的嵌入模型即使后续检索再精准召回的也可能是语义不相关的垃圾信息。BGE-M3在开源模型中表现均衡且出色是当前入门和生产的优选。2.3 检索策略关键词匹配与语义搜索的“左右互搏”有了向量化的知识库当用户提问时我们需要快速找到最相关的资料片段。这就是检索要干的事。主流方法有两派关键词检索如BM25这是搜索引擎用了数十年的经典算法。它统计查询词和文档中的词频进行加权打分。它的优势是精确。如果你问“Python中如何读取CSV文件”BM25能精准命中包含“Python”、“读取”、“CSV”这些关键词的文档。但它的问题是“词汇鸿沟”无法理解“编程语言”和“Python”是同一个意思。向量检索语义搜索将用户的查询也用同样的嵌入模型转化为向量然后在知识库的所有向量中计算与查询向量最相似距离最近的Top K个。它的优势是泛化。即使你的问题里没有出现“犬”只有“狗”它也能找到关于“犬”的文档。但它有时会过于“发散”召回一些语义相关但并非直接解答问题的内容。我们的项目将同时实现这两种方法并进行对比。更妙的是我们会尝试将它们结合起来即混合检索。简单来说就是让BM25和向量检索各自独立搜索然后对它们的结果进行融合与重排序以期兼得关键词的精确和语义的泛化能力。这是当前工业级RAG系统普遍采用的策略。3. 环境搭建与核心工具选型工欲善其事必先利其器。为了让实验过程清晰可控我们选择在本地Python环境中进行。以下是我们需要准备的核心“武器库”。3.1 Python环境与包管理建议使用Python 3.9或3.10版本过高版本可能遇到一些库的依赖冲突。使用conda或venv创建独立的虚拟环境是最佳实践。# 创建并激活虚拟环境以conda为例 conda create -n rag-007 python3.10 conda activate rag-0073.2 核心库安装与说明我们将通过pip安装一系列库每个库都有其不可替代的作用pip install torch transformers sentence-transformers rank-bm25 faiss-cpu pypdf langchain-chromatorch: PyTorch深度学习框架BGE-M3模型运行的基础。transformers sentence-transformers: Hugging Face出品的模型加载和推理库。sentence-transformers对嵌入模型的使用做了极大简化是我们的首选。rank-bm25: 一个纯Python实现的BM25算法库轻量且易用。faiss-cpu: Facebook AI研发的高效向量相似度搜索库。即使处理上万条向量也能在毫秒级返回结果。faiss-cpu是CPU版本适合入门和中小规模数据。pypdf: 用于读取我们的实验素材——PDF文档。langchain-chroma: LangChain是一个流行的AI应用框架我们这里仅用其Chroma组件作为一个轻量级的向量数据库来存储和查询向量。它内置了Faiss方便集成。3.3 素材准备构建微型知识库为了聚焦流程我们不需要庞大的数据。你可以找两三篇相关的技术博客、产品说明书或项目文档的PDF内容最好围绕一个主题比如“Python基础”或“机器学习简介”。我们将把这些PDF作为原始的“非结构化数据”通过后续流程将其转化为可供检索的“知识”。假设我们将这些PDF保存在项目根目录的./docs文件夹下。4. 从文档到向量完整数据处理流水线现在我们进入核心实操环节。这一步的目标是把一堆PDF文件变成一系列规整的文本片段块及其对应的向量。4.1 文档加载与文本提取首先我们需要读取PDF并提取其中的纯文本。import os from pypdf import PdfReader def load_pdfs_from_folder(folder_path): 从指定文件夹加载所有PDF文件并提取文本。 documents [] for filename in os.listdir(folder_path): if filename.endswith(.pdf): file_path os.path.join(folder_path, filename) print(f正在处理: {filename}) try: reader PdfReader(file_path) text for page in reader.pages: text page.extract_text() \n # 将整个文档文本作为一个条目后续再分块 documents.append({ source: filename, text: text.strip() }) except Exception as e: print(f读取 {filename} 时出错: {e}) return documents # 使用示例 docs_folder ./docs raw_documents load_pdfs_from_folder(docs_folder) print(f共加载了 {len(raw_documents)} 个文档。)4.2 文本分块艺术与科学的结合这是RAG中极其关键且微妙的一步。我们不能把整本书扔给检索系统那样太粗糙也不能切得太碎会丢失上下文。常见的策略有固定长度重叠分块比如每256个字符为一块相邻块重叠50个字符。这是最常用的方法简单有效。按语义分块利用句子或段落分隔符如\n\n,.,;进行分割。更符合人类阅读习惯但对文档格式要求高。递归分块先按大分隔符分如果块太大再用小分隔符继续分。我们采用第一种方法并使用LangChain提供的文本分割器它已经帮我们处理好了重叠等逻辑。from langchain.text_splitter import RecursiveCharacterTextSplitter def split_documents(raw_docs, chunk_size500, chunk_overlap50): 将文档文本分割成小块。 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , ] ) all_chunks [] for doc in raw_docs: chunks text_splitter.split_text(doc[text]) for i, chunk in enumerate(chunks): # 为每个块创建包含元数据的字典 all_chunks.append({ id: f{doc[source]}_chunk_{i}, text: chunk, source: doc[source], chunk_index: i }) return all_chunks # 使用示例 text_chunks split_documents(raw_documents, chunk_size500, chunk_overlap50) print(f文档被分割成 {len(text_chunks)} 个文本块。) print(第一个文本块预览, text_chunks[0][text][:200])实操心得分块大小的“黄金法则”分块大小没有绝对标准需根据你的文档类型和问题类型调整。事实性问答如“某产品的规格是什么”适合较小的块200-500字符答案通常集中在一两句话里。概括性/分析性问题如“总结某技术的优缺点”需要较大的块500-1000字符甚至更大以提供足够的上下文。重叠Overlap设置10%-20%的重叠能有效防止答案被恰好切在块边界而丢失。这是提升召回率的一个低成本技巧。4.3 向量化调用BGE-M3模型接下来是魔法发生的地方——将文本块转化为向量。我们使用sentence-transformers库来加载BGE-M3模型。from sentence_transformers import SentenceTransformer import torch # 设置设备如果有GPU则用GPU速度会快很多 device cuda if torch.cuda.is_available() else cpu print(f使用设备: {device}) # 加载BGE-M3模型。第一次运行会从Hugging Face下载模型需要一定时间。 # encode_kwargs 中的 normalize_embeddingsTrue 非常重要它将向量归一化方便后续使用余弦相似度计算。 model SentenceTransformer(BAAI/bge-m3, devicedevice) model.encode_kwargs {normalize_embeddings: True} # 关键设置 def generate_embeddings(text_chunks): 为所有文本块生成向量嵌入。 # 提取纯文本列表 texts [chunk[text] for chunk in text_chunks] print(f开始为 {len(texts)} 个文本块生成向量...) # 批量编码提高效率 embeddings model.encode(texts, batch_size32, # 根据你的GPU内存调整 show_progress_barTrue, convert_to_tensorTrue) # 返回PyTorch Tensor print(向量生成完成。) # 将向量赋值回每个块 for i, chunk in enumerate(text_chunks): chunk[embedding] embeddings[i].cpu().numpy() # 转成numpy数组方便后续存储 return text_chunks # 执行向量化 enriched_chunks generate_embeddings(text_chunks) # 查看一个向量的形状 print(f向量维度: {enriched_chunks[0][embedding].shape}) # BGE-M3通常是1024维这个过程可能需要一些时间取决于文本块的数量和你的硬件。完成后每个文本块都拥有了一个1024维的“数字指纹”。4.4 向量存储写入Chroma数据库生成向量后我们需要将其持久化存储以便快速检索。这里使用Chroma它非常简单易用。import chromadb from chromadb.config import Settings # 初始化一个持久化的Chroma客户端 chroma_client chromadb.PersistentClient(path./chroma_db_007) # 创建或获取一个集合Collection相当于一个表 collection_name rag_demo_collection # 如果集合已存在先删除为了演示每次重新创建 try: chroma_client.delete_collection(collection_name) except: pass collection chroma_client.create_collection(namecollection_name) # 准备批量插入的数据 ids [chunk[id] for chunk in enriched_chunks] embeddings [chunk[embedding].tolist() for chunk in enriched_chunks] # Chroma需要list格式 documents [chunk[text] for chunk in enriched_chunks] metadatas [{source: chunk[source], index: chunk[chunk_index]} for chunk in enriched_chunks] # 批量添加到集合 collection.add( idsids, embeddingsembeddings, documentsdocuments, metadatasmetadatas ) print(f成功将 {len(ids)} 个向量存入Chroma数据库。)至此我们的“向量知识库”就构建完成了。它包含了原始的文本、对应的向量以及一些元数据来源、块索引静静地躺在./chroma_db_007目录下等待被查询。5. 双路检索实战BM25与向量检索的实现与对比知识库准备好了现在我们来打造两把不同的“钥匙”BM25和向量检索并用同一个问题来测试它们。5.1 构建BM25检索器BM25不需要向量它直接在原始文本上工作。from rank_bm25 import BM25Okapi import jieba # 用于中文分词如果是英文文档可用nltk或直接按空格分 def build_bm25_index(text_chunks): 构建BM25检索索引。 # 对每个文本块进行分词。对于中文分词是关键步骤。 tokenized_corpus [] for chunk in text_chunks: # 使用jieba进行精确模式分词 tokens list(jieba.cut_for_search(chunk[text])) # 搜索引擎模式粒度较细 # 如果是英文可以这样tokens chunk[text].lower().split() tokenized_corpus.append(tokens) # 创建BM25对象 bm25 BM25Okapi(tokenized_corpus) return bm25, text_chunks # 构建索引 bm25_index, bm25_chunks build_bm25_index(enriched_chunks) def search_with_bm25(query, bm25_index, chunks, top_k5): 使用BM25进行检索。 # 对查询进行同样的分词处理 tokenized_query list(jieba.cut_for_search(query)) # 获取文档得分 scores bm25_index.get_scores(tokenized_query) # 获取得分最高的top_k个索引 top_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue)[:top_k] # 返回对应的文本块和得分 results [] for idx in top_indices: results.append({ text: chunks[idx][text], score: scores[idx], source: chunks[idx][source], id: chunks[idx][id] }) return results5.2 实现向量检索向量检索则利用我们之前存入Chroma的向量。def search_with_vector(query, collection, model, top_k5): 使用向量进行语义检索。 # 将查询语句转化为向量 query_embedding model.encode([query], convert_to_tensorTrue).cpu().numpy().tolist()[0] # 向Chroma数据库查询 vector_results collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, metadatas, distances] # 返回文档、元数据和距离 ) # 整理结果 results [] if vector_results[documents]: for i in range(len(vector_results[documents][0])): results.append({ text: vector_results[documents][0][i], score: 1 - vector_results[distances][0][i], # Chroma返回的是距离我们转为相似度分数余弦相似度 source: vector_results[metadatas][0][i][source], id: vector_results[ids][0][i] }) return results5.3 同场竞技设计测试查询让我们设计几个有代表性的问题来对比两种检索方式的效果。# 定义测试问题 test_queries [ Python中如何定义一个函数, # 精确关键词匹配 机器学习的基本步骤有哪些, // 概念性、概括性问题 如果程序运行报错‘内存不足’可能是什么原因 // 语义相关但关键词可能不直接匹配 ] print( 检索效果对比测试 \n) for query in test_queries: print(f查询: 「{query}」) print(- * 50) # BM25检索 print([BM25 结果]:) bm25_results search_with_bm25(query, bm25_index, bm25_chunks, top_k3) for i, res in enumerate(bm25_results): print(f {i1}. (得分: {res[score]:.4f}) {res[text][:150]}...) print() # 向量检索 print([向量检索 结果]:) vector_results search_with_vector(query, collection, model, top_k3) for i, res in enumerate(vector_results): print(f {i1}. (相似度: {res[score]:.4f}) {res[text][:150]}...) print(\n *80 \n)预期观察与分析对于第一个问题“如何定义函数”BM25可能会因为精确匹配“Python”、“定义”、“函数”等词而排名靠前向量检索也能找到相关内容但排序可能略有不同。对于第二个问题“机器学习基本步骤”向量检索的优势可能更明显因为它能理解“步骤”、“流程”、“阶段”等词的语义关联即使文档中没有完全相同的表述。对于第三个问题“内存不足报错”这是一个典型的语义搜索场景。文档中可能描述的是“内存溢出”、“RAM不足”、“OOM错误”BM25如果只匹配“内存不足”这个词可能一无所获而向量检索则可能找到相关段落。6. 进阶混合检索与重排序初步实践单一的检索方式总有局限工业级系统通常会采用“混合检索”策略。这里我们实现一个最简单的版本加权分数融合。6.1 分数归一化与融合BM25和向量检索的分数范围不同BM25分数无上限向量相似度在0-1之间直接相加不公平。我们需要先对分数进行归一化。def normalize_scores(scores): 将分数列表归一化到[0,1]区间。 if not scores: return scores min_s, max_s min(scores), max(scores) if max_s min_s: # 防止除零 return [0.5 for _ in scores] return [(s - min_s) / (max_s - min_s) for s in scores] def hybrid_search(query, bm25_index, bm25_chunks, collection, model, top_k5, bm25_weight0.4, vector_weight0.6): 混合检索结合BM25和向量检索的结果。 # 1. 分别检索获取更多候选结果 bm25_candidates search_with_bm25(query, bm25_index, bm25_chunks, top_ktop_k*2) vector_candidates search_with_vector(query, collection, model, top_ktop_k*2) # 2. 构建一个字典来聚合所有候选结果key是文本块的id candidate_dict {} for res in bm25_candidates: candidate_dict[res[id]] { text: res[text], source: res[source], bm25_score: res[score], vector_score: 0.0 # 先初始化为0 } for res in vector_candidates: if res[id] in candidate_dict: # 如果BM25也找到了更新向量分数 candidate_dict[res[id]][vector_score] res[score] else: # 如果只有向量找到了添加新条目 candidate_dict[res[id]] { text: res[text], source: res[source], bm25_score: 0.0, vector_score: res[score] } # 3. 提取分数分别归一化 all_candidates list(candidate_dict.values()) bm25_scores [c[bm25_score] for c in all_candidates] vector_scores [c[vector_score] for c in all_candidates] norm_bm25_scores normalize_scores(bm25_scores) norm_vector_scores normalize_scores(vector_scores) # 4. 计算加权综合分 for i, cand in enumerate(all_candidates): hybrid_score (bm25_weight * norm_bm25_scores[i]) (vector_weight * norm_vector_scores[i]) cand[hybrid_score] hybrid_score # 5. 按综合分排序返回top_k sorted_candidates sorted(all_candidates, keylambda x: x[hybrid_score], reverseTrue)[:top_k] return sorted_candidates6.2 测试混合检索效果现在用同样的查询来测试混合检索看看它是否能够取长补短。print( 混合检索测试 \n) for query in test_queries: print(f查询: 「{query}」) hybrid_results hybrid_search(query, bm25_index, bm25_chunks, collection, model, top_k3) for i, res in enumerate(hybrid_results): print(f {i1}. (综合分: {res[hybrid_score]:.4f}, BM25: {res[bm25_score]:.2f}, 向量: {res[vector_score]:.4f})) print(f 文本: {res[text][:120]}...) print(\n -*80 \n)通过调整bm25_weight和vector_weight参数你可以控制检索的“风格”。例如在需要高精确度的场景如法律条文查询可以调高BM25权重在需要泛化理解的场景如创意灵感搜索可以调高向量权重。注意事项权重调优混合检索的权重不是固定的需要根据你的具体数据和查询类型进行微调。一个实用的方法是准备一个“验证集”——一批问题及其对应的标准答案文档块然后尝试不同的权重组合选择那个能让正确答案排名最靠前的组合。这是一个简单的“超参数”调优过程。7. 常见问题、排查技巧与效能优化在实际操作中你肯定会遇到各种各样的问题。下面是我踩过坑后总结的一些经验。7.1 检索效果不佳的排查清单当你发现召回的内容总是不相关时可以按照以下顺序排查问题现象可能原因排查与解决思路所有检索方式结果都差文本分块不合理检查分块大小和重叠。对于问答尝试更小的块如200字。预览几个块的内容看是否把完整的句子或思想切碎了。文档质量差/噪声大检查原始PDF提取的文本是否包含大量乱码、页眉页脚。考虑增加文本清洗步骤如移除URL、特殊字符、连续空格等。向量检索效果差嵌入模型不匹配BGE-M3适合中英文混合。如果你纯英文可尝试all-MiniLM-L6-v2纯中文可尝试BAAI/bge-small-zh。在 Hugging Face MTEB榜单 上选择适合你任务的模型。向量未归一化确保生成向量时设置了normalize_embeddingsTrue。余弦相似度计算依赖于归一化后的向量。BM25检索效果差分词问题中文必须分词。检查分词后的词汇是否合理。对于专业术语可以考虑加载自定义词典到jieba中。停用词未处理BM25受常见词的、是、在影响大。可以考虑在分词后移除停用词但需谨慎有时停用词也有意义。混合检索结果奇怪权重设置不当调整bm25_weight和vector_weight。极端情况下可以先设为1和0或0和1看单路检索是否正常再慢慢调整。分数归一化问题检查normalize_scores函数是否正确处理了所有分数相同的情况除零错误。7.2 性能与扩展性考量嵌入速度慢这是离线处理过程慢点可以接受。如果急需加速可以1) 使用GPU2) 增大encode函数的batch_size参数需注意显存3) 考虑更小的模型如bge-small-zh-v1.5在精度和速度间权衡。检索速度慢对于千万级以下的向量Faiss在CPU上也能做到毫秒级响应。如果数据量极大需考虑1) 使用Faiss的GPU版本2) 使用量化索引如IndexIVFPQ在损失少量精度的情况下大幅提升速度和减少内存占用。内存占用高向量数据库和BM25索引都会驻留内存。对于超大知识库需要分布式方案或使用专业的云端向量数据库如Milvus、Pinecone、Weaviate。7.3 超越基础重排序的引入我们目前的混合检索只是简单的分数融合更高级的做法是引入一个重排序模型。它的思路是先用快速的检索器如BM25向量召回100个候选文档然后用一个更精细但更慢的模型如交叉编码器对这100个候选进行精排选出最相关的10个。sentence-transformers也提供了交叉编码器模型它不像嵌入模型那样输出向量而是直接计算查询和文档对的匹配分数精度更高。# 示例使用交叉编码器进行重排序可选进阶步骤 from sentence_transformers import CrossEncoder # 加载一个重排序模型 reranker CrossEncoder(BAAI/bge-reranker-base, devicedevice) def rerank_results(query, candidate_list, reranker, top_k5): 对候选列表进行重排序。 if not candidate_list: return [] # 构建查询文档对 pairs [[query, cand[text]] for cand in candidate_list] # 预测分数 scores reranker.predict(pairs) # 将分数附加到候选字典中 for cand, score in zip(candidate_list, scores): cand[rerank_score] score # 按重排序分数排序 reranked sorted(candidate_list, keylambda x: x[rerank_score], reverseTrue)[:top_k] return reranked # 使用示例先混合检索出10个候选再重排序选出3个最佳 candidates hybrid_search(test_queries[0], bm25_index, bm25_chunks, collection, model, top_k10) final_results rerank_results(test_queries[0], candidates, reranker, top_k3)重排序能显著提升最终答案的相关性但代价是增加了计算开销。它通常用在召回之后、将结果喂给大模型生成答案之前的最后一步是提升RAG答案质量的关键技术之一。走到这里你已经完成了一个完整、可运行的RAG核心流程实验。从文档处理、向量化到双路检索、混合排序每一步都亲手实现并理解了其背后的考量。这个“007”项目虽然简单但骨架已然清晰。接下来你可以在这个基础上尝试接入真实的大模型如通过OpenAI API或本地部署的Qwen、ChatGLM等构建一个完整的问答系统或者继续深入探索更复杂的切片策略、检索算法和重排序技术。RAG的世界很大但你已经拿到了第一把钥匙。
