Haystack 集成 Elasticsearch 检索指南DocumentStore 与 BM25 / Embedding / SQL 三类检索器完全解析v2.18【免费下载链接】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/haystackHaystack 通过elasticsearch-haystack集成包将 Elasticsearch 8 无缝接入 LLM 应用管线ElasticsearchDocumentStore提供文档存取与近似最近邻ANN检索能力ElasticsearchBM25Retriever、ElasticsearchEmbeddingRetriever与ElasticsearchSQLRetriever则分别覆盖关键词检索、语义向量检索与结构化 SQL 查询三类场景。本文以 version-2.18 的 Elasticsearch API 参考文档 为骨架结合当前仓库中的用户指南与检索器文档完整讲解从环境搭建、索引初始化到三类检索器的参数语义、序列化与异步 API帮助你在 RAG、语义搜索与混合检索管线中直接落地 Elasticsearch 后端。一、集成概览一个后端、三条检索路径Elasticsearch 集成的核心价值在于同一个ElasticsearchDocumentStore可以同时承载稀疏检索BM25 关键词匹配与稠密检索embedding 向量相似度方便在 PoC 阶段直接对比 dense 与 sparse 两种检索方案的效果并平滑迁移到生产环境。文档存储支持 ANN 近似最近邻搜索。从 API 参考文档 的模块结构看集成主要包含四部分模块组件检索方式...retrievers.elasticsearch.bm25_retrieverElasticsearchBM25RetrieverBM25 关键词算法...retrievers.elasticsearch.embedding_retrieverElasticsearchEmbeddingRetriever向量相似度ANN...retrievers.elasticsearch.sql_retrieverElasticsearchSQLRetrieverElasticsearch SQL 原生查询...document_stores.elasticsearch.document_storeElasticsearchDocumentStore文档存储与索引管理另外文档还包含haystack_integrations.document_stores.elasticsearch.filters模块元数据过滤相关支持 Haystack 元数据过滤语法在 Elasticsearch 端的落地。核心库中对 Elasticsearch 的引用也贯穿于 自动合并检索器 与 句子窗口检索器 等核心组件这些组件通过与兼容的文档存储配合可叠加父子文档、上下文窗口等高级检索策略。二、环境准备安装 Elasticsearch 与集成包Haystack 支持 Elasticsearch 8。官方推荐使用 Docker 快速拉起单节点实例docker pull docker.elastic.co/elasticsearch/elasticsearch:8.19.7 docker run -p 9200:9200 -e discovery.typesingle-node -e ES_JAVA_OPTS-Xms1024m -Xmx1024m -e xpack.security.enabledfalse docker.elastic.co/elasticsearch/elasticsearch:8.19.7随后安装集成包pip install elasticsearch-haystack如需运行向量检索示例还需要安装 Sentence Transformers 嵌入器集成包pip install sentence-transformers-haystack注意上述 Docker 命令中xpack.security.enabledfalse仅用于本地演示生产环境务必启用安全认证详见官方连接文档确保只有授权用户能访问数据。三、ElasticsearchDocumentStore索引初始化与文档管理3.1 初始化与参数语义ElasticsearchDocumentStore同时支持 Elastic Cloud 与自建集群。两种典型初始化方式# 方式一Elastic Cloud通过环境变量提供 API Key from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore document_store ElasticsearchDocumentStore( api_key_idSecret.from_env_var(ELASTIC_API_KEY_ID, strictFalse), api_keySecret.from_env_var(ELASTIC_API_KEY, strictFalse), ) # 方式二自建实例本地演示安全已禁用 document_store ElasticsearchDocumentStore(hostshttp://localhost:9200)完整构造函数签名__init__( *, hosts: Hosts | None None, custom_mapping: dict[str, Any] | None None, index: str default, api_key: Secret | str | None Secret.from_env_var(ELASTIC_API_KEY, strictFalse), api_key_id: Secret | str | None Secret.from_env_var(ELASTIC_API_KEY_ID, strictFalse), embedding_similarity_function: Literal[cosine, dot_product, l2_norm, max_inner_product] cosine, sparse_vector_field: str | None None, ingest_pipeline: str | None None, **kwargs: Any ) - None各参数的核心语义如下参数默认值说明hostsNoneElasticsearch 客户端连接的节点地址列表custom_mappingNone自定义索引映射不传则使用默认映射indexdefault使用的 Elasticsearch 索引名索引不存在时会自动创建api_key读取ELASTIC_API_KEYAPI Key 的 Secret 对象或 base64 编码的id:secret拼接串以:分隔api_key_id读取ELASTIC_API_KEY_IDAPI Key ID 的 Secret 对象可与api_key二选一或同时提供embedding_similarity_functioncosine文档 embedding 相似度函数可选cosine/dot_product/l2_norm/max_inner_product仅在索引不存在并新建时生效选择时需结合嵌入模型说明sparse_vector_fieldNone若设置为该名称的 Elasticsearch 字段类型sparse_vector存储稀疏 embedding未设置时 Document 上的sparse_embedding数据写入时会被静默丢弃ingest_pipelineNoneElasticsearch ingest pipeline 的 id用于在索引时通过 inference processor如 ELSER 或稠密模型生成 embedding而无需在 Haystack 侧运行 embedder 组件首尾空白会被去除**kwargs—透传给 Elasticsearch 客户端Elasticsearch/AsyncElasticsearch的其余参数认证方面默认从环境变量加载Secret也可用Secret.from_token()从 token 加载。存储对象同时暴露client同步Elasticsearch客户端按需惰性初始化与async_client异步AsyncElasticsearch客户端两个属性。3.2 使用 ingest pipeline 生成 embedding 的约束当通过ingest_pipeline使用 inference processor 时有三个关键要求input_output必须正确指向输出字段output_field必须等于embedding稠密检索或sparse_vector_field的值ELSER / 稀疏检索。Elasticsearch 默认写入的ml.inference.tag目标字段不会被 Haystack 检索器找到。不要同时运行 Haystack 的DocumentEmbedder若文档到达时已带有预计算的embeddingingest pipeline 会用自身模型的向量覆盖它导致存储向量与查询向量静默不一致检索时出现错配。自定义映射需包含输出字段如果提供了custom_mapping必须包含类型正确的输出字段dense_vector或sparse_vector。另外关于稀疏 embedding 有一个实现细节Elasticsearch 不会把 inference pipeline 生成的sparse_vector数据存进_source它只进入倒排索引。Haystack 通过在每次搜索时借助 ES 的fieldsAPI 请求该字段从而正确填充返回 Document 的sparse_embedding。3.3 文档写入、删除与刷新语义写入文档write_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE, refresh: Literal[wait_for, True, False] wait_for, ) - intpolicy遇到同 ID 文档时的DuplicatePolicy策略如NONE、SKIP、FAIL、OVERWRITE。当策略为FAIL或NONE且同 ID 文档已存在时抛出DuplicateDocumentError。refresh控制写入对搜索操作可见的时机True操作后立即强制刷新False不刷新批量操作时性能更好wait_for等待下一个刷新周期默认保证写入即读一致性。返回实际写入的文档数documents非 Document 列表时抛ValueError写入出错抛DocumentStoreError。删除文档delete_documents(document_ids, refreshwait_for)按文档 ID 列表删除。delete_all_documents(recreate_indexFalse, refreshTrue)清空整个存储。recreate_indexTrue时删除索引并按原映射/设置重建否则走delete_by_query批量删除。delete_by_filter(filters, refreshFalse) - int按元数据过滤条件删除返回删除数量。按条件更新update_by_filter(filters, meta, refreshFalse) - int可批量更新命中过滤条件的文档元数据。3.4 检索、统计与元数据管理方法DocumentStore 的主查询方法是filter_documents(filters)它返回所有匹配过滤条件的 Documentcount_documents()返回文档总数count_documents_by_filter(filters)统计命中过滤条件的数量。面向元数据管理还提供了一组实用方法count_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]统计各指定元数据字段的唯一值个数字段名可带或不带meta.前缀若请求的字段不在索引映射中抛ValueError。get_metadata_fields_info() - dict[str, dict[str, str]]返回索引中字段的类型信息。例如写入Document(contentDoc 1, meta{category: A, status: active, priority: 1})与Document(contentDoc 2, meta{category: B, status: inactive})后返回{ content: {type: text}, category: {type: keyword}, status: {type: keyword}, priority: {type: long}, }get_metadata_field_min_max(metadata_field)返回某元数据字段的最小值与最大值{min: ..., max: ...}。get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]分页获取字段唯一值。底层基于 composite 聚合仅支持游标迭代因此from_偏移需要通过重复抓取并丢弃前面桶来模拟成本随from_线性增长而非随size增长total_count基于近似基数聚合计算超高基数字段下可能不精确。search_term为大小写不敏感的模糊子串匹配匹配字段值本身而非文档内容该匹配通过服务端脚本实现在大语料上开销较大。3.5 同步与异步 API 全覆盖上述所有方法均有对应的*_async版本write_documents_async、delete_documents_async、delete_all_documents_async、delete_by_filter_async、update_by_filter_async、filter_documents_async、count_documents_async、count_documents_by_filter_async、count_unique_metadata_by_filter_async、get_metadata_fields_info_async、get_metadata_field_min_max_async、get_metadata_field_unique_values_async。此外close()与close_async()分别释放同步与异步客户端资源。四、ElasticsearchBM25Retriever轻量关键词检索4.1 原理与适用场景ElasticsearchBM25Retriever基于 BM25 算法从ElasticsearchDocumentStore检索文档通过计算查询与文档之间的加权词重叠度来判定相似性只兼容ElasticsearchDocumentStore。由于其本质是词面匹配特别适合人名、产品名、ID、明确的错误信息等精确匹配场景BM25 轻量简单在域外数据上往往不逊于复杂的 embedding 方案。4.2 基本用法from haystack import Document from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchBM25Retriever document_store ElasticsearchDocumentStore(hostshttp://localhost:9200) retriever ElasticsearchBM25Retriever(document_storedocument_store) documents [ Document(textMy name is Carla and I live in Berlin), Document(textMy name is Paul and I live in New York), Document(textMy name is Silvano and I live in Matera), Document(textMy name is Usagi Tsukino and I live in Tokyo), ] document_store.write_documents(documents) result retriever.run(queryWho lives in Berlin?) for doc in result[documents]: print(doc.content)4.3 初始化参数__init__( *, document_store: ElasticsearchDocumentStore, filters: dict[str, Any] | None None, fuzziness: str AUTO, top_k: int 10, scale_score: bool False, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数默认值说明document_store—ElasticsearchDocumentStore实例必填非该类型实例时抛ValueErrorfiltersNone应用于检索结果的过滤条件语法详见ElasticsearchDocumentStore.filter_documentsfuzzinessAUTO传给 Elasticsearch 的模糊匹配参数inexact fuzzy matching用于容忍拼写错误top_k10最多返回的 Document 数量scale_scoreFalse为True时将 Document 的分数缩放到 0–1 区间filter_policyFilterPolicy.REPLACE决定初始化过滤器与运行时过滤器如何合并应用的策略枚举来自 Haystack 核心库4.4 run / run_asyncrun(query: str, filters: dict[str, Any] | None None, top_k: int | None None) - dict[str, list[Document]]query为要在 Document 文本中搜索的字符串filters在运行时可覆盖/合并初始化时的过滤器具体行为取决于filter_policytop_k可覆盖初始化值。返回字典键为documents匹配查询的 Document 列表。run_async为异步版本签名与返回结构一致。4.5 在 RAG 管线中的位置按组件文档的定位它通常位于 RAG 管线的PromptBuilder之前、语义搜索管线的末端或抽取式 QA 管线的 Reader 之前。一个完整 RAG 示例使用ChatPromptBuilderOpenAIChatGeneratorAnswerBuilder写入时用DuplicatePolicy.SKIP避免重复运行报错from haystack import Document, Pipeline from haystack.components.builders import AnswerBuilder, ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchBM25Retriever from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore document_store ElasticsearchDocumentStore(hostshttp://localhost:9200/) documents [Document(contentThere are over 7,000 languages spoken around the world today.)] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever ElasticsearchBM25Retriever(document_storedocument_store) prompt_template [ChatMessage.from_user(Given these documents, answer the question.\nDocuments:\n{% for doc in documents %}{{ doc.content }}{% endfor %}\n\nQuestion: {{question}}\nAnswer:)] rag_pipeline Pipeline() rag_pipeline.add_component(nameretriever, instanceretriever) rag_pipeline.add_component(nameprompt_builder, instanceChatPromptBuilder(templateprompt_template, required_variables*)) rag_pipeline.add_component(namellm, instanceOpenAIChatGenerator()) rag_pipeline.add_component(nameanswer_builder, instanceAnswerBuilder()) rag_pipeline.connect(retriever, prompt_builder.documents) rag_pipeline.connect(prompt_builder.prompt, llm.messages) rag_pipeline.connect(llm.replies, answer_builder.replies) rag_pipeline.connect(retriever, answer_builder.documents) question How many languages are spoken around the world today? result rag_pipeline.run({retriever: {query: question}, prompt_builder: {question: question}, answer_builder: {query: question}}) print(result[answer_builder][answers][0].data)五、ElasticsearchEmbeddingRetriever语义向量检索5.1 原理与前置条件ElasticsearchEmbeddingRetriever通过向量相似度从ElasticsearchDocumentStore检索文档比较查询 embedding 与文档 embedding返回最相关的文档。使用时必须保证查询与文档两侧的 embedding 都可用——通常由索引管线中的 Document Embedder 和查询管线中的 Text Embedder 提供。embedding 相似度函数embedding_similarity_function必须在初始化ElasticsearchDocumentStore时定义仅在索引新建时生效。5.2 基本用法from haystack import Document from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersTextEmbedder from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchEmbeddingRetriever document_store ElasticsearchDocumentStore(hostshttp://localhost:9200) retriever ElasticsearchEmbeddingRetriever(document_storedocument_store) documents [ Document(textMy name is Carla and I live in Berlin), Document(textMy name is Paul and I live in New York), Document(textMy name is Silvano and I live in Matera), Document(textMy name is Usagi Tsukino and I live in Tokyo), ] document_store.write_documents(documents) te SentenceTransformersTextEmbedder() query_embeddings te.run(Who lives in Berlin?)[embedding] result retriever.run(queryquery_embeddings) for doc in result[documents]: print(doc.content)5.3 初始化参数__init__( *, document_store: ElasticsearchDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, num_candidates: int | None None, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数默认值说明document_store—ElasticsearchDocumentStore实例必填非该类型时抛ValueErrorfiltersNone检索过滤条件在近似 KNN 搜索期间应用以确保返回恰好top_k个匹配文档top_k10最多返回的 Document 数量num_candidatesNone每个分片上的近似最近邻候选数默认top_k * 10增大可提升检索准确率但会降低检索速度属于速度—精度权衡的高级调参项filter_policyFilterPolicy.REPLACE初始化过滤器与运行时过滤器的合并策略5.4 run / run_asyncrun(query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None) - dict[str, list[Document]]query_embedding为查询的 embedding 向量float 列表filters同样在近似 KNN 搜索期间应用以保证top_k命中top_k可覆盖初始化值。返回documents键值为与query_embedding最相似的 Document 列表。run_async提供异步等价实现。5.5 管线化用法在查询管线中将 Text Embedder 的输出连接到 Retriever 的query_embedding输入索引侧先使用SentenceTransformersDocumentEmbedder生成文档向量from haystack import Document, Pipeline from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchEmbeddingRetriever from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore model BAAI/bge-large-en-v1.5 document_store ElasticsearchDocumentStore(hostshttp://localhost:9200/) documents [Document(contentThere are over 7,000 languages spoken around the world today.)] doc_embedder SentenceTransformersDocumentEmbedder(modelmodel) docs_with_embeddings doc_embedder.run(documents) document_store.write_documents(docs_with_embeddings.get(documents), policyDuplicatePolicy.SKIP) query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder(modelmodel)) query_pipeline.add_component(retriever, ElasticsearchEmbeddingRetriever(document_storedocument_store)) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) result query_pipeline.run({text_embedder: {text: How many languages are there?}}) print(result[retriever][documents][0])六、ElasticsearchSQLRetriever直连 Elasticsearch SQL API6.1 定位结构化数据访问与前两类检索器不同ElasticsearchSQLRetriever不把查询匹配到文档而是把原生 Elasticsearch SQL 语句直接下发执行并返回 SQL API 的原始 JSON 响应。它适合在运行时获取元数据、做聚合统计计数、均值等以及其他结构化数据访问。6.2 初始化与调用__init__( *, document_store: ElasticsearchDocumentStore, raise_on_failure: bool True, fetch_size: int | None None ) - Noneraise_on_failure为True默认时SQL API 调用失败抛出异常为False时记录 warning 并返回空字典。fetch_size每页抓取的结果条数不传则使用 Elasticsearch 服务端默认值。from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchSQLRetriever document_store ElasticsearchDocumentStore(hostshttp://localhost:9200) retriever ElasticsearchSQLRetriever(document_storedocument_store) result retriever.run(querySELECT content, category FROM my_index WHERE category \A\) # result[result] 包含 Elasticsearch 原始 JSON 响应 # result[result][columns] - 列元数据 # result[result][rows] - 数据行run/run_async的完整签名为run(query: str, document_store: ElasticsearchDocumentStore | None None, fetch_size: int | None None) - dict[str, dict[str, Any]]其中document_store与fetch_size均可选传入以覆盖初始化值。返回字典键为result其值为 Elasticsearch 的原始 JSON 响应dict或出错时的空 dict。6.3 聚合查询示例由于返回的是原始响应可以执行普通文档检索器不支持的聚合操作output retriever.run(querySELECT COUNT(*) AS doc_count FROM my_index) print(output[result][rows]) # 例如 [[3]]对可能出错或格式非法的查询可设置raise_on_failureFalse失败时仅告警并返回空字典避免中断管线。七、进阶混合检索Hybrid与序列化虽然 v2.18 的 API 参考文档聚焦上述三类检索器与 DocumentStore但同一集成包还提供ElasticsearchHybridRetriever组件文档这是一个基于 Haystack SuperComponent 实现的单组件混合检索器内置 Text Embedder并行执行 BM25 与向量检索再通过DocumentJoiner默认 Reciprocal Rank Fusion 互惠排名融合合并重排结果。可通过top_k_bm25、fuzziness、filters_bm25、scale_score、filter_policy_bm25、top_k_embedding、filters_embedding、num_candidates、filter_policy_embedding等参数分别调优两条检索支路并直接暴露join_mode、weights、top_k、sort_by_score等合并参数。所有组件的序列化能力保持一致to_dict() - dict[str, Any]把组件序列化为字典ElasticsearchDocumentStore等存储对象同样支持。from_dict(data) - 组件实例从字典反序列化恢复组件便于通过 YAML/JSON 定义与分享管线。close()/close_async()释放底层文档存储的同步/异步资源用于资源生命周期管理。八、小结围绕 Elasticsearch 后端Haystack 提供了从索引管理到多路检索的完整闭环ElasticsearchDocumentStore负责索引自动创建、映射定制、ingest pipeline 集成、批量写入/删除/更新以及丰富的元数据统计能力ElasticsearchBM25Retriever提供轻量关键词检索支持fuzziness模糊匹配与分数缩放ElasticsearchEmbeddingRetriever提供基于 ANN 的语义检索支持num_candidates精度调优ElasticsearchSQLRetriever则打通了结构化 SQL 查询通道。三者共享同步/异步双 API 与统一的序列化机制可直接嵌入 RAG、语义搜索、抽取式 QA 与混合检索管线。若要进一步了解安装细节、检索器在管线中的典型位置或 Hybrid 用法可继续阅读仓库内的 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），仅供参考