Modbus + 边缘AI Agent实战：用本地网关实现工业设备异常检测（不上云方案）

> 本文是「工业AI实战手册」系列第1篇。这个系列不讲概念，不讲PPT，只讲怎么用边缘AI网关在真实工厂里干活。

---

先说问题

工厂里常见的场景：一台空压机、一个冷却塔、一套输送带，运行了十几年。PLC上只有几个红灯绿灯，什么数据都没有。

你想要的是：

• 设备什么时候会故障 → 提前预警
• 能耗是不是异常 → 自动识别
• 工人有没有违规操作 → 自动告警

但你面对的现实：

• ❌ 没有云平台、不想上云（数据安全、网络不稳定、每年还要交平台费）
• ❌ 设备太老，不支持智能协议
• ❌ 找厂家升级报价几万起步

边缘AI网关 + Modbus采集 + 本地AI推理 就是这题的解。

不需要云，不需要改造设备，插上线就能跑。

---

方案架构（一张图看懂）

[老旧设备] ←RS485/232→ [Modbus采集模块] ←→ [边缘AI网关] 
                                               │
                                               ├─ 本地运行AI模型（7B以内）
                                               ├─ 实时异常检测
                                               ├─ 本地Web管理界面
                                               └─ 告警（声光/短信/微信）

硬件部分：边缘AI网关（我们用的S200，6TOPS算力，¥5,999）

软件部分：Python采集引擎 + 本地部署的AI推理模型

---

第一步：Modbus数据采集（Python实现）

不管你设备是Modbus RTU还是TCP，以下的Python引擎都能跑。

import minimalmodbus
import time
import json
from datetime import datetime

# === 配置 ===
SLAVE_ID = 1
BAUDRATE = 9600
PORT = '/dev/ttyUSB0'  # 根据实际串口调整

# === 连接设备 ===
instrument = minimalmodbus.Instrument(PORT, SLAVE_ID)
instrument.serial.baudrate = BAUDRATE
instrument.serial.timeout = 1

# === 采集函数 ===
def read_registers(address, count=2):
    """读取Modbus寄存器（浮点数格式）"""
    try:
        values = instrument.read_registers(address, count)
        # 将两个16位寄存器合并为32位浮点数
        import struct
        packed = struct.pack('>HH', values[0], values[1])
        return struct.unpack('>f', packed)[0]
    except Exception as e:
        print(f"[{datetime.now()}] 读取失败: {e}")
        return None

# === 主循环 ===
if __name__ == '__main__':
    print("🔌 Modbus采集引擎启动...")
    while True:
        # 读取常用参数（地址需根据设备手册调整）
        temperature = read_registers(0x100)
        pressure = read_registers(0x102)
        vibration = read_registers(0x104)
        
        data = {
            'timestamp': datetime.now().isoformat(),
            'temperature': temperature,
            'pressure': pressure,
            'vibration': vibration
        }
        
        # 输出为JSON（后续会被AI模型消费）
        print(json.dumps(data))
        time.sleep(5)  # 每5秒采集一次

> 要点：采集频率不要太高。对于设备预测性维护场景，5-30秒一次完全够用，频率越高数据噪声越大。

---

第二步：在边缘网关上跑AI推理

采集到的数据怎么变成"智能判断"？用本地AI模型。

我们选的是 Qwen2.5-7B-Instruct量化版（4bit GGUF），在S200上推理速度约 8-12 tokens/s，对设备诊断完全够用。

from llama_cpp import Llama
import json

# === 加载本地模型 ===
MODEL_PATH = '/models/qwen2.5-7b-instruct-q4_k_m.gguf'
llm = Llama(
    model_path=MODEL_PATH,
    n_ctx=4096,
    n_threads=4
)

# === 设备诊断函数 ===
SYSTEM_PROMPT = """你是工厂设备诊断专家。根据采集到的传感器数据，判断设备是否异常。

已知一台空压机的正常运行参数范围：
- 温度：30-55°C
- 压力：0.6-0.8 MPa
- 振动：< 5 mm/s

请判断当前状态并给出建议。"""

def diagnose_equipment(temp, pressure, vibration):
    """基于传感器数据的AI诊断"""
    user_input = f"""当前传感器数据：
- 温度：{temp}°C
- 压力：{pressure} MPa
- 振动：{vibration} mm/s

请回答：
1. 设备状态是否正常？
2. 如果异常，可能的原因是什么？
3. 建议的操作是什么？"""
    
    response = llm.create_chat_completion(
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_input}
        ],
        max_tokens=256,
        temperature=0.1
    )
    
    return response['choices'][0]['message']['content']

# === 实时诊断循环 ===
if __name__ == '__main__':
    print("🤖 边缘AI诊断引擎启动...")
    # 模拟采集数据
    sample_data = {"temperature": 68.5, "pressure": 0.82, "vibration": 7.2}
    
    result = diagnose_equipment(
        sample_data['temperature'],
        sample_data['pressure'],
        sample_data['vibration']
    )
    print(f"诊断结果：\n{result}")

> 注意：7B模型在S200上跑实时推理，建议用异步队列方案——采集线程只管收集数据，推理线程空闲了再处理，互不阻塞。

---

第三步：异常告警（本地推送）

检测到异常怎么办？不需要SMS网关，用微信机器人或者本地声光告警就够。

import requests

def send_alert(message):
    """通过企业微信机器人发送告警"""
    WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"## ⚠️ 设备异常告警\n{message}\n> 时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        }
    }
    try:
        r = requests.post(WEBHOOK_URL, json=payload, timeout=5)
        return r.ok
    except:
        return False

---

完整效果

部署在某工厂空压机上的实测数据：

| 指标 | 改造前 | 改造后 |

|------|--------|--------|

| 故障发现时间 | 工人报修后（2-8小时） | AI自动检测（<30秒） |

| 非计划停机 | 每月平均3.2次 | 每月0.4次 |

| 能耗优化 | 无 | 识别出2台异常高能耗，年省电费4.7万 |

| 投入成本 | 无 | 网关¥5,999 + 部署1天 |

---

FAQ

Q：需要改设备本身的PLC吗？

A：不需要。Modbus采集是无侵入的，串口接上就能读，不改变原有设备控制逻辑。

Q：数据安全怎么保证？

A：所有数据在边缘网关本地处理，不上云。告警消息也只走企业微信内部通道。敏感数据不出厂区。

Q：支持哪些协议？

A：Modbus RTU、Modbus TCP、OPC UA、S7（西门子）均支持。本案例用的是Modbus RTU，最通用的工业协议。

Q：7B模型在边缘端跑得动吗？

A：S200（6TOPS）跑Qwen2.5-7B量化版（q4_k_m）实测单次推理2-3秒，对预测性维护场景完全够用。如果只需要文本分类或简单判断，甚至可以跑3B模型，延迟<1秒。

Q：没有编程经验能用吗？

A：上面的Python代码是开箱即用的。但如果要做定制化——比如接不同的设备、改告警逻辑——需要一些基础开发能力。也可以直接走技术合伙人模式，我们来搞定。

---

> 下期预告：「工业AI实战手册」第2篇：从Modbus到OPC UA——异构工业协议统一采集方案

---

本文由实际项目沉淀，首发于 Agent评测。转载需保留出处。