RAG（三）检索（3）混合检索

一、步骤混合检索Hybrid Retrieval是将多种检索方法通常是基于关键词的稀疏检索和基于向量的语义检索进行组合通过多路召回和统一排序来提升检索效果。dense向量语义检索擅长“意思相近”sparse关键词检索擅长“词面精确命中”在 RAG 里通常是这个流程文档切片Dense EmbeddingSparse/关键词表示向量库用户问题Dense QuerySparse QueryDense 检索Sparse 检索结果融合可选重排大模型回答具体怎么做入库时每个 chunk 同时保存两份表示一份是 dense embedding一份是 sparse 词项表示或者关键词索引查询时问题先做两次表示一次走 dense 检索一次走 sparse 检索合并结果把两路结果融合常见方法是 RRF也就是“两个榜单都靠前的文档最终排得更前”也可以用 DBSF 这类分数融合可选重排先召回 top-k再用 cross-encoder 或 reranker 重新排序二、demo1、向量检索BM25检索在RAG三检索2向量检索-CSDN博客的基础上改成混合检索。1.1、代码改动1common主要改动。2store创建 collection 时新增了 sparse_vectors_configvectors_config 负责 dense 向量sparse_vectors_config{langchain-sparse: SparseVectorParams()} 负责 BM25 sparse写入时改成通过 make_vectorstore(client) 入库这样会同时写 dense BM25 sparse3ask.pyimport 里改成用了 make_vectorstore为了支持关键词检索还需要做中文分词这里用了jieba 分词。看下完整代码1commonimport os from pathlib import Path import sys import httpx import jieba import regex from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_qdrant import FastEmbedSparse, QdrantVectorStore, RetrievalMode from qdrant_client import QdrantClient sys.stdout.reconfigure(encodingutf-8) # 知识库目录demo/docs/small_appliance_kb DEMO_DIR Path(__file__).resolve().parents[1] DOC_DIR DEMO_DIR / docs / small_appliance_kb FASTEMBED_CACHE_DIR DEMO_DIR / .cache / fastembed COLLECTION_NAME small_appliance_demo QDRANT_URL http://localhost:6333 CHUNK_SIZE 360 CHUNK_OVERLAP 80 EMBEDDING_MODEL text-embedding-3-small LLM_MODEL gpt-5.1 BASE_URL https://llm.xxx..xxxx/v1 API_KEY ****** PROXY_ENV_KEYS ( HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, GIT_HTTP_PROXY, GIT_HTTPS_PROXY, http_proxy, https_proxy, all_proxy, git_http_proxy, git_https_proxy, ) # 先抓中文再抓英文/数字最后交给 BM25 做词项检索。 SPARSE_TOKEN_RE regex.compile( r\p{Han}|[A-Za-z0-9](?:[-_./][A-Za-z0-9])*, regex.VERSION1, ) CHINESE_RE regex.compile(r\p{Han}, regex.VERSION1) class CompatQdrantClient(QdrantClient): # 兼容新版 qdrant-client让 langchain_qdrant 继续调用 search 接口。 def search( self, *, collection_name, query_vector, query_filterNone, search_paramsNone, limit10, offset0, with_payloadTrue, with_vectorsFalse, score_thresholdNone, consistencyNone, **kwargs, ): return self.query_points( collection_namecollection_name, queryquery_vector, query_filterquery_filter, search_paramssearch_params, limitlimit, offsetoffset, with_payloadwith_payload, with_vectorswith_vectors, score_thresholdscore_threshold, consistencyconsistency, **kwargs, ).points def normalize_for_sparse(text: str) - str: # 中中文分词英文统一小写给 BM25 这一路一个更稳的词面输入。 tokens: list[str] [] for chunk in SPARSE_TOKEN_RE.findall(text): if CHINESE_RE.search(chunk): tokens.extend(token.strip() for token in jieba.lcut(chunk) if token.strip()) else: tokens.append(chunk.lower()) return .join(tokens) class MixedLanguageSparseEmbeddings: # 先用 FastEmbed 的 BM25再叠一层最小的中英预处理。 def __init__(self) - None: os.environ.setdefault(HF_HUB_DISABLE_SYMLINKS, 1) os.environ.setdefault(HF_HUB_DISABLE_SYMLINKS_WARNING, 1) for key in PROXY_ENV_KEYS: os.environ.pop(key, None) FASTEMBED_CACHE_DIR.mkdir(parentsTrue, exist_okTrue) self._inner FastEmbedSparse(cache_dirstr(FASTEMBED_CACHE_DIR)) def embed_documents(self, texts: list[str]): return self._inner.embed_documents([normalize_for_sparse(text) for text in texts]) def embed_query(self, text: str): return self._inner.embed_query(normalize_for_sparse(text)) def make_client() - CompatQdrantClient: # 连接本地 Qdrant 服务。 return CompatQdrantClient(urlQDRANT_URL) def make_embeddings() - OpenAIEmbeddings: # 第 2 步生成 dense embedding。 return OpenAIEmbeddings( modelEMBEDDING_MODEL, api_keydummy, base_urlBASE_URL, default_headers{X-Api-Key: API_KEY}, http_clienthttpx.Client(trust_envFalse), ) def make_sparse_embeddings() - MixedLanguageSparseEmbeddings: # 第 2 步生成 sparse embedding。 # 这里仍然用 Qdrant/BM25但先把中文分词、英文小写化。 return MixedLanguageSparseEmbeddings() def make_vectorstore( client: QdrantClient, retrieval_mode: RetrievalMode RetrievalMode.HYBRID, ) - QdrantVectorStore: # 统一的向量库入口dense BM25 sparse 混合检索。 return QdrantVectorStore( clientclient, collection_nameCOLLECTION_NAME, embeddingmake_embeddings(), sparse_embeddingmake_sparse_embeddings(), retrieval_moderetrieval_mode, ) def make_llm() - ChatOpenAI: # 回答问题时用的大模型。 return ChatOpenAI( modelLLM_MODEL, base_urlBASE_URL, api_keydummy, default_headers{X-Api-Key: API_KEY}, http_clienthttpx.Client(trust_envFalse), ) def list_doc_paths() - list[Path]: # 第 0 步扫描知识库目录自动读取里面所有文档。 if not DOC_DIR.exists(): return [] return sorted( path for path in DOC_DIR.iterdir() if path.is_file() and path.suffix.lower() in {.md, .txt} ) def load_chunks(doc_paths: list[Path] | None None) - tuple[list[str], list[dict]]: # 第 1 步固定步长滑动切片保证相邻 chunk 有重叠。 if doc_paths is None: doc_paths list_doc_paths() step CHUNK_SIZE - CHUNK_OVERLAP texts: list[str] [] metadatas: list[dict] [] for path in doc_paths: text path.read_text(encodingutf-8) chunks [ text[start : start CHUNK_SIZE] for start in range(0, len(text), step) if text[start : start CHUNK_SIZE] ] texts.extend(chunks) # 第 3 步chunk 元数据只保留来源文件名和序号。 metadatas.extend( {source: path.name, chunk: i 1} for i in range(len(chunks)) ) return texts, metadatas2storefrom qdrant_client.models import Distance, SparseVectorParams, VectorParams from common import ( COLLECTION_NAME, DOC_DIR, QDRANT_URL, list_doc_paths, load_chunks, make_client, make_vectorstore, ) client make_client() doc_paths list_doc_paths() if not doc_paths: raise SystemExit(fno documents found in: {DOC_DIR}) texts, metadatas load_chunks(doc_paths) if not texts: raise SystemExit(fno chunks generated from: {DOC_DIR}) # 第 4 步删旧重建保证每次都是目录里的最新文档。 if client.collection_exists(COLLECTION_NAME): client.delete_collection(COLLECTION_NAME) # 第 4 步创建同时支持 dense BM25 sparse 的 collection。 client.create_collection( collection_nameCOLLECTION_NAME, vectors_configVectorParams(size1536, distanceDistance.COSINE), sparse_vectors_config{langchain-sparse: SparseVectorParams()}, ) # 第 2 步 第 3 步写入 dense embedding、BM25 sparse embedding 和元数据。 vectorstore make_vectorstore(client) vectorstore.add_texts(texts, metadatasmetadatas) print(fstored chunks: {len(texts)}) print(fcollection: {COLLECTION_NAME}) print(fqdrant: {QDRANT_URL}) print(sources:) for path in doc_paths: print(f- {path.name}) # 第 5 步数据已经持久化到本地 Qdrant 服务。 client.close()3askimport argparse from langchain_core.prompts import ChatPromptTemplate from common import COLLECTION_NAME, make_client, make_llm, make_vectorstore parser argparse.ArgumentParser(description小家电 demo 查询) parser.add_argument(question, nargs*, help要查询的问题) args parser.parse_args() client make_client() if not client.collection_exists(COLLECTION_NAME): raise SystemExit(先运行 store.py 生成 collection。) # 第 6 步查询时同时走 dense 和 BM25 sparse结果由 Qdrant 做混合检索。 vectorstore make_vectorstore(client) if args.question: question .join(args.question).strip() else: question input(Question: ).strip() or 第一次使用前要做什么 # 第 6(1)(2) 步把问题送进混合检索取回最相关的 chunk。 docs vectorstore.similarity_search(question, k3) print(retrieved chunks:) for i, doc in enumerate(docs, 1): source doc.metadata.get(source, unknown) chunk doc.metadata.get(chunk, ?) print(f\n--- chunk {i} ({source}#{chunk}) ---) print(doc.page_content[:200]) # 第 6(5) 步把检索结果拼成上下文再交给大模型生成答案。 context \n\n.join(doc.page_content for doc in docs) prompt ChatPromptTemplate.from_messages( [ (system, 你是小家电使用与报修助手只能根据上下文回答保持简短。), (human, 上下文:\n{context}\n\n问题: {question}), ] ) llm make_llm() resp llm.invoke(prompt.format_messages(contextcontext, questionquestion)) print(\nanswer:) print(resp.content) client.close()1.2、测试运行store.py重新建collection可以看到对比之前的point多了langchain-sparse也就是稀疏向量检索。运行ask.pyretrieved chunks:--- chunk 1 (repair.txt#2) ---现。报修步骤1. 准备城市、联系人、联系电话、购买日期和故障描述。2. 拍下铭牌、外观和故障视频方便客服判断。3. 拨打对应城市热线报出型号、序列号和故障现象。4. 如客服建议寄修按要求打包并保留发票和配件。5. 维修完成后先做一次空载检查再加水试烧。补充说明- 如果闻到焦糊味、看到冒烟或外壳发烫明显请立刻断电。- 如果发生进水、摔落或烧焦建议先停用再联系维修--- chunk 2 (usecase.md#1) ---# 迷你电热水壶使用手册## 适用范围适用于宿舍、办公室和值班室使用的迷你电热水壶。容量一般不超过 1L适合快速烧水、冲泡饮品和少量热水需求。## 开箱检查1. 检查壶体、底座、电源线、插头是否完好。2. 第一次使用前用清水冲洗内胆和壶盖不要直接空烧。3. 如果发现裂纹、异味或零件松动先停止使用并联系报修。## 正常使用方法1. 将水壶放在平稳、干燥、耐热的台面上。--- chunk 3 (usecase.md#2) ---倒水时握住手柄避免触碰壶身高温区域。## 清洁与保养- 每次使用后先断电等壶体冷却再清洁。- 定期擦拭壶身和底座底座不要进水。- 如果水垢明显可以用温水加少量柠檬酸浸泡 10 分钟再充分冲洗。- 长时间不用时清洗干净后晾干收纳。## 注意事项- 不要干烧不要把底座浸入水中。- 不要在台面边缘、潮湿环境或易燃物旁使用。- 如果出现异味、漏水、加热缓慢或自动断电异常answer:第一次使用前要用清水把内胆和壶盖冲洗干净不要直接空烧。1.3、进一步验证多路检索这里 hybrid_store make_vectorstore(client) 是混合一把检索了如果想验证不同的检索方式返回了哪些chunk我们来测试下dense_store make_vectorstore(client, RetrievalMode.DENSE) sparse_store make_vectorstore(client, RetrievalMode.SPARSE)# 去重再叠加from qdrant_client.models import Distance, SparseVectorParams, VectorParams from common import ( COLLECTION_NAME, DOC_DIR, QDRANT_URL, list_doc_paths, load_chunks, make_client, make_vectorstore, ) client make_client() doc_paths list_doc_paths() if not doc_paths: raise SystemExit(fno documents found in: {DOC_DIR}) texts, metadatas load_chunks(doc_paths) if not texts: raise SystemExit(fno chunks generated from: {DOC_DIR}) # 第 4 步删旧重建保证每次都是目录里的最新文档。 if client.collection_exists(COLLECTION_NAME): client.delete_collection(COLLECTION_NAME) # 第 4 步创建同时支持 dense BM25 sparse 的 collection。 client.create_collection( collection_nameCOLLECTION_NAME, vectors_configVectorParams(size1536, distanceDistance.COSINE), sparse_vectors_config{langchain-sparse: SparseVectorParams()}, ) # 第 2 步 第 3 步写入 dense embedding、BM25 sparse embedding 和元数据。 vectorstore make_vectorstore(client) vectorstore.add_texts(texts, metadatasmetadatas) print(fstored chunks: {len(texts)}) print(fcollection: {COLLECTION_NAME}) print(fqdrant: {QDRANT_URL}) print(sources:) for path in doc_paths: print(f- {path.name}) # 第 5 步数据已经持久化到本地 Qdrant 服务。 client.close()输出dense hits:1. repair.txt#22. usecase.md#13. usecase.md#24. repair.txt#15. flower.md#16. flower.md#27. flower.md#48. flower.md#59. flower.md#3bm25 hits:1. usecase.md#12. usecase.md#23. repair.txt#24. repair.txt#1hybrid hits:1. usecase.md#1 [vectorkeyword]# 迷你电热水壶使用手册## 适用范围适用于宿舍、办公室和值班室使用的迷你电热水壶。容量一般不超过 1L适合快速烧水、冲泡饮品和少量热水需求。## 开箱检查1. 检查壶体、底座、电源线、插头是否完好。2. 第一次使用前用清水冲洗内胆和壶盖不要直接空烧。3. 如果发现裂纹、异味或零件松动先停止使用并联系报修。## 正常使用方法1. 将水壶放在平稳、干燥、耐热的台面上。2. repair.txt#2 [vectorkeyword]现。报修步骤1. 准备城市、联系人、联系电话、购买日期和故障描述。2. 拍下铭牌、外观和故障视频方便客服判断。3. 拨打对应城市热线报出型号、序列号和故障现象。4. 如客服建议寄修按要求打包并保留发票和配件。5. 维修完成后先做一次空载检查再加水试烧。补充说明- 如果闻到焦糊味、看到冒烟或外壳发烫明显请立刻断电。- 如果发生进水、摔落或烧焦建议先停用再联系维修3. usecase.md#2 [vectorkeyword]倒水时握住手柄避免触碰壶身高温区域。## 清洁与保养- 每次使用后先断电等壶体冷却再清洁。- 定期擦拭壶身和底座底座不要进水。- 如果水垢明显可以用温水加少量柠檬酸浸泡 10 分钟再充分冲洗。- 长时间不用时清洗干净后晾干收纳。## 注意事项- 不要干烧不要把底座浸入水中。- 不要在台面边缘、潮湿环境或易燃物旁使用。- 如果出现异味、漏水、加热缓慢或自动断电异常answer:第一次使用前要用清水把内胆和壶盖冲洗干净不要直接空烧如果发现裂纹、异味或零件松动要先停止使用并联系报修。1.4、向量增量更新模拟下向量增量更新from argparse import ArgumentParser from pathlib import Path from qdrant_client import models from common import COLLECTION_NAME, DOC_DIR, QDRANT_URL, load_chunks, make_client, make_vectorstore DEMO_DIR Path(__file__).resolve().parents[1] def resolve_input_path(raw_path: str) - Path: # 允许传完整路径、demo 相对路径或者直接传文件名。 path Path(raw_path) candidates [ path, DEMO_DIR / path, DOC_DIR / path.name, ] for candidate in candidates: if candidate.exists(): return candidate raise SystemExit(ffile not found: {raw_path}) def ensure_collection_exists(client) - None: if not client.collection_exists(COLLECTION_NAME): raise SystemExit(先运行 store.py 创建 collection再用增量脚本。) def delete_by_filename(client, filename: str) - None: source_name Path(filename).name # 按 metadata.source 删除同名文件对应的所有 chunk。 payload_filter models.Filter( must[ models.FieldCondition( keymetadata.source, matchmodels.MatchValue(valuesource_name), ) ] ) client.delete(collection_nameCOLLECTION_NAME, points_selectorpayload_filter) print(fdeleted chunks for: {source_name}) def add_files(client, vectorstore, paths: list[str]) - None: for raw_path in paths: path resolve_input_path(raw_path) # 先删同名旧 chunk再写入新 chunk避免重复。 delete_by_filename(client, path.name) texts, metadatas load_chunks([path]) vectorstore.add_texts(texts, metadatasmetadatas) print(fadded chunks from {path.name}: {len(texts)}) def main() - None: parser ArgumentParser(descriptionQdrant 增量更新脚本) subparsers parser.add_subparsers(destcommand, requiredTrue) add_parser subparsers.add_parser(add, help新增或替换一个或多个文档) add_parser.add_argument(paths, nargs, help要写入的文档路径) delete_parser subparsers.add_parser(delete, help按文件名删除 chunk) delete_parser.add_argument(filenames, nargs, help要删除的文件名) args parser.parse_args() client make_client() ensure_collection_exists(client) vectorstore make_vectorstore(client) if args.command add: add_files(client, vectorstore, args.paths) elif args.command delete: for filename in args.filenames: delete_by_filename(client, filename) else: raise SystemExit(funknown command: {args.command}) print(fqdrant: {QDRANT_URL}) client.close() if __name__ __main__: main()执行python demo\qdrant_demo\store_incremental.py add docs\small_appliance_kb\info.txt可以看到info.txt新增到向量库了执行python demo\qdrant_demo\store_incremental.py delete info.txt删除成功