La función de modelo de ordenamiento personalizado permite desplegar en OpenSearch un modelo de ordenamiento de red neuronal profunda (DNN) basado en TensorFlow. Esta guía explica cómo implementar la clase CustomModel, configurar las columnas de funciones y cumplir los requisitos de compatibilidad para el inference en línea.
Cómo funciona
El framework construye un grafo de cómputo de TensorFlow mediante la llamada a la función build, la cual ejecuta en orden los métodos que se muestran a continuación:
build_placeholder()build_model()setup_global_step()reg_loss()loss_op()update_op()training_op()predictions_op()mark_output()metrics_op()summary_op()trace_sample_op()
Debe implementar cinco de estos métodos. Los restantes usan la lógica predeterminada del framework; no los sobrescriba.
Requisitos previos
Antes de empezar, asegúrese de contar con:
Un entorno de TensorFlow 1.x en funcionamiento
Familiaridad con las columnas de funciones de TensorFlow y con el entrenamiento de modelos DNN
Acceso a los datos de entrenamiento y al esquema de funciones
Implemente la clase CustomModel
CustomModel debe heredar de BaseModel. El flujo de trabajo de alto nivel es el siguiente:
Defina la clase y llame al constructor de la clase base.
Implemente los cinco métodos obligatorios.
Configure las columnas de funciones.
Asigne las salidas a los atributos obligatorios (
self.logits,self.loss,self.reg_loss).
Métodos obligatorios
Implemente exactamente estos cinco métodos; el framework se encarga de todo lo demás:
|
Método |
Qué implementar |
|
|
Defina la arquitectura del modelo (capa de embedding, capas DNN, capa de logits) |
|
|
Recopile las operaciones de actualización de normalización por lotes de |
|
|
Recopile las pérdidas de regularización de |
|
|
Defina la pérdida de entrenamiento (por ejemplo, entropía cruzada sigmoide + pérdida de regularización) |
|
|
Configure el optimizador y cree |
Métodos que se deben dejar sin implementar
No implemente build_placeholder, mark_output ni trace_sample_op. Estos métodos dependen de la lógica del framework que garantiza la compatibilidad en línea.
Ejemplo de implementación
El ejemplo que se muestra a continuación presenta una implementación completa de CustomModel que usa funciones de embedding y una DNN de tres capas:
from collections import OrderedDict
import tensorflow as tf
from tensorflow.contrib import layers
from tensorflow.contrib.framework.python.ops import arg_scope
from tensorflow.python.framework import ops
from tensorflow.python.ops import variable_scope
from model_ops.tflog import tflogger as logging
import model_ops.optimizer_ops as myopt
from model.base_model import BaseModel
from model_ops import ops as base_ops
from model_ops import utils
class CustomModel(BaseModel):
def __init__(self, config, name="CTR"):
# Always call the parent constructor with this exact signature
super(CustomModel, self).__init__(config, name)
# TensorFlow variable collections for tracking DNN layers
self.collections_dnn_hidden_layer = "{}_dnn_hidden_layer".format(self.name)
self.collections_dnn_hidden_output = "{}_dnn_hidden_output".format(self.name)
self.layer_dict = OrderedDict()
# Configure embedding feature columns for sparse categorical features
self.embedding_columns = ['feature1', 'feature2']
for feature_name in self.embedding_columns:
self.generate_embedding_feature_column(
feature_name,
hash_bucket_size=1000,
dimension=16,
initializer=tf.zeros_initializer,
combiner="sum",
is_share_embedding=False,
shared_embedding_name=None
)
# Uncomment to add real-valued or sparse ID columns:
# self.real_valued_columns = ['feature3', 'feature4']
# for feature_name in self.real_valued_columns:
# self.generate_real_valued_feature_column(feature_name, dtype="Float", value_dimension=1)
#
# self.sparse_id_columns = ['feature5', 'feature6']
# for feature_name in self.sparse_id_columns:
# self.generate_sparse_id_feature_column(feature_name, hash_bucket_size=1000, combiner="sum")
# Model hyperparameters
self.embedding_partition_size = 4 * 1024 * 1024
self.dnn_partition_size = 64 * 1024
self.dnn_l2_reg = 1e-6
self.clip_gradients = 5.0
self.dnn_hidden_units = [1024, 512, 256]
def build_placeholder(self):
try:
self.is_training = tf.get_default_graph().get_tensor_by_name("training:0")
except KeyError:
self.is_training = tf.placeholder(tf.bool, name="training")
def setup_global_step(self):
global_step = tf.Variable(
initial_value=0,
name="global_step",
trainable=False,
dtype=tf.int64,
collections=[tf.GraphKeys.GLOBAL_STEP, tf.GraphKeys.GLOBAL_VARIABLES])
self.global_step = global_step
def embedding_layer(self):
with tf.variable_scope(
name_or_scope="Embedding_Layer",
partitioner=base_ops.partitioner(self.config.ps_num, self.embedding_partition_size),
reuse=tf.AUTO_REUSE
) as scope:
logging.info('ps num: {}, embedding partition size: {} \n scope: {}'.format(
self.config.ps_num, self.embedding_partition_size, scope))
# Use input_from_feature_columns to embed sparse features — do not use other embedding functions
self.layer_dict['dnn'] = layers.input_from_feature_columns(
self.features,
self.feature_columns_from_column_names(self.embedding_columns),
scope=scope
)
def dnn_layer(self):
dnn_layer = [self.layer_dict['dnn']]
with tf.variable_scope(
name_or_scope="{}_Score_Network".format(self.name),
partitioner=base_ops.partitioner(self.config.ps_num, self.dnn_partition_size)
):
self.dnn_net = tf.concat(values=dnn_layer, axis=1)
with arg_scope(base_ops.model_arg_scope(weight_decay=self.dnn_l2_reg)):
for layer_id, num_hidden_units in enumerate(self.dnn_hidden_units):
with variable_scope.variable_scope("hiddenlayer_{}".format(layer_id)) as dnn_hidden_layer_scope:
tf.contrib.layers.apply_regularization(
regularizer=tf.contrib.layers.l2_regularizer(float(self.dnn_l2_reg)),
weights_list=[self.dnn_net])
self.dnn_net = layers.fully_connected(
self.dnn_net,
num_hidden_units,
utils.getActivationFunctionOp("llrelu"),
scope=dnn_hidden_layer_scope,
variables_collections=[self.collections_dnn_hidden_layer],
outputs_collections=[self.collections_dnn_hidden_output],
normalizer_fn=layers.batch_norm,
normalizer_params={"scale": True, "is_training": self.is_training})
def logits_layer(self):
with tf.variable_scope(
name_or_scope="{}_Logits".format(self.name),
partitioner=base_ops.partitioner(self.config.ps_num, self.dnn_partition_size)
) as dnn_logits_scope:
with arg_scope(base_ops.model_arg_scope(weight_decay=self.dnn_l2_reg)):
# Assign raw logits to self.logits — OpenSearch applies sigmoid automatically
self.logits = layers.linear(
self.dnn_net,
1,
scope=dnn_logits_scope,
variables_collections=[self.collections_dnn_hidden_layer],
outputs_collections=[self.collections_dnn_hidden_output])
def build_model(self):
self.embedding_layer()
self.dnn_layer()
self.logits_layer()
def update_op(self):
update_ops = tf.get_collection(tf.GraphKeys.UPDATE_OPS)
self.update_ops = [op for op in update_ops if op.name.startswith(self.name)]
def reg_loss(self):
reg_losses = tf.get_collection(tf.GraphKeys.REGULARIZATION_LOSSES)
self.reg_losses = [loss for loss in reg_losses if loss.name.startswith(self.name)]
self.reg_loss = tf.reduce_sum(self.reg_losses)
def training_op(self):
if self.config.predict:
self.train_op = None
return
with tf.variable_scope(name_or_scope="Optimize_Layer", reuse=tf.AUTO_REUSE):
gs = tf.train.get_or_create_global_step()
logging.info("Global_step: {}, {}".format(self.name, str(gs)))
logging.info("Model_name: {}, train_op_final_loss: {}".format(self.name, str(self.loss)))
self.train_op, _, _ = myopt.optimize_loss(
loss=self.loss,
global_step=self.global_step,
learning_rate=0.01,
optimizer=tf.train.AdamAsyncOptimizer(
learning_rate=0.01,
beta1=0.9,
beta2=0.999,
epsilon=1e-8,
use_locking=False),
update_ops=self.update_ops,
clip_gradients=self.clip_gradients,
variables=ops.get_collection(ops.GraphKeys.TRAINABLE_VARIABLES),
increment_global_step=True,
summaries=myopt.OPTIMIZER_SUMMARIES)
def loss_op(self):
with tf.name_scope("{}_Loss_Op".format(self.name)):
# Assign total loss (task loss + regularization) to self.loss
self.loss = tf.reduce_mean(
tf.nn.sigmoid_cross_entropy_with_logits(
logits=self.logits,
labels=self.label))
self.loss = self.loss + self.reg_loss
def metrics_op(self):
# Call the parent to let OpenSearch track common system metrics
super(CustomModel, self).metrics_op()
def summary_op(self):
with tf.name_scope("{}_Metrics_Scalar".format(self.name)):
for key, metric in self.metrics.items():
tf.summary.scalar(name=key, tensor=metric)
with tf.name_scope("{}_Layer_Summary".format(self.name)):
base_ops.add_norm2_summary(self.collections_dnn_hidden_layer)
base_ops.add_dense_output_summary(self.collections_dnn_hidden_output)
base_ops.add_weight_summary(self.collections_dnn_hidden_layer)Configure las columnas de funciones
Acceda a las funciones mediante self.features y use el nombre de la función como clave. Configure todas las columnas de funciones mediante las operaciones de API proporcionadas por OpenSearch; usar directamente funciones nativas de TensorFlow para las columnas de funciones puede provocar incompatibilidad con el servicio de inference en línea.
Use contrib.layers.input_from_feature_columns para obtener el embedding de las funciones. Otras funciones de embedding no son compatibles y causarán problemas en el servicio en línea.
Tipos de columna de funciones compatibles
|
Tipo |
Cuándo usar |
|
|
Funciones categóricas dispersas que requieren una búsqueda de embedding |
|
|
Funciones numéricas continuas |
|
|
Funciones de ID dispersas sin embedding |
|
|
Cuando se debe compartir el mismo embedding entre varias funciones |
Una columna de embedding no se puede usar dos veces en el mismo modelo. Si dos funciones deben compartir un embedding, configure esas funciones con is_share_embedding=True y el mismo shared_embedding_name.
Operaciones de API
Use las operaciones de API encapsuladas que se muestran a continuación para configurar las columnas de funciones:
# Embedding column — for sparse categorical features
self.generate_embedding_feature_column(
feature_name,
hash_bucket_size, # Number of hash buckets for the sparse feature
dimension, # Embedding dimension
initializer=tf.zeros_initializer,
combiner="sum", # How to combine multi-valued features: "sum", "mean", or "sqrtn"
is_share_embedding=False,
shared_embedding_name=None
)
# Real-valued column — for continuous numeric features
self.generate_real_valued_feature_column(
feature_name,
dtype="Float", # "Float" or "Int" only
value_dimension=1
)
# Sparse ID column — for sparse features without embedding
self.generate_sparse_id_feature_column(
feature_name,
hash_bucket_size,
combiner="sum"
)
# Retrieve configured feature columns by name
self.feature_columns_from_column_names(feature_list)Especificaciones del modelo
Cumpla todos los requisitos que se indican a continuación para garantizar la compatibilidad con el servicio en línea:
|
Atributo / método |
Requisito |
|
|
Llame a |
|
|
Asigne los logits sin procesar (puntuaciones previas a la sigmoide). OpenSearch aplica la función sigmoide automáticamente para convertirlos en la puntuación final de ordenamiento. Para usar una activación diferente, sobrescriba |
|
|
Asigne la pérdida total de entrenamiento |
|
|
Asigne la pérdida de regularización |
|
|
Llame a |
Notas de uso
Variables personalizadas fuera del módulo contrib
Si crea variables con tf.Variable o usa funciones fuera del módulo contrib, agregue esas variables a la colección MODEL_VARIABLES. OpenSearch carga los pesos con base en esta colección.
Agregue solo variables cuyos pesos deban cargarse en tiempo de servicio.
No agregue
global_stepa esta colección.
from tensorflow.python.framework import ops
from tensorflow.python.ops import variable_scope as vs
self._weights = vs.get_variable(
_WEIGHTS_VARIABLE_NAME,
[total_arg_size, output_size],
dtype=dtype,
initializer=kernel_initializer,
collections=[ops.GraphKeys.GLOBAL_VARIABLES, ops.GraphKeys.MODEL_VARIABLES]
)