完成“对接指南”页面的注册账号、实名认证、AK/SK 密钥获取和签名获取后,可调用 API 接口实现数据集(Collection)创建、数据集(Collection)数据写入、索引(Index)创建和检索查询等功能。
如下提供一个向量数据里 VikingDB 使用 Python 语言调用 API 接口的完整请求示例。其中,请求域名为 https://api-vikingdb.volces.com (华北)、 https://api-vikingdb.mlp.cn-shanghai.volces.com(华东)、 https://api-vikingdb.mlp.ap-mya.byteplus.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="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)
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="https://{}{}".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="https://{}{}".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="https://{}{}".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="https://{}{}".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="https://{}{}".format(DOMAIN, info_req.path), headers=info_req.headers, data=info_req.body, ) # 打印请求结果 print(r.text)