文档中心

向量数据库VikingDB

最佳实践

【向量库】多模态搜索实践(文搜图／图搜图)

以下最佳实践基于旧版API实现，后续我们会基于新版API V2更新，若使用新版API V2想参考最佳实践请联系我们。

一、背景介绍

1.1 需求场景

随着数据智能技术的飞速发展，图像搜索技术正在越来越广泛地应用于各行各业。传统的图像搜索方法大多依赖于图像的元数据或标签，存在信息更新滞后、知识局限性、搜索精度不足等问题。如何有效提升图像搜索的精度和效率，成为了当前亟待解决的关键问题。
在此背景下，基于向量检索的图像搜索技术逐渐成为一种突破性解决方案。通过将图像转换为高维向量并在向量空间内计算相似度，能够大幅提升图像搜索的效率和精度，解决传统图像搜索方式中的局限。VikingDB作为一款支持大规模数据场景的向量数据库，具备出色的高性能、高可用性和大数据处理能力，凭借对字节千亿级检索需求的丰富经验，成为在大数据规模下图像搜索领域的领先产品。
VikingDB结合最先进的doubao多模态embedding模型，能够有效处理文本与图像的多模态数据，并通过高效的向量检索，帮助用户在电商平台、图库平台、内容审核、安防监控等场景中实现精准的图像搜索与分析。结合VikingDB的强大性能和doubao模型的多模态能力，企业能够构建更加智能、精准且高效的图像搜索解决方案，推动图像搜索技术在各行业、各业务场景中的广泛应用。

1.2 场景示例

电商智能搜索

以文搜图：用户输入“复古风格皮鞋”，即使商品标题中没有“复古”二字，系统也能根据图片语义匹配到类似款式的皮鞋。
以图搜图：用户上传一张明星穿的衣服照片，系统自动找到相似款或相同品牌的商品，助力精准购物推荐。

社交媒体内容发现

以文搜图：用户搜索“森林系的婚纱摄影”，系统理解“森林系”的概念，返回具有自然、绿意背景的婚纱照，而不是简单匹配“森林”或“婚纱”标签的图片。
以图搜图：用户上传一张自己喜欢的插画风格图片，系统推荐相似风格的插画作品或相关创作者，提高内容发现的精准度。

版权保护与内容审核

以文搜图：输入“梵高风格的夜景油画”，系统搜索数据库中具有相同绘画风格的图片，而不仅仅是“梵高”相关标签的作品。
以图搜图：上传一张艺术作品，系统能找到相似构图、风格甚至被修改过的版本，帮助发现潜在侵权行为。

医疗影像智能检索

以文搜图：医生输入“CT显示肺部有磨玻璃影”，系统检索数据库中相似病变的医学影像，辅助医生参考类似病例。
以图搜图：医生上传患者的CT影像，系统从数据库中找到高度相似的病例，推荐相关诊断和治疗方案，提升诊疗效率。

1.3 方案思路

## 1.4 方案优势

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

二、快速搭建

2.1 案例场景与数据

下面我们将使用VikingDB向量数据库进行以文搜图和以图搜图，即输入文本或图片，检索出对应的图片。本案例使用pic_search_1000_images.zip数据，其中包含1000张野生动物图片。
请下载该数据压缩包并解压，解压后的图片文件夹需要和Python代码放在同一目录下。
pic_search_1000_images.zip

该数据抽样自 Sourav Banerjee 在 Kaggle 上发布的 Animal Image Dataset数据集，数据源为：https://www.kaggle.com/datasets/iamsouravbanerjee/animal-image-dataset-90-different-animals。

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桶，此处桶的命名为"best-practice-pic-search-tos"。可以修改TOS桶名称，但是不建议修改，因为此处修改后，后面的代码全部都要修改才能正常运行。

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

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

点击”创建桶“；

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

至此，您的TOS桶就创建完毕了。
此处初始化TOS服务的的Python SDK，然后创建一个TOS桶，此处桶的命名为"best-practice-pic-search-tos"。可以修改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 = "best-practice-pic-search-tos" # 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操作二选一即可，效果相同。
此处将图片文件夹pic_search_1000_images上传至上一步创建的TOS桶中。

点击刚才创建的“best-practice-pic-search-tos”桶；

点击”上传文件“；

将 pic_search_1000_images 文件夹拖拽进去，然后点击”上传“；

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

至此，上传图片到TOS桶中这一步骤就完成了。
此处将图片文件夹pic_search_1000_images上传至上一步创建的TOS桶中，所以需要您将pic_search_1000_images文件夹和该Python代码放在同一目录下。
需要您在代码开头填入您的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 = "best-practice-pic-search-tos" # tos桶名称
file_dir = "pic_search_1000_images" # 本地文件的当前路径
try:
    # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
    client = tos.TosClientV2(ak, sk, endpoint, region)

    # 递归获取所有文件的路径
    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)

    # 使用 tqdm 显示进度条
    for file_path in tqdm(all_files, desc="图片上传进度", unit="file"):
        client.put_object_from_file(bucket_name, file_path, file_path)

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))

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 = "best-practice-pic-search-tos" # TOS桶名称
prefix = "pic_search_1000_images/" # 前缀,即你的图片文件夹名称加一个斜杠，如果上传的是图片本身而不是文件夹，则prefix为空即可prefix = ""
output_file = "best_practice_pic_search_1000_image.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 = {
                    "image_id": str(image_id),
                    "animal_image": 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))

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

{"image_id": "1", "animal_image": "tos://best-practice-pic-search-tos/pic_search_1000_images/0009fc27d9.jpg"}

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

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

什么是字段？

字段定义：字段（fields）是数据表中的基本单位，每个字段存储特定类型的数据，比如“姓名”字段存储人的名字，“年龄”字段存储数字信息。
字段名：下方代码的字段名（"field_name"）中，"image_id"字段用于储存图片的id，"animal_image"字段就储存图片的tos地址。
字段类型：不同字段可以配置不同的字段类型（"field_type"），比如"image_id"字段的类型是"int64"，也就是64 位整数**，**"animal_image"字段的类型是"image"，也就是他存入的是图片（TOS地址）。

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

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

设置数据集名称为”best_practice_pic_search“；选择使用场景为“图像”，选择向量化模型为“Doubao-embedding-vision250615”；选择向量维度为2048维；

配置向量文本字段名称为other_text，图片向量化字段名称为animal_image；其他字段名为image_id，类型int64；选择“从 string/int64 字段选择”，选择image_id；最后点击“创建数据集”；

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

此处代码用于配置字段，然后在VikingDB中创建数据集（Collection），前面的图片tos地址就会被存在这个即将被创建的数据集中。

什么是字段？

字段定义：字段（fields）是数据表中的基本单位，每个字段存储特定类型的数据，比如“姓名”字段存储人的名字，“年龄”字段存储数字信息。
字段名：下方代码的字段名（"field_name"）中，"image_id"字段用于储存图片的id，"animal_image"字段就储存图片的tos地址。
字段类型：不同字段可以配置不同的字段类型（"field_type"），比如"image_id"字段的类型是"int64"，也就是64 位整数**，**"animal_image"字段的类型是"image"，也就是他存入的是图片（TOS地址）。

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

import json
import sys
import requests # pip3 install --upgrade requests
from volcengine.auth.SignerV4 import SignerV4 # pip3 install --upgrade volcengine 
from volcengine.base.Request import Request # pip3 install --upgrade volcengine 
from volcengine.Credentials import Credentials # pip3 install --upgrade volcengine 



AK = "*" # 双引号内填入您的Access Key ID（简称为AK）
SK ="*" # 双引号内填入您的Secret Access Key（简称为SK）
DOMAIN = "api-vikingdb.volces.com"


# 签名生成函数
def prepare_request(method, path, ak, sk, params=None, data=None, doseq=0):
    if params:
        for key in params:
            if (
                    type(params[key]) == int
                    or type(params[key]) == float
                    or type(params[key]) == bool
            ):
                params[key] = str(params[key])
            elif sys.version_info[0] != 3:
                if type(params[key]) == unicode:
                    params[key] = params[key].encode("utf-8")
            elif type(params[key]) == list:
                if not doseq:
                    params[key] = ",".join(params[key])
    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": "api-vikingdb.volces.com",
    }
    r.set_headers(mheaders)
    if params:
        r.set_query(params)
    r.set_host("api-vikingdb.volces.com")
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data))
    # 生成签名
    credentials = Credentials(ak, sk, "air", "cn-north-1")
    SignerV4.sign(r, credentials)
    return r


# 创建 Collection
request_params = {
    "collection_name": "best_practice_pic_search", # 数据集名称
    "description": "用于展示VikingDB的文搜图、图搜图能力", # 数据集描述
    "primary_key": "image_id", # 声明"id"这个字段是主键
    "fields": [ # 定义声明各个字段
        {
            "field_name": "image_id",
            "field_type": "int64",
        },
        {
            "field_name": "animal_image",
            "field_type": "image",
        },
        {
            "field_name": "other_text",
            "field_type": "text",
        }
    ],
    "vectorize": [ # 需要向量化的字段
        {
            "dense": {
                "image_field": "animal_image",
                "text_field": "other_text",
                "model_name": "doubao-embedding-vision",
                "dim": 2048,
                "model_version": "250615",
                
            }
        }
    ]
}


# 调用 prepare_request 生成请求并完成签名
info_collection_req = prepare_request(
    method="POST",
    path="/api/collection/create",
    ak=AK,
    sk=SK,
    data=request_params,
)
# 完成请求
r = requests.request(
    method=info_collection_req.method,
    url="https://{}{}".format(DOMAIN, info_collection_req.path),
    headers=info_collection_req.headers,
    params=info_collection_req.query,
    data=info_collection_req.body,
)


# 打印请求结果
print(r.text)

数据集创建成功后，VikingDB网页端数据集页面就可以看见该数据集。

2.5 导入图片TOS地址到VikingDB的数据集

该步骤只能使用代码完成。
此处代码将best_practice_pic_search_1000_image.jsonl中的图片tos地址上传给best_practice_pic_search数据集中的"animal_image"字段。
TOS地址传入VikingDB的同时，其对应的图片会被自动向量化（Embedding），这样后面就可以进行向量相似度检索。这里向量化的速率是有一定限制的，每秒钟15张图。
需要您在代码开头填入您的ak和sk。

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



AK = "*" # 双引号内填入您的Access Key ID（简称为AK）
SK = "*" # 双引号内填入您的Secret Access Key（简称为SK）
DOMAIN = "api-vikingdb.volces.com"
COLLECTION_NAME = "best_practice_pic_search"
FILE_PATH = "best_practice_pic_search_1000_image.jsonl"


# 签名生成函数
def prepare_request(method, path, ak, sk, params=None, data=None, doseq=0):
    if params:
        for key in params:
            if (
                    type(params[key]) == int
                    or type(params[key]) == float
                    or type(params[key]) == bool
            ):
                params[key] = str(params[key])
            elif sys.version_info[0] != 3:
                if type(params[key]) == unicode:
                    params[key] = params[key].encode("utf-8")
            elif type(params[key]) == list:
                if not doseq:
                    params[key] = ",".join(params[key])
    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": "api-vikingdb.volces.com",
    }
    r.set_headers(mheaders)
    if params:
        r.set_query(params)
    r.set_host("api-vikingdb.volces.com")
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data))
    # 生成签名
    credentials = Credentials(ak, sk, "air", "cn-north-1")
    SignerV4.sign(r, credentials)
    return r


def send_batch(batch_data, collection_name, ak, sk, domain):
    """
    上传一批数据到指定的 Collection
    """
    # 构造请求参数
    request_params = {
        "collection_name": collection_name,
        "fields": batch_data
    }


    # 生成带签名的请求
    req_data = prepare_request(
        method="POST",
        path="/api/collection/upsert_data",
        ak=ak,
        sk=sk,
        data=request_params,
    )


    # 拼接完整 URL，发送请求
    url = f"https://{domain}{req_data.path}"
    response = requests.request(
        method=req_data.method,
        url=url,
        headers=req_data.headers,
        data=req_data.body,
    )


    try:
        resp_json = response.json()
    except Exception as e:
        # 无法解析为 JSON 时，输出错误信息
        tqdm.write("上传该批数据的响应结果：无法解析响应为 JSON: " + response.text)
        return


    # 仅当 message 不等于 "success" 时输出响应结果
    if resp_json.get("message") != "success":
        tqdm.write("上传该批数据的响应结果：" + response.text)


def batch_upload(file_path, collection_name, ak, sk, domain, batch_size):
    """
    从指定的 JSONL 文件中读取数据，每 batch_size 条数据上传一次。
    文件中每一行格式类似：
        {"image_id": "1", "animal_image": "tos://bucket/path/to/image.jpg"}
    """
    batch = []
    # 统计总行数，用于 tqdm 显示总进度
    with open(file_path, "r", encoding="utf-8") as f:
        total_lines = sum(1 for _ in f)


    with open(file_path, "r", encoding="utf-8") as f:
        for line in tqdm(f, total=total_lines, desc="上传进度"):
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
                # 如果 other_text 字段不存在或为空，赋一个单空格作为默认值
                if not record.get("other_text"):
                    record["other_text"] = " "
                batch.append(record)
            except Exception as e:
                tqdm.write(f"解析该行数据失败: {line}\n错误信息: {e}")
                continue


            # 当累计到 batch_size 条数据时，进行一次上传
            if len(batch) >= batch_size:
                send_batch(batch, collection_name, ak, sk, domain)
                batch = []  # 重置 batch


    # 处理最后不足 batch_size 条的数据
    if batch:
        send_batch(batch, collection_name, ak, sk, domain)


if __name__ == "__main__":
    batch_upload(FILE_PATH, COLLECTION_NAME, AK, SK, DOMAIN, batch_size=5)

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

2.6 创建索引

网页操作和Python操作二选一即可，效果相同。
此处为刚才的数据创建一个名为best_practice_pic_search_index的索引。
索引是对数据中某些字段的快速查找结构，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。本案例中使用2个CPU配额。
description：对索引的描述。

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

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

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

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

此处代码为刚才的数据创建一个名为best_practice_pic_search_index的索引。索引是对数据中某些字段的快速查找结构，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 json
import sys
import requests # pip3 install --upgrade requests
from volcengine.auth.SignerV4 import SignerV4 # pip3 install --upgrade volcengine 
from volcengine.base.Request import Request # pip3 install --upgrade volcengine 
from volcengine.Credentials import Credentials # pip3 install --upgrade volcengine 



AK = "*" # 双引号内填入您的Access Key ID（简称为AK）
SK = "*" # 双引号内填入您的Secret Access Key（简称为SK）
DOMAIN = "api-vikingdb.volces.com"
COLLECTION_NAME = "best_practice_pic_search"

# 签名生成函数
def prepare_request(method, path, ak, sk, params=None, data=None, doseq=0):
    if params:
        for key in params:
            if (
                    type(params[key]) == int
                    or type(params[key]) == float
                    or type(params[key]) == bool
            ):
                params[key] = str(params[key])
            elif sys.version_info[0] != 3:
                if type(params[key]) == unicode:
                    params[key] = params[key].encode("utf-8")
            elif type(params[key]) == list:
                if not doseq:
                    params[key] = ",".join(params[key])
    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": "api-vikingdb.volces.com",
    }
    r.set_headers(mheaders)
    if params:
        r.set_query(params)
    r.set_host("api-vikingdb.volces.com")
    r.set_path(path)
    if data is not None:
        r.set_body(json.dumps(data))
    # 生成签名
    credentials = Credentials(ak, sk, "air", "cn-north-1")
    SignerV4.sign(r, credentials)
    return r


# 创建 Index
request_params = {
    "collection_name": "best_practice_pic_search",
    "index_name": "best_practice_pic_search_index",
    "cpu_quota":1,
    "description": "用于展示VikingDB的图搜图、文搜图能力",
    "vector_index": {
        "index_type": "hnsw",
        "distance": "cosine",
        "quant": "int8",
    },
}

# 调用 prepare_request 生成请求并完成签名
info_req = prepare_request(
    method="POST",
    path="/api/index/create",
    ak=AK,
    sk=SK,
    data=request_params,
)
# 完成请求
r = requests.request(
    method=info_req.method,
    url="https://{}{}".format(DOMAIN, info_req.path),
    headers=info_req.headers,
    data=info_req.body,
)

# 打印请求结果
print(r.text)

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

2.7 以图搜图&以文搜图

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

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

输入文本，查询图片。

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

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

# 文搜图
import json
import sys
import os
import requests # pip install requests
import tos # pip3 install --upgrade tos
from pprint import pprint
from volcengine.auth.SignerV4 import SignerV4 # pip3 install --upgrade volcengine
from volcengine.base.Request import Request # pip3 install --upgrade volcengine
from volcengine.Credentials import Credentials # pip3 install --upgrade volcengine

# 签名相关配置
AK = "*" # 输入您的AK，注意保留双引号
SK = "*" # 输入您的SK，注意保留双引号
DOMAIN = "api-vikingdb.volces.com"

# 检索参数配置
Collection_name = "best_practice_pic_search"
Index_name = "best_practice_pic_search_index"
Query ="熊猫"

# TOS 配置（请根据实际填写）
TOS_ENDPOINT = "tos-cn-beijing.volces.com"
TOS_REGION = "cn-beijing"
TOS_BUCKET = "best-practice-pic-search-tos"

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


# 签名请求生成函数
def prepare_request(method, path, ak, sk, params=None, data=None, doseq=0):
    if params:
        for key in params:
            if isinstance(params[key], (int, float, bool)):
                params[key] = str(params[key])
            elif sys.version_info[0] != 3 and isinstance(params[key], unicode):
                params[key] = params[key].encode("utf-8")
            elif isinstance(params[key], list) and not doseq:
                params[key] = ",".join(params[key])
    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, "air", "cn-north-1")
    SignerV4.sign(r, credentials)
    return r

# 检索请求参数
request_data = {
    "collection_name": "best_practice_pic_search",
    "index_name": "best_practice_pic_search_index",
    "search": {
        "order_by_raw": {
            "text": "熊猫"
        },
        "limit": 10
    }
}

# 发起检索请求
search_req = prepare_request(
    method="POST",
    path="/api/index/search",
    ak=AK,
    sk=SK,
    data=request_data
)

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

# 解析结果
result = json.loads(response.text)
pprint(result)

# 下载图片函数
def download_tos_image(tos_uri, save_path):
    object_key = tos_uri.replace("tos://{}/".format(TOS_BUCKET), "")
    client = tos.TosClientV2(AK, SK, TOS_ENDPOINT, TOS_REGION)
    try:
        client.get_object_to_file(TOS_BUCKET, object_key, save_path)
        print(f"Downloaded: {save_path}")
    except Exception as e:
        print(f"Failed to download {tos_uri}: {e}")

# 提取并下载图像
if result.get("code") == 0:
    data = result.get("data", [[]])[0]
    for item in data:
        tos_url = item.get("fields", {}).get("animal_image")
        if tos_url and tos_url.startswith("tos://"):
            filename = tos_url.split("/")[-1]
            local_path = os.path.join(SAVE_DIR, filename)
            download_tos_image(tos_url, local_path)
else:
    print("Search failed:", result.get("message"))

如下图所示，检索到的top10图片保存在了当前目录下的download_images文件夹中。

【以图搜图】操作说明：
请在代码开头的 Query_Image_Path 位置填写您希望用于检索的图片路径。
运行代码后，系统会自动将检索到的 Top10 图片下载到当前目录下的 download_images 文件夹中。

import json
import sys
import os
import base64
import requests        # pip install requests
import tos             # pip3 install --upgrade tos
from pprint import pprint
from volcengine.auth.SignerV4 import SignerV4        # pip3 install --upgrade volcengine
from volcengine.base.Request import Request          # pip3 install --upgrade volcengine
from volcengine.Credentials import Credentials       # pip3 install --upgrade volcengine


# ===== 配置项 =====
AK = "*"  # 输入您的AK，注意保留双引号
SK = "*"  # 输入您的SK，注意保留双引号
Query_Image_Path = "*"  # 本地用于检索的图片路径
DOMAIN = "api-vikingdb.volces.com"  # VikingDB 的 API 域名

COLLECTION_NAME = "best_practice_pic_search"  # 检索的 Collection 名称
INDEX_NAME = "best_practice_pic_search_index"  # 检索的向量索引名称

TOS_ENDPOINT = "tos-cn-beijing.volces.com"  # TOS 的 endpoint（北京区域）
TOS_REGION = "cn-beijing"  # TOS 的 region（北京区域）
TOS_BUCKET = "best-practice-pic-search-tos"  # 存储图片的 TOS bucket 名称

SAVE_DIR = "downloaded_images"  # 检索结果图片下载保存的本地目录




os.makedirs(SAVE_DIR, exist_ok=True)


# ===== 读取本地图片并转为 Base64 =====
def encode_image_to_base64(img_path):
    with open(img_path, "rb") as image_file:
        encoded_string = base64.b64encode(image_file.read()).decode("utf-8")
        return f"base64://{encoded_string}"


# ===== 签名请求生成函数 =====
def prepare_request(method, path, ak, sk, params=None, data=None, doseq=0):
    if params:
        for key in params:
            if isinstance(params[key], (int, float, bool)):
                params[key] = str(params[key])
            elif sys.version_info[0] != 3 and isinstance(params[key], unicode):
                params[key] = params[key].encode("utf-8")
            elif isinstance(params[key], list) and not doseq:
                params[key] = ",".join(params[key])
    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, "air", "cn-north-1")
    SignerV4.sign(r, credentials)
    return r


# ===== 下载图片函数 =====
def download_tos_image(tos_uri, save_path):
    object_key = tos_uri.replace(f"tos://{TOS_BUCKET}/", "")
    client = tos.TosClientV2(AK, SK, TOS_ENDPOINT, TOS_REGION)
    try:
        client.get_object_to_file(TOS_BUCKET, object_key, save_path)
        print(f"Downloaded: {save_path}")
    except Exception as e:
        print(f"Failed to download {tos_uri}: {e}")


# ===== 主函数：图搜图 + 下载 =====
def search_by_image(img_path):
    base64_img = encode_image_to_base64(img_path)

    request_data = {
        "collection_name": COLLECTION_NAME,
        "index_name": INDEX_NAME,
        "search": {
            "order_by_raw": {
                "image": base64_img
            },
            "limit": 10
        }
    }

    search_req = prepare_request(
        method="POST",
        path="/api/index/search",
        ak=AK,
        sk=SK,
        data=request_data
    )

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

    result = json.loads(response.text)
    pprint(result)

    # 下载检索到的图片
    if result.get("code") == 0:
        data = result.get("data", [[]])[0]
        for item in data:
            tos_url = item.get("fields", {}).get("animal_image")
            if tos_url and tos_url.startswith("tos://"):
                filename = tos_url.split("/")[-1]
                local_path = os.path.join(SAVE_DIR, filename)
                download_tos_image(tos_url, local_path)
    else:
        print("Search failed:", result.get("message"))


# ===== 示例调用 =====
if __name__ == "__main__":
    image_path = Query_Image_Path
    search_by_image(image_path)

如下图所示，检索到的top10图片保存在了当前目录下的download_images文件夹中。

注意，如果删除了TOS桶中的图片或者TOS桶本身，那么进行图片检索时则无法保存结果图片到本地。

示例修正

依赖安装注释修正为 # pip3 install --upgrade tos，与后续 import tos 一致。

最近更新时间：2026.09.08 15:30:03

这个页面对您有帮助吗？

有用

无用

向量数据库VikingDB

1.1 需求场景 #

1.2 场景示例 #

1.3 方案思路 #

2.1 案例场景与数据 #

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

2.3 数据准备 #

2.3.1 创建TOS桶 #

2.3.2 上传图片文件夹 #

2.3.3 导出TOS中图片的路径 #

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

2.5 导入图片TOS地址到VikingDB的数据集 #

2.6 创建索引 #

2.7 以图搜图&以文搜图 #

示例修正 #

1.1 需求场景

1.2 场景示例

1.3 方案思路

2.1 案例场景与数据

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

2.3 数据准备

2.3.1 创建TOS桶

2.3.2 上传图片文件夹

2.3.3 导出TOS中图片的路径

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

2.5 导入图片TOS地址到VikingDB的数据集

2.6 创建索引

2.7 以图搜图&以文搜图

示例修正