API鉴权机制说明--全域AI搜索-火山引擎

文档中心

导航

API鉴权机制说明

最近更新时间：2025.08.22 23:57:02首次发布时间：2025.04.24 20:27:18

鉴权机制说明

AI搜索数据面API（包括搜索、对话、推荐等服务API和数据的上传、删除、查询API）均支持2种鉴权方式：
API Key鉴权：简单方便，可直接在AI 搜索控制台配置使用。
Access Key鉴权：火山云产品体系的传统云资源访问鉴权方式，面向企业精细化资源和权限管控。

使用API Key鉴权

创建并获取API Key

签名构造

API Key 签名鉴权方式要求在 HTTP 请求 header 中按如下方式添加 Authorization header:

Authorization: Bearer <API_KEY>

API 调用示例

curl -X POST 'https://aisearch.cn-beijing.volces.com/api/v1/application/${application_id}/search' \ 
  -H 'Content-Type: application/json' \ 
  -H 'Authorization:Bearer <API_KEY>' \ 
  -d '{ 
    "query": { 
        "text": "科幻片" 
    }, 
    "page_number": 1, 
    "page_size": 5, 
    "user": { 
        "_user_id": "ldy_199432" 
    }, 
    "filter": {
        "op": "must",
        "field": "director",
        "conds": ["郭帆"]
    },
    "dataset_id": "106265704" 
}'

使用火山Access Key鉴权

获取Access Key

Access Key 包括 Access Key ID（简称为 AK）和 Access Key Secret（简称为 SK），其中，Access Key ID 用于标识用户，Access Key Secret 用于验证用户的密钥，请您妥善保管。

说明

如果您使用子账号操作，请与您的企业管理员确认您的子账号拥有Access Key的创建和读取权限。

AK/SK 密钥获取方式如下，详情请参考 Access Key（密钥）管理。

单击右上角账号名下拉框中的【API 访问密钥】进入对应页面。

单击【新建密钥】按钮，可获取 AK/SK，可以此为凭证调用上述已接入应用的接口。

注意：安全起见，建议新建子账户，并使用子账户的 AK/SK。

构造签名

使用 Access Key 构造签名，签名方法和签名构造详细机制请参见签名方法；推荐使用火山引擎提供的多语言官方SDK来进行构造签名。
- 签名用到的AI搜索平台相关参数

Service：aisearch
Region：cn-beijing

在 HTTP 请求 header 中添加 Authorization 为构造的签名字符串。

curl -X POST 'https://aisearch.cn-beijing.volces.com/api/v1/{api action}'
  -H 'Content-Type: application/json' \
  -H 'Authorization: HMAC-SHA256 Credential={AccessKeyId}/{CredentialScope}, SignedHeaders={SignedHeaders}, Signature={Signature}' \
  -d '{
  "key": "value"
}'

API 调用示例

安装火山引擎程序包

Python

Java

pip install volcengine

生成签名并测试调用

Python

Java

import json
import sys
import requests

from volcengine.auth.SignerV4 import SignerV4
from volcengine.base.Request import Request
from volcengine.Credentials import Credentials

Host = "aisearch.cn-beijing.volces.com"
Schema = "https"
Service = "aisearch"
Region = "cn-beijing"

# 用户的AK/SK
AK = "xxx"
SK = "xxx"


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(Schema)
    r.set_method(method)
    r.set_connection_timeout(10)
    r.set_socket_timeout(10)
    mheaders = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Host": Host
    }
    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, Service, Region)
    SignerV4.sign(r, credentials)
    return r


def search_demo():
    path = "/api/v1/application/{}/search".format("应用id")
    url_param = None
    body_data = {
        "query": {
            "text": "测试"
        },
        "user": {
            "_user_id": "mock_user_id",
        },
        "dataset_id": "106621109",
        "page_number": 1,
        "page_size": 10
    }

    sign_req = prepare_request("POST", path, AK, SK, url_param, body_data)

    resp = requests.request(
        method=sign_req.method,
        url="{}://{}{}".format(Schema, Host, path),
        headers=sign_req.headers,
        data=sign_req.body
    )
    resp.raise_for_status()

    print(resp.text)


def chat_search_demo():
    path = "/api/v1/application/{}/chat_search".format("应用id")
    url_param = None
    body_data = {
        "input_message": {
            "content": [
                {
                    "type": "text",
                    "text": "帮我推荐几部喜剧?"
                }
            ]
        },
        "user": {
            "_user_id": "mock_user_id",
        },
        "session_id": "mock_session_id",
    }

    sign_req = prepare_request("POST", path, AK, SK, url_param, body_data)

    resp = requests.request(
        method=sign_req.method,
        url="{}://{}{}".format(Schema, Host, path),
        headers=sign_req.headers,
        data=sign_req.body,
        stream=True,  # 注意需要开启流式
    )
    resp.raise_for_status()
    process_stream_response(resp)


def process_stream_response(response):
    for line in response.iter_lines(decode_unicode=True):
        if line:
            print(line)


if __name__ == "__main__":
    search_demo()
    chat_search_demo()

package javaTest;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.*;

import com.google.gson.Gson;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
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.methods.HttpPost;
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 Example {
    private static final String AK = "xxx";
    private static final String SK = "xxx";
    private static final String HOST = "aisearch.cn-beijing.volces.com";
    private static final String SERVICE = "aisearch";
    private static final String REGION = "cn-beijing";
    private static final String SCHEMA = "https";


    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", HOST);
        request.setEntity(new StringEntity(body, "utf-8"));

        URIBuilder builder = request.getUriBuilder();
        builder.setScheme(SCHEMA);
        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(REGION, SERVICE);
        credentials.setAccessKeyID(ak);
        credentials.setSecretAccessKey(sk);

        // 签名
        ISignerV4 ISigner = new SignerV4Impl();
        ISigner.sign(request, credentials);

        return request;
    }

    public static void search_demo() {
        try {
            String path = String.format("/api/v1/application/%s/search", "应用id");

            Map<String, Object> bodyData = new HashMap<>();
            bodyData.put("dataset_id", "实际数据集id");
            bodyData.put("page_number", 1);
            bodyData.put("page_size", 10);
            
            Map<String, String> query = new HashMap<>();
            query.put("text", "测试");
            bodyData.put("query", query);
            
            Map<String, String> user = new HashMap<>();
            user.put("_user_id", "mock_user_id");
            bodyData.put("user", user);

            Gson gson = new Gson();
            String body = gson.toJson(bodyData);

            SignableRequest signableRequest = prepareRequest(HOST, path, "POST", null, body, AK, SK);
            URI uri = new URIBuilder().setScheme(SCHEMA).setHost(HOST).setPath(path).build();

            HttpPost httpPost = new HttpPost(uri);
            httpPost.setConfig(signableRequest.getConfig());
            httpPost.setEntity(signableRequest.getEntity());
            for (Header header : signableRequest.getAllHeaders()) {
                httpPost.setHeader(header.getName(), header.getValue());
            }
            HttpClient httpClient = HttpClients.createDefault();
            HttpResponse response = httpClient.execute(httpPost);
            int statusCode = response.getStatusLine().getStatusCode();
            String responseBody = EntityUtils.toString(response.getEntity());

            System.out.println(statusCode);
            System.out.println(responseBody);

        } catch (IOException e) {
            System.out.println("请求出错: " + e.getMessage());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void chat_search_demo() {
        // 注意chat_search为流式接口
        try {
            String path = String.format("/api/v1/application/%s/chat_search", "应用id");

            Map<String, Object> bodyData = new HashMap<>();
            bodyData.put("session_id", "mock_session_id");

            Map<String, Object> inputMessage = new HashMap<>();
            Map<String, String> content = new HashMap<>();
            content.put("type", "text");
            content.put("text", "帮我推荐几部喜剧?");
            inputMessage.put("content", Collections.singletonList(content));
            bodyData.put("input_message", inputMessage);

            Map<String, String> user = new HashMap<>();
            user.put("_user_id", "mock_user_id");
            bodyData.put("user", user);

            Gson gson = new Gson();
            String body = gson.toJson(bodyData);

            SignableRequest signableRequest = prepareRequest(HOST, path, "POST", null, body, AK, SK);
            URI uri = new URIBuilder().setScheme(SCHEMA).setHost(HOST).setPath(path).build();

            HttpPost httpPost = new HttpPost(uri);
            httpPost.setConfig(signableRequest.getConfig());
            httpPost.setEntity(signableRequest.getEntity());
            for (Header header : signableRequest.getAllHeaders()) {
                httpPost.setHeader(header.getName(), header.getValue());
            }
            HttpClient httpClient = HttpClients.createDefault();
            HttpResponse response = httpClient.execute(httpPost);
            int statusCode = response.getStatusLine().getStatusCode();
            if (statusCode == 200) {
                // 注意需要走流式处理
                processStreamResponse(response.getEntity());
            } else {
                System.err.println("HTTP请求失败，状态码: " + statusCode);
                String responseBody = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
                System.err.println("响应内容: " + responseBody);
            }
        } catch (IOException e) {
            System.out.println("请求出错: " + e.getMessage());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void processStreamResponse(HttpEntity entity) throws IOException {
        try (InputStream inputStream = entity.getContent();
             BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8), 128)) {

            String line;
            // 逐行读取并打印响应内容
            while ((line = reader.readLine()) != null) {
                System.out.println(line);
            }
        } finally {
            EntityUtils.consumeQuietly(entity);
        }
    }

    public static void main(String[] args) {
        search_demo();
        chat_search_demo();
    }
}

package main

import (
    "bufio"
    "bytes"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
    "net/url"
    "strings"

    "github.com/volcengine/volc-sdk-golang/base"
)

const (
    Service = "aisearch"
    Host    = "aisearch.cn-beijing.volces.com"
    Schema  = "https"
    Region  = "cn-beijing"

    // 用户的AK/SK
    AK = "xxx"
    SK = "xxx"
)

func PrepareRequest(method string, path string, ak string, sk string, query url.Values, body []byte) *http.Request {
    u := url.URL{
       Scheme: Schema,
       Host:   Host,
       Path:   path,
    }
    if query != nil {
       u.RawQuery = query.Encode()
    }
    req, _ := http.NewRequest(strings.ToUpper(method), u.String(), bytes.NewReader(body))
    req.Header.Set("Accept", "application/json")
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("Host", Host)
    credential := base.Credentials{
       AccessKeyID:     ak,
       SecretAccessKey: sk,
       Service:         Service,
       Region:          Region,
    }
    req = credential.Sign(req)
    return req
}

func searchDemo() {
    path := fmt.Sprintf("/api/v1/application/%s/search", "应用id")
    urlParam := url.Values{}
    bodyData := map[string]interface{}{
       "query": map[string]string{
          "text": "测试",
       },
       "user": map[string]interface{}{
          "_user_id": "mock_user_id",
       },
       "dataset_id":  "实际数据集id",
       "page_number": 1,
       "page_size":   10,
    }
    body, _ := json.Marshal(bodyData)
    req := PrepareRequest("POST", path, AK, SK, urlParam, body)

    // send request
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
       fmt.Printf("读取响应内容出错: %v
", err)
       return
    }
    defer resp.Body.Close()

    respBody, err := io.ReadAll(resp.Body)
    if err != nil {
       fmt.Printf("读取响应内容出错: %v
", err)
       return
    }
    fmt.Printf("响应状态码: %d
", resp.StatusCode)
    fmt.Printf("响应内容: %s
", respBody)
}

func chatSearchDemo() {
    path := fmt.Sprintf("/api/v1/application/%s/chat_search", "应用id")
    urlParam := url.Values{}
    bodyData := map[string]interface{}{
       "input_message": map[string]interface{}{
          "content": []map[string]interface{}{
             {
                "type": "text",
                "text": "帮我推荐几部喜剧?",
             },
          },
       },
       "user": map[string]interface{}{
          "_user_id": "mock_user_id",
       },
       "session_id": "mock_session_id",
    }
    body, _ := json.Marshal(bodyData)
    req := PrepareRequest("POST", path, AK, SK, urlParam, body)

    // send request
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
       fmt.Printf("读取响应内容出错: %v", err)
       return
    }
    if resp.StatusCode != http.StatusOK {
       fmt.Printf("响应状态码异常: %v", resp.StatusCode)
       return
    }
    processStreamResponse(resp)
}

func processStreamResponse(resp *http.Response) {
    defer func() {
       _ = resp.Body.Close()
    }()

    reader := bufio.NewReader(resp.Body)
    for {
       line, _, err := reader.ReadLine()
       if err != nil {
          if errors.Is(err, io.EOF) {
             break
          }
          fmt.Printf("读取流式响应内容出错: %v", err)
       }
       fmt.Println(string(line))
    }
}

func main() {
    searchDemo()
    chatSearchDemo()
}