Haystack 检索器（Retrievers）API 全解析：从 BM25、向量语义检索到自动合并与句子窗口

Haystack 检索器RetrieversAPI 全解析从 BM25、向量语义检索到自动合并与句子窗口【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文系统梳理 Haystack 2.x 中haystack.components.retrievers核心检索器组件的 API 与实战用法涵盖InMemoryBM25Retriever、InMemoryEmbeddingRetriever、FilterRetriever、AutoMergingRetriever与SentenceWindowRetriever五个组件。你将掌握每个组件的构造参数、run/run_async调用契约、过滤器Filters与FilterPolicy策略的底层合并逻辑并能在 RAG 与语义搜索流水线中正确选用与编排这些检索器。文中所有 API 签名、示例与参数说明均基于本仓库docs-website/reference_versioned_docs/version-2.20/haystack-api/retrievers_api.md及对应源码可放心对照验证。检索器在 Haystack 流水线中的定位在 Haystack 中检索器Retriever负责“在 Document Store 中扫描并返回与查询相关的一组候选文档”是整个 RAG检索增强生成与语义搜索流水线的第一环。它通常接收用户查询或查询向量从文档库中筛选出最相关的一批文档再交给后续的 Ranker、Prompt Builder 与 Generator 处理。按检索机制可划分为四类关键词检索InMemoryBM25Retriever基于 BM25 词频-逆文档频率算法适合精确词匹配与 OOV词典外场景。向量语义检索InMemoryEmbeddingRetriever基于查询向量与文档向量的相似度适合理解语义近似。元数据/过滤检索FilterRetriever仅依据过滤器条件筛选文档不依赖查询文本。上下文增强检索AutoMergingRetriever层级文档自动合并与SentenceWindowRetriever句子窗口上下文扩展二者都是对其他检索器结果的二次加工。本文后续小节将逐一讲解上述组件的构造参数、输入输出契约与源码级实现细节。InMemoryBM25Retriever基于关键词的 BM25 检索组件概览InMemoryBM25Retriever使用基于关键词的 BM25 算法返回与查询最相似的文档必须搭配InMemoryDocumentStore使用。其核心实现位于 bm25_retriever.py检索逻辑委托给 Document Store 的bm25_retrieval方法见 document_store.py。快速上手示例from haystack import Document from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.document_stores.in_memory import InMemoryDocumentStore docs [ Document(contentPython is a popular programming language), Document(contentpython ist eine beliebte Programmiersprache), ] doc_store InMemoryDocumentStore() doc_store.write_documents(docs) retriever InMemoryBM25Retriever(doc_store) result retriever.run(queryProgrammiersprache) print(result[documents])构造参数详解def __init__(document_store: InMemoryDocumentStore, filters: Optional[dict[str, Any]] None, top_k: int 10, scale_score: bool False, filter_policy: FilterPolicy FilterPolicy.REPLACE)参数默认值说明document_store必填InMemoryDocumentStore实例检索的目标文档库。filtersNone过滤器字典用于收窄检索空间如只检索某个分类下的文档。top_k10返回的最大文档数。若top_k 0构造时抛出ValueError。scale_scoreFalse为True时将检索得分缩放到 0~1 区间1 表示极度相关为False时返回原始相似度得分。filter_policyREPLACE过滤器应用策略详见下文「FilterPolicy 过滤器策略」小节。从源码看构造时会先做类型与取值校验document_store必须是InMemoryDocumentStore实例否则抛TypeErrortop_k必须大于 0否则抛ValueError。run 与 run_async 调用契约component.output_types(documentslist[Document]) def run(query: str, filters: Optional[dict[str, Any]] None, top_k: Optional[int] None, scale_score: Optional[bool] None) component.output_types(documentslist[Document]) async def run_async(query: str, filters: Optional[dict[str, Any]] None, top_k: Optional[int] None, scale_score: Optional[bool] None)query查询字符串必填。若为空字符串Document Store 底层会抛出ValueError。filters/top_k/scale_score均为可选若未传入则回退到构造时设置的默认值filters会先经apply_filter_policy处理见下文。返回值字典{documents: [Document, ...]}即检索到的候选文档列表。run_async为异步版本内部调用bm25_retrieval_async在需要高并发查询吞吐的流水线中可替代run使用。InMemoryEmbeddingRetriever基于向量的语义检索组件概览InMemoryEmbeddingRetriever返回与查询语义最相似的文档同样必须搭配InMemoryDocumentStore。与 BM25 不同它需要查询向量与文档向量均已就绪在索引indexing流水线中使用DocumentEmbedder如SentenceTransformersDocumentEmbedder为文档生成向量在查询query流水线中使用TextEmbedder如SentenceTransformersTextEmbedder为查询生成向量再传给检索器。实现位于 embedding_retriever.py向量检索逻辑委托给embedding_retrieval方法见 document_store.py。快速上手示例from haystack import Document from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore docs [ Document(contentPython is a popular programming language), Document(contentpython ist eine beliebte Programmiersprache), ] doc_embedder SentenceTransformersDocumentEmbedder() doc_embedder.warm_up() docs_with_embeddings doc_embedder.run(docs)[documents] doc_store InMemoryDocumentStore() doc_store.write_documents(docs_with_embeddings) retriever InMemoryEmbeddingRetriever(doc_store) query Programmiersprache text_embedder SentenceTransformersTextEmbedder() text_embedder.warm_up() query_embedding text_embedder.run(query)[embedding] result retriever.run(query_embeddingquery_embedding) print(result[documents])构造参数详解def __init__(document_store: InMemoryDocumentStore, filters: Optional[dict[str, Any]] None, top_k: int 10, scale_score: bool False, return_embedding: bool False, filter_policy: FilterPolicy FilterPolicy.REPLACE)参数默认值说明document_store必填InMemoryDocumentStore实例。filtersNone过滤器字典收窄检索空间。top_k10返回的最大文档数必须大于 0。scale_scoreFalse是否将相似度得分缩放到 0~1。return_embeddingFalse为True时返回的文档中附带embedding字段为False时仅返回文档本体节省内存与传输开销。filter_policyREPLACE过滤器应用策略。run 与 run_async 调用契约component.output_types(documentslist[Document]) def run(query_embedding: list[float], filters: Optional[dict[str, Any]] None, top_k: Optional[int] None, scale_score: Optional[bool] None, return_embedding: Optional[bool] None) component.output_types(documentslist[Document]) async def run_async(query_embedding: list[float], filters: Optional[dict[str, Any]] None, top_k: Optional[int] None, scale_score: Optional[bool] None, return_embedding: Optional[bool] None)query_embedding查询向量list[float]必填通常来自TextEmbedder的输出。其余可选参数与构造参数同名传入后覆盖默认值。返回值{documents: [Document, ...]}。FilterRetriever纯过滤器检索组件概览FilterRetriever不依赖查询文本而是直接按过滤器条件返回匹配的文档适合作为元数据路由、文档筛选等场景的组件。它接受任意实现了DocumentStore协议的存储实例实现位于 filter_retriever.py底层调用filter_documents。快速上手示例from haystack import Document from haystack.components.retrievers import FilterRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore docs [ Document(contentPython is a popular programming language, meta{lang: en}), Document(contentpython ist eine beliebte Programmiersprache, meta{lang: de}), ] doc_store InMemoryDocumentStore() doc_store.write_documents(docs) retriever FilterRetriever(doc_store, filters{field: lang, operator: , value: en}) # 若在 run 方法中传入 filters则覆盖初始化时设置的 filters result retriever.run(filters{field: lang, operator: , value: de}) print(result[documents])构造与调用契约def __init__(document_store: DocumentStore, filters: Optional[dict[str, Any]] None) component.output_types(documentslist[Document]) def run(filters: Optional[dict[str, Any]] None)document_store任意DocumentStore实例。filters比较过滤器字典形如{field: lang, operator: , value: en}也可使用带operator与conditions的逻辑过滤器。run中未传filters时回退到初始化时设置的过滤器从源码看resolved_filters filters if filters is not None else self.filters。返回值{documents: [Document, ...]}。FilterRetriever的过滤器语法与 Document Store 的元数据过滤metadata filtering保持一致支持、!、、、、、IN、NOT IN等比较运算符以及AND、OR、NOT逻辑组合。FilterPolicy 过滤器策略REPLACE 与 MERGE 的底层逻辑在InMemoryBM25Retriever与InMemoryEmbeddingRetriever中filter_policy决定“初始化过滤器”与“运行时过滤器”如何协同其实现位于 filter_policy.pyREPLACE默认运行时传入的过滤器直接覆盖初始化时的过滤器适合在不同查询间动态切换过滤条件。底层apply_filter_policy在REPLACE分支直接返回runtime_filters or init_filters。MERGE将运行时过滤器与初始化过滤器合并进一步收窄检索范围。合并时根据过滤器形态比较过滤器 vs 逻辑过滤器选择不同的组合函数两个比较过滤器 →combine_two_comparison_filters若字段相同以运行时过滤器为准并记录警告否则用默认逻辑运算符默认AND组合。初始化比较过滤器 运行时逻辑过滤器 →combine_init_comparison_and_runtime_logical_filters仅当逻辑运算符一致时才并入否则忽略初始化过滤器。初始化逻辑过滤器 运行时比较过滤器 →combine_runtime_comparison_and_init_logical_filters字段冲突时运行时过滤器覆盖初始化过滤器。两个逻辑过滤器 →combine_two_logical_filters仅当二者operator相同时合并conditions否则采用运行时过滤器。例如初始化过滤器为{field: meta.type, operator: , value: article}运行时传入{field: meta.rating, operator: , value: 3}在MERGE策略下最终过滤器为{ operator: AND, conditions: [ {field: meta.type, operator: , value: article}, {field: meta.rating, operator: , value: 3}, ] }序列化时filter_policy以字符串形式replace/merge写入init_parameters反序列化时通过FilterPolicy.from_str还原为枚举见 filter_policy.py 与 bm25_retriever.py。AutoMergingRetriever层级文档自动合并检索组件概览AutoMergingRetriever解决“段落被切分成多个 chunk检索时只命中其中部分 chunk导致上下文碎片化”的问题。它假设文档存在层级树结构叶子节点被索引到 Document Store当同一父节点下的叶子文档被命中的比例超过阈值时返回父文档更完整的段落而非零散的叶子文档。其设计理念是如果一个段落被拆成多个叶子 chunk且查询命中了其中多个 chunk那么整个段落往往比单个 chunk 更有信息量。层级结构可通过HierarchicalDocumentSplitter创建。实现位于 auto_merging_retriever.py。支持的 Document Store来自原文档AstraDB、ElasticSearch、OpenSearch、PGVector、Qdrant。由于父文档需通过filter_documents按 ID 查询因此该组件对 Document Store 的过滤能力有要求。快速上手示例from haystack import Document from haystack.components.preprocessors import HierarchicalDocumentSplitter from haystack.components.retrievers.auto_merging_retriever import AutoMergingRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore # 创建 3 层层级文档结构父文档包含 3 个子文档 text The sun rose early in the morning. It cast a warm glow over the trees. Birds began to sing. original_document Document(contenttext) builder HierarchicalDocumentSplitter(block_sizes[10, 3], split_overlap0, split_byword) docs builder.run([original_document])[documents] # 存储 level-1 父文档并初始化检索器 doc_store_parents InMemoryDocumentStore() for doc in docs[documents]: if doc.meta[children_ids] and doc.meta[level] 1: doc_store_parents.write_documents([doc]) retriever AutoMergingRetriever(doc_store_parents, threshold0.5) # 假设从同一父文档检索到 2 个叶子文档该父文档应被返回 # 因为它有 3 个子文档且 threshold0.5而 2/3 0.66(6) 超过阈值 leaf_docs [doc for doc in docs[documents] if not doc.meta[children_ids]] docs retriever.run(leaf_docs[4:6]) # {documents: [Document(id538..), # content: warm glow over the trees. Birds began to sing., # meta: {block_size: 10, parent_id: 835.., children_ids: [c17..., 3ff..., 352...], level: 1, # source_id: 835..., page_number: 1, split_id: 1, split_idx_start: 45})]}构造与调用契约def __init__(document_store: DocumentStore, threshold: float 0.5) component.output_types(documentslist[Document]) def run(documents: list[Document])document_store用于检索父文档的 Document Store父文档需预先写入。threshold合并阈值决定是返回父文档还是保留叶子文档。源码在构造时校验0 threshold 1越界则抛ValueError。run(documents)输入为上游检索器命中的叶子文档列表输出为{documents: [...]}可能混合不同层级的文档。递归合并的底层原理从源码看run首先通过_check_valid_documents校验叶子文档必须具备__parent_id、__level、__block_size等元数据字段这些由HierarchicalDocumentSplitter写入。随后进入递归合并流程_try_merge_level将叶子文档按__parent_id分组无父文档的直接进入返回列表。对每个分组通过filter_documents按 ID 查询父文档计算合并得分score len(child_docs) / len(parent_doc.meta[__children_ids])。若score threshold将父文档作为合并结果否则保留叶子文档。若本轮产生了新合并则对合并出的父文档递归执行下一层合并直至无法继续合并为止。由于层级可以多层嵌套run返回的文档可能同时包含不同层级的结果run_async则使用filter_documents_async实现同样的递归逻辑。SentenceWindowRetriever句子窗口上下文扩展组件概览SentenceWindowRetriever旨在解决“检索命中的 chunk 上下文不足”的问题应放在其他检索器如 BM25、Embedding 检索器之后使用。它根据文档元数据中的source_id来源文档标识与split_idchunk 在原文中的顺序从 Document Store 中取回命中 chunk 前后相邻的文档拼成更完整的上下文窗口。文档必须包含如下元数据source_id标识 chunk 属于同一原始文档split_idchunk 在文档内的位置/顺序。可用source_id_meta_field与split_id_meta_field自定义这两个字段名。实现位于 sentence_window_retriever.py。支持的 Document Store来自原文档Astra、Elasticsearch、OpenSearch、Pgvector、Pinecone、Qdrant。快速上手示例与 BM25 检索器组成流水线from haystack import Document, Pipeline from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.components.retrievers import SentenceWindowRetriever from haystack.components.preprocessors import DocumentSplitter from haystack.document_stores.in_memory import InMemoryDocumentStore splitter DocumentSplitter(split_length10, split_overlap5, split_byword) text ( This is a text with some words. There is a second sentence. And there is also a third sentence. It also contains a fourth sentence. And a fifth sentence. And a sixth sentence. And a seventh sentence ) doc Document(contenttext) docs splitter.run([doc]) doc_store InMemoryDocumentStore() doc_store.write_documents(docs[documents]) rag Pipeline() rag.add_component(bm25_retriever, InMemoryBM25Retriever(doc_store, top_k1)) rag.add_component(sentence_window_retriever, SentenceWindowRetriever(document_storedoc_store, window_size2)) rag.connect(bm25_retriever, sentence_window_retriever) rag.run({bm25_retriever: {query: third}}) # {sentence_window_retriever: {context_windows: [some words. There is a second sentence. # And there is also a third sentence. It also contains a fourth sentence. And a fifth sentence. And a sixth # sentence. And a], context_documents: [[Document(id..., content: some words. There is a second sentence. # And there is , meta: {source_id: ..., page_number: 1, split_id: 1, split_idx_start: 20, # _split_overlap: [{doc_id: ..., range: (20, 43)}, {doc_id: ..., range: (0, 30)}]}), # Document(id..., content: second sentence. And there is also a third sentence. It , # meta: {source_id: 74ea87deb38012873cf8c07e..., page_number: 1, # split_id: 2, split_idx_start: 43, ...}), ...]]}}构造与调用契约def __init__(document_store: DocumentStore, window_size: int 3, *, source_id_meta_field: Union[str, list[str]] source_id, split_id_meta_field: str split_id, raise_on_missing_meta_fields: bool True) component.output_types(context_windowslist[str], context_documentslist[Document]) def run(retrieved_documents: list[Document], window_size: Optional[int] None) component.output_types(context_windowslist[str], context_documentslist[Document]) async def run_async(retrieved_documents: list[Document], window_size: Optional[int] None)构造参数参数默认值说明document_store必填用于取回相邻文档的 Document Store。window_size3命中 chunk 前后各取多少个文档。例如window_size: 2表示取前 2 个、后 2 个。源码校验其必须大于 0。source_id_meta_fieldsource_id来源 ID 元数据字段名。可传单个字段或字段列表传多个字段时要求所有字段值都匹配才视为同一来源。split_id_meta_fieldsplit_id位置 ID 元数据字段名。raise_on_missing_meta_fieldsTrue为True时缺失必需元数据字段的文档触发ValueError为False时跳过其上下文取回但原文档仍会保留在结果中并记录警告日志。run 参数retrieved_documents上游检索器返回的文档列表。window_size可选传入后覆盖构造时的窗口大小同样必须大于 0。返回值两个输出槽位context_windowslist[str]每个元素是对应命中文档拼接后的上下文文本context_documentslist[Document]命中文档与其上下文文档的合集按split_idx_start元数据字段排序。上下文窗口的底层实现从源码看run对每个命中文档执行以下步骤校验文档是否包含source_id_meta_field与split_id_meta_field指定的元数据。计算窗口边界min_before split_id - window_size、max_after split_id window_size。构建 AND 逻辑过滤器split_id落在[min_before, max_after]区间且各来源字段匹配见_build_filter_conditionssentence_window_retriever.py。调用filter_documents取回上下文文档并调用静态方法merge_documents_text拼接文本。merge_documents_text的拼接策略很巧妙若文档带split_idx_start元数据则先按起始位置排序再逐段跳过与已拼接内容的重叠部分重叠区域由DocumentSplitter的split_overlap产生从而得到无重复的连贯文本若缺少该元数据则直接拼接所有内容。context_documents最终按split_id排序返回方便下游组件按原文顺序消费。序列化与反序列化to_dict 与 from_dict上述五个组件均实现了 Haystack 组件的标准序列化协议to_dict()将组件含document_store、过滤策略、窗口大小等init_parameters序列化为字典用于流水线持久化、YAML 配置或分布式调度。filter_policy以字符串枚举值replace/merge写入反序列化时再还原为FilterPolicy枚举。from_dict(cls, data)类方法从字典还原组件实例。对于FilterPolicy字段from_dict会先调用FilterPolicy.from_str完成字符串到枚举的转换再交给default_from_dict完成常规反序列化参见 bm25_retriever.py 与 embedding_retriever.py。这保证了由这些检索器构成的流水线可以无损地保存到 YAML 文件并重新加载运行。选型建议与流水线编排需要精确关键词匹配、处理专业术语或 OOV 词选择InMemoryBM25Retriever无需预先生成向量冷启动成本低。需要语义理解、处理同义表达与跨语言近似选择InMemoryEmbeddingRetriever但需在索引流水线中用DocumentEmbedder生成文档向量、在查询流水线中用TextEmbedder生成查询向量。仅按元数据筛选文档不依赖查询文本选择FilterRetriever适合文档路由、数据治理或结合分类器使用。长文档被切碎导致上下文不完整在 BM25/Embedding 检索器之后串联SentenceWindowRetriever获取命中 chunk 前后文或使用HierarchicalDocumentSplitter构建层级结构后搭配AutoMergingRetriever命中多个子节点时自动回退到父文档。动态过滤需求利用filter_policyMERGE在保持全局过滤约束的同时为每次查询叠加运行时过滤条件。所有组件均提供run与run_async两种调用方式可灵活嵌入同步或异步流水线close/close_async方法用于释放底层 Document Store 的同步/异步资源。相关源码可进一步参阅 retrievers 目录 与 DocumentStore 类型定义。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考