在Tensorflow Serving上部署基于LSTM的文本分

一些重要的概念

---

Servables

• Servables 是客户端请求执行计算的基础对象，大小和粒度是灵活的。
• Servables 不会管理自己的运行周期。
• 典型的Servables包括：
  • a TensorFlow SavedModelBundle (tensorflow::Session)
  • a lookup table for embedding or vocabulary lookups

Servable Versions

• Tensorflow servables 可以管理多个版本的servable，可以通过版本管理来更新算法配置和模型参数；
• 可以支持同时加载多个servable

Servable Streams

• servable的版本序列。

Models

• TensorFlow Serving 将 model 表示为一个或者多个Servables，一个Servable可能对应着模型的一部分，例如，a large lookup table 可以被许多 TensorFlow Serving 共享。

Loaders

• Loaders 管理着对应的一个servable的生命周期

Source

• Sources 是可以寻找和提供 servables 的模块，每个 Source 提供了0个或者多个servable streams，对于每个servable stream，Source 都会提供一个Loader实例；
• Sources 可以保存在多个服务或者版本之间共享的state；

Aspired Versions

• Aspired Versions 是可以被加载服务版本集合；

Managers

• 管理 Servable 的整个的生命周期，包括：
  • loading Servables
  • serving Servables
  • unloading Servables
• 如果服务要求的资源不足，Manager 可能会拒绝加载新的模型；

---

流程

---

流程图

简单来说：

• Source 为 Servable 创建 Loader 对象；
• Loader 被作为 Aspired Versions 发送给 Manager，之后 Manager 加载指定的服务去处理客户端的请求；

具体来说：

• Source 为指定的服务创建Loader，Loader里包含了服务所需要的元数据（模型）；
• 之后 Source 使用 回掉函数通知 Manager 的 Aspired Version(servable version的集合);
• Manager 根据配置的Version Policy决定下一步的操作（是否 unload 之前的servable，或者 load 新的servable）；
• 如果 Manager 判定是操作安全的，就会给 Loader 要求的resource并让 Loader 加载新的版本;
• 客户端向 Manager 请求服务，可以指定服务版本或者只是请求最新的版本。Manager 返回服务端的处理结果；

Source 可以看作不断更新weights的graph，weights 存储在磁盘中

例如：

• Source 选择了一个新版本的模型，创建了一个Loader，Loader包含指向模型文件路径的指针；
• Source 使用回掉函数通知 Manager 的 Aspired Version；
• 根据 Version Policy ，Manager 决定加载新的模型；
• Manager 通知 Loader 有足够的内存用于模型加载，Loader 使用新的weights将graph实例化；
• 客户端发送请求，Manager 返回处理结果；

生成可用于tensorflow serving的模型

---

环境：Tensorflow 1.0.1
模型：基于LSTM的文本情感分类的模型：

import os,time
import tensorflow as tf

import online_lstm_model
from utils import data_utils,data_process
from lib import config
#载入词向量，生成字典
embedding_matrix, word_list = data_utils.load_pretained_vector(config.WORD_VECTOR_PATH)
data_utils.create_vocabulary(config.VOCABULARY_PATH, word_list)
#数据预处理
for file_name in os.listdir(config.ORGINAL_PATH):
    data_utils.data_to_token_ids(config.ORGINAL_PATH + file_name, config.TOCKEN_PATN + file_name,
                                   config.VOCABULARY_PATH)
vocabulary_size = len(word_list) + 2
#获取训练数据
x_train, y_train, x_test, y_test = data_process.data_split(choose_path=config.choose_car_path,
                                                           buy_path=config.buy_car_path,
                                                           no_path=config.no_car_path)

with tf.Graph().as_default():
    #build graph
    model = online_lstm_model.RNN_Model(vocabulary_size, config.BATCH_SIZE, embedding_matrix)
    logits = model.logits
    loss  = model.loss
    cost = model.cost
    acu = model.accuracy
    prediction = model.prediction
    train_op = model.train_op

    saver = tf.train.Saver()
    #GPU设置
    gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction=1.0)
    with tf.Session(config=tf.ConfigProto(gpu_options=gpu_options)) as sess:
        count = 0
        init = tf.global_variables_initializer()
        sess.run(init)
        while count<config.max_steps:
            start_time = time.time()
            #生成训练batch
            batch_x_train, batch_y_train = data_process.generate_batch(x_train, y_train)
            input_feed = {
                model.input_data: batch_x_train,
                model.target: batch_y_train
            }
            #进行训练
            train_loss, train_cost, train_acu, _ = sess.run(fetches=[loss, cost, acu, train_op],
                                                            feed_dict=input_feed)

            count += 1
            if count % 100 == 0:
                #每100轮，验证一次
                sum_valid_cost = 0
                sum_valid_acu = 0
                for i in range(len(x_test)//64):
                    batch_x_valid, batch_y_valid = data_process.generate_batch(x_test, y_test)
                    valid_feed = {
                        model.input_data: batch_x_valid,
                        model.target: batch_y_valid
                    }
                    valid_loss, valid_cost, valid_acu = sess.run(fetches=[loss, cost, acu],
                                                                feed_dict=valid_feed)
                    sum_valid_cost += valid_cost
                    sum_valid_acu += valid_acu
                print("current step: %f, train cost: %f, train accuracy: %f, cost_time: %f"%(count, train_cost, train_acu, time.time()-start_time))
                print("valid cost: %f, valid accuracy: %f"%(sum_valid_cost/(len(x_test)//64), sum_valid_acu/(len(x_test)//64)))
        #将模型保存为可用于线上服务的文件（一个.pb文件，一个variables文件夹）
        export_path_base = config.export_path_base
        export_path = os.path.join(
            tf.compat.as_bytes(export_path_base),
            tf.compat.as_bytes(str(count)))
        print('Exporting trained model to', export_path)

        builder = tf.saved_model.builder.SavedModelBuilder(export_path)

        # 建立签名映射
        """
        build_tensor_info：建立一个基于提供的参数构造的TensorInfo protocol buffer，
        输入：tensorflow graph中的tensor；
        输出：基于提供的参数（tensor）构建的包含TensorInfo的protocol buffer
        """
        input_sentence = tf.saved_model.utils.build_tensor_info(model.input_data)
        classification_outputs_classes = tf.saved_model.utils.build_tensor_info(model.prediction)
        classification_outputs_scores = tf.saved_model.utils.build_tensor_info(model.logits)

        """
        signature_constants：SavedModel保存和恢复操作的签名常量。
        
        如果使用默认的tensorflow_model_server部署模型，
        这里的method_name必须为signature_constants中CLASSIFY,PREDICT,REGRESS的一种
        """
        #定义模型的输入输出，建立调用接口与tensor签名之间的映射
        classification_signature = (
            tf.saved_model.signature_def_utils.build_signature_def(
                inputs={
                    "input_sentence":
                        input_sentence
                },
                outputs={
                    "predict_classification":
                        classification_outputs_classes,
                    "predict_scores":
                        classification_outputs_scores
                },
                method_name=tf.saved_model.signature_constants.CLASSIFY_METHOD_NAME))

        """
        tf.group : 创建一个将多个操作分组的操作，返回一个可以执行所有输入的操作
        """
        legacy_init_op = tf.group(tf.tables_initializer(), name='legacy_init_op')

        """
        add_meta_graph_and_variables：建立一个Saver来保存session中的变量，
                                      输出对应的原图的定义，这个函数假设保存的变量已经被初始化；
                                      对于一个SavedModelBuilder，这个API必须被调用一次来保存meta graph；
                                      对于后面添加的图结构，可以使用函数 add_meta_graph()来进行添加
        """
        #建立模型名称与模型签名之间的映射
        builder.add_meta_graph_and_variables(
            sess, [tf.saved_model.tag_constants.SERVING],
            #保存模型的方法名，与客户端的request.model_spec.signature_name对应
            signature_def_map={
                tf.saved_model.signature_constants.DEFAULT_SERVING_SIGNATURE_DEF_KEY:
                    classification_signature},
            legacy_init_op=legacy_init_op)

        builder.save()
        print("Build Done")

启动服务

在这里采用默认的 tensorflow_model_server 启动服务

bazel-bin/tensorflow_serving/model_servers/tensorflow_model_server  --port=8000 --model_name=your_model_name --model_base_path=your_model_path

例如：

bazel-bin/tensorflow_serving/model_servers/tensorflow_model_server  --port=8000 --model_name=sentiment_classification --model_base_path=/home/liyonghong/sentiment_analysis/builder_last

注意事项：

• 在实验过程中，采用保存checkpoint的方式（每n轮保存一次）保存serving模型时，进行本地调用时出错，所以最好在训练过程中只保存一次模型。

Tensorflow Serving Client端（在这里使用第三方包tensorflow_serving_client）

---

固定request_input的shape，向request中放一条数据

（如果构建模型时输入的维度是固定的，应该将shape字段设置为模型输入的shape）

from tensorflow_serving_client.protos import predict_pb2, prediction_service_pb2
from grpc.beta import implementations
import tensorflow as tf
import time

from keras.preprocessing.sequence import pad_sequences
from utils import data_utils,data_process
from lib import config

#文件读取和处理
vocb, rev_vocb = data_utils.initialize_vocabulary("./vocab")
test_sentence_ = ["我", "讨厌", "这", "车"]
test_token_sentence = [[vocb.get(i.encode('utf-8'), 1) for i in test_sentence_]]
#将多条数据放到一个request中：
# for i in range(128):
#     test_token_sentence.append([vocb.get(i.encode('utf-8'), 1) for i in test_sentence_])
padding_sentence = pad_sequences(test_token_sentence, maxlen=config.MAX_SEQUENCE_LENGTH)
#计时
start_time = time.time()
#建立连接
channel = implementations.insecure_channel("服务的IP地址", 端口号)
stub = prediction_service_pb2.beta_create_PredictionService_stub(channel)
request = predict_pb2.PredictRequest()
#这里由保存和运行时定义，第一个启动tensorflow serving时配置的model_name，第二个是保存模型时的方法名
request.model_spec.name = "sentiment_classification"
request.model_spec.signature_name = "serving_default"
#入参参照入参定义
request.inputs["input_sentence"].ParseFromString(tf.contrib.util.make_tensor_proto(padding_sentence,
                                                                                   dtype=tf.int64,
                                                                                   shape=[1, 50]).SerializeToString())
#第二个参数是最大等待时间，因为这里是block模式访问的
response = stub.Predict(request, 10.0)
results = {}
for key in response.outputs:
    tensor_proto = response.outputs[key]
    nd_array = tf.contrib.util.make_ndarray(tensor_proto)
    results[key] = nd_array
print("cost %ss to predict: " % (time.time() - start_time))
print(results["predict_classification"])
# print(results["predict_scores"])

结果：

cost 0.7014968395233154s to predict:
predict label is: [0]

不固定request_input的shape，将 n 条数据放在同一个请求中（实验中n=128）

from tensorflow_serving_client.protos import predict_pb2, prediction_service_pb2
from grpc.beta import implementations
import tensorflow as tf
import time

from keras.preprocessing.sequence import pad_sequences
from utils import data_utils,data_process
from lib import config

#文件读取和处理
vocb, rev_vocb = data_utils.initialize_vocabulary(config.VOCABULARY_PATH)
test_sentence_ = ["我", "讨厌", "这", "车"]
test_token_sentence = [[vocb.get(i.encode('utf-8'), 1) for i in test_sentence_]]
#将多条数据放到一个request中：
for i in range(128):
    test_token_sentence.append([vocb.get(i.encode('utf-8'), 1) for i in test_sentence_])
padding_sentence = pad_sequences(test_token_sentence, maxlen=config.MAX_SEQUENCE_LENGTH)
#计时
start_time = time.time()
#建立连接
channel = implementations.insecure_channel("部署服务的IP地址", 端口号)
stub = prediction_service_pb2.beta_create_PredictionService_stub(channel)
request = predict_pb2.PredictRequest()
#这里由保存和运行时定义，第一个启动tensorflow serving时配置的model_name，第二个是保存模型时的方法名
request.model_spec.name = "sentiment_classification"
request.model_spec.signature_name = "serving_default"
#入参参照入参定义
request.inputs["input_sentence"].ParseFromString(tf.contrib.util.make_tensor_proto(padding_sentence,
                                                                                   dtype=tf.int64).SerializeToString())
#第二个参数是最大等待时间，因为这里是block模式访问的
response = stub.Predict(request, 10.0)
results = {}
for key in response.outputs:
    tensor_proto = response.outputs[key]
    nd_array = tf.contrib.util.make_ndarray(tensor_proto)
    results[key] = nd_array
print("cost %ss to predict: " % (time.time() - start_time))
print("predict label is:",results["predict_classification"])
# print(results["predict_scores"])

结果：

cost 0.7941889762878418s to predict:
predict label is: [0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0]

第二种调用方式处理的数据远多于第一种调用方式处理的数据，但是所用的时间跟第一种方式差不多，这可能是由于Tensorflow Serving 在处理请求时，采用了并行计算的方式，降低了计算耗时。

注意事项

• 对request.inputs进行定义时，数据类型dtype要与保存模型时对应的tensor的数据类型保持一致。例如：定义graph中tensor的数据类型为tf.int64，那么request.inputs也应该定义为tf.int64,如果定义为别的数据类型，可能会损失精度;
• 客户端调用成功后，建议将客户端请求结果与本地调用模型的结果进行对比，验证结果是否正确；