# TensorFlow 模型部署使用 SavedModel 格式、适用于移动和边缘设备的 TensorFlow Lite、量化技术和服务基础设施将 TensorFlow 模型部署到生产环境。该技能涵盖模型导出、优化、转换和部署策略。 ## 保存模型导出 ### 基本 SavedModel 导出 # 将模型保存为 TensorFlow SavedModel 格式 model.save('path/to/saved_model') # 加载保存的模型 load_model = tf.keras.models.load_model('path/to/saved_model') # 使用加载的模型进行预测预测=loaded_model.predict(test_data) ### 创建服务模型 # 从分类器创建服务模型服务模型 = classifier.create_serving_model() # 检查模型输入和输出 print(f'模型的输入形状和类型：{serving_model.inputs}') print(f'模型的输出形状和类型：{serving_model.outputs}') # 保存服务模型 serving_model.save('model_path') ### 带签名导出 # 定义服务签名 @tf.function(input_signature=[tf.TensorSpec(shape=[无, 224, 224, 3], dtype=tf.float32)]) def 服务（图像）：返回模型（图像，训练=False） # 保存并签名 tf.saved_model.save( 模型, '保存的模型目录', 签名={'serving_default': 服务} ） ## TensorFlow Lite 转换 ### 基本 TFLite 转换 # 将 SavedModel 转换为 TFLite 转换器 = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir') tflite_model =转换器.convert() # 保存 TFLite 模型将 open('model.tflite', 'wb') 作为 f： f.write(tflite_model) ### 来自 Keras 模型 # 将 Keras 模型直接转换为 TFLite 转换器 = tf.lite.TFLiteConverter.from_keras_model(模型) tflite_model =转换器.convert() # 保存到文件导入路径库 tflite_models_dir = pathlib.Path("tflite_models/") tflite_models_dir.mkdir（exist_ok = True，parents = True） tflite_model_file = tflite_models_dir /“mnist_model.tflite” tflite_model_file.write_bytes(tflite_model) ### 来自具体函数 # 从具体函数转换具体函数 = model.signatures['serving_default'] 转换器 = tf.lite.TFLiteConverter.from_concrete_functions( [具体功能] ） tflite_model =转换器.convert() ### 使用模型制作工具导出 # 将训练好的模型连同元数据导出到 TFLite 模型.导出（ export_dir='输出/', tflite_filename='model.tflite', label_filename='标签.txt', vocab_filename='vocab.txt' ） # 导出多种格式模型.导出（ export_dir='输出/', 导出格式=[ mm.ExportFormat.TFLITE, mm.ExportFormat.SAVED_MODEL, mm.ExportFormat.LABEL ] ） ## 模型量化 ### 训练后 Float16 量化从 tflite_model_maker.config 导入 QuantizationConfig # 创建 float16 量化配置配置 = QuantizationConfig.for_float16() # 量化导出模型.导出（导出目录='.', tflite_filename='model_fp16.tflite', quantization_config=配置） ### 动态范围量化 # 使用动态范围量化进行转换转换器 = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir') converter.optimizations = [tf.lite.Optimize.DEFAULT] tflite_model =转换器.convert() # 保存量化模型使用 open('model_quantized.tflite', 'wb') 作为 f： f.write(tflite_model) ### 全整数量化 defrepresentative_dataset(): """生成用于校准的代表性数据集。""" 对于范围（100）内的 i：产量 [np.random.rand(1, 224, 224, 3).astype(np.float32)] # 使用全整数量化进行转换转换器 = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir') converter.optimizations = [tf.lite.Optimize.DEFAULT] converter.representative_dataset =representative_dataset converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8] 转换器.inference_input_type = tf.int8 转换器.inference_output_type = tf.int8 tflite_model =转换器.convert() ### 调试量化从tensorflow.lite.python导入转换 # 创建带有数值验证的调试模型 converter.optimizations = [tf.lite.Optimize.DEFAULT] converter.representative_dataset =calibration_gen converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8] # 通过验证进行校准和量化 converter._experimental_calibrate_only = True 校准=转换器.convert() debug_model = Convert.mlir_quantize（已校准，enable_numeric_verify=True） ### 获取量化转换器 # 将量化设置应用于转换器 def get_converter_with_quantization(转换器, **kwargs): """将量化配置应用于转换器。""" 配置=量化配置（**kwargs）返回 config.get_converter_with_quantization(转换器) # 与自定义设置一起使用转换器 = tf.lite.TFLiteConverter.from_keras_model(模型) 量化转换器 = get_converter_with_quantization( 转换器，优化=[tf.lite.Optimize.DEFAULT], 代表数据集=代表数据集） tflite_model = quantized_converter.convert() ## JAX 到 TFLite 的转换 ### 基本 JAX 转换从 orbax.export 导入 ExportManager 从 orbax.export 导入 JaxModule 从 orbax.export 导入 ServingConfig 将张量流导入为 tf 将 jax.numpy 导入为 jnp def model_fn(_, x): 返回 jnp.sin(jnp.cos(x)) jax_module = JaxModule({}, model_fn, input_polymorphic_shape='b, ...') # 选项1：直接SavedModel转换 tf.saved_model.save( jax_模块， '/一些/目录', 签名=jax_module.methods[JaxModule.DEFAULT_METHOD_KEY].get_concrete_function( tf.TensorSpec(shape=(None,), dtype=tf.float32, name="input") ), 选项=tf.saved_model.SaveOptions(experimental_custom_gradients=True), ）转换器 = tf.lite.TFLiteConverter.from_saved_model('/some/directory') tflite_model =转换器.convert() ### JAX 与前/后处理 # 选项 2：使用预处理和后处理服务配置 = 服务配置（ '默认服务', input_signature=[tf.TensorSpec(shape=(None,), dtype=tf.float32, name='input')], tf_preprocessor=lambda x: x, tf_postprocessor=lambda 输出：{'输出'：输出} ） export_mgr = ExportManager(jax_module, [serving_config]) export_mgr.save('/some/directory') 转换器 = tf.lite.TFLiteConverter.from_saved_model('/some/directory') tflite_model =转换器.convert() ### JAX ResNet50 示例从 orbax.export 导入 ExportManager、JaxModule、ServingConfig # 将模型参数和函数包装到 JaxModule 中 jax_module = JaxModule({}, jax_model.apply, 可训练=False) # 指定服务配置并导出模型服务配置 = 服务配置（ “服务默认”， input_signature=[tf.TensorSpec([480, 640, 3], tf.float32, name="输入")], tf_preprocessor=resnet_image_processor, tf_postprocessor=lambda x: tf.argmax(x, axis=-1), ） export_manager = ExportManager(jax_module, [serving_config]) save_model_dir =“resnet50_saved_model” 导出管理器.保存（保存的模型目录） # 转换为 TFLite 转换器 = tf.lite.TFLiteConverter.from_saved_model(saved_model_dir) tflite_model =转换器.convert() ## 模型优化 ### 图转换 # 构建图转换工具 bazel构建tensorflow/tools/graph_transforms:transform_graph # 优化部署 bazel-bin/tensorflow/tools/graph_transforms/transform_graph \ --in_graph=tensorflow_inception_graph.pb \ --out_graph=optimized_inception_graph.pb \ --inputs='Mul' \ --outputs='softmax' \ --变换=' strip_unused_nodes（类型=浮动，形状=“1,299,299,3”） remove_nodes(op=Identity, op=CheckNumerics) Fold_constants（ignore_errors = true）折叠批量规范 Fold_old_batch_norms' ### 修复移动内核错误 # 优化移动部署 bazel-bin/tensorflow/tools/graph_transforms/transform_graph \ --in_graph=tensorflow_inception_graph.pb \ --out_graph=optimized_inception_graph.pb \ --inputs='Mul' \ --outputs='softmax' \ --变换=' strip_unused_nodes（类型=浮动，形状=“1,299,299,3”） Fold_constants（ignore_errors = true）折叠批量规范 Fold_old_batch_norms' ## 高效Det 部署 ### 导出 SavedModel def 导出保存的模型( 模型：tf.keras.Model，保存的模型目录：str，批量大小：可选[int] =无， pre_mode: 可选[str] = '推断', post_mode: 可选[str] = '全局' ) -> 无: """将 EfficientDet 模型导出为 SavedModel 格式。参数： model：用于训练的EfficientDetNet模型 save_model_dir：保存模型的文件夹路径 batch_size：要保存在saved_model中的批量大小 pre_mode：预处理模式（'infer' 或 None） post_mode：后处理模式（'global'、'per_class'、'tflite' 或 None） ”“” # 实现导出具有指定配置的模型 tf.saved_model.save（模型，saved_model_dir） ### 完整的出口管道 # 导出所有格式的模型导出保存的模型（模型=my_keras_model， save_model_dir="./saved_model_export", 批量大小=1， pre_mode='推断', post_mode='全局' ） # 转换为 TFLite 转换器 = tf.lite.TFLiteConverter.from_saved_model('./saved_model_export') tflite_model =转换器.convert() # 保存 TFLite 模型使用 open('efficientdet.tflite', 'wb') 作为 f： f.write(tflite_model) ## 移动部署 ### 部署到 Android # 将 TFLite 模型推送到 Android 设备 adb 推送 mobilenet_quant_v1_224.tflite /data/local/tmp # 在设备上运行基准测试 adb shell /data/local/tmp/benchmark_model \ --graph=/data/local/tmp/mobilenet_quant_v1_224.tflite \ --num_threads=4 ### TFLite 解释器用法 # 加载TFLite模型并分配张量解释器 = tf.lite.Interpreter(model_path='model.tflite') 解释器.allocate_tensors() # 获取输入和输出详细信息 input_details =terpreter.get_input_details() 输出详细信息=解释器.get_output_details() # 准备输入数据 input_shape = input_details[0]['形状'] input_data = np.array(np.random.random_sample(input_shape), dtype=np.float32) # 运行推理 terpreter.set_tensor(input_details[0]['index'], input_data) 解释器.invoke() # 获取预测 output_data =terpreter.get_tensor(output_details[0]['index']) 打印（输出数据） ## 分布式训练和服务 ### 多 GPU 的 MirroredStrategy # 创建策略实例。它将自动检测所有 GPU。镜像策略 = tf.distribute.MirroredStrategy() # 在strategy.scope()下创建并编译keras模型使用mirrored_strategy.scope()：模型 = tf.keras.Sequential([tf.keras.layers.Dense(1, input_shape=(1,))]) model.compile(loss='mse', 优化器='sgd') # 像以前一样调用 model.fit 和 model.evaluate 。数据集 = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(10) model.fit(数据集, epochs=2) 模型.评估（数据集） # 保存分布式模型 model.save('分布式模型') ### TPU 变量优化 # 优化 MLIR 中的 TPU 变量重新格式化 # 优化前： var0 = ... 变量1 = ... tf.while_loop(..., var0, var1) { tf_device.replicate([var0, var1] as rvar) { 编译 = tf._TPUCompileMlir() tf.TPUExecuteAndUpdateVariablesOp(rvar, 编译) } } # 使用状态变量优化后： var0 = ... 变量1 = ... 状态_var0 = ... 状态_var1 = ... tf.while_loop(..., var0, var1, state_var0, state_var1) { tf_device.复制（ [var0, var1] 作为 rvar， [state_var0, state_var1] 作为 rstate ）{ 编译 = tf._TPUCompileMlir() tf.TPUReshardVariablesOp(rvar, 编译, rstate) tf.TPUExecuteAndUpdateVariablesOp(rvar, 编译) } } ## 使用 TensorFlow Serving 进行模型服务 ### 导出 TensorFlow 服务 # 导出带有版本号的模型 export_path = os.path.join('serving_models', 'my_model', '1') tf.saved_model.save（模型，导出路径） # 导出多个版本对于 [1, 2, 3] 中的版本： export_path = os.path.join('serving_models', 'my_model', str(版本)) tf.saved_model.save（模型，导出路径） ### Docker 部署 # 拉取 TensorFlow Serving 镜像 docker拉张量流/服务 # 运行 TensorFlow 服务容器 docker运行-p 8501:8501 \ --挂载类型=绑定，源=/path/to/my_model，目标=/models/my_model \ -e MODEL_NAME=我的模型 \ -t 张量流/服务 # 测试 REST API 卷曲 -d '{"实例": [[1.0, 2.0, 3.0, 4.0]]}' \ -X POST http://localhost:8501/v1/models/my_model:predict ## 模型验证和测试 ### 验证 TFLite 模型 # 将 TFLite 预测与原始模型进行比较 def validate_tflite_model（模型，tflite_model_path，test_data）： """根据原始模型验证 TFLite 模型。""" # 原始模型预测原始预测 = model.predict(test_data) # TFLite 模型预测解释器 = tf.lite.Interpreter(model_path=tflite_model_path) 解释器.allocate_tensors() input_details =terpreter.get_input_details() 输出详细信息=解释器.get_output_details() tflite_predictions = [] 对于 test_data 中的样本： terpreter.set_tensor(input_details[0]['index'], 样本[np.newaxis, ...]) 解释器.invoke() 输出 =terpreter.get_tensor(output_details[0]['index']) tflite_predictions.append(输出[0]) tflite_predictions = np.array(tflite_predictions) # 比较预测差异 = np.abs(original_predictions - tflite_predictions) print(f"平均绝对差: {np.mean(difference):.6f}") print(f"最大绝对差：{np.max(difference):.6f}") ### 模型尺寸比较导入操作系统 def Compare_model_sizes(saved_model_path, tflite_model_path): """比较 SavedModel 和 TFLite 的大小。""" # SavedModel 大小（所有文件的总和）保存的模型大小=总和（ os.path.getsize(os.path.join(dirpath, 文件名)) 对于 os.walk(saved_model_path) 中的 dirpath、_、文件名对于文件名中的文件名） # TFLite 模型大小 tflite_size = os.path.getsize(tflite_model_path) print(f"SavedModel 大小：{saved_model_size / 1e6:.2f} MB") print(f"TFLite 模型大小：{tflite_size / 1e6:.2f} MB") print(f"尺寸缩小：{(1 - tflite_size / saving_model_size) * 100:.1f}%") ## 何时使用此技能当您需要执行以下操作时，请使用tensorflow-model-deployment技能： * 导出经过训练的模型用于生产服务 * 将模型部署到移动设备（iOS、Android） * 优化边缘设备和物联网的模型 * 将模型转换为 TensorFlow Lite 格式 * 应用训练后量化进行模型压缩 * 设置 TensorFlow Serving 基础设施 * 使用 Docker 容器部署模型 * 使用 REST 或 gRPC 创建模型服务 API * 优化推理延迟和吞吐量 * 减小带宽受限环境的模型大小 * 将 JAX 或 PyTorch 模型转换为 TensorFlow 格式 * 实现多个模型版本的A/B测试 * 将模型部署到云平台（GCP、AWS、Azure） * 创建设备上的机器学习应用程序 * 针对特定硬件加速器优化模型 ## 最佳实践 1. **始终验证转换后的模型** \- 将 TFLite 预测与原始模型进行比较以确保准确性 2. **使用 SavedModel 格式** \- 用于生产部署和服务的标准格式 3. **应用适当的量化** \- Float16 用于平衡速度/精度，INT8 用于最大压缩 4. **在模型中包含预处理** \- 在 SavedModel 中嵌入预处理以实现一致的推理 5. **对模型进行版本控制** \- 在导出路径中使用版本号进行模型管理 6. **在目标设备上测试** \- 验证实际部署硬件上的性能 7. **监控模型大小** \- 跟踪优化前后的模型大小 8. **使用代表性数据集** \- 提供校准数据以进行准确量化 9. **启用 GPU 委派** \- 在支持的设备上使用 GPU/TPU 加速 10. **优化批量大小** \- 调整批量大小以实现吞吐量与延迟的权衡 11. **缓存常用模型** \- 加载模型一次并重用于多个预测 12. **使用 TensorFlow Serving** \- 利用内置服务基础设施实现可扩展性 13. **实现模型预热** \- 运行虚拟预测来初始化服务系统 14. **监控推理指标** \- 跟踪生产中的延迟、吞吐量和错误率 15. **在 TFLite 中使用元数据** \- 在模型元数据中包含标签和预处理信息 ## 常见陷阱 1. **不验证转换后的模型** \- TFLite 转换会导致精度下降 2. **过度量化** \- 未经校准的 INT8 量化会导致精度损失 3. **缺少代表性数据集** \- 未经校准的量化会产生较差的结果 4. **忽略模型大小** \- 大型模型无法部署在内存受限的设备上 5. **未在目标硬件上进行测试** \- 不同设备的性能差异很大 6. **硬编码预处理** \- 客户端预处理导致不一致 7. **错误的输入/输出类型** \- 模型和推理代码之间的类型不匹配 8. **不使用批量推理** \- 单样本推理对于高吞吐量来说效率低下 9. **缺少错误处理** \-生产系统需要强大的错误处理 10. **不监控模型漂移** \- 在不监控的情况下模型性能会随着时间的推移而下降 11. **不正确的张量形状** \- 形状不匹配导致运行时错误 12. **不针对目标设备进行优化** \- 通用优化不利用特定于设备的功能 13. **忘记模型版本控制** \- 没有版本很难回滚或 A/B 测试 14. **不使用 GPU 加速** \- 仅 CPU 推理在有能力的设备上要慢得多 15. **部署未经测试的模型** \- 始终在生产部署之前验证模型 ## 资源 * [SavedModel格式指南]() * [TensorFlow Lite 指南]() * [训练后量化]() * [TensorFlow 服务]() * [模型优化工具包]() * [TFLite Android 指南]() * [TFLite iOS 指南]() * [图变换]() * [模型制作]() * [Orbax 导出 (JAX)]()

Tensorflow模型部署

包含技能