Azure 搜索文档 Python SDK
Azure AI Search Python SDK
支持全文搜索、向量搜索和混合搜索,并具备 AI 增强能力。
安装
pip install azure-search-documents环境变量
AZURE_SEARCH_ENDPOINT=https://<service-name>.search.windows.net
AZURE_SEARCH_API_KEY=<your-api-key>
AZURE_SEARCH_INDEX_NAME=<your-index-name>身份验证
API 密钥
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 (推荐)
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 | 索引器、数据源、技能集 |
创建包含向量字段的索引
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)
上传文档
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)} 份文档")
关键词搜索
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
Yo
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="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}")
过滤器
results = client.search(
search_text="*",
filter="category eq 'Technology' and rating gt 4",
order_by=["rating desc"],
select=["id", "title", "category", "rating"]
)分面搜索 (Facets)
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']}")
自动完成与建议
# 自动完成 (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) 的索引器
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 代码。
额外模式的安装
pip install azure-search-documents azure-identity额外模式的环境变量
AZURE_SEARCH_ENDPOINT=https://<search-service>.search.windows.net
AZURE_SEARCH_INDEX_NAME=<index-name>
用于 API 密钥认证(生产环境不推荐)
AZURE_SEARCH_API_KEY=<api-key>额外模式的身份验证
DefaultAzureCredential (推荐):
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
credential = DefaultAzureCredential()
client = SearchClient(endpoint, index_name, credential)
API 密钥:
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 问答的智能体检索 |
索引创建模式
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)
## 文档操作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) # 删除## 搜索模式基础搜索
results = search_client.search(search_text="query")向量搜索
from azure.search.documents.models import VectorizedQueryresults = 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 )## Agentic 检索 (知识库)
关于基于 LLM 的问答与答案合成,请参阅 references/agentic-retrieval.md。
核心概念:
- 知识源 (Knowledge Source):指向一个搜索索引
- 知识库 (Knowledge Base):封装知识源 + LLM,用于查询规划和合成
- 输出模式:
EXTRACTIVE_DATA (原始分块) 或 ANSWER_SYNTHESIS (LLM 生成的答案)
异步模式
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 密钥
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] | 字符串数组 |
错误处理
try:
result = search_client.get_document(key="123")
except ResourceNotFoundError:
print("Document not found")
except HttpResponseError as e:
print(f"Search error: {e.message}")使用场景
本技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务明确符合上述范围时使用此技能。
- 不要将输出结果视为针对特定环境的验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准,请停止操作并请求澄清。