searchById--向量数据库VikingDB-火山引擎

文档中心

立即注册

导航

向量数据库VikingDB

searchById

最近更新时间：2025.11.14 17:04:48首次发布时间：2025.11.14 17:04:48

概述

search_by_id 用于主键 id 检索。根据主键 id，搜索与其距离最近的 limit 个向量。

说明

对于使用了 hnsw-hybrid 算法的混合索引，暂时不支持基于 id 进行检索。
Collection 数据写入/删除后，Index 数据更新时间预计20s，不能立即在 Index 检索到。
当请求参数 filter 配置时，表示混合检索；当请求参数 filter 没有配置时，表示纯向量检索。

前提条件

通过 create_collection 接口创建数据集时，定义字段 fields 已添加 vector 字段。
通过 upsert_data 接口写入数据时，已写入 vector 类型的字段名称和字段值。
通过 create_index 接口创建索引时，已创建 vector_index 向量索引。

请求参数

请求参数是 SearchByIDRequest，其字段如下表所示。

名称	类型	必选	描述
id	Any	是	要检索的主键值。
filter	Optional[Dict[str, Any]]	否	标量过滤条件，详见标量过滤。不填表示基于 id 的纯向量检索。支持 must、must_not、range、range_out 等算子，可用 and / or 组合。
output_fields	Optional[List[str]]	否	要返回的标量字段列表。未设置时返回集合内所有标量字段。传入空列表表示不返回任何标量字段。字段名必须存在于 collection schema，否则请求报错。
limit	Optional[int]	否	限制返回条数，最大 5000。
offset	Optional[int]	否	分页偏移量，默认 0，过大时会出现深分页性能开销。
partition	Optional[str]	否	仅检索指定分区，默认搜索全部分区。
advance	Optional[SearchAdvance]	否	高级参数集合（post_process_ops、ids_in 等），详见检索公共参数。

返回参数

名称	类型	描述
request_id	Optional[str]	请求链路 ID。
code	Optional[str]	服务返回码，Success 表示成功。
message	Optional[str]	错误或提示信息。
api	Optional[str]	具体调用的 API 名称。
result	Optional[SearchResult]	检索结果主体，结构见下。

SearchResult

名称	类型	描述
data	List[SearchItemResult]	召回到的结果列表，结构见下。
filter_matched_count	Optional[int]	满足过滤条件的文档总数。
total_return_count	Optional[int]	本次返回的结果数量。
real_text_query	Optional[str]	模型可能修正后的真实查询串。
token_usage	Dict[str, Any]	token 计量信息。

SearchItemResult

名称	类型	描述
id	Any	主键值。
fields	Dict[str, Any]	返回的标量字段内容。
score	Optional[float]	最终相似度得分。
ann_score	Optional[float]	ANN 粗排得分。

示例

请求参数

import os

from vikingdb import IAM
from vikingdb.vector import SearchByIDRequest, VikingVector

auth = IAM(
    ak=os.environ["VIKINGDB_AK"],
    sk=os.environ["VIKINGDB_SK"],
)
client = VikingVector(
    host=os.environ["VIKINGDB_HOST"],
    region=os.environ["VIKINGDB_REGION"],
    auth=auth,
    scheme="https",
)

index_client = client.index(
    collection_name=os.environ["VIKINGDB_COLLECTION"],
    index_name=os.environ["VIKINGDB_INDEX"],
)
request = SearchByIDRequest(
    id=1234567890,
    limit=5,
    output_fields=["title", "score"],
)
response = index_client.search_by_id(request)
print(f"request_id={response.request_id}")
if response.result:
    for item in response.result.data:
        print(item.id, item.score, item.fields.get("title"))