--- title: "云推理服务" description: "全面介绍云推理服务架构，包括弹性伸缩、多区域部署、成本优化和主流云平台LLM服务对比。" tags: ["云推理", "弹性伸缩", "多区域部署", "云平台"] category: "llm" icon: "🧠"

云推理服务

云推理的优势

云推理将大模型部署在云端，按需付费，无需自建GPU集群。用户通过API调用即可获得推理能力，云服务商负责基础设施运维、弹性伸缩和高可用保障。

云推理服务架构

典型架构

# 云推理服务组件
components:
  api_gateway:
    - 请求路由
    - 认证鉴权
    - 限流熔断
  
  load_balancer:
    - 请求分发
    - 健康检查
    - 会话保持
  
  inference_cluster:
    - GPU实例池
    - 模型缓存
    - 批处理优化
  
  storage:
    - 模型仓库
    - 日志存储
    - 缓存层

请求处理流程

class CloudInferenceService:
    def __init__(self):
        self.model_registry = {}
        self.gpu_pool = GPUPool()
        self.request_queue = asyncio.Queue()
    
    async def handle_request(self, request):
        # 1. 认证鉴权
        await self.authenticate(request)
        
        # 2. 路由到合适的模型实例
        instance = await self.route_to_instance(request.model_id)
        
        # 3. 批处理优化
        batched_input = await self.batch_request(request)
        
        # 4. 执行推理
        result = await instance.infer(batched_input)
        
        # 5. 返回结果
        return result

弹性伸缩策略

基于队列深度的伸缩

class AutoScaler:
    def __init__(self, min_instances=1, max_instances=10):
        self.min_instances = min_instances
        self.max_instances = max_instances
        self.current_instances = min_instances
        self.queue_depth = 0
        self.target_latency_ms = 100
    
    def evaluate_scaling(self, metrics: dict) -> int:
        queue_depth = metrics['queue_depth']
        avg_latency = metrics['avg_latency_ms']
        gpu_util = metrics['gpu_utilization']
        
        # 扩容条件
        if (queue_depth > 100 or avg_latency > self.target_latency_ms * 1.5
            or gpu_util > 85):
            new_count = min(self.current_instances + 1, self.max_instances)
        
        # 缩容条件
        elif (queue_depth < 10 and avg_latency < self.target_latency_ms * 0.5
              and gpu_util < 30):
            new_count = max(self.current_instances - 1, self.min_instances)
        
        else:
            new_count = self.current_instances
        
        self.current_instances = new_count
        return new_count

预测性伸缩

import numpy as np
from datetime import datetime

class PredictiveScaler:
    """基于历史模式预测未来负载"""
    
    def __init__(self):
        self.traffic_patterns = {}  # hour -> avg_traffic
    
    def record_traffic(self):
        hour = datetime.now().hour
        if hour not in self.traffic_patterns:
            self.traffic_patterns[hour] = []
        self.traffic_patterns[hour].append(1)
    
    def predict_next_hour(self) -> int:
        next_hour = (datetime.now().hour + 1) % 24
        if next_hour in self.traffic_patterns:
            avg = np.mean(self.traffic_patterns[next_hour])
            return max(1, int(avg * 1.2))  # 预留20%余量
        return 1

多区域部署

全球推理网络

class MultiRegionRouter:
    def __init__(self):
        self.regions = {
            'us-east': {'latency': 20, 'capacity': 100},
            'eu-west': {'latency': 80, 'capacity': 80},
            'ap-south': {'latency': 150, 'capacity': 60},
        }
    
    def route_request(self, user_location: str) -> str:
        """根据用户位置选择最优区域"""
        region_scores = {}
        for region, info in self.regions.items():
            score = info['capacity'] / (info['latency'] + 1)
            region_scores[region] = score
        
        return max(region_scores, key=region_scores.get)
    
    def replicate_model(self, model_id: str, target_regions: list):
        """跨区域复制模型"""
        for region in target_regions:
            print(f"复制模型 {model_id} 到 {region}")

数据一致性

class ModelSync:
    """模型版本同步"""
    
    def __init__(self):
        self.model_versions = {}  # region -> version
    
    def sync_model(self, model_id: str, version: str, source_region: str):
        """从源区域同步模型到所有区域"""
        for region in self.model_versions:
            if region != source_region:
                self._copy_model(model_id, version, source_region, region)
                self.model_versions[region] = version
    
    def _copy_model(self, model_id, version, src, dst):
        print(f"同步 {model_id}:{version} 从 {src} 到 {dst}")

主流云平台对比

cloud_services = {
    'AWS': {
        'service': 'SageMaker',
        'gpu': ['A10G', 'A100', 'H100'],
        'pricing': '按秒计费',
        'features': ['Spot实例', '自动伸缩', '多模型端点'],
    },
    'GCP': {
        'service': 'Vertex AI',
        'gpu': ['T4', 'A100', 'H100'],
        'pricing': '按秒计费',
        'features': ['Predictions API', '批量预测', '端点自动伸缩'],
    },
    'Azure': {
        'service': 'Azure ML',
        'gpu': ['NC系列', 'ND系列'],
        'pricing': '按秒计费',
        'features': ['Managed Online Endpoints', 'Kubernetes部署'],
    },
}

for provider, info in cloud_services.items():
    print(f"{provider} ({info['service']}): GPU={info['gpu']}, 定价={info['pricing']}")

成本优化

class CostOptimizer:
    def __init__(self):
        self.spot_discount = 0.7  # Spot实例折扣
        self.reserved_discount = 0.4  # 预留实例折扣
    
    def optimize(self, usage_pattern: str) -> dict:
        if usage_pattern == 'steady':
            return {'strategy': 'reserved', 'savings': self.reserved_discount}
        elif usage_pattern == 'bursty':
            return {'strategy': 'spot', 'savings': self.spot_discount}
        else:
            return {'strategy': 'on-demand', 'savings': 0}
    
    def estimate_monthly_cost(self, gpu_hours: float, gpu_type: str) -> float:
        prices = {
            'A10G': 1.0,
            'A100': 3.0,
            'H100': 8.0,
        }
        return gpu_hours * prices.get(gpu_type, 1.0)

最佳实践

• 选择靠近用户的区域部署，降低延迟
• 使用Spot/抢占式实例降低成本
• 实施预测性伸缩，提前扩容应对流量高峰
• 多区域部署实现高可用和灾备
• 监控推理延迟和成本，持续优化资源配置