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 應(yīng)用管線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)境務(wù)必啟用安全認證詳見官方連接文檔確保只有授權(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若設(shè)置為該名稱的 Elasticsearch 字段類型sparse_vector存儲稀疏 embedding未設(shè)置時 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若文檔到達時已帶有預(yù)計算的embeddingingest pipeline 會用自身模型的向量覆蓋它導(dǎo)致存儲向量與查詢向量靜默不一致檢索時出現(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時刪除索引并按原映射/設(shè)置重建否則走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_偏移需要通過重復(fù)抓取并丟棄前面桶來模擬成本隨from_線性增長而非隨size增長total_count基于近似基數(shù)聚合計算超高基數(shù)字段下可能不精確。search_term為大小寫不敏感的模糊子串匹配匹配字段值本身而非文檔內(nèi)容該匹配通過服務(wù)端腳本實現(xiàn)在大語料上開銷較大。3.5 同步與異步 API 全覆蓋上述所有方法均有對應(yīng)的*_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ù)上往往不遜于復(fù)雜的 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應(yīng)用于檢索結(jié)果的過濾條件語法詳見ElasticsearchDocumentStore.filter_documentsfuzzinessAUTO傳給 Elasticsearch 的模糊匹配參數(shù)inexact fuzzy matching用于容忍拼寫錯誤top_k10最多返回的 Document 數(shù)量scale_scoreFalse為True時將 Document 的分數(shù)縮放到 0–1 區(qū)間filter_policyFilterPolicy.REPLACE決定初始化過濾器與運行時過濾器如何合并應(yīng)用的策略枚舉來自 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避免重復(fù)運行報錯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 搜索期間應(yīng)用以確保返回恰好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 搜索期間應(yīng)用以保證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 響應(yīng)。它適合在運行時獲取元數(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 服務(wù)端默認值。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 響應(yīng) # 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 響應(yīng)dict或出錯時的空 dict。6.3 聚合查詢示例由于返回的是原始響應(yīng)可以執(zhí)行普通文檔檢索器不支持的聚合操作output retriever.run(querySELECT COUNT(*) AS doc_count FROM my_index) print(output[result][rows]) # 例如 [[3]]對可能出錯或格式非法的查詢可設(shè)置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) - 組件實例從字典反序列化恢復(fù)組件便于通過 YAML/JSON 定義與分享管線。close()/close_async()釋放底層文檔存儲的同步/異步資源用于資源生命周期管理。八、小結(jié)圍繞 Elasticsearch 后端Haystack 提供了從索引管理到多路檢索的完整閉環(huán)ElasticsearchDocumentStore負責(zé)索引自動創(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 指南 及對應(yīng)檢索器組件文檔。【免費下載鏈接】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），僅供參考