知识库支持通过 API 接口进行操作,在使用 API 接口之前,需要先完成以下操作。
已通过 注册账号及开通服务 页面操作完成注册账号及开通服务。
在调用知识库的各个能力之前,确保您已生成访问密钥 Access Key。Access Key 包括 Access Key ID(简称为 AK) 和 Access Key Secret(简称为 SK),其中,Access Key ID 用于标识用户,Access Key Secret 用于验证用户的密钥,请您妥善保管。
AK/SK 密钥获取方式如下,更多详情请参考 Access Key(密钥)管理。
注意
安全起见,建议新建子账户,并使用子账户的 AK/SK。
获取到 AK/SK 后,生成签名。签名过程说明详情请参考 签名过程Demo,通过源码生成签名详情请参考 签名源码示例,通过 SDK 生成签名详情请参考 SDK概览。
安装火山引擎程序包。
pip install volcengine
生成签名。
import json import sys import requests from volcengine.auth.SignerV4 import SignerV4 from volcengine.base.Request import Request from volcengine.Credentials import Credentials 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: # 在 Python 3.x 中,不需要这部分代码 # 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", } r.set_headers(mheaders) if params: r.set_query(params) 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
安装火山引擎程序包。
go get -u github.com/volcengine/volc-sdk-golang
生成签名。
package main import ( "bytes" "fmt" "net/http" "net/url" "strings" "io/ioutil" "github.com/volcengine/volc-sdk-golang/base" ) const ( testAk = "***" testSk = "***" ) func PrepareRequest(method string, path string, ak string, sk string, query url.Values, body []byte) *http.Request { u := url.URL{ Scheme: "https", Host: "api-knowledgebase.mlp.cn-beijing.volces.com", Path: "/api/knowledge/collection/search", } if query != nil { u.RawQuery = query.Encode() } req, _ := http.NewRequest(strings.ToUpper(method), u.String(), bytes.NewReader(body)) req.Header.Add("Accept", "application/json") req.Header.Add("Content-Type", "application/json") req.Header.Add("Host", "api-knowledgebase.mlp.cn-beijing.volces.com") credential := base.Credentials{ AccessKeyID: ak, SecretAccessKey: sk, Service: "air", Region: "cn-north-1", } req = credential.Sign(req) return req }
安装火山引擎程序包。
<dependency> <groupId>com.volcengine</groupId> <artifactId>volc-sdk-java</artifactId> <version>最新版本</version> </dependency>
生成签名
package javaTest; import java.util.List; import org.apache.http.HttpResponse; import org.apache.http.NameValuePair; import org.apache.http.client.HttpClient; import org.apache.http.client.config.RequestConfig; import org.apache.http.client.utils.URIBuilder; import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.HttpClients; import org.apache.http.util.EntityUtils; import com.volcengine.auth.ISignerV4; import com.volcengine.auth.impl.SignerV4Impl; import com.volcengine.model.Credentials; import com.volcengine.service.SignableRequest; public class sign { public static SignableRequest prepareRequest(String host, String path, String method, List<NameValuePair> params, String body, String ak, String sk) throws Exception { SignableRequest request = new SignableRequest(); request.setMethod(method); request.setHeader("Accept", "application/json"); request.setHeader("Content-Type", "application/json"); request.setHeader("Host", "api-knowledgebase.mlp.cn-beijing.volces.com"); request.setEntity(new StringEntity(body, "utf-8")); URIBuilder builder = request.getUriBuilder(); builder.setScheme("https"); builder.setHost(host); builder.setPath(path); if (params != null) { builder.setParameters(params); } RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(5000).setConnectTimeout(2000).build(); request.setConfig(requestConfig); Credentials credentials = new Credentials("cn-north-1", "air"); credentials.setAccessKeyID(ak); credentials.setSecretAccessKey(sk); // 签名 ISignerV4 ISigner = new SignerV4Impl(); ISigner.sign(request, credentials); return request; }
pnpm add @volcengine/openapi
import { Signer } from '@volcengine/openapi'; import type { RequestObj } from '@volcengine/openapi/lib/base/types'; interface Options { pathname: string method: 'GET' | 'POST' body?: string // json 字符串,当且仅当 post 请求 region: 'cn-beijing' | 'cn-shanghai' params?: Record<string, any> // 当且仅当 get 请求 } function signer({ pathname, method, body, region, params }: Options) { const requestObj: RequestObj = { region, headers: { Accept: 'application/json', 'Content-type': 'application/json' }, method, body, pathname, params, } const signer = new Signer(requestObj, 'air') signer.addAuthorization({ accessKeyId: 'ak', // 替换为用户 ak secretKey: 'sk', // 替换为用户 sk }) return requestObj.headers }
调用函数 signer
生成请求体 header 后,将生成好的 header 注入到用户请求函数中。
获取好AK/SK 密钥后,可以将签名生成函数和创建知识库逻辑写进同一个脚本文件中。这里给出将签名生成逻辑与具体接口调用对接的一般实践过程,以python代码为例:
import json import requests from volcengine.auth.SignerV4 import SignerV4 from volcengine.base.Request import Request from volcengine.Credentials import Credentials #签名生成逻辑 def prepare_request(method, path, params=None, data=None, doseq=0): #创建request if params: for key in params: if ( isinstance(params[key], int) or isinstance(params[key], float) or isinstance(params[key], bool) ): params[key] = str(params[key]) elif isinstance(params[key], list): if not doseq: params[key] = ",".join(params[key]) r = Request() r.set_shema("http") r.set_method(method) r.set_connection_timeout(10) r.set_socket_timeout(10) mheaders = { "Accept": "application/json", "Content-Type": "application/json", "Host": g_knowledge_base_domain, "V-Account-Id": account_id, } r.set_headers(mheaders) if params: r.set_query(params) r.set_host(g_knowledge_base_domain) 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 create(): #创建知识库的请求方式 method = "POST" #创建知识库的请求路径 path = "/api/knowledge/collection/create" #创建知识库所需要的一些参数 request_params = { "name": "lzm_test2", "data_type": "unstructured_data", "preprocessing": { "chunking_strategy":"custom_balance", "multi_modal":["image_ocr"] }, "index": { "cpu_quota": 1, "embedding_model": "doubao-embedding-and-m3", "embedding_dimension": 2048, "quant": "int8", "index_type": "hnsw_hybrid" }, } info_req = prepare_request(method=method, path=path, data=request_params) rsp = requests.request( method=info_req.method, url="https://{}{}".format(g_knowledge_base_domain, info_req.path), headers=info_req.headers, data=info_req.body ) print(rsp.text) if __name__ == "__main__": ak = '刚才申请到的ak' sk = '刚才申请到的sk' g_knowledge_base_domain = "api-knowledgebase.mlp.cn-beijing.volces.com" account_id = "你自己的火山账号id" #调用创建知识库的函数 create()
其中: