Platform For AI:TorchEasyRec Processor

背景情報

以下の図は、TorchEasyRec プロセッサに基づくレコメンデーションエンジンのアーキテクチャを示しています。

TorchEasyRec プロセッサは、以下のモジュールで構成されています。

• アイテム特徴量キャッシュ：フィーチャーストアからアイテム側の特徴量をメモリにキャッシュして、ネットワークオーバーヘッドを削減し、フィーチャーストアへのリクエスト負荷を軽減し、推論サービスのパフォーマンスを向上させます。アイテム側の特徴量にリアルタイムデータが含まれる場合、フィーチャーストアがその同期を処理します。

• 特徴量ジェネレーター (FG)：設定ファイルで特徴量変換を定義し、単一の C++ コードベースを使用してオフラインとオンラインの特徴量処理の一貫性を確保するモジュールです。

• TorchModel：TorchEasyRec または PyTorch を使用してトレーニングした後、ScriptedModel としてエクスポートされた PyTorch モデルです。

制限事項

汎用 g6、g7、または g8 インスタンスファミリーのインスタンスタイプ、および T4 や A10 などの GPU のみを使用できます。詳細については、「汎用インスタンスファミリー (g シリーズ)」をご参照ください。GPU サービスをデプロイする場合、CUDA ドライバーのバージョンが 535 以降であることを確認してください。

バージョン履歴

TorchEasyRec プロセッサは現在も活発に開発が進められています。最新の機能と最高のパフォーマンスを得るために、最新バージョンを使用して推論サービスをデプロイすることを推奨します。リリースされたバージョンは以下の通りです。

ステップ 1：サービスのデプロイ

1. torcheasyrec.json サービス設定ファイルを準備します。

   Processor を easyrec-torch-{version} に設定します。{version} には、「バージョン履歴」からバージョンを選択します。以下の JSON は設定例です。

   FG あり (fg_mode='normal')

   {
     "metadata": {
       "instance": 1,
       "name": "alirec_rank_with_fg",
       "rpc": {
         "enable_jemalloc": 1,
         "max_queue_size": 256,
         "worker_threads": 16
       }
     },
     "cloud": {
           "computing": {
               "instance_type": "ecs.gn6i-c16g1.4xlarge"
           }
     },
     "model_config": {
       "fg_mode": "normal",
       "fg_threads": 8,
       "region": "YOUR_REGION",
       "fs_project": "YOUR_FS_PROJECT",
       "fs_model": "YOUR_FS_MODEL",
       "fs_entity": "item",
       "load_feature_from_offlinestore": true,
       "access_key_id":"YOUR_ACCESS_KEY_ID",
       "access_key_secret":"YOUR_ACCESS_KEY_SECRET"
     },
     "storage": [
       {
         "mount_path": "/home/admin/docker_ml/workspace/model/",
         "oss": {
           "path": "oss://xxx/xxx/export",
           "readOnly": false
         },
         "properties": {
           "resource_type": "code"
         }
       }
     ],
     "processor":"easyrec-torch-1.12"
   }

   FG なし (fg_mode='bypass')

   {
     "metadata": {
       "instance": 1,
       "name": "alirec_rank_no_fg",
       "rpc": {
         "enable_jemalloc": 1,
         "max_queue_size": 256,
         "worker_threads": 16
       }
     },
     "cloud": {
           "computing": {
               "instance_type": "ecs.gn6i-c16g1.4xlarge"
           }
     },
     "model_config": {
       "fg_mode": "bypass"
     },
     "storage": [
       {
         "mount_path": "/home/admin/docker_ml/workspace/model/",
         "oss": {
           "path": "oss://xxx/xxx/export",
           "readOnly": false
         },
         "properties": {
           "resource_type": "code"
         }
       }
     ],
     "processor":"easyrec-torch-1.12"
   }

   以下の表に、主要なパラメーターを示します。その他のパラメーターについては、「JSON デプロイ」をご参照ください。

   パラメーター	必須	説明	例
   processor	はい	TorchEasyRec プロセッサの名前です。	"processor":"easyrec-torch-1.12"
   path	はい	モデルファイルを保存するためにサービスにマウントされる Object Storage Service (OSS) パスです。	"path": "oss://examplebucket/xxx/export"
   fg_mode	いいえ	特徴量エンジニアリングモードを指定します。有効な値： bypass (デフォルト)：特徴量エンジニアリング (FG) を無効にします。Torch モデルのみがデプロイされます。 このモードは、カスタム特徴量処理が必要なシナリオ向けです。 このモードでは、プロセッサがフィーチャーストアにアクセスするためのパラメーターを設定する必要はありません。 normal：FG を有効にします。このモードは通常、モデルトレーニングに TorchEasyRec と共に使用されます。	"fg_mode": "normal"
   fg_threads	いいえ	リクエストごとの特徴量エンジニアリング (FG) の並行スレッド数です。	"fg_threads": 15
   outputs	いいえ	Torch モデルの予測からの出力変数の名前 (probs_ctr など) です。複数の変数名はカンマ (,) で区切ります。省略した場合、すべての変数が返されます。	"outputs":"probs_ctr,probs_cvr"
   item_empty_score	いいえ	アイテム ID が見つからない場合に返すデフォルトのスコアです。デフォルト値は 0 です。	"item_empty_score": -1
   プロセッサのリコールパラメーター
   faiss_neigh_num	いいえ	FAISS ベクター再現率で取得するアイテムの数です。プロセッサは、リクエストボディの faiss_neigh_num フィールドからこの値を取得します。このフィールドが指定されていない場合、プロセッサは model_config セクションの faiss_neigh_num の値を使用します。デフォルト値は 1 です。	"faiss_neigh_num": 200
   faiss_nprobe	いいえ	nprobe パラメーターは、取得プロセス中に検索するクラスターの数を指定するパラメーターで、デフォルト値は 800 です。FAISS では、転置ファイルインデックスがデータを複数の small クラスターに分割し、各クラスターの転置リストを維持します。より大きい nprobe の値は、一般的に取得精度が高くなりますが、計算コストと検索時間が増加します。逆に、値が小さいほど精度は低下しますが、検索は高速化されます。	"faiss_nprobe" : 700
   フィーチャーストアアクセス用のプロセッサパラメーター
   fs_project	いいえ	フィーチャーストアプロジェクトの名前です。このパラメーターは、フィーチャーストアを使用する場合に必須です。詳細については、「フィーチャーストアプロジェクトの設定」をご参照ください。	"fs_project": "fs_demo"
   fs_model	いいえ	フィーチャーストア内の特徴量モデルの名前です。	"fs_model": "fs_rank_v1"
   fs_entity	いいえ	フィーチャーストア内のエンティティの名前です。	"fs_entity": "item"
   region	いいえ	フィーチャーストアサービスがデプロイされているリージョンです。例：中国 (北京) の場合は cn-beijing。リージョンの詳細については、「エンドポイント」をご参照ください。	"region": "cn-beijing"
   access_key_id	いいえ	フィーチャーストアにアクセスするための AccessKey ID です。	"access_key_id": "xxxxx"
   access_key_secret	いいえ	フィーチャーストアにアクセスするための AccessKey シークレットです。	"access_key_secret": "xxxxx"
   load_feature_from_offlinestore	いいえ	フィーチャーストアのオフラインストアからオフライン特徴量を直接ロードするかどうかを指定します。有効な値： True：フィーチャーストアのオフラインストアからデータをロードします。 False (デフォルト)：フィーチャーストアのオンラインストアからデータをロードします。	"load_feature_from_offlinestore": True
   featuredb_username	いいえ	FeatureDB のユーザー名です。	"featuredb_username":"xxx"
   featuredb_password	いいえ	FeatureDB のパスワードです。	"featuredb_password":"xxx"
   自動特徴量拡張 (INPUT_TILE)
   INPUT_TILE	いいえ	パフォーマンスを最適化するために自動特徴量拡張を有効にします。user_id のように、リクエスト内のすべてのアイテムで同じ値を持つ特徴量については、値を一度だけ送信できます。これにより、リクエストペイロードサイズ、ネットワーク転送時間、および計算が削減されます。 この機能は normal モードで、TorchEasyRec でトレーニングされたモデルと共に使用する必要があります。モデルのエクスポート中に対応する環境変数を設定する必要があります。デフォルトでは、システムはエクスポートされたモデルディレクトリ内の model_acc.json ファイルから INPUT_TILE の値を読み取ります。このファイルが存在しない場合、値は環境変数から読み取られます。 有効にした場合の動作： 環境変数の値が 2 に設定されている場合：ユーザー側特徴量の FG は一度だけ計算されます。 環境変数の値が 3 に設定されている場合：ユーザー側特徴量の FG は一度だけ計算されます。システムはユーザーとアイテムの特徴量の埋め込みを別々に計算し、ユーザー側の埋め込みも一度だけ計算されます。この設定は、ユーザー側の特徴量が多いシナリオに最適です。	"processor_envs": [ { "name": "INPUT_TILE", "value": "2" } ]
   NO_GRAD_GUARD	いいえ	推論中の勾配計算を無効にします。これにより、操作の追跡が停止し、計算グラフの構築が防がれ、パフォーマンスが向上する可能性があります。 説明 これを 1 に設定すると、一部のモデルで互換性の問題が発生する可能性があります。2 回目の推論実行中にストールが発生した場合は、PYTORCH_TENSOREXPR_FALLBACK=2 環境変数を設定して問題を解決してください。これにより、一部のグラフ最適化機能を保持しながら、コンパイルステップがバイパスされます。	"processor_envs": [ { "name": "NO_GRAD_GUARD", "value": "1" } ]
   モデルウォームアップパラメーター
   warmup_data_path	いいえ	モデルウォームアップ機能を有効にし、ウォームアップファイルが保存されるパスを指定します。データ損失を防ぐために、storage 設定でこのパスに OSS マウントを追加する必要があります。	"warmup_data_path": "/warmup"
   warmup_cnt_per_file	いいえ	各 Protobuf ファイルに対してウォームアッププロセスが実行される回数です。値が大きいほど、より徹底的なウォームアップが保証されますが、サービス起動時間が増加します。デフォルト値は 20 です。	"warmup_cnt_per_file": 20,
   warmup_pb_files_count	いいえ	次のモデルウォームアップのために Protobuf ファイルとして保存するオンラインリクエストの数です。ファイルは warmup_data_path で指定されたパスに保存されます。デフォルト値は 64 です。	"warmup_pb_files_count": 64
   低速リクエストのロギングと保存
   long_request_threshold	いいえ	低速リクエストを識別するための時間しきい値 (ミリ秒) です。リクエストの処理時間がこのしきい値を超えると、システムは各ステージの実行時間をログに自動的に記録します。デフォルト値は 200 ms です。	"long_request_threshold": 200
   save_long_request	いいえ	リクエストが long_request_threshold を超えた場合に、リクエストを Protobuf ファイルとして保存するかどうかを指定するブール値パラメーターです。デフォルト値は false です。	"save_long_request": true
   生リクエストとアイテム特徴量の OSS への保存
   request_log_path	いいえ	Protobuf ファイルを保存するためのディスクパスです。サービス設定で、この場所に OSS パスをマウントする必要があります。	"request_log_path": "/online_log_pb"
   background_feature_thread_num	いいえ	ディスクへのファイル保存専用のバックグラウンドスレッドの数です。ディスク書き込みのワークロードが重い場合は、この値を増やすことで Protobuf ファイルの保存速度を向上させることができます。デフォルト値は 4 です。	"background_feature_thread_num": 8

2. 以下のいずれかの方法で TorchEasyRec モデルサービスをデプロイします。

   JSON

   以下の手順に従ってください。

   1. PAI コンソールにログインします。ページ上部でリージョンを選択し、目的のワークスペースを選択して、Elastic Algorithm Service (EAS) をクリックします。

   2. Elastic Algorithm Service (EAS) ページで、Deploy Service をクリックします。Custom Model Deployment セクションで、JSON Deployment をクリックします。

   3. JSON エディターに JSON 設定を貼り付け、Deploy をクリックします。

   CLI

   1. クライアントをダウンロードして認証します。このトピックでは、Windows 64 ビット版を例として使用します。

   2. ターミナルを開きます。JSON ファイルを含むディレクトリで、以下のコマンドを実行してサービスを作成します。コマンドの詳細については、「コマンドリファレンス」をご参照ください。

      eascmdwin64.exe create <service.json>

      <service.json> をご利用の JSON ファイル名 (例：torcheasyrec.json) に置き換えます。

ステップ 2：サービスの呼び出し

TorchEasyRec モデルサービスのデプロイが完了したら、サービスの呼び出し情報を表示します。

1. PAI コンソールにログインし、ページ上部でターゲットリージョン、右側でワークスペースを選択し、[EAS に入る] をクリックします。

2. ターゲットサービスの Service Type 列で、Invocation Information をクリックして、サービスのエンドポイントとトークンを表示します。

TorchEasyRec モデルサービスは、入力と出力の両方に Protobuf フォーマットを使用し、FG を使用するかどうかに応じて 2 つの呼び出し方法を提供します。

package com.aliyun.openservices.eas.predict;

import com.aliyun.openservices.eas.predict.http.Compressor;
import com.aliyun.openservices.eas.predict.http.HttpConfig;
import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.proto.TorchRecPredictProtos;
import com.aliyun.openservices.eas.predict.request.TorchRecRequest;
import com.aliyun.openservices.eas.predict.proto.TorchPredictProtos.ArrayProto;

import java.util.*;


public class TorchRecPredictTest {
    public static PredictClient InitClient() {
        return new PredictClient(new HttpConfig());
    }

    public static TorchRecRequest buildPredictRequest() {
        TorchRecRequest TorchRecRequest = new TorchRecRequest();
        TorchRecRequest.appendItemId("7033");

        TorchRecRequest.addUserFeature("user_id", 33981,"int");

        ArrayList<Double> list = new ArrayList<>();
        list.add(0.24689289764507472);
        list.add(0.005758482924454689);
        list.add(0.6765301324940026);
        list.add(0.18137273055602343);
        TorchRecRequest.addUserFeature("raw_3", list,"List<double>");

        Map<String,Integer> myMap =new LinkedHashMap<>();
        myMap.put("866", 4143);
        myMap.put("1627", 2451);
        TorchRecRequest.addUserFeature("map_1", myMap,"map<string,int>");

        ArrayList<ArrayList<Float>> list2 = new ArrayList<>();
        ArrayList<Float> innerList1 = new ArrayList<>();
        innerList1.add(1.1f);
        innerList1.add(2.2f);
        innerList1.add(3.3f);
        list2.add(innerList1);
        ArrayList<Float> innerList2 = new ArrayList<>();
        innerList2.add(4.4f);
        innerList2.add(5.5f);
        list2.add(innerList2);
        TorchRecRequest.addUserFeature("click", list2,"list<list<float>>");

        TorchRecRequest.addContextFeature("id_2", list,"List<double>");
        TorchRecRequest.addContextFeature("id_2", list,"List<double>");

        System.out.println(TorchRecRequest.request);
        return TorchRecRequest;
    }

    public static void main(String[] args) throws Exception{
        PredictClient client = InitClient();
        client.setToken("tokenGeneratedFromService");
        client.setEndpoint("175805416243****.cn-beijing.pai-eas.aliyuncs.com");
        client.setModelName("alirec_rank_with_fg");
        client.setRequestTimeout(100000);


        testInvoke(client);
        testDebugLevel(client);
        client.shutdown();
    }

    public static void testInvoke(PredictClient client) throws Exception {
        long startTime = System.currentTimeMillis();
        TorchRecPredictProtos.PBResponse response = client.predict(buildPredictRequest());
        for (Map.Entry<String, ArrayProto> entry : response.getMapOutputsMap().entrySet()) {

            System.out.println("Key: " + entry.getKey() + ", Value: " + entry.getValue());
        }
        long endTime = System.currentTimeMillis();
        System.out.println("Spend Time: " + (endTime - startTime) + "ms");

    }

    public static void testDebugLevel(PredictClient client) throws Exception {
        long startTime = System.currentTimeMillis();
        TorchRecRequest request = buildPredictRequest();
        request.setDebugLevel(1);
        TorchRecPredictProtos.PBResponse response = client.predict(request);
        Map<String, String> genFeas = response.getGenerateFeaturesMap();
        for(String itemId: genFeas.keySet()) {
            System.out.println(itemId);
            System.out.println(genFeas.get(itemId));
        }
        long endTime = System.currentTimeMillis();
        System.out.println("Spend Time: " + (endTime - startTime) + "ms");

    }
}

from eas_prediction import PredictClient
from eas_prediction.torchrec_request import TorchRecRequest


if __name__ == '__main__':
    endpoint = 'http://localhost:6016'

    client = PredictClient(endpoint, '<YOUR_SERVICE_NAME>')
    client.set_token('<your_service_token>')
    client.init()
    torchrec_req = TorchRecRequest()

    torchrec_req.add_user_fea('user_id', 'u001d', "STRING")
    torchrec_req.add_user_fea('age', 12, "INT")
    torchrec_req.add_user_fea('weight', 129.8, "FLOAT")
    torchrec_req.add_item_id('item_0001')
    torchrec_req.add_item_id('item_0002')
    torchrec_req.add_item_id('item_0003')
    torchrec_req.add_user_fea("raw_3", [0.24689289764507472, 0.005758482924454689, 0.6765301324940026, 0.18137273055602343], "list<double>")
    torchrec_req.add_user_fea("raw_4", [0.9965264740966043, 0.659596586238391, 0.16396649403055896, 0.08364986620265635], "list<double>")
    torchrec_req.add_user_fea("map_1", {"0":0.37845234405201145}, "map<int,float>")
    torchrec_req.add_user_fea("map_2", {"866":4143,"1627":2451}, "map<int,int>")
    torchrec_req.add_context_fea("id_2", [866], "list<int>" )
    torchrec_req.add_context_fea("id_2", [7022,1], "list<int>" )
    torchrec_req.add_context_fea("id_2", [7022,1], "list<int>" )
    torchrec_req.add_user_fea("click", [[0.94433516,0.49145547], [0.94433516, 0.49145597]], "list<list<float>>")

    res = client.predict(torchrec_req)
    print(res)

package com.aliyun.openservices.eas.predict;

import java.util.List;
import java.util.Arrays;


import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.http.HttpConfig;
import com.aliyun.openservices.eas.predict.request.TorchDataType;
import com.aliyun.openservices.eas.predict.request.TorchRequest;
import com.aliyun.openservices.eas.predict.response.TorchResponse;

public class Test_Torch {
    public static PredictClient InitClient() {
        return new PredictClient(new HttpConfig());
    }

    public static TorchRequest buildPredictRequest() {
        TorchRequest request = new TorchRequest();
        float[] content = new float[2304000];
        for (int i = 0; i < content.length; i++) {
            content[i] = (float) 0.0;
        }
        long[] content_i = new long[900];
        for (int i = 0; i < content_i.length; i++) {
            content_i[i] = 0;
        }

        long[] a = Arrays.copyOfRange(content_i, 0, 300);
        float[] b = Arrays.copyOfRange(content, 0, 230400);
        request.addFeed(0, TorchDataType.DT_INT64, new long[]{300,3}, content_i);
        request.addFeed(1, TorchDataType.DT_FLOAT, new long[]{300,10,768}, content);
        request.addFeed(2, TorchDataType.DT_FLOAT, new long[]{300,768}, b);
        request.addFeed(3, TorchDataType.DT_INT64, new long[]{300}, a);
        request.addFetch(0);
        request.setDebugLevel(903);
        return request;
    }

    public static void main(String[] args) throws Exception {
        PredictClient client = InitClient();
        client.setToken("tokenGeneratedFromService");
        client.setEndpoint("175805416243****.cn-beijing.pai-eas.aliyuncs.com");
        client.setModelName("alirec_rank_no_fg");
        client.setIsCompressed(false);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 10; i++) {
            TorchResponse response = null;
            try {
                response = client.predict(buildPredictRequest());
                List<Float> result = response.getFloatVals(0);
                System.out.print("Predict Result: [");
                for (int j = 0; j < result.size(); j++) {
                    System.out.print(result.get(j).floatValue());
                    if (j != result.size() - 1) {
                        System.out.print(", ");
                    }
                }
                System.out.print("]\n");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("Elapsed Time: " + (endTime - startTime) + "ms");
        client.shutdown();
    }
}

from eas_prediction import PredictClient
from eas_prediction import TorchRequest

# snappy data
req = TorchRequest(False)

req.add_feed(0, [300, 3], TorchRequest.DT_INT64, [1] * 900)
req.add_feed(1, [300, 10, 768], TorchRequest.DT_FLOAT, [1.0] * 3 * 768000)
req.add_feed(2, [300, 768], TorchRequest.DT_FLOAT, [1.0] * 3 * 76800)
req.add_feed(3, [300], TorchRequest.DT_INT64, [1] * 300)


client = PredictClient('<your_endpoint>', '<your_service_name>')
client.set_token('<your_service_token>')

client.init()

resp = client.predict(req)
print(resp)

サービスから返されるステータスコードの詳細については、「サービスステータスコードの説明」をご参照ください。サービスリクエストを構築するには、「リクエストフォーマット」をご参照ください。

リクエストフォーマット

クライアントは .proto ファイルから予測リクエストコードを生成してサービスを呼び出すことができます。サービスリクエストを構築するには、以下の protobuf 定義を使用します。

syntax = "proto3";

package pytorch.eas;
option cc_enable_arenas = true;
option java_package = "com.aliyun.openservices.eas.predict.proto";
option java_outer_classname = "TorchPredictProtos";

enum ArrayDataType {
  // 無効な値。DataType フィールドが設定されていないことを示します。
  DT_INVALID = 0;
  
  // すべての計算デバイスでサポートされるデータ型。
  DT_FLOAT = 1;
  DT_DOUBLE = 2;
  DT_INT32 = 3;
  DT_UINT8 = 4;
  DT_INT16 = 5;
  DT_INT8 = 6;
  DT_STRING = 7;
  DT_COMPLEX64 = 8;  // 単精度複素数
  DT_INT64 = 9;
  DT_BOOL = 10;
  DT_QINT8 = 11;     // 量子化 int8
  DT_QUINT8 = 12;    // 量子化 uint8
  DT_QINT32 = 13;    // 量子化 int32
  DT_BFLOAT16 = 14;  // 16 ビットに切り捨てられた Float32。キャスト操作のみ。
  DT_QINT16 = 15;    // 量子化 int16
  DT_QUINT16 = 16;   // 量子化 uint16
  DT_UINT16 = 17;
  DT_COMPLEX128 = 18;  // 倍精度複素数
  DT_HALF = 19;
  DT_RESOURCE = 20;
  DT_VARIANT = 21;  // 任意の C++ データ型
}

// 配列の次元。
message ArrayShape {
  repeated int64 dim = 1 [packed = true];
}

// 配列のプロトコルバッファ。
message ArrayProto {
  // データ型。
  ArrayDataType dtype = 1;

  // 配列の形状。
  ArrayShape array_shape = 2;

  // DT_FLOAT。
  repeated float float_val = 3 [packed = true];

  // DT_DOUBLE。
  repeated double double_val = 4 [packed = true];

  // DT_INT32, DT_INT16, DT_INT8, DT_UINT8。
  repeated int32 int_val = 5 [packed = true];

  // DT_STRING。
  repeated bytes string_val = 6;

  // DT_INT64。
  repeated int64 int64_val = 7 [packed = true];

}


message PredictRequest {

  // 入力テンソル。
  repeated ArrayProto inputs = 1;

  // 出力フィルター。
  repeated int32 output_filter = 2;

  // レコメンデーション用の入力テンソル。
  map<string, ArrayProto> map_inputs = 3;

  // レコメンデーション用のデバッグレベル。
  int32 debug_level = 100;
}

// PredictRequest が成功した場合の応答。
message PredictResponse {
  // 出力テンソル。
  repeated ArrayProto outputs = 1;
  // レコメンデーション用の出力テンソル。
  map<string, ArrayProto> map_outputs = 2;
}

syntax = "proto3";

option go_package = ".;torch_predict_protos";
option java_package = "com.aliyun.openservices.eas.predict.proto";
option java_outer_classname = "TorchRecPredictProtos";
package com.alibaba.pairec.processor;
import "pytorch_predict.proto";

// int64 からのマッピング
message LongStringMap {
  map<int64, string> map_field = 1;
}
message LongIntMap {
  map<int64, int32> map_field = 1;
}
message LongLongMap {
  map<int64, int64> map_field = 1;
}
message LongFloatMap {
  map<int64, float> map_field = 1;
}
message LongDoubleMap {
  map<int64, double> map_field = 1;
}

// string からのマッピング
message StringStringMap {
  map<string, string> map_field = 1;
}
message StringIntMap {
  map<string, int32> map_field = 1;
}
message StringLongMap {
  map<string, int64> map_field = 1;
}
message StringFloatMap {
  map<string, float> map_field = 1;
}
message StringDoubleMap {
  map<string, double> map_field = 1;
}

// int32 からのマッピング
message IntStringMap {
  map<int32, string> map_field = 1;
}
message IntIntMap {
  map<int32, int32> map_field = 1;
}
message IntLongMap {
  map<int32, int64> map_field = 1;
}
message IntFloatMap {
  map<int32, float> map_field = 1;
}
message IntDoubleMap {
  map<int32, double> map_field = 1;
}

// 単一リスト
message IntList {
  repeated int32 features = 1;
}
message LongList {
  repeated int64 features  = 1;
}

message FloatList {
  repeated float features = 1;
}
message DoubleList {
  repeated double features = 1;
}
message StringList {
  repeated string features = 1;
}

// ネストされたリスト
message IntLists {
  repeated IntList lists = 1;
}
message LongLists {
  repeated LongList lists = 1;
}

message FloatLists {
  repeated FloatList lists = 1;
}
message DoubleLists {
  repeated DoubleList lists = 1;
}
message StringLists {
  repeated StringList lists = 1;
}

message PBFeature {
  oneof value {
    int32 int_feature = 1;
    int64 long_feature = 2;
    string string_feature = 3;
    float float_feature = 4;
    double double_feature=5;

    LongStringMap long_string_map = 6; 
    LongIntMap long_int_map = 7; 
    LongLongMap long_long_map = 8; 
    LongFloatMap long_float_map = 9; 
    LongDoubleMap long_double_map = 10; 
    
    StringStringMap string_string_map = 11; 
    StringIntMap string_int_map = 12; 
    StringLongMap string_long_map = 13; 
    StringFloatMap string_float_map = 14; 
    StringDoubleMap string_double_map = 15; 

    IntStringMap int_string_map = 16; 
    IntIntMap int_int_map = 17; 
    IntLongMap int_long_map = 18; 
    IntFloatMap int_float_map = 19; 
    IntDoubleMap int_double_map = 20; 

    IntList int_list = 21; 
    LongList long_list =22;
    StringList string_list = 23;
    FloatList float_list = 24;
    DoubleList double_list = 25;

    IntLists int_lists = 26;
    LongLists long_lists =27;
    StringLists string_lists = 28;
    FloatLists float_lists = 29;
    DoubleLists double_lists = 30;
    
  }
}

// コンテキスト特徴量。
message ContextFeatures {
  repeated PBFeature features = 1;
}

// アグリゲーターに送信されるリクエストを定義します。
message PBRequest {
  // デバッグモード。
  int32 debug_level = 1;

  // ユーザー特徴量。ユーザー入力名でキー付けされます。
  map<string, PBFeature> user_features = 2;

  // アイテム ID。
  repeated string item_ids = 3;

  // 各アイテムのコンテキスト特徴量。コンテキスト入力名でキー付けされます。
  map<string, ContextFeatures> context_features = 4;

  // faiss から取得する最近傍 (アイテム) の数。
  int32 faiss_neigh_num = 5;

  // 各アイテムのアイテム特徴量。アイテム入力名でキー付けされます。
  map<string, ContextFeatures> item_features = 6;
  
  // オプションのメタデータ。
  map<string, string> meta_data = 7;
}

// アグリゲーターから受信する応答を定義します。
message PBResponse {
  // Torch 出力テンソル。
  map<string, pytorch.eas.ArrayProto> map_outputs = 1;

  // FG 出力特徴量。
  map<string, string> generate_features = 2;

  // すべての FG 入力特徴量。
  map<string, string> raw_features = 3;

  // アイテム ID。
  repeated string item_ids = 4;

}

debug_level パラメーターは以下の通りです。

サービスステータスコード

このセクションでは、TorchEasyRec サービスから返される主要なステータスコードについて説明します。EAS サービスのステータスコードの完全なリストについては、「付録：サービスステータスコードと一般的なエラー」をご参照ください。

Protobuf リクエストの保存と解析

プロセッサバージョン 1.12 以降では、PAI-REC エンジンのリクエストボディで debug=True を設定してデバッグモードを有効にすると、プロセッサは特徴量分析と検証のためにデータをディスク上の protobuf ファイルに保存します。このファイルには、元のリクエストデータ、アイテム側の入力特徴量、および変換されたアイテム側の特徴量が含まれます。この機能を使用するには、request_log_path パラメーターを宛先パスに設定し、このパスを OSS 経由でマウントします。例：

"model_config": {
        "fg_mode": "normal",
        "fg_threads": 8,
        "request_log_path": "/request_log",
        "background_feature_thread_num": 8
},
 "storage": [
    {
        "mount_path": "/request_log",
        "oss": {
            "path": "oss://my-bucket/my-model/myrequests/",
            "readOnly": false
        }
    },
    {
        "mount_path": "/home/admin/docker_ml/workspace/model/",
        "oss": {
            "path": "oss://my-bucket/my-model/20260316",
            "readOnly": false
        }
    }
]

プロセッサは、request_log_path で指定されたパス内の date_hour サブディレクトリにリクエストデータを保存します。バックグラウンドスレッドがこのデータを非同期でディスクに書き込みます。model_config.background_feature_thread_num パラメーターを使用して、バックグラウンドスレッドの数を設定します。デフォルトは 4 です。この値を増やすと、書き込みスループットが向上する可能性があります。保存された protobuf ファイルは <request_id>_<random_str>.pb の形式で命名されます。OSS の書き込み帯域幅には限りがあるため、デバッグモードが有効な場合、PAI-REC エンジンからのリクエスト量は過剰にならないようにしてください。書き込み操作が遅延すると、モデルサービスの内部キューは新しいリクエストを破棄します。

protobuf ファイルを解析するには、EAS-Python-SDK バージョン 0.35 以降、または EAS-Java-SDK バージョン 2.0.29 以降が必要です。以下は Python の例です。

from eas_prediction.torchrec_predict_pb2 import PBLogData
with open('xxxx.pb', 'rb') as f:
    pb_data = f.read() 
pb_log = PBLogData()
pb_log.ParseFromString(pb_data)
print(pb_log) # ログ全体を出力
print(pb_log.request) # リクエストを出力
print(pb_log.raw_features) # 生のアイテム側特徴量を出力
print(pb_log.generate_features) # 生成されたアイテム側特徴量を出力

モデルサービスのウォームアップの有効化

モデルサービスが起動または更新されると、初期の応答時間が急上昇することがあります。これらのスパイクを防ぐために、プロセッサのウォームアップ機能を設定します。easyrec-torch-1.5 以降のバージョンでは、model_config ファイルに 3 つのパラメーターを追加することでこの機能を有効にできます。例：

"warmup_data_path": "/warmup",  # ウォームアップを有効にし、ウォームアップファイルのディレクトリを指定します。
"warmup_cnt_per_file": 20, # ファイルごとのウォームアップ反復回数を設定します。値が大きいほど、より徹底的なウォームアップが行われます。
"warmup_pb_files_count": 64 # ウォームアップ用に保存するオンラインリクエストの数を protobuf ファイルとして設定します。値が大きいほど、より多くのデータパターンをカバーできます。ファイルは `warmup_data_path` で指定されたディレクトリに保存されます。

protobuf ファイルを永続化するには、storage セクションで warmup_data_path を指す OSS マウントを設定します。例：

"storage": [
    ...,
    {
        "mount_path": "/warmup",
        "oss": {
            "path": "oss://<your-warmup-pb-file-path>",
            "readOnly": false
        }
    }
]

設定後に初めて起動すると、プロセッサは warmup_pb_files_count で指定された数のオンラインリクエストをキャプチャして保存します。その後の再起動時には、これらの保存された protobuf ファイルを使用してウォームアップします。