TensorFlow Serving模型部署与版本管理

# models.config 配置文件示例model_config_list {  config {    name: 'ctr_model'    base_path: '/models/ctr_model'    model_platform: 'tensorflow'    model_version_policy {      specific {        versions: [1, 2, 3]  # 保留版本1/2/3      }    }  }  config {    name: 'nlp_model'    base_path: '/models/nlp_model'    model_platform: 'tensorflow'    model_version_policy {      latest {        num_versions: 2  # 只保留最近2个版本      }    }  }}

# batching_parameters.config 配置示例max_batch_size { value: 32 }batch_timeout_micros { value: 2000 }num_batch_threads { value: 4 }max_enqueued_batches { value: 1000 } # 批处理效果模拟分析batch_sizes = [1, 8, 16, 32, 64, 128]for bs in batch_sizes:    # 模拟吞吐和延迟(简化模型)    throughput = min(bs * 500, 16000)  # 吞吐随batch增大但受GPU限制    latency = max(2, bs * 0.3)        # 延迟随batch增大    print(f"  batch={bs:3d}: 吞吐={throughput:5d} QPS, "          f"延迟={latency:.1f}ms")

# Kubernetes GPU共享部署配置(TF Serving Deployment)apiVersion: apps/v1kind: Deploymentmetadata:  name: tf-serving-ctrspec:  replicas: 2  template:    spec:      containers:      - name: tensorflow-serving        image: tensorflow/serving:latest-gpu        args:          - --model_name=ctr_model          - --model_base_path=/models/ctr_model          - --per_process_gpu_memory_fraction=0.4  # 限制40%显存        resources:          limits:            nvidia.com/gpu: 1  # 请求1个GPU(共享)          requests:            nvidia.com/gpu: 1

import jsonimport time  class TFServingSimulator:    """TF Serving推理服务模拟器"""     def __init__(self):        self.models = {}       # model_name -> {version: info}        self.traffic_split = {}  # model_name -> {version: ratio}        self.request_count = 0        self.batch_queue = []     def load_model(self, name, version, path):        """加载模型版本(模拟热加载)"""        if name not in self.models:            self.models[name] = {}        self.models[name][version] = {            'path': path,            'status': 'loading',            'load_time': time.time()        }        print(f"[加载] {name} v{version} 开始加载...")        # 模拟加载完成        self.models[name][version]['status'] = 'ready'        print(f"[加载] {name} v{version} 加载完成, 旧版本继续服务")     def set_version_policy(self, name, policy_type, num_versions=2):        """设置版本策略"""        if name not in self.models:            return        versions = sorted(self.models[name].keys(), reverse=True)        if policy_type == 'latest':            keep = versions[:1]        elif policy_type == 'specific':            keep = versions[:num_versions]        else:            keep = versions        # 卸载不需要的版本        for v in versions:            if v not in keep:                print(f"[卸载] {name} v{v} 已卸载")                del self.models[name][v]        print(f"[策略] {name}: 保留版本 {keep}")     def set_traffic_split(self, name, split_config):        """设置流量分流(A/B测试/金丝雀)"""        self.traffic_split[name] = split_config        print(f"[分流] {name}: {split_config}")     def canary_release(self, name, new_version, ratios):        """金丝雀发布: 逐步扩大新版本流量"""        print(f"\n=== 金丝雀发布: {name} v{new_version} ===")        for ratio in ratios:            old_ratio = round(1 - ratio, 2)            split = {                new_version - 1: old_ratio,                new_version: ratio            }            self.set_traffic_split(name, split)            # 模拟监控            error_rate = 0.001 if ratio < 0.5 else 0.002            print(f"  流量{ratio:.0%}: 错误率={error_rate:.2%}")            if error_rate > 0.01:                print("  !!! 错误率超阈值, 自动回滚 !!!")                self.set_traffic_split(name, {new_version - 1: 1.0})                return False        print(f"  金丝雀发布成功: 全量切换到 v{new_version}")        return True     def batch_inference(self, requests, max_batch_size=32,                        timeout_micros=2000):        """批处理推理模拟"""        results = []        batch = []        wait_time = 0        for req in requests:            batch.append(req)            if len(batch) >= max_batch_size:                # batch满了立即执行                results.extend(self._execute_batch(batch))                batch = []                wait_time = 0            else:                wait_time += timeout_micros / len(requests)        # 处理剩余请求        if batch:            results.extend(self._execute_batch(batch))        return results     def _execute_batch(self, batch):        """执行batch推理(模拟)"""        batch_size = len(batch)        # 模拟吞吐随batch增大但受GPU限制        throughput = min(batch_size * 500, 16000)        latency = max(2, batch_size * 0.3)        return [f"result_{i}" for i in range(batch_size)]  # 1. 模型热加载与版本管理print("=== 模型热加载与版本管理 ===")server = TFServingSimulator()server.load_model("ctr_model", 1, "/models/ctr_model/1")server.load_model("ctr_model", 2, "/models/ctr_model/2")server.set_version_policy("ctr_model", "specific", num_versions=3)server.load_model("ctr_model", 3, "/models/ctr_model/3")server.set_version_policy("ctr_model", "specific", num_versions=3) # 2. 金丝雀发布print("\n=== 金丝雀发布流程 ===")server.canary_release("ctr_model", 3, [0.05, 0.20, 0.50, 1.0]) # 3. 批处理性能分析print("\n=== 批处理性能分析 ===")requests = [f"req_{i}" for i in range(100)]for bs in [1, 8, 16, 32]:    results = server.batch_inference(requests, max_batch_size=bs)    throughput = min(bs * 500, 16000)    latency = max(2, bs * 0.3)    print(f"  batch={bs:2d}: 吞吐={throughput:5d} QPS, "          f"延迟={latency:.1f}ms") # 4. GPU共享配置print("\n=== GPU共享配置 ===")gpu_configs = [    {"model": "ctr_model", "mem_fraction": 0.4},    {"model": "nlp_model", "mem_fraction": 0.4},    {"model": "cv_model", "mem_fraction": 0.15},]total = sum(c["mem_fraction"] for c in gpu_configs)for c in gpu_configs:    print(f"  {c['model']}: GPU显存={c['mem_fraction']:.0%}")print(f"  总显存占用: {total:.0%} ({'OK' if total <= 1.0 else 'OOM风险'})")