Haystack 集成 ElasticsearchDocumentStore 与三大 Retriever 完整使用指南【免费下载链接】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.19 版 API 参考文档Elasticsearch 集成参考系统讲解如何通过elasticsearch-haystack集成包将 Elasticsearch 用作 Haystack 的文档存储并围绕它构建关键词检索BM25、向量检索Embedding与结构化 SQL 查询三种检索能力。读完本文你将掌握ElasticsearchDocumentStore的初始化与增删改查 API、三个官方 Retriever 的全部参数与同步/异步用法并能在 RAG 与语义搜索 Pipeline 中直接落地使用。一、集成概览为什么在 Haystack 中选择 ElasticsearchElasticsearch 是一个开源的分布式搜索与分析引擎在 Haystack 生态中由独立的elasticsearch-haystack集成包提供支持即haystack_integrations.document_stores.elasticsearch与haystack_integrations.components.retrievers.elasticsearch等命名空间。ElasticsearchDocumentStore同时支持关键词检索与稠密向量检索内置ANNApproximate Nearest Neighbours近似最近邻搜索能力支持大规模向量检索可在一套索引中对比稠密检索与稀疏检索BM25的效果便于从概念验证PoC平滑过渡到生产环境依据 ElasticsearchDocumentStore 使用文档提供ElasticsearchBM25Retriever、ElasticsearchEmbeddingRetriever、ElasticsearchSQLRetriever三种官方检索组件覆盖词法、语义与结构化查询三类场景。官方支持 Elasticsearch 8。在仓库内检索类组件的更多使用说明可参考 ElasticsearchBM25Retriever 使用文档、ElasticsearchEmbeddingRetriever 使用文档 与 ElasticsearchSQLRetriever 使用文档。二、环境准备与安装2.1 启动 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也可以使用elasticsearch-haystack集成仓库中自带的docker-compose.yml一键启动docker compose up2.2 安装 Python 集成包pip install elasticsearch-haystack若要在向量检索示例中使用 Sentence Transformers 嵌入器还需安装pip install sentence-transformers-haystack2.3 最小可用验证from haystack_integrations.document_stores.elasticsearch import ( ElasticsearchDocumentStore, ) from haystack import Document document_store ElasticsearchDocumentStore(hostshttp://localhost:9200) document_store.write_documents( [Document(contentThis is first), Document(contentThis is second)], ) print(document_store.count_documents())三、ElasticsearchDocumentStore连接、索引与核心 APIElasticsearchDocumentStore可连接 Elastic Cloud 或自建集群。初始化时会尝试创建索引若索引已存在则直接复用。3.1 连接方式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), )连接自建 Elasticsearch 实例from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore document_store ElasticsearchDocumentStore(hostshttp://localhost:9200)上面的示例关闭了安全认证仅用于演示基本用法。官方强烈建议启用安全功能确保只有授权用户能访问你的数据。更详细的连接与安全配置请参考 Elasticsearch 官方 Python 客户端连接文档。3.2 构造函数签名与参数详解__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参数类型默认值说明hostsHosts \| NoneNoneElasticsearch 客户端连接的节点地址列表例如[http://localhost:9200]custom_mappingdict[str, Any] \| NoneNone自定义索引映射不提供时使用默认映射indexstrdefaultElasticsearch 中的索引名称api_keySecret \| str \| None环境变量ELASTIC_API_KEY非强制用于认证的 API Key可以是id:secret拼接后的 base64 编码字符串api_key_idSecret \| str \| None环境变量ELASTIC_API_KEY_ID非强制API Key 的 IDembedding_similarity_functionLiteral[...]cosine文档嵌入的相似度函数可选cosine、dot_product、l2_norm、max_inner_product仅在索引不存在、由本组件创建时生效sparse_vector_fieldstr \| NoneNone存储稀疏向量的字段名使用sparse_vector字段类型未设置时写入时文档上的sparse_embedding数据会被静默丢弃ingest_pipelinestr \| NoneNoneElasticsearch 摄取管道ingest pipelineID用于在索引时通过推理处理器如 ELSER 或稠密模型生成嵌入首尾空白会被去除认证通过Secret对象提供默认从环境变量加载。你可以同时提供api_key_id和api_key也可以只提供包含id:secretbase64 编码字符串的api_keySecret实例还可以通过Secret.from_token()从令牌加载。**kwargs中的额外参数会全部透传给底层elasticsearch.Elasticsearch客户端例如verify_certs、timeout、retry_on_timeout等连接选项。3.3 客户端访问同步与异步client: Elasticsearch——返回同步客户端必要时惰性初始化async_client: AsyncElasticsearch——返回异步客户端必要时惰性构造。同步/异步资源分别通过close()与close_async()释放三大 Retriever 也各自提供close()/close_async()来释放底层 DocumentStore 的对应资源。3.4 索引时使用推理处理器inference processors设置ingest_pipeline后可以用 Elasticsearch 的推理处理器在索引时直接生成嵌入例如 ELSER 稀疏模型或稠密模型而无需在 Haystack 侧运行DocumentEmbedder组件。使用时必须满足以下约束处理器需配置input_output使嵌入直接写入正确的字段稠密检索时output_field必须为embeddingELSER/稀疏检索时必须等于sparse_vector_field的值。ES 默认目标ml.inference.tag不会被 Haystack 的检索器找到上游不要再运行 Haystack 的DocumentEmbedder如果文档已带有预计算的embedding摄取管道会用自身模型的向量覆盖它导致检索时存储向量与查询向量静默不一致如果提供了custom_mapping必须将输出字段声明为正确的类型dense_vector或sparse_vector。关于稀疏向量还有一个值得注意的实现细节Elasticsearch 不会把推理管道生成的sparse_vector数据存入_source它只进入倒排索引。Haystack 的做法是每次搜索都通过 ES 的fieldsAPI 请求该字段从而正确填充返回文档的Document.sparse_embedding见原文档 sparse embedding note。3.5 写入、删除与刷新语义写入文档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且文档重复会抛出DuplicateDocumentError若documents不是Document列表则抛出ValueError其余写入错误抛出DocumentStoreErrorrefresh控制写入对搜索操作何时可见True操作后立即强制刷新False不刷新批量操作性能更好wait_for等待下一个刷新周期默认值保证“写入后即可读到”的一致性。write_documents_async提供对应的异步版本。删除文档delete_documents(document_ids: list[str], refresh: Literal[wait_for, True, False] wait_for) - None delete_all_documents(recreate_index: bool False, refresh: bool True) - None delete_by_filter(filters: dict[str, Any], refresh: bool False) - intdelete_all_documents是清空数据的快速方式同时保留索引的设置与映射recreate_indexTrue时删除并重建索引否则通过delete_by_queryAPI 删除全部文档delete_by_filter按过滤器删除匹配文档并返回删除数量过滤器语法遵循 Haystack 元数据过滤规范。更新与统计update_by_filter(filters, meta, refreshFalse) - int——批量更新匹配文档的元数据字段返回更新数量count_documents() - int与count_documents_by_filter(filters) - int——统计文档总数与符合过滤条件的文档数以上方法均有对应的*_async异步版本。3.6 查询与元数据探索 APIfilter_documents(filters)是 DocumentStore 的主查询方法返回所有匹配过滤器的文档过滤器结构遵循 Elasticsearch Query DSL。针对元数据DocumentStore 还提供了一组实用方法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) - dict[str, int | None]——返回某元数据字段在所有文档中的最小/最大值键为min与maxcount_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]——统计各指定字段的唯一值数量字段名可带或不带meta.前缀若请求的字段不存在于索引映射中抛出ValueErrorget_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone)——分页返回某字段的唯一值及其总数返回(list[Any], int)元组。需要注意两个实现细节见原文档说明底层由 composite 聚合实现只支持基于游标的迭代因此from_偏移需要每次调用时重新抓取并丢弃前from_个分桶成本随from_线性增长total_count通过近似基数聚合计算对高基数字段可能不精确search_term的匹配是大小写不敏感的模糊子串匹配由服务端脚本完成在大语料上开销较高需谨慎使用。以上方法同样全部提供*_async异步版本。四、ElasticsearchBM25Retriever关键词检索ElasticsearchBM25Retriever使用BM25 算法从ElasticsearchDocumentStore中检索与查询最相似的文档。BM25 通过计算查询与文档之间的加权词重叠度来判定相似性因此非常适合精确匹配人名、产品名、ID 或定义明确的错误信息等场景算法轻量简单在域外数据上往往不输于更复杂的嵌入方法依据 BM25 Retriever 使用文档。该组件仅与ElasticsearchDocumentStore兼容。4.1 基本用法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) # Add documents to DocumentStore 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.2 初始化参数__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_storeElasticsearchDocumentStore必填与之绑定的 DocumentStore 实例若不是该类型会抛出ValueErrorfiltersdict[str, Any] \| NoneNone应用于检索结果的过滤器详见ElasticsearchDocumentStore.filter_documentsfuzzinessstrAUTO传给 Elasticsearch 的模糊匹配参数即不精确的模糊匹配取值规则遵循 Elasticsearch 官方fuzziness文档top_kint10最多返回的文档数量scale_scoreboolFalse为True时把文档得分缩放到 01 之间filter_policystr \| FilterPolicyFilterPolicy.REPLACE决定过滤器如何被应用见下文 4.44.3 run 与 run_asyncrun( query: str, filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query要在文档文本中搜索的字符串filters运行时过滤器其应用方式取决于初始化时选择的filter_policytop_k本次调用最多返回的文档数可覆盖初始化值。返回{documents: [...]}其中documents是匹配查询的Document列表。run_async提供完全等价的异步版本。4.4 理解 filter_policy 过滤器策略filter_policy是初始化参数类型为str | FilterPolicy决定运行时传入的过滤器与初始化时设置的过滤器之间的关系。其定义位于 Haystack 核心源码 filter_policy.py包含两个取值FilterPolicy.REPLACE值replace运行时过滤器替换初始化过滤器FilterPolicy.MERGE值merge运行时过滤器与初始化过滤器合并运行时值覆盖初始化值。初始化时也可直接传字符串replace或merge。该策略同样适用于下文ElasticsearchEmbeddingRetriever的运行时过滤器。4.5 在 RAG Pipeline 中使用ElasticsearchBM25Retriever最常见的 Pipeline 位置是RAG 中位于PromptBuilder之前、语义搜索管线的末端组件、抽取式问答中位于 Reader 之前。完整示例写入文档时使用DuplicatePolicy.SKIP便于重复运行from haystack_integrations.components.retrievers.elasticsearch import ( ElasticsearchBM25Retriever, ) from haystack_integrations.document_stores.elasticsearch import ( ElasticsearchDocumentStore, ) from haystack import Document, Pipeline from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.document_stores.types import DuplicatePolicy prompt_template [ ChatMessage.from_user( Given these documents, answer the question.\nDocuments: {% for doc in documents %} {{ doc.content }} {% endfor %} \nQuestion: {{question}} \nAnswer: , ), ] document_store ElasticsearchDocumentStore(hostshttp://localhost:9200/) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.), Document(contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.), ] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever ElasticsearchBM25Retriever(document_storedocument_store) 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)运行后可能得到类似输出Over 7,000 languages are spoken around the world today。五、ElasticsearchEmbeddingRetriever向量语义检索ElasticsearchEmbeddingRetriever通过向量相似度从ElasticsearchDocumentStore中检索文档适合语义匹配场景依据 Embedding Retriever 使用文档。使用时需要确保查询与文档的嵌入都可用在索引 Pipeline 中加 Document Embedder在查询 Pipeline 中加 Text Embedder。5.1 基本用法from haystack import Document # Requires: pip install sentence-transformers-haystack 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.2 初始化参数__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_storeElasticsearchDocumentStore必填绑定的 DocumentStore类型不符抛出ValueErrorfiltersdict[str, Any] \| NoneNone应用于检索结果的过滤器。过滤器会在近似 KNN 搜索阶段应用确保返回满足条件的top_k个文档top_kint10最多返回的文档数num_candidatesint \| NoneNone每个分片上近似最近邻候选数默认取top_k * 10。增大该值可提升搜索精度但会降低搜索速度详见 Elasticsearch 近似 KNN 调优文档filter_policystr \| FilterPolicyFilterPolicy.REPLACE过滤器应用策略取值同 4.4注意用于向量检索的embedding_similarity_function必须在初始化对应的ElasticsearchDocumentStore时指定见 3.2 节因为该参数只在索引创建时生效。5.3 run 与 run_asyncrun( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询的嵌入向量list[float]filters运行时过滤器在近似 KNN 搜索期间应用确保返回top_k个匹配文档应用方式由初始化时的filter_policy决定top_k本次调用最多返回的文档数。返回{documents: [...]}即与query_embedding最相似的文档列表。run_async提供异步版本。5.4 在查询 Pipeline 中使用from haystack_integrations.components.retrievers.elasticsearch import ( ElasticsearchEmbeddingRetriever, ) from haystack_integrations.document_stores.elasticsearch import ( ElasticsearchDocumentStore, ) from haystack.document_stores.types import DuplicatePolicy from haystack import Document, Pipeline from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) document_store ElasticsearchDocumentStore(hostshttp://localhost:9200/) model BAAI/bge-large-en-v1.5 documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.), Document(contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.), ] document_embedder SentenceTransformersDocumentEmbedder(modelmodel) documents_with_embeddings document_embedder.run(documents) document_store.write_documents(documents_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) query How many languages are there? result query_pipeline.run({text_embedder: {text: query}}) print(result[retriever][documents][0])示例输出形如含语义相似度得分与 1024 维嵌入向量Document(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.87717235, embedding: vector of size 1024)六、ElasticsearchSQLRetriever结构化 SQL 查询ElasticsearchSQLRetriever允许直接对ElasticsearchDocumentStore执行Elasticsearch SQL查询用于在运行时获取元数据、执行聚合如计数、平均值或其他结构化数据访问依据 SQL Retriever 使用文档。与另外两个检索器不同它不返回Document列表而是返回 Elasticsearch SQL API 的原始 JSON 响应。6.1 基本用法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] contains the raw Elasticsearch JSON response # result[result][columns] contains column metadata # result[result][rows] contains the data rows6.2 初始化参数__init__( *, document_store: ElasticsearchDocumentStore, raise_on_failure: bool True, fetch_size: int | None None, ) - None参数类型默认值说明document_storeElasticsearchDocumentStore必填要使用的 DocumentStore类型不符抛出ValueErrorraise_on_failureboolTrueAPI 调用失败时是否抛出异常为False时记录警告并返回空字典fetch_sizeint \| NoneNone每页抓取的结果条数不设置时使用 Elasticsearch 默认的 fetch size6.3 run 与 run_asyncrun( query: str, document_store: ElasticsearchDocumentStore | None None, fetch_size: int | None None, ) - dict[str, dict[str, Any]]query要执行的 Elasticsearch SQL 查询字符串document_store可选的 DocumentStore 实例用于覆盖初始化时绑定的实例fetch_size每页抓取条数可覆盖初始化值不提供时回退到初始化值或 ES 默认值。返回字典的result键保存 Elasticsearch SQL API 的原始 JSON 响应失败且raise_on_failureFalse时为空字典。run_async提供异步版本用法一致result await retriever.run_async( querySELECT content, category FROM my_index WHERE category \A\ )6.4 完整示例与聚合查询先向索引写入文档再执行 SQLfrom haystack import Document from haystack_integrations.components.retrievers.elasticsearch import ( ElasticsearchSQLRetriever, ) from haystack_integrations.document_stores.elasticsearch import ( ElasticsearchDocumentStore, ) from haystack.document_stores.types import DuplicatePolicy document_store ElasticsearchDocumentStore( hostshttp://localhost:9200/, indexmy_index ) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.), Document(contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.), ] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever ElasticsearchSQLRetriever(document_storedocument_store) output retriever.run(querySELECT content FROM my_index LIMIT 10) result output[result] print(result[columns]) # 列元数据例如 [{name: content, type: text}] for row in result[rows]: print(row)利用原始 SQL 响应执行文档型检索器不支持的聚合例如计数retriever ElasticsearchSQLRetriever(document_storedocument_store) output retriever.run(querySELECT COUNT(*) AS doc_count FROM my_index) result output[result] print(result[rows]) # 例如 [[3]]若担心查询语句错误导致整条链路中断可初始化ElasticsearchSQLRetriever(document_storedocument_store, raise_on_failureFalse)失败时记录警告并返回空字典。七、序列化to_dict 与 from_dict三个 Retriever 与ElasticsearchDocumentStore均实现了标准的 Haystack 序列化协议to_dict() - dict[str, Any]——将组件序列化为字典其中Secret等敏感信息会以安全的占位形式存储from_dict(data: dict[str, Any]) - ElasticsearchDocumentStore或对应的 Retriever 类型——从字典反序列化还原组件实例。这保证了组件可以无缝嵌入 Haystack 的 YAML/JSON Pipeline 描述与持久化流程。八、实战小结与选型建议场景推荐组件关键点精确匹配人名、ID、错误信息等关键词ElasticsearchBM25Retriever轻量可配fuzziness模糊匹配与scale_score得分归一化语义相似检索RAG、问答ElasticsearchEmbeddingRetriever需要 Embedder 组件配合num_candidates控制精度/速度取舍元数据、聚合、结构化字段查询ElasticsearchSQLRetriever返回原始 JSON可执行 SQL 聚合raise_on_failure控制失败行为索引侧直接生成嵌入ELSER 等ElasticsearchDocumentStore(ingest_pipeline...)注意output_field必须为embedding或sparse_vector_field且上游不要叠加 DocumentEmbedder所有组件都同时提供同步与异步 APIrun/run_async、write_documents/write_documents_async等可平滑对接异步 Pipeline序列化协议则让整套配置可以落入 Pipeline 描述文件进行版本管理。更深入的过滤器语法可参考 Haystack 元数据过滤文档与 Elasticsearch Query DSL 官方文档连接与安全配置可参考 Elasticsearch Python 客户端连接文档。【免费下载链接】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），仅供参考