一、背景：为什么要在边缘网关上加 AI？

工业现场的边缘网关通常只干三件事：采集 → 处理 → 上传。Modbus 轮询设备，MQTT 推到云端，完事。

但最近 AI Agent（Claude、GPT、各类大模型）越来越强，客户开始问一个问题：

能让 AI 直接看网关数据，甚至帮我操作网关吗？

传统方案是让 AI 对接云端平台，走一层转发。但延迟高、依赖网络、且云平台本身不一定开放了 AI 友好的接口。

于是我们尝试在网关本地实现了一套 AI 直接接入能力——让 AI Agent 通过标准协议（MCP）直接读写网关数据、查看告警、甚至执行操作。

二、架构概览：一条数据从现场到 AI 的完整路径

先把整体链路画出来：

网关内部的核心数据流：

现场设备 → Modbus 轮询 → 最新值缓存 → MQTT 周期推送
                               ↕
                          HTTP API / MCP 接口
                               ↕
                            AI Agent

关键设计：同一份最新值数据，同时服务于北向 MQTT 推送和 AI 接口查询，没有冗余通道。

三、南向数据采集：Modbus 的"体检式"轮询

Modbus 是工业领域最基础的协议之一。网关作为 Modbus Master，轮询现场设备。

采集配置采用三层模型：

数据源（Source）     →  例如：Modbus TCP 连接 192.168.1.100:502
  └─ 设备实例（Instance）→  例如：1号电表（关联物模型 meter）
       └─ 测点（Point） →  例如：电压、电流、功率（各1000ms采集一次）

技术选型上直接用 libmodbus，C++ 实现，轻量成熟，在嵌入式平台跑起来毫无压力。

值得分享的经验

关于轮询频率：我们让每个测点独立配置采集周期（100ms ~ 5000ms）。一开始觉得没必要，后来发现高压设备的状态量需要 100ms 高频，而温度这类缓变量 5s 一次就够了。按需采集比一刀切节省大量总线带宽。

关于失败处理：Modbus 从机断连是常事。我们的策略是：重试 N 次后标记 quality=Bad，但保留上次的好值不覆盖。这样即使设备离线，AI 仍然能读到"最近一次正常值"，而不是一个空值引发误判。

四、数据缓存：一个 map 撑起整个系统

系统只有一个内存数据结构是所有模块的中心——Latest Value Map。

每个测点最新值包含：

• 值（统一存为字符串）
• 质量（Good / Bad）
• 时间戳
• 版本号（每次更新递增，全局唯一）

这个版本号是 AI 集成中的一个关键设计。AI 可以通过 ?since=12345 查询自某版本以来的增量变更，避免每次都全量拉取。对于几十上百个测点的网关，全量才几 KB，但在 AI 对话场景中，减少 Token 消耗效果显著。

五、北向数据推送：MQTT 的周期批量模式

MQTT 作为北向通道，把数据推到云平台。

当前实现采用 QoS0 + 周期批量打包方式：

• 每个 MQTT 源独立配置 Broker、主题、推送间隔
• 到点时，将绑定的所有测点最新值打包成一个 JSON 发出去
• 支持 username/password 认证

为什么不 QoS1 或事件驱动？ 嵌入式资源有限，QoS0 最简单且可靠；事件驱动虽然实时性好，但会导致短时间内大量小包推送，在 NB-IoT/4G 网络下反而不可靠。周期性批量推送是"稳"字当头的选择。

六、AI 接入：让 AI 直接"看懂"网关

这是这篇文章想重点分享的部分——我们提供了两种 AI 接入路径。

路径一：REST API

标准的 HTTP JSON 接口，AI 应用可以直接 curl 调用：

GET  /api/ai/status           → 网关运行状态
GET  /api/ai/points/latest    → 测点最新值
GET  /api/ai/alarms/active    → 活跃告警
POST /api/ai/system/reboot    → 重启网关（高危操作）
...

共 14 个端点，覆盖数据查询、告警管理、系统操作三大类。

路径二：MCP 协议（重点）

MCP（Model Context Protocol）是 Anthropic 提出的开放协议，让 AI 模型可以像调用本地函数一样调用外部系统工具。

通俗理解：MCP 就是"AI 世界的 USB 协议"——即插即用，让 AI 发现并使用外部能力。

网关实现了 MCP over HTTP，AI Agent 的调用流程只有两步：

第一步：工具发现——AI 问网关：你能干什么？

POST /api/ai/mcp
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list"
}

网关返回 10 个 MCP 工具列表：

工具名	说明
gateway.management.summary	读取系统概览
gateway.latest.list	读取所有测点值
gateway.latest.changes	增量变更查询
gateway.channel.status	南向通道状态
gateway.mqtt.status	MQTT 连接状态
gateway.alarms.active	活跃告警
gateway.alarms.events	告警历史（分页）
gateway.runtime.logs	实时日志
gateway.system.reboot	⚠️ 重启网关
gateway.system.factory_reset	⚠️ 恢复出厂

第二步：工具调用——AI 说：帮我看看当前数据

POST /api/ai/mcp
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "gateway.latest.list",
    "arguments": {}
  }
}

网关返回所有测点的最新值 JSON。
但最大的区别就在这里，他不是单纯的返回大量的数据，而是自行对数据做了汇总整理。像管家一下帮你完成数据的分析：

安全设计

让 AI 直接操作设备，安全是第一位的。我们做了三件事：

1. API Key + 权限作用域

每个 API Key 绑定一组权限（Scope），以 JSON 数组形式配置：

{
  "scopes": ["point.read", "alarm.read", "gateway.read"]
}

Scope 分两类：

• 只读 Scope：查询数据、查看状态
• 高危 Scope：重启、恢复出厂、确认告警

API Key 还可以设置过期时间，到期自动失效。

2. 高危操作二次确认

reboot 和 factory_reset 这两个工具，即使有了匹配的 Scope，也必须携带 "confirm": true 参数才会被执行。相当于软件上多了一层保险。

3. 全链路审计

每一次 AI 调用都会被记入审计日志（到文件），记录谁、什么时候、调了什么工具、结果如何。即使一切正常，也能追溯。

速率限制

每个 API Key 每分钟有调用次数上限，防止 AI Agent 疯狂轮询把网关打满。

七、告警引擎：不只是简单的阈值判断

告警规则支持四种基本类型：模拟量高限/低限、布尔量 True/False。

几个在实际项目中踩过的坑：

死区（Deadband）——没有死区时，电压在 49.9 和 50.1 之间抖动，告警一秒内反复触发消除十几次。加上 ±2% 的死区后彻底安静了。

确认延迟（Confirm Delay）——瞬时毛刺不该触发告警，持续异常 N 秒才确认。我们把这个参数开放给用户配置，不同场景按需调整。

恢复延迟（Recover Delay）——和确认延迟对称，值回归正常后维持一段时间再消除告警，避免"闪恢复"。

八、一些值得分享的经验教训

1. 单线程事件循环足够了

最开始考虑过多线程：一个线程采集、一个线程 HTTP、一个线程 MQTT……后来发现 RK3506 两颗核心、资源有限，锁竞争带来的复杂度远大于收益。

最终采用 poll(2) 单线程事件循环，把所有定时任务统一调度。实测效果理想：同时管理 500+ 测点采集、MQTT 推送、HTTP 请求毫无压力。

2. JSON 解析别上重型库

一开始用了 nlohmann/json，交叉编译后体积增加了 200KB。索性手写了一个轻量递归下降解析器，只支持需要的特性（对象、数组、字符串、数字），体积只有原来的十分之一。

3. SQLite 要关线程安全

嵌入式场景、单线程应用，编译时加 SQLITE_THREADSAFE=0，省掉不必要的锁开销。同时关掉不需要的扩展加载功能，体积进一步缩小。

4. MQTT 库也是自裁的

没用完整版 paho-mqtt-c（太重），只提取了 paho-mqtt-packet（序列化层，仅 4 个 .c 文件），TCP socket 自己管理。这样编译出来的东西小到可以塞进任何固件。

九、写在最后

边缘网关 + AI 是一个很有意思的方向。网关不再只是"透传数据的盒子"，它可以是 AI 的"眼睛和手"——帮 AI 看现场数据，帮 AI 操作设备。

我们走的路线是开放的、标准的：MCP 协议让任何兼容的 AI Agent 都能接入，不需要私有 SDK；Scope 权限体系让安全可控；全链路审计让一切可追溯。

如果你也在做类似的方向，有几个问题值得思考：

• 你的网关数据用哪种方式开放给 AI？ REST / MCP / GraphQL / gRPC？
• 高危操作如何授权？ 二次确认、双人授权、还是全部只读？
• AI 调用的审计和限流怎么做？ 别让 AI 把现场设备"问死"了。

欢迎在评论区交流讨论！

---

参考资源：

• MCP 协议规范：https://modelcontextprotocol.io/
• libmodbus：https://libmodbus.org/
• paho.mqtt：https://www.eclipse.org/paho/