我是 TensorFlow Serving 的新手。我使用估算器训练了一个广泛而深入的模型。现在我想为我的模型服务。我创建我的服务输入接收器函数并保存模型。当我尝试使用保存的模型进行预测时，我总是收到 InternalError: Unable to get element as bytes。我真的不明白服务函数中应该包含什么或者我应该发送什么格式类型。有人可以先向我解释服务功能的概念以及如何正确创建功能。
可重现的例子:https://github.com/dangz90/wide_and_deep_debugging/blob/master/wide%20and%20deep%20debug.ipynb
服务功能:
def serving_input_receiver_fn():
serialized_tf_example = tf.placeholder(dtype=tf.string,
shape=[],
name='input_tensor')
receiver_tensors = {'inputs': serialized_tf_example}
parsed_features = tf.parse_single_example(
serialized_tf_example,
# Defaults are not specified since both keys are required.
features={
'var1': tf.FixedLenFeature(shape=[1], dtype=tf.string),
'var2': tf.FixedLenFeature(shape=[1], dtype=tf.string),
'var3': tf.FixedLenFeature(shape=[1], dtype=tf.string),
'var4': tf.VarLenFeature(dtype=tf.string),
})
return tf.estimator.export.ServingInputReceiver(parsed_features, receiver_tensors)
estimator_predictor = tf.contrib.predictor.from_estimator(m, serving_input_receiver_fn)
estimator_predictor({ 'inputs': examples.SerializeToString() })
我试过发送 pandas、示例等。但真的不知道数据应该是什么格式。对于我的训练，输入的数据被保存为 tfrecord，然后作为数据集加载。 注意:如果我直接使用模型 m.predict() 我能够得到正确的预测
完整错误:
--------------------------------------------------------------------------- InternalError Traceback (most recent call last) /databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in _do_call(self, fn, *args) 1333 try:
-> 1334 return fn(*args) 1335 except errors.OpError as e:
/databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in _run_fn(feed_dict, fetch_list, target_list, options, run_metadata) 1318 return self._call_tf_sessionrun(
-> 1319 options, feed_dict, fetch_list, target_list, run_metadata) 1320
/databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in _call_tf_sessionrun(self, options, feed_dict, fetch_list, target_list, run_metadata) 1406 self._session, options, feed_dict, fetch_list, target_list,
-> 1407 run_metadata) 1408
InternalError: Unable to get element as bytes.
During handling of the above exception, another exception occurred:
InternalError Traceback (most recent call last) <command-364073753108128> in <module>()
3
4 estimator_predictor = tf.contrib.predictor.from_estimator(m, serving_input_receiver_fn)
----> 5 estimator_predictor({ 'inputs': examples_ })
/databricks/python/lib/python3.6/site-packages/tensorflow/contrib/predictor/predictor.py in __call__(self, input_dict)
75 if value is not None:
76 feed_dict[self.feed_tensors[key]] = value
---> 77 return self._session.run(fetches=self.fetch_tensors, feed_dict=feed_dict)
/databricks/python/lib/python3.6/site-packages/tensorflow/python/training/monitored_session.py in run(self, fetches, feed_dict, options, run_metadata)
674 feed_dict=feed_dict,
675 options=options,
--> 676 run_metadata=run_metadata)
677
678 def run_step_fn(self, step_fn):
/databricks/python/lib/python3.6/site-packages/tensorflow/python/training/monitored_session.py in run(self, fetches, feed_dict, options, run_metadata) 1169 feed_dict=feed_dict, 1170 options=options,
-> 1171 run_metadata=run_metadata) 1172 except _PREEMPTION_ERRORS as e: 1173 logging.info('An error was raised. This may be due to a preemption in '
/databricks/python/lib/python3.6/site-packages/tensorflow/python/training/monitored_session.py in run(self, *args, **kwargs) 1268 raise six.reraise(*original_exc_info) 1269 else:
-> 1270 raise six.reraise(*original_exc_info) 1271 1272
/databricks/python/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None
/databricks/python/lib/python3.6/site-packages/tensorflow/python/training/monitored_session.py in run(self, *args, **kwargs) 1253 def run(self, *args,
**kwargs): 1254 try:
-> 1255 return self._sess.run(*args, **kwargs) 1256 except _PREEMPTION_ERRORS: 1257 raise
/databricks/python/lib/python3.6/site-packages/tensorflow/python/training/monitored_session.py in run(self, fetches, feed_dict, options, run_metadata) 1325 feed_dict=feed_dict, 1326 options=options,
-> 1327 run_metadata=run_metadata) 1328 1329 for hook in self._hooks:
/databricks/python/lib/python3.6/site-packages/tensorflow/python/training/monitored_session.py in run(self, *args, **kwargs) 1089 1090 def run(self, *args,
**kwargs):
-> 1091 return self._sess.run(*args, **kwargs) 1092 1093 def run_step_fn(self, step_fn, raw_session, run_with_hooks):
/databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in run(self, fetches, feed_dict, options, run_metadata)
927 try:
928 result = self._run(None, fetches, feed_dict, options_ptr,
--> 929 run_metadata_ptr)
930 if run_metadata:
931 proto_data = tf_session.TF_GetBuffer(run_metadata_ptr)
/databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in _run(self, handle, fetches, feed_dict, options, run_metadata) 1150 if final_fetches or final_targets or (handle and feed_dict_tensor): 1151 results = self._do_run(handle, final_targets, final_fetches,
-> 1152 feed_dict_tensor, options, run_metadata) 1153 else: 1154 results = []
/databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in _do_run(self, handle, target_list, fetch_list, feed_dict, options, run_metadata) 1326 if handle is None: 1327 return self._do_call(_run_fn, feeds, fetches, targets, options,
-> 1328 run_metadata) 1329 else: 1330 return self._do_call(_prun_fn, handle, feeds, fetches)
/databricks/python/lib/python3.6/site-packages/tensorflow/python/client/session.py in _do_call(self, fn, *args) 1346 pass 1347 message = error_interpolation.interpolate(message, self._graph)
-> 1348 raise type(e)(node_def, op, message) 1349 1350 def _extend_graph(self):
InternalError: Unable to get element as bytes.
以下是示例在序列化之前的样子。
features {
feature {
key: "var4"
value {
bytes_list {
value: "43"
value: "65"
value: "89"
value: "02"
}
}
}
feature {
key: "var3"
value {
bytes_list {
value: "0123194"
}
}
}
feature {
key: "var2"
value {
bytes_list {
value: "1243"
}
}
}
feature {
key: "var1"
value {
bytes_list {
value: "54"
}
}
}
}
为了获取示例，我运行以下脚本:
def serialise_input(data):
dict_feature = {}
for e in data.items():
if e[0] == "var4":
dict_feature.update({e[0]: Feature(bytes_list=BytesList(value=[m.encode('utf-8') for m in e[1]]))})
else:
dict_feature.update({e[0]: Feature(bytes_list=BytesList( value=[e[1].encode()] ))})
example = Example(features=Features(feature=dict_feature))
return example.SerializeToString()
# Serialize Input
raw_data = test.toPandas().dropna().iloc[0,:-1]
examples_ = serialise_input(raw_data)
提前致谢。
请您参考如下方法:
看起来你的 serialized_tf_example 占位符有 shape=[] 这是一个例子。你应该传递一个 tf.train.Example , 序列化为字符串:
# I assume here examples_ is a list of tf.train.Examples
example = examples_[0].SerializeToString()
estimator_predictor({ 'inputs': example })
如果您想提供一批示例而不是单个序列化示例，则需要改用 shape=[None] 和 tf.io.parse_example。
此外，您提供的示例应该定义了您在特征字典(例如 var1)中引用的特征，以便它们可以被正确解析。
服务函数指定您的模型在预测时应如何接收其输入 - 当您将经过训练的 TensorFlow 图/模型导出为 SavedModel 时。不像在训练中那样有一个事件的 TF session ，您通常使用占位符或 tf.data.Dataset 并且需要有状态的 python/TF 运行时执行，您希望您的 SavedModel 能够被序列化并写入磁盘，以便您可以使用 TF 进行部署在移动设备等上提供或运行它。
所以你导出这个 SavedModel - 服务输入是定义如何解析它发送的请求然后将其连接到你的模型的部分你可能想要发送它们的方式是 tf.train.Examples Protocol Buffer (序列化为字符串，以便可以在 RPC 中发送)。然后您将它们解析为一个特征字典，这样您的估算器就可以像理解您的训练数据一样理解它们。
