快速入门：在Azure Cosmos DB中使用Python进行矢量搜索

将Azure Cosmos DB中的矢量搜索与Python客户端库配合使用。 在应用程序中有效地存储和查询矢量数据。

本快速入门使用 JSON 文件中的示例酒店数据集，其中包含 文本嵌入-3-small 模型中的矢量。 数据集包括酒店名称、位置、说明和矢量嵌入。

在 GitHub 上查找包含资源预配的示例代码。

Prerequisites

• Azure 订阅服务

  • 如果没有 Azure 订阅，请创建一个免费帐户

• 现有的 Azure Cosmos DB 资源数据平面访问

  • 如果没有资源，请创建新资源
  • 防火墙已配置为允许访问您的客户端 IP 地址
  • 分配的基于角色的访问控制（RBAC）角色：
    • Cosmos DB 内置数据参与者 （数据平面）
    • 角色 ID： 00000000-0000-0000-0000-000000000002

• Azure OpenAI 资源

  • 配置的自定义域
  • 分配了基于角色的访问控制（RBAC）角色：
    • 认知服务 OpenAI 用户
    • 角色 ID： 5e0bd9bd-7b93-4f28-af87-19fc36ad61bd
  • text-embedding-3-small 已部署的模型

• Visual Studio Code

  • Cosmos DB 扩展

• Python 3.10 或更高版本

使用矢量创建数据文件

1. 为酒店数据文件创建新的数据目录：

   • Linux/macOS
   • Windows操作系统

   mkdir data
   

2. 将包含矢量的 raw 数据文件下载到 data 目录：

   • Linux/macOS
   • Windows操作系统

   curl -o data/HotelsData_toCosmosDB_Vector.json https://raw.githubusercontent.com/Azure-Samples/cosmos-db-vector-samples/refs/heads/main/data/HotelsData_toCosmosDB_Vector.json
   

创建Python项目

1. 在与数据目录相同的级别为项目创建新的同级目录，并在Visual Studio Code中打开它：

   mkdir vector-search-quickstart
   code vector-search-quickstart
   

2. 在终端中，创建并激活Python虚拟环境：

   python -m venv .venv
   

   • Linux/macOS
   • Windows操作系统

   source .venv/bin/activate
   

3. requirements.txt在项目根目录中创建包含以下内容的文件：

   azure-cosmos>=4.7.0
   azure-identity>=1.18.0
   openai>=1.57.0
   python-dotenv>=1.0.1
   

4. 安装所需的包：

   pip install -r requirements.txt
   

   • azure-cosmos - 用于数据库操作的Azure Cosmos DB客户端库
   • azure-identity - 用于无密码（托管标识）连接的Azure身份验证库
   • openai - OpenAI SDK，用于使用 Azure OpenAI 生成嵌入内容
   • python-dotenv - 从 .env 文件加载环境变量

5. .env在项目根目录中为环境变量创建文件：

   # Identity for local developer authentication with Azure CLI
   AZURE_TOKEN_CREDENTIALS=AzureCliCredential
   
   # Azure OpenAI Embedding Settings
   AZURE_OPENAI_EMBEDDING_MODEL=text-embedding-3-small
   AZURE_OPENAI_EMBEDDING_API_VERSION=2024-08-01-preview
   AZURE_OPENAI_EMBEDDING_ENDPOINT=
   
   # Cosmos DB configuration
   AZURE_COSMOSDB_ENDPOINT=
   
   # Data file
   DATA_FILE_WITH_VECTORS=../data/HotelsData_toCosmosDB_Vector.json
   FIELD_TO_EMBED=Description
   EMBEDDED_FIELD=DescriptionVector
   EMBEDDING_DIMENSIONS=1536
   

   将文件中的 .env 占位符值替换为你自己的信息：

   • AZURE_OPENAI_EMBEDDING_ENDPOINT：您的 Azure OpenAI 资源端点 URL
   • AZURE_COSMOSDB_ENDPOINT：您的 Azure Cosmos DB 终端 URL

了解文档架构

在生成应用程序之前，请了解向量如何存储在 Azure Cosmos DB 文档中。 每个酒店文档都包含：

• 标准字段：HotelId、、HotelNameDescription、Category等。
• 矢量字段： DescriptionVector - 一个由 1536 个浮点数构成的数组，表示酒店描述的语义含义

下面是酒店文档结构的简化示例：

{
  "HotelId": "1",
  "HotelName": "Stay-Kay City Hotel",
  "Description": "This classic hotel is fully-refurbished...",
  "Rating": 3.6,
  "DescriptionVector": [
    -0.04886505,
    -0.02030743,
    0.01763356,
    ...
    // 1536 dimensions total
  ]
}

有关存储嵌入的要点：

• 矢量数组 作为标准 JSON 数组存储在文档中
• 矢量策略定义路径（）、数据类型（/DescriptionVectorfloat32）、维度（1536）和距离函数（余弦）
• 索引策略 在矢量字段上创建矢量索引，以便进行高效的相似性搜索
• 应 从标准索引中排除 向量字段以优化插入性能

这些策略在此示例项目的 距离度量 的 Bicep 模板中定义。 有关矢量策略和索引的详细信息，请参阅 Azure Cosmos DB 中的矢量搜索。

为矢量搜索创建代码文件

为 Python 文件创建一个src目录。 添加两个文件：vector_search.py 和 utils.py 用于您的矢量搜索实现：

mkdir src
touch src/__init__.py
touch src/vector_search.py
touch src/utils.py

New-Item -ItemType Directory -Force -Path src | Out-Null
New-Item -ItemType File -Path src/__init__.py | Out-Null
New-Item -ItemType File -Path src/vector_search.py | Out-Null
New-Item -ItemType File -Path src/utils.py | Out-Null

为矢量搜索创建代码

将以下代码粘贴到 vector_search.py 文件中。

"""Azure Cosmos DB NoSQL Vector Search — main entry point.

Loads hotel data, bulk-inserts into the selected container (DiskANN or
QuantizedFlat), generates a query embedding via Azure OpenAI, and
executes a VectorDistance() similarity search.
"""

import os
import sys
from pathlib import Path

from dotenv import load_dotenv

sys.path.insert(0, str(Path(__file__).parent))

from utils import (
    get_clients_passwordless,
    get_clients,
    insert_data,
    print_search_results,
    read_file_return_json,
    validate_field_name,
    get_query_activity_id,
)

# ---------------------------------------------------------------------------
# Load environment
# ---------------------------------------------------------------------------
load_dotenv()

ALGORITHM_CONFIGS: dict[str, dict[str, str]] = {
    "diskann": {
        "container_name": "hotels_diskann",
        "algorithm_name": "DiskANN",
    },
    "quantizedflat": {
        "container_name": "hotels_quantizedflat",
        "algorithm_name": "QuantizedFlat",
    },
}


def _build_config() -> dict[str, str | int]:
    """Build runtime configuration from environment variables."""
    return {
        "query": "quintessential lodging near running trails, eateries, retail",
        "db_name": os.getenv("AZURE_COSMOSDB_DATABASENAME", "Hotels"),
        "algorithm": os.getenv("VECTOR_ALGORITHM", "diskann").strip().lower(),
        "data_file": os.getenv("DATA_FILE_WITH_VECTORS", "../data/HotelsData_toCosmosDB_Vector.json"),
        "embedded_field": os.getenv("EMBEDDED_FIELD", "DescriptionVector"),
        "embedding_dimensions": int(os.getenv("EMBEDDING_DIMENSIONS", "1536")),
        "deployment": os.getenv("AZURE_OPENAI_EMBEDDING_MODEL", "text-embedding-3-small"),
        "distance_function": os.getenv("VECTOR_DISTANCE_FUNCTION", "cosine"),
    }


def main() -> None:
    """Run the vector search demonstration."""
    config = _build_config()

    # Try passwordless auth first, fall back to key-based
    clients = get_clients_passwordless()
    if not clients["ai_client"] or not clients["db_client"]:
        clients = get_clients()

    ai_client = clients["ai_client"]
    db_client = clients["db_client"]

    try:
        algorithm = config["algorithm"]
        if algorithm not in ALGORITHM_CONFIGS:
            valid = ", ".join(ALGORITHM_CONFIGS)
            raise ValueError(
                f"Invalid algorithm '{algorithm}'. Must be one of: {valid}"
            )

        if not ai_client:
            raise RuntimeError(
                "Azure OpenAI client is not configured. "
                "Please check your environment variables."
            )
        if not db_client:
            raise RuntimeError(
                "Cosmos DB client is not configured. "
                "Please check your environment variables."
            )

        algo_cfg = ALGORITHM_CONFIGS[algorithm]
        container_name = algo_cfg["container_name"]

        database = db_client.get_database_client(config["db_name"])
        print(f"Connected to database: {config['db_name']}")

        container = database.get_container_client(container_name)
        print(f"Connected to container: {container_name}")
        print(f"\n📊 Vector Search Algorithm: {algo_cfg['algorithm_name']}")
        print(f"📏 Distance Function: {config['distance_function']}")

        # Verify the container exists
        try:
            container.read()
        except Exception as e:
            status_code = getattr(e, "status_code", None)
            if status_code == 404:
                raise RuntimeError(
                    f"Container or database not found. Ensure database "
                    f"'{config['db_name']}' and container '{container_name}' "
                    f"exist before running this script."
                ) from e
            raise

        data_path = Path(__file__).parent.parent / config["data_file"]
        data = read_file_return_json(str(data_path))
        insert_data(container, data)

        embedding_response = ai_client.embeddings.create(
            model=config["deployment"],
            input=[config["query"]],
        )
        query_embedding = embedding_response.data[0].embedding

        safe_field = validate_field_name(config["embedded_field"])
        query_text = (
            f"SELECT TOP 5 c.HotelName, c.Description, c.Rating, "
            f"VectorDistance(c.{safe_field}, @embedding) AS SimilarityScore "
            f"FROM c "
            f"ORDER BY VectorDistance(c.{safe_field}, @embedding)"
        )

        print("\n--- Executing Vector Search Query ---")
        print(f"Query: {query_text}")
        print(
            f"Parameters: @embedding (vector with {len(query_embedding)} dimensions)"
        )
        print("--------------------------------------\n")

        results = list(
            container.query_items(
                query=query_text,
                parameters=[{"name": "@embedding", "value": query_embedding}],
                enable_cross_partition_query=True,
            )
        )

        # Extract diagnostics
        response_headers = container.client_connection.last_response_headers
        activity_id = get_query_activity_id(response_headers)
        if activity_id:
            print(f"Query activity ID: {activity_id}")

        request_charge_raw = response_headers.get("x-ms-request-charge", "0") if response_headers else "0"
        try:
            request_charge = float(request_charge_raw)
        except (ValueError, TypeError):
            request_charge = 0.0

        print_search_results(results, request_charge)

    except Exception as error:
        print(f"App failed: {error}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

此代码：

• 从环境变量配置 DiskANN 或 quantizedFlat 失量算法。
• 使用无密码身份验证连接到 Azure OpenAI 和 Azure Cosmos DB。
• 从 JSON 文件加载预矢量化酒店数据。
• 将数据插入相应的容器。
• 生成自然语言查询的嵌入（quintessential lodging near running trails, eateries, retail）。
• VectorDistance执行 SQL 查询，以检索按相似性分数排名的前 5 个语义上相似的酒店。
• 处理缺少的客户端、无效的算法选择以及不存在的容器/数据库的错误。

了解代码：使用 Azure OpenAI 生成嵌入内容

该代码为查询文本创建嵌入内容：

embedding_response = ai_client.embeddings.create(
    model=config["deployment"],  # OpenAI embedding model, e.g. "text-embedding-3-small"
    input=[config["query"]],     # List of description strings to embed
)
query_embedding = embedding_response.data[0].embedding

此 OpenAI API 调用 用于 client.embeddings.create，将“经典住宿毗邻跑步小径”等文本转换为一个 1536 维的向量，以捕捉其语义含义。 有关生成嵌入的更多详细信息，请参阅 Azure OpenAI 嵌入文档。

了解代码：在 Azure Cosmos DB 中存储矢量

通过 upsert_item 函数插入所有具有矢量数组的文档。

for item in data:
    doc = {"id": item["HotelId"], **item}
    response = container.upsert_item(body=doc)

这会将酒店文档（包括其预生成的 DescriptionVector 数组）插入到容器中。 每个文档都从中获取一个 id 映射 HotelId的字段，该函数处理 upsert，以便可以安全地重新插入文档。

了解代码：运行矢量相似性搜索

该代码使用 VectorDistance 函数执行矢量搜索：

query_text = (
    f"SELECT TOP 5 c.HotelName, c.Description, c.Rating, "
    f"VectorDistance(c.{safe_field}, @embedding) AS SimilarityScore "
    f"FROM c "
    f"ORDER BY VectorDistance(c.{safe_field}, @embedding)"
)

results = list(
    container.query_items(
        query=query_text,
        parameters=[{"name": "@embedding", "value": query_embedding}],
        enable_cross_partition_query=True,
    )
)

此代码生成参数化的 SQL 查询，该查询使用 VectorDistance 函数将查询的嵌入向量（@embedding）与每个文档的存储向量字段（DescriptionVector）进行比较，并返回名称与相似性分数最高的前 5 家酒店（从最相似到最不相似）。 查询嵌入被作为参数传递，以防止注入，并且该嵌入来自先前的 Azure OpenAI 调用中的 embeddings.create。

此查询返回的内容：

• 基于矢量距离的前 5 家最相似的酒店
• 酒店属性：HotelName，Description，Rating
• SimilarityScore：一个数值，该值指示每个酒店与查询的相似程度
• 从最类似于最不相似的结果排序

有关函数 VectorDistance 的详细信息，请参阅 VectorDistance 文档。

创建实用工具函数

将以下代码粘贴到 utils.py：

"""Shared utilities for Azure Cosmos DB NoSQL vector search.

Provides client initialization (passwordless and key-based), JSON I/O,
bulk insert with RU tracking, field validation, and result formatting.
"""

import json
import os
import re
import time
from typing import Any, Optional


def get_clients() -> dict[str, Any]:
    """Get Azure OpenAI and Cosmos DB clients using key-based authentication.

    Returns dict with 'ai_client' and 'db_client' (either may be None if
    the required environment variables are missing).
    """
    from azure.cosmos import CosmosClient
    from openai import AzureOpenAI

    ai_client = None
    db_client = None

    api_key = os.getenv("AZURE_OPENAI_EMBEDDING_KEY", "")
    api_version = os.getenv("AZURE_OPENAI_EMBEDDING_API_VERSION", "")
    endpoint = os.getenv("AZURE_OPENAI_EMBEDDING_ENDPOINT", "")
    deployment = os.getenv("AZURE_OPENAI_EMBEDDING_MODEL", "")

    if api_key and api_version and endpoint and deployment:
        ai_client = AzureOpenAI(
            api_key=api_key,
            api_version=api_version,
            azure_endpoint=endpoint,
            azure_deployment=deployment,
        )

    cosmos_endpoint = os.getenv("AZURE_COSMOSDB_ENDPOINT", "")
    cosmos_key = os.getenv("AZURE_COSMOSDB_KEY", "")

    if cosmos_endpoint and cosmos_key:
        db_client = CosmosClient(url=cosmos_endpoint, credential=cosmos_key)

    return {"ai_client": ai_client, "db_client": db_client}


def get_clients_passwordless() -> dict[str, Any]:
    """Get Azure OpenAI and Cosmos DB clients using DefaultAzureCredential.

    Uses managed identity / Azure CLI credentials for passwordless auth.
    Returns dict with 'ai_client' and 'db_client' (either may be None).
    """
    from azure.cosmos import CosmosClient
    from azure.identity import DefaultAzureCredential, get_bearer_token_provider
    from openai import AzureOpenAI

    ai_client = None
    db_client = None

    api_version = os.getenv("AZURE_OPENAI_EMBEDDING_API_VERSION", "")
    endpoint = os.getenv("AZURE_OPENAI_EMBEDDING_ENDPOINT", "")
    deployment = os.getenv("AZURE_OPENAI_EMBEDDING_MODEL", "")

    if api_version and endpoint and deployment:
        credential = DefaultAzureCredential()
        token_provider = get_bearer_token_provider(
            credential, "https://cognitiveservices.azure.com/.default"
        )
        ai_client = AzureOpenAI(
            api_version=api_version,
            azure_endpoint=endpoint,
            azure_deployment=deployment,
            azure_ad_token_provider=token_provider,
        )

    cosmos_endpoint = os.getenv("AZURE_COSMOSDB_ENDPOINT", "")

    if cosmos_endpoint:
        credential = DefaultAzureCredential()
        db_client = CosmosClient(url=cosmos_endpoint, credential=credential)

    return {"ai_client": ai_client, "db_client": db_client}


def read_file_return_json(file_path: str) -> list[dict[str, Any]]:
    """Read a JSON file and return its parsed contents."""
    print(f"Reading JSON file from {file_path}")
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        print(f"Error: File '{file_path}' not found")
        raise
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in file '{file_path}': {e}")
        raise


def write_file_json(file_path: str, json_data: Any) -> None:
    """Serialize data to a JSON file."""
    try:
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump(json_data, f, indent=2, ensure_ascii=False)
        print(f"Wrote JSON file to {file_path}")
    except IOError as e:
        print(f"Error writing to file '{file_path}': {e}")
        raise


def _get_document_count(container: Any) -> int:
    """Return the number of documents in a Cosmos DB container."""
    query = "SELECT VALUE COUNT(1) FROM c"
    results = list(container.query_items(query=query, enable_cross_partition_query=True))
    return results[0] if results else 0


def insert_data(
    container: Any, data: list[dict[str, Any]]
) -> dict[str, Any]:
    """Bulk-insert documents into a Cosmos DB container.

    Skips insertion if the container already has documents.
    Each item gets an 'id' field mapped from 'HotelId'.

    Returns a dict with total, inserted, failed, skipped, and requestCharge.
    """
    existing_count = _get_document_count(container)

    if existing_count > 0:
        print(f"Container already has {existing_count} documents. Skipping insert.")
        return {
            "total": 0,
            "inserted": 0,
            "failed": 0,
            "skipped": existing_count,
            "requestCharge": 0.0,
        }

    print(f"Inserting {len(data)} items...")

    inserted = 0
    failed = 0
    total_request_charge = 0.0

    start_time = time.time()

    for item in data:
        doc = {"id": item["HotelId"], **item}
        try:
            response = container.upsert_item(body=doc)
            inserted += 1
            ru = _extract_ru_from_headers(container.client_connection.last_response_headers)
            total_request_charge += ru
        except Exception as e:
            status_code = getattr(e, "status_code", None)
            if status_code == 409:
                inserted += 1
            else:
                failed += 1
                print(f"  Insert failed for item {item.get('HotelId', '?')}: {e}")

    duration = time.time() - start_time
    print(f"Bulk insert completed in {duration:.2f}s")
    print(f"\nInsert Request Charge: {total_request_charge:.2f} RUs\n")

    return {
        "total": len(data),
        "inserted": inserted,
        "failed": failed,
        "skipped": 0,
        "requestCharge": total_request_charge,
    }


def _extract_ru_from_headers(headers: Optional[dict[str, str]]) -> float:
    """Extract the request charge (RU) from Cosmos DB response headers."""
    if not headers:
        return 0.0
    raw = headers.get("x-ms-request-charge", "0")
    try:
        return float(raw)
    except (ValueError, TypeError):
        return 0.0


def validate_field_name(field_name: str) -> str:
    """Validate a field name is a safe SQL identifier.

    Prevents NoSQL injection when interpolating field names into queries.
    Allows only letters, digits, and underscores; must start with a letter
    or underscore.

    Raises ValueError if the field name is invalid.
    """
    pattern = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
    if not pattern.match(field_name):
        raise ValueError(
            f'Invalid field name: "{field_name}". '
            "Field names must start with a letter or underscore and "
            "contain only letters, numbers, and underscores."
        )
    return field_name


def print_search_results(
    search_results: list[dict[str, Any]],
    request_charge: Optional[float] = None,
) -> None:
    """Print vector search results in a consistent format."""
    print("\n--- Search Results ---")
    if not search_results:
        print("No results found.")
        return

    for i, result in enumerate(search_results, 1):
        score = result.get("SimilarityScore", 0.0)
        name = result.get("HotelName", "Unknown")
        print(f"{i}. {name}, Score: {score:.4f}")

    if request_charge is not None:
        print(f"\nVector Search Request Charge: {request_charge:.2f} RUs")
    print("")


def get_query_activity_id(response_headers: Optional[dict[str, str]]) -> Optional[str]:
    """Extract the activity ID from Cosmos DB query response headers."""
    if not response_headers:
        return None
    return response_headers.get("x-ms-activity-id")


def get_bulk_operation_rus(headers: Optional[dict[str, str]]) -> float:
    """Extract total RU cost from Cosmos DB response headers."""
    return _extract_ru_from_headers(headers)

此实用工具模块提供以下 关键 功能：

• get_clients_passwordless：使用无密码身份验证为 Azure OpenAI 和 Azure Cosmos DB 创建和返回客户端。 在资源上启用 RBAC 并登录到 Azure CLI
• insert_data：将数据插入Azure Cosmos DB容器，并跟踪每个操作的请求单位（RU）
• print_search_results：打印矢量搜索的结果，包括分数和酒店名称
• validate_field_name：验证数据中是否存在字段名称
• get_bulk_operation_rus：从Azure Cosmos DB响应标头中提取总 RU 成本

使用 Azure CLI 进行身份验证

在运行应用程序之前登录到 Azure CLI，以便应用可以安全地访问 Azure 资源。

az login

该代码通过get_clients_passwordless中的utils.py函数使用本地开发者身份验证访问 Azure Cosmos DB 和 Azure OpenAI。 设置 AZURE_TOKEN_CREDENTIALS=AzureCliCredential后，可以确定性地从其凭据链中选择使用哪个凭据 DefaultAzureCredential 。 该函数依赖于DefaultAzureCredential，来自azure-identity包，它会遍历一个有序的凭据提供程序链，但优先按照环境变量通过解析获取Azure CLI凭据。 详细了解如何使用 Azure 标识库向 Azure 服务验证 Python 应用。

运行应用程序

VECTOR_ALGORITHM使用环境变量选择要运行的向量索引实现。 变量控制Azure Cosmos DB应用程序连接到的容器。

VECTOR_ALGORITHM=diskann python -m src.vector_search

$env:VECTOR_ALGORITHM="diskann"; python -m src.vector_search

VECTOR_ALGORITHM=quantizedflat python -m src.vector_search

$env:VECTOR_ALGORITHM="quantizedflat"; python -m src.vector_search

应用日志记录和输出显示：

• 容器连接状态
• 数据插入状态
• 具有酒店名称和相似性分数的搜索结果

Connected to database: Hotels
Connected to container: hotels_diskann

📊 Vector Search Algorithm: DiskANN
📏 Distance Function: cosine

Reading JSON file from ..\data\HotelsData_toCosmosDB_Vector.json
Container already has 50 documents. Skipping insert.

--- Executing Vector Search Query ---
Query: SELECT TOP 5 c.HotelName, c.Description, c.Rating, VectorDistance(c.DescriptionVector, @embedding) AS SimilarityScore FROM c ORDER BY VectorDistance(c.DescriptionVector, @embedding)
Parameters: @embedding (vector with 1536 dimensions)
--------------------------------------

Query activity ID: <ACTIVITY_ID>

--- Search Results ---
1. Royal Cottage Resort, Score: 0.4991
2. Country Comfort Inn, Score: 0.4786
3. Nordick's Valley Motel, Score: 0.4635
4. Economy Universe Motel, Score: 0.4461
5. Roach Motel, Score: 0.4388

Vector Search Request Charge: 5.33 RUs

Connected to database: Hotels
Connected to container: hotels_quantizedflat

📊 Vector Search Algorithm: QuantizedFlat
📏 Distance Function: cosine

Reading JSON file from ..\data\HotelsData_toCosmosDB_Vector.json
Container already has 50 documents. Skipping insert.

--- Executing Vector Search Query ---
Query: SELECT TOP 5 c.HotelName, c.Description, c.Rating, VectorDistance(c.DescriptionVector, @embedding) AS SimilarityScore FROM c ORDER BY VectorDistance(c.DescriptionVector, @embedding)
Parameters: @embedding (vector with 1536 dimensions)
--------------------------------------

Query activity ID: <ACTIVITY_ID>

--- Search Results ---
1. Royal Cottage Resort, Score: 0.4991
2. Country Comfort Inn, Score: 0.4786
3. Nordick's Valley Motel, Score: 0.4635
4. Economy Universe Motel, Score: 0.4461
5. Roach Motel, Score: 0.4388

Vector Search Request Charge: 5.33 RUs

距离指标

Azure Cosmos DB 支持三个距离函数，实现矢量相似性：

创建容器时，在 矢量嵌入策略 中设置距离函数。 这在示例存储库的 基础结构 中提供。 它定义为容器定义的一部分。

{
    name: 'hotels_diskann'
    partitionKeyPaths: [
        '/HotelId'
    ]
    indexingPolicy: {
        indexingMode: 'consistent'
        automatic: true
        includedPaths: [
        {
            path: '/*'
        }
        ]
        excludedPaths: [
        {
            path: '/_etag/?'
        }
        {
            path: '/DescriptionVector/*'
        }
        ]
        vectorIndexes: [
        {
            path: '/DescriptionVector'
            type: 'diskANN'
        }
        ]
    }
    vectorEmbeddingPolicy: {
        vectorEmbeddings: [
        {
            path: '/DescriptionVector'
            dataType: 'float32'
            dimensions: 1536
            distanceFunction: 'cosine'
        }
        ]
    }
}

{
    name: 'hotels_quantizedflat'
    partitionKeyPaths: [
        '/HotelId'
    ]
    indexingPolicy: {
        indexingMode: 'consistent'
        automatic: true
        includedPaths: [
        {
            path: '/*'
        }
        ]
        excludedPaths: [
        {
            path: '/_etag/?'
        }
        {
            path: '/DescriptionVector/*'
        }
        ]
        vectorIndexes: [
        {
            path: '/DescriptionVector'
            type: 'quantizedFlat'
        }
        ]
    }
    vectorEmbeddingPolicy: {
        vectorEmbeddings: [
        {
            path: '/DescriptionVector'
            dataType: 'float32'
            dimensions: 1536
            distanceFunction: 'cosine'
        }
        ]
    }
}

此 Bicep 代码定义了一个 Azure Cosmos DB 容器配置，用于存储具有矢量搜索功能的酒店文档。

解释相似性分数

在使用 余弦相似性的示例输出中：

• 0.4991 （皇家小屋度假村） - 最高相似性，最佳匹配“提供靠近跑道、餐馆和商店的住宿选项”
• 0.4388 （罗奇汽车旅馆） - 较低的相似性，仍然相关，但不太匹配
• 分数接近 1.0 表示更强的语义相似性
• 接近 0 的分数表示几乎没有相似性

重要说明：

• 绝对分数值取决于嵌入模型和数据
• 关注 相对排名 而不是绝对阈值
• Azure OpenAI 嵌入在使用余弦相似性时效果最好。

有关距离函数的详细信息，请参阅 什么是距离函数？

在 Visual Studio Code 中查看和管理数据

1. 在 Visual Studio Code 中选择 Cosmos DB 扩展 以连接到 Azure Cosmos DB 帐户。

2. 查看 Hotels 数据库中的数据和索引。

   

清理资源

不再需要用于NoSQL帐户的 API 时，可以删除相应的资源组。

az group delete --name $resourceGroupName

$parameters = @{
    Name = $RESOURCE_GROUP_NAME
}
Remove-AzResourceGroup @parameters

相关内容

• Azure Cosmos DB 中的向量搜索
• Azure Cosmos DB 的文档索引器（预览版）
• 在 Azure Cosmos DB 中嵌入矢量
• Azure RBAC 内置角色