AI 边缘计算与端侧部署完全教程

1. AI 边缘计算概述与硬件选型

AI 模型训练通常在云端 GPU 集群上完成，但推理部署面临不同挑战：延迟要求（自动驾驶需要毫秒级响应）、隐私约束（医疗数据不能离院）、带宽限制（工业场景网络不稳定）、成本控制（云端推理费用随请求量线性增长）。边缘计算将推理能力下沉到靠近数据源的设备上，解决上述问题。

主流边缘硬件平台

硬件选型决策流程

def select_hardware(requirements):
    """边缘硬件选型决策辅助"""
    platforms = {
        'jetson_orin': {'tops': 275, 'power_w': 60, 'price': 999, 'framework': 'TensorRT'},
        'snapdragon_8g3': {'tops': 73, 'power_w': 8, 'price': 0, 'framework': 'SNPE'},
        'apple_m4': {'tops': 35, 'power_w': 22, 'price': 0, 'framework': 'CoreML'},
        'edge_tpu': {'tops': 4, 'power_w': 2, 'price': 75, 'framework': 'TF Lite'},
        'rk3588': {'tops': 6, 'power_w': 10, 'price': 80, 'framework': 'RKNN'},
    }

    candidates = []
    for name, spec in platforms.items():
        # 算力是否满足
        if spec['tops'] < requirements.get('min_tops', 0):
            continue
        # 功耗是否满足
        if spec['power_w'] > requirements.get('max_power', 999):
            continue
        # 模型大小是否满足
        candidates.append((name, spec))

    if not candidates:
        return "无满足条件的平台，考虑模型压缩或云端部署"
    # 按功耗效率排序
    candidates.sort(key=lambda x: x[1]['tops'] / x[1]['power_w'], reverse=True)
    return candidates[0]

req = {'min_tops': 5, 'max_power': 15}
best = select_hardware(req)
print(f"推荐平台: {best[0]}, 算力: {best[1]['tops']} TOPS")

2. 模型量化（INT8/INT4/PTQ/QAT）

量化是将模型权重和激活值从浮点数（FP32/FP16）映射到低比特整数的技术，能显著减小模型体积、加速推理。

量化基础原理

对称量化公式：\(x_q = \text{round}(\frac{x}{s})\)，其中缩放因子 \(s = \frac{\max(|x|)}{2^{b-1} - 1}\)。

非对称量化还增加了零点偏移：\(x_q = \text{round}(\frac{x}{s}) + z\)。

import torch
import torch.quantization as quant

# ============================================
# 方法1：训练后量化（PTQ - Post-Training Quantization）
# ============================================
class SimpleModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = torch.nn.Linear(784, 256)
        self.fc2 = torch.nn.Linear(256, 128)
        self.fc3 = torch.nn.Linear(128, 10)
        self.relu = torch.nn.ReLU()

    def forward(self, x):
        x = self.relu(self.fc1(x))
        x = self.relu(self.fc2(x))
        return self.fc3(x)

model = SimpleModel()
model.eval()

# PyTorch 动态量化（权重静态量化，激活动态量化）
quantized_model = torch.quantization.quantize_dynamic(
    model,
    {torch.nn.Linear},  # 量化目标层
    dtype=torch.qint8   # 量化精度
)

# 比较模型大小
def model_size_mb(m):
    torch.save(m.state_dict(), '/tmp/model.pt')
    import os
    size = os.path.getsize('/tmp/model.pt') / (1024 * 1024)
    os.remove('/tmp/model.pt')
    return size

print(f"原始模型: {model_size_mb(model):.2f} MB")
print(f"量化模型: {model_size_mb(quantized_model):.2f} MB")

# ============================================
# 方法2：静态量化（需要校准数据）
# ============================================
model_static = SimpleModel()
model_static.eval()

# 配置量化方案
model_static.qconfig = torch.quantization.get_default_qconfig('qnnpack')
# 准备：插入观察者
model_prepared = torch.quantization.prepare(model_static)
# 校准：用真实数据运行
calibration_data = torch.randn(100, 784)
with torch.no_grad():
    model_prepared(calibration_data)
# 转换
model_quantized = torch.quantization.convert(model_prepared)
print(f"静态量化模型: {model_size_mb(model_quantized):.2f} MB")

# ============================================
# 方法3：量化感知训练（QAT）
# ============================================
model_qat = SimpleModel()
model_qat.qconfig = torch.quantization.get_default_qat_qconfig('qnnpack')
model_qat_prepared = torch.quantization.prepare_qat(model_qat.train())

# 模拟训练过程
optimizer = torch.optim.SGD(model_qat_prepared.parameters(), lr=0.01)
for epoch in range(5):
    inputs = torch.randn(32, 784)
    targets = torch.randint(0, 10, (32,))
    outputs = model_qat_prepared(inputs)
    loss = torch.nn.functional.cross_entropy(outputs, targets)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

model_qat_prepared.eval()
model_qat_quantized = torch.quantization.convert(model_qat_prepared)
print(f"QAT模型: {model_size_mb(model_qat_quantized):.2f} MB")

INT4 量化与 GPTQ/AWQ

对于大语言模型，INT4 量化（如 GPTQ、AWQ）可以在几乎不损失精度的情况下将模型压缩到原来的 1/4。

# 使用 bitsandbytes 进行 INT4 量化（以 Transformer 模型为例）
# pip install bitsandbytes accelerate

# 示例伪代码（需要 GPU 环境运行）
"""
from transformers import AutoModelForCausalLM, BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_compute_dtype=torch.float16,
    bnb_4bit_quant_type="nf4",  # NormalFloat4
    bnb_4bit_use_double_quant=True,  # 双重量化
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    quantization_config=quantization_config,
    device_map="auto"
)
# 7B模型从~14GB压缩到~4GB
"""

3. 模型剪枝与知识蒸馏

模型剪枝

剪枝通过移除不重要的权重或神经元来减小模型。结构化剪枝（移除整个通道/注意力头）对硬件更友好，因为不需要稀疏计算支持。

import torch.nn.utils.prune as prune

model = SimpleModel()

# 非结构化剪枝：按比例移除最小权重
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        prune.l1_unstructured(module, name='weight', amount=0.3)  # 剪掉30%

# 查看稀疏度
def check_sparsity(model):
    total, zeros = 0, 0
    for name, param in model.named_parameters():
        if 'weight' in name:
            total += param.numel()
            zeros += (param == 0).sum().item()
    print(f"稀疏度: {zeros/total*100:.1f}%")

check_sparsity(model)

# 结构化剪枝：移除整个输出通道
model2 = SimpleModel()
prune.ln_structured(
    model2.fc1, name='weight', amount=0.2, n=2, dim=0
)
# fc1的20%输出通道被置零，可直接裁剪减小模型

# 永久化剪枝（移除mask，直接修改权重）
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        prune.remove(module, 'weight')

知识蒸馏

知识蒸馏用大模型（Teacher）指导小模型（Student）学习，Student 不仅学习硬标签，还学习 Teacher 的软输出分布。

class DistillationLoss(torch.nn.Module):
    def __init__(self, temperature=4.0, alpha=0.7):
        super().__init__()
        self.temperature = temperature
        self.alpha = alpha
        self.ce_loss = torch.nn.CrossEntropyLoss()
        self.kl_loss = torch.nn.KLDivLoss(reduction='batchmean')

    def forward(self, student_logits, teacher_logits, labels):
        # 硬标签损失
        hard_loss = self.ce_loss(student_logits, labels)

        # 软标签损失（蒸馏损失）
        soft_student = torch.log_softmax(student_logits / self.temperature, dim=1)
        soft_teacher = torch.softmax(teacher_logits / self.temperature, dim=1)
        soft_loss = self.kl_loss(soft_student, soft_teacher)
        soft_loss *= self.temperature ** 2

        return self.alpha * soft_loss + (1 - self.alpha) * hard_loss

# 蒸馏训练循环
teacher = SimpleModel()  # 假设已训练好
student = torch.nn.Sequential(
    torch.nn.Linear(784, 64),
    torch.nn.ReLU(),
    torch.nn.Linear(64, 10)
)

distill_loss_fn = DistillationLoss(temperature=4.0, alpha=0.7)
optimizer = torch.optim.Adam(student.parameters(), lr=1e-3)

for epoch in range(10):
    inputs = torch.randn(32, 784)
    targets = torch.randint(0, 10, (32,))
    with torch.no_grad():
        teacher_logits = teacher(inputs)
    student_logits = student(inputs)
    loss = distill_loss_fn(student_logits, teacher_logits, targets)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

4. ONNX Runtime 与模型转换

ONNX（Open Neural Network Exchange）是模型交换的开放标准，ONNX Runtime 是微软开发的高性能推理引擎，支持多种硬件后端。

import torch

# 导出模型为ONNX格式
model = SimpleModel()
model.eval()
dummy_input = torch.randn(1, 784)

torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    input_names=['input'],
    output_names=['output'],
    dynamic_axes={'input': {0: 'batch_size'}, 'output': {0: 'batch_size'}},
    opset_version=13
)
print("模型已导出为 model.onnx")

# 使用ONNX Runtime推理
# pip install onnxruntime
import onnxruntime as ort

session = ort.InferenceSession("model.onnx",
    providers=['CPUExecutionProvider'])

# 推理
inputs_np = torch.randn(1, 784).numpy()
result = session.run(None, {'input': inputs_np})
print(f"ONNX输出: {result[0].shape}")

# 性能对比
import time

def benchmark(func, inputs, n_runs=100):
    # 预热
    for _ in range(10):
        func(inputs)
    start = time.perf_counter()
    for _ in range(n_runs):
        func(inputs)
    elapsed = (time.perf_counter() - start) / n_runs * 1000
    return elapsed

# PyTorch推理
pytorch_time = benchmark(
    lambda x: model(torch.from_numpy(x)),
    inputs_np
)

# ONNX推理
onnx_time = benchmark(
    lambda x: session.run(None, {'input': x}),
    inputs_np
)

print(f"PyTorch: {pytorch_time:.2f}ms, ONNX: {onnx_time:.2f}ms")
print(f"加速比: {pytorch_time / onnx_time:.1f}x")

ONNX 模型优化

# ONNX Runtime 图优化
from onnxruntime.transformers import optimizer

# 设置优化级别
sess_options = ort.SessionOptions()
sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
sess_options.optimized_model_filepath = "model_optimized.onnx"

session_opt = ort.InferenceSession("model.onnx", sess_options)

# 量化ONNX模型
from onnxruntime.quantization import quantize_dynamic, QuantType

quantize_dynamic(
    model_input="model.onnx",
    model_output="model_quantized.onnx",
    weight_type=QuantType.QInt8
)
print("ONNX量化模型已生成")

5. 移动端部署（TFLite / Core ML / MNN）

TensorFlow Lite

# TFLite模型转换
# pip install tensorflow

"""
import tensorflow as tf

# 加载SavedModel
converter = tf.lite.TFLiteConverter.from_saved_model("saved_model_dir")

# FP16量化
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.float16]

# INT8全量化（需要代表性数据集）
def representative_dataset():
    for _ in range(100):
        yield [tf.random.uniform([1, 224, 224, 3])]

converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type = tf.int8
converter.inference_output_type = tf.int8

tflite_model = converter.convert()
with open("model.tflite", "wb") as f:
    f.write(tflite_model)
"""

# Android端TFLite推理（Java/Kotlin伪代码示例）
"""
// build.gradle 依赖
// implementation 'org.tensorflow:tensorflow-lite:2.14.0'

// 加载模型
Interpreter tflite = new Interpreter(loadModelFile("model.tflite"));

// 推理
float[][] input = new float[1][784];
float[][] output = new float[1][10];
tflite.run(input, output);
"""

Core ML（Apple 生态）

# Core ML模型转换
# pip install coremltools

"""
import coremltools as ct
import torch

model = SimpleModel()
model.eval()
traced = torch.jit.trace(model, torch.randn(1, 784))

coreml_model = ct.convert(
    traced,
    inputs=[ct.TensorType(name="input", shape=(1, 784))],
    minimum_deployment_target=ct.target.iOS16,
)

coreml_model.save("model.mlpackage")

# Swift端推理代码
# let model = try model(configuration: .init())
# let input = modelInput(input: MLMultiArray(...))
# let output = try model.prediction(input: input)
"""

MNN（阿里开源）

# MNN 转换示例（命令行工具）
"""
# 将ONNX模型转为MNN格式
MNNConvert -f ONNX --modelFile model.onnx --MNNModel model.mnn --bizCode biz

# Python推理
import MNN

interpreter = MNN.Interpreter("model.mnn")
session = interpreter.createSession()
input_tensor = interpreter.getSessionInput(session)

# 设置输入数据
input_data = MNN.Tensor((1, 784), MNN.Halide_Type_Float, ...)
input_tensor.copyFrom(input_data)
interpreter.runSession(session)
output_tensor = interpreter.getSessionOutput(session)
"""

6. 浏览器端 AI（WebGPU / WASM / Transformers.js）

浏览器端推理可以保护用户隐私（数据不离开设备）、消除服务器成本、支持离线使用。

WebGPU 推理

// WebGPU 矩阵乘法（WGSL着色器）
const shaderCode = `
@group(0) @binding(0) var<storage, read> a: array<f32>;
@group(0) @binding(1) var<storage, read> b: array<f32>;
@group(0) @binding(2) var<storage, read_write> c: array<f32>;

@compute @workgroup_size(8, 8)
fn main(@builtin(global_invocation_id) id: vec3<u32>) {
    let row = id.x;
    let col = id.y;
    let N: u32 = 64u;  // 矩阵维度
    var sum: f32 = 0.0;
    for (var k: u32 = 0u; k < N; k = k + 1u) {
        sum = sum + a[row * N + k] * b[k * N + col];
    }
    c[row * N + col] = sum;
}
`;

async function webgpuMatmul() {
    if (!navigator.gpu) {
        console.log("WebGPU 不可用");
        return;
    }

    const adapter = await navigator.gpu.requestAdapter();
    const device = await adapter.requestDevice();

    const N = 64;
    const size = N * N * 4; // f32

    // 创建缓冲区
    const bufA = device.createBuffer({
        size, usage: GPUBufferUsage.STORAGE | GPUBufferUsage.COPY_DST
    });
    const bufB = device.createBuffer({
        size, usage: GPUBufferUsage.STORAGE | GPUBufferUsage.COPY_DST
    });
    const bufC = device.createBuffer({
        size, usage: GPUBufferUsage.STORAGE | GPUBufferUsage.COPY_SRC
    });

    // 加载数据
    const dataA = new Float32Array(N * N).fill(1.0);
    const dataB = new Float32Array(N * N).fill(0.5);
    device.queue.writeBuffer(bufA, 0, dataA);
    device.queue.writeBuffer(bufB, 0, dataB);

    // 创建计算管线
    const shader = device.createShaderModule({ code: shaderCode });
    const pipeline = device.createComputePipeline({
        layout: 'auto',
        compute: { module: shader, entryPoint: 'main' }
    });

    const bindGroup = device.createBindGroup({
        layout: pipeline.getBindGroupLayout(0),
        entries: [
            { binding: 0, resource: { buffer: bufA } },
            { binding: 1, resource: { buffer: bufB } },
            { binding: 2, resource: { buffer: bufC } },
        ]
    });

    const encoder = device.createCommandEncoder();
    const pass = encoder.beginComputePass();
    pass.setPipeline(pipeline);
    pass.setBindGroup(0, bindGroup);
    pass.dispatchWorkgroups(8, 8);
    pass.end();
    device.queue.submit([encoder.finish()]);

    console.log("WebGPU 计算完成");
}

Transformers.js（Hugging Face）

// 使用 Transformers.js 在浏览器中运行模型
// <script src="https://cdn.jsdelivr.net/npm/@huggingface/transformers"></script>

/*
import { pipeline } from '@huggingface/transformers';

// 文本分类
const classifier = await pipeline(
    'text-classification',
    'Xenova/distilbert-base-uncased-finetuned-sst-2-english'
);
const result = await classifier('This movie is amazing!');
console.log(result);
// [{ label: 'POSITIVE', score: 0.9998 }]

// 图像分类
const imageClassifier = await pipeline(
    'image-classification',
    'Xenova/vit-base-patch16-224'
);
const imgResult = await imageClassifier('https://example.com/cat.jpg');

// 文本生成
const generator = await pipeline(
    'text-generation',
    'Xenova/gpt2'
);
const generated = await generator('Once upon a time', {
    max_new_tokens: 50
});

// 使用 WebGPU 后端（需要浏览器支持）
const model = await pipeline(
    'text-classification',
    'Xenova/distilbert-base-uncased-finetuned-sst-2-english',
    { device: 'webgpu' }
);
*/

ONNX Runtime Web

// ONNX Runtime Web（WASM后端）
/*
import * as ort from 'onnxruntime-web';

// 配置WASM执行后端
ort.env.wasm.numThreads = 4;

async function runInference() {
    const session = await ort.InferenceSession.create('model.onnx', {
        executionProviders: ['wasm'],  // 或 'webgpu'
    });

    const input = new ort.Tensor('float32', new Float32Array(784), [1, 784]);
    const results = await session.run({ input: input });
    const output = results.output.data;
    console.log('推理结果:', output);
}
*/

7. 嵌入式部署（TensorRT / OpenVINO / NCNN）

TensorRT（NVIDIA）

# TensorRT 模型优化与部署
"""
import tensorrt as trt

# 1. ONNX转TensorRT引擎
logger = trt.Logger(trt.Logger.WARNING)
builder = trt.Builder(logger)
network = builder.create_network(
    1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
)
parser = trt.OnnxParser(network, logger)

with open("model.onnx", "rb") as f:
    parser.parse(f.read())

# 配置优化参数
config = builder.create_builder_config()
config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 28)  # 256MB
config.set_flag(trt.BuilderFlag.FP16)  # 启用FP16

# 构建引擎
engine = builder.build_serialized_network(network, config)

# 保存引擎
with open("model.engine", "wb") as f:
    f.write(engine)

# 2. 推理
runtime = trt.Runtime(logger)
engine = runtime.deserialize_cuda_engine(engine)
context = engine.create_execution_context()

# 分配GPU内存并执行推理
# ... (需要pycuda或cuda-python)
"""

OpenVINO（Intel）

# OpenVINO 模型优化
"""
from openvino.runtime import Core

core = Core()

# 读取并编译模型（自动优化）
model = core.read_model("model.onnx")
compiled = core.compile_model(model, "CPU")  # 也可选 GPU, AUTO

# 推理
import numpy as np
input_data = np.random.randn(1, 784).astype(np.float32)
result = compiled(input_data)

# INT8量化（Post-Training）
from openvino.tools.pot import compress_model_weights
from openvino.tools.pot import IEEngine

# 配置量化参数
engine = IEEngine(config={"device": "CPU"}, data_loader=calibration_loader)
# ... 量化流程
"""

NCNN（腾讯开源）

# NCNN 特别适合移动端和嵌入式设备，纯C++实现，无第三方依赖
"""
// C++ 推理代码
#include <net.h>

ncnn::Net net;
net.opt.use_vulkan_compute = true;  // GPU加速（如支持）
net.load_param("model.param");
net.load_model("model.bin");

ncnn::Mat in = ncnn::Mat(224, 224, 3, input_data);
ncnn::Mat out;

ncnn::Extractor ex = net.create_extractor();
ex.input("input", in);
ex.extract("output", out);

// Python绑定
# import ncnn
# net = ncnn.Net()
# net.load_param("model.param")
# net.load_model("model.bin")
"""

8. 端云协同推理架构

端云协同将推理任务在边缘设备和云端之间动态分配，兼顾延迟、精度和成本。

class EdgeCloudOrchestrator:
    """端云协同推理调度器"""
    def __init__(self, edge_model, cloud_model, confidence_threshold=0.85):
        self.edge_model = edge_model  # 轻量边缘模型
        self.cloud_model = cloud_model  # 重量云端模型
        self.threshold = confidence_threshold
        self.stats = {'edge': 0, 'cloud': 0}

    def infer(self, input_data, latency_budget_ms=100):
        """
        推理策略：
        1. 边缘设备先推理
        2. 如果置信度足够高，直接返回
        3. 否则将输入发送到云端推理
        """
        import time

        # 边缘推理
        start = time.perf_counter()
        edge_result = self.edge_model(input_data)
        edge_latency = (time.perf_counter() - start) * 1000

        confidence = edge_result.max().item()

        if confidence >= self.threshold:
            self.stats['edge'] += 1
            return {
                'result': edge_result,
                'confidence': confidence,
                'source': 'edge',
                'latency_ms': edge_latency
            }

        # 置信度不足，发送到云端
        cloud_result = self.cloud_model(input_data)
        self.stats['cloud'] += 1
        return {
            'result': cloud_result,
            'confidence': cloud_result.max().item(),
            'source': 'cloud',
            'latency_ms': edge_latency + 50  # 模拟网络延迟
        }

    def get_stats(self):
        total = self.stats['edge'] + self.stats['cloud']
        return {
            'edge_ratio': self.stats['edge'] / max(total, 1),
            'cloud_ratio': self.stats['cloud'] / max(total, 1),
            'total': total
        }

# 使用示例
edge_net = torch.nn.Linear(784, 10)
cloud_net = torch.nn.Sequential(
    torch.nn.Linear(784, 512),
    torch.nn.ReLU(),
    torch.nn.Linear(512, 256),
    torch.nn.ReLU(),
    torch.nn.Linear(256, 10)
)

orchestrator = EdgeCloudOrchestrator(edge_net, cloud_net, confidence_threshold=0.9)

for _ in range(10):
    data = torch.randn(1, 784)
    result = orchestrator.infer(data)
    print(f"来源: {result['source']}, 置信度: {result['confidence']:.3f}")

print(f"统计: {orchestrator.get_stats()}")

早退机制（Early Exit）

在模型中间层设置分支，简单样本在浅层就完成推理，复杂样本才走完整网络。

class EarlyExitModel(torch.nn.Module):
    def __init__(self, n_input, n_classes, exit_threshold=0.9):
        super().__init__()
        self.threshold = exit_threshold

        self.layer1 = torch.nn.Linear(n_input, 256)
        self.exit1 = torch.nn.Linear(256, n_classes)

        self.layer2 = torch.nn.Linear(256, 128)
        self.exit2 = torch.nn.Linear(128, n_classes)

        self.layer3 = torch.nn.Linear(128, 64)
        self.final = torch.nn.Linear(64, n_classes)

    def forward(self, x):
        # 第一层 + 早退检查
        h = torch.relu(self.layer1(x))
        out1 = torch.softmax(self.exit1(h), dim=1)
        if out1.max().item() > self.threshold:
            return out1, 1  # 返回结果和退出层数

        # 第二层 + 早退检查
        h = torch.relu(self.layer2(h))
        out2 = torch.softmax(self.exit2(h), dim=1)
        if out2.max().item() > self.threshold:
            return out2, 2

        # 最终层
        h = torch.relu(self.layer3(h))
        out3 = torch.softmax(self.final(h), dim=1)
        return out3, 3

model = EarlyExitModel(784, 10)
data = torch.randn(5, 784)
for i in range(5):
    result, exit_layer = model(data[i:i+1])
    print(f"样本{i}: 退出层={exit_layer}, 置信度={result.max():.3f}")

9. 隐私保护与联邦推理

在医疗、金融等隐私敏感场景，数据不能离开本地设备。联邦推理让多个设备协作完成推理而不共享原始数据。

import torch
import torch.nn as nn

class SplitLearning:
    """分割学习：模型在设备和服务器之间分割"""
    def __init__(self, client_model, server_model):
        self.client_model = client_model  # 前几层，在设备上
        self.server_model = server_model  # 后几层，在服务器上

    def client_forward(self, x):
        """设备端：运行模型前半部分"""
        with torch.no_grad():
            intermediate = self.client_model(x)
        return intermediate  # 只传输中间表示，不传输原始数据

    def server_forward(self, intermediate):
        """服务器端：运行模型后半部分"""
        output = self.server_model(intermediate)
        return output

# 示例
client_net = nn.Sequential(
    nn.Linear(784, 128),
    nn.ReLU(),
    nn.Linear(128, 32)
)
server_net = nn.Sequential(
    nn.ReLU(),
    nn.Linear(32, 10)
)

split = SplitLearning(client_net, server_net)
data = torch.randn(1, 784)

# 设备端处理
intermediate = split.client_forward(data)
print(f"传输的中间表示大小: {intermediate.shape}")  # [1, 32] vs 原始 [1, 784]
print(f"数据压缩比: {784 / 32:.0f}x")

# 服务器端推理
output = split.server_forward(intermediate)
print(f"最终输出: {output.shape}")

# ============================================
# 差分隐私：在中间表示上添加噪声
# ============================================
def add_dp_noise(intermediate, epsilon=1.0):
    """添加差分隐私噪声"""
    sensitivity = 1.0  # L2敏感度
    noise_scale = sensitivity / epsilon
    noise = torch.randn_like(intermediate) * noise_scale
    return intermediate + noise

private_intermediate = add_dp_noise(intermediate, epsilon=2.0)
private_output = split.server_forward(private_intermediate)
print(f"差分隐私推理完成, epsilon=2.0")

10. 实战案例：手机端实时目标检测

以 YOLOv8-nano 为例，展示从模型导出到手机部署的完整流程。

# ============================================
# 步骤1：导出ONNX模型
# ============================================
"""
from ultralytics import YOLO

model = YOLO("yolov8n.pt")  # nano版本，仅3.2M参数
model.export(format="onnx", imgsz=320, simplify=True, opset=13)
# 生成 yolov8n.onnx
"""

# ============================================
# 步骤2：转换为TFLite（Android）或CoreML（iOS）
# ============================================
"""
# TFLite转换
import onnx
from onnx_tf.backend import prepare
import tensorflow as tf

onnx_model = onnx.load("yolov8n.onnx")
tf_rep = prepare(onnx_model)
tf_rep.export_graph("yolov8n_tf")

converter = tf.lite.TFLiteConverter.from_saved_model("yolov8n_tf")
converter.optimizations = [tf.lite.Optimize.DEFAULT]

# INT8量化
def representative_dataset():
    import cv2
    import numpy as np
    for img_path in calibration_images:
        img = cv2.imread(img_path)
        img = cv2.resize(img, (320, 320))
        img = img.astype(np.float32) / 255.0
        img = np.expand_dims(img, 0)
        yield [img]

converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
tflite_model = converter.convert()
with open("yolov8n_int8.tflite", "wb") as f:
    f.write(tflite_model)
"""

# ============================================
# 步骤3：性能评估
# ============================================
def evaluate_detection_model(onnx_path, test_images):
    """评估检测模型的精度和速度"""
    import onnxruntime as ort
    import numpy as np
    import time

    session = ort.InferenceSession(onnx_path,
        providers=['CPUExecutionProvider'])

    latencies = []
    for img in test_images:
        input_data = img.astype(np.float32) / 255.0
        input_data = np.transpose(input_data, (2, 0, 1))
        input_data = np.expand_dims(input_data, 0)

        start = time.perf_counter()
        outputs = session.run(None, {'images': input_data})
        latencies.append((time.perf_counter() - start) * 1000)

    avg_latency = np.mean(latencies)
    p95_latency = np.percentile(latencies, 95)
    fps = 1000 / avg_latency

    print(f"平均延迟: {avg_latency:.1f}ms")
    print(f"P95延迟: {p95_latency:.1f}ms")
    print(f"FPS: {fps:.1f}")

    return {'avg_ms': avg_latency, 'p95_ms': p95_latency, 'fps': fps}

# 模拟测试
test_imgs = [np.random.randint(0, 255, (320, 320, 3)) for _ in range(50)]
# evaluate_detection_model("yolov8n.onnx", test_imgs)

Android 端部署要点

// Android Kotlin 伪代码
/*
class Detector(context: Context) {
    private val interpreter: Interpreter

    init {
        val model = FileUtil.loadMappedFile(context, "yolov8n_int8.tflite")
        val options = Interpreter.Options().apply {
            setNumThreads(4)
            // 使用GPU代理
            addDelegate(GpuDelegate())
        }
        interpreter = Interpreter(model, options)
    }

    fun detect(bitmap: Bitmap): List<Detection> {
        // 预处理：缩放到320x320，归一化
        val input = preprocessImage(bitmap)

        // 推理
        val output = Array(1) { Array(25200) { FloatArray(85) } }
        interpreter.run(input, output)

        // 后处理：NMS过滤
        return postprocess(output, confThreshold = 0.25f, iouThreshold = 0.45f)
    }
}
*/

11. 性能优化与功耗管理

推理性能优化技巧

import torch
import time

class InferenceOptimizer:
    """端侧推理优化工具集"""

    @staticmethod
    def optimize_pytorch_model(model, input_shape):
        """PyTorch模型优化"""
        model.eval()

        # 1. TorchScript追踪
        dummy = torch.randn(input_shape)
        traced = torch.jit.trace(model, dummy)

        # 2. 融合BN层
        torch.quantization.fuse_modules(
            model, [['conv', 'bn', 'relu']], inplace=True
        )

        # 3. 禁用梯度计算
        with torch.no_grad():
            # 4. 使用Channels Last内存格式（对ARM CPU友好）
            model = model.to(memory_format=torch.channels_last)
            dummy = dummy.to(memory_format=torch.channels_last)

            # 预热
            for _ in range(5):
                model(dummy)

            # 基准测试
            latencies = []
            for _ in range(100):
                start = time.perf_counter()
                model(dummy)
                latencies.append((time.perf_counter() - start) * 1000)

        return {
            'avg_ms': sum(latencies) / len(latencies),
            'min_ms': min(latencies),
            'max_ms': max(latencies),
            'p95_ms': sorted(latencies)[94]
        }

    @staticmethod
    def batch_inference(model, inputs, max_batch=8, timeout_ms=50):
        """动态批处理：积累请求到一定数量或超时后一起推理"""
        import queue
        import threading

        request_queue = queue.Queue()
        results = {}

        def inference_worker():
            batch = []
            while True:
                try:
                    item = request_queue.get(timeout=timeout_ms / 1000)
                    batch.append(item)
                except queue.Empty:
                    pass

                if len(batch) >= max_batch or (batch and request_queue.empty()):
                    # 批量推理
                    batch_input = torch.stack([b['input'] for b in batch])
                    with torch.no_grad():
                        output = model(batch_input)
                    for i, b in enumerate(batch):
                        results[b['id']] = output[i]
                    batch = []

        worker = threading.Thread(target=inference_worker, daemon=True)
        worker.start()
        return request_queue, results

    @staticmethod
    def measure_power_estimation(flops, memory_bytes, frequency_mhz=1000):
        """功耗估算（简化模型）"""
        # 算术功耗：每FLOP约 4pJ（INT8）到 20pJ（FP32）
        compute_energy_nj = flops * 4e-9  # INT8

        # 内存访问功耗：每次访问约 100pJ（片上）到 10nJ（片外）
        memory_energy_nj = memory_bytes * 1e-9  # 片外访问

        total_energy_mj = (compute_energy_nj + memory_energy_nj) * 1e-6
        power_mw = total_energy_mj * frequency_mhz / 1e3

        return {
            'energy_per_inference_mj': total_energy_mj,
            'estimated_power_mw': power_mw,
            'inferences_per_second': frequency_mhz * 1e6 / flops if flops > 0 else 0
        }

# 使用示例
model = torch.nn.Sequential(
    torch.nn.Conv2d(3, 16, 3, padding=1),
    torch.nn.BatchNorm2d(16),
    torch.nn.ReLU(),
    torch.nn.AdaptiveAvgPool2d(1),
    torch.nn.Flatten(),
    torch.nn.Linear(16, 10)
)

optimizer = InferenceOptimizer()
result = optimizer.optimize_pytorch_model(model, (1, 3, 224, 224))
print(f"推理性能: {result}")

# 功耗估算（假设模型需要 500M FLOPs，10MB 内存访问）
power = optimizer.measure_power_estimation(500e6, 10e6)
print(f"单次推理能耗: {power['energy_per_inference_mj']:.3f} mJ")
print(f"估算功耗: {power['estimated_power_mw']:.1f} mW")

内存优化策略

class MemoryOptimizedInference:
    """内存优化推理"""

    @staticmethod
    def activation_checkpointing(model, input_data):
        """激活检查点：用计算换内存"""
        # 对于大模型，不在前向传播中保存所有中间激活值
        # 而是在反向传播时重新计算
        from torch.utils.checkpoint import checkpoint

        def forward_with_checkpoint(module, x):
            return checkpoint(module, x, use_reentrant=False)

        return forward_with_checkpoint(model, input_data)

    @staticmethod
    def weight_sharing(model):
        """权重共享：多个层共享同一份权重"""
        # 适用于Transformer等有重复结构的模型
        layers = list(model.children())
        if len(layers) >= 4:
            # 让第2层和第4层共享权重
            layers[3] = layers[1]
        return model

    @staticmethod
    def estimate_memory(model, input_shape, dtype=torch.float32):
        """估算模型推理内存需求"""
        dtype_size = {
            torch.float32: 4, torch.float16: 2,
            torch.int8: 1, torch.int4: 0.5
        }

        param_memory = sum(
            p.numel() * dtype_size.get(p.dtype, 4)
            for p in model.parameters()
        )

        # 激活内存（粗略估计）
        input_memory = 1
        for s in input_shape:
            input_memory *= s
        input_memory *= dtype_size.get(dtype, 4)

        total_mb = (param_memory + input_memory * 10) / (1024 * 1024)
        print(f"参数内存: {param_memory / (1024*1024):.1f} MB")
        print(f"激活内存估计: {input_memory * 10 / (1024*1024):.1f} MB")
        print(f"总内存估计: {total_mb:.1f} MB")
        return total_mb

MemoryOptimizedInference.estimate_memory(model, (1, 3, 224, 224))

边缘 AI 部署是一个涉及模型优化、硬件适配、系统工程的综合性课题。关键原则是：先量化再剪枝，先基准测试再优化，先在目标硬件上验证再投入生产。随着端侧算力的持续增长和模型压缩技术的进步，越来越多的 AI 能力将从云端下沉到设备端，实现真正的智能无处不在。