API簡介

提供通用檢索、檢索后處理階段結果供客戶按需使用。

• 通用搜索結果（common_search）：將覆蓋標準網頁搜索，并支持多種卡片結果，覆蓋網頁標題、動態摘要、來源網站、發布時間等關鍵字段。

• 檢索后處理（post_retrieval）：提供對通用檢索內容的重排、精選。降低對通用搜索結果的解析、選擇成本；

  多階段結果通過SSE流式返回。

基本概念

快速開始

前置條件

• 與阿里云同學進行需求溝通，完成產品購買，并聯系阿里云同學進行開通審核。

• 開通阿里云賬號，并申請子賬號AK/SK

• 前往RAM系統，對子賬號添加權限策略：AliyunLinkedMallFullAccess

服務接入點

示例代碼

Java SDK

• SDK版本

<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>alibabacloud-linkedmallretrieval20240501</artifactId>
    <version>2.0.0</version>
</dependency>

• 調用示例

package com.aliyun.linkedretrieval.example;

import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.core.http.HttpClient;
import com.aliyun.httpcomponent.httpclient.ApacheAsyncHttpClientBuilder;
import com.aliyun.sdk.gateway.pop.Configuration;
import com.aliyun.sdk.gateway.pop.auth.SignatureVersion;
import com.aliyun.sdk.gateway.pop.exception.PopClientException;
import com.aliyun.sdk.service.linkedmallretrieval20240501.AsyncClient;
import com.aliyun.sdk.service.linkedmallretrieval20240501.models.AISearchV2Request;
import com.aliyun.sdk.service.linkedmallretrieval20240501.models.AISearchV2ResponseBody;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import darabonba.core.ResponseIterable;
import darabonba.core.ResponseIterator;
import darabonba.core.client.ClientOverrideConfiguration;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.time.Duration;
import java.util.UUID;

public class AISearchSSEInvoker {

    private AsyncClient     asyncClient;

    @Before
    public void setup(){
        StaticCredentialProvider normalProvider = StaticCredentialProvider.create(
            Credential.builder()
                .accessKeyId(System.getenv("ACCESS_KEY"))
                .accessKeySecret(System.getenv("ACCESS_SECRET"))
                .build()
        );
        asyncClient = getAsyncClient(normalProvider);
    }

    @After
    public void cleanup(){
        if(asyncClient != null){
            asyncClient.close();
        }
    }


    @Test
    public void testSuccessQuery(){
        String sessionId = UUID.randomUUID().toString();
        invokeSearch(
            asyncClient,
            "無錫拈花灣旅游攻略",
            sessionId);
        invokeSearch(
            asyncClient,
            "歷年高考分數線",
            sessionId);

    }

    @Test
    public void testBizErrorEvent(){
        // query長度超過100，報on_error_event
        invokeSearch(
            asyncClient,
            "從9月11日召開的2023北京數字交通大會獲悉，目前我國超過多少公里公路完成智能化升級改造，京雄高速河北段、滬杭甬高速、杭州繞城 復線、成宜高速等一批智慧公路已建成運行。本次大會由中國交通運輸協會等單位主辦，以“數字·新時代、交通·新未來”為主題？",
            UUID.randomUUID().toString());
    }

    @Test
    public void testPopClientException(){
        StaticCredentialProvider noPermissionProvider = StaticCredentialProvider.create(
            Credential.builder()
                .accessKeyId(System.getenv("ACCESS_KEY_NO_PERMISSION"))
                .accessKeySecret(System.getenv("ACCESS_SECRET_NO_PERMISSION"))
                .build());
        AsyncClient asyncClient = getAsyncClient(noPermissionProvider);

        // 使用無權限的子賬號AK/SK請求，報錯
        invokeSearch(asyncClient, "無錫拈花灣旅游攻略", UUID.randomUUID().toString());
    }

    /**
     * 提供阿里云子賬號AK/SK進行的認證信息，建議您使用子賬號AK/SK，并且需要對：AliyunLinkedMallFullAccess 進行授權
     * AsyncClient支持跨請求復用，如果不復用，需要在使用后調用close進行關閉
     */
    private static AsyncClient getAsyncClient(StaticCredentialProvider credentialProvider){

        //2. 構建http client；SSE類請求超時時間與連接數可適當調大
        HttpClient httpClient = new ApacheAsyncHttpClientBuilder()
            .connectionTimeout(Duration.ofSeconds(60))
            .responseTimeout(Duration.ofSeconds(60))
            .maxConnections(256)
            .maxConnectionsPerRoute(256)
            // 如果需要，可以設置proxy信息
            //.proxy()
            .maxIdleTimeOut(Duration.ofSeconds(60))
            .build();

        //3. 構建client
        AsyncClient client = AsyncClient.builder()
            .region("cn-zhangjiakou")
            .httpClient(httpClient)
            .credentialsProvider(credentialProvider)
            .serviceConfiguration(Configuration.create().setSignatureVersion(SignatureVersion.V3))
            .overrideConfiguration(
                ClientOverrideConfiguration.create().setProtocol("HTTPS")
                    .setEndpointOverride("linkedmallretrieval.cn-zhangjiakou.aliyuncs.com")
            ).build();

        return client;
    }

    private static void invokeSearch(AsyncClient client, String query, String sessionId){
        AISearchV2Request request = AISearchV2Request.builder()
            .query(query)
            .sessionId(sessionId)
            .timeRange("OneYear")
            .build();
        try{
            ResponseIterable<AISearchV2ResponseBody> responseBodies = client.aISearchV2WithResponseIterable(request);
            ResponseIterator<AISearchV2ResponseBody> bodyIterator = responseBodies.iterator();
            while(bodyIterator.hasNext()){
                AISearchV2ResponseBody event = bodyIterator.next();
                String requestId = event.getRequestId();
                String eventName = event.getHeader().getEvent();
                String payload = event.getPayload();

                switch (eventName){
                    case "on_common_search_end":
                        // 關注通用搜索結果，可以使用此結果；payload為JSON格式,詳細的格式說明可以參考文檔；
                        JsonObject commonSearchRoot = JsonParser.parseString(payload).getAsJsonObject();
                        JsonArray pageItems = commonSearchRoot.getAsJsonArray("pageItems");
                        System.out.println("[on_common_search_end] requestId:" + requestId + ", pageItems:" + pageItems.size() + ", payload:" + payload);
                        break;
                    case "on_post_retrieval_end":
                        JsonArray postRetrievalRoot = JsonParser.parseString(payload).getAsJsonArray();
                        System.out.println("[on_post_retrieval_end] requestId: " + requestId + ", documents:" + postRetrievalRoot.size());
                        break;
                    case "on_error_event":
                        System.err.println("[on_error_event]:" + requestId +", payload: " + payload);
                        break;
                    default:
                }
            }
        }catch (PopClientException pe){
            pe.printStackTrace();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

Python SDK

• SDK依賴

pip install alibabacloud-tea-openapi-sse

• 調用示例

import asyncio
import json
import uuid

from alibabacloud_tea_sse.exceptions import TeaException

# 注意：此依賴包不是SDK中提供，參見下面的代碼塊
from aliyun.linked_retrieval.base import AISearch


async def test_success():
    session_id = str(uuid.uuid4())
    # 復用請求Connection
    ai_search = AISearch()

    await invoke_search(
        ai_search,
        query="敬銀山一日游",
        time_range="OneYear",
        session_id=session_id)
    await invoke_search(
        ai_search,
        query="如何在家啟蒙孩子英文",
        time_range="NoLimit",
        session_id=session_id)

    await ai_search.async_close()


# query 超長event報錯
async def test_biz_error_event():
    ai_search = AISearch()
    session_id = str(uuid.uuid4())
    try:
        await invoke_search(
            ai_search,
            query="從9月11日召開的2023北京數字交通大會獲悉，目前我國超過多少公里公路完成智能化升級改造，京雄高速河北段、滬杭甬高速、杭州繞城 復線、成宜高速等一批智慧公路已建成運行。本次大會由中國交通運輸協會等單位主辦，以“數字·新時代、交通·新未來”為主題？",
            time_range="NoLimit",
            session_id=session_id)
    finally:
        await ai_search.async_close()


# AK/SK不存在時，會有TeaException:
# 打印：api exception, requestId:7F076157-CB45-5DEF-B84D-77D951C5BB72, code:InvalidAccessKeyId.NotFound, message:Specified access key is not found.
async def test_api_error():
    ai_search = AISearch(access_key_id="not_exist", access_key_secret="not_exist")
    try:
        session_id = str(uuid.uuid4())
        await invoke_search(
            ai_search,
            query="敬銀山一日游",
            time_range="NoLimit",
            session_id=session_id)
    finally:
        await ai_search.async_close()


async def invoke_search(ai_search: AISearch, query: str, session_id: str, time_range: str):
    try:
        async for res in await ai_search.do_sse_query(query, session_id, time_range):
            event = json.loads(res.get("event").data)
            request_id = event.get("requestId")
            header = event.get("header")
            payload = event.get("payload")

            event_name = header.get("event")
            server_cost = header.get("responseTime")
            if event_name == "on_common_search_end":
                common_search_root = json.loads(payload)
                page_items = common_search_root.get("pageItems")
                print(
                    f"[on_common_search_end] requestId:{request_id}, serverCost:{server_cost}, pageItems:{len(page_items)}, payload:{payload}")
            elif event_name == "on_post_retrieval_end":
                post_retrieval_root = json.loads(payload)
                print(
                    f"[on_post_retrieval_end] requestId:{request_id}, serverCost:{server_cost}, post_retrieval:{len(post_retrieval_root)}")
            elif event_name == "on_error_event":
                print(f"[on_error_event] requestId:{request_id}, serverCost:{server_cost}, payload:{payload}")
    except TeaException as e:
        code = e.code
        request_id = e.data.get("RequestId")
        message = e.data.get("Message")
        print(f"api exception, requestId:{request_id}, code:{code}, message:{message}")


if __name__ == "__main__":
    asyncio.run(test_success())
    # asyncio.run(test_biz_error_event())
    # asyncio.run(test_api_error())

• aliyun.linked_retrieval.base

  這部分代碼SDK不包含，可以直接使用，無需修改

import os
import ssl
from typing import Optional

import aiohttp
import certifi
from alibabacloud_tea_openapi_sse import models as open_api_models
from alibabacloud_tea_openapi_sse.client import Client as OpenApiClient
from alibabacloud_tea_util_sse import models as util_models

"""
    AISearch客戶端，支持多請求復用；
"""


class AISearch:
    def __init__(self, access_key_id: str = None, access_key_secret: str = None) -> None:
        self.endpoint = "linkedmallretrieval.cn-zhangjiakou.aliyuncs.com"
        self._api_info = self._create_api_info()
        self._runtime = util_models.RuntimeOptions(
            read_timeout=60 * 1000,
            connect_timeout=60 * 1000
        )
        # 自定義連接池 & ssl配置；
        ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        ssl_context.load_verify_locations(certifi.where())
        self._session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                limit=256,
                limit_per_host=256,
                keepalive_timeout=600,
                ssl=ssl_context
            )
        )
    
        self.access_key_id = access_key_id if access_key_id is not None else os.environ.get('ACCESS_KEY')
        self.access_key_secret = access_key_secret if access_key_secret is not None else os.environ.get(
            'ACCESS_SECRET')

        assert self.access_key_id is not None and self.access_key_secret is not None
        self._init_app()

    def _init_app(self):
        config = open_api_models.Config(
            access_key_id=self.access_key_id,
            access_key_secret=self.access_key_secret,
            session=self._session,
            endpoint=self.endpoint,
        )
        self._client = OpenApiClient(config)

    # AISearch使用完以后，記得關閉，釋放連接池
    async def async_close(self):
        await self._session.close()

    def close(self):
        self._session.close()

    def _create_api_info(self) -> open_api_models.Params:
        """
        API 相關
        @param path: params
        @return: OpenApi.Params
        """
        params = open_api_models.Params(
            # 接口名稱
            action='AISearchV2',
            # 接口版本
            version='2024-05-01',
            # 接口協議
            protocol='HTTPS',
            # 接口 HTTP 方法
            method='GET',
            auth_type='AK',
            style='ROA',
            # 接口 PATH,
            pathname='/linked-retrieval/linked-retrieval-entry/v2/linkedRetrieval/commands/aiSearch',
            # 接口請求體內容格式,
            req_body_type='formData',
            # 接口響應體內容格式,
            body_type='sse'
        )
        return params

    async def do_sse_query(self,
                           search_query: str,
                           session_id: Optional[str] = None,
                           time_range: Optional[str] = None):
        """
        發起 SSE 請求
        :param search_query: 用戶Query， 長度: [2,100]
        :param session_id: 多輪交互的session_id
        :param time_range: 時間范圍，支持: OneDay, OneWeek, OneMonth, OneYear, NoLimit
        :return:
        """

        assert self._client is not None
        request = open_api_models.OpenApiRequest(
            query={
                "query": search_query,
                "sessionId": session_id,
                "timeRange": time_range
            }
        )
        sse_receiver = self._client.call_sse_api_async(params=self._api_info, request=request, runtime=self._runtime)
        return sse_receiver

Go SDK

• SDK依賴

require (
  github.com/alibabacloud-go/tea-utils/v2 v2.0.7
)

• 調用示例

package main

import (
	"encoding/json"
	"fmt"
	"go_example/aliyun/linked_retrieval"
	"time"
)

func main() {
	invokeQuery("董宇輝")
	invokeQuery("黑神話悟空")
	time.Sleep(1 * time.Second)
	invokeQuery("董")

}

func invokeQuery(query string) {
	timeRange := "OneYear"
	sessionId := "testSessionId"
	fmt.Printf("\n\n\n\n************ %s", query)

	events, err := linked_retrieval.DoSseQuery(query, &sessionId, &timeRange)
	if err != nil {
		fmt.Printf("query from linked_retrieval failed, %s", err)
		return
	}

	for event := range events {

		var eventData linked_retrieval.EventData
		json.Unmarshal([]byte(*event.Data), &eventData)

		// 請求RequestId, 排查問題時可以提供此信息
		requestId := eventData.RequestID

		// 當前的eventName，支持: on_common_search_end, on_post_retrieval_end 兩種事件，可以根據需要選擇
		eventName := eventData.Header.Event

		// 服務端當前Event的時延(從服務端接收到請求開始)
		serverRT := eventData.Header.ResponseTime

		// 具體消息的內容，參考文檔說明: 內部是一個JSON，
		payload := eventData.Payload

		if eventName == "on_common_search_end" {
			fmt.Printf("[%s] %s serverRt:%d, payload:%s \n\n", requestId, eventName, serverRT, payload)
		}
		if eventName == "on_post_retrieval_end" {
			fmt.Printf("[%s] %s serverRt:%d, payload:%s \n\n", requestId, eventName, serverRT, payload)
		}
		if eventName == "on_error_event" {
			fmt.Printf("[%s] %s errorPayload:%s \n\n", requestId, eventName, payload)
		}
	}
}

• go_example/aliyun/linked_retrieval

package linked_retrieval

import (
	openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client"
	openapiutil "github.com/alibabacloud-go/openapi-util/service"
	util "github.com/alibabacloud-go/tea-utils/v2/service"
	"github.com/alibabacloud-go/tea/tea"
	"io"
	"os"
)

const (
	endpoint           = "linkedmallretrieval.cn-zhangjiakou.aliyuncs.com"
	pathName           = "/linked-retrieval/linked-retrieval-entry/v2/linkedRetrieval/commands/aiSearch"
	timeoutMillSeconds = 5000
)

func createApiInfo() *openapi.Params {
	params := &openapi.Params{
		// 接口名稱
		Action: tea.String("AISearchV2"),
		// 接口版本
		Version: tea.String("2024-05-01"),
		// 接口協議
		Protocol: tea.String("HTTPS"),
		// 接口 HTTP 方法
		Method:   tea.String("GET"),
		AuthType: tea.String("AK"),
		Style:    tea.String("ROA"),
		// 接口 PATH
		Pathname: tea.String(pathName),
		// 接口請求體內容格式
		ReqBodyType: tea.String("json"),
		// 接口響應體內容格式，注意一定得是binary格式，CallApi才會透傳出response body進行ReadAsSSE
		BodyType: tea.String("binary"),
	}
	return params
}

func DoSseQuery(query string, sessionId *string, timeRange *string) (<-chan util.SSEEvent, error) {
	// 在環境變量中設置您子賬號的AK/SK，并修改此處的AK/SK的環境變量名
	accessKeyID := os.Getenv("ACCESS_KEY_AIGC_01")
	accessKeySecret := os.Getenv("ACCESS_SECRET_AIGC_01")

	config := &openapi.Config{
		AccessKeyId:     tea.String(accessKeyID),
		AccessKeySecret: tea.String(accessKeySecret),
		Endpoint:        tea.String(endpoint),
		ReadTimeout:     tea.Int(timeoutMillSeconds),
	}
	client, err := openapi.NewClient(config)
	if err != nil {
		return nil, err
	}

	params := createApiInfo()
	// query params
	queries := map[string]interface{}{
		"query":     tea.String(query),
		"sessionId": tea.StringValue(sessionId),
		"timeRange": tea.StringValue(timeRange),
	}

	// runtime options
	runtime := &util.RuntimeOptions{}
	request := &openapi.OpenApiRequest{
		Query: openapiutil.Query(queries),
	}
	// 復制代碼運行請自行打印 API 的返回值
	// 返回值為 Map 類型，可從 Map 中獲得三類數據：響應體 body、響應頭 headers、HTTP 返回的狀態碼 statusCode。
	resp, err := client.CallApi(params, request, runtime)
	if err != nil {
		return nil, err
	}

	// 迭代讀取SSE內容
	events, sseErrors := util.ReadAsSSE(resp["body"].(io.ReadCloser))

	select {
	case sseError := <-sseErrors:
		err = sseError
	default:
		// 沒有錯誤的情況
		err = nil
	}
	return events, err
}

// 定義主結構體
type EventData struct {
	Payload   string `json:"payload"`
	RequestID string `json:"requestId"`
	Header    Header `json:"header"`
}

// 定義 Header 結構體
type Header struct {
	EventID      string `json:"eventId"`
	ResponseTime int    `json:"responseTime"`
	Event        string `json:"event"`
}