第 8 章：模型部署与服务化

对称量化（Symmetric）:
    范围: [-127, 127]
    零点 Z = 0
    适用: 权重量化

非对称量化（Asymmetric）:
    范围: [0, 255]  
    零点 Z ≠ 0
    适用: 激活值量化

模型组件量化配置:
├── 视觉编码器: INT8 动态量化
├── 投影层: FP16 保持
├── 语言模型
│   ├── Embedding: INT8
│   ├── Attention: INT8 + FP16 (QK计算)
│   └── FFN: INT8
└── LM Head: FP16 (关键层保护)

for layer in model.layers:
    # 收集该层输入激活值
    X = collect_activations(layer, calibration_data)

    # 计算 Hessian 矩阵
    H = 2 * X @ X.T

    # 逐列量化权重
    for col in range(W.shape[1]):
        w_q = quantize_column(W[:, col], H)
        # 更新剩余列以补偿量化误差
        update_remaining_columns(W, w_q, col)

决策树：
显存充足？
├── 是 → FP16/BF16 推理
└── 否 → 需要量化
    ├── 延迟敏感？
    │   ├── 是 → INT8 量化（最快）
    │   └── 否 → 继续评估
    └── 精度要求？
        ├── 高 → GPTQ 4-bit
        └── 中 → AWQ 4-bit（推荐）

传统 KV Cache:
[连续内存块] → 浪费严重

PagedAttention:
[页表管理] → [按需分配] → [内存共享]
优势: 减少 50-80% 内存浪费

# 伪代码展示原理
def flash_attention(Q, K, V, block_size=64):
    # 分块遍历，减少 HBM 访问
    for q_block in split(Q, block_size):
        for kv_block in split(K, V, block_size):
            # 在 SRAM 中计算
            attn_block = softmax(q_block @ kv_block.T)
            out_block = attn_block @ v_block
            # 增量更新结果
            update_output(out_block)

传统 Static Batching:
[等待所有请求完成] → GPU 利用率低

Continuous Batching:
[持续加入新请求] → [动态调度] → GPU 利用率高

class VLMBatchScheduler:
    def schedule(self, requests):
        # 按图像大小分组
        groups = group_by_image_size(requests)

        # 视觉编码批处理
        for group in groups:
            vision_features = batch_encode_images(group)
            cache_features(vision_features)

        # 文本生成动态batching
        while active_requests:
            batch = select_compatible_requests()
            tokens = generate_batch(batch)
            update_requests(batch, tokens)

┌─────────────────────────────────────┐
│         Load Balancer               │
└────────────┬────────────────────────┘
             │
    ┌────────┴────────┐
    │                 │
┌───▼───┐       ┌────▼────┐
│ API   │       │  API    │
│Server │       │ Server  │
└───┬───┘       └────┬────┘
    │                │
    └────────┬───────┘
             │
    ┌────────▼────────┐
    │  Request Queue  │
    └────────┬────────┘
             │
    ┌────────▼────────────┐
    │  Inference Engine   │
    │  ┌──────────────┐  │
    │  │ Vision       │  │
    │  │ Encoder Pool │  │
    │  └──────┬───────┘  │
    │         │          │
    │  ┌──────▼───────┐  │
    │  │   Language   │  │
    │  │  Model Pool  │  │
    │  └──────────────┘  │
    └─────────────────────┘

class RequestRouter:
    def route(self, request):
        # 根据模型版本路由
        if request.model_version:
            return self.version_pools[request.model_version]

        # 根据负载均衡
        return self.select_least_loaded()

    def health_check(self):
        # 定期检查后端健康状态
        for backend in self.backends:
            if not backend.is_healthy():
                self.remove_backend(backend)

class VLMCache:
    def __init__(self):
        # 图像特征缓存
        self.vision_cache = LRUCache(size=10000)
        # Prompt 缓存
        self.prompt_cache = LRUCache(size=5000)

    def get_vision_features(self, image_hash):
        if image_hash in self.vision_cache:
            return self.vision_cache[image_hash]
        return None

    def cache_vision_features(self, image_hash, features):
        self.vision_cache[image_hash] = features

class ResourceManager:
    def allocate_request(self, request):
        required_memory = self.estimate_memory(request)

        # 等待资源可用
        while not self.has_available_memory(required_memory):
            time.sleep(0.1)

        # 分配资源
        self.current_memory += required_memory
        return self.process_request(request)

class ModelManager:
    def update_model(self, new_model_path):
        # 加载新模型
        new_model = load_model(new_model_path)

        # 逐步切换流量
        for ratio in [0.1, 0.3, 0.5, 0.7, 1.0]:
            self.traffic_ratio = ratio
            time.sleep(60)  # 观察指标

            if self.has_errors():
                self.rollback()
                break

@app.post("/v1/chat/completions")
async def chat_completion(request: ChatRequest):
    # 请求验证
    validate_request(request)

    # 图像预处理
    if request.images:
        vision_features = await encode_images(request.images)

    # 生成响应
    response = await generate_response(
        prompt=request.messages,
        vision_features=vision_features,
        **request.parameters
    )

    return response

@app.post("/v1/chat/completions/stream")
async def stream_chat_completion(request: ChatRequest):
    async def generate():
        async for token in generate_tokens(request):
            yield f"data: {json.dumps({'token': token})}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

class MetricsCollector:
    def __init__(self):
        self.metrics = {
            'ttft': [],
            'tps': [],
            'gpu_util': [],
            'memory_usage': []
        }

    def record_inference(self, request_id, start_time, tokens):
        ttft = time.time() - start_time
        tps = len(tokens) / (time.time() - start_time)

        self.metrics['ttft'].append(ttft)
        self.metrics['tps'].append(tps)

        # 记录到 Prometheus
        TTFT_HISTOGRAM.observe(ttft)
        TPS_GAUGE.set(tps)

# 使用 nsys 进行性能分析
nsys profile -o model_profile python inference_server.py

# 使用 nvprof 分析 kernel 执行
nvprof --print-gpu-trace python benchmark.py

def analyze_memory():
    # 显存快照
    snapshot = torch.cuda.memory_snapshot()

    # 分析内存分配
    for block in snapshot:
        if block['allocated']:
            print(f"Size: {block['size']}, Stream: {block['stream']}")

    # 内存统计
    print(f"Allocated: {torch.cuda.memory_allocated() / 1e9:.2f} GB")
    print(f"Reserved: {torch.cuda.memory_reserved() / 1e9:.2f} GB")

class ABTestManager:
    def __init__(self):
        self.experiments = {}

    def create_experiment(self, name, variants):
        self.experiments[name] = {
            'variants': variants,
            'metrics': defaultdict(list)
        }

    def route_request(self, request, experiment_name):
        # 基于用户 ID 的一致性哈希
        user_hash = hash(request.user_id)
        variant_index = user_hash % len(self.experiments[experiment_name]['variants'])
        return self.experiments[experiment_name]['variants'][variant_index]

    def record_metric(self, experiment_name, variant, metric_name, value):
        self.experiments[experiment_name]['metrics'][f"{variant}_{metric_name}"].append(value)

class DynamicBatchSizer:
    def __init__(self):
        self.current_batch_size = 1
        self.latency_history = []

    def adjust_batch_size(self):
        avg_latency = np.mean(self.latency_history[-100:])

        if avg_latency < TARGET_LATENCY * 0.8:
            # 延迟充裕，增加批大小
            self.current_batch_size = min(self.current_batch_size + 1, MAX_BATCH)
        elif avg_latency > TARGET_LATENCY:
            # 延迟超标，减小批大小
            self.current_batch_size = max(self.current_batch_size - 1, 1)

class AutoScaler:
    def scale_decision(self, metrics):
        # 基于队列长度和延迟决策
        if metrics['queue_length'] > QUEUE_THRESHOLD:
            return 'scale_up'
        elif metrics['avg_gpu_util'] < 0.3:
            return 'scale_down'
        return 'maintain'

# 安装 vLLM (支持 VLM)
pip install vllm>=0.3.0

# 验证 GPU 支持
python -c "import torch; print(torch.cuda.get_device_capability())"
# 需要 compute capability >= 7.0

from vllm import LLM, SamplingParams
from vllm.multimodal import MultiModalData

class VLMDeployment:
    def __init__(self, model_path):
        self.llm = LLM(
            model=model_path,
            # 关键参数配置
            tensor_parallel_size=2,  # TP 并行度
            max_model_len=4096,      # 最大序列长度
            gpu_memory_utilization=0.9,  # GPU 内存利用率

            # VLM 特定配置
            image_input_type="pixel_values",
            image_token_id=32000,
            image_input_shape=(3, 336, 336),
            image_feature_size=576,  # 24*24 patches

            # 优化参数
            enable_prefix_caching=True,  # 启用前缀缓存
            enable_chunked_prefill=True,  # 分块预填充
            max_num_batched_tokens=8192,
            max_num_seqs=256,

            # 量化配置（可选）
            quantization="awq",  # 使用 AWQ 4-bit 量化
        )

        self.sampling_params = SamplingParams(
            temperature=0.7,
            top_p=0.9,
            max_tokens=512,
        )

# 1. 启用 Flash Attention
os.environ["VLLM_USE_FLASH_ATTN"] = "1"

# 2. 配置 CUDA Graph
os.environ["VLLM_USE_CUDA_GRAPH"] = "1"
os.environ["VLLM_CUDA_GRAPH_MAX_SEQS"] = "32"

# 3. 调整调度策略
engine_args = {
    "scheduler_config": {
        "max_num_batched_tokens": 8192,
        "max_num_seqs": 256,
        "max_paddings": 512,
        "delay_factor": 0.1,  # 控制批处理等待时间
    }
}

def optimized_batch_inference(requests):
    # 按图像大小分组
    grouped = defaultdict(list)
    for req in requests:
        img_size = req.image.shape
        grouped[img_size].append(req)

    results = []
    for size, batch in grouped.items():
        # 同尺寸图像批处理
        outputs = llm.generate(
            prompts=[r.prompt for r in batch],
            multi_modal_data=[r.image for r in batch],
            sampling_params=sampling_params
        )
        results.extend(outputs)

    return results

# 监控内存使用
def monitor_memory():
    stats = llm.get_model_memory_usage()
    print(f"KV Cache: {stats['kv_cache_usage'] / 1e9:.2f} GB")
    print(f"Model Weights: {stats['model_weights'] / 1e9:.2f} GB")

    # 动态调整 KV cache 大小
    if stats['kv_cache_usage'] > MEMORY_THRESHOLD:
        llm.reduce_max_num_seqs(factor=0.8)

# 对不同层使用不同量化
quantization_config = {
    "vision_encoder": "int8",      # 视觉编码器用 INT8
    "projection": None,             # 投影层不量化
    "llm_layers_0_15": "awq_4bit", # 前半部分用 AWQ
    "llm_layers_16_31": "gptq_4bit", # 后半部分用 GPTQ
    "lm_head": None                # 输出层不量化
}

class PriorityBatchScheduler:
    def __init__(self):
        self.queues = {
            'high': PriorityQueue(),
            'normal': Queue(),
            'low': Queue()
        }

    def schedule_next_batch(self, max_batch_size):
        batch = []

        # 优先处理高优先级请求
        for priority in ['high', 'normal', 'low']:
            while len(batch) < max_batch_size and not self.queues[priority].empty():
                batch.append(self.queues[priority].get())

        return batch

def adaptive_padding(sequences):
    lengths = [len(seq) for seq in sequences]

    # 计算最优 padding 长度
    # 考虑硬件特性（如 tensor core 需要 8 的倍数）
    max_len = max(lengths)
    optimal_len = ((max_len + 7) // 8) * 8

    # 如果浪费超过阈值，考虑分批
    waste_ratio = (optimal_len * len(sequences) - sum(lengths)) / (optimal_len * len(sequences))

    if waste_ratio > 0.3:  # 30% 浪费阈值
        # 分成两批处理
        return split_by_length(sequences)

    return pad_sequences(sequences, optimal_len)

class PredictiveBatcher:
    def __init__(self):
        self.arrival_predictor = ArrivalRatePredictor()

    def should_wait_for_batch(self, current_batch_size):
        # 预测未来请求到达
        expected_arrivals = self.arrival_predictor.predict(window=100)  # 100ms

        # 计算等待收益
        current_efficiency = batch_efficiency(current_batch_size)
        future_efficiency = batch_efficiency(current_batch_size + expected_arrivals)

        # 决策：等待 vs 立即处理
        if future_efficiency / current_efficiency > 1.2:  # 20% 提升阈值
            return True, 100  # 等待 100ms
        return False, 0

# ❌ 错误：所有层都用 2-bit
model = quantize_model(model, bits=2)  # 精度严重下降

# ✅ 正确：混合精度策略
critical_layers = identify_sensitive_layers(model)
for name, layer in model.named_modules():
    if name in critical_layers:
        quantize_layer(layer, bits=8)  # 关键层保持高精度
    else:
        quantize_layer(layer, bits=4)

# ❌ 错误：使用随机数据校准
calibration_data = torch.randn(100, 3, 224, 224)

# ✅ 正确：使用真实分布的数据
calibration_data = load_representative_samples(
    dataset, 
    n_samples=200,
    stratified=True  # 确保覆盖各种情况
)

# ❌ 错误：只优化吞吐量，忽视延迟
config = {"max_batch_size": 128}  # P99 延迟爆炸

# ✅ 正确：平衡多个指标
config = {
    "max_batch_size": 32,
    "max_wait_time": 50,  # ms
    "target_latency": 200  # ms
}

# ❌ 错误：不清理已完成请求的 cache
kv_cache[request_id] = compute_kv(request)
# 请求完成后未删除...

# ✅ 正确：及时清理
try:
    kv_cache[request_id] = compute_kv(request)
    result = generate(kv_cache[request_id])
finally:
    del kv_cache[request_id]  # 确保清理

# ❌ 错误：直接处理第一个请求
@app.on_event("startup")
async def startup():
    global model
    model = load_model()  # 加载完就结束

# ✅ 正确：预热模型
@app.on_event("startup") 
async def startup():
    global model
    model = load_model()
    # 预热：运行几个推理避免首次调用慢
    warmup_inputs = create_dummy_inputs()
    for _ in range(3):
        model.generate(warmup_inputs)

# ❌ 错误：同步图像处理阻塞事件循环
def process_request(image, text):
    processed_image = cv2.resize(image, (336, 336))  # 阻塞
    return model.generate(processed_image, text)

# ✅ 正确：异步处理
async def process_request(image, text):
    processed_image = await asyncio.to_thread(
        cv2.resize, image, (336, 336)
    )
    return await model.generate_async(processed_image, text)

# ❌ 错误：平均延迟看起来很好
print(f"Avg latency: {np.mean(latencies)}ms")  # 200ms

# ✅ 正确：关注分位数
print(f"P50: {np.percentile(latencies, 50)}ms")  # 150ms
print(f"P95: {np.percentile(latencies, 95)}ms")  # 800ms！
print(f"P99: {np.percentile(latencies, 99)}ms")  # 2000ms！！