05 - 边缘部署¶

⚠️ 时效性说明：本章涉及前沿模型/价格/榜单等信息，可能随版本快速变化；请以论文原文、官方发布页和 API 文档为准。

让大模型在边缘设备上运行

📖 章节概述¶

本章将深入探讨边缘部署技术，包括边缘计算、移动端部署和嵌入式设备等内容。这些技术可以帮助你在资源受限的边缘设备上运行大模型。

🎯 学习目标¶

完成本章后，你将能够：

理解边缘部署的挑战和机遇
掌握移动端部署的方法
了解嵌入式设备的优化技巧
能够在边缘设备上部署和运行大模型

1. 边缘部署概述¶

1.1 什么是边缘部署？¶

边缘部署将AI模型部署到靠近数据源的设备上，如手机、物联网设备、边缘服务器等。

优势： - 低延迟：无需云端往返 - 隐私保护：数据不离开设备 - 离线能力：无需网络连接 - 成本降低：减少云端费用

挑战： - 资源受限：计算和内存有限 - 能耗限制：电池续航 - 硬件多样：不同设备差异大 - 更新困难：设备分散

1.2 边缘部署场景¶

Text Only

边缘部署场景
├── 移动设备
│   ├── 智能手机
│   ├── 平板电脑
│   └── 可穿戴设备
├── 物联网设备
│   ├── 智能家居
│   ├── 工业设备
│   └── 传感器
├── 边缘服务器
│   ├── 零售终端
│   ├── 工厂边缘节点
│   └── 车载系统
└── 嵌入式系统
    ├── 树莓派
    ├── Jetson系列
    └── 其他开发板

2. 移动端部署¶

2.1 移动端优化技术¶

模型优化： - 量化：INT8/INT4量化 - 剪枝：移除冗余参数 - 知识蒸馏：小模型学习大模型

推理优化： - 模型转换：ONNX、TFLite、CoreML - 硬件加速：GPU、NPU、DSP - 批处理：合并推理请求

2.2 使用ONNX Runtime Mobile¶

Python

import onnxruntime as ort
import numpy as np

# 加载ONNX模型
session = ort.InferenceSession("model.onnx", providers=['CPUExecutionProvider'])

# 获取输入输出信息
input_name = session.get_inputs()[0].name
output_name = session.get_outputs()[0].name

# 准备输入
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)

# 推理
outputs = session.run([output_name], {input_name: input_data})

# 获取结果
result = outputs[0]
print(f"推理结果: {result}")

2.3 转换模型为ONNX¶

Python

import torch
import torch.onnx
from transformers import AutoModelForCausalLM, AutoTokenizer

def convert_to_onnx(model_name, output_path="model.onnx"):
    """
    转换模型为ONNX格式

    Args:
        model_name: 模型名称或路径
        output_path: 输出路径
    """
    # 加载模型和分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)
    model.eval()  # eval()评估模式

    # 准备示例输入
    dummy_input = tokenizer("Hello, world!", return_tensors="pt")

    # 导出ONNX模型
    torch.onnx.export(
        model,
        (dummy_input["input_ids"], dummy_input["attention_mask"]),
        output_path,
        input_names=["input_ids", "attention_mask"],
        output_names=["logits"],
        dynamic_axes={
            "input_ids": {0: "batch_size", 1: "sequence_length"},
            "attention_mask": {0: "batch_size", 1: "sequence_length"},
            "logits": {0: "batch_size", 1: "sequence_length"}
        },
        opset_version=14
    )

    print(f"模型已导出到: {output_path}")

# 使用示例
# convert_to_onnx("meta-llama/Llama-2-7b-hf", "llama2.onnx")

2.4 使用TensorFlow Lite¶

Python

import tensorflow as tf

def convert_to_tflite(model_path, output_path="model.tflite"):
    """
    转换模型为TFLite格式

    Args:
        model_path: 模型路径
        output_path: 输出路径
    """
    # 加载模型
    model = tf.keras.models.load_model(model_path)

    # 转换为TFLite
    converter = tf.lite.TFLiteConverter.from_keras_model(model)

    # 优化选项
    converter.optimizations = [tf.lite.Optimize.DEFAULT]

    # 转换
    tflite_model = converter.convert()

    # 保存
    with open(output_path, 'wb') as f:  # with自动管理文件关闭
        f.write(tflite_model)

    print(f"模型已导出到: {output_path}")

# 使用示例
# convert_to_tflite("model.h5", "model.tflite")

# 推理
def tflite_inference(model_path, input_data):
    """
    TFLite推理

    Args:
        model_path: TFLite模型路径
        input_data: 输入数据
    """
    # 加载模型
    interpreter = tf.lite.Interpreter(model_path=model_path)
    interpreter.allocate_tensors()

    # 获取输入输出张量
    input_details = interpreter.get_input_details()
    output_details = interpreter.get_output_details()

    # 设置输入
    interpreter.set_tensor(input_details[0]['index'], input_data)

    # 推理
    interpreter.invoke()

    # 获取输出
    output = interpreter.get_tensor(output_details[0]['index'])

    return output

# 使用示例
# output = tflite_inference("model.tflite", input_data)

2.5 Android部署示例¶

Java

// Android应用中使用ONNX Runtime
import ai.onnxruntime.OrtEnvironment;
import ai.onnxruntime.OrtSession;

public class ONNXModel {
    private OrtEnvironment env;
    private OrtSession session;

    public ONNXModel(String modelPath) throws OrtException {
        env = OrtEnvironment.getEnvironment();
        session = env.createSession(modelPath);
    }

    public float[] predict(float[][][] input) throws OrtException {
        // 准备输入
        OnnxTensor inputTensor = OnnxTensor.createTensor(env, input);

        // 推理
        OrtSession.Result results = session.run(
            Map.of("input", inputTensor)
        );

        // 获取输出
        float[][] output = (float[][]) results.get(0).getValue();

        return output[0];
    }

    public void close() {
        if (session != null) {
            session.close();
        }
        if (env != null) {
            env.close();
        }
    }
}

// 使用示例
// ONNXModel model = new ONNXModel("model.onnx");
// float[] result = model.predict(inputData);
// model.close();

2.6 iOS部署示例¶

Swift

// iOS应用中使用CoreML
import CoreML

class CoreMLModel {
    private var model: MLModel?

    init() {
        do {
            let config = MLModelConfiguration()
            config.computeUnits = .all
            self.model = try MyModel(configuration: config)
        } catch {
            print("模型加载失败: \(error)")
        }
    }

    func predict(input: MLMultiArray) -> MLMultiArray? {
        guard let model = model else {
            return nil
        }

        do {
            let input = MyModelInput(input: input)
            let output = try model.prediction(input: input)
            return output.output
        } catch {
            print("推理失败: \(error)")
            return nil
        }
    }
}

// 使用示例
// let model = CoreMLModel()
// let result = model.predict(input: inputData)

3. 嵌入式设备部署¶

3.1 树莓派部署¶

⚠️ 重要警告：bitsandbytes 不支持 ARM 架构

BitsAndBytesConfig 依赖的 bitsandbytes 库仅支持 x86_64 架构的 GPU 环境（需要 NVIDIA GPU 和 CUDA）。树莓派使用的是 ARM 架构 CPU，无法使用 bitsandbytes 量化方案。

树莓派推荐的替代方案： - llama.cpp：支持 ARM NEON 优化的 GGUF 格式量化 - ONNX Runtime：支持 ARM 架构的 INT8 量化 - TensorFlow Lite：专为移动/边缘设备优化

方案一：使用 llama.cpp（推荐）¶

llama.cpp 是在 ARM 设备上运行 LLM 的最佳选择，支持 GGUF 格式的多种量化级别。

Python

# 注意：树莓派上推荐使用 llama.cpp 的 Python 绑定
# 安装：pip install llama-cpp-python

from llama_cpp import Llama

def deploy_on_raspberry_pi_llamacpp(model_path, n_ctx=2048, n_gpu_layers=0):
    """
    在树莓派上使用 llama.cpp 部署模型

    Args:
        model_path: GGUF 格式模型路径（需要预先转换）
        n_ctx: 上下文长度
        n_gpu_layers: GPU 层数（树莓派通常为 0，纯 CPU 推理）

    Returns:
        Llama: 加载的模型实例
    """
    # 加载 GGUF 格式的量化模型
    # 常用量化级别：Q4_K_M, Q5_K_M, Q8_0 等
    llm = Llama(
        model_path=model_path,
        n_ctx=n_ctx,
        n_gpu_layers=n_gpu_layers,  # 树莓派无 GPU，设为 0
        verbose=False
    )

    return llm

def inference_llamacpp(llm, prompt, max_tokens=100):
    """
    使用 llama.cpp 进行推理

    Args:
        llm: Llama 模型实例
        prompt: 输入提示词
        max_tokens: 最大生成token数
    """
    output = llm(
        prompt,
        max_tokens=max_tokens,
        temperature=0.7,
        top_p=0.9,
        echo=False
    )

    return output['choices'][0]['text']

# 使用示例
# llm = deploy_on_raspberry_pi_llamacpp("llama-2-7b.Q4_K_M.gguf")
# result = inference_llamacpp(llm, "请介绍一下人工智能")
# print(result)

方案二：使用 ONNX Runtime¶

ONNX Runtime 支持 ARM 架构的 INT8 量化推理。

Python

import numpy as np
import onnxruntime as ort
from transformers import AutoTokenizer

def deploy_on_raspberry_pi_onnx(model_path, tokenizer_name):
    """
    在树莓派上使用 ONNX Runtime 部署模型

    Args:
        model_path: ONNX 模型路径
        tokenizer_name: 分词器名称

    Returns:
        session: ONNX Runtime 会话
        tokenizer: 分词器
    """
    # 加载分词器
    tokenizer = AutoTokenizer.from_pretrained(tokenizer_name)

    # 创建 ONNX Runtime 会话
    # 使用 CPUExecutionProvider，支持 ARM 架构
    sess_options = ort.SessionOptions()
    sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL

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

    return session, tokenizer

def inference_onnx(session, tokenizer, prompt, max_length=100):
    """
    使用 ONNX Runtime 进行推理
    """
    # 编码输入
    inputs = tokenizer(prompt, return_tensors="np")

    # 获取输入名称
    input_names = [inp.name for inp in session.get_inputs()]
    output_names = [out.name for out in session.get_outputs()]

    # 准备输入字典
    ort_inputs = {name: inputs[name] for name in input_names if name in inputs}

    # 执行推理
    outputs = session.run(output_names, ort_inputs)

    # 解码输出（简化版，实际需要更复杂的解码逻辑）
    return outputs

# 使用示例
# session, tokenizer = deploy_on_raspberry_pi_onnx("model.onnx", "model_name")
# result = inference_onnx(session, tokenizer, "请介绍一下人工智能")

GGUF 模型转换说明¶

在部署到树莓派之前，需要先将模型转换为 GGUF 格式：

Bash

# 1. 克隆 llama.cpp
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# 2. 安装依赖
pip install -r requirements.txt

# 3. 转换 HuggingFace 模型为 GGUF 格式
python convert-hf-to-gguf.py /path/to/model --outfile model.gguf --outtype q4_k_m

# 4. 将生成的 GGUF 文件复制到树莓派

3.2 NVIDIA Jetson部署¶

Python

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

def deploy_on_jetson(model_name):
    """
    在NVIDIA Jetson上部署模型

    Args:
        model_name: 模型名称
    """
    # 设置CUDA设备
    torch.cuda.set_device(0)

    # 加载分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name)

    # 加载模型（使用TensorRT优化）
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype=torch.float16,
        device_map="cuda:0"
    )

    # 推理
    prompt = "请介绍一下人工智能"
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_length=100,
            do_sample=True,
            temperature=0.7
        )

    result = tokenizer.decode(outputs[0], skip_special_tokens=True)
    print(result)

    return model, tokenizer

# 使用示例
# model, tokenizer = deploy_on_jetson("meta-llama/Llama-2-7b-hf")

3.3 使用TensorRT优化¶

Python

import torch
import tensorrt as trt
import numpy as np
from transformers import AutoModelForCausalLM, AutoTokenizer

def convert_to_tensorrt(model_path, output_path="model.trt"):
    """
    转换ONNX模型为TensorRT格式

    Args:
        model_path: ONNX模型路径
        output_path: TensorRT引擎输出路径
    """
    # 创建TensorRT构建器
    TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
    builder = trt.Builder(TRT_LOGGER)
    network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
    parser = trt.OnnxParser(network, TRT_LOGGER)

    # 解析ONNX模型
    with open(model_path, 'rb') as model:
        if not parser.parse(model.read()):
            for i in range(parser.num_errors):
                print(f"ONNX解析错误: {parser.get_error(i)}")
            raise RuntimeError("ONNX模型解析失败")

    # 构建引擎配置
    config = builder.create_builder_config()
    config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30)  # 1GB

    # 启用FP16
    if builder.platform_has_fast_fp16:
        config.set_flag(trt.BuilderFlag.FP16)

    # 序列化引擎
    serialized_engine = builder.build_serialized_network(network, config)
    if serialized_engine is None:
        raise RuntimeError("TensorRT引擎构建失败")

    # 保存引擎
    with open(output_path, 'wb') as f:
        f.write(serialized_engine)

    print(f"模型已导出到: {output_path}")

# 使用示例
# convert_to_tensorrt("model.onnx", "model.trt")

# TensorRT推理
def tensorrt_inference(engine_path, input_data):
    """
    TensorRT推理

    Args:
        engine_path: TensorRT引擎路径
        input_data: 输入数据 (numpy array)
    """
    import pycuda.driver as cuda
    import pycuda.autoinit

    # 加载引擎
    TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
    runtime = trt.Runtime(TRT_LOGGER)

    with open(engine_path, 'rb') as f:
        engine = runtime.deserialize_cuda_engine(f.read())

    # 创建执行上下文
    context = engine.create_execution_context()

    # 分配GPU内存并执行推理
    d_input = cuda.mem_alloc(input_data.nbytes)
    output = np.empty([1, 1000], dtype=np.float32)
    d_output = cuda.mem_alloc(output.nbytes)

    cuda.memcpy_htod(d_input, input_data)
    context.execute_v2([int(d_input), int(d_output)])
    cuda.memcpy_dtoh(output, d_output)

    return output

# 使用示例
# output = tensorrt_inference("model.trt", input_data)

⚠️ Jetson平台注意事项：TensorRT Python API 在 Jetson 设备上可能存在性能开销。对于生产环境，建议使用 TensorRT C++ API 以获得最佳性能。Python API 适合原型开发和快速验证。

4. 边缘部署优化¶

4.1 模型压缩¶

Python

import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM

def compress_model(model, compression_ratio=0.5):
    """
    压缩模型

    Args:
        model: 要压缩的模型
        compression_ratio: 压缩比例
    """
    # 非结构化剪枝
    for name, module in model.named_modules():
        if isinstance(module, nn.Linear):  # isinstance检查类型
            # 计算权重幅度的阈值
            weight_abs = module.weight.abs()
            threshold = torch.quantile(weight_abs, compression_ratio)

            # 创建掩码
            mask = weight_abs > threshold

            # 应用剪枝
            module.weight.data = module.weight.data * mask.float()

    return model

# 使用示例
# model = AutoModelForCausalLM.from_pretrained("model-name")
# compressed_model = compress_model(model, compression_ratio=0.5)

4.2 量化优化¶

Python

import torch
import torch.quantization

def quantize_model_for_edge(model, dataloader):
    """
    为边缘设备量化模型

    Args:
        model: 要量化的模型
        dataloader: 校准数据加载器
    """
    # 设置模型为评估模式
    model.eval()

    # 配置量化
    model.qconfig = torch.quantization.get_default_qconfig('qnnpack')

    # 准备量化
    model_prepared = torch.quantization.prepare(model)

    # 使用校准数据进行校准
    with torch.no_grad():
        for batch_x, _ in dataloader:
            model_prepared(batch_x)

    # 转换为量化模型
    quantized_model = torch.quantization.convert(model_prepared)

    return quantized_model

# 使用示例
# quantized_model = quantize_model_for_edge(model, dataloader)

4.3 动态批处理¶

Python

import torch
from collections import defaultdict
import time

class DynamicBatchProcessor:
    """
    动态批处理器
    """
    def __init__(self, model, max_batch_size=8, max_wait_time=0.1):
        self.model = model
        self.max_batch_size = max_batch_size
        self.max_wait_time = max_wait_time
        self.requests = []
        self.results = defaultdict(list)  # defaultdict访问不存在的键时返回默认值

    async def add_request(self, request_id, input_data):  # async定义异步函数
        """
        添加请求
        """
        self.requests.append((request_id, input_data))

        # 如果达到最大批次大小，处理请求
        if len(self.requests) >= self.max_batch_size:
            await self._process_batch()  # await等待异步操作完成

    async def _process_batch(self):
        """
        处理批次
        """
        if not self.requests:
            return

        # 准备批次
        batch_inputs = [req[1] for req in self.requests]
        request_ids = [req[0] for req in self.requests]

        # 批量推理
        batch_outputs = self.model(batch_inputs)

        # 分发结果
        for request_id, output in zip(request_ids, batch_outputs):  # zip按位置配对
            self.results[request_id].append(output)

        # 清空请求
        self.requests = []

    async def get_result(self, request_id, timeout=1.0):
        """
        获取结果
        """
        start_time = time.time()

        while time.time() - start_time < timeout:
            if request_id in self.results and self.results[request_id]:
                return self.results[request_id].pop(0)

            # 检查是否需要处理批次
            if len(self.requests) >= self.max_batch_size:
                await self._process_batch()

            await asyncio.sleep(0.01)

        return None

# 使用示例
# processor = DynamicBatchProcessor(model)
# await processor.add_request("req1", input1)
# result = await processor.get_result("req1")

5. 边缘部署实践¶

5.1 完整的边缘部署流程¶

Python

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import onnxruntime as ort
import numpy as np

class EdgeDeploymentPipeline:
    """
    边缘部署流水线
    """
    def __init__(self, model_name):
        self.model_name = model_name
        self.tokenizer = None
        self.session = None

    def prepare_model(self, quantization="int8"):
        """
        准备模型

        Args:
            quantization: 量化类型
        """
        # 加载分词器
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)

        # 加载模型
        if quantization == "int8":
            from transformers import BitsAndBytesConfig
            quantization_config = BitsAndBytesConfig(
                load_in_8bit=True
            )
            model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                quantization_config=quantization_config,
                device_map="auto"
            )
        else:
            model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16,
                device_map="auto"
            )

        return model

    def convert_to_onnx(self, model, output_path="model.onnx"):
        """
        转换为ONNX
        """
        # 准备示例输入
        dummy_input = self.tokenizer("Hello", return_tensors="pt")

        # 导出ONNX
        torch.onnx.export(
            model,
            (dummy_input["input_ids"], dummy_input["attention_mask"]),
            output_path,
            input_names=["input_ids", "attention_mask"],
            output_names=["logits"],
            opset_version=14
        )

        print(f"模型已导出到: {output_path}")

    def load_onnx_model(self, model_path="model.onnx"):
        """
        加载ONNX模型
        """
        self.session = ort.InferenceSession(
            model_path,
            providers=['CPUExecutionProvider']
        )

    def inference(self, prompt, max_length=100):
        """
        推理
        """
        # 编码输入
        inputs = self.tokenizer(prompt, return_tensors="np")
        input_ids = inputs["input_ids"].astype(np.int64)
        attention_mask = inputs["attention_mask"].astype(np.int64)

        # 推理
        input_name = self.session.get_inputs()[0].name
        output_name = self.session.get_outputs()[0].name

        outputs = self.session.run(
            [output_name],
            {input_name: input_ids}
        )

        # 解码输出
        logits = outputs[0]
        predicted_token_id = np.argmax(logits[0, -1, :])
        result = self.tokenizer.decode(predicted_token_id)

        return result

# 使用示例
# pipeline = EdgeDeploymentPipeline("meta-llama/Llama-2-7b-hf")
# model = pipeline.prepare_model(quantization="int8")
# pipeline.convert_to_onnx(model, "model.onnx")
# pipeline.load_onnx_model("model.onnx")
# result = pipeline.inference("请介绍一下人工智能")
# print(result)

6. 练习题¶

基础练习¶

转换模型为ONNX

Python

# TODO: 实现模型转换为ONNX
def convert_to_onnx(model, output_path):
    # 你的代码
    pass

实现简单的边缘推理

Python

# TODO: 实现简单的边缘推理
class EdgeInference:
    def __init__(self, model_path):
        # 你的代码
        pass

    def predict(self, input_data):
        # 你的代码
        pass

进阶练习¶

实现动态批处理

Python

# TODO: 实现动态批处理推理
class DynamicBatchInference:
    def __init__(self, model, max_batch_size):
        # 你的代码
        pass

    async def process(self, requests):
        # 你的代码
        pass

实现边缘部署优化

Python

# TODO: 实现边缘部署优化
class EdgeOptimizer:
    def __init__(self, model):
        # 你的代码
        pass

    def optimize(self):
        # 你的代码
        pass

项目练习¶

创建边缘部署框架
支持多种设备
自动优化模型
监控和日志

7. 最佳实践¶

✅ 推荐做法¶

充分优化模型
量化、剪枝、蒸馏
使用合适的精度
测试优化效果
测试设备兼容性
在目标设备上测试
验证性能和精度
考虑硬件限制
实现离线能力
支持离线推理
缓存常用结果
优雅降级

❌ 避免做法¶

忽视设备限制
不要假设资源充足
考虑内存和计算限制
优化模型大小
忽略能耗
考虑电池续航
优化计算效率
避免过度计算
缺乏测试
在实际设备上测试
测试各种场景
验证稳定性

8. 总结¶

本章介绍了边缘部署的核心技术：

移动端部署: ONNX、TFLite、CoreML
嵌入式设备: 树莓派、Jetson
优化技术: 量化、剪枝、批处理
部署实践: 完整的部署流程

边缘部署需要综合考虑性能、功耗和设备限制。

9. 下一步¶

继续学习06-实战项目，通过完整的项目实践所学知识。