Platform For AI:EasyRec Processor

Pustaka FeatureGenerator baru (fg_mode=normal)

Contoh ini menggunakan tipe node PyOdps3 dengan pustaka FeatureGenerator baru. Pustaka ini mendukung operator transformasi bawaan dan kustom, tipe input kompleks (array, map), serta dependensi fitur berbasis DAG.

Contoh berikut menggunakan PAI-FeatureStore untuk manajemen data fitur. Ganti ${fs_project},${fs_model} dengan nilai aktual. Langkah 2: Buat dan terapkan layanan model EAS menyediakan prosedur lengkapnya.

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.5",
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20250722/export/final_with_fg"
      }
    }
  ],
  # Saat Anda mengubah fg_mode, metode pemanggilan juga harus diubah.
  # Jika fg_mode adalah 'normal' atau 'tf', gunakan SDK EasyRecRequest.
  # Jika fg_mode adalah 'bypass', gunakan SDK TFRequest.
  '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')

Ganti nilai featuredb_username dan featuredb_password dengan kredensial yang valid.

Versi operator TF FeatureGenerator (fg_mode=tf)

Penting: Versi operator TF FeatureGenerator hanya mendukung sekumpulan fitur bawaan terbatas: id_feature, raw_feature, combo_feature, lookup_feature, match_feature, dan sequence_feature. Operator FeatureGenerator kustom tidak didukung.

Skrip penerapan berikut menyertakan pasangan AccessKey dalam teks biasa. Skrip ini tidak menggunakan PAI-FeatureStore atau memuat data dari MaxCompute untuk mengurangi beban Hologres.

Gunakan PAI-FeatureStore dengan MaxCompute untuk penerapan di lingkungan produksi. Langkah 2: Buat dan terapkan layanan model EAS menunjukkan metode yang lebih aman menggunakan skrip Python, objek o DataWorks, dan token STS sementara dengan load_feature_from_offlinestore diatur ke True.

bizdate=$1
# Ubah metode pemanggilan berdasarkan fg_mode: EasyRecRequest untuk 'normal'/'tf', TFRequest untuk 'bypass'
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>@<endpoint>:<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.5",
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20221122/export/final_with_fg"
      }
    }
  ]
}

EOF
# Jalankan perintah penerapan.
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# Jalankan perintah pembaruan.
eascmd update ali_rec_rnk_with_fg -s echo.json

Lewati FeatureGenerator (fg_mode=bypass)

Tanpa FeatureGenerator, susun permintaan di sisi klien. Cara menggunakan EAS untuk inferensi tanpa pelatihan dengan EasyRec.

bizdate=$1
# Ubah metode pemanggilan berdasarkan fg_mode: EasyRecRequest untuk 'normal'/'tf', TFRequest untuk 'bypass'
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.5",
  "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
# Jalankan perintah penerapan.
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# Jalankan perintah pembaruan.
eascmd update ali_rec_rnk_no_fg -s echo.json

Tabel berikut menjelaskan parameter utama. Penerapan JSON mencakup parameter tambahan.

Optimasi inferensi processor EasyRec

2.1 Konfigurasi jaringan

Mesin PAI-Rec dan layanan penilaian keduanya berjalan di PAI-EAS dan memerlukan koneksi jaringan langsung. Di halaman instans PAI-EAS, klik 'VPC' untuk mengonfigurasi VPC, vSwitch, dan security group yang sama. Akses sumber daya publik atau on-premises dari EAS. Jika Anda menggunakan Hologres, konfigurasikan juga VPC yang sama. Gambar berikut menunjukkan contohnya.

2.2 Dapatkan informasi layanan

Setelah penerapan, buka halaman Elastic Algorithm Service (EAS). Temukan layanan Anda dan klik Invocation Information di kolom Service Method untuk melihat titik akhir dan token.

2.3 Contoh kode SDK

Layanan model EasyRec menggunakan Protocol Buffers (Protobuf) untuk input dan output, sehingga Anda tidak dapat mengujinya dari konsol PAI-EAS.

Sebelum memanggil layanan, pastikan fg_mode dalam model_config dari Langkah 1. Setiap mode memerlukan SDK klien yang berbeda.

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

PredictClient client = new PredictClient(new HttpConfig());
// Saat Anda mengakses layanan melalui gateway publik, gunakan titik akhir yang diawali dengan UID Anda. Anda dapat memperoleh titik akhir ini dari informasi pemanggilan layanan di konsol EAS.
client.setEndpoint("xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com");
client.setModelName("ali_rec_rnk_with_fg");
// Ganti ini dengan token layanan Anda.
client.setToken("******");

EasyRecRequest easyrecRequest = new EasyRecRequest(separator);
// userFeatures: Fitur pengguna. Fitur dipisahkan oleh \u0002 (CTRL_B). Nama fitur dan nilai dipisahkan oleh titik dua (:).
//  user_fea0:user_fea0_val\u0002user_fea1:user_fea1_val
// Untuk informasi lebih lanjut tentang format nilai fitur, lihat: https://easyrec.readthedocs.io/en/latest/feature/rtp_fg.html
easyrecRequest.appendUserFeatureString(userFeatures);
// Anda juga dapat menambahkan satu fitur pengguna sekaligus:
// easyrecRequest.addUserFeature(String userFeaName, T userFeaValue).
// Tipe data T nilai fitur dapat berupa String, float, long, atau int.

// contextFeatures: Fitur konteks. Fitur dipisahkan oleh \u0002 (CTRL_B). Nama fitur dan nilai dipisahkan oleh titik dua (:). Beberapa nilai untuk fitur yang sama juga dipisahkan oleh titik dua.
//   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);
// Anda juga dapat menambahkan satu fitur konteks sekaligus:
// easyrecRequest.addContextFeature(String ctxtFeaName, List<Object> ctxtFeaValue).
// Tipe data ctxtFeaValue dapat berupa String, Float, Long, atau Integer.

// itemIdStr: Daftar ID item untuk prediksi, dipisahkan oleh koma (,).
easyrecRequest.appendItemStr(itemIdStr, ",");
// Anda juga dapat menambahkan satu ID item sekaligus:
// 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));
    }
}

// Dapatkan fitur setelah pemrosesan FG untuk memeriksa konsistensi dengan fitur offline.
// Atur DebugLevel ke 1 untuk mengembalikan fitur yang dihasilkan.
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});
        // Catatan: Jika Anda mengatur INPUT_TILE=2, untuk fitur yang memiliki nilai sama, Anda hanya perlu mengirimkan nilai sekali:
        //    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());

        // Untuk menggunakan koneksi jaringan langsung, gunakan metode setDirectEndpoint. Contoh: 
        //   client.setDirectEndpoint("pai-eas-vpc.cn-shanghai.aliyuncs.com");
        // Anda harus mengaktifkan koneksi jaringan langsung di konsol EAS dan menyediakan vSwitch sumber yang digunakan untuk mengakses layanan EAS.
        // Koneksi jaringan langsung menawarkan stabilitas dan kinerja lebih baik.
        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" adalah bidang output model. Anda dapat menggunakan perintah curl untuk melihat input dan output model:
                //   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()
    
    # Catatan: Ganti server_default dengan signature_name aktual model Anda. Untuk informasi lebih lanjut, lihat panduan SDK yang disebutkan di atas.
    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)
    # Catatan: Setelah mengaktifkan optimasi INPUT_TILE=2, Anda dapat mengirimkan satu nilai untuk fitur-fitur di atas.
    #   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 Bangun permintaan layanan kustom

Untuk bahasa selain Python dan Java, hasilkan kode permintaan prediksi dari berkas .proto berikut:

• tf_predict.proto: Definisi permintaan untuk model 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 {
    // Bukan nilai legal untuk DataType. Digunakan untuk menunjukkan bahwa bidang DataType
    // belum diatur.
    DT_INVALID = 0;
  
    // Tipe data yang diharapkan dapat didukung oleh semua perangkat komputasi.
    DT_FLOAT = 1;
    DT_DOUBLE = 2;
    DT_INT32 = 3;
    DT_UINT8 = 4;
    DT_INT16 = 5;
    DT_INT8 = 6;
    DT_STRING = 7;
    DT_COMPLEX64 = 8;  // Bilangan kompleks presisi tunggal
    DT_INT64 = 9;
    DT_BOOL = 10;
    DT_QINT8 = 11;     // int8 terkuantisasi
    DT_QUINT8 = 12;    // uint8 terkuantisasi
    DT_QINT32 = 13;    // int32 terkuantisasi
    DT_BFLOAT16 = 14;  // Float32 dipotong menjadi 16 bit. Hanya untuk operasi cast.
    DT_QINT16 = 15;    // int16 terkuantisasi
    DT_QUINT16 = 16;   // uint16 terkuantisasi
    DT_UINT16 = 17;
    DT_COMPLEX128 = 18;  // Bilangan kompleks presisi ganda
    DT_HALF = 19;
    DT_RESOURCE = 20;
    DT_VARIANT = 21;  // Tipe data C++ arbitrer
  }
  
  // Dimensi array
  message ArrayShape {
    repeated int64 dim = 1 [packed = true];
  }
  
  // Protokol buffer yang merepresentasikan array
  message ArrayProto {
    // Tipe Data.
    ArrayDataType dtype = 1;
  
    // Bentuk 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];
  
    // DT_BOOL.
    repeated bool bool_val = 8 [packed = true];
  }
  
  // PredictRequest menentukan model TensorFlow mana yang akan dijalankan, serta
  // bagaimana input dipetakan ke tensor dan bagaimana output difilter sebelum
  // dikembalikan ke pengguna.
  message PredictRequest {
    // Signature bernama untuk dievaluasi. Jika tidak ditentukan, signature default
    // akan digunakan
    string signature_name = 1;
  
    // Tensor input.
    // Nama tensor input adalah nama alias. Pemetaan dari alias ke nama tensor input
    // sesungguhnya diharapkan disimpan sebagai signature generik bernama
    // di bawah kunci "inputs" dalam ekspor model.
    // Setiap alias yang tercantum dalam signature generik bernama "inputs" harus disediakan
    // tepat satu kali agar prediksi dapat dijalankan.
    map<string, ArrayProto> inputs = 2;
  
    // Filter output.
    // Nama yang ditentukan adalah nama alias. Pemetaan dari alias ke nama output
    // tensor sesungguhnya diharapkan disimpan sebagai signature generik bernama di bawah
    // kunci "outputs" dalam ekspor model.
    // Hanya tensor yang ditentukan di sini yang akan dijalankan/diambil dan dikembalikan, dengan
    // pengecualian bahwa ketika tidak ada yang ditentukan, semua tensor yang ditentukan dalam
    // signature bernama akan dijalankan/diambil dan dikembalikan.
    repeated string output_filter = 3;
    
    // Bendera debug
    // 0: hanya mengembalikan hasil prediksi, tanpa informasi debug
    // 100: mengembalikan hasil prediksi, dan menyimpan permintaan ke model_dir 
    // 101: menyimpan timeline ke model_dir
    int32 debug_level = 100;
  }
  
  // Respons untuk PredictRequest pada eksekusi yang berhasil.
  message PredictResponse {
    // Tensor output.
    map<string, ArrayProto> outputs = 1;
  }

• easyrec_predict.proto: Definisi permintaan untuk model TensorFlow dengan FG.

  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";
  
  // fitur konteks
  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 menentukan permintaan untuk aggregator
  message PBRequest {
    // Bendera debug
    // 0: hanya mengembalikan hasil prediksi, tanpa informasi debug
    // 3: mengembalikan fitur yang dihasilkan oleh modul FG, format string, nilai fitur dipisahkan oleh \u0002, 
    //    dapat digunakan untuk memeriksa konsistensi fitur dan menghasilkan sampel pembelajaran mendalam online 
    // 100: mengembalikan hasil prediksi, dan menyimpan permintaan ke model_dir 
    // 101: menyimpan timeline ke model_dir
    // 102: untuk model recall seperti DSSM dan MIND, tidak hanya mengembalikan hasil pengambilan Faiss
    //      tetapi juga mengembalikan vektor embedding pengguna.
    int32 debug_level = 1;
  
    // fitur pengguna
    map<string, PBFeature> user_features = 2;
  
    // id item, fitur item statis (diperbarui harian) 
    // diambil dari cache fitur yang berada di 
    // setiap node processor berdasarkan item_ids
    repeated string item_ids = 3;
  
    // fitur konteks untuk setiap item, fitur item real-time
    //    dapat dikirimkan sebagai fitur konteks.
    map<string, ContextFeatures> context_features = 4;
  
    // jumlah tetangga pengambilan penyematan.
    int32 faiss_neigh_num = 5;
  }
  
  // hasil yang dikembalikan
  message Results {
    repeated double scores = 1 [packed = true];
  }
  
  enum StatusCode {
    OK = 0;
    INPUT_EMPTY = 1;
    EXCEPTION = 2;
  }
  
  // PBResponse menentukan respons untuk aggregator
  message PBResponse {
    // hasil
    map<string, Results> results = 1;
  
    // fitur item
    map<string, string> item_features = 2;
  
    // fitur yang dihasilkan fg
    map<string, string> generate_features = 3;
  
    // fitur konteks
    map<string, ContextFeatures> context_features = 4;
  
    string error_msg = 5;
  
    StatusCode status_code = 6;
  
    // id item
    repeated string item_ids = 7;
  
    repeated string outputs = 8;
  
    // semua fitur input fg
    map<string, string> raw_features = 9;
  
    // tensor output
    map<string, ArrayProto> tf_outputs = 10;
  }