如果你最近在折腾 LLM Agent 接入工业设备，大概率会有同样的感受：让大模型去直接解码 Modbus 寄存器，就像让一个文科生去逐字节翻译二进制协议报文。不是大模型不聪明，而是这件事从一开始就不该交给它。
Modbus 寄存器本质上是 16 位无符号整数的数组，它本身不携带数据类型、单位、小数点位置、字节顺序这些信息。设备手册里写一句“供水温度，地址 0，2 个寄存器，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 或 MCP（Model Context Protocol）工具，调用第二层暴露出来的查询函数。LLM 只需要理解工具描述，比如“读取供水温度，返回摄氏度和时间戳”，至于这个函数内部是走 Modbus TCP 还是解析 Float32，LLM 完全不需要知道。
这个设计最大的变化是责任边界被划清楚了：
- 设备接入层管连接和通信，保证报文不丢、不重、不错；
- 语义解析层管数据还原，保证字节到物理量的映射准确；
- 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-endian，2-3 回水温度 Float32 big-endian # 4 水泵状态 uint16，bit0=运行，bit1=故障 # 5 室外温度 uint16，缩放 0.1 datablock = ModbusSequentialDataBlock(0x0000, [ 0x419A, 0x999A, # 25.3℃ 0x4190, 0xCCCD, # 18.1℃ 0x0001, # 水泵运行 0x00FE # 25.4℃ ]) slave_context = ModbusSlaveContext(di=datablock, co=datablock, hr=datablock, ir=datablock) server_context = ModbusServerContext(slaves=slave_context, single=True) StartTcpServer(context=server_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(path="modbus_registry.json"): with open(path, "r", encoding="utf-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, byteorder="big") 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, byteorder="big"))[0] * scale return value raise ValueError(f"不支持的数据类型: {data_type}") def read_point(host, port, slave_id, point_name, registry=None): 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, port=port) if not client.connect(): return {"error": "Modbus 连接失败"} try: rr = client.read_holding_registers( address=point_cfg["address"], count=point_cfg["count"], slave=slave_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_url=BASE_URL, api_key=API_KEY) messages = [ {"role": "system", "content": "你是工业设备值班助手，回答必须基于工具返回的实时数据，不要编造数值。"}, {"role": "user", "content": "当前供水温度是多少？水泵状态正常吗？"} ] for _ in range(5): resp = client.chat.completions.create( model=MODEL, messages=messages, tools=TOOLS, ) 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_ascii=False) }) 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. 运行与效果验证
先确认模拟器已经启动，然后运行 Agent：
python 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 |
| 水泵状态显示为 3 | bit 位同时包含运行和故障 | 检查 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 寄存器解码属于前者，设备状态分析属于后者。与其花大量精力让模型学会字节解析，不如在中间加一个工具层，让模型永远接触不到原始寄存器的复杂性。
文章里的代码示例可以直接跑通一个最小闭环：模拟从站提供数据，语义解析层把寄存器变成带单位的 JSON，LLM 通过 Function Calling 查询数据并回答用户问题。这个模式不只适用于 Modbus，同样适用于 OPC UA、BACnet、MQTT 等其他工业协议。协议栈负责和硬件打交道，映射表负责把协议数据翻译成业务数据，LLM 负责在业务数据之上提供智能交互。
下一步可以做的事很多：把这套工具封装成 MCP Server，让更多 Agent 客户端直接复用；把历史数据接入时序数据库，给 LLM 增加趋势分析和异常预测能力；把设备点位组织成领域知识图谱，让模型理解“供回水温差”这种跨点位的业务概念。
如果你正在做 LLM 和工业设备结合的实践，建议从最小的一个点位开始，先把“读取温度”这条链路跑通，再去扩展复杂逻辑。记住，LLM 不需要学会 Modbus，你只需要让它永远不用碰 Modbus。
