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 サービス設定ファイルを準備します。

   プロセッサタイプを 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	はい	モデルファイルを保存するためにサービスにマウントされる 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 パラメーターは、取得時に検索するクラスターの数を指定します。FAISS では、転置ファイルインデックスがデータをより小さなクラスターに分割し、各クラスターの転置リストを維持します。大きな nprobe 値は、計算量と検索時間が増加する代わりに、一般的に再現率の精度を向上させます。一方、小さな値は精度を低下させますが、検索を高速化します。デフォルト値は 800 です。	"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 Secret。	"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_passwd":"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。	"long_request_threshold": 200
   save_long_request	いいえ	long_request_threshold を超えるリクエストを Protobuf ファイルとして保存するかどうかを指定します。true に設定すると、ファイルはモデルディレクトリ以下の torch_req ディレクトリに保存されます。デフォルト値：false。	"save_long_request": true
   生リクエストとアイテム特徴量の OSS への保存
   request_log_path	いいえ	Protobuf ファイルを保存するためのディスクパス。サービス設定でこの場所に OSS パスをマウントする必要があります。	"request_log_path": "/online_log_pb"
   background_feature_thread_num	いいえ	ファイルをディスクに書き込むための専用のバックグラウンドスレッド数。ディスク書き込みのワークロードが重い場合、この値を増やすことで保存プロセスを高速化できます。デフォルト値：4。	"background_feature_thread_num": 8
   パススルーデータ設定
   pass_through_data	いいえ	応答にパススルーするデータを指定します。これは、下流サービスに情報を渡すのに役立ちます。値は JSON オブジェクトでなければなりません。	"pass_through_data": {"model_version": "20260513"}

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 をクリックします。

   eascmd CLI

   1. クライアントをダウンロードして認証します。以下の手順では、64 ビット Windows バージョンを例として使用します。

   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("Spend 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 {
  // Not a legal value for DataType. Used to indicate that a DataType field
  // has not been set.
  DT_INVALID = 0;
  
  // Data types that all computation devices are expected to support.
  DT_FLOAT = 1;
  DT_DOUBLE = 2;
  DT_INT32 = 3;
  DT_UINT8 = 4;
  DT_INT16 = 5;
  DT_INT8 = 6;
  DT_STRING = 7;
  DT_COMPLEX64 = 8;  // Single-precision complex
  DT_INT64 = 9;
  DT_BOOL = 10;
  DT_QINT8 = 11;     // Quantized int8
  DT_QUINT8 = 12;    // Quantized uint8
  DT_QINT32 = 13;    // Quantized int32
  DT_BFLOAT16 = 14;  // Float32 truncated to 16 bits. Only for cast ops.
  DT_QINT16 = 15;    // Quantized int16
  DT_QUINT16 = 16;   // Quantized uint16
  DT_UINT16 = 17;
  DT_COMPLEX128 = 18;  // Double-precision complex
  DT_HALF = 19;
  DT_RESOURCE = 20;
  DT_VARIANT = 21;  // Arbitrary C++ data types
}

// Dimensions of an array.
message ArrayShape {
  repeated int64 dim = 1 [packed = true];
}

// Represents an array.
message ArrayProto {
  // Data type.
  ArrayDataType dtype = 1;

  // Array shape.
  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 {

  // Input tensors.
  repeated ArrayProto inputs = 1;

  // Output filter.
  repeated int32 output_filter = 2;

  // Input tensors for the recommendation model.
  map<string, ArrayProto> map_inputs = 3;

  // Debug level for the recommendation model.
  int32 debug_level = 100;
}

// Response for a successful PredictRequest.
message PredictResponse {
  // Output tensors.
  repeated ArrayProto outputs = 1;
  // Output tensors from the recommendation model.
  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";

// Mappings from 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;
}

// Mappings from 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;
}

// Mappings from 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;
}

// Single-level lists.
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;
}

// Nested lists.
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;
    
  }
}

// Context features.
message ContextFeatures {
  repeated PBFeature features = 1;
}

// Defines the request sent to the aggregator.
message PBRequest {
  // Debug level.
  int32 debug_level = 1;

  // User features, keyed by the input name.
  map<string, PBFeature> user_features = 2;

  // Item IDs.
  repeated string item_ids = 3;

  // Context features for each item, keyed by the input name. 
  map<string, ContextFeatures> context_features = 4;

  // The number of nearest neighbors to retrieve from Faiss.
  int32 faiss_neigh_num = 5;

  // Item features for each item, keyed by the input name. 
  map<string, ContextFeatures> item_features = 6;
  
  // Optional metadata.
  map<string, string> meta_data = 7;
}

// Defines the response from the aggregator.
message PBResponse {
  // Torch output tensors.
  map<string, pytorch.eas.ArrayProto> map_outputs = 1;

  // Output features from the feature generator (FG).
  map<string, string> generate_features = 2;

  // All input features for the feature generator (FG).
  map<string, string> raw_features = 3;

  // Item IDs.
  repeated string item_ids = 4;

  // Pass-through data configured in the model.
  map<string, string> pass_through_data = 5;
}

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 ファイルを永続化するには、storage セクションの warmup_data_path に OSS マウントを設定します。例：

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

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