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則分別覆蓋關(guān)鍵詞檢索、語義向量檢索與結(jié)構(gòu)化 SQL 查詢?nèi)悎鼍?。本文?version-2.18 的 Elasticsearch API 參考文檔 為骨架結(jié)合當前倉庫中的用戶指南與檢索器文檔完整講解從環(huán)境搭建、索引初始化到三類檢索器的參數(shù)語義、序列化與異步 API幫助你在 RAG、語義搜索與混合檢索管線中直接落地 Elasticsearch 后端。一、集成概覽一個后端、三條檢索路徑Elasticsearch 集成的核心價值在于同一個ElasticsearchDocumentStore可以同時承載稀疏檢索BM25 關(guān)鍵詞匹配與稠密檢索embedding 向量相似度方便在 PoC 階段直接對比 dense 與 sparse 兩種檢索方案的效果并平滑遷移到生產(chǎn)環(huán)境。文檔存儲支持 ANN 近似最近鄰搜索。從 API 參考文檔 的模塊結(jié)構(gòu)看集成主要包含四部分模塊組件檢索方式...retrievers.elasticsearch.bm25_retrieverElasticsearchBM25RetrieverBM25 關(guān)鍵詞算法...retrievers.elasticsearch.embedding_retrieverElasticsearchEmbeddingRetriever向量相似度ANN...retrievers.elasticsearch.sql_retrieverElasticsearchSQLRetrieverElasticsearch SQL 原生查詢...document_stores.elasticsearch.document_storeElasticsearchDocumentStore文檔存儲與索引管理另外文檔還包含haystack_integrations.document_stores.elasticsearch.filters模塊元數(shù)據(jù)過濾相關(guān)支持 Haystack 元數(shù)據(jù)過濾語法在 Elasticsearch 端的落地。核心庫中對 Elasticsearch 的引用也貫穿于 自動合并檢索器 與 句子窗口檢索器 等核心組件這些組件通過與兼容的文檔存儲配合可疊加父子文檔、上下文窗口等高級檢索策略。二、環(huán)境準備安裝 Elasticsearch 與集成包Haystack 支持 Elasticsearch 8。官方推薦使用 Docker 快速拉起單節(jié)點實例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僅用于本地演示生產(chǎn)環(huán)境務必啟用安全認證詳見官方連接文檔確保只有授權(quán)用戶能訪問數(shù)據(jù)。三、ElasticsearchDocumentStore索引初始化與文檔管理3.1 初始化與參數(shù)語義ElasticsearchDocumentStore同時支持 Elastic Cloud 與自建集群。兩種典型初始化方式# 方式一Elastic Cloud通過環(huán)境變量提供 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)完整構(gòu)造函數(shù)簽名__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各參數(shù)的核心語義如下參數(shù)默認值說明hostsNoneElasticsearch 客戶端連接的節(jié)點地址列表custom_mappingNone自定義索引映射不傳則使用默認映射indexdefault使用的 Elasticsearch 索引名索引不存在時會自動創(chuàng)建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 相似度函數(shù)可選cosine/dot_product/l2_norm/max_inner_product僅在索引不存在并新建時生效選擇時需結(jié)合嵌入模型說明sparse_vector_fieldNone若設置為該名稱的 Elasticsearch 字段類型sparse_vector存儲稀疏 embedding未設置時 Document 上的sparse_embedding數(shù)據(jù)寫入時會被靜默丟棄ingest_pipelineNoneElasticsearch ingest pipeline 的 id用于在索引時通過 inference processor如 ELSER 或稠密模型生成 embedding而無需在 Haystack 側(cè)運行 embedder 組件首尾空白會被去除**kwargs—透傳給 Elasticsearch 客戶端Elasticsearch/AsyncElasticsearch的其余參數(shù)認證方面默認從環(huán)境變量加載Secret也可用Secret.from_token()從 token 加載。存儲對象同時暴露client同步Elasticsearch客戶端按需惰性初始化與async_client異步AsyncElasticsearch客戶端兩個屬性。3.2 使用 ingest pipeline 生成 embedding 的約束當通過ingest_pipeline使用 inference processor 時有三個關(guān)鍵要求input_output必須正確指向輸出字段output_field必須等于embedding稠密檢索或sparse_vector_field的值ELSER / 稀疏檢索。Elasticsearch 默認寫入的ml.inference.tag目標字段不會被 Haystack 檢索器找到。不要同時運行 Haystack 的DocumentEmbedder若文檔到達時已帶有預計算的embeddingingest pipeline 會用自身模型的向量覆蓋它導致存儲向量與查詢向量靜默不一致檢索時出現(xiàn)錯配。自定義映射需包含輸出字段如果提供了custom_mapping必須包含類型正確的輸出字段dense_vector或sparse_vector。另外關(guān)于稀疏 embedding 有一個實現(xiàn)細節(jié)Elasticsearch 不會把 inference pipeline 生成的sparse_vector數(shù)據(jù)存進_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等待下一個刷新周期默認保證寫入即讀一致性。返回實際寫入的文檔數(shù)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按元數(shù)據(jù)過濾條件刪除返回刪除數(shù)量。按條件更新update_by_filter(filters, meta, refreshFalse) - int可批量更新命中過濾條件的文檔元數(shù)據(jù)。3.4 檢索、統(tǒng)計與元數(shù)據(jù)管理方法DocumentStore 的主查詢方法是filter_documents(filters)它返回所有匹配過濾條件的 Documentcount_documents()返回文檔總數(shù)count_documents_by_filter(filters)統(tǒng)計命中過濾條件的數(shù)量。面向元數(shù)據(jù)管理還提供了一組實用方法count_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]統(tǒng)計各指定元數(shù)據(jù)字段的唯一值個數(shù)字段名可帶或不帶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)返回某元數(shù)據(jù)字段的最小值與最大值{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基于近似基數(shù)聚合計算超高基數(shù)字段下可能不精確。search_term為大小寫不敏感的模糊子串匹配匹配字段值本身而非文檔內(nèi)容該匹配通過服務端腳本實現(xiàn)在大語料上開銷較大。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輕量關(guān)鍵詞檢索4.1 原理與適用場景ElasticsearchBM25Retriever基于 BM25 算法從ElasticsearchDocumentStore檢索文檔通過計算查詢與文檔之間的加權(quán)詞重疊度來判定相似性只兼容ElasticsearchDocumentStore。由于其本質(zhì)是詞面匹配特別適合人名、產(chǎn)品名、ID、明確的錯誤信息等精確匹配場景BM25 輕量簡單在域外數(shù)據(jù)上往往不遜于復雜的 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 初始化參數(shù)__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參數(shù)默認值說明document_store—ElasticsearchDocumentStore實例必填非該類型實例時拋ValueErrorfiltersNone應用于檢索結(jié)果的過濾條件語法詳見ElasticsearchDocumentStore.filter_documentsfuzzinessAUTO傳給 Elasticsearch 的模糊匹配參數(shù)inexact fuzzy matching用于容忍拼寫錯誤top_k10最多返回的 Document 數(shù)量scale_scoreFalse為True時將 Document 的分數(shù)縮放到 0–1 區(qū)間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為異步版本簽名與返回結(jié)構(gòu)一致。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返回最相關(guān)的文檔。使用時必須保證查詢與文檔兩側(cè)的 embedding 都可用——通常由索引管線中的 Document Embedder 和查詢管線中的 Text Embedder 提供。embedding 相似度函數(shù)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 初始化參數(shù)__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參數(shù)默認值說明document_store—ElasticsearchDocumentStore實例必填非該類型時拋ValueErrorfiltersNone檢索過濾條件在近似 KNN 搜索期間應用以確保返回恰好top_k個匹配文檔top_k10最多返回的 Document 數(shù)量num_candidatesNone每個分片上的近似最近鄰候選數(shù)默認top_k * 10增大可提升檢索準確率但會降低檢索速度屬于速度—精度權(quán)衡的高級調(diào)參項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提供異步等價實現(xiàn)。5.5 管線化用法在查詢管線中將 Text Embedder 的輸出連接到 Retriever 的query_embedding輸入索引側(cè)先使用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 定位結(jié)構(gòu)化數(shù)據(jù)訪問與前兩類檢索器不同ElasticsearchSQLRetriever不把查詢匹配到文檔而是把原生 Elasticsearch SQL 語句直接下發(fā)執(zhí)行并返回 SQL API 的原始 JSON 響應。它適合在運行時獲取元數(shù)據(jù)、做聚合統(tǒng)計計數(shù)、均值等以及其他結(jié)構(gòu)化數(shù)據(jù)訪問。6.2 初始化與調(diào)用__init__( *, document_store: ElasticsearchDocumentStore, raise_on_failure: bool True, fetch_size: int | None None ) - Noneraise_on_failure為True默認時SQL API 調(diào)用失敗拋出異常為False時記錄 warning 并返回空字典。fetch_size每頁抓取的結(jié)果條數(shù)不傳則使用 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] - 列元數(shù)據(jù) # result[result][rows] - 數(shù)據(jù)行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 聚合查詢示例由于返回的是原始響應可以執(zhí)行普通文檔檢索器不支持的聚合操作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 實現(xiàn)的單組件混合檢索器內(nèi)置 Text Embedder并行執(zhí)行 BM25 與向量檢索再通過DocumentJoiner默認 Reciprocal Rank Fusion 互惠排名融合合并重排結(jié)果?？赏ㄟ^top_k_bm25、fuzziness、filters_bm25、scale_score、filter_policy_bm25、top_k_embedding、filters_embedding、num_candidates、filter_policy_embedding等參數(shù)分別調(diào)優(yōu)兩條檢索支路并直接暴露join_mode、weights、top_k、sort_by_score等合并參數(shù)。所有組件的序列化能力保持一致to_dict() - dict[str, Any]把組件序列化為字典ElasticsearchDocumentStore等存儲對象同樣支持。from_dict(data) - 組件實例從字典反序列化恢復組件便于通過 YAML/JSON 定義與分享管線。close()/close_async()釋放底層文檔存儲的同步/異步資源用于資源生命周期管理。八、小結(jié)圍繞 Elasticsearch 后端Haystack 提供了從索引管理到多路檢索的完整閉環(huán)ElasticsearchDocumentStore負責索引自動創(chuàng)建、映射定制、ingest pipeline 集成、批量寫入/刪除/更新以及豐富的元數(shù)據(jù)統(tǒng)計能力ElasticsearchBM25Retriever提供輕量關(guān)鍵詞檢索支持fuzziness模糊匹配與分數(shù)縮放ElasticsearchEmbeddingRetriever提供基于 ANN 的語義檢索支持num_candidates精度調(diào)優(yōu)ElasticsearchSQLRetriever則打通了結(jié)構(gòu)化 SQL 查詢通道。三者共享同步/異步雙 API 與統(tǒng)一的序列化機制可直接嵌入 RAG、語義搜索、抽取式 QA 與混合檢索管線。若要進一步了解安裝細節(jié)、檢索器在管線中的典型位置或 Hybrid 用法可繼續(xù)閱讀倉庫內(nèi)的 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創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考