名称: tensorflow-model-deployment 描述: 部署和服务TensorFlow模型允许工具: [Bash, Read]

TensorFlow模型部署

使用SavedModel格式、TensorFlow Lite用于移动和边缘设备、量化技术以及服务基础设施，将TensorFlow模型部署到生产环境。此技能涵盖模型导出、优化、转换和部署策略。

SavedModel导出

基本SavedModel导出

# 保存模型到TensorFlow SavedModel格式
model.save('path/to/saved_model')

# 加载SavedModel
loaded_model = tf.keras.models.load_model('path/to/saved_model')

# 使用加载的模型进行预测
predictions = loaded_model.predict(test_data)

创建服务模型

# 从分类器创建服务模型
serving_model = classifier.create_serving_model()

# 检查模型输入和输出
print(f'Model\'s input shape and type: {serving_model.inputs}')
print(f'Model\'s output shape and type: {serving_model.outputs}')

# 保存服务模型
serving_model.save('model_path')

导出带签名

# 定义服务签名
@tf.function(input_signature=[tf.TensorSpec(shape=[None, 224, 224, 3], dtype=tf.float32)])
def serve(images):
    return model(images, training=False)

# 保存带签名
tf.saved_model.save(
    model,
    'saved_model_dir',
    signatures={'serving_default': serve}
)

TensorFlow Lite转换

基本TFLite转换

# 转换SavedModel到TFLite
converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
tflite_model = converter.convert()

# 保存TFLite模型
with open('model.tflite', 'wb') as f:
    f.write(tflite_model)

从Keras模型转换

# 直接转换Keras模型到TFLite
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()

# 保存到文件
import pathlib
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)

从具体函数转换

# 从具体函数转换
concrete_function = model.signatures['serving_default']

converter = tf.lite.TFLiteConverter.from_concrete_functions(
    [concrete_function]
)
tflite_model = converter.convert()

使用Model Maker导出

# 使用元数据导出训练模型到TFLite
model.export(
    export_dir='output/',
    tflite_filename='model.tflite',
    label_filename='labels.txt',
    vocab_filename='vocab.txt'
)

# 导出多格式
model.export(
    export_dir='output/',
    export_format=[
        mm.ExportFormat.TFLITE,
        mm.ExportFormat.SAVED_MODEL,
        mm.ExportFormat.LABEL
    ]
)

模型量化

后训练Float16量化

from tflite_model_maker.config import QuantizationConfig

# 创建float16量化配置
config = QuantizationConfig.for_float16()

# 带量化导出
model.export(
    export_dir='.',
    tflite_filename='model_fp16.tflite',
    quantization_config=config
)

动态范围量化

# 使用动态范围量化转换
converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

# 保存量化模型
with open('model_quantized.tflite', 'wb') as f:
    f.write(tflite_model)

全整数量化

def representative_dataset():
    """生成用于校准的代表性数据集。"""
    for i in range(100):
        yield [np.random.rand(1, 224, 224, 3).astype(np.float32)]

# 使用全整数量化转换
converter = 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]
converter.inference_input_type = tf.int8
converter.inference_output_type = tf.int8

tflite_model = converter.convert()

调试量化

from tensorflow.lite.python import convert

# 创建带数值验证的调试模型
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
calibrated = converter.convert()
debug_model = convert.mlir_quantize(calibrated, enable_numeric_verify=True)

获取量化转换器

# 应用量化设置到转换器
def get_converter_with_quantization(converter, **kwargs):
    """应用量化配置到转换器。"""
    config = QuantizationConfig(**kwargs)
    return config.get_converter_with_quantization(converter)

# 使用自定义设置
converter = tf.lite.TFLiteConverter.from_keras_model(model)
quantized_converter = get_converter_with_quantization(
    converter,
    optimizations=[tf.lite.Optimize.DEFAULT],
    representative_dataset=representative_dataset
)
tflite_model = quantized_converter.convert()

JAX到TFLite转换

基本JAX转换

from orbax.export import ExportManager
from orbax.export import JaxModule
from orbax.export import ServingConfig
import tensorflow as tf
import jax.numpy as jnp

def model_fn(_, x):
    return jnp.sin(jnp.cos(x))

jax_module = JaxModule({}, model_fn, input_polymorphic_shape='b, ...')

# 选项1: 直接SavedModel转换
tf.saved_model.save(
    jax_module,
    '/some/directory',
    signatures=jax_module.methods[JaxModule.DEFAULT_METHOD_KEY].get_concrete_function(
        tf.TensorSpec(shape=(None,), dtype=tf.float32, name="input")
    ),
    options=tf.saved_model.SaveOptions(experimental_custom_gradients=True),
)
converter = tf.lite.TFLiteConverter.from_saved_model('/some/directory')
tflite_model = converter.convert()

JAX带预处理/后处理

# 选项2: 带预处理和后处理
serving_config = ServingConfig(
    'Serving_default',
    input_signature=[tf.TensorSpec(shape=(None,), dtype=tf.float32, name='input')],
    tf_preprocessor=lambda x: x,
    tf_postprocessor=lambda out: {'output': out}
)
export_mgr = ExportManager(jax_module, [serving_config])
export_mgr.save('/some/directory')

converter = tf.lite.TFLiteConverter.from_saved_model('/some/directory')
tflite_model = converter.convert()

JAX ResNet50示例

from orbax.export import ExportManager, JaxModule, ServingConfig

# 将模型参数和函数包装到JaxModule
jax_module = JaxModule({}, jax_model.apply, trainable=False)

# 指定服务配置并导出模型
serving_config = ServingConfig(
    "serving_default",
    input_signature=[tf.TensorSpec([480, 640, 3], tf.float32, name="inputs")],
    tf_preprocessor=resnet_image_processor,
    tf_postprocessor=lambda x: tf.argmax(x, axis=-1),
)

export_manager = ExportManager(jax_module, [serving_config])

saved_model_dir = "resnet50_saved_model"
export_manager.save(saved_model_dir)

# 转换到TFLite
converter = tf.lite.TFLiteConverter.from_saved_model(saved_model_dir)
tflite_model = converter.convert()

模型优化

图变换

# 构建图变换工具
bazel build 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' \
--transforms='
  strip_unused_nodes(type=float, shape="1,299,299,3")
  remove_nodes(op=Identity, op=CheckNumerics)
  fold_constants(ignore_errors=true)
  fold_batch_norms
  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' \
--transforms='
  strip_unused_nodes(type=float, shape="1,299,299,3")
  fold_constants(ignore_errors=true)
  fold_batch_norms
  fold_old_batch_norms'

EfficientDet部署

导出SavedModel

def export_saved_model(
    model: tf.keras.Model,
    saved_model_dir: str,
    batch_size: Optional[int] = None,
    pre_mode: Optional[str] = 'infer',
    post_mode: Optional[str] = 'global'
) -> None:
    """导出EfficientDet模型到SavedModel格式。

    参数:
        model: 用于训练的EfficientDetNet模型
        saved_model_dir: 保存模型的文件夹路径
        batch_size: 保存到saved_model的批次大小
        pre_mode: 预处理模式（'infer' 或 None）
        post_mode: 后处理模式（'global', 'per_class', 'tflite', 或 None）
    """
    # 实现导出带指定配置的模型
    tf.saved_model.save(model, saved_model_dir)

完整导出流水线

# 导出模型带所有格式
export_saved_model(
    model=my_keras_model,
    saved_model_dir="./saved_model_export",
    batch_size=1,
    pre_mode='infer',
    post_mode='global'
)

# 转换到TFLite
converter = tf.lite.TFLiteConverter.from_saved_model('./saved_model_export')
tflite_model = converter.convert()

# 保存TFLite模型
with open('efficientdet.tflite', 'wb') as f:
    f.write(tflite_model)

移动部署

部署到Android

# 推送TFLite模型到Android设备
adb push 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模型并分配张量
interpreter = tf.lite.Interpreter(model_path='model.tflite')
interpreter.allocate_tensors()

# 获取输入和输出详情
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# 准备输入数据
input_shape = input_details[0]['shape']
input_data = np.array(np.random.random_sample(input_shape), dtype=np.float32)

# 运行推理
interpreter.set_tensor(input_details[0]['index'], input_data)
interpreter.invoke()

# 获取预测
output_data = interpreter.get_tensor(output_details[0]['index'])
print(output_data)

分布式训练和服务

多GPU镜像策略

# 创建策略实例。它会自动检测所有GPU。
mirrored_strategy = tf.distribute.MirroredStrategy()

# 在strategy.scope()下创建和编译keras模型
with mirrored_strategy.scope():
    model = tf.keras.Sequential([tf.keras.layers.Dense(1, input_shape=(1,))])
    model.compile(loss='mse', optimizer='sgd')

# 调用model.fit和model.evaluate，如前所述。
dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(10)
model.fit(dataset, epochs=2)
model.evaluate(dataset)

# 保存分布式模型
model.save('distributed_model')

TPU变量优化

# 在MLIR中优化TPU变量重新格式化
# 优化前:
var0 = ...
var1 = ...
tf.while_loop(..., var0, var1) {
    tf_device.replicate([var0, var1] as rvar) {
        compile = tf._TPUCompileMlir()
        tf.TPUExecuteAndUpdateVariablesOp(rvar, compile)
    }
}

# 优化后带状态变量:
var0 = ...
var1 = ...
state_var0 = ...
state_var1 = ...
tf.while_loop(..., var0, var1, state_var0, state_var1) {
    tf_device.replicate(
        [var0, var1] as rvar,
        [state_var0, state_var1] as rstate
    ) {
        compile = tf._TPUCompileMlir()
        tf.TPUReshardVariablesOp(rvar, compile, rstate)
        tf.TPUExecuteAndUpdateVariablesOp(rvar, compile)
    }
}

使用TensorFlow Serving服务模型

导出到TensorFlow Serving

# 使用版本号导出模型
export_path = os.path.join('serving_models', 'my_model', '1')
tf.saved_model.save(model, export_path)

# 导出多个版本
for version in [1, 2, 3]:
    export_path = os.path.join('serving_models', 'my_model', str(version))
    tf.saved_model.save(model, export_path)

Docker部署

# 拉取TensorFlow Serving镜像
docker pull tensorflow/serving

# 运行TensorFlow Serving容器
docker run -p 8501:8501 \
  --mount type=bind,source=/path/to/my_model,target=/models/my_model \
  -e MODEL_NAME=my_model \
  -t tensorflow/serving

# 测试REST API
curl -d '{"instances": [[1.0, 2.0, 3.0, 4.0]]}' \
  -X POST http://localhost:8501/v1/models/my_model:predict

模型验证和测试

验证TFLite模型

# 比较TFLite预测与原始模型
def validate_tflite_model(model, tflite_model_path, test_data):
    """验证TFLite模型与原始模型。"""
    # 原始模型预测
    original_predictions = model.predict(test_data)

    # TFLite模型预测
    interpreter = tf.lite.Interpreter(model_path=tflite_model_path)
    interpreter.allocate_tensors()

    input_details = interpreter.get_input_details()
    output_details = interpreter.get_output_details()

    tflite_predictions = []
    for sample in test_data:
        interpreter.set_tensor(input_details[0]['index'], sample[np.newaxis, ...])
        interpreter.invoke()
        output = interpreter.get_tensor(output_details[0]['index'])
        tflite_predictions.append(output[0])

    tflite_predictions = np.array(tflite_predictions)

    # 比较预测
    difference = np.abs(original_predictions - tflite_predictions)
    print(f"平均绝对差异: {np.mean(difference):.6f}")
    print(f"最大绝对差异: {np.max(difference):.6f}")

模型大小比较

import os

def compare_model_sizes(saved_model_path, tflite_model_path):
    """比较SavedModel和TFLite的大小。"""
    # SavedModel大小（所有文件总和）
    saved_model_size = sum(
        os.path.getsize(os.path.join(dirpath, filename))
        for dirpath, _, filenames in os.walk(saved_model_path)
        for filename in filenames
    )

    # 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 / saved_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）
创建设备上ML应用
为特定硬件加速器优化模型

最佳实践

始终验证转换后的模型 - 比较TFLite预测与原始模型以确保准确性
使用SavedModel格式 - 生产部署和服务的标准格式
应用适当的量化 - Float16用于平衡速度/准确性，INT8用于最大压缩
将预处理嵌入模型 - 在SavedModel中嵌入预处理以实现一致推理
版本化您的模型 - 在导出路径中使用版本号进行模型管理
在目标设备上测试 - 在实际部署硬件上验证性能
监控模型大小 - 跟踪优化前后的模型大小
使用代表性数据集 - 提供校准数据以进行准确量化
启用GPU代理 - 在支持的设备上使用GPU/TPU加速
优化批次大小 - 调整批次大小以平衡吞吐量与延迟
缓存常用模型 - 加载模型一次并重复用于多个预测
使用TensorFlow Serving - 利用内置服务基础设施实现可扩展性
实施模型预热 - 运行虚拟预测以初始化服务系统
监控推理指标 - 在生产中跟踪延迟、吞吐量和错误率
在TFLite中使用元数据 - 在模型元数据中包含标签和预处理信息

常见陷阱

未验证转换后的模型 - TFLite转换可能导致准确性降低
过度激进的量化 - 没有校准的INT8量化导致准确性损失
缺少代表性数据集 - 没有校准的量化产生差的结果
忽略模型大小 - 大型模型无法在内存受限的设备上部署
未在目标硬件上测试 - 性能在不同设备间显著变化
硬编码预处理 - 客户端预处理导致不一致
错误的输入/输出类型 - 模型和推理代码之间的类型不匹配
未使用批次推理 - 单样本推理对高吞吐量效率低下
缺少错误处理 - 生产系统需要健壮的错误处理
未监控模型漂移 - 模型性能随时间退化而没有监控
不正确的张量形状 - 形状不匹配导致运行时错误
未为目设备优化 - 通用优化不利用设备特定功能
忘记模型版本化 - 没有版本难以回滚或A/B测试
不使用GPU加速 - 在有能力设备上仅CPU推理慢得多
部署未测试的模型 - 在生产部署前始终验证模型

名称: tensorflow-model-deployment 描述: 部署和服务TensorFlow模型 允许工具: [Bash, Read]

TensorFlow模型部署

SavedModel导出

基本SavedModel导出

创建服务模型

导出带签名

TensorFlow Lite转换

基本TFLite转换

从Keras模型转换

从具体函数转换

使用Model Maker导出

模型量化

后训练Float16量化

动态范围量化

全整数量化

调试量化

获取量化转换器

JAX到TFLite转换

基本JAX转换

JAX带预处理/后处理

JAX ResNet50示例

模型优化

图变换

修复移动内核错误

EfficientDet部署

导出SavedModel

完整导出流水线

移动部署

部署到Android

TFLite解释器使用

分布式训练和服务

多GPU镜像策略

TPU变量优化

使用TensorFlow Serving服务模型

导出到TensorFlow Serving

Docker部署

模型验证和测试

验证TFLite模型

模型大小比较

何时使用此技能

最佳实践

常见陷阱

资源

名称: tensorflow-model-deployment 描述: 部署和服务TensorFlow模型允许工具: [Bash, Read]