Azure 搜索文档 Python SDK

azure-search-documents-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.70/5
使用5.3K

Azure AI Search Python SDK

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

安装

bash
pip install azure-search-documents

环境变量

bash
AZURE_SEARCH_ENDPOINT=https://<service-name>.search.windows.net
AZURE_SEARCH_API_KEY=<your-api-key>
AZURE_SEARCH_INDEX_NAME=<your-index-name>

身份验证

API 密钥

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

client = SearchClient(
endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
credential=AzureKeyCredential(os.environ["AZURE_SEARCH_API_KEY"])
)

Entra ID (推荐)

python
from azure.search.documents import SearchClient
from azure.identity import DefaultAzureCredential

client = SearchClient(
endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
index_name=os.environ["AZURE_SEARCH_INDEX_NAME"],
credential=DefaultAzureCredential()
)

客户端类型

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

创建包含向量字段的索引

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

index_client = SearchIndexClient(endpoint, AzureKeyCredential(key))

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
)

index_client.create_or_update_index(index)

上传文档

python
from azure.search.documents import SearchClient

client = SearchClient(endpoint, "my-index", AzureKeyCredential(key))

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

result = client.upload_documents(documents)
print(f"已上传 {len(result)} 份文档")

关键词搜索

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']}")

向量搜索

python
from azure.search.documents.models import VectorizedQuery

Yo

查询嵌入 (1536 维)
python
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']}")

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

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
)

语义排序

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}")

过滤器

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  # 仅获取分面结果,不返回文档
)

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']}")

自动完成与建议

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"] )

带有技能集 (Skillset) 的索引器

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

indexer_client = SearchIndexerClient(endpoint, AzureKeyCredential(key))

创建数据源

data_source = SearchIndexerDataSourceConnection( name="my-datasource", type="azureblob", connection_string=connection_string, 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. 使用混合搜索:结合向量和关键字以获得最佳相关性。
2. 启用语义排序:优化自然语言查询。
3. 分批进行索引
为了提高效率,建议处理 100-1000 份文档
4. 使用过滤器在排序前缩小结果范围
5. 配置向量维度以匹配您的嵌入模型
6. 使用 HNSW 算法进行大规模向量搜索
7. 在创建索引时创建建议项 (suggesters)(之后无法添加)

参考文件

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

---

额外的 Azure AI Search 模式

SDK 重点

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

额外模式的安装

bash
pip install azure-search-documents azure-identity

额外模式的环境变量

bash
AZURE_SEARCH_ENDPOINT=https://<search-service>.search.windows.net
AZURE_SEARCH_INDEX_NAME=<index-name>

用于 API 密钥认证(生产环境不推荐)

AZURE_SEARCH_API_KEY=<api-key>

额外模式的身份验证

DefaultAzureCredential (推荐):

python
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient

credential = DefaultAzureCredential()
client = SearchClient(endpoint, index_name, credential)

API 密钥:

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

client = SearchClient(endpoint, index_name, AzureKeyCredential(api_key))

客户端选择

| 客户端 | 用途 |
|--------|---------|
| SearchClient | 查询索引,上传/更新/删除文档 |
| SearchIndexClient | 创建/管理索引、知识源、知识库 |
| SearchIndexerClient | 管理索引器、技能集、数据源 |
| KnowledgeBaseRetrievalClient | 基于 LLM 问答的智能体检索 |

索引创建模式

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=[SemanticConfig


uration(
name="semantic-config",
prioritized_fields=SemanticPrioritizedFields(
content_fields=[SemanticField(field_name="content")]
)
)]
)
)

index_client = SearchIndexClient(endpoint, credential)
index_client.create_or_update_index(index)

code
## 文档操作
python
from azure.search.documents import SearchIndexingBufferedSender

使用自动分批进行批量上传

with SearchIndexingBufferedSender(endpoint, index_name, credential) as sender: sender.upload_documents(documents)

通过 SearchClient 直接操作

search_client = SearchClient(endpoint, index_name, credential) search_client.upload_documents(documents) # 新增 search_client.merge_documents(documents) # 更新现有文档 search_client.merge_or_upload_documents(documents) # Upsert (更新或新增) search_client.delete_documents(documents) # 删除
code
## 搜索模式
python

基础搜索

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 )
code
## Agentic 检索 (知识库)

关于基于 LLM 的问答与答案合成,请参阅 references/agentic-retrieval.md。

核心概念:

  • 知识源 (Knowledge Source):指向一个搜索索引

  • 知识库 (Knowledge Base):封装知识源 + LLM,用于查询规划和合成

  • 输出模式EXTRACTIVE_DATA (原始分块) 或 ANSWER_SYNTHESIS (LLM 生成的答案)

异步模式

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"])

code
## 最佳实践

1. 使用环境变量 存储端点、密钥和部署名称
2. 生产环境优先使用 DefaultAzureCredential 而非 API 密钥
3. 批量上传建议使用 SearchIndexingBufferedSender (支持分批和重试)
4. 为 Agentic 检索索引务必定义语义配置 (semantic configuration)
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 格式 |
| Collection(Edm.Single) | List[float] | 向量嵌入 |
| Collection(Edm.String) | List[str] | 字符串数组 |

错误处理

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

try:
result = search_client.get_document(key="123")
except ResourceNotFoundError:

code
print("Document not found")
except HttpResponseError as e:
print(f"Search error: {e.message}")

使用场景

本技能适用于执行概览中所描述的工作流或操作。

局限性

  • 仅在任务明确符合上述范围时使用此技能。
  • 不要将输出结果视为针对特定环境的验证、测试或专家评审的替代方案。
  • 如果缺少必要的输入、权限、安全边界或成功标准,请停止操作并请求澄清。