Go 用 SDK - Platform For AI - Alibaba Cloud ドキュメントセンター

公式の Elastic Algorithm Service（EAS）SDK は、モデルに基づいてデプロイされたサービスを呼び出すために提供されています。 EAS SDK は、呼び出しロジックを定義するために必要な時間を短縮し、呼び出しの安定性を向上させます。このトピックでは、Go 用 EAS SDK について説明します。デモが提供されており、Go 用 EAS SDK を使用してサービスを呼び出す方法を示しています。これらのデモでは、入力と出力は一般的に使用される型です。

背景情報

Go 用 EAS SDK を事前にインストールする必要はありません。 SDK は、コードのコンパイル中に GO 言語のパッケージマネージャによって GitHub から自動的にダウンロードされます。呼び出しロジックの特定の部分をカスタマイズするには、Go 用 EAS SDK をダウンロードしてコードを変更します。 SDK をダウンロードするには、eas-golang-sdk にアクセスしてください。

メソッド

クラス	メソッド	説明
PredictClient	`NewPredictClient(endpoint string, serviceName string) *PredictClient`	説明: PredictClient クラスのクライアントオブジェクトを作成します。パラメーター: endpoint: 必須。サーバーのエンドポイント。通常モードでサービスを呼び出すには、このパラメーターをデフォルトゲートウェイのエンドポイントに設定します。 serviceName: 必須。呼び出すサービスの名前。戻り値: 作成されたクライアントオブジェクト。
	`SetEndpoint(endpointName string)`	説明: サーバーのエンドポイントを指定します。パラメーター: endpointName: サーバーのエンドポイント。通常モードでサービスを呼び出すには、このパラメーターをデフォルトゲートウェイのエンドポイントに設定します。
	`SetServiceName(serviceName string)`	説明: サービスの名前を指定します。パラメーター: serviceName: サービスの名前。
	`SetEndpointType(endpointType string)`	説明: サーバーのゲートウェイの種類を指定します。パラメーター: endpointType: 使用するゲートウェイの種類。次のゲートウェイの種類がサポートされています: "DEFAULT": デフォルトゲートウェイ。ゲートウェイの種類を指定しない場合は、デフォルトゲートウェイが使用されます。 "DIRECT": Virtual Private Cloud（VPC）直接接続チャンネル。
	`SetToken(token string)`	説明: サービスアクセストークンを指定します。パラメーター: token: サービスの認証に使用されるトークン。
	`SetHttpTransport(transport *http.Transport)`	説明: HTTP クライアントの transport 属性を設定します。パラメーター: transport: HTTP リクエストを送信するために使用される transport オブジェクト。
	`SetRetryCount(max_retry_count int)`	説明: リクエストの失敗後に許可される最大リトライ回数を指定します。パラメーター: max_retry_count: リクエストの失敗後に許可される最大リトライ回数。デフォルト値: 5. 重要サーバーで処理エラーが発生した場合、サーバーエラーが発生した場合、またはゲートウェイへの持続的接続が閉じられた場合、クライアントはリクエストを再送信する必要があります。したがって、このパラメーターを 0 に設定しないことをお勧めします。
	`SetTimeout(timeout int)`	説明: リクエストのタイムアウト期間を設定します。パラメーター: timeout: リクエストのタイムアウト期間。単位: ミリ秒。デフォルト値: 5000。
	`Init()`	説明: クライアントオブジェクトを初期化します。パラメーターを設定するために使用される上記のいずれかのメソッドが呼び出された場合、`Init()` メソッドを呼び出すまで、パラメーターは有効になりません。
	`Predict(request Request) Response`	説明: オンライン予測サービスに予測リクエストを送信します。パラメーター: Request: 送信するリクエスト。文字列、TFRequest オブジェクト、または TorchRequest オブジェクトを指定できます。戻り値: 予測リクエストへの応答。文字列、TFResponse オブジェクト、または TorchResponse リクエストを指定できます。
	`StringPredict(request string) string`	説明: オンライン予測サービスに予測リクエストを送信します。パラメーター: request: 送信するリクエスト文字列。戻り値: 予測リクエストへの応答。文字列として返されます。
	`TorchPredict(request TorchRequest) TorchResponse`	説明: オンライン予測サービスに PyTorch 予測リクエストを送信します。パラメーター: request: 送信するリクエスト。TorchRequest オブジェクトです。戻り値: 予測リクエストへの応答。 TorchResponse オブジェクトです。
	`TFPredict(request TFRequest) TFResponse`	説明: オンライン予測サービスに予測リクエストを送信します。パラメーター: request: 送信するリクエスト。TFRequest オブジェクトです。戻り値: 予測リクエストへの応答。 TFResponse オブジェクトです。
TFRequest	`TFRequest(signatureName string)`	説明: TFRequest クラスのオブジェクトを作成します。パラメーター: signatureName: 呼び出すサービスのモデルのシグネチャ名。
	`AddFeed(?)(inputName string, shape []int64{}, content []?)`	説明: 呼び出すオンライン予測サービスの TensorFlow モデルの入力テンソルを指定します。パラメーター: inputName: 入力テンソルのエイリアス。 shape: 入力テンソルの形状。 content: 入力テンソルのデータ。 1 次元配列の形式で値を指定します。データの型は、INT32、INT64、FLOAT32、FLOAT64、STRING、または BOOL を指定できます。このメソッドの名前は、使用される特定のデータの型によって決まります。例: `AddFeedInt32()`。他のデータの型を使用する場合、指定されたコードに基づいて protocol buffer（PB）形式でデータの型を構築します。
	`AddFetch(outputName string)`	説明: エクスポートする TensorFlow モデルの出力テンソルのエイリアスを指定します。パラメーター: outputName: エクスポートする出力テンソルのエイリアス。 TensorFlow モデルが SavedModel 形式の場合、このパラメーターはオプションです。このパラメーターを指定しない場合、すべての出力テンソルがエクスポートされます。 TensorFlow モデルがフリーズモデルの場合、このパラメーターは必須です。
TFResponse	`GetTensorShape(outputName string) []int64`	説明: 指定されたエイリアスで識別される出力テンソルの形状をクエリします。パラメーター: outputName: 形状をクエリする出力テンソルのエイリアス。戻り値: 出力テンソルの形状。各ディメンションは配列として表示されます。
TFResponse	`Get(?)Val(outputName string) [](?)`	説明: 指定された出力テンソルのデータをクエリします。戻り値は 1 次元配列です。このメソッドを `GetTensorShape()` メソッドと一緒に呼び出して、出力テンソルの形状をクエリできます。戻り値は多次元配列です。データの型は、FLOAT、DOUBLE、INT、INT64、STRING、または BOOL を指定できます。このメソッドの名前は、使用される特定のデータの型によって決まります。例: `GetFloatVal()`。パラメーター: outputName: データをクエリする出力テンソルのエイリアス。戻り値: 取得した出力テンソルデータから変換された 1 次元配列。
TorchRequest	`TorchRequest()`	説明: TFRequest クラスのオブジェクトを作成します。
	`AddFeed(?)(index int, shape []int64{}, content []?)`	説明: 呼び出すオンライン予測サービスの PyTorch モデルの入力テンソルを指定します。パラメーター: index: 入力テンソルのインデックス。 shape: 入力テンソルの形状。 content: 入力テンソルのデータ。 1 次元配列の形式で値を指定します。データの型は、INT32、INT64、FLOAT32、または FLOAT64 を指定できます。このメソッドの名前は、使用される特定のデータの型によって決まります。例: `AddFeedInt32()`。他のデータの型を使用する場合、指定されたコードに基づいて protocol buffer（PB）形式でデータの型を構築します。
	`AddFetch(outputIndex int)`	説明: エクスポートする PyTorch モデルの出力テンソルのインデックスを指定します。このメソッドはオプションです。このメソッドを呼び出して出力テンソルのインデックスを設定しない場合、すべての出力テンソルがエクスポートされます。パラメーター: outputIndex: エクスポートする出力テンソルのインデックス。
TorchResponse	`GetTensorShape(outputIndex int) []int64`	説明: 指定されたインデックスで識別される出力テンソルの形状をクエリします。パラメーター: outputIndex: 出力テンソルのインデックス。戻り値: 出力テンソルの形状。各ディメンションは配列として表示されます。
TorchResponse	`Get(?)Val(outputIndex int) [](?)`	説明: 指定された出力テンソルのデータをクエリします。戻り値は 1 次元配列です。このメソッドを `GetTensorShape()` メソッドと一緒に呼び出して、出力テンソルの形状を取得できます。戻り値は多次元配列です。データの型は、FLOAT、DOUBLE、INT、または INT64 を指定できます。このメソッドの名前は、使用される特定のデータの型によって決まります。例: `GetFloatVal()`。パラメーター: outputName: データをクエリする出力テンソルのインデックス。戻り値: 取得した出力テンソルデータから変換された 1 次元配列。
QueueClient	`NewQueueClient(endpoint, queueName, token string) (*QueueClient, error)`	説明: QueueClient クラスのクライアントオブジェクトを作成します。パラメーター: endpoint: サーバーのエンドポイント。 queueName: 作成するキューの名前。 token: 作成するキューのトークン。戻り値: 作成されたクライアントオブジェクト。
	`Truncate(ctx context.Context, index uint64) error`	説明: 特定のインデックス値より前のデータを切り捨て、インデックス値より後のデータのみを保持します。パラメーター: ctx: この操作のコンテキスト。 index: データを切り捨てるために使用されるインデックス値。
	`Put(ctx context.Context, data []byte, tags types.Tags) (index uint64, requestId string, err error)`	説明: データレコードをキューに書き込みます。パラメーター: ctx: この操作のコンテキスト。 data: キューに書き込むデータレコード。戻り値: index: 書き込まれたデータレコードのインデックス値。この値を使用して、キュー内のデータをクエリできます。 requestId: キュー内に書き込まれたデータレコードに対して自動的に生成されるリクエスト ID。 requestId は、キュー内のデータをクエリするための特別なタグとして使用できます。
	`GetByIndex(ctx context.Context, index uint64) (dfs []types.DataFrame, err error)`	説明: レコードのインデックス値に基づいてキューからデータレコードをクエリし、レコードを削除します。パラメーター: ctx: この操作のコンテキスト。 index: クエリするデータレコードのインデックス値。戻り値: dfs: DataFrame 形式で取得されたデータレコード。
	`GetByRequestId(ctx context.Context, requestId string) (dfs []types.DataFrame, err error)`	説明: レコードのリクエスト ID に基づいてキューからデータレコードをクエリし、レコードを削除します。パラメーター: ctx: この操作のコンテキスト。 requestId: クエリするデータレコードのリクエスト ID。戻り値: dfs: DataFrame 形式で取得されたデータレコード。
	`Get(ctx context.Context, index uint64, length int, timeout time.Duration, autoDelete bool, tags types.Tags) (dfs []types.DataFrame, err error)`	説明: 特定の条件に基づいてキュー内のデータをクエリします。 `GetByIndex()` メソッドと `GetByRequestId()` メソッドは、`get()` メソッドのカプセル化です。パラメーター: ctx: この操作のコンテキスト。 index: クエリの開始インデックス。 length: クエリするデータレコードの数。このパラメーターを指定すると、index から始まる最大数のデータレコードが返されます。 index 値に一致するデータレコードも返されます。 timeout: クエリのタイムアウト期間。タイムアウト期間中にキューにデータが含まれている場合、指定された条件を満たすデータレコードの数が返されます。それ以外の場合、タイムアウト期間が経過するとクエリは停止します。 auto_delete: 取得したデータレコードをキューから自動的に削除するかどうかを指定します。 auto_delete を False に設定すると、データレコードを繰り返しクエリできます。この場合、`GetByIndex()` メソッドとメソッドを使用して、キューからデータレコードを取得して削除できます。タグ: データレコードを照会するために使用されるタグ。タグのキーと値は両方とも STRING 型である必要があります。このパラメーターが指定されている場合、指定されたタグで追加された index から始まるデータレコードが返されます。戻り値: dfs: DataFrame フォーマットで取得されたデータレコードです。
	`Del(ctx context.Context, indexes ...uint64)`	説明: 指定されたインデックス値に一致するデータレコードをキューから削除します。パラメータ: ctx: この操作のコンテキスト。 indexes: データレコードの削除に使用する、指定されたインデックス値。
	`Attributes() (attrs types.Attributes, err error)`	説明: キューの属性をクエリします。属性には、キュー内のデータレコードの総数と現在のキュー内のデータレコードの数が含まれます。戻り値: attrs: キューの属性。属性のキーと値は両方とも STRING 型である必要があります。
	`Watch(ctx context.Context, index, window uint64, indexOnly bool, autocommit bool) (watcher types.Watcher, err error)`	説明: キュー内のデータレコードをサブスクライブします。その後、キューイングサービスは、指定された条件に基づいてクライアントにデータをプッシュします。パラメータ: ctx: この操作のコンテキスト。 index: サブスクライブされるデータレコードの開始インデックス。 window: キューイングサービスによって単一のクライアントにプッシュされることが許可されるデータレコードの最大数。説明データがコミットされていない場合、サーバーは他のデータレコードをクライアントにプッシュしません。その後、N 個のデータレコードがコミットされると、N 個のデータレコードがサーバーにプッシュされます。これにより、クライアントによって処理されるデータレコードの数が window に指定された値を超えないことが保証されます。このようにして、クライアント側のコンピューティングの同時実行性が制御されます。 index_only: インデックス値のみをプッシュするかどうかを指定します。 auto_commit: レコードがプッシュされた後にデータレコードを自動的にコミットするかどうかを指定します。 auto_commit を False に設定することをお勧めします。この場合、レコードを受信して計算した後に、データレコードを手動でコミットする必要があります。計算が完了する前にインスタンスで例外が発生した場合、コミットされていないデータレコードは、キューイングサービスによって他のインスタンスにプッシュされます。戻り値: プッシュされたデータを読み取るために使用されるウォッチャー。
	`Commit(ctx context.Context, indexes ...uint64) error`	説明: 指定されたデータレコードをコミットします。説明データレコードが処理され、他のインスタンスにプッシュする必要がない場合は、コミットされます。その後、データレコードはキューから削除できます。パラメータ: ctx: この操作のコンテキスト。 indexes: コミットされたデータレコードに一致する、指定されたインデックス値。
types.Watcher	`FrameChan() <-chan types.DataFrame`	説明: サーバーからプッシュされたデータを含むパイプラインを返します。パイプラインを使用して、データを繰り返し読み取ることができます。戻り値: プッシュされたデータを読み取るために使用できるパイプライン。
types.Watcher	`Close()`	説明：バックエンド接続を閉じるためにウォッチャーを停止します。説明 1 つのクライアントに対して開始できるウォッチャーは 1 つだけです。別のウォッチャーを開始する前に、ウォッチャーを閉じる必要があります。

デモ

文字列としての入力と出力

カスタムプロセッサを使用してモデルをサービスとしてデプロイする場合、Predictive Model Markup Language（PMML）モデルに基づいてデプロイされたサービスなど、サービスを呼び出すために文字列がよく使用されます。詳細については、次のデモをご参照ください。

package main

// 必要なパッケージをインポートします
import (
        "fmt"
        "github.com/pai-eas/eas-golang-sdk/eas"
)

func main() {
    client := eas.NewPredictClient("182848887922****.cn-shanghai.pai-eas.aliyuncs.com", "scorecard_pmml_example")
    client.SetToken("YWFlMDYyZDNmNTc3M2I3MzMwYmY0MmYwM2Y2MTYxMTY4NzBkNzdj****")
    client.Init()
    req := "[{\"fea1\": 1, \"fea2\": 2}]"
    for i := 0; i < 100; i++ {
        resp, err := client.StringPredict(req)
        if err != nil {
            fmt.Printf("failed to predict: %v\n", err.Error())
        } else {
            fmt.Printf("%v\n", resp)
        }
    }
}

テンソルとしての入力と出力

TensorFlow を使用してモデルをサービスとしてデプロイする場合は、サービスを呼び出すために TFRequest クラスと TFResponse クラスを使用する必要があります。詳細については、次のデモをご参照ください。

package main

// 必要なパッケージをインポートします
import (
        "fmt"
        "github.com/pai-eas/eas-golang-sdk/eas"
)

func main() {
    client := eas.NewPredictClient("182848887922****.cn-shanghai.pai-eas.aliyuncs.com", "mnist_saved_model_example")
    client.SetToken("YTg2ZjE0ZjM4ZmE3OTc0NzYxZDMyNmYzMTJjZTQ1YmU0N2FjMTAy****")
    client.Init()

    tfreq := eas.TFRequest{}
    tfreq.SetSignatureName("predict_images")
    tfreq.AddFeedFloat32("images", []int64{1, 784}, make([]float32, 784))

    for i := 0; i < 100; i++ {
        resp, err := client.TFPredict(tfreq)
        if err != nil {
            fmt.Printf("failed to predict: %v", err)
        } else {
            fmt.Printf("%v\n", resp)
        }
    }
}

PyTorch モデルの呼び出し

PyTorch を使用してモデルをサービスとしてデプロイする場合は、サービスを呼び出すために TorchRequest クラスと TorchResponse クラスを使用する必要があります。詳細については、次のデモをご参照ください。

package main

// 必要なパッケージをインポートします
import (
        "fmt"
        "github.com/pai-eas/eas-golang-sdk/eas"
)

func main() {
    client := eas.NewPredictClient("182848887922****.cn-shanghai.pai-eas.aliyuncs.com", "pytorch_resnet_example")
    client.SetTimeout(500)
    client.SetToken("ZjdjZDg1NWVlMWI2NTU5YzJiMmY5ZmE5OTBmYzZkMjI0YjlmYWVl****")
    client.Init()
    req := eas.TorchRequest{}
    req.AddFeedFloat32(0, []int64{1, 3, 224, 224}, make([]float32, 150528))
    req.AddFetch(0)
    for i := 0; i < 10; i++ {
        resp, err := client.TorchPredict(req)
        if err != nil {
            fmt.Printf("failed to predict: %v", err)
        } else {
            fmt.Println(resp.GetTensorShape(0), resp.GetFloatVal(0))
        }
    }
}

VPC 直接接続チャネルを使用してサービスを呼び出す
VPC 直接接続チャネルを使用して、EAS 専用リソースグループにデプロイされているサービスのみにアクセスできます。さらに、チャネルを使用するには、EAS 専用リソースグループと指定された vSwitch が VPC に接続されている必要があります。詳細については、「EAS リソースグループの操作」および「ネットワーク接続の設定」をご参照ください。通常モードと比較して、このモードには追加のコード行 client.SetEndpointType(eas.EndpointTypeDirect) が含まれています。このモードは、高並列性と大量トラフィックのシナリオで使用できます。詳細については、次のデモをご参照ください。
```
package main

// 必要なパッケージをインポートします
import (
        "fmt"
        "github.com/pai-eas/eas-golang-sdk/eas"
)

func main() {
    client := eas.NewPredictClient("pai-eas-vpc.cn-shanghai.aliyuncs.com", "scorecard_pmml_example")
    client.SetToken("YWFlMDYyZDNmNTc3M2I3MzMwYmY0MmYwM2Y2MTYxMTY4NzBkNzdj****")
    client.SetEndpointType(eas.EndpointTypeDirect) // エンドポイントタイプを直接接続に設定
    client.Init()
    req := "[{\"fea1\": 1, \"fea2\": 2}]"
    for i := 0; i < 100; i++ {
        resp, err := client.StringPredict(req)
        if err != nil {
            fmt.Printf("failed to predict: %v\n", err.Error())
        } else {
            fmt.Printf("%v\n", resp)
        }
    }
}
```

クライアントの接続パラメータを設定する

http.Transport 属性を使用して、クライアントの接続パラメータを設定できます。詳細については、次のデモをご参照ください。

package main

// 必要なパッケージをインポートします
import (
        "fmt"
        "github.com/pai-eas/eas-golang-sdk/eas"
        "net/http"
        "time"
)

func main() {
    client := eas.NewPredictClient("pai-eas-vpc.cn-shanghai.aliyuncs.com", "network_test")
    client.SetToken("MDAwZDQ3NjE3OThhOTI4ODFmMjJiYzE0MDk1NWRkOGI1MmVhMGI0****")
    client.SetEndpointType(eas.EndpointTypeDirect)
    // クライアントの HTTP トランスポートパラメータを設定します
    client.SetHttpTransport(&http.Transport{
        MaxConnsPerHost:       300,
        TLSHandshakeTimeout:   100 * time.Millisecond,
        ResponseHeaderTimeout: 200 * time.Millisecond,
        ExpectContinueTimeout: 200 * time.Millisecond,
    })
}

キューイングサービスを使用してデータを送受信する

キュー内のデータの送受信、キューの状態のクエリ、およびキューによってプッシュされたデータのサブスクライブを行うことができます。次のデモでは、1 つのスレッドがキューにデータをプッシュし、別のスレッドがウォッチャーを使用してプッシュされたデータをサブスクライブします。詳細については、次のデモをご参照ください。

説明

EAS に非同期推論サービスをデプロイすると、入力キューと出力キューが次の形式で自動的に生成されます。

入力キュー: <domain>/api/predict/<service_name>

出力キュー: <domain>/api/predict/<service_name>/sink

実際のニーズに応じて、<service_name> または <service_name>/sink を使用して QueueClient を構築します。

    const (
        QueueEndpoint = "182848887922****.cn-shanghai.pai-eas.aliyuncs.com"
        // 例: EAS サービス名が test_qservice の場合、入力キュー名は test_qservice、出力キュー名は test_qservice/sink になります。
        QueueName     = "test_qservice"
        QueueToken    = "YmE3NDkyMzdiMzNmMGM3ZmE4ZmNjZDk0M2NiMDA3OTZmNzc1MTUx****"
    )
    queue, err := NewQueueClient(QueueEndpoint, QueueName, QueueToken)

    // キュー内のすべてのメッセージを切り詰めます
    attrs, err := queue.Attributes()
    if index, ok := attrs["stream.lastEntry"]; ok {
        idx, _ := strconv.ParseUint(index, 10, 64)
        queue.Truncate(context.Background(), idx+1)
    }

    ctx, cancel := context.WithCancel(context.Background())

    // キューにメッセージを送信するゴルーチンを作成します
    go func() {
        i := 0
        for {
            select {
            case <-time.NewTicker(time.Microsecond * 1).C:
                _, _, err := queue.Put(context.Background(), []byte(strconv.Itoa(i)), types.Tags{})
                if err != nil {
                    fmt.Printf("Error occured, retry to handle it: %v\n", err)
                }
                i += 1
            case <-ctx.Done():
                break
            }
        }
    }()

    // キューからのメッセージを監視するウォッチャーを作成します
    watcher, err := queue.Watch(context.Background(), 0, 5, false, false)
    if err != nil {
        fmt.Printf("Failed to create a watcher to watch the queue: %v\n", err)
        return
    }

    // キューからメッセージを読み取り、手動でコミットします
    for i := 0; i < 100; i++ {
        df := <-watcher.FrameChan()
        err := queue.Commit(context.Background(), df.Index.Uint64())
        if err != nil {
            fmt.Printf("Failed to commit index: %v(%v)\n", df.Index, err)
        }
    }

    // すべて完了したら、ウォッチャーを閉じます
    watcher.Close()
    cancel()