:DAS Agent と Chat API を統合するためのベストプラクティス

SDK のインストール

pip3 install alibabacloud_das20200116==3.0.0

サンプルコード

# -*- coding: utf-8 -*-

import argparse
import json
import sys
import uuid

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class ChatSample:

    @staticmethod
    def create_client() -> OpenApiClient:
        credential = CredentialClient()
        config = open_api_models.Config(credential=credential)
        config.endpoint = 'das.cn-shanghai.aliyuncs.com'
        return OpenApiClient(config)

    @staticmethod
    def create_api_info() -> open_api_models.Params:
        return open_api_models.Params(
            action='Chat',
            version='2020-01-16',
            protocol='HTTPS',
            pathname='/chat',
            method='POST',
            auth_type='AK',
            style='RPC',
            req_body_type='json',
            body_type='sse',
        )

    @staticmethod
    def build_message(text: str) -> str:
        message = {
            'id': str(uuid.uuid4()),
            'role': 'user',
            'content': [{'type': 'text', 'text': text}],
        }
        return json.dumps(message, ensure_ascii=False)

    @staticmethod
    def build_request(
        text: str,
        session_id: str = None,
        agent_id: str = None,
        summary: str = None,
    ) -> open_api_models.OpenApiRequest:
        query = {'Message': ChatSample.build_message(text)}
        if not UtilClient.is_unset(session_id):
            query['SessionId'] = session_id
        if not UtilClient.is_unset(agent_id):
            query['AgentId'] = agent_id
        if not UtilClient.is_unset(summary):
            query['Summary'] = summary
        return open_api_models.OpenApiRequest(query=query, headers={})

    @staticmethod
    def parse_event_data(event) -> dict:
        if not hasattr(event, 'data'):
            return {}
        data = event.data
        if isinstance(data, str):
            try:
                data = json.loads(data)
            except json.JSONDecodeError:
                return {}
        return data if isinstance(data, dict) else {}

    @staticmethod
    def run(
        text: str,
        session_id: str = None,
        agent_id: str = None,
        summary: str = None,
    ) -> None:
        client = ChatSample.create_client()
        params = ChatSample.create_api_info()
        runtime = util_models.RuntimeOptions()
        request = ChatSample.build_request(text, session_id, agent_id, summary)

        response = client.call_sseapi(params, request, runtime)
        full_content = []

        for res in response:
            data = ChatSample.parse_event_data(res.event)
            if not data:
                continue

            activity = data.get('ActivityType')
            delta = data.get('Delta')
            content = data.get('Content')
            ext_name = data.get('Name')
            ext_value = data.get('Value')

            if delta:
                print(delta, end='', flush=True)
                full_content.append(delta)
            elif content and not delta:
                print(f"\n[Content] {content}", flush=True)
            elif activity:
                print(f"\n[Activity] {activity}", file=sys.stderr, flush=True)

            if ext_name == 'summary' and ext_value:
                print(f"\n[Summary] {ext_value}", flush=True)

        if full_content:
            print()


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='DAS Chat API のサンプル')
    parser.add_argument(
        'query',
        nargs='?',
        default='Describe DAS Agent in about 1000 words',
        help='ユーザーのクエリ',
    )
    parser.add_argument('--session-id', dest='session_id', help='セッション ID (UUID)')
    parser.add_argument('--agent-id', dest='agent_id', help='エージェント ID')
    parser.add_argument('--summary', choices=['true', 'false'], help='概要情報を返すかどうか')
    return parser.parse_args()


if __name__ == '__main__':
    args = parse_args()
    ChatSample.run(
        text=args.query,
        session_id=args.session_id,
        agent_id=args.agent_id,
        summary=args.summary,
    )

実行コマンド

# シングルターンの会話
python3 chat_sample.py "DAS Agent について 1000 ワード程度で説明してください"

# 複数ターンの会話 (SessionId と AgentId を渡します)
python3 chat_sample.py "スロークエリ分析について詳しく教えてください" \
    --session-id 123e4567-e89b-12d3-a456-xxxxxxxxxxxx \
    --agent-id ag-472T0DxtmjIxxxxx \
    --summary true

SDK のインストール

go mod init das_agent_chat_demo
go get github.com/alibabacloud-go/darabonba-openapi/v2/client
go get github.com/alibabacloud-go/tea-utils/v2/service
go get github.com/alibabacloud-go/tea/tea
go get github.com/google/uuid

サンプルコード

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"strings"

	openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client"
	openapiutil "github.com/alibabacloud-go/darabonba-openapi/v2/utils"
	"github.com/alibabacloud-go/tea/dara"
	"github.com/alibabacloud-go/tea/tea"
	"github.com/google/uuid"
)

func createClient() (*openapi.Client, error) {
	config := &openapi.Config{
		AccessKeyId:     tea.String(os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")),
		AccessKeySecret: tea.String(os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")),
		Endpoint:        tea.String("das.cn-shanghai.aliyuncs.com"),
	}
	return openapi.NewClient(config)
}

func createAPIInfo() *openapi.Params {
	return &openapi.Params{
		Action:      tea.String("Chat"),
		Version:     tea.String("2020-01-16"),
		Protocol:    tea.String("HTTPS"),
		Pathname:    tea.String("/chat"),
		Method:      tea.String("POST"),
		AuthType:    tea.String("AK"),
		Style:       tea.String("RPC"),
		ReqBodyType: tea.String("json"),
		BodyType:    tea.String("sse"),
	}
}

func buildMessage(text string) (string, error) {
	payload := map[string]interface{}{
		"id":   uuid.NewString(),
		"role": "user",
		"content": []map[string]string{
			{"type": "text", "text": text},
		},
	}
	bs, err := json.Marshal(payload)
	if err != nil {
		return "", err
	}
	return string(bs), nil
}

func buildRequest(text, sessionID, agentID, summary string) (*openapi.OpenApiRequest, error) {
	msg, err := buildMessage(text)
	if err != nil {
		return nil, err
	}
	query := map[string]interface{}{
		"Message": msg,
	}
	if sessionID != "" {
		query["SessionId"] = sessionID
	}
	if agentID != "" {
		query["AgentId"] = agentID
	}
	if summary != "" {
		query["Summary"] = summary
	}
	return &openapi.OpenApiRequest{
		Query:   openapiutil.Query(query),
		Headers: map[string]*string{},
	}, nil
}

func parseEventData(raw *string) map[string]interface{} {
	if raw == nil || *raw == "" {
		return nil
	}
	var data map[string]interface{}
	if err := json.Unmarshal([]byte(*raw), &data); err != nil {
		return nil
	}
	return data
}

func run(text, sessionID, agentID, summary string) error {
	client, err := createClient()
	if err != nil {
		return err
	}
	params := createAPIInfo()
	request, err := buildRequest(text, sessionID, agentID, summary)
	if err != nil {
		return err
	}
	runtime := &dara.RuntimeOptions{}

	sseChan := make(chan *openapi.SSEResponse, 100)
	errChan := make(chan error, 1)
	go client.CallSSEApi(params, request, runtime, sseChan, errChan)

	for event := range sseChan {
		if event.Event == nil {
			continue
		}
		data := parseEventData(event.Event.Data)
		if data == nil {
			continue
		}

		delta, _ := data["Delta"].(string)
		content, _ := data["Content"].(string)
		activity, _ := data["ActivityType"].(string)
		extName, _ := data["Name"].(string)
		extValue := data["Value"]

		switch {
		case delta != "":
			fmt.Print(delta)
		case content != "":
			fmt.Println()
			fmt.Println("[Content]", content)
		case activity != "":
			fmt.Fprintln(os.Stderr, "[Activity]", activity)
		}

		if extName == "summary" && extValue != nil {
			fmt.Println()
			fmt.Println("[Summary]", extValue)
		}
	}
	fmt.Println()
	if err := <-errChan; err != nil {
		return err
	}
	return nil
}

func main() {
	fs := flag.NewFlagSet("chat_sample", flag.ExitOnError)
	sessionID := fs.String("session-id", "", "セッション ID (UUID)")
	agentID := fs.String("agent-id", "", "エージェント ID")
	summary := fs.String("summary", "", "概要を返すかどうか (true/false)")
	fs.Usage = func() {
		fmt.Fprintf(os.Stderr, "Usage: %s [query] [--session-id ID] [--agent-id ID] [--summary true|false]\n", os.Args[0])
		fs.PrintDefaults()
	}
	if err := fs.Parse(os.Args[1:]); err != nil {
		os.Exit(2)
	}

	query := "Describe DAS Agent in about 1000 words"
	if fs.NArg() > 0 {
		query = strings.Join(fs.Args(), " ")
	}

	if err := run(query, *sessionID, *agentID, *summary); err != nil {
		fmt.Fprintln(os.Stderr, "リクエストに失敗しました:", err)
		os.Exit(1)
	}
}

実行コマンド

# シングルターンの会話
go run chat_sample.go "DAS Agent について 1000 ワード程度で説明してください"

# 複数ターンの会話 (SessionId と AgentId を渡します)
go run chat_sample.go "スロークエリ分析について詳しく教えてください" \
    --session-id 123e4567-e89b-12d3-a456-xxxxxxxxxxxx \
    --agent-id ag-472T0DxtmjIxxxxx \
    --summary true