Haystack 与 Azure AI Search 集成指南：AzureAISearchDocumentStore 与三大检索器实战解析-编程实验室

Haystack 与 Azure AI Search 集成指南：AzureAISearchDocumentStore 与三大检索器实战解析

【免费下载链接】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

Azure AI Search 是微软推出的企业级云端搜索与检索服务，专为在 Azure 上构建 RAG 应用而设计，并原生集成了 LLM 能力。Haystack 通过azure-ai-search-haystack集成包，将 Azure AI Search 封装为标准的 Document Store 与 Retriever 组件，让开发者可以在 Haystack 的 Pipeline 中直接使用向量检索、BM25 关键词检索、混合检索与语义重排能力。本文基于 Haystack 2.19 版本的 API 参考文档（docs-website/reference_versioned_docs/version-2.19/integrations-api/azure_ai_search.md）及配套使用指南，系统讲解AzureAISearchDocumentStore的完整 API、AzureAISearchEmbeddingRetriever的用法，以及如何在 RAG 管道中落地，读完即可动手搭建一套基于 Azure AI Search 的检索增强生成应用。

集成概览与适用场景

AzureAISearchDocumentStore是一个以 Azure AI Search 索引为后端的 Document Store，支持语义重排（semantic reranking）以及元数据/内容过滤。它适用于多种生产场景，包括：

知识库洞察：目录检索、文档搜索；
信息发现：数据探索与筛选；
RAG（检索增强生成）：为 LLM 提供高质量上下文；
自动化流程：与各类业务系统集成。

配套的检索器组件共有三个，均基于 Azure AI Search API 实现，可根据管道需求选用：

检索器	输入	特点
`AzureAISearchEmbeddingRetriever`	`query_embedding`（向量）	基于向量相似度检索，需要先用 Embedder 对查询编码
`AzureAISearchBM25Retriever`	`query`（文本）	基于 BM25 关键词打分检索
`AzureAISearchHybridRetriever`	`query`+`query_embedding`	同时执行向量检索与 BM25 检索，用 RRF 融合排序

对应完整使用指南可参考 AzureAISearchDocumentStore、AzureAISearchEmbeddingRetriever、AzureAISearchBM25Retriever 与 AzureAISearchHybridRetriever。

环境准备与安装

使用该集成的前提是拥有一个有效的 Azure 订阅，并已部署 Azure AI Search 服务。随后安装集成包：

pip install azure-ai-search-haystack

认证需要两类信息，推荐通过环境变量注入：

AZURE_AI_SEARCH_ENDPOINT：搜索服务的 URL 端点（必填，strict=True）；
AZURE_AI_SEARCH_API_KEY：API 密钥（若未提供，DefaultAzureCredential会尝试通过浏览器完成登录认证）。

需要说明的是：Azure AI Search 索引的字段在创建后无法通过 API 修改。因此，除默认字段外的任何附加字段，都必须在 Document Store 初始化时通过metadata_fields声明；如需调整字段定义，只能借助 Azure 门户在不删除索引的前提下修改。

AzureAISearchDocumentStore 完整 API 解析

AzureAISearchDocumentStore的构造签名如下：

__init__( *, api_key: Secret = Secret.from_env_var("AZURE_AI_SEARCH_API_KEY", strict=False), azure_endpoint: Secret = Secret.from_env_var("AZURE_AI_SEARCH_ENDPOINT", strict=True), index_name: str = "default", embedding_dimension: int = 768, metadata_fields: dict[str, SearchField | type] | None = None, vector_search_configuration: VectorSearch | None = None, include_search_metadata: bool = False, azure_token_credential: TokenCredential | None = None, **index_creation_kwargs: Any ) -> None

核心参数说明

azure_endpoint（Secret）：Azure AI Search 服务的 URL 端点。
api_key（Secret）：用于认证的 API 密钥，默认从环境变量AZURE_AI_SEARCH_API_KEY读取（非严格模式，允许缺失）。
index_name（str，默认"default"）：索引名称。初始化时若索引不存在会自动创建。
embedding_dimension（int，默认768）：嵌入向量的维度，必须与写入文档时 Embedder 输出的向量维度一致。
metadata_fields（dict[str, SearchField | type] | None）：元数据字段映射，每个字段可以有两种定义方式：
- 传SearchField对象，精细化配置字段类型、是否可搜索（searchable）、是否可过滤（filterable）等；
- 传 Python 类型（str、bool、int、float或datetime），自动创建一个可过滤字段。
这些字段会在创建索引时自动加入索引结构。示例：
```
metadata_fields={ "Title": SearchField( name="Title", type="Edm.String", searchable=True, filterable=True ), "Pages": int }
```
vector_search_configuration（VectorSearch | None）：向量搜索相关配置。默认配置使用 HNSW 算法配合余弦相似度（cosine similarity）处理向量检索。
include_search_metadata（bool，默认False）：是否将 Azure AI Search 返回的元数据字段写入文档的meta。置为True时，返回文档的meta会包含@search.score、@search.reranker_score、@search.highlights、@search.captions等字段。
azure_token_credential（TokenCredential | None）：AzureTokenCredential实例，用于基于令牌的认证；一旦提供，其优先级高于api_key。
index_creation_kwargs（Any）：透传给SearchIndex类的可选关键字参数，常见包括：
- semantic_search：定义索引的语义配置，用于在索引上启用语义搜索能力（启用语义重排的入口）；
- similarity：匹配查询时用于打分排序的相似度算法。该算法只能在索引创建时定义，已有索引无法修改。

索引与客户端访问

client属性返回 AzureSearchClient，并且在首次访问时会自动创建索引（若不存在）：

client: SearchClient

生命周期与序列化

to_dict() -> dict[str, Any]：将组件序列化为字典；
from_dict(data: dict[str, Any]) -> AzureAISearchDocumentStore：从字典反序列化还原组件；
close() -> None：释放关联的同步资源。

文档写入与删除

write_documents(documents: list[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int：将文档写入索引，返回成功写入的文档数。当文档类型不是Document时抛出ValueError；文档 ID 不是字符串时抛出TypeError。注意，AzureAISearchDocumentStore实际默认的重复策略为DuplicatePolicy.OVERWRITE。
delete_documents(document_ids: list[str]) -> None：按 ID 删除索引中的文档。
delete_all_documents(recreate_index: bool = False) -> None：清空所有文档。recreate_index=True时先删除索引再按原 schema 重建；False时保留索引结构仅清空文档。
delete_by_filter(filters: dict[str, Any]) -> int：删除所有匹配过滤条件的文档，返回删除数量。由于 Azure AI Search 不支持服务端按查询删除，该方法会先搜索匹配文档，再通过批量操作删除——这是实现层面需要注意的性能特征。

文档更新

update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) -> int：更新所有匹配过滤条件文档的字段，返回更新数量。同理，Azure AI Search 不支持服务端按查询更新，因此该方法先搜索匹配文档，再使用合并（merge）操作更新。注意meta中的字段必须已存在于索引 schema 中，否则更新无法落库。

查询与计数

count_documents() -> int：返回索引中文档总数。
count_documents_by_filter(filters: dict[str, Any]) -> int：返回匹配过滤条件的文档数。
count_unique_metadata_by_filter(filters: dict[str, Any], metadata_fields: list[str]) -> dict[str, int]：对匹配过滤条件的文档，统计每个指定元数据字段的唯一值个数。
get_metadata_fields_info() -> dict[str, dict[str, str]]：返回索引中元数据字段的类型信息。
get_metadata_field_min_max(metadata_field: str) -> dict[str, Any]：返回某元数据字段的最小值与最大值，结果字典包含"min"与"max"两个键。
get_metadata_field_unique_values(metadata_field: str, search_term: str | None = None, from_: int = 0, size: int = 10, filters: dict[str, Any] | None = None) -> tuple[list[Any], int]：带搜索与分页地获取某元数据字段的唯一值，返回(唯一值列表, 匹配总数)；未在索引 schema 中定义的字段返回([], 0)。
query_sql(query: str) -> Any：执行 SQL 查询。Azure AI Search 不支持 SQL 查询，调用该方法是无效的。
get_documents_by_id(document_ids: list[str]) -> list[Document]：按 ID 批量获取文档。
search_documents(search_text: str = '*', top_k: int = 10) -> list[Document]：返回匹配search_text的所有文档；search_text为空时返回全部文档。
filter_documents(filters: dict[str, Any] | None = None) -> list[Document]：按元数据过滤条件返回文档。过滤条件遵循 Haystack 的元数据过滤语法。

以上过滤类方法的filters均遵循 Haystack 元数据过滤（metadata filtering）规范，属于该集成的通用接口能力。

初始化与基础写入示例

推荐在执行示例前先通过环境变量提供认证信息：

from haystack_integrations.document_stores.azure_ai_search import ( AzureAISearchDocumentStore, ) from haystack import Document document_store = AzureAISearchDocumentStore(index_name="haystack-docs") document_store.write_documents( [ Document(content="This is the first document."), Document(content="This is the second document."), ], ) print(document_store.count_documents())

:::note 索引延迟提示由于 Azure 搜索索引存在传播延迟，示例执行后立即count_documents()的结果可能是 0。在从索引检索文档时应留意这一延迟，必要时等待片刻再查询。 :::

启用语义重排的方式：在初始化时通过index_creation_kwargs传入SemanticSearch配置，之后即可在某个 Retriever 中调用语义查询。这一步是使用语义检索功能的先决条件。

AzureAISearchEmbeddingRetriever：向量检索器

AzureAISearchEmbeddingRetriever使用向量相似度指标从AzureAISearchDocumentStore检索文档，必须连接到该 Document Store 才能运行。

初始化参数

__init__( *, document_store: AzureAISearchDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, filter_policy: str | FilterPolicy = FilterPolicy.REPLACE, **kwargs: Any ) -> None

document_store（AzureAISearchDocumentStore）：与该检索器配合使用的 Document Store 实例。
filters（dict[str, Any] | None）：拉取文档时应用的过滤条件。
top_k（int，默认10）：最多返回的文档数量。
filter_policy（str | FilterPolicy，默认FilterPolicy.REPLACE）：过滤策略，决定运行时传入的过滤器如何与初始化时的过滤器合并。
kwargs（Any）：透传给 Azure AI Search 端点的附加参数，常用包括：
- query_type：查询类型字符串，可选'simple'、'full'、'semantic'；
- semantic_configuration_name：处理语义查询时使用的语义配置名称（需索引已配置semantic_search）。

run 方法

run( query_embedding: list[float], filters: dict[str, Any] | None = None, top_k: int | None = None, ) -> dict[str, list[Document]]

query_embedding（list[float]）：查询文本的向量表示，必须由上游 Embedder 组件（如 Text Embedder）预先计算；
filters：运行时过滤条件，其生效方式取决于初始化时选择的filter_policy；
top_k：运行时覆盖最大返回文档数。

返回字典包含键documents，值为从 Document Store 检索到的文档列表。

序列化与资源释放

to_dict() -> dict[str, Any]：序列化为字典；
from_dict(data: dict[str, Any]) -> AzureAISearchEmbeddingRetriever：反序列化还原；
close() -> None：释放底层 Document Store 的同步资源。

独立使用

from haystack_integrations.document_stores.azure_ai_search import ( AzureAISearchDocumentStore, ) from haystack_integrations.components.retrievers.azure_ai_search import ( AzureAISearchEmbeddingRetriever, ) document_store = AzureAISearchDocumentStore() retriever = AzureAISearchEmbeddingRetriever(document_store=document_store) ## 示例查询 retriever.run(query_embedding=[0.1] * 384)

语义重排的适用范围

需要特别注意：Azure AI Search 的语义重排能力不适用于纯向量检索。若希望在检索流程中加入语义重排，应改用AzureAISearchBM25Retriever或AzureAISearchHybridRetriever。

AzureAISearchBM25Retriever：关键词检索器

AzureAISearchBM25Retriever是基于关键词的检索器，使用 BM25 算法计算查询与文档之间的加权词重叠度来确定相似性。它接受文本查询，也支持带布尔运算符的组合词条，例如"pool"、"pool spa"、"pool spa +airport"都是合法的查询形式。

运行接口：run(query: str, filters: dict | None = None, top_k: int | None = None)，返回documents列表。top_k与filters用于收窄检索范围。

独立使用

from haystack import Document from haystack_integrations.components.retrievers.azure_ai_search import ( AzureAISearchBM25Retriever, ) from haystack_integrations.document_stores.azure_ai_search import ( AzureAISearchDocumentStore, ) document_store = AzureAISearchDocumentStore(index_name="haystack_docs") documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.", ), Document( content="In 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(documents=documents) retriever = AzureAISearchBM25Retriever(document_store=document_store) retriever.run(query="How many languages are spoken around the world today?")

启用语义排名

如果搜索索引配置了语义配置（semantic configuration），可以通过在初始化时传入相应 kwargs 为 BM25 检索结果启用语义排名。若想同时结合 BM25 与向量检索，则应使用AzureAISearchHybridRetriever。

AzureAISearchHybridRetriever：混合检索器

AzureAISearchHybridRetriever在同一请求中并行执行向量检索与 BM25 文本检索，再使用倒数排名融合（Reciprocal Rank Fusion, RRF）合并并重排结果，从而获得更相关的统一结果集。

运行接口需要同时提供query: str与query_embedding: list[float]，可选top_k与filters；初始化时同样可传入附加关键字参数做进一步定制。

独立使用

from haystack import Document from haystack_integrations.components.retrievers.azure_ai_search import ( AzureAISearchHybridRetriever, ) from haystack_integrations.document_stores.azure_ai_search import ( AzureAISearchDocumentStore, ) document_store = AzureAISearchDocumentStore(index_name="haystack_docs") documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.", ), Document( content="In 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(documents=documents) retriever = AzureAISearchHybridRetriever(document_store=document_store) ## 用假向量简化示例 retriever.run( query="How many languages are spoken around the world today?", query_embedding=[0.1] * 384, )

选择建议

纯关键词检索 →AzureAISearchBM25Retriever；
纯向量检索 →AzureAISearchEmbeddingRetriever；
兼顾语义与关键词、追求更稳的召回 →AzureAISearchHybridRetriever。

实战：在 Haystack Pipeline 中落地

场景一：Embedding 检索的索引 + 查询双管道

索引管道负责将文档送入 Document Embedder 编码后写入 Document Store；查询管道先用 Text Embedder 编码查询，再交给AzureAISearchEmbeddingRetriever取回结果。

from haystack import Document, Pipeline from haystack.components.embedders import ( SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, ) from haystack.components.writers import DocumentWriter from haystack_integrations.components.retrievers.azure_ai_search import ( AzureAISearchEmbeddingRetriever, ) from haystack_integrations.document_stores.azure_ai_search import ( AzureAISearchDocumentStore, ) document_store = AzureAISearchDocumentStore(index_name="retrieval-example") model = "sentence-transformers/all-mpnet-base-v2" documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="""Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.""", ), Document( content="""In certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.""", ), ] document_embedder = SentenceTransformersDocumentEmbedder(model=model) document_embedder.warm_up() ## 索引管道 indexing_pipeline = Pipeline() indexing_pipeline.add_component(instance=document_embedder, name="doc_embedder") indexing_pipeline.add_component( instance=DocumentWriter(document_store=document_store), name="doc_writer", ) indexing_pipeline.connect("doc_embedder", "doc_writer") indexing_pipeline.run({"doc_embedder": {"documents": documents}}) ## 查询管道 query_pipeline = Pipeline() query_pipeline.add_component( "text_embedder", SentenceTransformersTextEmbedder(model=model), ) query_pipeline.add_component( "retriever", AzureAISearchEmbeddingRetriever(document_store=document_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])

要点：text_embedder.embedding必须显式连接到retriever.query_embedding；使用AzureAISearchHybridRetriever时，查询管道还需要在run时同时传入{"retriever": {"query": query}}。

场景二：BM25 + LLM 的完整 RAG 管道

AzureAISearchBM25Retriever可以直接串入标准 RAG 链路：Retriever → PromptBuilder → OpenAIGenerator → AnswerBuilder。运行前将OPENAI_API_KEY配置为环境变量。

from haystack_integrations.components.retrievers.azure_ai_search import ( AzureAISearchBM25Retriever, ) from haystack_integrations.document_stores.azure_ai_search import ( AzureAISearchDocumentStore, ) from haystack import Document from haystack import Pipeline from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders.prompt_builder import PromptBuilder from haystack.components.generators import OpenAIGenerator from haystack.document_stores.types import DuplicatePolicy import os api_key = os.environ["OPENAI_API_KEY"] ## 创建 RAG 查询管道 prompt_template = """ Given these documents, answer the question.\nDocuments: {% for doc in documents %} {{ doc.content }} {% endfor %} \nQuestion: {{question}} \nAnswer: """ document_store = AzureAISearchDocumentStore(index_name="haystack-docs") ## 添加文档 documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.", ), Document( content="In certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.", ), ] ## policy 参数可选，AzureAISearchDocumentStore 默认策略为 DuplicatePolicy.OVERWRITE document_store.write_documents(documents=documents, policy=DuplicatePolicy.OVERWRITE) retriever = AzureAISearchBM25Retriever(document_store=document_store) rag_pipeline = Pipeline() rag_pipeline.add_component(name="retriever", instance=retriever) rag_pipeline.add_component( instance=PromptBuilder(template=prompt_template), name="prompt_builder", ) rag_pipeline.add_component(instance=OpenAIGenerator(), name="llm") rag_pipeline.add_component(instance=AnswerBuilder(), name="answer_builder") rag_pipeline.connect("retriever", "prompt_builder.documents") rag_pipeline.connect("prompt_builder", "llm") rag_pipeline.connect("llm.replies", "answer_builder.replies") rag_pipeline.connect("llm.meta", "answer_builder.meta") rag_pipeline.connect("retriever", "answer_builder.documents") question = "Tell me something about languages?" result = rag_pipeline.run( { "retriever": {"query": question}, "prompt_builder": {"question": question}, "answer_builder": {"query": question}, }, ) print(result["answer_builder"]["answers"][0])

场景三：Hybrid 检索的索引 + 查询双管道

混合检索的索引管道与 Embedding 场景完全一致（同样需要 Document Embedder + DocumentWriter）；查询管道在连接text_embedder.embedding → retriever.query_embedding的同时，run阶段还需为retriever提供文本query：

result = query_pipeline.run( {"text_embedder": {"text": query}, "retriever": {"query": query}}, )

这是 Hybrid Retriever 与纯 Embedding Retriever 在用法上最核心的区别。

常见陷阱与最佳实践

索引 schema 不可变：字段在创建后无法通过 API 修改，务必在初始化时一次性通过metadata_fields声明全部附加字段；similarity算法同样只能在创建时指定。
认证优先级：azure_token_credential优先于api_key；未提供api_key时会回退到DefaultAzureCredential。
索引传播延迟：写入后立即查询可能拿不到数据，生产代码中应容忍或处理这一延迟。
按查询更新/删除的开销：delete_by_filter与update_by_filter因 Azure 侧不支持服务端按查询操作，内部是先检索再批量操作，数据量大时耗时与消耗会明显上升。
语义重排的适用边界：纯向量检索不支持语义排名，需要语义重排时应走 BM25 或 Hybrid 检索器，并确保索引已配置semantic_search。
过滤语法统一：所有filters参数均遵循 Haystack 元数据过滤语法；初始化时的filter_policy决定运行时过滤器如何与之合并。

Haystack 与 Azure AI Search 集成指南：AzureAISearchDocumentStore 与三大检索器实战解析