Azure AI Search中的向量搜索：HNSW、配置文件以及Python RAG检索器
使用 Python 在 Azure AI Search 中构建向量索引：HNSW 与全量 KNN 的对比、向量特征描述、嵌入模型、向量查询及 RAG 提示词，此外还有混合搜索的实用技巧。
来源: Building a Vector Index in Azure AI Search: HNSW, Profiles, and RAG Retrieval
检索是RAG系统中决定最终答案是否有依据还是模糊不清的关键部分。本指南将介绍Azure AI Search中的向量搜索工作原理，随后会构建一个简单的Python服务，该服务能够接收问题、根据语义找到最相关的文本，并为语言模型生成提示词。需要说明的是：该服务曾被称为Azure Cognitive Search，许多旧文章和示例仍使用这个名称，但其核心概念并无差异。
RAG的两大组成部分
检索增强生成模型分为检索和生成两个阶段。当有问题出现时，系统会搜索知识库并提取最相关的段落，然后将这些段落连同问题一起传递给语言模型，从而使答案有真实的文本依据。
模型的表现仅取决于它所获得的输入质量。如果检索返回了错误的段落，那么无论模型能力多强，都会给出错误或含糊的答案。正因如此，搜索技术的选择至关重要。
为何关键词匹配不够理想
传统搜索是按字面词进行匹配的。对于“car”这样的查询，可能无法找到仅使用“automobile”一词的文档。而向量搜索则通过理解语义来解决这一问题。
每段文本都会被转换成一组数字，即嵌入向量。含义相似的文本在向量空间中会彼此靠近。查询时，问题也会被转换为嵌入向量，服务便会寻找与之最接近的已存储向量，这一过程被称为最近邻搜索。在 Azure AI Search 中，你可以在索引中定义一个向量字段并存储嵌入向量，该服务会构建一种结构，从而能够快速扫描海量向量。
Azure AI Search 中的算法与配置
该服务为向量提供了两种搜索算法：
- HNSW（分层可导航小世界）是一种近似最近邻搜索方法。它速度较快，通常是生产环境的首选。
算法并非直接关联到某个字段。索引会包含向量搜索配置，其中列有算法列表和配置文件列表，每个配置文件为一种算法选择指定名称。向量字段则通过名称来引用对应的配置文件，这样相同的算法设置就可以在多个字段之间共享。
该字段的类型为Collection(Edm.Single)，即单精度浮点数的集合。必须为它指定维度数，且该数值必须与你所使用的嵌入模型的输出尺寸相同。
系统的结构
存在两种路径。一种是在数据变更时或定期执行一次数据摄取操作，从而更新索引；另一种则是针对每个查询都进行检索。两者有一个共同要求：必须使用相同的嵌入模型。如果用不同的模型对文档和查询进行嵌入，向量就无法相互比较，结果也就毫无意义。
文档通常以分块的形式而非完整文件存储。较小的分块能实现更精准的检索，也更容易被纳入提示词中。每个分块都构成一个独立的搜索文档，其中包含文本、嵌入向量以及用于追溯来源的元数据。此示例使用的都是短文档，因此每个文档都算作一个分块。在实际系统中会添加分块步骤，但索引结构保持不变。
实际构建
你需要拥有 Azure 订阅、Python 3.9 或更高版本，以及一定的脚本运行经验。嵌入向量是通过一个小型开源模型在本地生成的，因此无需使用 Azure OpenAI 进行部署。关于切换到 Azure OpenAI 的内容将在文末介绍。
步骤 1：创建搜索服务
在 Azure 门户中创建一个 Azure AI Search 资源。此处免费套餐即可满足需求。创建完成后，复制两个值：服务端点，格式为 https://<your-service>.search.windows.net，以及“密钥”部分中的管理密钥。请将此密钥视为机密信息。在真实项目中，它应存储在环境变量或 Azure Key Vault 中，绝不能出现在源代码中。
步骤 2：安装库
需要安装两个包：Azure Search SDK 和 sentence-transformers。
pip install azure-search-documents sentence-transformers
步骤 3：定义向量索引
接下来的脚本会先使用向量场构建索引，然后在创建索引之前通过算法和配置文件来定义向量搜索参数。
from azure.core.credentials import AzureKeyCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
SearchIndex,
SimpleField,
SearchableField,
SearchField,
SearchFieldDataType,
VectorSearch,
HnswAlgorithmConfiguration,
HnswParameters,
VectorSearchProfile,
)
endpoint = "https://<your-service>.search.windows.net"
admin_key = "<your-admin-key>"
index_name = "rag-docs"
index_client = SearchIndexClient(
endpoint=endpoint,
credential=AzureKeyCredential(admin_key),
)
# 1. Define the fields. The embedding field is the vector field.
fields = [
SimpleField(name="doc_id", type=SearchFieldDataType.String, key=True, filterable=True),
SearchableField(name="text", type=SearchFieldDataType.String),
SimpleField(name="source", type=SearchFieldDataType.String, filterable=True),
SearchField(
name="embedding",
type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
searchable=True,
vector_search_dimensions=384, # must match the embedding model
vector_search_profile_name="my-vector-profile",
),
]
# 2. Define the vector search configuration: one algorithm and one profile.
vector_search = VectorSearch(
algorithms=[
HnswAlgorithmConfiguration(
name="my-hnsw",
parameters=HnswParameters(
m=4,
ef_construction=400,
ef_search=500,
metric="cosine",
),
)
],
profiles=[
VectorSearchProfile(
name="my-vector-profile",
algorithm_configuration_name="my-hnsw",
)
],
)
# 3. Create (or update) the index.
index = SearchIndex(name=index_name, fields=fields, vector_search=vector_search)
index_client.create_or_update_index(index)
print("Index created:", index_name)
该脚本中的几个选项需要解释。维度数为384，是因为所选的嵌入模型输出的维度就是如此。向量场指向一个名为my-vector-profile的配置文件，而该配置文件又指向一个名为my-hnsw的HNSW算法。距离度量标准为余弦值，这是文本嵌入常用的默认值。
HNSW提供了可调整的参数，Azure也设置了合理的默认值。文档中规定的参数为：每个节点的双向链接数m为4，构建图时候选列表的大小efConstruction为400，查询时候选列表的大小efSearch为500。提高这些参数值可以提升召回率，但会增加延迟和构建时间，因此建议先使用默认值，再根据自身需求进行调整。
第4步：嵌入并上传
该脚本会加载模型，创建一个包含五篇简短文档的微型知识库，计算这些文档的嵌入向量，并将它们上传到索引中。
from azure.search.documents import SearchClient
from sentence_transformers import SentenceTransformer
search_client = SearchClient(
endpoint=endpoint,
index_name=index_name,
credential=AzureKeyCredential(admin_key),
)
# Load the embedding model (produces 384-dimensional vectors)
model = SentenceTransformer("all-MiniLM-L6-v2")
# Our small knowledge base
documents = [
{"doc_id": "1", "source": "billing-faq",
"text": "You can update your payment method from the account settings page under Billing."},
{"doc_id": "2", "source": "billing-faq",
"text": "Refunds are processed within five to seven business days to the original payment method."},
{"doc_id": "3", "source": "shipping-faq",
"text": "Standard delivery takes three to five working days within the country."},
{"doc_id": "4", "source": "account-faq",
"text": "To reset your password, click the forgot password link on the login screen."},
{"doc_id": "5", "source": "shipping-faq",
"text": "International orders may take up to fourteen working days depending on customs."},
]
# Create embeddings for the text of each document
texts = [d["text"] for d in documents]
vectors = model.encode(texts)
# Attach the embedding to each document and upload
to_upload = []
for doc, vector in zip(documents, vectors):
to_upload.append({
"doc_id": doc["doc_id"],
"source": doc["source"],
"text": doc["text"],
"embedding": vector.tolist(),
})
result = search_client.upload_documents(documents=to_upload)
print("Uploaded", len(result), "documents.")
一个实际的项目会从文件或数据库中读取数据，将文本分割成小块，然后通常以批次形式，通过同一个upload_documents接口上传成千上万甚至数百万份文档。
第5步：基于向量的查询
检索过程会使用相同的模型将用户的问题进行嵌入处理，随后发起向量查询。VectorizedQuery对象用于告知服务应使用哪个向量、返回多少个邻近向量以及搜索哪个字段。
from azure.search.documents.models import VectorizedQuery
def search(question, k=3):
# Embed the question with the same model
query_vector = model.encode([question])[0]
vector_query = VectorizedQuery(
vector=query_vector.tolist(),
k_nearest_neighbors=k,
fields="embedding",
)
results = search_client.search(
search_text=None, # pure vector search
vector_queries=[vector_query],
select=["doc_id", "source", "text"],
)
output = []
for r in results:
output.append({
"score": r["@search.score"],
"source": r["source"],
"text": r["text"],
})
return output
for item in search("how do I get my money back", k=3):
print(round(item["score"], 4), "|", item["source"], "|", item["text"])
试着用“帮我退钱”这样的方式来表述问题。所有示例文档中都没有出现这些字眼，但关于退款的文档应当排在最前面，因为比较的是含义而非关键词。这正是RAG系统所需要的行为。
第6步：构建提示语
检索已完成。最后一步就是根据检索结果来构建提示语。这里并不指定具体的模型，因为你可以使用Azure OpenAI或任何其他服务提供商。该脚本仅负责准备输入内容。
def build_prompt(question, k=3):
hits = search(question, k=k)
context = "\n\n".join(f"- {hit['text']}" for hit in hits)
prompt = (
"You are a support assistant. Use only the context below to answer "
"the question. If the answer is not in the context, say that you do "
"not have enough information.\n\n"
f"Context:\n{context}\n\n"
f"Question: {question}\n"
"Answer:"
)
return prompt
print(build_prompt("how do I get my money back"))
生成的提示词包含问题以及最相关的段落。将其发送给模型即可完成基于检索的循环处理，其中Azure AI Search充当向量存储和检索工具。
替换为Azure OpenAI嵌入模型
许多Azure团队使用如text-embedding-3-small这样的托管嵌入模型，而非本地模型。改动很小：只需调用Azure OpenAI客户端来生成每个嵌入，并将索引维度数设置为与模型匹配的值，例如text-embedding-ada-002的维度数为1536。索引和查询代码中的其他部分保持不变。
from openai import AzureOpenAI
client = AzureOpenAI(
api_key="<your-azure-openai-key>",
api_version="2024-10-21",
azure_endpoint="https://<your-resource>.openai.azure.com/",
)
def get_embedding(text):
response = client.embeddings.create(
input=text,
model="text-embedding-3-small", # your deployment name
)
return response.data[0].embedding
Azure AI Search还支持集成式向量化功能：该服务会自行调用Azure OpenAI模型，在索引过程中将文本转换为向量，每次查询时也会再次进行转换。这样无需自行编写嵌入代码，也更适合大型处理流程，但上述基本流程已足以帮助理解向量搜索的工作原理。
可能出错的情况
导致首次结果最令人失望的通常有三种错误。第一种是维度不匹配：如果索引显示为384，但模型输出的数值却是1536，上传就会失败，解决方法是读取模型的输出尺寸并相应设置该字段。第二种是在数据导入和查询过程中混用不同的嵌入模型，这样得到的结果看似合理，但实际上完全是随机的。第三种是在不重新构建模型的情况下更换模型，因为此时所有存储的向量都属于旧模型的空间，必须重新生成并上传。
一个能节省时间的习惯是将模型名称和维度数保存在同一个配置值中，由数据导入脚本和查询函数共同读取，这样两者就不会出现偏差。
检查检索质量
在信任该检索工具之前，先写出大约十个实际的问题，并注明每个问题对应的文档应该是哪一份。通过搜索功能运行这些查询，查看排名靠前的三个结果。如果找不到正确的段落，常见的原因包括文本块大小过大或过小、针对您所在领域的嵌入模型性能不佳，或是HNSW的设置更注重速度而非召回率。可以将该近似搜索结果与在同一数据上使用的完整KNN算法进行对比，从而了解其丢失了多少召回率，因为KNN能够提供精确的答案作为参照。
超越演示阶段的注意事项
- 嵌入模型。它决定了向量的维度，并对检索质量有着重要影响，因此需谨慎选择。
- 文本分块。在建立索引之前，先将长文档分割成独立的段落。
search_text的值。该服务会将BM25关键词搜索结果与向量搜索结果合并。从这里开始，你可以添加自己的文档、分块功能、混合搜索机制以及自选的模型，从而将这个演示版本发展成完整的RAG应用。