核心流程

完成”对接指南“页面的注册账号、实名认证、AK/SK 密钥获取和签名获取后，可调用 API 接口实现数据集（Collection）创建、数据集（Collection）数据写入、索引（Index）创建和检索查询等功能。
如下提供一个向量数据里 VikingDB 使用 Python 语言调用 API 接口的完整请求示例。其中，请求域名为 https://api-vikingdb.volces.com 或 http://api-vikingdb.volces.com。

import requests
from volc_auth import prepare_request
# volc_auth 请参考《对接指南》页面的获取签名介绍。
AK = "*"
SK = "*"
DOMAIN = "api-vikingdb.volces.com"

# 创建 Collection
request_params = {
    "collection_name": "your_collection",
    "description": "this is an example",
    # 声明主键
    "primary_key": "id",
    # 定义声明各个字段
    "fields": [
        {
            "field_name": "id",
            "field_type": "int64",
            "default_val": 0,
        },
        {"field_name": "vector_field", "field_type": "vector", "dim": 64}
        # 其他字段已省略
    ],
}

# 调用 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="http://{}{}".format(DOMAIN, info_collection_req.path),
    headers=info_collection_req.headers,
    params=info_collection_req.query,
    data=info_collection_req.body,
)

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

import random

import requests
from volc_auth import prepare_request


def gen_random_vector(dim):
    res = [
        0,
    ] * dim
    for i in range(dim):
        res[i] = random.random() - 0.5
    return res


AK = "*"
SK = "*"
DOMAIN = "api-vikingdb.volces.com"

# 写入 Collection
request_params = {
    "collection_name": "your_collection",
    "fields": [
        {
            "id": 1,
            # 向量维度与创建时指定 dim 保持一致
            "vector_field": gen_random_vector(64),
        },
        {
            "id": 2,
            # 向量维度与创建时指定 dim 保持一致
            "vector_field": gen_random_vector(64),
        },
        {
            "id": 3,
            # 向量维度与创建时指定 dim 保持一致
            "vector_field": gen_random_vector(64),
        },
    ],
}

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

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

import requests
from volc_auth import prepare_request

AK = "*"
SK = "*"
DOMAIN = "api-vikingdb.volces.com"

# 查找 Collection 数据
request_params = {"collection_name": "your_collection", "primary_keys": 1}

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

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

import requests
from volc_auth import prepare_request

AK = "*"
SK = "*"
DOMAIN = "api-vikingdb.volces.com"

# 创建 Index
request_params = {
    "collection_name": "your_collection",
    "index_name": "your_index",
    "description": "test",
    "vector_index": {
        "index_type": "hnsw",
        "distance": "ip",
        "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="http://{}{}".format(DOMAIN, info_req.path),
    headers=info_req.headers,
    data=info_req.body,
)

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

import requests
from volc_auth import prepare_request

AK = "*"
SK = "*"
DOMAIN = "api-vikingdb.volces.com"

# Index 数据查询
request_params = {
    "collection_name": "your_collection",
    "index_name": "your_index",
    "primary_keys": 1,
}

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

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

import random

import requests
from volc_auth import prepare_request


def gen_random_vector(dim):
    res = [0] * dim
    for i in range(dim):
        res[i] = random.random() + 3
    return res


AK = "*"
SK = "*"
DOMAIN = "api-vikingdb.volces.com"

# Index 检索
request_params = {
    "collection_name": "your_collection",
    "index_name": "your_index",
    "search": {
        # 使用向量进行检索
        "order_by_vector": {"vectors": [gen_random_vector(64)]},
        "limit": 2,
    },
}

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

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