边侧部署垂直模型：架构设计与实现指南

flowchart TD
    subgraph 云端
        A[模型训练与优化] --> B[模型仓库与版本管理]
        B --> C[模型分发与编排中心]
    end

    subgraph 边缘层
        C --> D[边缘节点/网关]
        D --> E[推理引擎]
        E --> F[本地推理服务]
        F --> G[业务应用]
    end

    subgraph 终端层
        H[传感器/摄像头/用户终端] --> G
    end

    G --> I[结果上报与反馈]
    I --> A

# 1. 安装 ONNX Runtime（ARM64 版本）
wget https://github.com/microsoft/onnxruntime/releases/download/v1.17.0/onnxruntime-linux-aarch64-1.17.0.tgz
tar -xzf onnxruntime-linux-aarch64-1.17.0.tgz
sudo cp -r onnxruntime-linux-aarch64-1.17.0/lib/* /usr/local/lib/
sudo ldconfig

# 2. 安装 Python 依赖（若使用 Python 推理）
pip install onnxruntime numpy opencv-python pillow

import torch
import torch.onnx
import onnx
from onnxruntime.quantization import quantize_dynamic, QuantType

# 假设 model 是训练好的 PyTorch 模型
model = torch.load('vertical_model.pth')
model.eval()

# 转换为 ONNX
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(model, dummy_input, "vertical_model.onnx",
                  input_names=['input'], output_names=['output'],
                  dynamic_axes={'input': {0: 'batch_size'}, 'output': {0: 'batch_size'}})

# 动态量化（INT8）
model_fp32 = 'vertical_model.onnx'
model_int8 = 'vertical_model_int8.onnx'
quantized_model = quantize_dynamic(model_fp32, model_int8, weight_type=QuantType.QInt8)
print(f"量化完成，模型大小从 {os.path.getsize(model_fp32)/1024:.1f} KB 减小到 {os.path.getsize(model_int8)/1024:.1f} KB")

# inference_server.py
import numpy as np
import onnxruntime as ort
from flask import Flask, request, jsonify
from PIL import Image
import io

app = Flask(__name__)

class EdgeInferenceEngine:
    def __init__(self, model_path):
        self.session = ort.InferenceSession(model_path)
        self.input_name = self.session.get_inputs()[0].name
        self.output_name = self.session.get_outputs()[0].name

    def preprocess(self, image_bytes):
        image = Image.open(io.BytesIO(image_bytes)).convert('RGB')
        image = image.resize((224, 224))
        input_array = np.array(image).astype(np.float32) / 255.0
        input_array = np.transpose(input_array, (2, 0, 1))  # HWC -> CHW
        input_array = np.expand_dims(input_array, axis=0)   # 添加 batch 维度
        return input_array

    def infer(self, input_array):
        outputs = self.session.run([self.output_name], {self.input_name: input_array})
        return outputs[0]

    def postprocess(self, output_array):
        predicted_class = int(np.argmax(output_array[0]))
        confidence = float(np.max(output_array[0]))
        return {"class_id": predicted_class, "confidence": confidence}

engine = EdgeInferenceEngine('vertical_model_int8.onnx')

@app.route('/predict', methods=['POST'])
def predict():
    if 'image' not in request.files:
        return jsonify({"error": "No image provided"}), 400
    image_file = request.files['image'].read()
    try:
        input_tensor = engine.preprocess(image_file)
        output = engine.infer(input_tensor)
        result = engine.postprocess(output)
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    # 在生产环境中应使用 Gunicorn 或 uWSGI
    app.run(host='0.0.0.0', port=8080, threaded=True)

# 启动推理服务
python inference_server.py

# 使用 curl 测试
curl -X POST -F "image=@test_image.jpg" http://localhost:8080/predict
# 预期输出: {"class_id": 5, "confidence": 0.987}

@app.route('/health')
def health():
    return jsonify({"status": "ok", "model_version": "v1.2.3"})

@app.route('/metrics')
def metrics():
    # 返回 Prometheus 格式的指标
    return f"""
# HELP inference_latency_ms 推理延迟（毫秒）
# TYPE inference_latency_ms gauge
inference_latency_ms {current_latency}
# HELP inference_count 推理请求总数
# TYPE inference_count counter
inference_count {total_requests}
"""