选项
首页首页 Skill 数据库管理 azure-search-documents-py

azure-search-documents-py

microsoft/skills microsoft/skills

使用 Python SDK 搜索 Azure AI Search 索引,支持带 AI 增强功能的全文搜索、向量搜索、混合搜索和语义搜索。

...展开全部
2
更新时间 2026-09-14

Azure AI Search Python SDK

支持全文、向量和混合搜索,并具备 AI 增强功能。

安装

pip install azure-search-documents

环境变量

AZURE_SEARCH_ENDPOINT=https://.search.windows.net  # 所有身份验证方法均需此项
AZURE_SEARCH_INDEX_NAME= # 所有身份验证方法均需此项
AZURE_TOKEN_CREDENTIALS=prod # 仅当在生产环境中使用 DefaultAzureCredential 时才需要
AZURE_SEARCH_API_KEY= # 仅适用于下文所述的旧版 API 密钥认证路径

身份验证与生命周期

🔑 以下所有代码示例均遵循两条规则:

  1. 优先使用DefaultAzureCredential它在本地(Azure CLI / VS Code / 开发者 CLI)和 Azure 环境(托管身份、工作负载身份)中均可使用,且无需修改代码。请避免使用连接字符串、账户/API 密钥——它们会绕过 Entra 的审计和密钥轮换机制。
    • 本地开发:DefaultAzureCredential可直接使用。
    • 生产环境:将AZURE_TOKEN_CREDENTIALS设置为prod(或AZURE_TOKEN_CREDENTIALS= ),以将凭据链限制为生产环境安全的凭据。
  2. 将每个客户端封装在上下文管理器中,以确保 HTTP 传输、套接字和令牌缓存能以可预测的方式释放:
    • 同步模式:使用 `(...) as client:`
    • 异步:async with(...) as client: 以及 async with DefaultAzureCredential() as credential:(来自azure.identity.aio

代码片段可能会简化此配置,但生产环境中的代码应始终遵循这两条规则。

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.search.documents import SearchClient

# 本地开发:DefaultAzureCredential。生产环境:将 AZURE_TOKEN_CREDENTIALS 设置为 prod 或 AZURE_TOKEN_CREDENTIALS 设置为
credential = DefaultAzureCredential(require_envvar=True)
# 或者在生产环境中直接使用特定凭据:
# 参见 https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

with SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
    credential=credential,
) as client:
    results = list(client.search(search_text="*", top=5))

旧版:API 密钥(现有基于密钥的部署)

新代码应使用上文提到的DefaultAzureCredential。 仅当您拥有尚未迁移到 Entra ID 的现有基于密钥的部署时,才应使用AzureKeyCredential—— 例如,仍在完成 Entra 部署的受监管环境。同一AzureKeyCredential也可用于SearchIndexClientSearchIndexerClient进行管理操作。

import os
from azure.core.credentials import AzureKeyCredential
from azure.search.documents import SearchClient

with SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
    credential=AzureKeyCredential(os.environ["AZURE_SEARCH_API_KEY"]),
) as client:
    results = list(client.search(search_text="*", top=5))

客户端类型

客户端 用途
SearchClient 搜索和文档操作
SearchIndexClient 索引管理、同义词映射
SearchIndexerClient 索引器、数据源、技能集

使用向量字段创建索引

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    SearchFieldDataType,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    SearchableField,
    SimpleField
)

fields = [
    SimpleField(name="id", type=SearchFieldDataType.String, key=True),
    SearchableField(name="title", type=SearchFieldDataType.String),
    SearchableField(name="content", type=SearchFieldDataType.String),
    SearchField(
        name="content_vector",
        type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
        searchable=True,
        vector_search_dimensions=1536,
        vector_search_profile_name="my-vector-profile"
    )
]

vector_search = VectorSearch(
    algorithms=[
        HnswAlgorithmConfiguration(name="my-hnsw")
    ],
    profiles=[
        VectorSearchProfile(
            name="my-vector-profile",
            algorithm_configuration_name="my-hnsw"
        )
    ]
)

index = SearchIndex(
    name="my-index",
    fields=fields,
    vector_search=vector_search
)

with SearchIndexClient(endpoint, DefaultAzureCredential()) as index_client:
    index_client.create_or_update_index(index)

上传文档

from azure.search.documents import SearchClient

documents = [
    {
        "id": "1",
        "title": "Azure AI Search",
        "content": "全文和向量搜索服务",
        "content_vector": [0.1, 0.2, ...]  # 1536 个维度
    }
]

with SearchClient(endpoint, "my-index", DefaultAzureCredential()) as client:
    result = client.upload_documents(documents)
    print(f"已上传 {len(result)} 份文档")

关键词搜索

results = client.search(
    search_text="azure search",
    select=["id", "title", "content"],
    top=10
)

for result in results:
    print(f"{result['title']}: {result['@search.score']}")

向量搜索

from azure.search.documents.models import VectorizedQuery

# 您的查询嵌入向量(1536维)
query_vector = get_embedding("semantic search capabilities")

vector_query = VectorizedQuery(
    vector=query_vector,
    k_nearest_neighbors=10,
    fields="content_vector"
)

results = client.search(
    vector_queries=[vector_query],
    select=["id", "title", "content"]
)

for result in results:
    print(f"{result['title']}: {result['@search.score']}")

混合搜索(向量 + 关键词)

from azure.search.documents.models import VectorizedQuery

vector_query = VectorizedQuery(
    vector=query_vector,
    k_nearest_neighbors=10,
    fields="content_vector"
)

results = client.search(
    search_text="azure search",
    vector_queries=[vector_query],
    select=["id", "title", "content"],
    top=10
)

语义排序

from azure.search.documents.models import QueryType

results = client.search(
    search_text="什么是 Azure Search",
    query_type=QueryType.SEMANTIC,
    semantic_configuration_name="my-semantic-config",
    select=["id", "title", "content"],
    top=10
)

for result in results:
    print(f"{result['title']}")
    if result.get("@search.captions"):
        print(f"  标题说明:{result['@search.captions'][0].text}")

筛选条件

results = client.search(
    search_text="*",
    filter="category eq 'Technology' and rating gt 4",
    order_by=["rating desc"],
    select=["id", "title", "category", "rating"]
)

分面

results = client.search(
    search_text="*",
    facets=["category,count:10", "rating"],
    top=0  # 仅获取分面,不获取文档
)

for facet_name, facet_values in results.get_facets().items():
    print(f"{facet_name}:")
    for facet in facet_values:
        print(f"  {facet['value']}: {facet['count']}")

自动完成与推荐

# 自动完成
results = client.autocomplete(
    search_text="sea",
    suggester_name="my-suggester",
    mode="twoTerms"
)

# 推荐
results = client.suggest(
    search_text="sea",
    suggester_name="my-suggester",
    select=["title"]
)

带技能集的索引器

from azure.search.documents.indexes import SearchIndexerClient
from azure.search.documents.indexes.models import (
    SearchIndexer,
    SearchIndexerDataSourceConnection,
    SearchIndexerSkillset,
    EntityRecognitionSkill,
    InputFieldMappingEntry,
    OutputFieldMappingEntry
)

with SearchIndexerClient(endpoint, DefaultAzureCredential()) as indexer_client:
    # 使用托管身份(搜索服务必须在存储账户上拥有 RBAC 角色)。避免使用包含嵌入式密钥的存储连接字符串。
    data_source = SearchIndexerDataSourceConnection(
        name="my-datasource",
        type="azureblob",
        connection_string="ResourceId=/subscriptions//resourceGroups//providers/Microsoft.Storage/storageAccounts/",
        container={"name": "documents"}
    )
    indexer_client.create_or_update_data_source_connection(data_source)

    # 创建技能集
    skillset = SearchIndexerSkillset(
        name="my-skillset",
        skills=[
            EntityRecognitionSkill(
                inputs=[InputFieldMappingEntry(name="text", source="/document/content")],
                outputs=[OutputFieldMappingEntry(name="organizations", target_name="organizations")]
            )
        ]
    )
    indexer_client.create_or_update_skillset(skillset)

    # 创建索引器
    indexer = SearchIndexer(
        name="my-indexer",
        data_source_name="my-datasource",
        target_index_name="my-index",
        skillset_name="my-skillset"
    )
    indexer_client.create_or_update_indexer(indexer)

最佳实践

  1. 请选择同步或异步模式并保持一致。请勿在同一调用路径中混合使用azure.xxx同步客户端与azure.xxx.aio异步客户端。每个模块应选择一种模式。
  2. 始终为客户端和异步凭据使用上下文管理器。将每个客户端用 `Client(...) as client:`(同步)或`Client(...) as client:`异步)进行封装。 对于来自azure.identity.aio 的异步DefaultAzureCredential,也应配合 async 模式使用credential:,以便对令牌和传输进行清理。
  3. 使用混合搜索,结合向量和关键词以获得最佳相关性
  4. 针对自然语言查询启用语义排序
  5. 为提高效率,请以100-1000 份文档为一批进行索引
  6. 在排序前使用过滤器缩小结果范围
  7. 配置向量维度以匹配您的嵌入模型
  8. 使用 HNSW 算法进行大规模向量搜索
  9. 在创建索引时生成建议器(无法在后续添加)

参考文件

文件 内容
references/vector-search.md HNSW 配置、集成向量化、多向量查询
references/semantic-ranking.md 语义配置、图片说明、答案、混合模式
scripts/setup_vector_index.py 用于创建支持向量搜索的索引的 CLI 脚本

其他 Azure AI Search 模式

Azure AI Search Python SDK

使用azure-search-documents 为 Azure AI Search 编写简洁、符合惯例的 Python 代码。

安装

pip install azure-search-documents azure-identity

环境变量

AZURE_SEARCH_ENDPOINT=https://.search.windows.net  # 所有身份验证方法均需此项
AZURE_SEARCH_INDEX_NAME= # 所有身份验证方法均需此项
AZURE_TOKEN_CREDENTIALS=prod # 仅当在生产环境中使用 DefaultAzureCredential 时才需要

身份验证

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.search.documents import SearchClient

# 本地开发环境:DefaultAzureCredential。生产环境:设置 AZURE_TOKEN_CREDENTIALS=prod 或 AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# 或者在生产环境中直接使用特定凭据:
# 参见 https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

with SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
    credential=credential,
) as client:
    results = list(client.search(search_text="*", top=5))

客户端选择

客户端 用途
SearchClient 查询索引、上传/更新/删除文档
SearchIndexClient 创建/管理索引、知识源、知识库
SearchIndexerClient 管理索引器、技能集、数据源
KnowledgeBaseRetrievalClient 基于大语言模型(LLM)的问答代理式检索

索引创建模式

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SearchField, VectorSearch, VectorSearchProfile,
    HnswAlgorithmConfiguration, AzureOpenAIVectorizer,
    AzureOpenAIVectorizerParameters, SemanticSearch,
    SemanticConfiguration, SemanticPrioritizedFields, SemanticField
)

index = SearchIndex(
    name=index_name,
    fields=[
        SearchField(name="id", type="Edm.String", key=True),
        SearchField(name="content", type="Edm.String", searchable=True),
        SearchField(name="embedding", type="Collection(Edm.Single)",
                   vector_search_dimensions=3072,
                   vector_search_profile_name="vector-profile"),
    ],
    vector_search=VectorSearch(
        profiles=[VectorSearchProfile(
            name="vector-profile",
            algorithm_configuration_name="hnsw-algo",
            vectorizer_name="openai-vectorizer"
        )],
        algorithms=[HnswAlgorithmConfiguration(name="hnsw-algo")],
        vectorizers=[AzureOpenAIVectorizer(
            vectorizer_name="openai-vectorizer",
            parameters=AzureOpenAIVectorizerParameters(
                resource_url=aoai_endpoint,
                deployment_name=embedding_deployment,
                model_name=embedding_model
            )
        )]
    ),
    semantic_search=SemanticSearch(
        default_configuration_name="semantic-config",
        configurations=[SemanticConfiguration(
            name="semantic-config",
            prioritized_fields=SemanticPrioritizedFields(
                content_fields=[SemanticField(field_name="content")]
            )
        )]
    )
)

with SearchIndexClient(endpoint, credential) as index_client:
    index_client.create_or_update_index(index)

文档操作

from azure.search.documents import SearchIndexingBufferedSender

# 支持自动分批的批量上传
with SearchIndexingBufferedSender(endpoint, index_name, credential) as sender:
    sender.upload_documents(documents)

# 通过 SearchClient 进行直接操作
with SearchClient(endpoint, index_name, credential) as search_client:
    search_client.upload_documents(documents)      # 添加新文档
    search_client.merge_documents(documents)       # 更新现有文档
    search_client.merge_or_upload_documents(documents)  # 插入或更新
    search_client.delete_documents(documents)      # 删除

搜索模式

# 基本搜索
results = search_client.search(search_text="query")

# 向量搜索
from azure.search.documents.models import VectorizedQuery

results = search_client.search(
    search_text=None,
    vector_queries=[VectorizedQuery(
        vector=embedding,
        k_nearest_neighbors=5,
        fields="embedding"
    )]
)

# 混合搜索(向量 + 关键词)
results = search_client.search(
    search_text="query",
    vector_queries=[VectorizedQuery(vector=embedding, k_nearest_neighbors=5, fields="embedding")],
    query_type="semantic",
    semantic_configuration_name="semantic-config"
)

# 带筛选条件
results = search_client.search(
    search_text="query",
    filter="category eq 'technology'",
    select=["id", "title", "content"],
    top=10
)

代理式检索(知识库)

有关基于大语言模型(LLM)且支持答案生成的问答功能,请参阅 references/agentic-retrieval.md。

关键概念:

  • 知识源:指向搜索索引
  • 知识库:封装知识源与 LLM,用于查询规划和答案合成
  • 输出模式EXTRACTIVE_DATA(原始片段)或ANSWER_SYNTHESIS(LLM 生成的答案)

异步模式

from azure.search.documents.aio import SearchClient

async with SearchClient(endpoint, index_name, credential) as client:
    results = await client.search(search_text="query")
    async for result in results:
        print(result["title"])

最佳实践

  1. 请使用环境变量设置端点、密钥和部署名称
  2. 对于在本地运行的代码,请使用DefaultAzureCredential(而非 API 密钥)。对于在 Azure 中运行的代码,请使用特定的令牌凭据。
  3. 批量上传时请使用SearchIndexingBufferedSender(负责处理批处理和重试)
  4. 始终为代理检索索引定义语义配置
  5. 使用create_or_update_index进行幂等索引创建
  6. 使用上下文管理器或显式调用 close()关闭客户端

字段类型参考

EDM 类型 Python 备注
Edm.String str 可搜索文本
Edm.Int32 int 整数
Edm.Int64 int 长整数
Edm.Double float 浮点数
Edm.Boolean bool 真/假
Edm.DateTimeOffset datetime ISO 8601
集合(Edm.Single) List[float] 向量嵌入
集合(Edm.String) List[str] 字符串数组

错误处理

from azure.core.exceptions import (
    HttpResponseError,
    ResourceNotFoundError,
    ResourceExistsError
)

try:
    result = search_client.get_document(key="123")
except ResourceNotFoundError:
    print("未找到文档")
except HttpResponseError as e:
    print(f"搜索错误:{e.message}")
在 GitHub 上查看
---
name: azure-search-documents-py
description: Search Azure AI Search indexes using the Python SDK for full-text, vector, hybrid, and semantic search with AI enrichment.
license: MIT
---

# Azure AI Search SDK for Python

Full-text, vector, and hybrid search with AI enrichment capabilities.

## Installation

```bash
pip install azure-search-documents
```

## Environment Variables

```bash
AZURE_SEARCH_ENDPOINT=https://<service-name>.search.windows.net  # Required for all auth methods
AZURE_SEARCH_INDEX_NAME=<your-index-name>  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
AZURE_SEARCH_API_KEY=<your-api-key>  # Only required for the legacy API-key auth path below
```

## Authentication & Lifecycle

> **🔑 Two rules apply to every code sample below:**
>
> 1. **Prefer `DefaultAzureCredential`.** It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
>    - Local dev: `DefaultAzureCredential` works as-is.
>    - Production: set `AZURE_TOKEN_CREDENTIALS=prod` (or `AZURE_TOKEN_CREDENTIALS=<specific_credential>`) to constrain the credential chain to production-safe credentials.
> 2. **Wrap every client in a context manager** so HTTP transports, sockets, and token caches are released deterministically:
>    - Sync: `with <Client>(...) as client:`
>    - Async: `async with <Client>(...) as client:` **and** `async with DefaultAzureCredential() as credential:` (from `azure.identity.aio`)
>
> Snippets may abbreviate this setup, but production code should always follow both rules.

```python
import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.search.documents import SearchClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

with SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
    credential=credential,
) as client:
    results = list(client.search(search_text="*", top=5))
```

### Legacy: API Key (existing keyed deployments)

New code should use `DefaultAzureCredential` above. Use `AzureKeyCredential` only if you have an existing keyed deployment that hasn't been migrated to Entra ID yet — for example, regulated environments still completing their Entra rollout. The same `AzureKeyCredential` works with `SearchIndexClient` and `SearchIndexerClient` for admin operations.

```python
import os
from azure.core.credentials import AzureKeyCredential
from azure.search.documents import SearchClient

with SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
    credential=AzureKeyCredential(os.environ["AZURE_SEARCH_API_KEY"]),
) as client:
    results = list(client.search(search_text="*", top=5))
```

## Client Types

| Client | Purpose |
|--------|---------|
| `SearchClient` | Search and document operations |
| `SearchIndexClient` | Index management, synonym maps |
| `SearchIndexerClient` | Indexers, data sources, skillsets |

## Create Index with Vector Field

```python
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    SearchFieldDataType,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    SearchableField,
    SimpleField
)

fields = [
    SimpleField(name="id", type=SearchFieldDataType.String, key=True),
    SearchableField(name="title", type=SearchFieldDataType.String),
    SearchableField(name="content", type=SearchFieldDataType.String),
    SearchField(
        name="content_vector",
        type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
        searchable=True,
        vector_search_dimensions=1536,
        vector_search_profile_name="my-vector-profile"
    )
]

vector_search = VectorSearch(
    algorithms=[
        HnswAlgorithmConfiguration(name="my-hnsw")
    ],
    profiles=[
        VectorSearchProfile(
            name="my-vector-profile",
            algorithm_configuration_name="my-hnsw"
        )
    ]
)

index = SearchIndex(
    name="my-index",
    fields=fields,
    vector_search=vector_search
)

with SearchIndexClient(endpoint, DefaultAzureCredential()) as index_client:
    index_client.create_or_update_index(index)
```

## Upload Documents

```python
from azure.search.documents import SearchClient

documents = [
    {
        "id": "1",
        "title": "Azure AI Search",
        "content": "Full-text and vector search service",
        "content_vector": [0.1, 0.2, ...]  # 1536 dimensions
    }
]

with SearchClient(endpoint, "my-index", DefaultAzureCredential()) as client:
    result = client.upload_documents(documents)
    print(f"Uploaded {len(result)} documents")
```

## Keyword Search

```python
results = client.search(
    search_text="azure search",
    select=["id", "title", "content"],
    top=10
)

for result in results:
    print(f"{result['title']}: {result['@search.score']}")
```

## Vector Search

```python
from azure.search.documents.models import VectorizedQuery

# Your query embedding (1536 dimensions)
query_vector = get_embedding("semantic search capabilities")

vector_query = VectorizedQuery(
    vector=query_vector,
    k_nearest_neighbors=10,
    fields="content_vector"
)

results = client.search(
    vector_queries=[vector_query],
    select=["id", "title", "content"]
)

for result in results:
    print(f"{result['title']}: {result['@search.score']}")
```

## Hybrid Search (Vector + Keyword)

```python
from azure.search.documents.models import VectorizedQuery

vector_query = VectorizedQuery(
    vector=query_vector,
    k_nearest_neighbors=10,
    fields="content_vector"
)

results = client.search(
    search_text="azure search",
    vector_queries=[vector_query],
    select=["id", "title", "content"],
    top=10
)
```

## Semantic Ranking

```python
from azure.search.documents.models import QueryType

results = client.search(
    search_text="what is azure search",
    query_type=QueryType.SEMANTIC,
    semantic_configuration_name="my-semantic-config",
    select=["id", "title", "content"],
    top=10
)

for result in results:
    print(f"{result['title']}")
    if result.get("@search.captions"):
        print(f"  Caption: {result['@search.captions'][0].text}")
```

## Filters

```python
results = client.search(
    search_text="*",
    filter="category eq 'Technology' and rating gt 4",
    order_by=["rating desc"],
    select=["id", "title", "category", "rating"]
)
```

## Facets

```python
results = client.search(
    search_text="*",
    facets=["category,count:10", "rating"],
    top=0  # Only get facets, no documents
)

for facet_name, facet_values in results.get_facets().items():
    print(f"{facet_name}:")
    for facet in facet_values:
        print(f"  {facet['value']}: {facet['count']}")
```

## Autocomplete & Suggest

```python
# Autocomplete
results = client.autocomplete(
    search_text="sea",
    suggester_name="my-suggester",
    mode="twoTerms"
)

# Suggest
results = client.suggest(
    search_text="sea",
    suggester_name="my-suggester",
    select=["title"]
)
```

## Indexer with Skillset

```python
from azure.search.documents.indexes import SearchIndexerClient
from azure.search.documents.indexes.models import (
    SearchIndexer,
    SearchIndexerDataSourceConnection,
    SearchIndexerSkillset,
    EntityRecognitionSkill,
    InputFieldMappingEntry,
    OutputFieldMappingEntry
)

with SearchIndexerClient(endpoint, DefaultAzureCredential()) as indexer_client:
    # Use managed identity (search service must have RBAC role on the storage account). Avoid storage connection strings with embedded keys.
    data_source = SearchIndexerDataSourceConnection(
        name="my-datasource",
        type="azureblob",
        connection_string="ResourceId=/subscriptions/<sub>/resourceGroups/<rg>/providers/Microsoft.Storage/storageAccounts/<acct>",
        container={"name": "documents"}
    )
    indexer_client.create_or_update_data_source_connection(data_source)

    # Create skillset
    skillset = SearchIndexerSkillset(
        name="my-skillset",
        skills=[
            EntityRecognitionSkill(
                inputs=[InputFieldMappingEntry(name="text", source="/document/content")],
                outputs=[OutputFieldMappingEntry(name="organizations", target_name="organizations")]
            )
        ]
    )
    indexer_client.create_or_update_skillset(skillset)

    # Create indexer
    indexer = SearchIndexer(
        name="my-indexer",
        data_source_name="my-datasource",
        target_index_name="my-index",
        skillset_name="my-skillset"
    )
    indexer_client.create_or_update_indexer(indexer)
```

## Best Practices

1. **Pick sync OR async and stay consistent.** Do not mix `azure.xxx` sync clients with `azure.xxx.aio` async clients in the same call path. Choose one mode per module.
2. **Always use context managers for clients and async credentials.** Wrap every client in `with Client(...) as client:` (sync) or `async with Client(...) as client:` (async). For async `DefaultAzureCredential` from `azure.identity.aio`, also use `async with credential:` so tokens and transports are cleaned up.
3. **Use hybrid search** for best relevance combining vector and keyword
4. **Enable semantic ranking** for natural language queries
5. **Index in batches** of 100-1000 documents for efficiency
6. **Use filters** to narrow results before ranking
7. **Configure vector dimensions** to match your embedding model
8. **Use HNSW algorithm** for large-scale vector search
9. **Create suggesters** at index creation time (cannot add later)

## Reference Files

| File | Contents |
|------|----------|
| [references/vector-search.md](references/vector-search.md) | HNSW configuration, integrated vectorization, multi-vector queries |
| [references/semantic-ranking.md](references/semantic-ranking.md) | Semantic configuration, captions, answers, hybrid patterns |
| [scripts/setup_vector_index.py](scripts/setup_vector_index.py) | CLI script to create vector-enabled search index |


---

## Additional Azure AI Search Patterns

# Azure AI Search Python SDK

Write clean, idiomatic Python code for Azure AI Search using `azure-search-documents`.

## Installation

```bash
pip install azure-search-documents azure-identity
```

## Environment Variables

```bash
AZURE_SEARCH_ENDPOINT=https://<search-service>.search.windows.net  # Required for all auth methods
AZURE_SEARCH_INDEX_NAME=<index-name>  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
```

## Authentication

```python
import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.search.documents import SearchClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

with SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
    credential=credential,
) as client:
    results = list(client.search(search_text="*", top=5))
```

## Client Selection

| Client | Purpose |
|--------|---------|
| `SearchClient` | Query indexes, upload/update/delete documents |
| `SearchIndexClient` | Create/manage indexes, knowledge sources, knowledge bases |
| `SearchIndexerClient` | Manage indexers, skillsets, data sources |
| `KnowledgeBaseRetrievalClient` | Agentic retrieval with LLM-powered Q&A |

## Index Creation Pattern

```python
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SearchField, VectorSearch, VectorSearchProfile,
    HnswAlgorithmConfiguration, AzureOpenAIVectorizer,
    AzureOpenAIVectorizerParameters, SemanticSearch,
    SemanticConfiguration, SemanticPrioritizedFields, SemanticField
)

index = SearchIndex(
    name=index_name,
    fields=[
        SearchField(name="id", type="Edm.String", key=True),
        SearchField(name="content", type="Edm.String", searchable=True),
        SearchField(name="embedding", type="Collection(Edm.Single)",
                   vector_search_dimensions=3072,
                   vector_search_profile_name="vector-profile"),
    ],
    vector_search=VectorSearch(
        profiles=[VectorSearchProfile(
            name="vector-profile",
            algorithm_configuration_name="hnsw-algo",
            vectorizer_name="openai-vectorizer"
        )],
        algorithms=[HnswAlgorithmConfiguration(name="hnsw-algo")],
        vectorizers=[AzureOpenAIVectorizer(
            vectorizer_name="openai-vectorizer",
            parameters=AzureOpenAIVectorizerParameters(
                resource_url=aoai_endpoint,
                deployment_name=embedding_deployment,
                model_name=embedding_model
            )
        )]
    ),
    semantic_search=SemanticSearch(
        default_configuration_name="semantic-config",
        configurations=[SemanticConfiguration(
            name="semantic-config",
            prioritized_fields=SemanticPrioritizedFields(
                content_fields=[SemanticField(field_name="content")]
            )
        )]
    )
)

with SearchIndexClient(endpoint, credential) as index_client:
    index_client.create_or_update_index(index)
```

## Document Operations

```python
from azure.search.documents import SearchIndexingBufferedSender

# Batch upload with automatic batching
with SearchIndexingBufferedSender(endpoint, index_name, credential) as sender:
    sender.upload_documents(documents)

# Direct operations via SearchClient
with SearchClient(endpoint, index_name, credential) as search_client:
    search_client.upload_documents(documents)      # Add new
    search_client.merge_documents(documents)       # Update existing
    search_client.merge_or_upload_documents(documents)  # Upsert
    search_client.delete_documents(documents)      # Remove
```

## Search Patterns

```python
# Basic search
results = search_client.search(search_text="query")

# Vector search
from azure.search.documents.models import VectorizedQuery

results = search_client.search(
    search_text=None,
    vector_queries=[VectorizedQuery(
        vector=embedding,
        k_nearest_neighbors=5,
        fields="embedding"
    )]
)

# Hybrid search (vector + keyword)
results = search_client.search(
    search_text="query",
    vector_queries=[VectorizedQuery(vector=embedding, k_nearest_neighbors=5, fields="embedding")],
    query_type="semantic",
    semantic_configuration_name="semantic-config"
)

# With filters
results = search_client.search(
    search_text="query",
    filter="category eq 'technology'",
    select=["id", "title", "content"],
    top=10
)
```

## Agentic Retrieval (Knowledge Bases)

For LLM-powered Q&A with answer synthesis, see [references/agentic-retrieval.md](references/agentic-retrieval.md).

Key concepts:
- **Knowledge Source**: Points to a search index
- **Knowledge Base**: Wraps knowledge sources + LLM for query planning and synthesis
- **Output modes**: `EXTRACTIVE_DATA` (raw chunks) or `ANSWER_SYNTHESIS` (LLM-generated answers)

## Async Pattern

```python
from azure.search.documents.aio import SearchClient

async with SearchClient(endpoint, index_name, credential) as client:
    results = await client.search(search_text="query")
    async for result in results:
        print(result["title"])
```

## Best Practices

1. **Use environment variables** for endpoints, keys, and deployment names
2. **Use `DefaultAzureCredential`** for code that runs locally (instead of API keys). Use a specific token credential for code that runs in Azure.
3. **Use `SearchIndexingBufferedSender`** for batch uploads (handles batching/retries)
4. **Always define semantic configuration** for agentic retrieval indexes
5. **Use `create_or_update_index`** for idempotent index creation
6. **Close clients** with context managers or explicit `close()`

## Field Types Reference

| EDM Type | Python | Notes |
|----------|--------|-------|
| `Edm.String` | str | Searchable text |
| `Edm.Int32` | int | Integer |
| `Edm.Int64` | int | Long integer |
| `Edm.Double` | float | Floating point |
| `Edm.Boolean` | bool | True/False |
| `Edm.DateTimeOffset` | datetime | ISO 8601 |
| `Collection(Edm.Single)` | List[float] | Vector embeddings |
| `Collection(Edm.String)` | List[str] | String arrays |

## Error Handling

```python
from azure.core.exceptions import (
    HttpResponseError,
    ResourceNotFoundError,
    ResourceExistsError
)

try:
    result = search_client.get_document(key="123")
except ResourceNotFoundError:
    print("Document not found")
except HttpResponseError as e:
    print(f"Search error: {e.message}")
```

所有文件

0 个文件

安装 azure-search-documents-py

下载技能文件并将其解压到 .claude/skills/ 目录中。

下载ZIP

克隆仓库并复制技能文件到您的项目中。

git clone https://github.com/microsoft/skills/tree/main/.github/plugins/azure-sdk-python/skills/azure-search-documents-py # Copy SKILL.md to your .claude/skills/ directory

复制 复制
快速设置: 将技能文件夹复制到 .claude/skills/ Claude 会自动检测并使用该技能

相关技能

microservices-patterns
更新时间 2026-06-29
jpa-patterns
更新时间 2026-06-30
fabric-lakehouse
更新时间 2026-06-30
prisma-expert
更新时间 2026-06-29
OR