从向量检索到混合搜索：手把手教你搭建生产级RAG检索层（附全代码）

读完这篇你将得到成果说明✅ 一个可运行的 ChromaDB 向量库支持持久化、相似度检索✅ 一个可运行的 BM25 检索器中文分词、精确匹配✅ 一个混合检索器RRF 融合向量 关键词 双路召回✅ 一个 Rerank 精排模块Cross-Encoder 提升 Top 准确率✅ 一个完整的 RAGRetriever 类复制粘贴即可用环境要求Python 3.88GB 内存即可运行。第一部分环境准备5分钟bash# 创建虚拟环境 python -m venv rag_retriever source rag_retriever/bin/activate # Linux/Mac # rag_retriever\Scripts\activate # Windows # 安装依赖 pip install chromadb # 向量数据库 pip install sentence-transformers # Embedding 模型 pip install rank-bm25 # BM25 检索 pip install jieba # 中文分词 pip install FlagEmbedding # BGE Reranker可选较大 # 或轻量版pip install sentence-transformers验证安装pythonpython -c import chromadb; print(ChromaDB OK) python -c import sentence_transformers; print(Sentence Transformers OK)第二部分准备测试数据创建一个docs.txt文件text华为云 ModelArts 是一个端到端的 AI 开发平台支持数据处理、模型训练、模型部署。 MindSpore 是华为开源的全场景 AI 框架支持端边云协同。 ModelArts Pro 是企业级 AI 应用开发套件提供了预训练模型和自动化标注。 PyTorch 是一个流行的深度学习框架由 Meta 开源。 TensorFlow 是 Google 开源的机器学习框架广泛应用于生产环境。 ModelArts 支持 PyTorch、TensorFlow、MindSpore 等多种深度学习框架。 华为云 ModelArts 提供了自动学习功能无需编码即可训练模型。 ModelArts 支持大规模分布式训练可以处理亿级参数模型。读取文档pythonwith open(docs.txt, r, encodingutf-8) as f: documents [line.strip() for line in f if line.strip()] print(f共加载 {len(documents)} 条文档) for i, doc in enumerate(documents): print(f{i}: {doc[:50]}...)第三部分ChromaDB 向量检索实战初始化pythonimport chromadb from sentence_transformers import SentenceTransformer # 1. 创建持久化客户端数据保存在 ./chroma_db 目录 client chromadb.PersistentClient(path./chroma_db) # 2. 创建 Collection类似 SQL 的表 collection client.get_or_create_collection( namedocs, metadata{hnsw:space: cosine} # 使用余弦相似度 ) print(fCollection 已创建当前文档数: {collection.count()})加载 Embedding 模型python# 使用多语言模型支持中英文 model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) print(f向量维度: {model.get_sentence_embedding_dimension()}) def get_embedding(text): return model.encode(text).tolist()批量插入文档python# 生成 ID ids [fdoc_{i} for i in range(len(documents))] # 生成向量首次运行会慢一些 embeddings [get_embedding(doc) for doc in documents] # 插入 collection.add( idsids, documentsdocuments, embeddingsembeddings ) print(f成功插入 {len(documents)} 条文档)向量检索函数pythondef vector_search(query, top_k3): 向量检索 query_embedding get_embedding(query) results collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, distances] ) print(f\n【向量检索】查询: {query}) print(- * 50) for i, (doc, dist) in enumerate(zip(results[documents][0], results[distances][0])): # 余弦距离转相似度1 - dist/2 similarity 1 - dist / 2 print(f{i1}. 相似度: {similarity:.3f}) print(f {doc}\n) return results # 测试 vector_search(ModelArts 支持哪些深度学习框架)预期输出text【向量检索】查询: ModelArts 支持哪些深度学习框架 -------------------------------------------------- 1. 相似度: 0.892 ModelArts 支持 PyTorch、TensorFlow、MindSpore 等多种深度学习框架。 2. 相似度: 0.754 华为云 ModelArts 是一个端到端的 AI 开发平台... 3. 相似度: 0.721 ModelArts Pro 是企业级 AI 应用开发套件...第四部分BM25 关键词检索实战安装中文分词pythonimport jieba def tokenize_chinese(text): 中文分词 return list(jieba.cut(text)) # 测试 print(tokenize_chinese(华为云 ModelArts 是 AI 平台)) # 输出: [华为云, , ModelArts, , 是, , AI, , 平台]构建 BM25 索引pythonfrom rank_bm25 import BM25Okapi # 对所有文档进行分词 tokenized_docs [tokenize_chinese(doc) for doc in documents] # 构建 BM25 索引 bm25 BM25Okapi(tokenized_docs) print(BM25 索引构建完成)BM25 检索函数pythondef bm25_search(query, top_k3): BM25 关键词检索 tokenized_query tokenize_chinese(query) scores bm25.get_scores(tokenized_query) # 获取 Top-K 索引 top_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue)[:top_k] print(f\n【BM25检索】查询: {query}) print(- * 50) for i, idx in enumerate(top_indices): print(f{i1}. BM25分数: {scores[idx]:.3f}) print(f {documents[idx]}\n) return top_indices, scores # 测试 bm25_search(ModelArts Pro)预期输出text【BM25检索】查询: ModelArts Pro -------------------------------------------------- 1. BM25分数: 8.234 ModelArts Pro 是企业级 AI 应用开发套件... 2. BM25分数: 4.123 华为云 ModelArts 是一个端到端的 AI 开发平台... 3. BM25分数: 0.000 MindSpore 是华为开源的全场景 AI 框架...BM25 调参python# 自定义参数 # k1: 词频饱和度默认1.5越大词频影响越大 # b: 文档长度惩罚默认0.75越大长文档惩罚越重 bm25_custom BM25Okapi(tokenized_docs, k11.2, b0.75) # 不同场景推荐配置 configs { 短文档标题/摘要: {k1: 1.2, b: 0.5}, 长文档论文/财报: {k1: 1.5, b: 0.8}, 精确匹配优先: {k1: 0.0, b: 0.75}, }第五部分混合检索 RRF 融合为什么需要 RRF向量分数0-1和 BM25 分数0-10不在同一量纲不能直接相加。RRF 只看排名不看分数完整混合检索器pythonimport numpy as np from sentence_transformers import util class HybridRetriever: def __init__(self, documents, embedding_modelparaphrase-multilingual-MiniLM-L12-v2): self.documents documents self.n_docs len(documents) print(初始化 HybridRetriever...) # 1. BM25 索引 print( - 构建 BM25 索引...) self.tokenized_docs [tokenize_chinese(doc) for doc in documents] self.bm25 BM25Okapi(self.tokenized_docs) # 2. 向量索引 print( - 加载 Embedding 模型...) self.encoder SentenceTransformer(embedding_model) print(f - 生成文档向量...) self.doc_embeddings self.encoder.encode(documents, convert_to_tensorTrue) print(f初始化完成文档数: {self.n_docs}) def search_vector(self, query): 向量检索返回所有文档的相似度 query_embedding self.encoder.encode(query, convert_to_tensorTrue) scores util.cos_sim(query_embedding, self.doc_embeddings)[0].cpu().numpy() return scores def search_bm25(self, query): BM25 检索返回所有文档的分数 tokenized_query tokenize_chinese(query) scores self.bm25.get_scores(tokenized_query) return scores def hybrid_search(self, query, top_k3, rrf_k60, print_resultsTrue): 混合检索 RRF 融合 参数: query: 查询文本 top_k: 返回结果数 rrf_k: RRF 平滑参数默认 60 # 1. 获取两种检索的全量分数 bm25_scores self.search_bm25(query) vector_scores self.search_vector(query) # 2. 获取排名分数越高排名越靠前 bm25_rank {idx: i for i, idx in enumerate(np.argsort(bm25_scores)[::-1])} vector_rank {idx: i for i, idx in enumerate(np.argsort(vector_scores)[::-1])} # 3. RRF 融合 final_scores {} for idx in range(self.n_docs): rrf_score 0 if idx in bm25_rank: rrf_score 1.0 / (rrf_k bm25_rank[idx] 1) if idx in vector_rank: rrf_score 1.0 / (rrf_k vector_rank[idx] 1) final_scores[idx] rrf_score # 4. 排序并返回 sorted_indices sorted(final_scores.keys(), keylambda x: final_scores[x], reverseTrue)[:top_k] results [(idx, final_scores[idx], self.documents[idx]) for idx in sorted_indices] if print_results: print(f\n【混合检索】查询: {query}) print( * 60) for i, (idx, score, doc) in enumerate(results): print(f{i1}. [RRF: {score:.4f}] {doc}) return results # 使用 retriever HybridRetriever(documents) retriever.hybrid_search(ModelArts 支持哪些深度学习框架)预期输出text初始化 HybridRetriever... - 构建 BM25 索引... - 加载 Embedding 模型... - 生成文档向量... 初始化完成文档数: 8 【混合检索】查询: ModelArts 支持哪些深度学习框架 1. [RRF: 0.0328] ModelArts 支持 PyTorch、TensorFlow、MindSpore 等多种深度学习框架。 2. [RRF: 0.0161] 华为云 ModelArts 是一个端到端的 AI 开发平台... 3. [RRF: 0.0159] ModelArts Pro 是企业级 AI 应用开发套件...第六部分Rerank 结果重排为什么需要 Rerank混合检索返回的 Top-5 可能只有 2 条真正相关。Rerank 用更精细的模型重新排序。使用 BGE Rerankerpython# 方式1使用 FlagEmbedding推荐精度高 try: from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-large, use_fp16False) print(BGE Reranker 加载成功) except Exception as e: print(fBGE Reranker 加载失败: {e}) reranker None # 方式2轻量级 Cross-Encoder备选 from sentence_transformers import CrossEncoder light_reranker CrossEncoder(cross-encoder/ms-marco-MiniLM-L-6-v2)Rerank 函数pythondef rerank_results(query, documents, top_k3): 对检索结果进行重排 if reranker is None: # 使用轻量级 reranker pairs [[query, doc] for doc in documents] scores light_reranker.predict(pairs) else: pairs [[query, doc] for doc in documents] scores [reranker.compute_score(pair) for pair in pairs] # 按分数排序 ranked sorted(zip(documents, scores), keylambda x: x[1], reverseTrue) print(f\n【Rerank 重排】查询: {query}) print( * 60) for i, (doc, score) in enumerate(ranked[:top_k]): print(f{i1}. [相关度: {score:.3f}] {doc}) return ranked[:top_k] # 测试先用混合检索拿候选再重排 candidates retriever.hybrid_search(ModelArts 支持 PyTorch 吗, top_k5, print_resultsFalse) candidate_docs [doc for _, _, doc in candidates] rerank_results(ModelArts 支持 PyTorch 吗, candidate_docs, top_k3)Rerank 前后对比pythondef compare_rerank(query, top_k3): 对比 Rerank 前后的结果 # 混合检索原始结果 raw_results retriever.hybrid_search(query, top_ktop_k, print_resultsFalse) raw_docs [doc for _, _, doc in raw_results] # Rerank 后结果 reranked rerank_results(query, raw_docs, top_ktop_k) reranked_docs [doc for doc, _ in reranked] print(f\n查询: {query}) print(f原始 Top-{top_k}: {[d[:40] for d in raw_docs]}) print(fRerank Top-{top_k}: {[d[:40] for d in reranked_docs]}) compare_rerank(ModelArts 的自动学习功能)第七部分完整 RAGRetriever 类把上面所有功能整合成一个类复制粘贴即可用pythonclass RAGRetriever: 完整的 RAG 检索器 - 开箱即用 def __init__(self, documents, embedding_modelparaphrase-multilingual-MiniLM-L12-v2): self.documents documents # BM25 self.tokenized_docs [tokenize_chinese(doc) for doc in documents] self.bm25 BM25Okapi(self.tokenized_docs) # 向量检索 self.encoder SentenceTransformer(embedding_model) self.doc_embeddings self.encoder.encode(documents, convert_to_tensorTrue) # Reranker懒加载 self._reranker None property def reranker(self): if self._reranker is None: try: from FlagEmbedding import FlagReranker self._reranker FlagReranker(BAAI/bge-reranker-large, use_fp16False) except: from sentence_transformers import CrossEncoder self._reranker CrossEncoder(cross-encoder/ms-marco-MiniLM-L-6-v2) return self._reranker def search(self, query, top_k5, use_rerankTrue): 统一检索接口 参数: query: 查询文本 top_k: 返回文档数 use_rerank: 是否使用 Rerank # 1. 混合检索 bm25_scores self.bm25.get_scores(tokenize_chinese(query)) query_emb self.encoder.encode(query, convert_to_tensorTrue) vector_scores util.cos_sim(query_emb, self.doc_embeddings)[0].cpu().numpy() # RRF 融合 bm25_rank {i: r for r, i in enumerate(np.argsort(bm25_scores)[::-1])} vector_rank {i: r for r, i in enumerate(np.argsort(vector_scores)[::-1])} final_scores {} for i in range(len(self.documents)): score 0 if i in bm25_rank: score 1 / (60 bm25_rank[i] 1) if i in vector_rank: score 1 / (60 vector_rank[i] 1) final_scores[i] score candidates sorted(final_scores.items(), keylambda x: x[1], reverseTrue)[:top_k] candidate_docs [(idx, self.documents[idx]) for idx, _ in candidates] # 2. Rerank可选 if use_rerank: pairs [[query, doc] for _, doc in candidate_docs] scores self.reranker.predict(pairs) if hasattr(self.reranker, predict) else [ self.reranker.compute_score(pair) for pair in pairs ] ranked sorted(zip(candidate_docs, scores), keylambda x: x[1], reverseTrue) results [(idx, doc, score) for (idx, doc), score in ranked] else: results [(idx, doc, final_scores[idx]) for idx, doc in candidate_docs] return results def print_results(self, query, top_k3, use_rerankTrue): 打印检索结果 results self.search(query, top_k, use_rerank) print(f\n查询: {query}) print( * 60) for i, (idx, doc, score) in enumerate(results[:top_k]): print(f{i1}. [分数: {score:.4f}] {doc}) return results # 使用 retriever RAGRetriever(documents) retriever.print_results(ModelArts 支持哪些框架) retriever.print_results(华为的 AI 开发平台有什么功能)第八部分性能对比总结python# 对比三种检索方式的效果 def benchmark(query): print(f\n{*60}) print(f基准测试: {query}) print(f{*60}) # 向量检索 vector_search(query, top_k3) # BM25 bm25_search(query, top_k3) # 混合检索 retriever.print_results(query, top_k3, use_rerankFalse) # 混合检索 Rerank retriever.print_results(query, top_k3, use_rerankTrue) benchmark(ModelArts 的自动学习功能)传统产品经理正在成为下个被淘汰的“传统岗位”。过去画原型、写 PRD、跟进度的“传统技能包”在AI时代正迅速贬值。63% 的企业转型做 AI 产品当下的问题不再是“要不要学 AI ”而是“如何构建 AI 产品”。前段时间还跟字节、腾讯的资深 AI 产品经理沟通他们反馈在大量招人只要有 AI 相关的项目经验基本都能拿到面试机会而且领导很舍得给钱涨薪 40-60% 很正常01接下来的产品人得卷AI能力了如今AI大火行业极速发展的背后懂AI 产品人才却严重稀缺。这不是要你转技术岗而是要掌握构建 AI 产品的核心方法如何将你的领域知识转化为 AI 产品的核心竞争力如何用 AI 技术实现你的产品需求如何设计真正懂用户的 AI 交互体验……懂AI就是产品经理的“救命稻草”风口之下与其焦虑被行业淘汰不如先人一步享受AI技术带来的红利我把AI产品经理的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】不限年龄不限岗位没有代码基础也能学现在扫码完课还送《AI产品面试题库》《AI大模型应用案例集》02掌握技术实战快速转型想成为一名卓越的AI大模型产品经理需要从技术、到项目实战的全方位转型指南**1**AI产品应用原理解析产品经理也能听懂对于产品经理来说如果你不懂技术做不了业务和AI大模型技术衔接、定义不了数据需求是没法完整的落地一个产品的本次课程专门面向产品经理人群解析当下最热门的AI产品应用的必备的「大模型」、「多模态」的实际应用和算法原理解析AI产品应用技术积累大模型能力简单易懂不需要会代码小白也能掌握大模型微调掌握主流大模型如DeepSeek、Qwen等的微调技术针对特定场景优化模型性能。学习如何利用领域数据如制造、医药、金融等进行模型定制AI Agent智能体搭建学习如何设计和开发AI Agent实现多任务协同、自主决策和复杂问题解决。构建垂类场景下的智能助手产品如制造业中的设备故障诊断Agent、金融领域的投资分析Agent等2超全行业案例解析课程详细讲解现阶段大模型在各个行业和领域的应用现状包括零售与电商、教育、医疗、泛娱乐、法律等等10大行业详细讲解案例的思路、应用场景以及背后的技术原理、核心技术揭秘各个行业、场景的真实现状和未来产品的发展与机遇可以说讲解完一个案例就能积累一个AI产品实践的经验课程中所涉及到的实战项目都可以直接在自己的工作中使用让自己的产品/项目有可借鉴的成功案例3AI产品经理求职专项辅导课程中会系统的帮助大家拆解字节、腾讯、百度等大厂AI PM岗位JD关键词掌握AI PM高频面试题型与回答框架展示 AI 相关能力的关键技巧Prompt设计、模型评估、A/B测试、成本意识、与算法/工程协作经验To B类AI产品经理突出“行业理解 技术落地 商业闭环”能力的简历结构设计展示项目成果从客户需求洞察到技术方案设计展现端到产品思维如何评估To B AI产品的可行性、客户付费意愿与实施成本To C类AI产品经理拆解头部公司岗位JD将过往尽力转化为AI产品叙事逻辑从行业趋势、产品设计题、案例分析数据分析题、技术理解边界等全流程辅导面试避免无效海投、锁定最适合的AI产品岗位03本次课程全程直播讲解能直接对话大佬和专业助教不懂就问超详细的案例小白也能轻松get完课后还赠送《AI产品经理面试题库》、《AI大模型应用案例集》不断更新中……适合人群想转型AI产品经理、AI项目管理专家、AI产品解决方案等岗位想进行AI产品创业的创业者想成为制作AI产品的程序员想利用AI解决企业问题的管理岗想在AI方向寻找就业方向的毕业生AI方向前景广阔、待遇好目前很多产品人已经通过完整学习拿到大厂高薪offer收入嗷嗷涨我把AI产品经理的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】