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	いいえ	FeatureStore プロジェクトが配置されているリージョンです。たとえば、中国 (北京) リージョンの場合は、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_passwd":"xxx"
   自動特徴量拡張 (input_tile) のパラメーター
   INPUT_TILE	いいえ	自動的な特徴の展開を有効にすることで、パフォーマンスを最適化します。たとえば、user_id のように、単一のリクエスト内のすべてのアイテムで同じ値を共有する特徴の場合、その値を 1 回だけ送信できます。これにより、リクエストのペイロードサイズ、ネットワーク遅延、および計算時間を削減できます。 この機能は 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) ページで、サービスのデプロイメント をクリックします。カスタムモデルのデプロイ セクションで、JSON 独立デプロイメント をクリックします。

   3. JSON エディターに JSON 設定を貼り付け、デプロイメント をクリックします。

   eascmd CLI

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

   2. JSON ファイルが含まれるディレクトリから、次のコマンドを実行してサービスを作成します。コマンドの詳細については、「コマンドリファレンス」をご参照ください。

      eascmdwin64.exe create <service.json>

      <service.json> を、お使いの JSON ファイル名に置き換えます。例: torcheasyrec.json。

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

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

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

2. 対象のサービスの サービス方式 列にある 呼び出し情報 をクリックして、サービスエンドポイントとトークンを確認します。

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 {
  // DataType の有効な値ではありません。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;

  // モデルで設定されたパススルーデータ。
  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 ファイルを使用してウォームアップします。