文档中心

向量数据库VikingDB

最佳实践

【向量库】视频搜索实践(文搜视频/图搜视频/视频搜视频)

以下操作基于API V2版本实现

一、背景介绍

1.1 需求场景

随着5G、短视频、直播等技术的普及，视频数据正以前所未有的速度爆发式增长，成为信息传播的主要载体。传统的视频搜索方法严重依赖于标题、标签、简介等手动标注的文本信息，这种方式面临着诸多挑战：语义鸿沟巨大（标签无法描述视频丰富的动态内容）、人工成本高昂（海量视频无法有效标注）、检索维度单一（无法理解画面、声音、动作等深层信息），导致用户难以精准、高效地找到所需内容。
在此背景下，基于向量检索的视频搜索技术应运而生，为解决上述难题提供了革命性的方案。通过将视频内容（包括视频帧、音频、文本等）理解并转换为高维特征向量，我们可以在向量空间中通过计算相似度来精准地匹配视频内容，极大地提升了视频搜索的广度、深度和精度。VikingDB作为一款专为海量数据场景设计的向量数据库，具备卓越的高性能、高可用性与强大的分布式处理能力，凭借其在字节跳动千亿级检索需求下的丰富实践经验，成为应对大数据时代视频搜索挑战的理想选择。
VikingDB结合业界领先的 doubao 多模态 embedding 模型，能够深度理解和处理视频中的图像、声音、文本等多模态信息，通过高效的向量检索，帮助用户在短视频平台、媒体资料库、在线教育、安防监控等多元化场景中，实现前所未有的精准视频搜索与内容分析。结合VikingDB的强劲性能和doubao模型的多模态理解能力，企业能够构建更智能、更高效、更懂用户的次世代视频搜索解决方案，全面提升用户体验与运营效率。

1.2 场景示例

短视频检索

以文搜视频： 用户输入“日落时分在海边弹吉他”，系统能深度理解这一场景的视觉元素（日落、海边）、物体（吉他）和氛围感，精准返回符合意境的视频，而不仅仅是标题或标签中包含这些关键词的视频。
以图搜视频： 用户上传一张网红打卡地的照片（如圣托里尼的蓝顶教堂），系统能够识别照片中的地点和风格，为其推荐相关的旅游Vlog、风景记录或打卡攻略视频。
以视频搜视频（精彩片段检索）： 用户上传一段包含“C罗标志性庆祝动作”的短视频，系统能快速在海量视频库中找到其他包含该特定动作的视频片段，极大地提升了内容发现和二次创作的效率。

媒体资产管理

以文搜视频： 视频制作人需要一段“充满未来感的城市夜景，搭配快节奏电子音乐”的素材，系统能够同时理解并检索视频画面的风格与音频的节奏、曲风，快速定位到最匹配的素材片段。
以图搜视频： 设计师提供一张设计图或情绪板（Mood Board），系统可以理解其色彩、构图和主题，从庞大的视频素材库中检索出视觉风格和艺术感相匹配的视频片段，用于广告或电影制作。
以视频搜视频： 导演上传一段样片，希望寻找素材库中与之光影风格、镜头运动方式或色调相似的视频片段，以保持影视作品整体风格的统一性。

智驾的数据标注检索

以文搜视频： 算法工程师输入“寻找在雨夜中，行人推着婴儿车横穿马路的视频片段”，系统能从海量的路采数据中精准找出这类稀有但关键的Corner Case，用于模型的针对性训练和评测。
以图搜视频： 上传一张罕见的临时交通指示牌照片，系统能从海量路采数据中，快速定位所有包含该指示牌的视频，用于感知模型的专项训练，提升自动驾驶系统对长尾场景的识别能力。
以视频搜视频： 截取一段导致系统紧急制动（Disengagement）的危险切入（Cut-in）场景视频，系统可以检索出所有具有相似车辆运动轨迹和交互模式的视频片段，帮助算法工程师进行归因分析和模型迭代。

智能家居与家庭安防

以文搜视频： 户主输入“快递员在门口放下包裹”，系统能快速从一整天的录像中，定位到快递员出现并完成派送的几秒钟视频，无需手动拖动时间轴翻找。
以图搜视频： 用户手机抓拍了一张在家门口徘徊的陌生人照片，上传后系统能自动检索所有家庭摄像头，找出该陌生人在所有时间点、所有摄像头下的活动录像。
以视频搜视频： 用户发现后院的植物被破坏，截取了一段监控中松鼠的视频，系统可以找出过去一个月所有松鼠在后院活动的视频片段，帮助用户了解其活动规律。

1.3 方案思路

## 1.4 方案优势

精准的视觉匹配：借助深层次特征提取，能将风格、场景和对象相似的视频准确检索；
超大规模高效检索：针对十亿量级的视频，VikingDB向量数据库能极快完成相似度匹配，让检索效率成倍提升；
跨模态搜索：不仅能实时接收并索引新视频，还可结合文本信息进行搜索，方便快速定位和分析目标素材；

二、快速搭建

2.1 案例场景与数据

下面我们将使用VikingDB向量数据库进行以文搜视频、以图搜视频和以视频搜视频，即输入文本或图片或视频，检索出对应的视频。本案例使用video_demo.zip数据，其中包含5条自然景观视频。（视频的生成依赖于https://console.volcengine.com/ark/region:ark+cn-beijing/model/detail?Id=doubao-seedance-1-0-pro平台）
请下载该数据压缩包并解压，解压后的视频文件夹需要和Python代码放在同一目录下。
video_demo.zip

2.2 服务开通、密钥与环境准备

请注意，以下每一步都是完成多模态检索的必要条件，请勿遗漏。

VikingDB账号注册与服务开通：注册账号及开通服务--向量数据库VikingDB-火山引擎；
TOS（对象存储Torch Object Storage）服务开通：去开通，开通后您可以把视频存在TOS桶中；
授权 VikingDB 服务可以访问您的 TOS 权限：去授权，授权后VikingDB就可以访问TOS中的视频；
获取安全凭证Access Key ID（简称为AK）和Secret Access Key（简称为SK）：Access Key(密钥)管理--API访问密钥(Access Key)-火山引擎；
推荐Python 3.7及以上；
运行环境准备，在终端（Terminal）输入以下指令：

VikingDB SDK安装	pip3 install --upgrade volcengine	用于使用VikingDB的服务
TOS安装	pip3 install --upgrade tos	用于使用TOS服务
requests安装	pip3 install --upgrade requests	HTTP请求库
aiohttp安装	pip3 install --upgrade aiohttp	异步HTTP通信框架
tqdm安装	pip3 install --upgrade tqdm	用于进度条显示

2.3 数据准备

2.3.1 创建TOS桶

网页操作和Python操作二选一即可，效果相同。
此处创建一个TOS桶，此处桶的命名为"video-demo1"。可以修改TOS桶名称，但是不建议修改，因为此处修改后，后面的代码全部都要修改才能正常运行。

TOS是什么？
火山引擎对象存储 TOS（Torch Object Storage）是火山引擎提供的海量、安全、低成本、易用、高可靠、高可用的分布式云存储服务。您可以通过 RESTful API 接口、SDK 和工具等多种形式使用火山引擎 TOS。通过网络，您可以在任何应用、任何时间、任何地点管理和访问火山引擎 TOS 上的数据。
更多关于TOS的信息可以查阅TOS帮助文档。

在火山引擎官网打开对象存储（TOS），点击”桶列表“

点击”创建桶“；

给TOS桶命名、选择区域、桶策略，然后点击确定，就能看见列表里有

至此，您的TOS桶就创建完毕了。
此处初始化TOS服务的的Python SDK，然后创建一个TOS桶，此处桶的命名为"video-demo1"。可以修改TOS桶名称。

TOS是什么？
火山引擎对象存储 TOS（Torch Object Storage）是火山引擎提供的海量、安全、低成本、易用、高可靠、高可用的分布式云存储服务。您可以通过 RESTful API 接口、SDK 和工具等多种形式使用火山引擎 TOS。通过网络，您可以在任何应用、任何时间、任何地点管理和访问火山引擎 TOS 上的数据。
更多关于TOS的信息可以查阅TOS帮助文档。

需要您在代码开头填入您的ak和sk。

import tos # pip3 install --upgrade tos

# 初始化tos客户端
ak = "*"  # 双引号内填入您的Access Key ID（简称为AK）
sk = "*"  # 双引号内填入您的Secret Access Key（简称为SK）
endpoint = "tos-cn-beijing.volces.com" # 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com
region = "cn-beijing" # region 填写 cn-beijing。
client = tos.TosClientV2(ak, sk, endpoint, region)


# 创建TOS桶
bucket_name = "video-demo1" # TOS桶名称
try:
    # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
    client = tos.TosClientV2(ak, sk, endpoint, region)
    # 在创建存储空间时指定桶存储类型，可设置可选参数storage_class
    # 在创建存储空间时指定桶访问权限，可设置可选参数acl
    # 在创建存储空间时指定桶指定AZ属性, 可设置可选参数az_redundancy
    # 以配置桶为标准存储类型, 访问权限为私有, 数据容灾为3az存储为例
    client.create_bucket(bucket_name, acl=tos.ACLType.ACL_Private,
                         storage_class=tos.StorageClassType.Storage_Class_Standard,
                         az_redundancy=tos.AzRedundancyType.Az_Redundancy_Multi_Az)
except tos.exceptions.TosClientError as e:
    # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
    print('fail with client error, message:{}, cause: {}'.format(e.message, e.cause))
except tos.exceptions.TosServerError as e:
    # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
    print('fail with server error, code: {}'.format(e.code))
    # request id 可定位具体问题，强烈建议日志中保存
    print('error with request id: {}'.format(e.request_id))
    print('error with message: {}'.format(e.message))
    print('error with http code: {}'.format(e.status_code))
    print('error with ec: {}'.format(e.ec))
    print('error with request url: {}'.format(e.request_url))
except Exception as e:
    print('fail with unknown error: {}'.format(e))

至此，您的TOS桶就创建完毕了。

2.3.2 上传视频文件夹网页操作和Python操作二选一即可，效果相同。

此处将视频文件夹上传至上一步创建的TOS桶中。

点击刚才创建的“video-demo1”桶；

点击”上传文件“；

拖拽video-demo1文件夹拖拽进去，然后点击”上传“；

稍作等待，视频就会上传完成，任务状态会显示”执行成功“；

至此，上传视频到TOS桶中这一步骤就完成了。
需要您在代码开头填入您的ak和sk。

import tos # pip3 install --upgrade tos
import os
from tqdm import tqdm # pip3 install --upgrade tqdm 用于进度条显示


# 初始化tos客户端
ak = "*"  # 双引号内填入您的Access Key ID（简称为AK）
sk = "*"  # 双引号内填入您的Secret Access Key（简称为SK）
endpoint = "tos-cn-beijing.volces.com" # 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com
region = "cn-beijing" # region 填写 cn-beijing。
client = tos.TosClientV2(ak, sk, endpoint, region)


# 开始上传
bucket_name = "video-demo1" # tos桶名称
file_dir = "/path/to/your/video_demo" # 本地文件的当前路径


try:
    # 递归获取所有文件的路径
    def get_all_files(root_dir):
        file_paths = []
        for root, dirs, files in os.walk(root_dir):
            for file in files:
                file_paths.append(os.path.join(root, file))
        return file_paths


    all_files = get_all_files(file_dir)


    print(f"准备上传 {len(all_files)} 个文件...")


    # 使用 tqdm 显示进度条
    for file_path in tqdm(all_files, desc="视频上传进度", unit="file"):
        # --- 这是修改的关键 ---
        # 计算文件相对于根目录的相对路径，作为在TOS中的对象key
        object_key = os.path.relpath(file_path, file_dir)


        # 在Windows系统上，路径分隔符是'\'，为保证TOS上的路径统一，建议替换为'/'
        if os.name == 'nt':
            object_key = object_key.replace('\\', '/')


        # 使用相对路径作为 object_key 进行上传
        client.put_object_from_file(bucket_name, object_key, file_path)
    
    print("所有文件上传完成！")


except tos.exceptions.TosClientError as e:
    # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
    print('客户端错误, message:{}, cause: {}'.format(e.message, e.cause))
except tos.exceptions.TosServerError as e:
    # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
    print('服务端错误, code: {}'.format(e.code))
    print('Request ID: {}'.format(e.request_id))
    print('Message: {}'.format(e.message))
except Exception as e:
    print('未知错误: {}'.format(e))

2.3.3 导出TOS中视频的路径

该步骤只能使用代码完成。
使用VikingDB进行视频检索时，并不是将视频本身直接传入VikingDB，而是将TOS桶中视频的地址存入VikingDB。VikingDB会访问视频的TOS地址，自动将视频向量化，并把向量也存入VikingDB。
以下代码作用就是导出您存在TOS中视频的tos地址，并保存为一个jsonl文件。
需要您在代码开头填入您的ak和sk。

import tos # pip3 install --upgrade tos
import json
# 初始化tos客户端
ak = "*"  # 双引号内填入您的Access Key ID（简称为AK）
sk = "*"  # 双引号内填入您的Secret Access Key（简称为SK）
endpoint = "tos-cn-beijing.volces.com" # 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com
region = "cn-beijing" # region 填写 cn-beijing。
client = tos.TosClientV2(ak, sk, endpoint, region)
bucket_name = "video-demo1" # TOS桶名称
prefix = "video_demo/" # 前缀,即你的视频文件夹名称加一个斜杠，如果上传的是视频本身而不是文件夹，则prefix为空即可prefix = ""
output_file = "video.jsonl"  # 输出文件


try:
    # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
    client = tos.TosClientV2(ak, sk, endpoint, region)
    truncated = True
    continuation_token = ''
    image_id = 1  # 初始 id
    # 打开文件，准备写入 jsonl 格式
    with open(output_file, "w", encoding="utf-8") as f:
        while truncated:
            result = client.list_objects_type2(bucket_name, prefix=prefix, continuation_token=continuation_token)
            for item in result.contents:
                record = {
                    "video_id": str(image_id),
                    "landscape_video": f"tos://{bucket_name}/{item.key}" # 视频路径格式为：【tos://{bucket}/{object_key}】，不包含【】，可用于向量化
                }
                # 写入一行 JSON
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
                image_id += 1
            truncated = result.is_truncated
            continuation_token = result.next_continuation_token


except tos.exceptions.TosClientError as e:
    # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
    print('fail with client error, message:{}, cause: {}'.format(e.message, e.cause))
except tos.exceptions.TosServerError as e:
    # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
    print('fail with server error, code: {}'.format(e.code))
    print('error with request id: {}'.format(e.request_id))
    print('error with message: {}'.format(e.message))
    print('error with http code: {}'.format(e.status_code))
    print('error with ec: {}'.format(e.ec))
    print('error with request url: {}'.format(e.request_url))
except Exception as e:
    print('fail with unknown error: {}'.format(e))

代码运行完成后，会在代码所在目录下生成一个名为video.jsonl的文件。
文件中存有多条json，每一条json包含视频的id（"video_id"）和视频的tos地址（"
landscape_video"），tos地址会在后续步骤被传入VikingDB，json具体形式如下：

{"video_id": "1", "landscape_video": "tos://video-demo1/video_demo/0002.mp4"}

2.4 字段配置与创建VikingDB数据集

网页操作和Python操作二选一即可，效果相同。
此处在VikingDB中创建数据集（Collection）并配置字段，前面的视频tos地址就会被存在这个即将被创建的数据集中。

什么是字段？

字段定义：字段（fields）是数据表中的基本单位，每个字段存储特定类型的数据，比如“姓名”字段存储人的名字，“年龄”字段存储数字信息。

在VikingDB页面点击”创建数据集“；

选择”从向量化开始“，点击”开始创建“；

设置数据集名称为”video_demo1“；选择使用场景为“视频”，选择向量化模型为“Doubao-embedding- vision 250615”；选择向量维度为2048维；

视频向量化字段名称为landscape_video；其他字段名为video_id，类型int64；选择“从 string/int64 字段选择”，选择video_id；最后点击“创建数据集”；

此后您就能在数据集页面看到刚刚创建完成的”video-demo1“数据集；

需要您在代码中填入您的ak和sk。

import os
import datetime
import hashlib
import hmac
from urllib.parse import quote
import requests, json
def norm_query(params):
    query = ""
    for key in sorted(params.keys()):
        if type(params[key]) == list:
            for k in params[key]:
                query = (
                        query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
                )
        else:
            query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
    query = query[:-1]
    return query.replace("+", "%20")
def hmac_sha256(key: bytes, content: str):
    return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()
def hash_sha256(content: str):
    return hashlib.sha256(content.encode("utf-8")).hexdigest()
class ClientForConsoleApi:
    def __init__(self, ak, sk, host, region):
        self.ak = ak
        self.sk = sk
        self.host = host
        self.region = region
        self.service_code = "vikingdb"
        self.version = "2025-06-09"
    def request(self, method, action, body, query=None, header=None):
        if query is None:
            query = {}
        if header is None:
            header = {}
        credential = {
            "access_key_id": self.ak,
            "secret_access_key": self.sk,
            "service": self.service_code,
            "region": self.region,
        }
        request_param = {
            "body": json.dumps(body),
            "host": self.host,
            "path": "/",
            "method": method,
            "content_type": 'application/json',
            "date": datetime.datetime.utcnow(),
            "query": {"Action": action, "Version": self.version, **query},
        }
        if body is None:
            request_param["body"] = ""
        x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ")
        short_x_date = x_date[:8]
        x_content_sha256 = hash_sha256(request_param["body"])
        sign_result = {
            "Host": request_param["host"],
            "X-Content-Sha256": x_content_sha256,
            "X-Date": x_date,
            "Content-Type": request_param["content_type"],
        }
        signed_headers_str = ";".join(
            ["content-type", "host", "x-content-sha256", "x-date"]
        )
        canonical_request_str = "\n".join(
            [request_param["method"].upper(),
             request_param["path"],
             norm_query(request_param["query"]),
             "\n".join(
                 [
                     "content-type:" + request_param["content_type"],
                     "host:" + request_param["host"],
                     "x-content-sha256:" + x_content_sha256,
                     "x-date:" + x_date,
                 ]
             ),
             "",
             signed_headers_str,
             x_content_sha256,
             ]
        )
        hashed_canonical_request = hash_sha256(canonical_request_str)
        credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"])
        string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])
        k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date)
        k_region = hmac_sha256(k_date, credential["region"])
        k_service = hmac_sha256(k_region, credential["service"])
        k_signing = hmac_sha256(k_service, "request")
        signature = hmac_sha256(k_signing, string_to_sign).hex()
        sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
            credential["access_key_id"] + "/" + credential_scope,
            signed_headers_str,
            signature,
        )
        header = {**header, **sign_result}
        r = requests.request(method=method,
                             url="https://{}{}".format(request_param["host"], request_param["path"]),
                             headers=header,
                             params=request_param["query"],
                             data=request_param["body"],
                             )
        return r.status_code, r.headers, r.json()
if __name__ == '__main__':
    client = ClientForConsoleApi(
        ak = "*",#替换为您的ak
        sk ="*" ,#替换为您的sk
        host="vikingdb.cn-beijing.volcengineapi.com",
        region="cn-beijing"
    )
    http_code, _, result_json = client.request(
        method="POST",
        action="CreateVikingdbCollection",
        body={
        "CollectionName": "video_demo1",
        "Description": "test vectorize collection",
        "ProjectName": "default",
        "Fields": [
            {
                "FieldName": "video_id",
                "FieldType": "int64",
                "IsPrimaryKey": True
            }, {
                "FieldName": "video_demo",
                "FieldType": "video",
            }, 
        ],
        "Vectorize": {
            "Dense": {
                "ModelName": "doubao-embedding-vision",
                "VideoField": "video_demo",
                "ModelVersion": "250615"
            }
        }
    }    
    )
    print("req http status code: ", http_code)
    print("req result: \n", result_json)

2.5 导入视频TOS地址到VikingDB的数据集

该步骤只能使用代码完成。
此处代码将video.jsonl中的视频tos地址上传给video-demo1数据集中的"landscape_video"字段。
TOS地址传入VikingDB的同时，其对应的视频会被自动向量化（Embedding），这样后面就可以进行向量相似度检索。这里向量化的速率是有一定限制的，所以上传的视频数量越多，等待的时间越长。
需要您在代码中填入您的ak和sk。及前面导出的video.jsonl文件

"""
pip3 install volcengine
"""
import os
from volcengine.auth.SignerV4 import SignerV4
from volcengine.Credentials import Credentials
from volcengine.base.Request import Request
import requests, json
# 举例：从 JSONL 文件逐条导入
import time
import json
class ClientForDataApi:
    def __init__(self, ak, sk, host):
        self.ak = ak
        self.sk = sk
        self.host = host


    def prepare_request(self, method, path, params=None, data=None):
        r = Request()
        r.set_shema("https")
        r.set_method(method)
        r.set_connection_timeout(10)
        r.set_socket_timeout(10)
        mheaders = {
            'Accept': 'application/json',
            'Content-Type': 'application/json',
            'Host': self.host,
        }
        r.set_headers(mheaders)
        if params:
            r.set_query(params)
        r.set_host(self.host)
        r.set_path(path)
        if data is not None:
            r.set_body(json.dumps(data))
        credentials = Credentials(self.ak, self.sk, 'vikingdb', 'cn-beijing')
        SignerV4.sign(r, credentials)
        return r
        
    def do_req(self, req_method, req_path, req_params, req_body):
        req = self.prepare_request(method=req_method, path=req_path, params=req_params, data=req_body)
        return requests.request(method=req.method, url="http://{}{}".format(self.host, req.path),
                                  headers=req.headers, data=req.body, timeout=10000)
                                  
if __name__ == '__main__':
    client = ClientForDataApi(
        ak = "*",  # your ak
        sk = "*",  # your sk
        host = "api-vikingdb.vikingdb.cn-beijing.volces.com",
    )


    req_path = "/api/vikingdb/data/upsert"
    req_method = "POST"
    collection_name = "video_demo1"
    jsonl_path = "/path/to/your/video.jsonl"
    with open(jsonl_path, "r", encoding="utf-8") as f:
        for idx, line in enumerate(f, 1):
            obj = json.loads(line)
            # 按你的字段构造单条 data
            one = {
                "video_id": obj["video_id"],
                "landscape_video": {"value": obj["landscape_video"]},  # 若服务端字段是对象需 value，若是字符串就直接传字符串
            }
            req_body = {
                "collection_name": collection_name,
                "data": [one],  # 单条
            }
            resp = client.do_req(req_method=req_method, req_path=req_path, req_params=None, req_body=req_body)
            print(f"[{idx}] status={resp.status_code} body={resp.text}")
            # 可选：限速避免触发 QPS 限制
            # time.sleep(0.01)

上传数据成功，等待一段时间同步后，在VikingDB向量数据库网页端-数据集页面，video-demo1数据集后方就可以看见数据量更新（该数据量是近似准确的）。

2.6 创建索引

网页操作和Python操作二选一即可，效果相同。
此处为刚才的数据创建一个名为video_demo1的索引。
索引是对数据中某些字段的快速查找结构，VikingDB 索引支持对向量、标量的混合搜索需求。

向量索引参数说明：

distance：指衡量向量之间距离（相似度）的算法，可以选择IP（计算向量内积，内积越大向，量相似度越高）、L2（计算向量之间的欧氏距离，欧氏距离越小，向量相似度越高）、COSINE（计算向量之间的余弦相似度，余弦相似度越大，向量相似度越高）.
index_type：向量索引类型，可以选择的类型有HNSW、HNSW_HYBRID、FLAT、IVF、DISKANN。本案例中选择HNSW算法，HNSW通过构建多层网络减少搜索过程中需要访问的节点数量，实现快速高效地搜索最近邻，适合对搜索效率要求较高的场景。
quant：量化方式，指索引中对向量的压缩方式，可以降低向量间相似性计算的复杂度，可选类型有Int8、Float、Fix16、PQ。本案例中选择Float，也就是不对向量压缩，保留向量全部精度。
cpu_quota：指索引检索消耗的 CPU 配额，1CPU 核约为 100QPS，取值范围是2到10240。本案例中使用1个CPU配额。
description：对索引的描述。

找到刚才创建的video_demo数据集，点击其右侧的“创建索引”；

配置索引，设置索引名称为“video-demo1”，填写索引描述（非必填），选择数据集为video-demo1，选择 CPU 配额为 1 个 CPU 核，选择索引分片方式为“自动化分片”，选择索引类型为“内存索引”，选择索引算法为“HNSW”，选择距离类型为“COSINE”，选择量化方式为“Int8”，点击“提交”；

操作完成后，就能在索引页面看到您刚才创建的索引video-demo1；

稍作等待后，索引状态就会变成“已就绪”，此后就可以进行检索了；

此处代码为刚才的数据创建一个名为video_demo1的索引。索引是对数据中某些字段的快速查找结构，VikingDB 索引支持对向量、标量的混合搜索需求。

向量索引参数说明：

distance：指衡量向量之间距离（相似度）的算法，可以选择IP（计算向量内积，内积越大向，量相似度越高）、L2（计算向量之间的欧氏距离，欧氏距离越小，向量相似度越高）、COSINE（计算向量之间的余弦相似度，余弦相似度越大，向量相似度越高）.
index_type：向量索引类型，可以选择的类型有HNSW、HNSW_HYBRID、FLAT、IVF、DISKANN。本案例中选择HNSW算法，HNSW通过构建多层网络减少搜索过程中需要访问的节点数量，实现快速高效地搜索最近邻，适合对搜索效率要求较高的场景。
quant：量化方式，指索引中对向量的压缩方式，可以降低向量间相似性计算的复杂度，可选类型有Int8、Float、Fix16、PQ。本案例中选择Float，也就是不对向量压缩，保留向量全部精度。
cpu_quota：指索引检索消耗的 CPU 配额，1CPU 核约为 100QPS，取值范围是2到10240。
description：对索引的描述。

需要您在代码开头填入您的ak和sk。

import os
import datetime
import hashlib
import hmac
from urllib.parse import quote
import requests, json
def norm_query(params):
    query = ""
    for key in sorted(params.keys()):
        if type(params[key]) == list:
            for k in params[key]:
                query = (
                        query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
                )
        else:
            query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
    query = query[:-1]
    return query.replace("+", "%20")
def hmac_sha256(key: bytes, content: str):
    return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()
def hash_sha256(content: str):
    return hashlib.sha256(content.encode("utf-8")).hexdigest()
class ClientForConsoleApi:
    def __init__(self, ak, sk, host, region):
        self.ak = ak
        self.sk = sk
        self.host = host
        self.region = region
        self.service_code = "vikingdb"
        self.version = "2025-06-09"
    def request(self, method, action, body, query=None, header=None):
        if query is None:
            query = {}
        if header is None:
            header = {}
        credential = {
            "access_key_id": self.ak,
            "secret_access_key": self.sk,
            "service": self.service_code,
            "region": self.region,
        }
        request_param = {
            "body": json.dumps(body),
            "host": self.host,
            "path": "/",
            "method": method,
            "content_type": 'application/json',
            "date": datetime.datetime.utcnow(),
            "query": {"Action": action, "Version": self.version, **query},
        }
        if body is None:
            request_param["body"] = ""
        x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ")
        short_x_date = x_date[:8]
        x_content_sha256 = hash_sha256(request_param["body"])
        sign_result = {
            "Host": request_param["host"],
            "X-Content-Sha256": x_content_sha256,
            "X-Date": x_date,
            "Content-Type": request_param["content_type"],
        }
        signed_headers_str = ";".join(
            ["content-type", "host", "x-content-sha256", "x-date"]
        )
        canonical_request_str = "\n".join(
            [request_param["method"].upper(),
             request_param["path"],
             norm_query(request_param["query"]),
             "\n".join(
                 [
                     "content-type:" + request_param["content_type"],
                     "host:" + request_param["host"],
                     "x-content-sha256:" + x_content_sha256,
                     "x-date:" + x_date,
                 ]
             ),
             "",
             signed_headers_str,
             x_content_sha256,
             ]
        )
        hashed_canonical_request = hash_sha256(canonical_request_str)
        credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"])
        string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])
        k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date)
        k_region = hmac_sha256(k_date, credential["region"])
        k_service = hmac_sha256(k_region, credential["service"])
        k_signing = hmac_sha256(k_service, "request")
        signature = hmac_sha256(k_signing, string_to_sign).hex()
        sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
            credential["access_key_id"] + "/" + credential_scope,
            signed_headers_str,
            signature,
        )
        header = {**header, **sign_result}
        r = requests.request(method=method,
                             url="https://{}{}".format(request_param["host"], request_param["path"]),
                             headers=header,
                             params=request_param["query"],
                             data=request_param["body"],
                             )
        return r.status_code, r.headers, r.json()
if __name__ == '__main__':
    client = ClientForConsoleApi(
        ak = "*",#替换为您的ak
        sk ="*" ,#替换为您的sk
        host="vikingdb.cn-beijing.volcengineapi.com",
        region="cn-beijing"
    )
    http_code, _, result_json = client.request(
        method="POST",
        action="CreateVikingdbIndex",
        body={
        "CollectionName": "video_demo1",
        "IndexName": "video_demo1",
        "cpu_quota":1,
        "Description": "test create index",
        "Vector_index": {
        "Index_type": "hnsw",
        "Distance": "cosine",
        "quant": "int8",
    },
    }    
    )
    print("req http status code: ", http_code)
    print("req result: \n", result_json)

代码运行成功后进入VikingDB网页端的索引页面，就可以看见索引best_practice_pic_search_index。一般刚创建完时，其执行状态是”初始化中“，稍作等待后刷新页面，状态就会变为”已就绪“。接下来就可以进行视频检索了。

2.7 以文搜视频&以图搜视频&以视频搜视频

成功完成以上步骤后，就可以使用API进行检索了，在页面上也可以进行快速检索测试。

点击对应索引（video-demo1）右侧的”检索测试“。

输入文本，查询视频。

也可以输入图片，查询视频。

也可以输入视频，查询视频。

注意，如果删除了TOS桶中的视频或者TOS桶本身，那么进行视频检索时则无法显示视频。
成功完成以上步骤后，即可开始进行视频检索。以下示例展示了如何使用 Python 进行【以文搜视频】和【以图+文本搜视频】和【以视频搜视频】，并将检索到的 Top5 视频下载到本地。
【以文搜视频】操作说明：
请在代码开头的 QUERY_TEXT 位置填写您希望检索的文本内容。
运行代码后，系统会自动将检索到的 Top5 视频下载到当前目录下的 download_videos 文件夹中。

import json
import os
import sys
from pprint import pprint


import requests  # pip install requests
import tos  # pip3 install --upgrade tos
from volcengine.auth.SignerV4 import SignerV4  # pip3 install --upgrade volcengine
from volcengine.base.Request import Request
from volcengine.Credentials import Credentials

AK = "*"  # 替换为您的AK，注意保留双引号
SK = "*"  # 替换为您的SK，注意保留双引号
# VikingDB域名，请确认您实例所在的区域
DOMAIN = "api-vikingdb.vikingdb.cn-beijing.volces.com"
COLLECTION_NAME = "video_demo1"  # 替换为您的视频集合名称
INDEX_NAME = "video_demo1"      # 替换为您的视频索引名称
QUERY_TEXT = "风景"             # 您想要搜索的文本
SEARCH_LIMIT = 5                # 希望返回的结果数量
# TOS (对象存储) 配置
TOS_ENDPOINT = "tos-cn-beijing.volces.com"  # 您的TOS Endpoint
TOS_REGION = "cn-beijing"                   # 您的TOS Region
TOS_BUCKET = "video-demo1"             
# 本地保存路径配置
SAVE_DIR = "downloaded_videos"
os.makedirs(SAVE_DIR, exist_ok=True)

def prepare_request(method, path, params=None, data=None):
    """生成带V4签名的请求对象"""
    r = Request()
    r.set_shema("https")
    r.set_method(method)
    r.set_connection_timeout(10)
    r.set_socket_timeout(10)
    mheaders = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Host": DOMAIN,
    }
    r.set_headers(mheaders)
    if params:
        r.set_query(params)
    r.set_host(DOMAIN)
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data, ensure_ascii=False))
        
    credentials = Credentials(AK, SK, 'vikingdb', 'cn-beijing')
    SignerV4.sign(r, credentials)
    return r


def download_tos_video(tos_uri, save_path):
    """使用TOS SDK从TOS URI下载视频"""
    object_key = tos_uri.replace(f"tos://{TOS_BUCKET}/", "")
    client = tos.TosClientV2(AK, SK, TOS_ENDPOINT, TOS_REGION)
    try:
        print(f"⬇️  准备从TOS下载: {object_key}")
        client.get_object_to_file(TOS_BUCKET, object_key, save_path)
        print(f"✅  成功下载并保存到: {save_path}")
    except Exception as e:
        print(f"❌  下载失败 {tos_uri}: {e}")

if __name__ == "__main__":
    print("--- 步骤 1: 开始检索视频 ---")
    
    search_path = "/api/vikingdb/data/search/multi_modal"
    search_data = {
        "collection_name": COLLECTION_NAME,
        "index_name": INDEX_NAME,
        "text": QUERY_TEXT,
        "need_instruction": True,
        "limit": SEARCH_LIMIT
    }


    search_req = prepare_request(
        method="POST",
        path=search_path,
        data=search_data
    )


    response = requests.request(
        method=search_req.method,
        url=f"https://{DOMAIN}{search_req.path}",
        headers=search_req.headers,
        data=search_req.body
    )

    print("\n--- 步骤 2: 解析结果并下载视频 ---")
    if response.status_code != 200:
        print(f"❌ 检索请求失败，状态码: {response.status_code}")
        print("返回内容:", response.text)
        sys.exit(1)


    result = response.json()
    print("检索API返回原始结果:")
    pprint(result)

    if result.get("code") == 'Success':
        search_hits = result.get('result', {}).get('data', [])
        
        if not search_hits:
            print("\nℹ️  检索成功，但未找到匹配的视频。")
        else:
            print(f"\n检索到 {len(search_hits)} 个结果，开始下载...")
            for item in search_hits:
                fields = item.get("fields", {})
                
                landscape_video_info = fields.get('landscape_video', {})
                tos_url = landscape_video_info.get('value')


                if tos_url and tos_url.startswith("tos://"):
                    filename = tos_url.split("/")[-1]
                    local_path = os.path.join(SAVE_DIR, filename)
                    download_tos_video(tos_url, local_path)
                else:
                    print(f"⚠️  警告: 在条目中未找到有效的 'tos://' 链接，跳过。条目内容: {fields}")
    else:
        # 现在这个分支只会在真正失败时进入
        print(f"\n❌ 检索失败，错误信息: {result.get('message')}")

如下图所示，检索到的top5视频保存在了当前目录下的download_videos文件夹中。

【以文本+图搜视频】操作说明：
search_data中的text字段和image字段位置填写您希望用于检索的文本内容和图片路径。
运行代码后，系统会自动将检索到的 Top5 视频下载到当前目录下的 download_videos 文件夹中。

import json
import os
import sys
from pprint import pprint



import requests  # pip install requests
import tos  # pip3 install --upgrade tos
from volcengine.auth.SignerV4 import SignerV4  # pip3 install --upgrade volcengine
from volcengine.base.Request import Request
from volcengine.Credentials import Credentials

AK = "*"  # 替换为您的AK，注意保留双引号
SK = "*"  # 替换为您的SK，注意保留双引号
# VikingDB域名，请确认您实例所在的区域
DOMAIN = "api-vikingdb.vikingdb.cn-beijing.volces.com"



# VikingDB 视频检索参数配置
COLLECTION_NAME = "video_demo1"  # 替换为您的视频集合名称
INDEX_NAME = "video_demo1"      # 替换为您的视频索引名称
SEARCH_LIMIT = 5                # 希望返回的结果数量


# TOS (对象存储) 配置
TOS_ENDPOINT = "tos-cn-beijing.volces.com"  # 您的TOS Endpoint
TOS_REGION = "cn-beijing"                   # 您的TOS Region
TOS_BUCKET = "video-demo1"             



# 本地保存路径配置
SAVE_DIR = "downloaded_videos"
os.makedirs(SAVE_DIR, exist_ok=True)

def prepare_request(method, path, params=None, data=None):
    """生成带V4签名的请求对象"""
    r = Request()
    r.set_shema("https")
    r.set_method(method)
    r.set_connection_timeout(10)
    r.set_socket_timeout(10)
    mheaders = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Host": DOMAIN,
    }
    r.set_headers(mheaders)
    if params:
        r.set_query(params)
    r.set_host(DOMAIN)
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data, ensure_ascii=False))
        
    credentials = Credentials(AK, SK, 'vikingdb', 'cn-beijing')
    SignerV4.sign(r, credentials)
    return r



def download_tos_video(tos_uri, save_path):
    """使用TOS SDK从TOS URI下载视频"""
    object_key = tos_uri.replace(f"tos://{TOS_BUCKET}/", "")
    client = tos.TosClientV2(AK, SK, TOS_ENDPOINT, TOS_REGION)
    try:
        print(f"⬇️  准备从TOS下载: {object_key}")
        client.get_object_to_file(TOS_BUCKET, object_key, save_path)
        print(f"✅  成功下载并保存到: {save_path}")
    except Exception as e:
        print(f"❌  下载失败 {tos_uri}: {e}")


if __name__ == "__main__":
    # 步骤一：构造并发送视频检索请求
    print("--- 步骤 1: 开始检索视频 ---")
    
    search_path = "/api/vikingdb/data/search/multi_modal"
    search_data = {
        "collection_name": COLLECTION_NAME,
        "index_name": INDEX_NAME,
        "text": "风景",
        "need_instruction": True,
        "limit": SEARCH_LIMIT,
        "image":"tos://landscape/landscape/00000000.jpg"
    }



    search_req = prepare_request(
        method="POST",
        path=search_path,
        data=search_data
    )



    response = requests.request(
        method=search_req.method,
        url=f"https://{DOMAIN}{search_req.path}",
        headers=search_req.headers,
        data=search_req.body
    )

    print("\n--- 步骤 2: 解析结果并下载视频 ---")
    if response.status_code != 200:
        print(f"❌ 检索请求失败，状态码: {response.status_code}")
        print("返回内容:", response.text)
        sys.exit(1)



    result = response.json()
    print("检索API返回原始结果:")
    pprint(result)

    if result.get("code") == 'Success':
        search_hits = result.get('result', {}).get('data', [])
        
        if not search_hits:
            print("\nℹ️  检索成功，但未找到匹配的视频。")
        else:
            print(f"\n检索到 {len(search_hits)} 个结果，开始下载...")
            for item in search_hits:
                fields = item.get("fields", {})
                
                landscape_video_info = fields.get('landscape_video', {})
                tos_url = landscape_video_info.get('value')

                if tos_url and tos_url.startswith("tos://"):
                    filename = tos_url.split("/")[-1]
                    local_path = os.path.join(SAVE_DIR, filename)
                    download_tos_video(tos_url, local_path)
                else:
                    print(f"⚠️  警告: 在条目中未找到有效的 'tos://' 链接，跳过。条目内容: {fields}")
    else:
        print(f"\n❌ 检索失败，错误信息: {result.get('message')}")

如下图所示，检索到的top5视频保存在了当前目录下的download_videos文件夹中。

注意，如果删除了TOS桶中的视频或者TOS桶本身，那么进行视频检索时则无法保存结果视频到本地。
【以视频搜视频】操作说明：
search_data中的video字段即视频搜视频的输入，运行代码后，系统会自动将检索到的 Top5 （SEARCH_LIMIT参数控制）视频下载到当前目录下的 download_videos 文件夹中。

import json
import os
import sys
from pprint import pprint
import requests  # pip install requests
import tos  # pip3 install --upgrade tos
from volcengine.auth.SignerV4 import SignerV4  # pip3 install --upgrade volcengine
from volcengine.base.Request import Request
from volcengine.Credentials import Credentials
# 签名与认证配置
AK = "*"  # 替换为您的AK，注意保留双引号
SK = "*"  # 替换为您的SK，注意保留双引号
# VikingDB域名，请确认您实例所在的区域
DOMAIN = "api-vikingdb.vikingdb.cn-beijing.volces.com"

# VikingDB 视频检索参数配置
COLLECTION_NAME = "video_demo1"  # 替换为您的视频集合名称
INDEX_NAME = "video_demo1"      # 替换为您的视频索引名称
SEARCH_LIMIT = 5                # 希望返回的结果数量
# TOS (对象存储) 配置
TOS_ENDPOINT = "tos-cn-beijing.volces.com"  # 您的TOS Endpoint
TOS_REGION = "cn-beijing"                   # 您的TOS Region

TOS_BUCKET = "video-demo1"             
# 本地保存路径配置
SAVE_DIR = "downloaded_videos"
os.makedirs(SAVE_DIR, exist_ok=True)
def prepare_request(method, path, params=None, data=None):
    """生成带V4签名的请求对象"""
    r = Request()
    r.set_shema("https")
    r.set_method(method)
    r.set_connection_timeout(10)
    r.set_socket_timeout(10)
    mheaders = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Host": DOMAIN,
    }
    r.set_headers(mheaders)
    if params:
        r.set_query(params)
    r.set_host(DOMAIN)
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data, ensure_ascii=False))
        
    credentials = Credentials(AK, SK, 'vikingdb', 'cn-beijing')
    SignerV4.sign(r, credentials)
    return r
def download_tos_video(tos_uri, save_path):
    """使用TOS SDK从TOS URI下载视频"""
    object_key = tos_uri.replace(f"tos://{TOS_BUCKET}/", "")
    client = tos.TosClientV2(AK, SK, TOS_ENDPOINT, TOS_REGION)
    try:
        print(f"⬇️  准备从TOS下载: {object_key}")
        client.get_object_to_file(TOS_BUCKET, object_key, save_path)
        print(f"✅  成功下载并保存到: {save_path}")
    except Exception as e:
        print(f"❌  下载失败 {tos_uri}: {e}")

if __name__ == "__main__":
    # 步骤一：构造并发送视频检索请求
    print("--- 步骤 1: 开始检索视频 ---")
    
    search_path = "/api/vikingdb/data/search/multi_modal"
    search_data = {
        "collection_name": COLLECTION_NAME,
        "index_name": INDEX_NAME,
        "limit": SEARCH_LIMIT,
        "need_instruction": False,
         "video": 
            {   
                "value": "tos://video-demo1/video_demo/0012.mp4",
                "fps": 1.0,
            },
        
    }
    search_req = prepare_request(
        method="POST",
        path=search_path,
        data=search_data
    )
    response = requests.request(
        method=search_req.method,
        url=f"https://{DOMAIN}{search_req.path}",
        headers=search_req.headers,
        data=search_req.body
    )

    # 步骤二：解析结果并下载视频
    print("\n--- 步骤 2: 解析结果并下载视频 ---")
    if response.status_code != 200:
        print(f"❌ 检索请求失败，状态码: {response.status_code}")
        print("返回内容:", response.text)
        sys.exit(1)

    result = response.json()
    print("检索API返回原始结果:")
    pprint(result)


    if result.get("code") == 'Success':
        search_hits = result.get('result', {}).get('data', [])
        
        if not search_hits:
            print("\nℹ️  检索成功，但未找到匹配的视频。")
        else:
            print(f"\n检索到 {len(search_hits)} 个结果，开始下载...")
            for item in search_hits:
                fields = item.get("fields", {})
                landscape_video_info = fields.get('landscape_video', {})
                tos_url = landscape_video_info.get('value')



                if tos_url and tos_url.startswith("tos://"):
                    filename = tos_url.split("/")[-1]
                    local_path = os.path.join(SAVE_DIR, filename)
                    download_tos_video(tos_url, local_path)
                else:
                    print(f"⚠️  警告: 在条目中未找到有效的 'tos://' 链接，跳过。条目内容: {fields}")
    else:
        print(f"\n❌ 检索失败，错误信息: {result.get('message')}")

如下图所示，检索到的top5视频保存在了当前目录下的download_videos文件夹中。

最近更新时间：2026.09.08 12:02:51

这个页面对您有帮助吗？

有用

无用

向量数据库VikingDB

1.1 需求场景

1.2 场景示例

1.3 方案思路 #

2.1 案例场景与数据 #

2.2 服务开通、密钥与环境准备 #

2.3 数据准备 #

2.3.1 创建TOS桶 #

2.3.2 上传视频文件夹 网页操作和Python操作二选一即可，效果相同。 #

2.3.3 导出TOS中视频的路径 #

2.4 字段配置与创建VikingDB数据集 #

2.5 导入视频TOS地址到VikingDB的数据集 #

2.6 创建索引 #

2.7 以文搜视频&以图搜视频&以视频搜视频 #