第22章 边缘与端侧 AI 基础设施

第22章 边缘与端侧 AI 基础设施

云端推理的延迟是 100-500ms，端侧推理的延迟是 10-50ms。对于实时应用（自动驾驶、AR/VR、工业控制），这不是优化——这是”能不能用”的区别。边缘 AI 不是云端 AI 的”缩小版”，它是一套完全不同的基础设施。

22.1 边缘推理场景与挑战

为什么需要边缘推理

graph LR
    subgraph "云端推理"
        U1[用户设备] -->|上传数据<br/>100ms| CLOUD[云数据中心]
        CLOUD -->|推理 + 返回<br/>100ms| U1
    end
    
    subgraph "边缘推理"
        U2[用户设备] -->|本地推理<br/>10-50ms| EDGE[边缘节点 / 端侧 NPU]
    end
    
    style CLOUD fill:#e3f2fd
    style EDGE fill:#e8f5e9

边缘推理的核心驱动力不仅是延迟：

驱动力	说明	典型场景
低延迟	往返延迟 < 50ms	自动驾驶、机器人、AR/VR
隐私保护	数据不出设备	医疗影像、家庭摄像头、语音助手
带宽节省	避免传输大量原始数据	视频分析、工业传感器
离线能力	无网络也能工作	户外设备、偏远地区
成本控制	减少云服务器费用	海量 IoT 设备

边缘硬件的约束三角

graph triangle
    P((性能 Performance<br/>TOPS)) --- M((内存 Memory<br/>MB - GB))
    M --- W((功耗 Power<br/>mW - W))
    W --- P
    
    style P fill:#e1f5fe
    style M fill:#fff3e0
    style W fill:#fce4ec

典型边缘设备的资源约束：

设备类型	算力 (TOPS)	内存 (RAM)	功耗	典型价格
智能手机 (NPU)	10-50	8-16 GB	2-5 W	$400-1200
Jetson Orin Nano	40	8 GB	7-15 W	$249
Jetson Orin AGX	275	64 GB	15-60 W	$1,999
Raspberry Pi 5 + Hailo	13	8 GB	5-8 W	$150
智能手表	0.5-2	0.5-1 GB	< 1 W	$200-800
工业网关	5-20	4-8 GB	5-15 W	$200-500

Tip

设备选型经验法则：首先确定你的延迟目标（如 30ms）和功耗预算（如 5W）。然后选择能在这个功耗预算内提供足够 TOPS 的设备。内存往往是最大的限制——一个 INT8 量化的 7B 模型需要约 7GB 内存，这已经排除了大多数微控制器。

22.2 模型轻量化与端侧部署

量化：最直接的压缩手段

量化是将浮点模型转换为低精度表示的过程。INT8 量化可以将模型大小减小 4 倍，同时推理速度提升 2-3 倍：

# 模型量化全流程示例

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

# 步骤 1: 加载原始模型
model_id = "Qwen/Qwen2.5-1.5B"
model = AutoModelForCausalLM.from_pretrained(
    model_id, 
    torch_dtype=torch.float16,
    device_map="cpu",
)
tokenizer = AutoTokenizer.from_pretrained(model_id)

# 原始模型大小
original_size = sum(p.nelement() * p.element_size() for p in model.parameters())
print(f"FP16 模型大小: {original_size / 1e9:.2f} GB")

# 步骤 2: 动态量化（最简单的方式，适合 CPU 推理）
from torch.quantization import quantize_dynamic

model_int8 = quantize_dynamic(
    model,
    {torch.nn.Linear},  # 只量化 Linear 层
    dtype=torch.qint8,
)

quantized_size = sum(
    p.nelement() * (1 if p.dtype == torch.qint8 else p.element_size())
    for p in model_int8.parameters()
)
print(f"INT8 模型大小: {quantized_size / 1e9:.2f} GB")
print(f"压缩比: {original_size / quantized_size:.1f}x")

# 步骤 3: 性能对比测试
import time

def benchmark_generation(model, prompt, num_tokens=50, runs=3):
    inputs = tokenizer(prompt, return_tensors="pt")
    
    times = []
    for _ in range(runs):
        torch.cuda.synchronize() if torch.cuda.is_available() else None
        start = time.perf_counter()
        
        with torch.no_grad():
            output = model.generate(
                **inputs,
                max_new_tokens=num_tokens,
                do_sample=False,  # 贪心解码，结果可复现
            )
        
        torch.cuda.synchronize() if torch.cuda.is_available() else None
        elapsed = time.perf_counter() - start
        times.append(elapsed)
    
    avg_time = sum(times) / len(times)
    tokens_per_sec = num_tokens / avg_time
    return avg_time, tokens_per_sec

# 对比
fp16_time, fp16_tps = benchmark_generation(model, "Explain AI:")
int8_time, int8_tps = benchmark_generation(model_int8, "Explain AI:")

print(f"FP16: {fp16_tps:.1f} tokens/sec ({fp16_time:.2f}s)")
print(f"INT8: {int8_tps:.1f} tokens/sec ({int8_time:.2f}s)")
print(f"加速比: {int8_tps / fp16_tps:.1f}x")

GPTQ / AWQ：更先进的后训练量化

# 使用 AWQ (Activation-aware Weight Quantization) 量化
# AWQ 在量化时考虑激活值的分布，比朴素的 RTN 量化精度更高

# 安装：pip install autoawq

from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

model_path = "Qwen/Qwen2.5-7B"
quant_path = "Qwen2.5-7B-AWQ"

# 加载模型
model = AutoAWQForCausalLM.from_pretrained(model_path, device_map="auto")
tokenizer = AutoTokenizer.from_pretrained(model_path)

# 量化配置
quant_config = {
    "zero_point": True,        # 使用零点（对称+非对称量化）
    "q_group_size": 128,       # 分组量化，每组独立缩放
    "w_bit": 4,                # 4-bit 量化
    "version": "GEMM",         # GEMM 版本更快，GEMV 版本更省内存
}

# 执行量化（需要校准数据集）
# 校准数据应尽可能接近实际使用场景
calibration_data = [
    "The future of AI is",
    "In machine learning,",
    # ... 更多校准样本
]

model.quantize(
    tokenizer,
    quant_config=quant_config,
    calib_data=calibration_data,
)

# 保存量化模型
model.save_quantized(quant_path)
tokenizer.save_pretrained(quant_path)

# AWQ 4-bit 量化效果：
# 原始 FP16 7B: ~14 GB
# AWQ INT4 7B:  ~3.5 GB  (4x 压缩)
# 精度损失通常 < 2%（在标准 benchmark 上）

知识蒸馏

# 知识蒸馏：用大模型（教师）训练小模型（学生）

import torch
import torch.nn as nn
import torch.nn.functional as F

class DistillationTrainer:
    """
    知识蒸馏训练器
    
    核心：学生不仅要学习正确答案，还要学习教师的输出分布
    """
    
    def __init__(self, teacher_model, student_model, 
                 temperature=4.0, alpha=0.5):
        self.teacher = teacher_model
        self.student = student_model
        self.temperature = temperature  # 温度参数（软化概率分布）
        self.alpha = alpha              # 蒸馏损失 vs 任务损失的权重
        
        # 冻结教师模型
        for param in self.teacher.parameters():
            param.requires_grad = False
        self.teacher.eval()
    
    def compute_loss(self, student_logits, teacher_logits, labels):
        """
        蒸馏损失 = α * KL(teacher || student) + (1-α) * CE(labels)
        """
        # 软目标损失（KL 散度）
        soft_teacher = F.log_softmax(
            teacher_logits / self.temperature, dim=-1
        )
        soft_student = F.log_softmax(
            student_logits / self.temperature, dim=-1
        )
        distillation_loss = F.kl_div(
            soft_student, soft_teacher.exp(),  # 注意 KL 方向
            reduction="batchmean"
        ) * (self.temperature ** 2)  # 温度补偿
        
        # 硬目标损失（标准交叉熵）
        task_loss = F.cross_entropy(student_logits, labels)
        
        # 总损失
        total_loss = self.alpha * distillation_loss + (1 - self.alpha) * task_loss
        return total_loss, distillation_loss, task_loss
    
    def train_step(self, inputs, labels, optimizer):
        optimizer.zero_grad()
        
        # 教师前向（不需要梯度）
        with torch.no_grad():
            teacher_logits = self.teacher(inputs)
        
        # 学生前向
        student_logits = self.student(inputs)
        
        # 计算损失
        loss, dist_loss, task_loss = self.compute_loss(
            student_logits, teacher_logits, labels
        )
        
        loss.backward()
        optimizer.step()
        
        return {
            "loss": loss.item(),
            "distillation_loss": dist_loss.item(),
            "task_loss": task_loss.item(),
        }

ONNX Runtime：跨平台推理引擎

# 将 PyTorch 模型导出为 ONNX 并使用 ONNX Runtime 推理

import torch
import onnxruntime as ort
import numpy as np

# 步骤 1: 导出为 ONNX
class SimpleModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = torch.nn.Linear(768, 1024)
        self.linear2 = torch.nn.Linear(1024, 768)
    
    def forward(self, x):
        x = torch.relu(self.linear1(x))
        return self.linear2(x)

model = SimpleModel().eval()
dummy_input = torch.randn(1, 768)

torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    input_names=["input"],
    output_names=["output"],
    dynamic_axes={
        "input": {0: "batch_size"},   # 动态 batch
        "output": {0: "batch_size"},
    },
    opset_version=17,  # 使用较新的 opset
)

# 步骤 2: 使用 ONNX Runtime 推理
# 在边缘设备上，ONNX Runtime 提供了优化的推理引擎

# 基础推理
session = ort.InferenceSession("model.onnx")

input_data = np.random.randn(1, 768).astype(np.float32)
outputs = session.run(None, {"input": input_data})

# 步骤 3: 使用优化配置
# 针对特定硬件优化
session_options = ort.SessionOptions()
session_options.graph_optimization_level = (
    ort.GraphOptimizationLevel.ORT_ENABLE_ALL
)

# 线程配置（边缘设备通常核数有限）
session_options.intra_op_num_threads = 4  # 算子内并行
session_options.inter_op_num_threads = 1  # 算子间并行（边缘设备建议 1）

# 针对特定硬件的 Execution Provider
providers = [
    # 优先使用硬件加速
    "CoreMLExecutionProvider",     # Apple Neural Engine (ANE)
    "CUDAExecutionProvider",       # NVIDIA GPU
    "QNNExecutionProvider",        # Qualcomm NPU
    "CPUExecutionProvider",        # 回退到 CPU
]

session = ort.InferenceSession(
    "model.onnx",
    sess_options=session_options,
    providers=providers,
)

print(f"使用的 Provider: {session.get_providers()}")

Tip

Apple Silicon 特别提示：在 M1/M2/M3/M4 上，CoreML Execution Provider 可以利用 Apple Neural Engine (ANE)，推理速度通常比 CPU 快 5-10 倍。但并非所有 ONNX 算子都支持 ANE——不支持的算子会回退到 CPU，频繁切换会降低性能。建议使用 coremltools 转换模型并检查 ANE 支持情况。

22.3 边云协同架构

模型分发管道

graph TB
    subgraph "云端"
        T[模型训练] --> R[模型注册中心]
        R --> C[压缩 / 量化]
        C --> V[版本管理]
        V --> D[CDN 分发]
    end
    
    subgraph "边缘节点"
        D -->|OTA 更新| E1[设备 1]
        D -->|OTA 更新| E2[设备 2]
        D -->|OTA 更新| EN[设备 N]
    end
    
    subgraph "监控反馈"
        E1 --> M[遥测数据]
        E2 --> M
        EN --> M
        M --> T
    end
    
    style T fill:#e1f5fe
    style D fill:#fff3e0
    style M fill:#e8f5e9

# 边云协同的模型分发系统

class ModelDistributionSystem:
    """
    模型分发与管理系统
    
    功能：
    - 版本管理（支持灰度发布）
    - 差异更新（减少传输量）
    - A/B 测试
    - 回滚机制
    """
    
    def __init__(self, registry_url, cdn_url):
        self.registry = registry_url  # 模型注册中心
        self.cdn = cdn_url            # CDN 分发地址
        self.device_registry = {}     # 设备注册表
    
    def publish_model(self, model_path, version, channels=["canary"]):
        """
        发布新版本模型
        
        channels: 
        - "canary": 灰度（1% 设备）
        - "beta": 测试（10% 设备）
        - "stable": 全量（100% 设备）
        """
        manifest = {
            "version": version,
            "channels": channels,
            "created_at": datetime.now().isoformat(),
            "checksum": self._compute_checksum(model_path),
            "size_bytes": os.path.getsize(model_path),
            "min_device_memory_mb": self._estimate_memory(model_path),
            "supported_devices": self._get_supported_devices(model_path),
        }
        
        # 上传到 CDN
        self._upload_to_cdn(model_path, version)
        
        # 注册到模型注册中心
        self._register(manifest)
        
        return manifest
    
    def get_update_plan(self, device_id):
        """
        为特定设备生成更新计划
        
        考虑因素：
        - 设备型号和硬件能力
        - 当前模型版本
        - 网络条件
        - 电量状态
        """
        device = self.device_registry.get(device_id)
        if not device:
            return None
        
        current_version = device.get("model_version", "unknown")
        target_version = self._get_target_version(device)
        
        if current_version == target_version:
            return None  # 无需更新
        
        # 检查更新条件
        if not self._can_update_now(device):
            return {"deferred": True, "reason": "conditions_not_met"}
        
        # 差异更新 vs 全量更新
        if self._supports_delta_update(current_version, target_version):
            update_url = self._get_delta_url(current_version, target_version)
            update_type = "delta"
            size = self._estimate_delta_size(current_version, target_version)
        else:
            update_url = self._get_full_url(target_version)
            update_type = "full"
            size = self._get_model_size(target_version)
        
        return {
            "update_type": update_type,
            "current_version": current_version,
            "target_version": target_version,
            "download_url": update_url,
            "size_bytes": size,
            "checksum": self._get_checksum(target_version),
            "actions": {
                "pre_download": "backup_current",
                "post_download": "verify_checksum",
                "post_install": "health_check + rollback_if_fail",
            },
        }
    
    def _can_update_now(self, device):
        """检查设备是否满足更新条件"""
        return (
            device.get("battery_level", 0) > 0.5      # 电量 > 50%
            and device.get("wifi_connected", False)    # WiFi 连接
            and device.get("idle_state", False)        # 设备空闲
        )

联邦学习

# 联邦学习：数据不动，模型动

import torch
from collections import OrderedDict

class FederatedLearningServer:
    """
    联邦学习服务器端
    
    流程：
    1. 服务器下发全局模型
    2. 客户端在本地数据上训练
    3. 客户端上传模型更新（梯度或权重差）
    4. 服务器聚合更新，生成新的全局模型
    5. 回到步骤 1
    """
    
    def __init__(self, model_template, num_clients=10):
        self.global_model = model_template()
        self.num_clients = num_clients
        self.current_round = 0
    
    def aggregate(self, client_updates, client_weights=None):
        """
        FedAvg: 加权平均客户端模型更新
        
        client_updates: list of state_dicts
        client_weights: 各客户端的样本数权重
        """
        if client_weights is None:
            client_weights = [1.0 / len(client_updates)] * len(client_updates)
        
        # 归一化权重
        total_weight = sum(client_weights)
        client_weights = [w / total_weight for w in client_weights]
        
        # 加权平均
        global_state = self.global_model.state_dict()
        
        for key in global_state.keys():
            # 只平均浮点参数，跳过 buffer
            if global_state[key].dtype in [torch.float32, torch.float16]:
                weighted_sum = sum(
                    update[key].float() * weight
                    for update, weight in zip(client_updates, client_weights)
                )
                global_state[key] = weighted_sum.to(global_state[key].dtype)
        
        self.global_model.load_state_dict(global_state)
        self.current_round += 1
        
        return self.global_model.state_dict()


class FederatedLearningClient:
    """
    联邦学习客户端（运行在边缘设备上）
    """
    
    def __init__(self, client_id, local_data, model_template):
        self.client_id = client_id
        self.local_data = local_data
        self.model = model_template()
    
    def local_train(self, global_state, epochs=3, lr=1e-4):
        """
        使用本地数据训练
        
        关键：数据永远不离开设备
        只上传模型权重差或梯度
        """
        # 加载全局模型
        self.model.load_state_dict(global_state)
        self.model.train()
        
        optimizer = torch.optim.SGD(self.model.parameters(), lr=lr)
        
        for epoch in range(epochs):
            for batch in self.local_data:
                optimizer.zero_grad()
                output = self.model(batch["inputs"])
                loss = torch.nn.functional.cross_entropy(
                    output, batch["labels"]
                )
                loss.backward()
                optimizer.step()
        
        # 返回更新后的权重（不含原始数据）
        return self.model.state_dict()
    
    def compute_update_diff(self, old_state, new_state):
        """
        只上传权重差异（减少通信量）
        """
        diff = OrderedDict()
        for key in old_state.keys():
            delta = new_state[key] - old_state[key]
            # 可以进一步压缩（如 top-k 稀疏化、量化）
            diff[key] = delta
        return diff


# 差分隐私增强
class DPFederatedClient(FederatedLearningClient):
    """带差分隐私保护的联邦学习客户端"""
    
    def local_train(self, global_state, epochs=3, lr=1e-4, 
                    epsilon=8.0, delta=1e-5):
        # 在梯度中加入噪声，保护训练数据隐私
        from opacus import PrivacyEngine
        
        optimizer = torch.optim.SGD(self.model.parameters(), lr=lr)
        
        privacy_engine = PrivacyEngine()
        self.model, optimizer, dataloader = privacy_engine.make_private(
            module=self.model,
            optimizer=optimizer,
            data_loader=self.local_data,
            noise_multiplier=1.0,
            max_grad_norm=1.0,
        )
        
        # 训练...
        # 上传的梯度已经包含 DP 噪声

Warning

联邦学习的现实挑战：(1) 设备异构性——不同设备的算力和内存差异巨大；(2) 数据非独立同分布（Non-IID）——每个用户的数据分布不同，影响聚合效果；(3) 通信效率——数百次 round 的通信开销可能比直接收集数据还大；(4) 设备可用性——移动设备可能在训练中途离线。联邦学习在特定场景（医疗、金融）有价值，但不是万灵药。

22.4 实时性与能耗约束

推理调度

# 边缘设备上的推理调度器

import time
import threading
from queue import PriorityQueue
from dataclasses import dataclass, field
from enum import Enum

class TaskPriority(Enum):
    CRITICAL = 0     # 安全关键（如碰撞检测）
    HIGH = 1         # 用户交互（如语音识别）
    NORMAL = 2       # 常规分析（如场景理解）
    BACKGROUND = 3   # 后台任务（如模型更新）

@dataclass(order=True)
class InferenceTask:
    priority: int                          # 优先级（越小越高）
    deadline: float                        # 截止时间（timestamp）
    task_id: str = field(compare=False)    
    model_name: str = field(compare=False) 
    input_data: object = field(compare=False)
    callback: object = field(compare=False)

class EdgeInferenceScheduler:
    """
    边缘设备推理调度器
    
    策略：
    - 优先级调度：安全关键任务优先
    - 截止时间感知：满足延迟约束
    - 能耗感知：在低电量时降低非关键推理频率
    - 动态批处理：将短时间内的请求合并
    """
    
    def __init__(self, config):
        self.task_queue = PriorityQueue()
        self.running = False
        self.battery_level = 1.0  # 0.0 - 1.0
        self.models = {}          # model_name -> loaded_model
        
        # 能耗策略
        self.power_modes = {
            "full": {"max_concurrent": 4, "batch_timeout_ms": 10},
            "balanced": {"max_concurrent": 2, "batch_timeout_ms": 30},
            "low_power": {"max_concurrent": 1, "batch_timeout_ms": 100},
        }
        self.current_mode = "balanced"
    
    def update_power_mode(self, battery_level):
        """根据电量调整运行模式"""
        self.battery_level = battery_level
        
        if battery_level > 0.5:
            self.current_mode = "full"
        elif battery_level > 0.2:
            self.current_mode = "balanced"
        else:
            self.current_mode = "low_power"
        
        mode = self.power_modes[self.current_mode]
        print(f"切换到 {self.current_mode} 模式: {mode}")
    
    def submit_task(self, task: InferenceTask):
        """提交推理任务"""
        # 检查是否可以降级
        if self.current_mode == "low_power":
            if task.priority >= TaskPriority.BACKGROUND.value:
                # 低电量模式下跳过后台任务
                return False
        
        self.task_queue.put(task)
        return True
    
    def run(self):
        """主调度循环"""
        self.running = True
        
        while self.running:
            try:
                # 带超时的获取（支持动态批处理）
                timeout = self.power_modes[self.current_mode]["batch_timeout_ms"] / 1000
                task = self.task_queue.get(timeout=timeout)
                
                # 收集更多任务用于批处理
                batch = [task]
                while len(batch) < self.power_modes[self.current_mode]["max_concurrent"]:
                    try:
                        next_task = self.task_queue.get_nowait()
                        batch.append(next_task)
                    except:
                        break
                
                # 检查截止时间
                now = time.time()
                valid_batch = []
                for t in batch:
                    if t.deadline > now:
                        valid_batch.append(t)
                    else:
                        # 错过截止时间
                        self._handle_missed_deadline(t)
                
                if valid_batch:
                    self._execute_batch(valid_batch)
                    
            except:
                # 队列为空，进入低功耗等待
                time.sleep(0.01)
    
    def _execute_batch(self, batch):
        """执行一批推理任务"""
        # 可以使用动态批处理优化
        for task in batch:
            model = self.models[task.model_name]
            result = model(task.input_data)
            if task.callback:
                task.callback(result)

功耗管理

# 功耗监控与管理

class PowerManager:
    """
    边缘设备功耗管理器
    
    监控维度：
    - GPU/NPU 功耗
    - 内存带宽利用
    - CPU 使用率
    - 温度
    """
    
    def __init__(self):
        self.power_budget_watts = 15.0  # 功耗预算
        self.current_consumption = 0
        self.throttle_threshold = 0.85   # 85% 预算时开始降频
        self.critical_threshold = 0.95   # 95% 预算时停止非关键任务
        
    def monitor_power(self):
        """实时监控功耗"""
        while True:
            # 读取传感器
            gpu_power = self._read_gpu_power()
            cpu_power = self._read_cpu_power()
            memory_power = self._read_memory_power()
            
            self.current_consumption = gpu_power + cpu_power + memory_power
            utilization = self.current_consumption / self.power_budget_watts
            
            if utilization > self.critical_threshold:
                self._emergency_throttle()
            elif utilization > self.throttle_threshold:
                self._gradual_throttle()
            
            # 温度检查
            temp = self._read_temperature()
            if temp > 85:
                self._thermal_throttle()
            
            yield {
                "timestamp": time.time(),
                "gpu_power_w": gpu_power,
                "cpu_power_w": cpu_power,
                "memory_power_w": memory_power,
                "total_power_w": self.current_consumption,
                "budget_utilization": utilization,
                "temperature_c": temp,
            }
            
            time.sleep(0.1)  # 10Hz 监控
    
    def _thermal_throttle(self):
        """热保护降频"""
        # 降低推理频率
        # 降低 GPU 频率
        # 如果温度持续升高，暂停推理
        pass
    
    def estimate_inference_power(self, model_name, batch_size):
        """
        预估单次推理的能耗
        
        用于在功耗预算内最大化吞吐
        """
        # 基于模型 FLOPs 和硬件能效比估算
        model_flops = self._get_model_flops(model_name)  # GFLOPs
        hardware_efficiency = 5.0  # TOPS/W (INT8)
        
        inference_energy_joules = (model_flops * batch_size) / hardware_efficiency
        return inference_energy_joules


# 推理频率 vs 精度 vs 功耗的权衡
class AdaptiveInferenceFrequency:
    """
    自适应推理频率
    
    场景：视频流分析（如行人检测）
    - 高运动场景：需要高频推理（30 FPS）
    - 静止场景：可以降低频率（5 FPS）
    """
    
    def __init__(self, target_fps=30, min_fps=5, max_power=10.0):
        self.target_fps = target_fps
        self.min_fps = min_fps
        self.max_power = max_power
        self.current_fps = target_fps
    
    def adapt(self, motion_score, power_usage, latency):
        """
        根据运动强度、功耗和延迟自适应调整
        """
        # 如果功耗过高，降低 FPS
        if power_usage > self.max_power * 0.9:
            self.current_fps = max(self.min_fps, self.current_fps * 0.8)
        
        # 如果场景运动量大，提高 FPS
        elif motion_score > 0.5 and power_usage < self.max_power * 0.7:
            self.current_fps = min(self.target_fps, self.current_fps * 1.2)
        
        # 如果延迟过高，降低 FPS
        if latency > 1.0 / self.current_fps * 0.8:  # 接近帧间隔
            self.current_fps = max(self.min_fps, self.current_fps * 0.9)
        
        return self.current_fps

热设计考量

# 边缘设备的热设计规格（参考）

thermal_design:
  # 关键温度阈值
  thresholds:
    normal: 45         # °C，正常工作
    warm: 60           # °C，开始降频
    hot: 75            # °C，限制推理频率
    critical: 85       # °C，暂停非关键推理
    shutdown: 95       # °C，系统关机保护
  
  # 散热方案选择
  cooling_solutions:
    - name: "被动散热（散热片）"
      max_heat_dissipation: "5-8 W"
      suitable_for: "智能手机、平板、小型 IoT"
      pros: ["无噪音", "无故障点"]
      cons: ["散热能力有限"]
    
    - name: "主动散热（风扇）"
      max_heat_dissipation: "30-100 W"
      suitable_for: ["Jetson AGX", "工业 PC", "边缘服务器"]
      pros: ["散热能力强"]
      cons: ["噪音", "风扇故障", "灰尘积累"]
    
    - name: "液冷"
      max_heat_dissipation: "200+ W"
      suitable_for: ["高性能边缘服务器", "车载计算平台"]
      pros: ["散热效率最高", "噪音低"]
      cons: ["成本高", "复杂度高", "泄漏风险"]
  
  # 热节流策略
  throttle_strategy:
    level_1:  # 60-75°C
      actions:
        - "降低 GPU 频率 10%"
        - "降低推理批处理大小"
    level_2:  # 75-85°C
      actions:
        - "降低 GPU 频率 25%"
        - "跳过低优先级推理任务"
        - "增加推理间隔"
    level_3:  # 85-95°C
      actions:
        - "暂停所有非关键推理"
        - "仅保留安全关键功能"
        - "最大风扇转速"
        - "如果持续 60 秒，准备优雅关机"

Tip

车载场景的特殊考量：车载 AI 计算平台面临极端温度（-40°C 到 +85°C 环境温度）和持续高负载。建议：(1) 使用工业级芯片（宽温范围）；(2) 设计冗余散热（主风扇 + 被动散热备份）；(3) 在温度达到 level 2 时就开始卸载非关键任务到云端。

小结

边缘 AI 是 AI 基础设施中被低估的重要领域。与云端训练不同，边缘推理面临的是”戴着镣铐跳舞”——在严格的功耗、内存和延迟约束下提供高质量的推理服务。本章覆盖了四个关键领域：

1. 场景与约束：理解”为什么边缘”和”边缘有多受限”是设计的出发点。不同场景（手机、车端、工业、IoT）的约束差异巨大
2. 模型轻量化：量化（INT8/INT4）、知识蒸馏和剪枝是三板斧。ONNX Runtime 提供了跨平台的推理引擎
3. 边云协同：模型分发管道和联邦学习实现了”数据不动，模型动”的理念。OTA 更新和 A/B 测试是生产环境的必备能力
4. 实时性与能耗：推理调度器、功耗管理和热设计是让系统”可持续运行”的基础。自适应推理频率可以在保证体验的同时节省能耗

边缘 AI 的核心思维转变是：从追求峰值性能到追求能效比。一个在基准测试中慢 2 倍但功耗低 3 倍的方案，在实际产品中往往是更优的选择。

延伸阅读

• ONNX Runtime：https://onnxruntime.ai/ — 微软维护的跨平台推理引擎，文档详尽
• Apple Core ML：Apple 的端侧 ML 框架文档，关于 ANE 利用有详细说明
• TensorFlow Lite：Google 的移动端推理框架，支持 Android NNAPI 和 GPU Delegate
• Federated Learning：McMahan et al., “Communication-Efficient Learning of Deep Networks from Decentralized Data” — FedAvg 原始论文
• AWQ：Lin et al., “AWQ: Activation-aware Weight Quantization for LLM Compression and Acceleration”
• GPTQ：Frantar et al., “GPTQ: Accurate Post-Training Quantization for Generative Pre-trained Transformers”
• Jetson AI Lab：NVIDIA 的边缘 AI 实验室，有丰富的实战教程