Platform For AI:EasyRec プロセッサ

例：新規 FeatureGenerator ライブラリ (fg_mode=normal)

以下の例は、PyOdps3 ノードタイプを使用したデプロイ方法を示しています。このモードでは、新しいバージョンの FeatureGenerator を使用でき、豊富な組み込み特徴量変換演算子を提供し、FeatureGenerator 演算子のカスタマイズを可能にし、配列やマップなどの複雑な型の入力特徴量をサポートし、有向非循環グラフ (DAG) モードで特徴量間の依存関係を処理します。

この例では、PAI-FeatureStore を使用して特徴量データを管理しています。スクリプト内で、${fs_project} and ${fs_model} 変数を実際の値に置き換えてください。詳細については、「ステップ 2：PAI-EAS モデルサービスの作成およびデプロイ」をご参照ください。

import json
import os

service_name = 'ali_rec_rnk_with_fg'

config = {
  'name': service_name,
  'metadata': {
    "cpu": 8,
    #"cuda": "11.2",
    "gateway": "default",
    "gpu": 0,
    "memory": 32000,
    "rolling_strategy": {
        "max_unavailable": 1
    },
    "rpc": {
        "enable_jemalloc": 1,
        "max_queue_size": 256
    }
  },
  "processor_envs": [
    {
      "name": "ADAPTE_FG_CONFIG",
      "value": "true"
    }
  ],
  "model_path": "",
  "processor": "easyrec-3.1",
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20250722/export/final_with_fg"
      }
    }
  ],
  # fg_mode を変更する場合、呼び出し方法も変更する必要があります。
  # fg_mode が normal または tf に設定されている場合は、EasyRecRequest SDK を使用します。
  # fg_mode が bypass に設定されている場合は、TFRequest SDK を使用します。
  'model_config': {
    'outputs': 'probs_ctr,probs_cvr',
    'fg_mode': 'normal',
    'steady_mode': True,
    'period': 2880,
    'access_key_id': f'{o.account.access_id}',
    'access_key_secret': f'{o.account.secret_access_key}',
    "load_feature_from_offlinestore": True,
    'region': 'cn-shanghai',
    'fs_project': '${fs_project}',
    'fs_model': '${fs_model}',
    'fs_entity': 'item',
    'featuredb_username': 'guest',
    'featuredb_password': '123456',
    'log_iterate_time_threshold': 100,
    'iterate_featuredb_interval': 5,
    'mc_thread_pool_num': 1,
  }
}

with open('echo.json', 'w') as output_file:
    json.dump(config, output_file)

os.system(f'/home/admin/usertools/tools/eascmd -i {o.account.access_id} -k {o.account.secret_access_key} -e pai-eas.cn-shanghai.aliyuncs.com create echo.json')
# os.system(f'/home/admin/usertools/tools/eascmd -i {o.account.access_id} -k {o.account.secret_access_key} -e pai-eas.cn-shanghai.aliyuncs.com modify {service_name} -s echo.json')

注：featuredb_username および featuredb_password パラメーターの値を、有効なユーザー名およびパスワードに置き換えてください。

例：FG の TF 演算子モード (fg_mode=tf)

注：FeatureGenerator の TF 演算子バージョンは、限定された組み込み特徴量のみをサポートしています：id_feature, raw_feature, combo_feature, lookup_feature, match_feature, sequence_feature。カスタム FeatureGenerator 演算子はサポートされていません。

以下の例は、シェルスクリプトを使用したデプロイ方法を示しています。スクリプトには AccessKeyID および AccessKeySecret がプレーンテキストで含まれています。この方法はシンプルですが、PAI-FeatureStore を使用しておらず、Hologres への負荷を軽減するために MaxCompute からテーブルデータをロードする方法についても説明していません。

PAI-FeatureStore を使用し、MaxCompute からデータをロードすることを推奨します。詳細については、「ステップ 2：PAI-EAS モデルサービスの作成およびデプロイ」をご参照ください。参照ドキュメントでは、Python スクリプト、DataWorks 組み込みオブジェクト o、および一時的な Security Token Service (STS) トークンを使用した、より安全なデプロイ方法を紹介しています。その例では、load_feature_from_offlinestore が True に設定されています。

bizdate=$1
# fg_mode を変更する場合、呼び出し方法も変更する必要があります。fg_mode が normal または tf に設定されている場合は、EasyRecRequest SDK を使用します。fg_mode が bypass に設定されている場合は、TFRequest SDK を使用します。
cat << EOF > echo.json
{
  "name":"ali_rec_rnk_with_fg",
  "metadata": {
    "instance": 2,
    "rpc": {
      "enable_jemalloc": 1,
      "max_queue_size": 100
    }
  },
  "cloud": {
    "computing": {
      "instance_type": "ecs.g7.large",
      "instances": null
    }
  },
  "model_config": {
    "remote_type": "hologres",
    "url": "postgresql://<AccessKeyID>:<AccessKeySecret>@<domain_name>:<port>/<database>",
    "tables": [{"name":"<schema>.<table_name>","key":"<index_column_name>","value": "<column_name>"}],
    "period": 2880,
    "fg_mode": "tf",
    "outputs":"probs_ctr,probs_cvr",
  },
  "model_path": "",
  "processor": "easyrec-3.1",
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20221122/export/final_with_fg"
      }
    }
  ]
}

EOF
# デプロイコマンドを実行します。
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# 更新コマンドを実行します。
eascmd update ali_rec_rnk_with_fg -s echo.json

例: FeatureGenerator を使用しないでください (fg_mode=bypass)

FeatureGenerator を使用しない場合は、EasyRec プロセッサを呼び出す前にリクエストをアセンブルする必要があります。詳細については、「EasyRec トレーニングを使用せずに PAI-EAS で推論サービスを統合する」をご参照ください。

bizdate=$1
# fg_mode を変更する場合、呼び出し方法も変更する必要があります。fg_mode が normal または tf に設定されている場合は、EasyRecRequest SDK を使用します。fg_mode が bypass に設定されている場合は、TFRequest SDK を使用します。
cat << EOF > echo.json
{
  "name":"ali_rec_rnk_no_fg",
  "metadata": {
    "instance": 2,
    "rpc": {
      "enable_jemalloc": 1,
      "max_queue_size": 100
    }
  },
  "cloud": {
    "computing": {
      "instance_type": "ecs.g7.large",
      "instances": null
    }
  },
  "model_config": {
    "fg_mode": "bypass"
  },
  "processor": "easyrec-3.1",
  "processor_envs": [
    {
      "name": "INPUT_TILE",
      "value": "2"
    }
  ],
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20221122/export/final/"
      }
    }
  ],
  "warm_up_data_path": "oss://easyrec/ali_rec_sln_acc_rnk/rnk_warm_up.bin"
}

EOF
# デプロイコマンドを実行します。
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# 更新コマンドを実行します。
eascmd update ali_rec_rnk_no_fg -s echo.json

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

EasyRec プロセッサの推論最適化パラメーター

2.1 ネットワーク構成

PAI-Rec エンジンおよびモデルスコアリングサービスは、PAI EAS 上にデプロイされます。この構成には、直接的なネットワーク接続が必要です。PAI EAS インスタンスページで右上隅の VPC をクリックし、サービスが同じ VPC、vSwitch、およびセキュリティグループを使用するように構成します。詳細については、「EAS からのパブリックリソースまたは内部リソースへのアクセス」をご参照ください。Hologres を使用する場合は、同じ VPC 設定を使用していることを確認してください。次の図に例を示します。

2.2 サービス情報の取得

EasyRec モデルサービスをデプロイした後、Elastic Algorithm Service (EAS) ページに移動します。サービス方法 列で目的のサービスの 呼び出し方法 をクリックし、アクセスエンドポイントおよびトークンを確認します。

2.3 SDK 呼び出しの例

EasyRec モデルサービスは、入力および出力の両方に Protocol Buffers (Protobuf) 形式を使用します。そのため、PAI EAS コンソールから直接サービスをテストすることはできません。

サービスを呼び出す前に、ステップ 1 のデプロイ時に model_config で設定した fg_mode を特定する必要があります。モードによって、クライアントの呼び出し方法が完全に異なります。

import com.aliyun.openservices.eas.predict.http.*;
import com.aliyun.openservices.eas.predict.request.EasyRecRequest;

PredictClient client = new PredictClient(new HttpConfig());
// パブリックゲートウェイ経由でサービスにアクセスするには、ユーザー ID (UID) で始まるエンドポイントを使用します。この情報は、EAS コンソールのサービス呼び出し詳細で確認できます。
client.setEndpoint("xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com");
client.setModelName("ali_rec_rnk_with_fg");
// ご利用のサービストークンに置き換えてください。
client.setToken("******");

EasyRecRequest easyrecRequest = new EasyRecRequest(separator);
// userFeatures: ユーザー特徴量。特徴量は \u0002 (CTRL_B) で区切られ、特徴量名と値はコロン (:) で区切られます。
//  user_fea0:user_fea0_val\u0002user_fea1:user_fea1_val
// 特徴量値の形式については、https://easyrec.readthedocs.io/latest/feature/rtp_fg.html をご参照ください。
easyrecRequest.appendUserFeatureString(userFeatures);
// 1 つのユーザー特徴量を個別に追加することもできます：
// easyrecRequest.addUserFeature(String userFeaName, T userFeaValue)。
// 特徴量値の型 T は、String、float、long、または int です。

// contextFeatures: コンテキスト特徴量。特徴量は \u0002 (CTRL_B) で区切られます。特徴量名と値はコロン (:) で区切られます。単一の特徴量に対する複数の値もコロンで区切られます。
//   ctxt_fea0:ctxt_fea0_ival0:ctxt_fea0_ival1:ctxt_fea0_ival2\u0002ctxt_fea1:ctxt_fea1_ival0:ctxt_fea1_ival1:ctxt_fea1_ival2
easyrecRequest.appendContextFeatureString(contextFeatures);
// 1 つのコンテキスト特徴量を個別に追加することもできます：
// easyrecRequest.addContextFeature(String ctxtFeaName, List<Object> ctxtFeaValue)。
// ctxtFeaValue の型は、String、Float、Long、または Integer です。

// itemIdStr: 予測対象のアイテム ID リスト。カンマ (,) で区切られます。
easyrecRequest.appendItemStr(itemIdStr, ",");
// 1 つのアイテム ID を個別に追加することもできます：
// easyrecRequest.appendItemId(String itemId)

easyrecPredictProtos.PBResponse response = client.predict(easyrecRequest);

for (Map.Entry<String, easyrecPredictProtos.Results> entry : response.getResultsMap().entrySet()) {
    String key = entry.getKey();
    easyrecPredictProtos.Results value = entry.getValue();
    System.out.print("key: " + key);
    for (int i = 0; i < value.getScoresCount(); i++) {
        System.out.format("value: %.6g\n", value.getScores(i));
    }
}

// FG によって処理された特徴量を取得し、オフライン特徴量との一貫性を確認します。
// DebugLevel を 1 に設定すると、生成された特徴量が返されます。
easyrecRequest.setDebugLevel(1);
easyrecPredictProtos.PBResponse response = client.predict(easyrecRequest);
Map<String, String> genFeas = response.getGenerateFeaturesMap();
for(String itemId: genFeas.keySet()) {
    System.out.println(itemId);
    System.out.println(genFeas.get(itemId));
}

from eas_prediction import PredictClient

from eas_prediction.easyrec_request import EasyRecRequest
from eas_prediction.easyrec_predict_pb2 import PBFeature
from eas_prediction.easyrec_predict_pb2 import PBRequest

if __name__ == '__main__':
    endpoint = 'http://xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com'
    service_name = 'ali_rec_rnk_with_fg'
    token = '******'

    client = PredictClient(endpoint, service_name)
    client.set_token(token)
    client.init()

    req = PBRequest()
    uid = PBFeature()
    uid.string_feature = 'u0001'
    req.user_features['user_id'] = uid
    age = PBFeature()
    age.int_feature = 12
    req.user_features['age'] = age
    weight = PBFeature()
    weight.float_feature = 129.8
    req.user_features['weight'] = weight

    req.item_ids.extend(['item_0001', 'item_0002', 'item_0003'])
    
    easyrec_req = EasyRecRequest()
    easyrec_req.add_feed(req, debug_level=0)
    res = client.predict(easyrec_req)
    print(res)

import java.util.List;

import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.http.HttpConfig;
import com.aliyun.openservices.eas.predict.request.TFDataType;
import com.aliyun.openservices.eas.predict.request.TFRequest;
import com.aliyun.openservices.eas.predict.response.TFResponse;

public class TestEasyRec {
    public static TFRequest buildPredictRequest() {
        TFRequest request = new TFRequest();
 
        request.addFeed("user_id", TFDataType.DT_STRING, 
                        new long[]{3}, new String []{ "u0001", "u0001", "u0001"});
      	request.addFeed("age", TFDataType.DT_FLOAT, 
                        new long[]{3}, new float []{ 18.0f, 18.0f, 18.0f});
        // 注：INPUT_TILE=2 を設定した場合、すべてのアイテムで同じ値を持つ特徴量は 1 回だけ渡す必要があります：
        //    request.addFeed("user_id", TFDataType.DT_STRING,
        //            new long[]{1}, new String []{ "u0001" });
        //    request.addFeed("age", TFDataType.DT_FLOAT, 
        //            new long[]{1}, new float []{ 18.0f});
      	request.addFeed("item_id", TFDataType.DT_STRING, 
                        new long[]{3}, new String []{ "i0001", "i0002", "i0003"});  
        request.addFetch("probs");
      	return request;
    }

    public static void main(String[] args) throws Exception {
        PredictClient client = new PredictClient(new HttpConfig());

        // 直接ネットワーク接続を使用するには、setDirectEndpoint メソッドを呼び出します。例：
        //   client.setDirectEndpoint("pai-eas-vpc.cn-shanghai.aliyuncs.com");
        // 直接ネットワーク接続は、EAS コンソールで有効にする必要があります。EAS サービスにアクセスするために使用するソース vSwitch を指定する必要があります。
        // 直接ネットワーク接続により、安定性とパフォーマンスが向上します。
        client.setEndpoint("xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com");
        client.setModelName("ali_rec_rnk_no_fg");
        client.setToken("");
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            try {
                TFResponse response = client.predict(buildPredictRequest());
                // 'probs' はモデルの出力フィールド名です。curl コマンドを使用してモデルの入力および出力を確認できます：
                //   curl xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com -H "Authorization:{token}"
                List<Float> result = response.getFloatVals("probs");
                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();
    }
}

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import StringRequest
from eas_prediction import TFRequest

if __name__ == '__main__':
    client = PredictClient('http://xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com', 'ali_rec_rnk_no_fg')
    client.set_token('')
    client.init()
    
    # 注：'server_default' をご利用のモデルの実際の signature_name に置き換えてください。詳細については、SDK ドキュメントをご参照ください。
    req = TFRequest('server_default') 
    req.add_feed('user_id', [3], TFRequest.DT_STRING, ['u0001'] * 3)
    req.add_feed('age', [3], TFRequest.DT_FLOAT, [18.0] * 3)
    # 注：INPUT_TILE=2 最適化を有効にした後、すべてのアイテムで同じ値を持つ特徴量は 1 回だけ渡す必要があります。
    #   req.add_feed('user_id', [1], TFRequest.DT_STRING, ['u0001'])
    #   req.add_feed('age', [1], TFRequest.DT_FLOAT, [18.0])
    req.add_feed('item_id', [3], TFRequest.DT_STRING, 
        ['i0001', 'i0002', 'i0003'])
    for x in range(0, 100):
        resp = client.predict(req)
        print(resp)

2.4 カスタムサービスリクエストの構築

Python または Java 以外の言語を使用してサービスを呼び出すには、.proto ファイルから予測リクエストコードを手動で生成する必要があります。以下の Protocol Buffers 定義を使用して、必要なコードを生成できます。

• tf_predict.proto：TensorFlow モデルのリクエスト定義

  syntax = "proto3";
  
  option cc_enable_arenas = true;
  option go_package = ".;tf";
  option java_package = "com.aliyun.openservices.eas.predict.proto";
  option java_outer_classname = "PredictProtos";
  
  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];
  
    // DT_BOOL。
    repeated bool bool_val = 8 [packed = true];
  }
  
  // PredictRequest は、実行する TensorFlow モデル、
  // 入力がテンソルにどのようにマッピングされるか、
  // 出力がユーザーに返される前にどのようにフィルタリングされるかを指定します。
  message PredictRequest {
    // 評価する名前付き署名。指定しない場合、デフォルトの署名が使用されます。
    string signature_name = 1;
  
    // 入力テンソル。
    // 入力テンソルの名前はエイリアス名です。エイリアスから実際の入力テンソル名へのマッピングは、
    // モデルエクスポート内のキー "inputs" の下に名前付き汎用署名として格納されていることが期待されます。
    // "inputs" という名前の汎用署名にリストされた各エイリアスは、
    // 予測を実行するために正確に 1 回提供する必要があります。
    map<string, ArrayProto> inputs = 2;
  
    // 出力フィルター。
    // 指定された名前はエイリアス名です。エイリアスから実際の出力テンソル名へのマッピングは、
    // モデルエクスポート内のキー "outputs" の下に名前付き汎用署名として格納されていることが期待されます。
    // ここで指定されたテンソルのみが実行/フェッチされ、返されます。何も指定しない場合、
    // 名前付き署名で指定されたすべてのテンソルが実行/フェッチされ、返されます。
    repeated string output_filter = 3;
    
    // デバッグフラグ
    // 0: 予測結果のみを返し、デバッグ情報を返しません。
    // 100: 予測結果を返し、リクエストを model_dir に保存します。
    // 101: タイムラインを model_dir に保存します。
    int32 debug_level = 100;
  }
  
  // 成功した PredictRequest の応答。
  message PredictResponse {
    // 出力テンソル。
    map<string, ArrayProto> outputs = 1;
  }

• easyrec_predict.proto：FG を含む TensorFlow モデルのリクエスト定義

  syntax = "proto3";
  
  option cc_enable_arenas = true;
  option go_package = ".;easyrec";
  option java_package = "com.aliyun.openservices.eas.predict.proto";
  option java_outer_classname = "EasyRecPredictProtos";
  
  import "tf_predict.proto";
  
  // コンテキスト特徴量
  message ContextFeatures {
    repeated PBFeature features = 1;
  }
  
  message PBFeature {
    oneof value {
      int32 int_feature = 1;
      int64 long_feature = 2;
      string string_feature = 3;
      float float_feature = 4;
    }
  }
  
  // PBRequest はアグリゲーターへのリクエストを指定します。
  message PBRequest {
    // デバッグフラグ
    // 0: 予測結果のみを返し、デバッグ情報を返しません。
    // 3: FG モジュールによって生成された特徴量を文字列形式で返します。特徴量の値は \u0002 で区切られます。
    //    これは、特徴量の一貫性を確認したり、オンラインディープ ラーニング用サンプルを生成したりするために使用できます。
    // 100: 予測結果を返し、リクエストを model_dir に保存します。
    // 101: タイムラインを model_dir に保存します。
    // 102: DSSM や MIND などのリコールモデルの場合、Faiss による検索結果だけでなく、ユーザーの埋め込みベクターも返します。
    int32 debug_level = 1;
  
    // ユーザー特徴量
    map<string, PBFeature> user_features = 2;
  
    // アイテム ID。静的（毎日更新される）アイテム特徴量は、各プロセッサノード上に存在する特徴量キャッシュから item_ids を使用してフェッチされます。
    repeated string item_ids = 3;
  
    // 各アイテムのコンテキスト特徴量。リアルタイムアイテム特徴量は、コンテキスト特徴量として渡すことができます。
    map<string, ContextFeatures> context_features = 4;
  
    // 埋め込み取得のための近傍数。
    int32 faiss_neigh_num = 5;
  }
  
  // 戻り値
  message Results {
    repeated double scores = 1 [packed = true];
  }
  
  enum StatusCode {
    OK = 0;
    INPUT_EMPTY = 1;
    EXCEPTION = 2;
  }
  
  // PBResponse はアグリゲーターへの応答を指定します。
  message PBResponse {
    // 結果
    map<string, Results> results = 1;
  
    // アイテム特徴量
    map<string, string> item_features = 2;
  
    // FG によって生成された特徴量
    map<string, string> generate_features = 3;
  
    // コンテキスト特徴量
    map<string, ContextFeatures> context_features = 4;
  
    string error_msg = 5;
  
    StatusCode status_code = 6;
  
    // アイテム ID
    repeated string item_ids = 7;
  
    repeated string outputs = 8;
  
    // すべての FG 入力特徴量
    map<string, string> raw_features = 9;
  
    // 出力テンソル
    map<string, ArrayProto> tf_outputs = 10;
  }