Platform For AI:EasyRec processor (recommendation scoring service)

Example using the new FG library (fg_mode=normal)

This example uses the PyOdps3 node type for deployment. This mode supports the new FeatureGenerator with built-in feature transformation operators and custom FG operators. It supports complex input types (array and map) and feature dependencies in Directed Acyclic Graph (DAG) mode.

This example uses PAI-FeatureStore to manage feature data. In the script, you must replace the ${fs_project},${fs_model} variables with your actual values. For more information, see Step 2: Create and deploy an EAS model service.

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-2.9",
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20250722/export/final_with_fg"
      }
    }
  ],
  # When you change fg_mode, you must also change the corresponding invocation method.
  # If fg_mode is normal or tf, use the EasyRecRequest SDK for invocation.
  # If fg_mode is bypass, use the TFRequest SDK for invocation.
  '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')

Example using the TF OP version of FG (fg_mode=tf)

Note: The TF OP version of FG supports only these built-in features: id_feature, raw_feature, combo_feature, lookup_feature, match_feature, and sequence_feature. Custom FG operators are not supported.

The following example uses a shell script for deployment. The script contains the AccessKey ID and AccessKey secret in plaintext. This method is simple and easy to understand, but it does not include PAI-FeatureStore or explain how to load table data from MaxCompute to reduce the load on Hologres. For information about using PAI-FeatureStore and loading data from MaxCompute, see Step 2: Create and deploy an EAS model service. Note that the referenced document uses a Python script for deployment, which uses the DataWorks built-in object o and a temporary Security Token Service (STS) token for improved security, and sets load_feature_from_offlinestore to True.

bizdate=$1
# When you change fg_mode, you must also change the corresponding invocation method. If fg_mode is normal or tf, use the EasyRecRequest SDK. If fg_mode is bypass, use the 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
# Execute the deployment command.
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# Execute the update command.
eascmd update ali_rec_rnk_with_fg -s echo.json

Example without FG (fg_mode=bypass)

bizdate=$1
# When you change fg_mode, you must also change the corresponding invocation method. If fg_mode is normal or tf, use the EasyRecRequest SDK. If fg_mode is bypass, use the 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
# Execute the deployment command.
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# Execute the update command.
eascmd update ali_rec_rnk_no_fg -s echo.json

Key parameters. For other parameters, see JSON deployment.

Inference optimization parameters for EasyRecProcessor

2.1 Network configuration

The PAI-Rec engine and the model scoring service are both deployed on PAI EAS. Therefore, the products must be configured for a direct network connection. On the PAI EAS instance page, as shown in the following figure, click VPC in the upper-right corner. Set the same VPC, vSwitch, and security group. For more information, see Enable EAS to access the internet and private resources. If you use Hologres, you must also configure the same VPC information.

2.2 Get service information

After you deploy the EasyRec model service, go to the Elastic Algorithm Service (EAS) page. In the Service Method column, find the service that you want to call and click Invocation Method to view its endpoint and token information.

2.3 SDK code examples for invocation

The input and output formats for the EasyRec model service are both protobuf. Therefore, you cannot test the service on the PAI EAS product page.

Before you call the service, you must identify the fg_mode that was set in the model_config during deployment in Step 1. Different modes require different client invocation methods.

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

PredictClient client = new PredictClient(new HttpConfig());
// When accessing through a public gateway, use the Endpoint that starts with the user UID. You can get this information from the service's invocation details in the EAS console.
client.setEndpoint("xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com");
client.setModelName("ali_rec_rnk_with_fg");
// Replace with the service token information.
client.setToken("******");

EasyRecRequest easyrecRequest = new EasyRecRequest(separator);
// userFeatures: User features. Features are separated by \u0002 (CTRL_B). The feature name and value are separated by a colon (:).
//  user_fea0:user_fea0_val\u0002user_fea1:user_fea1_val
// For the format of feature values, see: https://easyrec.readthedocs.io/en/latest/feature/rtp_fg.html
easyrecRequest.appendUserFeatureString(userFeatures);
// You can also add one user feature at a time:
// easyrecRequest.addUserFeature(String userFeaName, T userFeaValue).
// The type T of the feature value can be: String, float, long, int.

// contextFeatures: Context features. Features are separated by \u0002 (CTRL_B). The feature name and value are separated by a colon (:). Multiple feature values are separated by colons (:).
//   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);
// You can also add one context feature at a time:
// easyrecRequest.addContextFeature(String ctxtFeaName, List<Object> ctxtFeaValue).
// The type of ctxtFeaValue can be: String, Float, Long, Integer.

// itemIdStr: A list of itemIds to predict, separated by a comma (,).
easyrecRequest.appendItemStr(itemIdStr, ",");
// You can also add one itemId at a time:
// easyrecRequest.appendItemId(String itemId)

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

for (Map.Entry<String, PredictProtos.Results> entry : response.getResultsMap().entrySet()) {
    String key = entry.getKey();
    PredictProtos.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));
    }
}

// Get the features after FG processing to compare them with offline features for consistency.
// Set DebugLevel to 1 to return the generated features.
easyrecRequest.setDebugLevel(1);
PredictProtos.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});
        // Note: If you set INPUT_TILE=2, you can pass the same value for the feature only once:
        //    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());

        // To use the direct network connection feature, use the setDirectEndpoint method, for example: 
        //   client.setDirectEndpoint("pai-eas-vpc.cn-shanghai.aliyuncs.com");
        // The direct network connection must be enabled in the EAS console. Provide the source vSwitch for accessing the EAS service.
        // Direct network connection offers better stability and performance.
        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 is the output field name of the model. You can use the curl command to view the model's input and output:
                //   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()
    
    # Note: Replace server_default with the actual signature_name of your model. For details, see the SDK instructions mentioned above.
    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)
    # Note: After enabling the INPUT_TILE=2 optimization, you can pass the value for the above features only once.
    #   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 Build a custom service request

For other languages, manually generate prediction request code from the .proto file. Use these protobuf definitions:

• tf_predict.proto: Request definition for a TensorFlow model

  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 {
    // 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];
  
    // DT_BOOL.
    repeated bool bool_val = 8 [packed = true];
  }
  
  // PredictRequest specifies which TensorFlow model to run, along with
  // how inputs are mapped to tensors and how outputs are filtered before
  // returning to user.
  message PredictRequest {
    // A named signature to evaluate. If unspecified, the default signature
    // will be used
    string signature_name = 1;
  
    // Input tensors.
    // Names of input tensor are alias names. The mapping from aliases to real
    // input tensor names is expected to be stored as named generic signature
    // under the key "inputs" in the model export.
    // Each alias listed in a generic signature named "inputs" should be provided
    // exactly once to run the prediction.
    map<string, ArrayProto> inputs = 2;
  
    // Output filter.
    // Names specified are alias names. The mapping from aliases to real output
    // tensor names is expected to be stored as named generic signature under
    // the key "outputs" in the model export.
    // Only tensors specified here will be run/fetched and returned, with the
    // exception that when none is specified, all tensors specified in the
    // named signature will be run/fetched and returned.
    repeated string output_filter = 3;
    
    // Debug flags
    // 0: just return prediction results, no debug information
    // 100: return prediction results, and save request to model_dir 
    // 101: save timeline to model_dir
    int32 debug_level = 100;
  }
  
  // Response for PredictRequest on successful run.
  message PredictResponse {
    // Output tensors.
    map<string, ArrayProto> outputs = 1;
  }

• easyrec_predict.proto: Request definition for a TensorFlow model with 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";
  
  // context features
  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 specifies the request for aggregator
  message PBRequest {
    // Debug flags
    // 0: just return prediction results, no debug information
    // 3: return features generated by FG module, string format, feature values are separated by \u0002, 
    //    could be used for checking feature consistency check and generating online deep learning samples 
    // 100: return prediction results, and save request to model_dir 
    // 101: save timeline to model_dir
    // 102: for recall models such as DSSM and MIND, only only return Faiss retrieved results
    //      but also return user embedding vectors.
    int32 debug_level = 1;
  
    // user features
    map<string, PBFeature> user_features = 2;
  
    // item ids, static(daily updated) item features 
    // are fetched from the feature cache resides in 
    // each processor node by item_ids
    repeated string item_ids = 3;
  
    // context features for each item, realtime item features
    //    could be passed as context features.
    map<string, ContextFeatures> context_features = 4;
  
    // embedding retrieval neighbor number.
    int32 faiss_neigh_num = 5;
  }
  
  // return results
  message Results {
    repeated double scores = 1 [packed = true];
  }
  
  enum StatusCode {
    OK = 0;
    INPUT_EMPTY = 1;
    EXCEPTION = 2;
  }
  
  // PBResponse specifies the response for aggregator
  message PBResponse {
    // results
    map<string, Results> results = 1;
  
    // item features
    map<string, string> item_features = 2;
  
    // fg generate features
    map<string, string> generate_features = 3;
  
    // context features
    map<string, ContextFeatures> context_features = 4;
  
    string error_msg = 5;
  
    StatusCode status_code = 6;
  
    // item ids
    repeated string item_ids = 7;
  
    repeated string outputs = 8;
  
    // all fg input features
    map<string, string> raw_features = 9;
  
    // output tensors
    map<string, ArrayProto> tf_outputs = 10;
  }