Platform For AI:TorchEasyRec プロセッサ

背景情報

次の図は、TorchEasyRec プロセッサに基づく推薦エンジンのアーキテクチャを示しています。

TorchEasyRec プロセッサには、次のモジュールが含まれています。

• Item Feature Cache：フィーチャーストアからのアイテム側特徴量をメモリにキャッシュします。これにより、ネットワークオーバーヘッドと フィーチャーストア へのリクエストプレッシャーが軽減されます。また、推論サービスのパフォーマンスも向上します。アイテム側特徴量にリアルタイム特徴量が含まれる場合、フィーチャーストアが同期を処理します。

• Feature Generator (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	はい	サービスストレージにマウントされる 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 の転置インデックスは、データを複数の小さなクラスター (グループ) に分割し、各クラスターの転置インデックスを維持します。大きな 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 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 のように、単一のリクエスト内で同じ値を持つ特徴量については、1 つの値を渡すだけで済みます。これにより、リクエストサイズ、ネットワーク転送時間、計算時間が削減されます。 この機能は、通常モードで TorchEasyRec と共に使用する必要があります。また、モデルをエクスポートする際に、対応する環境変数を設定する必要があります。現在、システムは TorchEasyRec からエクスポートされたモデルディレクトリ内の model_acc.json ファイルから INPUT_TILE 値を読み取ります。ファイルが存在しない場合、システムは環境変数から値を読み取ります。 この機能が有効になった後： 環境変数が 2 に設定されている場合、ユーザー側特徴量の FG は 1 回だけ計算されます。 環境変数が 3 に設定されている場合、ユーザー側特徴量の FG は 1 回だけ計算されます。システムはユーザーとアイテムの埋め込みを別々に計算し、ユーザー側の埋め込みは 1 回だけ計算されます。これは、ユーザー側特徴量が多いシナリオに適しています。	"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	いいえ	モデルウォームアップ機能を有効にし、ウォームアップファイルの保存パスを指定します。ウォームアップファイルが失われないように、ストレージ設定でこのパスに 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 ファイルとして保存するかどうかを指定します。デフォルト値は false です。	"save_long_request": true
   元のリクエストとアイテム特徴量の OSS ストレージへの書き込み
   request_log_path	いいえ	protobuf ファイルがディスクに保存されるパス。モデルサービス設定で、このパスに OSS マウントを使用します。	"request_log_path": "/online_log_pb"
   background_feature_thread_num	いいえ	ディスクへの保存というバックエンドタスクを担当するスレッド数。デフォルト値は 4 です。ディスク保存タスクが重い場合は、この値を増やすことで protobuf ファイルの保存速度を向上させることができます。	"background_feature_thread_num": 8

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

   JSON を使用したオンプレミスデプロイ (推奨)

   操作手順：

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

   2. 「Elastic Algorithm Service (EAS)」ページで、「サービスのデプロイ」をクリックし、次に「カスタムモデルのデプロイメント」セクションで、「JSON オンプレミスデプロイメント」をクリックします。

   3. JSON テキストボックスに、準備した JSON 設定コンテンツを入力し、[デプロイ] をクリックします。

   eascmd を使用したデプロイ

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

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

      eascmdwin64.exe create <service.json>

      <service.json> を、作成した JSON ファイル名 (例：torcheasyrec.json) に置き換えます。

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

TorchEasyRec モデルサービスがデプロイされた後、次の手順に従ってサービス呼び出し情報を表示します。

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

2. ターゲットサービスの [サービスメソッド] 列にある [呼び出し情報] をクリックして、サービスエンドポイントとトークン情報を表示します。

TorchEasyRec モデルサービスの入力と出力は protobuf 形式です。呼び出し方法は、FG を使用するかどうかによって異なります。

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 a DataType field
  // has not been set.
  DT_INVALID = 0;
  
  // Data types that all computation devices are expected to be
  // capable 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];
}

// Protocol buffer representing an array
message ArrayProto {
  // Data Type.
  ArrayDataType dtype = 1;

  // Shape of the array.
  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 rec
  map<string, ArrayProto> map_inputs = 3;

  // debug_level for rec
  int32 debug_level = 100;
}

// Response for PredictRequest on successful run.
message PredictResponse {
  // Output tensors.
  repeated ArrayProto outputs = 1;
  // Output tensors for rec.
  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";

//long->others
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->others
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->others
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;
}

// list
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;
}

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

// PBRequest specifies the request for aggregator
message PBRequest {
  // debug mode
  int32 debug_level = 1;

  // user features, key is user input name
  map<string, PBFeature> user_features = 2;

  // item ids
  repeated string item_ids = 3;

  // context features for each item, key is context input name 
  map<string, ContextFeatures> context_features = 4;

  // number of nearest neighbors(items) to retrieve
  // from faiss
  int32 faiss_neigh_num = 5;

  // item features for each item, key is item input name 
  map<string, ContextFeatures> item_features = 6;
  
  // optional meta data
  map<string, string> meta_data = 7;
}

// PBResponse specifies the response for aggregator
message PBResponse {
  // torch output tensors
  map<string, pytorch.eas.ArrayProto> map_outputs = 1;

  // fg ouput features
  map<string, string> generate_features = 2;

  // all fg input features
  map<string, string> raw_features = 3;

  // item ids
  repeated string item_ids = 4;

}

次の表に debug_level を示します。

サービスステータスコードの説明

次の表は、TorchEasyRec サービスにアクセスする際に返される可能性のある主なステータスコードを示しています。EAS サービスへのアクセス時に返されるステータスコードの詳細については、「付録：サービスステータスコードと一般的なエラー」をご参照ください。

リクエスト pb ファイルの保存と解析

プロセッサバージョン 1.12 以降では、PAI-Rec エンジンのリクエストボディで debug=True が有効になっている場合、プロセッサは元のリクエストとアイテム側の入力および出力特徴量をディスク上の protobuf ファイルに保存します。これにより、後続の特徴量の分析と検証がサポートされます。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) # 特徴量生成 (fg) 後のアイテム側特徴量を出力