Platform For AI:TorchEasyRec Processor

O TorchEasyRec Processor nativo do EAS implanta modelos de recomendação treinados com TorchEasyRec ou PyTorch como serviços de scoring com engenharia de features integrada. O processor otimiza conjuntamente a engenharia de features e o modelo PyTorch para oferecer serviços de scoring de alto desempenho. Este tópico explica como implantar e chamar um service de modelo TorchEasyRec.

Contexto

O diagrama a seguir mostra a arquitetura de um mecanismo de recomendação baseado no TorchEasyRec Processor.

O TorchEasyRec Processor é composto pelos seguintes módulos:

• Item Feature Cache: Armazena em memória as features do lado do item provenientes do FeatureStore para reduzir a sobrecarga de rede, aliviar a carga sobre o FeatureStore e melhorar o desempenho do service de inferência. Quando as features do lado do item incluem features em tempo real, o FeatureStore gerencia a sincronização.

• feature generator (FG): Define transformações de features em um arquivo de configuração, utilizando uma base de código C++ unificada para garantir consistência lógica entre o processamento de features offline e online.

• TorchModel: Modelo PyTorch treinado com TorchEasyRec ou PyTorch e exportado como ScriptedModel.

Limitações

Há suporte apenas para os modelos g6, g7 e g8 da família de instâncias de uso geral, além de modelos GPU como RTX Pro 5000, L20, A10 e T4. Para mais informações, consulte general-purpose instance family (g series). Ao implantar um service com GPU, certifique-se de que a versão do CUDA Driver seja 535 ou superior.

Histórico de versões

O TorchEasyRec Processor está em desenvolvimento ativo. Recomendamos usar a versão mais recente para implantar seu service de inferência, pois ela oferece mais recursos e maior desempenho. As versões lançadas são as seguintes:

Etapa 1: Implantar um service

1. Prepare o arquivo de configuração do service torcheasyrec.json.

   Especifique o Tipo de Processor como easyrec-torch-{version} e selecione um valor para {version} no histórico de versões. Veja abaixo um exemplo do arquivo de configuração JSON:

   Exemplo com 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-2.4"
   }

   Exemplo sem 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-2.4"
   }

   Para outros parâmetros, consulte JSON deployment.

   Parâmetro	Obrigatório	Descrição	Exemplo
   processor	Sim	O TorchEasyRec Processor.	"processor":"easyrec-torch-1.12"
   path	Sim	Caminho oss montado no service para armazenar arquivos de modelo.	"path": "oss://examplebucket/xxx/export"
   fg_mode	Não	Define o modo de engenharia de features. Valores válidos: bypass (padrão): Desativa a engenharia de features (FG). Apenas o modelo Torch é implantado. Use este modo para processamento personalizado de features. Neste modo, o Processor não requer parâmetros de acesso ao FeatureStore. normal: Ativa o FG. Este modo é geralmente usado com TorchEasyRec para treinamento de modelos.	"fg_mode": "normal"
   fg_threads	Não	Número de threads concorrentes para execução do FG por requisição.	"fg_threads": 15
   outputs	Não	Nomes das variáveis de saída da previsão do modelo Torch, como probs_ctr. Use vírgulas (,) para separar múltiplos nomes. Se este parâmetro não for especificado, o service retorna todas as variáveis.	"outputs":"probs_ctr,probs_cvr"
   item_empty_score	Não	Score padrão retornado quando um ID de item não existe. Valor padrão: 0.	"item_empty_score": -1
   Parâmetros de recall vetorial do Processor
   faiss_neigh_num	Não	Número de itens a serem recuperados para recall vetorial FAISS. O service prioriza o valor do campo faiss_neigh_num no corpo da requisição. Se esse campo não for fornecido, utiliza o valor do parâmetro faiss_neigh_num na seção model_config. O valor padrão é 1.	"faiss_neigh_num": 200
   faiss_nprobe	Não	O parâmetro nprobe especifica o número de clusters a serem buscados durante a recuperação. No FAISS, um índice de arquivo invertido divide os dados em clusters menores e mantém uma lista invertida para cada um. Um valor maior de nprobe geralmente melhora a precisão do recall ao custo de maior computação e tempo de busca, enquanto um valor menor reduz a precisão, mas acelera a busca. O valor padrão é 800.	"faiss_nprobe" : 700
   Parâmetros do Processor para acesso ao FeatureStore
   fs_project	Não	Nome do seu projeto FeatureStore. Este parâmetro é obrigatório ao usar o FeatureStore. Para mais informações, consulte Configure a FeatureStore project.	"fs_project": "fs_demo"
   fs_model	Não	Nome do modelo de features no FeatureStore.	"fs_model": "fs_rank_v1"
   fs_entity	Não	Nome da entidade no FeatureStore.	"fs_entity": "item"
   region	Não	Região onde seu projeto FeatureStore está localizado. Por exemplo, especifique cn-beijing para a região China (Beijing). Para obter uma lista de regiões suportadas e seus endpoints, consulte Endpoints.	"region": "cn-beijing"
   access_key_id	Não	AccessKey ID para acessar o FeatureStore.	"access_key_id": "xxxxx"
   access_key_secret	Não	AccessKey Secret para acessar o FeatureStore.	"access_key_secret": "xxxxx"
   load_feature_from_offlinestore	Não	Define se as features offline devem ser carregadas diretamente do OfflineStore do FeatureStore. Valores válidos: True: Carrega dados do OfflineStore do FeatureStore. False (padrão): Carrega dados do OnlineStore do FeatureStore.	"load_feature_from_offlinestore": True
   featuredb_username	Não	Nome de usuário para o FeatureDB.	"featuredb_username":"xxx"
   featuredb_password	Não	Senha para o FeatureDB.	"featuredb_passwd":"xxx"
   Parâmetros para expansão automática de features (input_tile)
   INPUT_TILE	Não	Ativa a expansão automática de features para otimizar o desempenho. Para features que compartilham o mesmo valor em todos os itens de uma única requisição, como um user_id, envie o valor apenas uma vez. Isso ajuda a reduzir o tamanho do payload da requisição, a latência de rede e o tempo de computação. Este recurso deve ser usado no modo normal e com um modelo treinado com TorchEasyRec. A variável de ambiente correspondente também deve ser definida durante a exportação do modelo. Por padrão, o sistema lê o valor INPUT_TILE do arquivo model_acc.json no diretório do modelo exportado. Se este arquivo estiver ausente, o sistema lê o valor da variável de ambiente. Quando este recurso está ativado: Se a variável de ambiente estiver definida como 2, o FG para features do lado do usuário é calculado apenas uma vez. Se a variável de ambiente estiver definida como 3, o FG para features do lado do usuário é calculado apenas uma vez. O sistema calcula embeddings para features de usuário e item separadamente, e o embedding do lado do usuário é calculado apenas uma vez. Esta configuração é ideal para cenários com muitas features do lado do usuário.	"processor_envs": [ { "name": "INPUT_TILE", "value": "2" } ]
   NO_GRAD_GUARD	Não	Desativa o cálculo de gradientes durante a inferência. Isso interrompe o rastreamento de operações e impede a construção de um grafo de computação. Nota Definir este parâmetro como 1 pode causar problemas de incompatibilidade com alguns modelos. Se o service travar durante uma segunda execução de inferência, resolva o problema definindo a variável de ambiente PYTORCH_TENSOREXPR_FALLBACK=2. Isso ignora a etapa de compilação, mantendo algumas capacidades de otimização de grafo.	"processor_envs": [ { "name": "NO_GRAD_GUARD", "value": "1" } ]
   Parâmetros de warm-up do modelo
   warmup_data_path	Não	Ativa o recurso de warm-up do modelo e especifica o caminho para salvar os arquivos de warm-up. Para persistir os arquivos de warm-up, monte um caminho oss neste local na configuração de storage.	"warmup_data_path": "/warmup"
   warmup_cnt_per_file	Não	Número de vezes para executar o processo de warm-up para cada arquivo Protobuf. Um valor maior garante um warm-up mais completo, mas aumenta o tempo de inicialização. Valor padrão: 20.	"warmup_cnt_per_file": 20,
   warmup_pb_files_count	Não	Número de requisições online a serem salvas como arquivos Protobuf para a próxima inicialização do service. Os arquivos são usados para warm-up do modelo e salvos no caminho especificado por warmup_data_path. Valor padrão: 64.	"warmup_pb_files_count": 64
   Registro e salvamento de requisições lentas
   long_request_threshold	Não	Limiar de tempo em milissegundos (ms) para identificar uma requisição lenta. Se o tempo de processamento de uma requisição exceder esse limiar, o sistema registra automaticamente o tempo de execução de cada etapa. Valor padrão: 200.	"long_request_threshold": 200
   save_long_request	Não	Define se requisições que excedem o long_request_threshold devem ser salvas como arquivos Protobuf. Se definido como true, os arquivos são salvos no diretório torch_req dentro do diretório do modelo. Valor padrão: false.	"save_long_request": true
   Salvamento de requisições brutas e features de item no oss
   request_log_path	Não	Caminho em disco para salvar os arquivos Protobuf. Monte um caminho oss neste local na configuração do service.	"request_log_path": "/online_log_pb"
   background_feature_thread_num	Não	Número de threads em segundo plano dedicadas a gravar arquivos em disco. Se a carga de trabalho de gravação em disco for alta, aumente este valor para acelerar o processo de salvamento. Valor padrão: 4.	"background_feature_thread_num": 8
   Configuração de dados de passagem direta (pass-through)
   pass_through_data	Não	Especifica dados para passar diretamente para a resposta. Útil para transmitir informações a serviços downstream. O valor deve ser um objeto JSON.	"pass_through_data": {"model_version": "20260513"}

2. Implante o service de modelo TorchEasyRec usando um dos métodos a seguir:

   JSON (Recomendado)

   Siga estas etapas:

   1. Faça login no PAI console. Selecione uma região no topo da página. Em seguida, selecione o workspace desejado e clique em Enter Elastic Algorithm Service (EAS).

   2. Na página Elastic Algorithm Service (EAS), clique em Deploy Service. Na seção Custom Model Deployment, clique em JSON Deployment.

   3. No editor JSON, cole sua configuração JSON e clique em Deploy.

   eascmd CLI

   1. Download and authenticate the client. As etapas a seguir usam a versão Windows 64 bits como exemplo.

   2. No diretório contendo seu arquivo JSON, execute o comando a seguir para criar um service. Para mais informações sobre o comando, consulte Command reference.

      eascmdwin64.exe create <service.json>

      Substitua <service.json> pelo nome do seu arquivo JSON, por exemplo, torcheasyrec.json.

Etapa 2: Chamar um service

Após implantar o service de modelo TorchEasyRec, siga estas etapas para visualizar as informações de chamada do service:

1. Faça login no PAI console, selecione a região no topo da página e o workspace à direita, e então clique em Go to EAS.

2. Clique em Invocation Information na coluna Service Type do service alvo para ver o endpoint e o token do service.

   Na caixa de diálogo Invocation Information, selecione a aba Public Address para obter o public endpoint e o Token para chamadas subsequentes ao service.

O service de modelo TorchEasyRec usa Protobuf como formato de entrada/saída. Existem dois métodos de chamada, dependendo se o FG está ativado:

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)

Para detalhes sobre códigos de status do service, consulte Service status codes. Para construir uma requisição de service, veja também Request format.

Formato da requisição

Para chamar o service, gere o código de requisição de previsão a partir do arquivo .proto. Alternativamente, para construir a requisição manualmente, use as seguintes definições 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 that a DataType field
  // has not been set.
  DT_INVALID = 0;
  
  // Data types that all computation devices are expected 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];
}

// Represents an array.
message ArrayProto {
  // Data type.
  ArrayDataType dtype = 1;

  // Array shape.
  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 the recommendation model.
  map<string, ArrayProto> map_inputs = 3;

  // Debug level for the recommendation model.
  int32 debug_level = 100;
}

// Response for a successful PredictRequest.
message PredictResponse {
  // Output tensors.
  repeated ArrayProto outputs = 1;
  // Output tensors from the recommendation model.
  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";

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

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

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

// Single-level lists.
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;
}

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

// Defines the request sent to the aggregator.
message PBRequest {
  // Debug level.
  int32 debug_level = 1;

  // User features, keyed by the input name.
  map<string, PBFeature> user_features = 2;

  // Item IDs.
  repeated string item_ids = 3;

  // Context features for each item, keyed by the input name. 
  map<string, ContextFeatures> context_features = 4;

  // The number of nearest neighbors to retrieve from Faiss.
  int32 faiss_neigh_num = 5;

  // Item features for each item, keyed by the input name. 
  map<string, ContextFeatures> item_features = 6;
  
  // Optional metadata.
  map<string, string> meta_data = 7;
}

// Defines the response from the aggregator.
message PBResponse {
  // Torch output tensors.
  map<string, pytorch.eas.ArrayProto> map_outputs = 1;

  // Output features from the feature generator (FG).
  map<string, string> generate_features = 2;

  // All input features for the feature generator (FG).
  map<string, string> raw_features = 3;

  // Item IDs.
  repeated string item_ids = 4;

  // Pass-through data configured in the model.
  map<string, string> pass_through_data = 5;
}

O parâmetro debug_level funciona da seguinte forma:

Códigos de status do service

Esta seção descreve os principais códigos de status do service TorchEasyRec. Para códigos de status de serviços EAS, consulte Appendix: Service status codes and common errors.

Salvar e analisar uma requisição Protobuf

Na versão 1.12 ou superior do processor, ative o modo debug definindo debug=True no corpo da requisição do mecanismo PAI-REC. Isso salva a requisição original e as features de entrada e saída do lado do item em um arquivo protobuf para análise e validação. Para usar este recurso, defina o parâmetro request_log_path para o caminho de destino e monte o oss nesse caminho. Por exemplo:

"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
        }
    }
]

O processor cria um subdiretório date_hour no caminho especificado por request_log_path para salvar os dados da requisição. Threads em segundo plano gravam esses dados em disco assincronamente. Use o parâmetro model_config.background_feature_thread_num para configurar o número de threads em segundo plano. O padrão é 4, e aumentar esse valor pode melhorar o throughput de gravação. Arquivos Protobuf gravados em disco usam o formato <request_id>_<random_str>.pb. Devido à largura de banda limitada de gravação do oss, não ative o modo debug para altos volumes de requisições no mecanismo PAI-REC. Se as gravações em disco ficarem atrasadas, a fila interna do service de modelo descartará novas requisições.

Para analisar o arquivo protobuf, use o EAS-Python-SDK 0.35 ou superior, ou o EAS-Java-SDK 2.0.29 ou superior. O exemplo a seguir mostra como fazer isso em 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 the entire log

print(pb_log.request) # Print the request
print(pb_log.raw_features) # Print the raw item-side features
print(pb_log.generate_features) # Print the item-side generated features

Warm-up do service de modelo

Quando um service de modelo inicia ou é atualizado, podem ocorrer picos no tempo de resposta. Para mitigar esses picos, configure o recurso de warm-up para o processor. Por exemplo, no easyrec-torch-1.5 e versões posteriores, ative este recurso adicionando três parâmetros ao model_config.

"warmup_data_path": "/warmup",  # Enables warmup and sets the path for the warmup files.
"warmup_cnt_per_file": 20, # Number of warmup iterations per file. A higher value results in a more thorough warmup.
"warmup_pb_files_count": 64 # Number of online requests to save as protobuf files for warmup. A higher value helps cover more data patterns.

Para persistir arquivos protobuf, configure uma montagem oss na seção storage no caminho warmup_data_path. Por exemplo:

"storage": [
    ...,
    {
        "mount_path": "/warmup",
        "oss": {
            "path": "oss://<warmup Protobuf file path>",
            "readOnly": false
        }
    }
]

Na primeira inicialização após a configuração, o processor captura e salva o número de requisições online especificado por warmup_pb_files_count. Nas reinicializações subsequentes, ele usa esses arquivos protobuf salvos para realizar o warm-up.

Treinamento e implantação de aprendizado online

Em cenários de recomendação e publicidade, os interesses dos usuários, a popularidade dos itens e os dados de negócios mudam continuamente. Os modelos precisam absorver novos sinais de feedback oportunamente e sincronizar os resultados do treinamento com os serviços online. O aprendizado online reduz o tempo entre a participação de novos dados no treinamento e seu impacto nos resultados de previsão através de treinamento contínuo e atualizações de parâmetros.

A partir do easyrec-torch-2.5, o TorchEasyRec Processor suporta atualizações incrementais de embedding e hot updates de modelos densos no modo de inferência com embedding distribuído. Combinado com jobs de treinamento do TorchEasyRec e FeatureDB, você pode construir um pipeline ponta a ponta desde o treinamento contínuo até a publicação de parâmetros e inferência online, sem reiniciar ou reimplantar o service de inferência a cada atualização de parâmetro. O TorchEasyRec é responsável por treinar e exportar parâmetros do modelo, o FeatureDB armazena e fornece dados de atualização de embedding, e o Processor aplica as atualizações na inferência online.

Princípio de funcionamento

O aprendizado online divide as atualizações do modelo em dois caminhos:

• Atualização incremental de embedding: O lado do treinamento captura embeddings alterados e os grava no FeatureDB. O Processor puxa e atualiza as tabelas de embedding locais usadas para inferência, o que reduz a sobrecarga de recarregar repetidamente todos os parâmetros. Este mecanismo suporta embeddings de ponto flutuante e quantizados, além de permitir a limpeza de chaves de embedding dinâmicas eliminadas pelo treinamento.

• Hot update de modelo denso: O lado do treinamento publica uma nova versão do modelo denso. Após detectar a atualização, o Processor realiza uma verificação de compatibilidade e warm-up, e então troca para a nova versão em quente. Durante o carregamento e warm-up, o modelo existente continua atendendo requisições de inferência.

Para coordenar os dois caminhos, o TorchEasyRec e o Processor fornecem um mecanismo de alinhamento de atualizações Densa/Esparsa para garantir a qualidade do scoring online.

Uso

1. Pré-requisitos

   1. TorchEasyRec: >= 1.4.3 ou superior

   2. Processor: >= easyrec-torch-2.5

   3. O Processor roda em um ambiente NVIDIA GPU.

   4. FeatureDB ativado. Crie um projeto e configure o FeatureDB como source de dados online.

2. Procedimento

   1. Fase de treinamento

      A fase de treinamento consiste em duas etapas: um treinamento offline completo e exportação para inicializar o Processor, e um treinamento online incremental para atualizar continuamente o Processor.

      1. Treinamento completo com dados offline: Treine o modelo da mesma forma que antes e exporte no modo distribuído.

         Comando de treinamento:

         torchrun --master_addr=localhost --master_port=<port> \
             --nnodes=1 --nproc-per-node=1 --node_rank=0 \
             -m tzrec.train_eval \
             --pipeline_config_path ${work_dir}/your_training_config.config \
             --train_input_path ${data_dir}/\*.parquet \
             --eval_input_path ${data_dir}\*.parquet \
             --model_dir ${work_dir}/your_model_dir

         Comando de exportação: Defina a variável de ambiente USE_DISTRIBUTED_EMBEDDING como 1. Se a quantização for necessária, defina também DIST_QUANT=INT8.

         export USE_DISTRIBUTED_EMBEDDING=1
         torchrun --master_addr=localhost --master_port=<port> \
             --nnodes=1 --nproc-per-node=1 --node_rank=0 \
             -m tzrec.export \
             --pipeline_config_path ${model_dir}/pipeline.config \
             --export_dir ${model_dir}/export_dist  # Export directory for full offline training. This directory must also be specified during online training.

      2. Treinamento incremental online: Carregue continuamente dados de embedding para o FeatureDB e exporte continuamente modelos densos para o oss.

         Para carregar continuamente dados de embedding no FeatureDB, adicione a seguinte configuração ao campo train_config na configuração de treinamento:

            train_config{
               delta_embedding_dump_config {
                   paired_dump_interval_minutes: 5 # Required. Overall frequency switch. Export dense model every 5 minutes, and also dump and upload delta embeddings.
                   sparse_dump_interval_minutes: 1 # Optional. Fine-grained frequency switch for sparse embeddings. Upload embeddings more frequently.
         
                   quant_type: DELTA_EMBEDDING_QUANT_INT8
         
                   feature_store_config {
                       region: "cn-beijing"
                       project_name: "dynemb_test"
                       feature_view_name: "demo_dynemb_feature_view"
                       version: "20260725"
                   }
               }
           }

         Descrição dos parâmetros:

         paired_dump_interval_minutes		Frequência geral do aprendizado online. Especifica que a cada N minutos, os embeddings alterados no intervalo são coletados e carregados no FeatureStore, e um snapshot do modelo denso também é exportado.
         sparse_dump_interval_minutes		Frequência granular para uploads de embeddings esparsos. Especifica que a cada M minutos, os embeddings alterados no intervalo são coletados e carregados no FeatureStore, sem exportar um snapshot do modelo denso. Este parâmetro é opcional. Se definido, deve ser menor que paired_dump_interval_minutes.
         paired_dump_interval_steps / sparse_dump_interval_steps		Mesmo que acima, mas usa passos de treinamento como base de frequência em vez de minutos. Mutuamente exclusivo com os parâmetros baseados em minutos.
         quant_type		Tipo de quantização. Valores válidos: DELTA_EMBEDDING_QUANT_NONE (sem quantização) DELTA_EMBEDDING_QUANT_INT8 (quantização INT8) A quantização é desativada por padrão. Recomendamos ativá-la quando o modelo for grande.
         feature_store_config	region	Região onde o FeatureDB está ativado, por exemplo, cn-beijing.
         	project_name	Nome do projeto FeatureStore pré-criado.
         	feature_view_name	Nome da feature view. Se não existir, o sistema a cria automaticamente.
         	version	Número da versão atual do treinamento. O treinamento online carrega dados de embedding no pool de dados especificado por esta versão, gerenciado pelo FeatureStore. O Processor também deve especificar esta versão.

         Comando de treinamento:

         export ALIBABA_CLOUD_ACCESS_KEY_ID=your key
         export ALIBABA_CLOUD_ACCESS_KEY_SECRET=your secret
         export FEATUREDB_USERNAME=your fdb user
         export FEATUREDB_PASSWORD=your fdb pw
         
         export ONLINE_DENSE_EXPORT=1
         export USE_DISTRIBUTED_EMBEDDING=1
         export ONLINE_DENSE_EXPORT_DIR=your full offline export directory
         export ONLINE_DENSE_EXPORT_KEEP_VERSIONS=20
         
         torchrun --master_addr=localhost --master_port=32555 \
             --nnodes=1 --nproc-per-node=1 --node_rank=0 \
             -m tzrec.train_eval \
             --pipeline_config_path online_train.config \
             --continue_train \
             --train_input_path /<data_dir>/\*.parquet \
             --eval_input_path /<data-dir>/\*.parquet \
             --model_dir your_model_dir

         Descrição das variáveis de ambiente:

         ONLINE_DENSE_EXPORT=1	Ativa a exportação contínua de modelos densos.
         USE_DISTRIBUTED_EMBEDDING=1	Ativa exportação distribuída com separação esparso/denso.
         ONLINE_DENSE_EXPORT_DIR	Diretório para a exportação offline completa.
         ONLINE_DENSE_EXPORT_KEEP_VERSIONS	Controla a limpeza de modelos densos exportados. Valor padrão: 0, significando nenhuma limpeza. Quando definido com um valor maior que 0, o sistema retém as últimas K versões. O valor mínimo de K é 3. Isso evita uso excessivo de recursos devido a muitos arquivos.

   2. Fase de inferência

      Configuração do Processor:

      "model_config": {
              "fg_mode": "normal",
              "featuredb_username": "your user",
              "featuredb_password": "your pw",
              "access_key_id": "your ak",
              "access_key_secret": "your as",
              "region": "your region"
              "fs_project": "your fs project",
      
              "dense_hot_load": {
                  "enable": true,
                  "dense_poll_interval_minutes": 1
              },
              "feature_store_embedding_polling": {
                  "feature_view_name": "your feature view name",
                  "mode": "apply",
                  "version": "your feature store version",
                  "sparse_poll_interval_secs": 10
              }
          },
          "processor": "easyrec-torch-2.5",

      Descrição dos parâmetros:

      dense_hot_load	enable: true	Permite que o Processor faça polling no diretório do modelo. Quando um novo modelo denso é detectado, o Processor o carrega e troca para ele automaticamente. Antes da troca, o Processor puxa dados do FeatureDB para sincronizar embeddings esparsos.
      	dense_poll_interval_minutes	Frequência com que o Processor faz polling no diretório do modelo, em minutos. Valor padrão: 1.
      feature_store_embedding_polling	feature_view_name	Nome da feature view para verificar atualizações de embedding via polling.
      	mode	Modo de atualização. Defina este parâmetro como apply.
      	version	Versão do FeatureStore que corresponde à configuração de treinamento.
      	sparse_poll_interval_secs	Intervalo em que o Processor verifica o FeatureStore para atualizações de embeddings esparsos, em segundos. Valor padrão: 10.