databricks-vector-search

Databricks Vector Search

首先：使用父级 databricks-core 技能了解 CLI 基础知识、身份验证和配置文件选择。

介绍为 RAG 和语义搜索应用创建、管理和查询向量索引的模式。

适用场景

在以下情况下使用此技能：

• 构建 RAG（检索增强生成）应用
• 实现语义搜索或相似度匹配
• 从 Delta 表创建向量索引
• 在存储优化型和标准端点之间进行选择
• 使用筛选条件查询向量索引

概述

Databricks Vector Search 提供托管式向量相似度搜索，并支持自动生成嵌入和 Delta Lake 集成。

组件	描述
端点	承载索引的计算资源（标准型或存储优化型）
索引	用于相似度搜索的向量数据结构
Delta Sync	自动与源 Delta 表同步
直接访问	手动对向量执行 CRUD 操作

端点类型

类型	延迟	容量	成本	最适用场景
标准型	20-50ms	320M 个向量（768 维）	较高	实时、低延迟场景
存储优化型	300-500ms	1B+ 个向量（768 维）	低 7 倍	大规模、成本敏感型场景

索引类型

类型	嵌入	同步	用例
Delta Sync（托管）	由 Databricks 计算	从 Delta 自动同步	设置最简单
Delta Sync（自管理）	由你提供	从 Delta 自动同步	自定义嵌入
直接访问	由你提供	手动 CRUD	实时更新

快速入门

创建端点

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Create a standard endpoint
endpoint = w.vector_search_endpoints.create_endpoint(
    name="my-vs-endpoint",
    endpoint_type="STANDARD"  # or "STORAGE_OPTIMIZED"
)
# Note: Endpoint creation is asynchronous; check status with get_endpoint()

创建 Delta Sync 索引（托管嵌入）

# Source table must have: primary key column + text column
index = w.vector_search_indexes.create_index(
    name="catalog.schema.my_index",
    endpoint_name="my-vs-endpoint",
    primary_key="id",
    index_type="DELTA_SYNC",
    delta_sync_index_spec={
        "source_table": "catalog.schema.documents",
        "embedding_source_columns": [
            {
                "name": "content",  # Text column to embed
                "embedding_model_endpoint_name": "databricks-gte-large-en"
            }
        ],
        "pipeline_type": "TRIGGERED"  # or "CONTINUOUS"
    }
)

查询索引

results = w.vector_search_indexes.query_index(
    index_name="catalog.schema.my_index",
    columns=["id", "content", "metadata"],
    query_text="What is machine learning?",
    num_results=5
)

for doc in results.result.data_array:
    score = doc[-1]  # Similarity score is last column
    print(f"Score: {score}, Content: {doc[1][:100]}...")

常见模式

创建存储优化型端点

# For large-scale, cost-effective deployments
endpoint = w.vector_search_endpoints.create_endpoint(
    name="my-storage-endpoint",
    endpoint_type="STORAGE_OPTIMIZED"
)

使用自管理嵌入进行 Delta Sync

# Source table must have: primary key + embedding vector column
index = w.vector_search_indexes.create_index(
    name="catalog.schema.my_index",
    endpoint_name="my-vs-endpoint",
    primary_key="id",
    index_type="DELTA_SYNC",
    delta_sync_index_spec={
        "source_table": "catalog.schema.documents",
        "embedding_vector_columns": [
            {
                "name": "embedding",  # Pre-computed embedding column
                "embedding_dimension": 768
            }
        ],
        "pipeline_type": "TRIGGERED"
    }
)

直接访问索引

import json

# Create index for manual CRUD
index = w.vector_search_indexes.create_index(
    name="catalog.schema.direct_index",
    endpoint_name="my-vs-endpoint",
    primary_key="id",
    index_type="DIRECT_ACCESS",
    direct_access_index_spec={
        "embedding_vector_columns": [
            {"name": "embedding", "embedding_dimension": 768}
        ],
        "schema_json": json.dumps({
            "id": "string",
            "text": "string",
            "embedding": "array<float>",
            "metadata": "string"
        })
    }
)

# Upsert data
w.vector_search_indexes.upsert_data_vector_index(
    index_name="catalog.schema.direct_index",
    inputs_json=json.dumps([
        {"id": "1", "text": "Hello", "embedding": [0.1, 0.2, ...], "metadata": "doc1"},
        {"id": "2", "text": "World", "embedding": [0.3, 0.4, ...], "metadata": "doc2"},
    ])
)

# Delete data
w.vector_search_indexes.delete_data_vector_index(
    index_name="catalog.schema.direct_index",
    primary_keys=["1", "2"]
)

使用嵌入向量查询

# When you have pre-computed query embedding
results = w.vector_search_indexes.query_index(
    index_name="catalog.schema.my_index",
    columns=["id", "text"],
    query_vector=[0.1, 0.2, 0.3, ...],  # Your 768-dim vector
    num_results=10
)

混合搜索（语义 + 关键词）

混合搜索结合了向量相似度（ANN）和 BM25 关键词评分。当查询中包含必须精确匹配的词项（例如 SKU、错误代码、专有名词或技术术语）时，请使用混合搜索；纯语义搜索可能会漏掉依赖关键词的结果。有关如何选择 ANN 或混合搜索的详细指南，请参阅 references/search-modes.md。

# Combines vector similarity with keyword matching
results = w.vector_search_indexes.query_index(
    index_name="catalog.schema.my_index",
    columns=["id", "content"],
    query_text="SPARK-12345 executor memory error",
    query_type="HYBRID",
    num_results=10
)

筛选

标准端点筛选（字典）

# filters_json uses dictionary format
results = w.vector_search_indexes.query_index(
    index_name="catalog.schema.my_index",
    columns=["id", "content"],
    query_text="machine learning",
    num_results=10,
    filters_json='{"category": "ai", "status": ["active", "pending"]}'
)

存储优化型筛选（类 SQL）

存储优化型端点通过 databricks-vectorsearch 包的 filters 参数（接受字符串）使用类 SQL 筛选语法：

from databricks.vector_search.client import VectorSearchClient

vsc = VectorSearchClient()
index = vsc.get_index(endpoint_name="my-storage-endpoint", index_name="catalog.schema.my_index")

# SQL-like filter syntax for storage-optimized endpoints
results = index.similarity_search(
    query_text="machine learning",
    columns=["id", "content"],
    num_results=10,
    filters="category = 'ai' AND status IN ('active', 'pending')"
)

# More filter examples
# filters="price > 100 AND price < 500"
# filters="department LIKE 'eng%'"
# filters="created_at >= '2024-01-01'"

触发索引同步

# For TRIGGERED pipeline type, manually sync
w.vector_search_indexes.sync_index(
    index_name="catalog.schema.my_index"
)

扫描所有索引条目

# Retrieve all vectors (for debugging/export)
scan_result = w.vector_search_indexes.scan_index(
    index_name="catalog.schema.my_index",
    num_results=100
)

参考文件

主题	文件	描述
索引类型	references/index-types.md	Delta Sync（托管/自管理）与直接访问的详细比较
端到端 RAG	references/end-to-end-rag.md	完整演练：源表 → 端点 → 索引 → 查询 → Agent 集成
搜索模式	references/search-modes.md	何时使用语义搜索（ANN）与混合搜索，以及选择指南
运维	references/troubleshooting-and-operations.md	监控、成本优化、容量规划、迁移

CLI 快速参考

# List endpoints
databricks vector-search-endpoints list-endpoints

# Create endpoint (positional args: NAME ENDPOINT_TYPE)
databricks vector-search-endpoints create-endpoint my-endpoint STANDARD

# List indexes on endpoint (positional arg: ENDPOINT_NAME)
databricks vector-search-indexes list-indexes my-endpoint

# Get index status (positional arg: INDEX_NAME)
databricks vector-search-indexes get-index catalog.schema.my_index

# Sync index (positional arg: INDEX_NAME)
databricks vector-search-indexes sync-index catalog.schema.my_index

# Delete index (positional arg: INDEX_NAME)
databricks vector-search-indexes delete-index catalog.schema.my_index

常见问题

问题	解决方案
索引同步缓慢	使用 Storage-Optimized 端点（索引速度快 20 倍）
查询延迟较高	使用 Standard 端点，将延迟控制在 100 毫秒以内
filters_json 无效	Storage-Optimized 使用类似 SQL 的字符串筛选器，通过 databricks-vectorsearch 包的 filters 参数传入
嵌入维度不匹配	确保查询和索引的维度一致
索引未更新	检查 pipeline_type；对于 TRIGGERED，请使用 sync_index()
容量不足	升级到 Storage-Optimized（支持 10 亿个以上向量）
query_vector 被截断	大型向量（例如 1024 维）序列化为 JSON 时可能会被截断。请改用 query_text（适用于托管嵌入索引），或使用 Databricks SDK 传递原始向量

嵌入模型

Databricks 提供内置嵌入模型：

模型	维度	上下文窗口	用例
databricks-gte-large-en	1024	8192 个 token	英文文本，高质量
databricks-bge-large-en	1024	512 个 token	英文文本，通用

# Use with managed embeddings
embedding_source_columns=[
    {
        "name": "content",
        "embedding_model_endpoint_name": "databricks-gte-large-en"
    }
]

说明

• Storage-Optimized 较新——除非需要低于 100 毫秒的延迟，否则更适合大多数用例
• 推荐使用 Delta Sync——对于大多数场景，它比 Direct Access 更简单
• 混合搜索——Delta Sync 和 Direct Access 索引均支持
• columns_to_sync 很重要——查询结果中仅包含已同步的列；请纳入所需的所有列
• 筛选语法因端点而异——Standard 使用字典格式的筛选器，Storage-Optimized 使用类似 SQL 的字符串筛选器。请使用 databricks-vectorsearch 包的 filters 参数，该参数支持这两种格式
• 管理与运行时——CLI 和 SDK 用于管理生命周期；运行时进行 Agent 工具调用时，请使用 VectorSearchRetrieverTool

相关技能

• databricks-model-serving - 部署使用 VectorSearchRetrieverTool 的 Agent
• databricks-agent-bricks - Knowledge Assistants 使用 RAG 检索已建立索引的文档
• databricks-unstructured-pdf-generation - 生成用于 Vector Search 建立索引的文档
• databricks-unity-catalog - 管理 Delta Sync 索引所依托的目录和表
• databricks-pipelines - 构建用作 Vector Search 数据源的 Delta 表