LLM接入工业Modbus设备：寄存器映射与Function Calling的架构实践

如果你最近在折腾 LLM Agent 接入工业设备大概率会有同样的感受让大模型去直接解码 Modbus 寄存器就像让一个文科生去逐字节翻译二进制协议报文。不是大模型不聪明而是这件事从一开始就不该交给它。Modbus 寄存器本质上是 16 位无符号整数的数组它本身不携带数据类型、单位、小数点位置、字节顺序这些信息。设备手册里写一句“供水温度地址 02 个寄存器Float32大端缩放 0.1单位 ℃”到了程序里就是一条解码规则。LLM 如果要直接处理原始寄存器值就得在对话上下文里做位运算、端序判断、浮点解析而这些恰恰是文本模型最不擅长、最容易产生幻觉的环节。错一个字节温度从 25℃ 变成 2500℃ 都不奇怪这在工业现场是不可接受的。所以真正的解法不是“让 LLM 更努力地学 Modbus”而是在架构上让 LLM 永远不需要解码寄存器。把确定性工作交给确定性代码把语义理解和决策交给 LLM各自做最擅长的事。这篇文章会用一个可运行的示例把这条链路完整串起来。1. 为什么 LLM 一处理 Modbus 寄存器就翻车先看一个真实场景现场有一台温控机组PLC 通过 Modbus TCP 把运行数据送到中控室你想做一个“设备值班助手”让值班人员用自然语言查询设备状态。传统做法是写死一组查询接口但设备点位多、变更频繁于是你想到用 LLM 来顶替这部分工作。于是你把一段原始报文直接丢给大模型原始寄存器值[100, 0, 80, 0, 1, 0, 0, 0]问题立刻出现。这 8 个寄存器到底是 4 个 Float32还是 8 个 uint16还是 2 个 Float32 加 4 个 uint16单位是什么缩放系数是多少寄存器是低字节在前还是高字节在前这些信息在原始数值里完全看不出来。大模型只能猜。它可能根据上下文推断出“这看起来像是温度数据”但那是概率判断不是确定性解析。更麻烦的是LLM 在计算 0x0064即 100转成浮点再乘 0.1 的时候很容易算错而且它自己不觉得错会非常自信地输出一个错误答案。这不是“换个更大参数的模型”能解决的问题。事实上参数越大模型越倾向于把不完整的知识“补全”成看起来合理的答案这在数据解码场景里更加危险。相比之下一小段 Python 代码用struct.unpack解析寄存器永远不会把端序搞错——前提是有人把配置写对。所以这里要形成一个明确判断寄存器解码是确定性的工程问题适合用代码解决设备状态分析和语言交互是非确定性语义问题适合用 LLM 解决。把前者交给后者是典型的架构错位。2. 问题的本质确定性解码不该交给非确定性模型理解这个问题的关键是分清“解码”和“理解”的边界。解码Decoding指的是从字节序列还原出物理量的过程。它要求结果必须完全可重现同一条报文无论谁来解答案都必须一样。这背后是 Modbus 协议规范、设备制造商的数据映射表、IEEE 754 浮点格式这些确定性的规则。理解Understanding指的是根据多个可信的数据点结合业务上下文判断设备当前处于什么状态、应该采取什么措施。比如“供水温度和回水温度温差过大可能是板式换热器结垢”这是语义推理允许模型有不同的表达方式但结论要有依据。LLM 的问题在于它把解码和理解的步骤混在一起了。你让它看原始寄存器它既要算字节序又要判断语义任何一步出错都会污染后面的所有判断。而且多轮对话中LLM 可能会在某一轮把寄存器地址记错下一轮又顺着错误继续推理形成“自信地胡说八道”。对比一下让 LLM 直接读 pcap 抓包文件去分析 HTTP 协议问题和让它看 Wireshark 已经把报文解析好的结果再分析两者的准确率天差地别。Modbus 也是同样的道理。Wireshark 已经替我们完成了协议解析LLM 只需要阅读解析结果即可。因此在设计 LLM 接入工业协议时第一原则不是“教模型懂协议”而是“让模型接触不到协议的原始形态”。LLM 应该面对的是这样的数据{ point: 供水温度, value: 25.3, unit: ℃, timestamp: 2025-01-18 10:30:00 }而不是这样的数据{ register_address: 0, register_value: [100, 0] }前一种数据LLM 会理解后一种数据LLM 只会猜。3. 方案设计让 LLM 永远不需要解码寄存器既然结论已经清楚剩下的问题就是如何设计一条让 LLM 只接触语义数据的链路。整体架构分为三个层次第一层是设备接入层。使用 pymodbus、libmodbus 这类成熟协议栈负责与 PLC、传感器等设备建立连接读写保持寄存器或输入寄存器。这一层的输入是 IP、端口、从站号、寄存器地址输出是原始寄存器数组。第二层是语义解析层。这里有一个“寄存器映射表”它是整个方案的核心资产。映射表描述每个设备点位的寄存器地址、长度、数据类型、字节顺序、缩放系数和单位。解码函数读入原始寄存器数组和一条映射配置输出一个具有业务含义的 JSON 对象。第三层是 LLM 决策层。LLM 不再直接请求 Modbus 数据而是通过 Function Calling 或 MCPModel Context Protocol工具调用第二层暴露出来的查询函数。LLM 只需要理解工具描述比如“读取供水温度返回摄氏度和时间戳”至于这个函数内部是走 Modbus TCP 还是解析 Float32LLM 完全不需要知道。这个设计最大的变化是责任边界被划清楚了设备接入层管连接和通信保证报文不丢、不重、不错语义解析层管数据还原保证字节到物理量的映射准确LLM 决策层管语义理解和交互保证回答有依据、可追踪。还有一个容易被忽视的好处可审计性。当 LLM 说“供水温度为 25.3℃”时我们可以把语义解析层输出的 JSON 日志拿出来对照。如果答案是错的能立刻定位是解码层的问题还是 LLM 理解层的问题而不是让两边互相甩锅。这在工业场景里极为重要。4. 环境准备与前置条件本文的示例基于 Python 实现主要依赖两个核心库pymodbus负责 Modbus 通信openaiSDK 负责调用兼容 OpenAI 接口的大模型服务可以是云端模型也可以是本地部署的模型服务。版本请以实际项目为准本文重点演示通用思路。建议使用 Python 3.9 及以上版本创建独立虚拟环境python -m venv venv source venv/bin/activate pip install pymodbus openai为了让示例可以在没有真实 PLC 的情况下跑通我们需要一个 Modbus 从站模拟器。这里用 pymodbus 自带的服务器功能在本地启动一个模拟从站预置一些寄存器数据。注意pymodbus的 server API 在不同版本中有差异以下代码基于常见写法如果报错请先查看当前版本的官方示例。# 文件路径modbus_simulator.py from pymodbus.server import StartTcpServer from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext def run_simulator(): # 保持寄存器0-1 供水温度 Float32 big-endian2-3 回水温度 Float32 big-endian # 4 水泵状态 uint16bit0运行bit1故障 # 5 室外温度 uint16缩放 0.1 datablock ModbusSequentialDataBlock(0x0000, [ 0x419A, 0x999A, # 25.3℃ 0x4190, 0xCCCD, # 18.1℃ 0x0001, # 水泵运行 0x00FE # 25.4℃ ]) slave_context ModbusSlaveContext(didatablock, codatablock, hrdatablock, irdatablock) server_context ModbusServerContext(slavesslave_context, singleTrue) StartTcpServer(contextserver_context, address(127.0.0.1, 5020)) if __name__ __main__: run_simulator()这段代码在本地 5020 端口启动了一个 Modbus TCP 从站。注意 0x419A 和 0x999A 是 IEEE 754 浮点数 25.3 按大端切分后的两个 16 位寄存器值。如果版本 API 不兼容换成其他 Modbus Slave 模拟工具也可以不影响后面的核心逻辑。启动模拟器python modbus_simulator.py终端停留在阻塞状态即说明从站运行正常。5. 完整实现寄存器映射表、解码工具与 Function Calling现在进入核心代码部分。整个实现分三个文件寄存器映射表 JSON、解码工具模块、LLM Agent 主程序。这样拆分的好处是接线人员可以只维护映射表不会碰 Python 代码。5.1 寄存器映射表先用 JSON 描述设备的点位配置。这是整个方案的“知识底座”也是让 LLM 从原始字节中解脱出来的关键。{ device: 温控机组, slave_id: 1, points: [ { name: 供水温度, address: 0, count: 2, data_type: float32, word_order: big, scale: 1.0, unit: ℃, description: 一次侧供水温度 }, { name: 回水温度, address: 2, count: 2, data_type: float32, word_order: big, scale: 1.0, unit: ℃, description: 一次侧回水温度 }, { name: 水泵状态, address: 4, count: 1, data_type: uint16, bit_mask: 3, unit: , description: bit0运行, bit1故障 }, { name: 室外温度, address: 5, count: 1, data_type: uint16, scale: 0.1, unit: ℃, description: 室外环境温度 } ] }这个表解决的是“寄存器到底代表什么”的问题。address是 Modbus 寄存器起始地址count是寄存器个数data_type说明数据类型word_order表示 32 位数据跨寄存器时的排列方式scale是工程缩放系数unit是显示单位。实际项目中这张表直接来源于设备手册或点表是现场工程师的日常工作产物。5.2 解码工具模块解码工具负责连接 Modbus 从站根据映射表读取寄存器并解析成语义 JSON。这里有一点容易搞错Modbus 标准规定寄存器内是高位字节在前但 32 位浮点跨两个寄存器时有的设备是“低字在前”有的是“高字在前”这就是word_order存在的意义。# 文件路径modbus_tools.py import json import struct from datetime import datetime from pymodbus.client import ModbusTcpClient def load_registry(pathmodbus_registry.json): with open(path, r, encodingutf-8) as f: return json.load(f) def decode_point(raw_regs, cfg): data_type cfg.get(data_type, uint16) scale cfg.get(scale, 1.0) if data_type float32: regs list(raw_regs) if cfg.get(word_order) little: regs list(reversed(regs)) # Modbus 寄存器内默认大端字节序 byte_data b.join(r.to_bytes(2, byteorderbig) for r in regs) value struct.unpack(f, byte_data)[0] * scale return round(value, 4) if data_type uint16: value raw_regs[0] * scale bit_mask cfg.get(bit_mask) if bit_mask is not None: value raw_regs[0] bit_mask return value if data_type int16: value struct.unpack(h, raw_regs[0].to_bytes(2, byteorderbig))[0] * scale return value raise ValueError(f不支持的数据类型: {data_type}) def read_point(host, port, slave_id, point_name, registryNone): registry registry or load_registry() point_cfg None for p in registry[points]: if p[name] point_name: point_cfg p break if point_cfg is None: return {error: f未找到点位: {point_name}, available: [p[name] for p in registry[points]]} client ModbusTcpClient(host, portport) if not client.connect(): return {error: Modbus 连接失败} try: rr client.read_holding_registers( addresspoint_cfg[address], countpoint_cfg[count], slaveslave_id ) if rr.isError(): return {error: f读取寄存器失败: {rr}} value decode_point(rr.registers, point_cfg) return { point: point_cfg[name], value: value, unit: point_cfg.get(unit, ), description: point_cfg.get(description, ), timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } finally: client.close()这段代码有几个值得注意的地方。第一decode_point是纯函数输入原始寄存器和配置输出数值不依赖任何全局状态容易写单元测试。第二read_point做了错误返回任何失败都会返回一个 JSON 结构的 error而不是抛出一个让 LLM 摸不着头脑的异常。第三finally里保证关闭客户端连接避免长期运行时的句柄泄漏。模拟器里第一个点“供水温度”的寄存器是 0x419A 和 0x999A按大端浮点解析后正好是 25.3和模拟器预置的数据一致。5.3 暴露给 LLM 的 Function Calling函数已经能读取数据了接下来要让 LLM 知道什么时候该调用它。OpenAI 兼容接口的 tools 定义如下[ { type: function, function: { name: read_modbus_point, description: 读取指定 Modbus 数据点的实时值返回工程单位值和采集时间, parameters: { type: object, properties: { point_name: { type: string, enum: [供水温度, 回水温度, 水泵状态, 室外温度], description: 设备数据点名称 } }, required: [point_name] } } } ]注意enum里的点位名是给 LLM 看的“面板”它不需要知道地址 0 代表什么。只要用户问“供水温度怎么样”模型就会填入point_name: 供水温度来触发工具调用。这样即使后续点位增多只需要更新注册表和 tools 定义不需要改模型逻辑。5.4 LLM Agent 主循环下面是完整的 Agent 示例。模型部分以 OpenAI SDK 为例base_url可以指向云端服务也可以指向本地兼容 OpenAI 接口的模型服务。# 文件路径agent_llm.py import json import os from openai import OpenAI from modbus_tools import load_registry, read_point MODEL os.getenv(MODEL_NAME, gpt-4o-mini) BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) API_KEY os.getenv(OPENAI_API_KEY, ) MODBUS_HOST os.getenv(MODBUS_HOST, 127.0.0.1) MODBUS_PORT int(os.getenv(MODBUS_PORT, 5020)) SLAVE_ID 1 TOOLS [ { type: function, function: { name: read_modbus_point, description: 读取指定 Modbus 数据点的实时值返回工程单位值和采集时间, parameters: { type: object, properties: { point_name: { type: string, enum: [供水温度, 回水温度, 水泵状态, 室外温度], description: 设备数据点名称 } }, required: [point_name] } } } ] def call_tool(name, arguments, registry): if name read_modbus_point: return read_point(MODBUS_HOST, MODBUS_PORT, SLAVE_ID, arguments[point_name], registry) return {error: f未知工具: {name}} def main(): registry load_registry() client OpenAI(base_urlBASE_URL, api_keyAPI_KEY) messages [ {role: system, content: 你是工业设备值班助手回答必须基于工具返回的实时数据不要编造数值。}, {role: user, content: 当前供水温度是多少水泵状态正常吗} ] for _ in range(5): resp client.chat.completions.create( modelMODEL, messagesmessages, toolsTOOLS, ) msg resp.choices[0].message if msg.tool_calls: messages.append(msg) for tc in msg.tool_calls: args json.loads(tc.function.arguments) result call_tool(tc.function.name, args, registry) print(f[工具调用] {tc.function.name}({args}) - {result}) messages.append({ role: tool, tool_call_id: tc.id, content: json.dumps(result, ensure_asciiFalse) }) else: print([助手回复], msg.content) break else: print(达到最大工具调用轮次请检查工具调用逻辑。) if __name__ __main__: main()这个循环实现了标准的 Function Calling 流程把用户问题和系统提示发送给模型模型判断是否要调用工具如果要调用返回tool_calls程序执行工具函数拿到 JSON 结果把工具结果作为role: tool的消息追加到对话里再次请求模型模型基于真实数据生成最终回答。for _ in range(5)是调用轮次上限防止模型在工具调用中死循环。实际生产中可以做成while True加超时控制但轮次上限更直观。5.5 一个更复杂的多点位查询值班助手经常要一次查多个点位直接调一次工具拿到的只是单点数据。一个改进方案是让工具内部支持“批量读取”def read_points(host, port, slave_id, point_names, registry): result {} for name in point_names: result[name] read_point(host, port, slave_id, name, registry) return result对应的 tools 定义可以把参数改成数组类型。这样 LLM 在回答“把四个点位都报一遍”时只需要一次工具调用既节省 token 又能保证数据的一致性。在 Modbus 现场多个点位分两三次请求比逐个请求更符合工程习惯。6. 运行与效果验证先确认模拟器已经启动然后运行 Agentpython agent_llm.py如果模型服务配置正确会在终端看到两类输出。第一类是工具调用的中间日志[工具调用] read_modbus_point({point_name: 供水温度}) - {point: 供水温度, value: 25.3, unit: ℃, description: 一次侧供水温度, timestamp: 2025-01-18 10:30:00} [工具调用] read_modbus_point({point_name: 水泵状态}) - {point: 水泵状态, value: 1, unit: , description: bit0运行, bit1故障, timestamp: 2025-01-18 10:30:01}第二类是模型基于工具结果的最终回答[助手回复] 当前供水温度为 25.3℃水泵状态值为 1表示 bit0 为 1即水泵处于运行状态未发现故障位。验证是否成功重点看两件事。第一模型答案中的所有数值必须能在工具日志里找到对应来源不能出现日志里没有的数值第二如果工具返回 error模型不应该编造一个数值来填补而是应该如实说明读取失败。如果模型在没有任何数据的情况下给出“温度正常”说明 system prompt 里的约束还不够强需要补充“没有数据时只回答无法获取”。为了更直观地看出方案的价值可以做一个对照实验把原始寄存器数组[4190, 0, 0, 0]直接发给同一个模型问它“这是什么”。模型大概率会给出模棱两可的猜测甚至自信地说“这是 4190.0 伏特”。这个对照不需要写代码直接到任意模型聊天页面就能测试。结果会让你理解为什么要做语义解析层。7. 常见问题与排查问题现象可能原因排查方式解决方案Modbus TCP 连接失败从站未启动、IP/端口错误、防火墙拦截先用nc -zv 127.0.0.1 5020测试端口连通性确认从站地址和端口检查防火墙规则读取寄存器返回异常状态从站地址错误、功能码不支持、从站号不对查看 pymodbus 返回的 exception code对照设备手册确认 slave_id 和寄存器范围水温显示成巨大数值float32 解析错误、word_order 配置反了打印原始寄存器值和解析后的 bytes在设备手册确认 32 位数据的字顺序LLM 不调用工具而是直接回答tools 参数未传、模型不支持 Function Calling查看请求日志确认 tools 字段完整换用支持 tools 的模型检查 SDK 版本工具返回 JSON 被模型截断返回内容太长、模型上下文窗口不够简化返回字段移除无关 description工具返回只保留 point、value、unit、timestamp多轮对话后数据混乱工具结果没有作为 tool 消息回传检查循环里是否 append 了 msg 和 tool 消息确保每次 tool_calls 后都追加对应的 tool response水泵状态显示为 3bit 位同时包含运行和故障检查 bit_mask 和描述拆分成多个布尔点或用 bit 状态枚举返回这里特别说明“LLM 不调用工具”这个坑。有些模型虽然兼容 OpenAI 接口但工具调用能力弱或者模型名称映射不正确。排查时先打印模型返回的原始resp对象确认tool_calls字段是否为空而不是靠猜测。还有一个容易忽略的问题Modbus 读写并发冲突。如果你的系统同时有写操作比如远程启停设备和读操作多个线程共用一个客户端连接可能导致报文交错。生产环境建议读操作使用连接池写操作单独使用专用连接并且设置合理的超时和重试机制。8. 工程最佳实践基于上面的实现再补充几条从实际项目中沉淀下来的经验。第一寄存器映射表必须是唯一的配置源。所有解码工具都从modbus_registry.json读取点位信息不要在使用时硬编码地址。现场点表经常调整如果地址散落在代码里改一次就是一次事故。映射表本身应该纳入版本管理并且用 diff 工具做变更评审。第二给 LLM 的工具描述要写清楚“语义”不是“协议”。比如read_modbus_point的 description 写“读取指定 Modbus 数据点的实时值返回工程单位值和采集时间”比写“读取保持寄存器第 0 号地址”有效得多。模型不需要知道协议细节它只需要知道这个函数能完成什么业务操作。第三解码函数必须做单元测试。用decode_point这种纯函数特性把手册里的典型值写成测试用例比如“输入 0x419A 0x999A期望输出 25.3”。一旦现场出现点表配置错误改完测试能快速回归。没有测试的解码代码等于把现场设备的正确性押在“这次改对了”上。第四区分只读工具和写工具。Modbus 不只是读取数据还经常用来做遥控写线圈、写寄存器。面向 LLM 暴露写工具时权限边界必须收紧。建议对所有写操作增加二次确认让 LLM 先返回操作意图由系统确认后再执行。在工业场景中“安全地拒绝”比“聪明地操作”更可靠。第五日志记录要贯穿全链路。设备接入层记录 Modbus 报文语义解析层记录解码前后的数据LLM 决策层记录工具调用参数和模型回答。这样才能在故障发生时回答“模型是不是说错了”和“解码是不是错了”这两个完全不同的问题。第六考虑加缓存。Modbus 设备的数据变化频率通常不高如果 Agent 需要在同一轮里多次查询同一个点位可以用 1 秒级缓存放行避免频繁占用设备连接资源。但注意缓存时间不能太长否则设备状态已经变化模型还在用旧数据分析。9. 总结与后续方向这篇文章的核心原则只有一句话确定性任务必须交给确定性代码LLM 只做它擅长的语义理解和交互。Modbus 寄存器解码属于前者设备状态分析属于后者。与其花大量精力让模型学会字节解析不如在中间加一个工具层让模型永远接触不到原始寄存器的复杂性。文章里的代码示例可以直接跑通一个最小闭环模拟从站提供数据语义解析层把寄存器变成带单位的 JSONLLM 通过 Function Calling 查询数据并回答用户问题。这个模式不只适用于 Modbus同样适用于 OPC UA、BACnet、MQTT 等其他工业协议。协议栈负责和硬件打交道映射表负责把协议数据翻译成业务数据LLM 负责在业务数据之上提供智能交互。下一步可以做的事很多把这套工具封装成 MCP Server让更多 Agent 客户端直接复用把历史数据接入时序数据库给 LLM 增加趋势分析和异常预测能力把设备点位组织成领域知识图谱让模型理解“供回水温差”这种跨点位的业务概念。如果你正在做 LLM 和工业设备结合的实践建议从最小的一个点位开始先把“读取温度”这条链路跑通再去扩展复杂逻辑。记住LLM 不需要学会 Modbus你只需要让它永远不用碰 Modbus。