AI API边缘计算实战：在树莓派和IoT设备上跑通大模型推理的完整方案

# 树莓派5上安装Ollama (ARM64)
curl -fsSL https://ollama.com/install.sh | sh

# 验证安装
ollama --version
# ollama version is 1.5.0

# 拉取Qwen2.5 1.5B（自动选择Q4量化版本）
ollama pull qwen2.5:1.5b

# 运行测试
ollama run qwen2.5:1.5b "检测到温度传感器异常：当前温度85°C，阈值70°C。请生成告警摘要。"

# 输出示例：
# 【温度告警】检测到温度异常超标。当前读数85°C，超出安全阈值70°C，
# 偏差幅度达21.4%。建议立即检查散热系统并降低负载。

import requests
import json

def edge_inference(prompt, model="qwen2.5:1.5b"):
    """调用本地Ollama模型做推理"""
    resp = requests.post(
        "http://localhost:11434/v1/chat/completions",
        headers={"Content-Type": "application/json"},
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": "你是IoT设备数据分析助手，用简洁的中文回复。"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 200
        },
        timeout=30
    )
    return resp.json()["choices"][0]["message"]["content"]

# 使用示例
result = edge_inference("传感器A振动值0.8mm/s，正常范围0-0.5mm/s，请判断状态。")
print(result)
# 输出：传感器A振动值0.8mm/s，超出正常范围0-0.5mm/s，
# 判定为异常状态。建议检查设备轴承和安装紧固情况。

// Cloudflare Worker: 边缘AI推理网关
export default {
  async fetch(request, env) {
    const { pathname } = new URL(request.url);

    if (pathname === '/v1/chat' && request.method === 'POST') {
      const { prompt, model } = await request.json();

      const response = await env.AI.run(
        model || '@cf/qwen/qwen2.5-7b-instruct',
        {
          prompt: prompt,
          max_tokens: 256,
          temperature: 0.3
        }
      );

      return new Response(JSON.stringify(response), {
        headers: { 'Content-Type': 'application/json' }
      });
    }

    return new Response('Edge AI Gateway', { status: 404 });
  }
};

# MQTT消息格式设计（JSON）
# 传感器数据上报
{
  "device_id": "sensor-temp-001",
  "timestamp": "2026-06-16T10:30:00.000Z",
  "data_type": "temperature",
  "values": [85.2, 84.8, 86.1, 85.7],
  "unit": "celsius"
}

# AI推理结果
{
  "device_id": "sensor-temp-001",
  "inference_id": "inf-20260616-103000-a3f2",
  "status": "alert",
  "confidence": 0.95,
  "summary": "温度持续升高，过去4个读数均超过70°C阈值，判定为过热风险。",
  "recommendation": "建议降低设备负载并检查冷却系统。"
}

import paho.mqtt.client as mqtt
import json
import requests
from datetime import datetime

# MQTT Broker配置
MQTT_BROKER = "192.168.1.100"
MQTT_PORT = 1883
TOPIC_SENSOR = "factory/sensor/+"
TOPIC_ALERT = "factory/alert"
OLLAMA_URL = "http://localhost:11434/v1/chat/completions"

def analyze_sensor_data(payload):
    """调用本地AI模型分析传感器数据"""
    resp = requests.post(OLLAMA_URL, json={
        "model": "qwen2.5:1.5b",
        "messages": [
            {
                "role": "system",
                "content": "你是工厂IoT数据分析助手。分析传感器数据，判断状态，用JSON格式回复：{"status":"normal/warning/alert","confidence":0-1,"summary":"摘要","recommendation":"建议"}
            },
            {
                "role": "user",
                "content": f"分析以下传感器数据：{json.dumps(payload, ensure_ascii=False)}"
            }
        ],
        "temperature": 0.2,
        "max_tokens": 200
    }, timeout=15)
    return resp.json()["choices"][0]["message"]["content"]

def on_message(client, userdata, msg):
    """MQTT消息回调"""
    try:
        payload = json.loads(msg.payload.decode())
        print(f"[{datetime.now()}] 收到数据: {payload['device_id']}")

        # 调用本地AI分析
        analysis = analyze_sensor_data(payload)

        # 发布分析结果
        result = {
            "device_id": payload["device_id"],
            "inference_id": f"inf-{datetime.now().strftime('%Y%m%d-%H%M%S')}",
            "raw_analysis": analysis,
            "timestamp": datetime.now().isoformat()
        }
        client.publish(TOPIC_ALERT, json.dumps(result, ensure_ascii=False))
        print(f"  -> 分析完成，已发布到 {TOPIC_ALERT}")

    except Exception as e:
        print(f"  -> 错误: {e}")

# 初始化MQTT客户端
client = mqtt.Client()
client.on_message = on_message
client.connect(MQTT_BROKER, MQTT_PORT, 60)
client.subscribe(TOPIC_SENSOR)

print("边缘AI推理网关已启动，等待传感器数据...")
client.loop_forever()

# 查看Jetson设备内存使用
sudo jetson-stats

# 关闭不必要的GUI服务释放内存
sudo systemctl set-default multi-user.target
sudo systemctl disable gdm3

# 增加swap空间
sudo fallocate -l 8G /var/swapfile
sudo mkswap /var/swapfile
sudo swapon /var/swapfile

# Ollama指定量化精度拉取模型
ollama pull qwen2.5:1.5b-q4_K_M
ollama pull qwen2.5:3b-q5_K_M

# 对比测试不同量化版本的输出
for model in qwen2.5:1.5b-q4_K_M qwen2.5:1.5b-q5_K_M; do
    echo "=== $model ==="
    time ollama run $model "计算 2.7 * 3.14 的结果"
done

# Jetson温度监控脚本
import subprocess
import time

def get_jetson_temp():
    """读取Jetson核心温度"""
    result = subprocess.run(
        ["cat", "/sys/devices/virtual/thermal/thermal_zone0/temp"],
        capture_output=True, text=True
    )
    return int(result.stdout.strip()) / 1000  # 转换为摄氏度

while True:
    temp = get_jetson_temp()
    if temp > 75:
        print(f"[警告] 温度过高: {temp}°C，建议降低负载")
    elif temp > 85:
        print(f"[严重] 温度临界: {temp}°C，即将降频！")
    time.sleep(10)

# 简单的OTA更新脚本
import requests
import subprocess
import hashlib
import os

MODEL_REGISTRY = "https://internal.example.com/models"
MODEL_DIR = "/opt/edge-ai/models"
CURRENT_MODEL = "qwen2.5:1.5b"

def check_update():
    """检查模型更新"""
    resp = requests.get(f"{MODEL_REGISTRY}/{CURRENT_MODEL}/latest")
    latest = resp.json()
    return latest["version"], latest["checksum"], latest["download_url"]

def rollback():
    """回滚到上一个版本"""
    backup = os.path.join(MODEL_DIR, "backup")
    if os.path.exists(backup):
        subprocess.run(["ollama", "serve", backup])
        print("已回滚到备份版本")
    else:
        print("无可用备份")

# 定时检查更新（crontab每小时执行）
version, checksum, url = check_update()
print(f"最新版本: {version}")