azure-search-documents-py
microsoft/skills
使用 Python SDK 搜索 Azure AI Search 索引,支持带 AI 增强功能的全文搜索、向量搜索、混合搜索和语义搜索。
...展开全部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 密钥认证路径
身份验证与生命周期
🔑 以下所有代码示例均遵循两条规则:
- 优先使用
DefaultAzureCredential。它在本地(Azure CLI / VS Code / 开发者 CLI)和 Azure 环境(托管身份、工作负载身份)中均可使用,且无需修改代码。请避免使用连接字符串、账户/API 密钥——它们会绕过 Entra 的审计和密钥轮换机制。
- 本地开发:
DefaultAzureCredential可直接使用。- 生产环境:将
AZURE_TOKEN_CREDENTIALS设置为prod(或AZURE_TOKEN_CREDENTIALS=),以将凭据链限制为生产环境安全的凭据。- 将每个客户端封装在上下文管理器中,以确保 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也可用于SearchIndexClient和SearchIndexerClient进行管理操作。
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)
最佳实践
- 请选择同步或异步模式并保持一致。请勿在同一调用路径中混合使用
azure.xxx同步客户端与azure.xxx.aio异步客户端。每个模块应选择一种模式。 - 始终为客户端和异步凭据使用上下文管理器。将每个客户端
用 `Client(...) as client:`(同步)或`Client(...) as client:`(异步)进行封装。 对于来自azure.identity.aio的异步DefaultAzureCredential,也应配合 async 模式使用credential:,以便对令牌和传输进行清理。 - 使用混合搜索,结合向量和关键词以获得最佳相关性
- 针对自然语言查询启用语义排序
- 为提高效率,请以100-1000 份文档为一批进行索引
- 在排序前使用过滤器缩小结果范围
- 配置向量维度以匹配您的嵌入模型
- 使用 HNSW 算法进行大规模向量搜索
- 在创建索引时生成建议器(无法在后续添加)
参考文件
| 文件 | 内容 |
|---|---|
| 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"])
最佳实践
- 请使用环境变量设置端点、密钥和部署名称
- 对于在本地运行的代码,请使用
DefaultAzureCredential(而非 API 密钥)。对于在 Azure 中运行的代码,请使用特定的令牌凭据。 - 批量上传时请使用
SearchIndexingBufferedSender(负责处理批处理和重试) - 始终为代理检索索引定义语义配置
- 使用
create_or_update_index进行幂等索引创建 - 使用上下文管理器或显式
调用 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}")
---
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
复制





首页
