64_模型部署：推理优化策略

目录
├── 1. 引言：大模型部署的挑战与意义
├── 2. 部署架构设计：从云端到边缘
├── 3. 硬件要求与选型：不同规模模型的部署配置
├── 4. 模型压缩技术：减小模型体积与计算需求
├── 5. 量化技术：精度与效率的平衡
├── 6. KV缓存优化：提升推理性能的关键
├── 7. 服务化部署：构建高性能推理服务
├── 8. 边缘部署：低延迟场景的优化方案
├── 9. 多模态模型部署：特殊考量与优化
└── 10. 行业最佳实践：成功案例分析

部署挑战分布：计算资源(35%) | 内存管理(30%) | 延迟优化(20%) | 能耗成本(10%) | 隐私安全(5%)

部署架构谱系：
纯云端部署 → 云边协同 → 混合部署 → 边缘部署 → 设备端部署

云边协同架构：
用户请求 → 边缘节点(预处理/缓存/轻量级推理) → 云端服务(复杂推理/模型更新) → 结果返回

def select_deployment_architecture(requirements):
    """选择合适的大模型部署架构"""
    # 延迟要求
    if requirements["latency_max_ms"] <= 10:
        if requirements["model_size"] <= "7B":
            return "设备端部署"
        else:
            return "边缘部署"
    elif requirements["latency_max_ms"] <= 100:
        if requirements["privacy_level"] >= 0.8:
            return "边缘部署"
        else:
            return "云边协同"
    elif requirements["latency_max_ms"] <= 500:
        # 成本优化
        if requirements["cost_sensitive"]:
            return "云边协同"
        else:
            return "纯云端部署"
    else:
        return "纯云端部署"
    
    # 其他因素考量
    if requirements["offline_support"]:
        return "边缘部署" if requirements["model_size"] <= "13B" else "混合部署"
    
    if requirements["scale"] >= 10000:  # 大规模服务
        return "纯云端部署"

def optimize_hardware_configuration(config, model_size, batch_size):
    """优化硬件配置以适应特定模型和负载"""
    # 根据模型大小调整GPU需求
    if model_size > "70B" and not config["multi_gpu"]:
        config["multi_gpu"] = True
        config["gpu_count"] = calculate_min_gpu_count(model_size)
    
    # 根据批次大小调整内存需求
    required_memory = estimate_memory_usage(model_size, batch_size)
    if config["system_memory"] < required_memory * 1.5:  # 预留50%缓冲
        config["system_memory"] = int(required_memory * 1.5)
    
    # 优化存储配置
    if config["model_storage_type"] != "NVMe SSD":
        config["model_storage_type"] = "NVMe SSD"
    
    # 对于多卡配置，确保网络支持
    if config["multi_gpu"] and config["gpu_count"] > 2:
        config["network_type"] = "InfiniBand"
    
    return config

压缩技术谱系：
量化 → 剪枝 → 知识蒸馏 → 低秩分解 → 结构重参数化 → 稀疏化训练

def prune_model(model, pruning_ratio=0.3):
    """对Transformer模型进行结构化剪枝"""
    # 逐层剪枝
    for name, module in model.named_modules():
        # 只剪枝注意力层和前馈网络
        if isinstance(module, nn.MultiheadAttention) or isinstance(module, nn.Linear):
            # 获取权重
            weight = module.weight.data
            
            # 计算权重重要性（绝对值）
            importance = weight.abs()
            
            # 对于线性层，对每个输出通道计算重要性
            if isinstance(module, nn.Linear):
                # 计算每个输出通道的重要性（L1范数）
                channel_importance = importance.sum(dim=1)
                # 排序获取阈值
                threshold = torch.sort(channel_importance)[0][int(len(channel_importance) * pruning_ratio)]
                # 创建掩码
                mask = channel_importance > threshold
                # 应用掩码
                module.weight.data = module.weight.data[mask]
                if module.bias is not None:
                    module.bias.data = module.bias.data[mask]
            
            # 更新下游层的输入维度
            update_downstream_layers(model, name, mask.sum().item())
    
    return model

def distillation_loss(student_logits, teacher_logits, targets, temperature=2.0, alpha=0.5):
    """计算知识蒸馏损失"""
    # 软目标损失（KL散度）
    soft_targets = F.softmax(teacher_logits / temperature, dim=-1)
    soft_prob = F.log_softmax(student_logits / temperature, dim=-1)
    soft_loss = F.kl_div(soft_prob, soft_targets, reduction='batchmean') * (temperature ** 2)
    
    # 硬目标损失（交叉熵）
    hard_loss = F.cross_entropy(student_logits, targets)
    
    # 组合损失
    loss = alpha * soft_loss + (1 - alpha) * hard_loss
    
    return loss

def train_student_model(student_model, teacher_model, dataloader, optimizer, args):
    """训练学生模型进行知识蒸馏"""
    student_model.train()
    teacher_model.eval()
    
    for epoch in range(args.epochs):
        total_loss = 0
        
        for batch in dataloader:
            input_ids = batch["input_ids"].to(args.device)
            attention_mask = batch["attention_mask"].to(args.device)
            labels = batch["labels"].to(args.device)
            
            # 冻结教师模型，获取输出
            with torch.no_grad():
                teacher_outputs = teacher_model(input_ids, attention_mask=attention_mask)
                teacher_logits = teacher_outputs.logits
            
            # 学生模型前向传播
            student_outputs = student_model(input_ids, attention_mask=attention_mask)
            student_logits = student_outputs.logits
            
            # 计算蒸馏损失
            loss = distillation_loss(
                student_logits=student_logits,
                teacher_logits=teacher_logits,
                targets=labels,
                temperature=args.temperature,
                alpha=args.alpha
            )
            
            # 反向传播和优化
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        print(f"Epoch {epoch+1}/{args.epochs}, Loss: {total_loss/len(dataloader):.4f}")
    
    return student_model

def apply_low_rank_decomposition(model, rank_ratio=0.1):
    """对模型应用低秩分解"""
    for name, module in model.named_modules():
        if isinstance(module, nn.Linear) and module.weight.size(0) > 100 and module.weight.size(1) > 100:
            # 计算目标秩
            m, n = module.weight.size()
            target_rank = max(10, int(min(m, n) * rank_ratio))
            
            # 使用SVD进行低秩分解
            with torch.no_grad():
                weight = module.weight.data
                U, S, V = torch.svd_lowrank(weight, q=target_rank)
                
                # 重建权重矩阵
                U_approx = U[:, :target_rank]
                S_approx = torch.diag(S[:target_rank])
                V_approx = V[:, :target_rank]
                
                # 创建分解后的线性层
                linear1 = nn.Linear(n, target_rank, bias=False)
                linear2 = nn.Linear(target_rank, m, bias=module.bias is not None)
                
                # 初始化权重
                linear1.weight.data = V_approx.t() @ torch.sqrt(S_approx)
                linear2.weight.data = torch.sqrt(S_approx) @ U_approx.t()
                
                if module.bias is not None:
                    linear2.bias.data = module.bias.data
            
            # 替换原始模块
            parent_name, child_name = name.rsplit('.', 1)
            parent = dict(model.named_modules())[parent_name]
            
            # 创建Sequential替换原始Linear
            setattr(parent, child_name, nn.Sequential(linear1, linear2))
    
    return model

量化技术谱系：
FP16/BF16 → INT8对称 → INT8非对称 → INT4 → FP8 → 混合精度量化

from transformers import AutoModelForCausalLM, AutoTokenizer
from optimum.gptq import GPTQQuantizer

def quantize_model_with_gptq(model_name, bits=4, group_size=128, device="cuda"):
    """使用GPTQ方法量化模型"""
    # 加载模型
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        device_map="auto",
        torch_dtype=torch.float16,
        trust_remote_code=True
    )
    
    tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
    
    # 准备校准数据
    def get_calibration_dataset(tokenizer, nsamples=128, seqlen=4096):
        # 使用随机数据作为校准集
        calibration_data = [torch.randint(0, tokenizer.vocab_size, (seqlen,)) for _ in range(nsamples)]
        return calibration_data
    
    calibration_dataset = get_calibration_dataset(tokenizer)
    
    # 创建量化器
    quantizer = GPTQQuantizer(
        bits=bits,
        group_size=group_size,
        damp_percent=0.01,
        desc_act=False,
    )
    
    # 执行量化
    quantized_model = quantizer.quantize_model(
        model=model,
        tokenizer=tokenizer,
        calibration_dataset=calibration_dataset
    )
    
    # 保存量化模型
    output_dir = f"{model_name}_gptq_{bits}bit"
    quantized_model.save_pretrained(output_dir)
    tokenizer.save_pretrained(output_dir)
    
    return output_dir

def quantize_model_with_awq(model_name, bits=4, device="cuda"):
    """使用AWQ方法量化模型"""
    from awq import AutoAWQForCausalLM
    
    # 加载和量化模型
    model = AutoAWQForCausalLM.from_pretrained(
        model_name,
        torch_dtype=torch.float16,
        device_map="auto"
    )
    
    # 量化配置
    quant_config = {
        "zero_point": True,
        "q_group_size": 128,
        "w_bit": bits,
        "version": "GEMM"
    }
    
    # 执行量化
    model.quantize(tokenizer, quant_config=quant_config)
    
    # 保存量化模型
    output_dir = f"{model_name}_awq_{bits}bit"
    model.save_quantized(output_dir)
    
    return output_dir

def quantize_model_with_smoothquant(model_name, bits=8, device="cuda"):
    """使用SmoothQuant方法量化模型"""
    from smoothquant import smooth_lm
    
    # 加载模型
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype=torch.float16,
        device_map="auto"
    )
    
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    # 应用SmoothQuant转换
    smoothed_model = smooth_lm(model, tokenizer, alpha=0.8)
    
    # 执行量化
    quantized_model = apply_int8_quantization(smoothed_model, bits)
    
    # 保存量化模型
    output_dir = f"{model_name}_smoothquant_{bits}bit"
    quantized_model.save_pretrained(output_dir)
    
    return output_dir

def optimized_inference_pipeline(model_name, quantization_method="gptq", bits=4, use_kv_cache_quant=True):
    """构建优化的推理管道"""
    # 1. 加载量化模型
    if quantization_method == "gptq":
        from transformers import AutoTokenizer, GPTQConfig
        
        gptq_config = GPTQConfig(
            bits=bits,
            group_size=128,
            dataset="wikitext2",
            disable_exllama=False
        )
        
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            quantization_config=gptq_config,
            device_map="auto"
        )
    elif quantization_method == "awq":
        from awq import AutoAWQForCausalLM
        
        model = AutoAWQForCausalLM.from_pretrained(
            model_name,
            low_cpu_mem_usage=True,
            device_map="auto"
        )
    
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    # 2. 启用KV缓存量化
    if use_kv_cache_quant and hasattr(model.config, "use_cache"):
        model.config.use_cache = True
        # 对于支持的模型，可以启用KV缓存量化
        if hasattr(model, "enable_input_require_grads"):
            model.enable_input_require_grads()
    
    # 3. 创建优化的推理管道
    from transformers import pipeline
    
    pipe = pipeline(
        "text-generation",
        model=model,
        tokenizer=tokenizer,
        torch_dtype=torch.float16,
        device_map="auto",
        # 推理优化参数
        do_sample=True,
        temperature=0.7,
        top_p=0.95,
        max_new_tokens=512,
        # 优化生成速度
        pad_token_id=tokenizer.pad_token_id,
        batch_size=1,
        # 启用缓存
        use_cache=True,
    )
    
    return pipe

KV缓存工作流程：
第1次生成 → 计算所有token的K和V → 存储到缓存
第2次生成 → 计算新token的Q → 与缓存中的K和V计算注意力 → 更新缓存
第n次生成 → 重复上述过程

def calculate_kv_cache_size(model_config, batch_size, sequence_length):
    """计算KV缓存的内存需求"""
    # 获取模型配置
    num_layers = model_config.num_hidden_layers
    num_attention_heads = model_config.num_attention_heads
    hidden_size = model_config.hidden_size
    
    # 计算每个注意力头的维度
    head_dim = hidden_size // num_attention_heads
    
    # K和V的大小（每个token每个层每个头）
    per_token_per_layer_per_head = head_dim * 2  # K和V各一个
    
    # 总体KV缓存大小
    kv_cache_size_bytes = batch_size * sequence_length * num_layers * num_attention_heads * per_token_per_layer_per_head * 2  # FP16
    
    # 转换为GB
    kv_cache_size_gb = kv_cache_size_bytes / (1024 ** 3)
    
    return kv_cache_size_gb

def quantized_kv_cache_implementation():
    """实现量化KV缓存的伪代码"""
    class QuantizedKVAttention(nn.Module):
        def __init__(self, hidden_size, num_attention_heads, head_dim, kv_bits=8):
            super().__init__()
            self.hidden_size = hidden_size
            self.num_attention_heads = num_attention_heads
            self.head_dim = head_dim
            self.kv_bits = kv_bits
            
            # 初始化量化器
            if kv_bits == 8:
                self.kv_quantizer = Int8Quantizer()
            elif kv_bits == 4:
                self.kv_quantizer = Int4Quantizer()
            elif kv_bits == 16:
                self.kv_quantizer = None  # FP16，无需量化
            
            # 注意力计算相关参数
            self.q_proj = nn.Linear(hidden_size, hidden_size)
            self.k_proj = nn.Linear(hidden_size, hidden_size)
            self.v_proj = nn.Linear(hidden_size, hidden_size)
            self.out_proj = nn.Linear(hidden_size, hidden_size)
        
        def forward(self, hidden_states, attention_mask=None, past_key_value=None):
            batch_size, seq_len, _ = hidden_states.size()
            
            # 计算查询Q
            query = self.q_proj(hidden_states)
            query = query.view(batch_size, seq_len, self.num_attention_heads, self.head_dim).transpose(1, 2)
            
            # 计算键K和值V
            key = self.k_proj(hidden_states)
            value = self.v_proj(hidden_states)
            key = key.view(batch_size, seq_len, self.num_attention_heads, self.head_dim).transpose(1, 2)
            value = value.view(batch_size, seq_len, self.num_attention_heads, self.head_dim).transpose(1, 2)
            
            # 处理缓存
            if past_key_value is not None:
                # 量化并合并缓存
                if self.kv_quantizer is not None:
                    # 使用量化版本的KV缓存
                    past_key = self.kv_quantizer.dequantize(past_key_value[0])
                    past_value = self.kv_quantizer.dequantize(past_key_value[1])
                else:
                    past_key, past_value = past_key_value
                
                key = torch.cat([past_key, key], dim=2)
                value = torch.cat([past_value, value], dim=2)
            
            # 更新缓存（可选量化）
            if self.kv_quantizer is not None:
                present_key_value = (
                    self.kv_quantizer.quantize(key),
                    self.kv_quantizer.quantize(value)
                )
            else:
                present_key_value = (key, value)
            
            # 计算注意力分数
            attention_scores = torch.matmul(query, key.transpose(-1, -2)) / math.sqrt(self.head_dim)
            
            if attention_mask is not None:
                attention_scores = attention_scores + attention_mask
            
            attention_probs = F.softmax(attention_scores, dim=-1)
            
            # 计算上下文向量
            context = torch.matmul(attention_probs, value)
            context = context.transpose(1, 2).contiguous().view(batch_size, seq_len, self.hidden_size)
            
            # 输出投影
            output = self.out_proj(context)
            
            return output, present_key_value
    
    return QuantizedKVAttention

class KVCacheManager:
    """高效的KV缓存管理器"""
    def __init__(self, max_batch_size, max_seq_len, num_layers, num_heads, head_dim, precision=torch.float16):
        self.max_batch_size = max_batch_size
        self.max_seq_len = max_seq_len
        self.num_layers = num_layers
        self.num_heads = num_heads
        self.head_dim = head_dim
        self.precision = precision
        
        # 预分配缓存空间
        self.k_cache = torch.zeros(
            (num_layers, max_batch_size, num_heads, max_seq_len, head_dim),
            dtype=precision, device="cuda"
        )
        self.v_cache = torch.zeros(
            (num_layers, max_batch_size, num_heads, max_seq_len, head_dim),
            dtype=precision, device="cuda"
        )
        
        # 跟踪每个序列的当前长度
        self.seq_lens = torch.zeros(max_batch_size, dtype=torch.int32, device="cuda")
    
    def update_cache(self, layer_idx, batch_idx, new_k, new_v):
        """更新KV缓存"""
        batch_size = new_k.size(0)
        new_seq_len = new_k.size(2)
        
        for i in range(batch_size):
            batch_pos = batch_idx[i]
            start_pos = self.seq_lens[batch_pos].item()
            end_pos = start_pos + new_seq_len
            
            # 更新缓存
            self.k_cache[layer_idx, batch_pos, :, start_pos:end_pos, :] = new_k[i]
            self.v_cache[layer_idx, batch_pos, :, start_pos:end_pos, :] = new_v[i]
            
            # 更新序列长度
            self.seq_lens[batch_pos] = end_pos
    
    def get_cache(self, layer_idx, batch_idx, current_seq_len):
        """获取KV缓存"""
        batch_size = len(batch_idx)
        
        # 创建输出缓存
        cache_k = torch.zeros(
            (batch_size, self.num_heads, current_seq_len, self.head_dim),
            dtype=self.precision, device="cuda"
        )
        cache_v = torch.zeros_like(cache_k)
        
        # 填充缓存
        for i in range(batch_size):
            batch_pos = batch_idx[i]
            seq_len = self.seq_lens[batch_pos].item()
            
            if seq_len > 0:
                actual_seq_len = min(seq_len, current_seq_len)
                cache_k[i, :, :actual_seq_len, :] = self.k_cache[layer_idx, batch_pos, :, :actual_seq_len, :]
                cache_v[i, :, :actual_seq_len, :] = self.v_cache[layer_idx, batch_pos, :, :actual_seq_len, :]
        
        return cache_k, cache_v
    
    def reset_cache(self, batch_idx):
        """重置指定批次的缓存"""
        for idx in batch_idx:
            self.seq_lens[idx] = 0

多级KV缓存架构：
显存(VRAM) → 内存(RAM) → 本地存储(SSD) → 远端存储

大模型推理服务架构：
客户端 → API网关 → 负载均衡器 → 推理服务集群 → 监控告警系统

class LLMAPIGateway:
    """大模型推理服务API网关"""
    def __init__(self, config):
        self.config = config
        self.auth_manager = AuthManager(config["auth"])
        self.rate_limiter = RateLimiter(config["rate_limit"])
        self.request_router = RequestRouter(config["routing"])
        self.monitor = Monitor(config["monitoring"])
    
    async def handle_request(self, request):
        """处理推理请求"""
        # 记录请求开始
        self.monitor.record_request_start(request)
        
        try:
            # 认证与授权
            if not await self.auth_manager.authenticate(request):
                return self._create_error_response(401, "Unauthorized")
            
            # 速率限制
            if not await self.rate_limiter.check(request):
                return self._create_error_response(429, "Too Many Requests")
            
            # 请求路由
            target_service = await self.request_router.route(request)
            
            # 转发请求到推理服务
            response = await self._forward_to_service(target_service, request)
            
            # 记录请求成功
            self.monitor.record_request_success(request, response)
            
            return response
        except Exception as e:
            # 记录请求失败
            self.monitor.record_request_error(request, e)
            return self._create_error_response(500, "Internal Server Error")
    
    async def _forward_to_service(self, service, request):
        """转发请求到推理服务"""
        # 实现请求转发逻辑
        pass
    
    def _create_error_response(self, status_code, message):
        """创建错误响应"""
        return {"error": {"code": status_code, "message": message}}

class LLMInferenceService:
    """大模型推理服务"""
    def __init__(self, config):
        self.config = config
        self.model = self._load_model(config["model"])
        self.tokenizer = self._load_tokenizer(config["model"])
        self.batch_processor = BatchProcessor(config["batching"])
        self.cache_manager = KVCacheManager(**config["kv_cache"])
    
    def _load_model(self, model_config):
        """加载模型"""
        # 根据配置加载合适的模型（量化、优化等）
        model_type = model_config.get("type", "transformer")
        
        if model_type == "gptq":
            # 加载GPTQ量化模型
            from transformers import AutoModelForCausalLM, GPTQConfig
            
            gptq_config = GPTQConfig(
                bits=model_config["bits"],
                group_size=model_config["group_size"]
            )
            
            model = AutoModelForCausalLM.from_pretrained(
                model_config["name"],
                quantization_config=gptq_config,
                device_map="auto"
            )
        else:
            # 加载标准模型
            model = AutoModelForCausalLM.from_pretrained(
                model_config["name"],
                torch_dtype=getattr(torch, model_config["dtype"]),
                device_map="auto"
            )
        
        return model
    
    def _load_tokenizer(self, model_config):
        """加载分词器"""
        from transformers import AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained(model_config["name"])
        return tokenizer
    
    async def process_request(self, request):
        """处理单个推理请求"""
        # 解析请求参数
        prompt = request["prompt"]
        max_tokens = request.get("max_tokens", 100)
        temperature = request.get("temperature", 0.7)
        
        # 预处理输入
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
        
        # 生成响应
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=max_tokens,
                temperature=temperature,
                do_sample=temperature > 0,
                use_cache=True
            )
        
        # 后处理输出
        response_text = self.tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
        
        return {"text": response_text}
    
    async def process_batch(self, batch_requests):
        """处理批量推理请求"""
        # 实现批量处理逻辑
        pass

class LLMLoadBalancer:
    """大模型推理服务负载均衡器"""
    def __init__(self, config):
        self.config = config
        self.service_instances = []
        self.health_checker = HealthChecker(config["health_check"])
        self.strategy = self._create_strategy(config["strategy"])
    
    def _create_strategy(self, strategy_config):
        """创建负载均衡策略"""
        strategy_type = strategy_config["type"]
        
        if strategy_type == "round_robin":
            return RoundRobinStrategy()
        elif strategy_type == "least_connection":
            return LeastConnectionStrategy()
        elif strategy_type == "resource_aware":
            return ResourceAwareStrategy(strategy_config)
        else:
            return RoundRobinStrategy()  # 默认策略
    
    def register_service(self, service_instance):
        """注册服务实例"""
        self.service_instances.append(service_instance)
    
    def unregister_service(self, service_instance):
        """注销服务实例"""
        if service_instance in self.service_instances:
            self.service_instances.remove(service_instance)
    
    async def get_next_service(self, request):
        """获取下一个服务实例"""
        # 获取健康的服务实例
        healthy_instances = await self.health_checker.get_healthy_instances(self.service_instances)
        
        if not healthy_instances:
            raise RuntimeError("No healthy service instances available")
        
        # 使用策略选择服务实例
        return self.strategy.select(healthy_instances, request)
    
    async def update_service_metrics(self, service_instance, metrics):
        """更新服务实例指标"""
        # 更新服务实例的负载、延迟等指标
        pass

# docker-compose.yml 示例
version: '3.8'

services:
  llm-api-gateway:
    image: llm-api-gateway:latest
    ports:
      - "8000:8000"
    environment:
      - CONFIG_PATH=/app/config/config.yaml
    volumes:
      - ./config:/app/config
    depends_on:
      - llm-load-balancer

  llm-load-balancer:
    image: llm-load-balancer:latest
    environment:
      - MONITORING_ENABLED=true
    depends_on:
      - llm-inference-1
      - llm-inference-2

  llm-inference-1:
    image: llm-inference:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - MODEL_NAME=meta-llama/Llama-3-70B
      - QUANTIZATION=gptq
      - BITS=4
      - MAX_BATCH_SIZE=32

  llm-inference-2:
    image: llm-inference:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - MODEL_NAME=meta-llama/Llama-3-70B
      - QUANTIZATION=gptq
      - BITS=4
      - MAX_BATCH_SIZE=32

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

边缘部署架构：
用户设备 → 边缘节点(模型推理) → 云服务器(模型更新/复杂任务)

def create_edge_optimized_model(base_model, target_size_mb=50):
    """创建边缘优化模型"""
    # 1. 知识蒸馏 - 创建小型学生模型
    student_config = AutoConfig.from_pretrained(base_model)
    # 减少层数和隐藏层大小
    student_config.num_hidden_layers = max(4, student_config.num_hidden_layers // 4)
    student_config.hidden_size = max(256, student_config.hidden_size // 2)
    student_config.num_attention_heads = max(4, student_config.num_attention_heads // 2)
    
    student_model = AutoModelForCausalLM.from_config(student_config)
    
    # 2. 量化 - 应用INT4量化
    from transformers import AutoTokenizer
    from optimum.onnxruntime import ORTQuantizer
    from optimum.onnxruntime.configuration import AutoQuantizationConfig
    
    # 导出为ONNX
    tokenizer = AutoTokenizer.from_pretrained(base_model)
    onnx_path = f"{base_model}_student.onnx"
    
    # 使用ONNX Runtime进行静态量化
    quantizer = ORTQuantizer.from_pretrained(onnx_path)
    quantization_config = AutoQuantizationConfig.avx512_vnni(is_static=False, per_channel=False)
    quantizer.quantize(quantization_config=quantization_config, save_dir="./quantized_model")
    
    # 3. 模型优化 - 应用图优化
    # 使用ONNX Runtime优化器
    from onnxruntime.quantization import quantize_dynamic
    
    # 进一步优化和压缩
    optimized_model_path = optimize_model_for_edge("quantized_model/model_quantized.onnx", target_size_mb)
    
    return optimized_model_path

def configure_edge_inference(model_path, device="cpu"):
    """配置边缘推理环境"""
    # 根据设备类型选择推理引擎
    if device == "cpu":
        # CPU优化
        import onnxruntime as ort
        
        # 创建优化的会话选项
        session_options = ort.SessionOptions()
        session_options.intra_op_num_threads = max(1, os.cpu_count() - 1)
        session_options.inter_op_num_threads = 1
        session_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        
        # 加载量化模型
        session = ort.InferenceSession(
            model_path,
            sess_options=session_options,
            providers=['CPUExecutionProvider']
        )
        
        return session
    
    elif device.startswith("nvidia"):
        # NVIDIA GPU优化
        import tensorrt as trt
        import numpy as np
        
        # 创建TensorRT引擎
        logger = trt.Logger(trt.Logger.WARNING)
        runtime = trt.Runtime(logger)
        
        # 加载优化的TensorRT引擎
        with open(model_path, 'rb') as f:
            engine = runtime.deserialize_cuda_engine(f.read())
        
        return engine
    
    elif device.startswith("arm"):
        # ARM设备优化（如手机、树莓派等）
        # 使用TensorFlow Lite或TensorRT for ARM
        import tensorflow as tf
        
        interpreter = tf.lite.Interpreter(model_path=model_path)
        interpreter.allocate_tensors()
        
        return interpreter

def optimize_multimodal_model(model, config):
    """优化多模态模型"""
    # 1. 文本编码器优化
    if config["optimize_text_encoder"]:
        # 对文本编码器应用量化
        text_quant_config = GPTQConfig(bits=4, group_size=128)
        model.text_encoder = apply_quantization(model.text_encoder, text_quant_config)
    
    # 2. 视觉编码器优化
    if config["optimize_vision_encoder"]:
        # 视觉编码器剪枝和量化
        model.vision_encoder = apply_pruning(model.vision_encoder, pruning_ratio=0.2)
        model.vision_encoder = apply_int8_quantization(model.vision_encoder)
    
    # 3. 跨模态连接器优化
    if config["optimize_connector"]:
        # 低秩分解跨模态连接器
        model.connector = apply_low_rank_decomposition(model.connector, rank_ratio=0.1)
    
    # 4. 解码器优化
    if config["optimize_decoder"]:
        # 应用KV缓存优化和量化
        model.decoder = enable_optimized_kv_cache(model.decoder, kv_bits=8)
    
    return model

class OptimizedMultimodalPipeline:
    """优化的多模态推理管道"""
    def __init__(self, model_path, config):
        self.config = config
        self.model = self._load_model(model_path)
        self.text_processor = self._create_text_processor()
        self.image_processor = self._create_image_processor()
        self.audio_processor = self._create_audio_processor()
        
        # 启用优化
        self._enable_optimizations()
    
    def _load_model(self, model_path):
        """加载优化后的模型"""
        # 根据配置加载模型
        pass
    
    def _enable_optimizations(self):
        """启用各种优化技术"""
        # 1. 启用ONNX Runtime优化
        if self.config["use_onnxruntime"]:
            self._convert_to_onnx()
        
        # 2. 启用内存优化
        if self.config["memory_optimization"]:
            self._setup_memory_optimization()
        
        # 3. 启用并行处理
        if self.config["parallel_processing"]:
            self._setup_parallel_processing()
    
    async def process_multimodal_input(self, text=None, image=None, audio=None):
        """处理多模态输入并生成响应"""
        # 并行预处理不同模态
        processed_inputs = {}
        
        if text is not None:
            processed_inputs["text"] = await self.text_processor.process(text)
        
        if image is not None:
            processed_inputs["image"] = await self.image_processor.process(image)
        
        if audio is not None:
            processed_inputs["audio"] = await self.audio_processor.process(audio)
        
        # 执行模型推理
        with torch.no_grad():
            outputs = self.model(**processed_inputs)
        
        # 后处理输出
        result = self._postprocess(outputs)
        
        return result