核心思路LLM 负责说话Modbus 工具负责干活这次我们来聊一个有点具体的工程问题当 LLM 需要处理 Modbus 寄存器时直接让大模型去解码寄存器值结果往往不靠谱。Modbus 寄存器本质上是二进制结构化数据涉及功能码、字节序、数据类型、缩放系数、位映射等一堆规则而大模型擅长的自然语言模式识别在这一场景里恰恰派不上用场。与其反复提示、纠正 LLM 的错误判断不如从架构上绕开这个问题让 LLM 永远不需要解码 Modbus 寄存器把解码和通信逻辑下沉到专用工具层。LLM 只负责理解用户意图、整理参数、生成工具调用真正的寄存器读写和解码全部由工具函数完成。这就是本文要拆解的方案。文章会围绕一条主线展开先用具体例子说明 LLM 在 Modbus 寄存器解码上的短板然后给出一个“LLM Function Calling Modbus 工具层”的架构接着用 Python 代码实现 Modbus 工具层、工具注册、LLM 调用链路再做一轮功能测试和效果验证最后补充 API 封装、批量任务、常见问题和最佳实践。这套思路适合三类读者一是正在做工业数据问答、设备巡检助手等项目的开发者二是想把大模型接到 PLC、传感器、能源仪表等 Modbus 设备上的工程师三是刚开始接触 Function Calling想找一个实际问题练手的 LLM 应用开发者。1. 核心能力速览能力项说明项目定义通过 Function Calling 让 LLM 调用 Modbus 工具避免 LLM 直接解码寄存器核心思路意图理解交给 LLM解码和通信交给工具层技术栈Python、pymodbus、OpenAI 兼容 Function Calling、FastAPI可选支持协议Modbus TCP / Modbus RTU以 pymodbus 支持范围为准主要功能自然语言查询寄存器、写入寄存器、批量读取、异常检测适用场景工业设备数据问答、巡检助手、MES/ERP 数据对接、调试工具硬件要求普通 PC 或服务器即可无需 GPULLM 可走云端 API 或本地模型部署方式命令行脚本、HTTP 服务、定时批量任务是否支持 API支持可将工具封装为 HTTP 接口是否支持批量任务支持可配置点位表批量轮询安全边界工业设备操作需授权写入操作必须增加确认机制这套方案并不试图提升 LLM 对二进制数据的“理解能力”而是从架构层面把 LLM 不擅长的部分彻底拿掉。这也是它和“给 LLM 堆提示词”“用 RAG 喂协议文档”这类思路最本质的区别。2. 为什么 LLM 不适合直接解码 Modbus 寄存器先看一个典型场景。设备手册上写着某个温度传感器的当前值在保持寄存器地址 40001协议地址 0x0000数据类型是 32 位浮点数字节序是 Big-Endian字序是 Little-Endian。如果让 LLM 直接根据寄存器原始值0x4195 0x999A推测它可能会尝试用 ASCII、十六进制或“推理”的方式解读结果大概率是错的。原因不是 LLM 不够聪明而是这个问题本质上是一段确定的二进制解码程序不是文本生成任务。Modbus 寄存器解码的难点集中在几个地方功能码决定了操作类型。读保持寄存器是 0x03读输入寄存器是 0x04写单个寄存器是 0x06写多个寄存器是 0x10。LLM 如果不熟悉 Modbus 协议很容易选错功能码。地址存在协议地址和设备手册地址的映射关系。Modbus 协议地址从 0 开始但设备手册常写成 40001、30001 这样的“PLC 风格地址”换算规则因厂商而异。数据类型不唯一。同一个 16 位寄存器可以解释成无符号整数、有符号整数、高字节位合并的 32 位整数、浮点数、甚至是多个开关位。没有设备手册和字节序信息LLM 无从判断。字节序和字序的处理非常严格。32 位以上数据需要跨多个寄存器寄存器顺序和字节顺序稍有不同结果就是完全错误的数值。设备返回错误码的情况很多。非法地址、非法数据值、从站无响应这些都需要协议层判断LLM 的“生成式猜测”在这里没有任何优势。所以结论不是“LLM 需要更多提示词”而是“解码任务根本不应该交给 LLM”。这也正是标题的意思既然 LLM 不擅长解码 Modbus 寄存器那就让它永远不需要这么做。把确定性交给代码把灵活性留给大模型这才是 LLM 应用落地时更合理的分工。3. 整体架构设计LLM 只做意图理解工具层只做解码通信这套方案的核心是“职责分离”。LLM 只做它擅长的事理解用户的自然语言转换成结构化的工具调用参数。Modbus 工具层做它擅长的事连接设备、读取寄存器、按规则解码、返回确定性的数值。两层通过 Function Calling 机制衔接。用户提问 3号从站当前温度是多少 | v ----------------------- | LLM 应用 | | 识别意图、生成工具调用 | ----------------------- | v 调用 read_holding_registers(address0, count2, data_typefloat32) ----------------------- | Modbus 工具层 | | 读取寄存器、按规则解码 | ----------------------- | v Modbus TCP / RTU ----------------------- | Modbus 从站设备 | -----------------------架构上有几个关键决策工具函数的返回值必须是解码后的、人可读的数值而不是原始字节。这样 LLM 拿到结构化结果后只需要组织语言回答用户。工具注册时要把参数约束放在 JSON Schema 里让 LLM 按枚举生成参数减少非法参数出现。所有写操作必须在工具层做二次保护例如独立的写入函数、必填确认字段、写入值范围检查。LLM 服务本身可以是云端 API也可以是本地模型只要支持 Function Calling 格式即可。这种架构还有一个好处Modbus 工具层可以被任意多个 LLM 应用复用。不管上层用的是 GPT、Claude还是某个开源模型只要它们都能输出标准 Function Calling 格式工具层可以保持不变。这相当于把“看懂自然语言”和“读懂设备协议”两个问题彻底解耦。4. 环境准备与前置条件以下环境准备以最常见的 Python 部署方式为例实际环境可能因操作系统和网络条件有所不同。4.1 Python 与依赖建议使用 Python 3.10 以上版本。需要安装的核心依赖pip install pymodbus openai fastapi uvicorn pyyamlpymodbusModbus TCP/RTU 通信库负责和现场设备通信。openai调用 OpenAI 兼容的 LLM 接口代码里会用到 Function Calling。fastapi uvicorn把工具层封装成 HTTP 服务方便和 LLM 解耦。pyyaml读取点位配置文件方便批量任务管理。如果要用本地 LLM 并支持 Function Calling可以选择支持工具调用的本地推理框架或模型服务接口保持 OpenAI 兼容即可。4.2 Modbus 设备或模拟环境测试时需要有一个 Modbus 从站。如果没有真实设备可以使用 Modbus Slave 模拟器在本地启动一个从站服务配置好寄存器初始值。另一台测试机用 Modbus Poll 也可以但这里我们是自己写代码推荐使用支持命令行或脚本启动的模拟器。Modbus Poll 是日常调试常用的上位机工具它本身也是验证工具层读写结果是否正确的一个参照物。实测环境建议使用局域网内的一台 Modbus TCP 设备或者本机启动模拟从站。文章后续代码以 Modbus TCP 为例Modbus RTU 的区别主要在 client 初始化时使用串口配置协议层面的寄存器读写逻辑完全一致。4.3 网络与端口检查Modbus TCP 默认端口是 502。如果设备在远程服务器或 PLC 网段内需要先确认本机到设备 502 端口能通信。可以用下面的命令检查# 检查 TCP 502 端口是否可达IP 替换为实际设备地址 nc -zv 192.168.1.100 502如果是不通的状态优先排查防火墙、交换机 VLAN、设备自身使能状态。这里有一个常见坑“能 ping 通但 Modbus 扫描不通”。原因通常是 502 端口只对特定 IP 开放或者设备 Modbus Server 功能没有开启。Ping 通只代表网络层通不代表应用层服务可用。所以调试时不要只看 ICMP 结果直接用 Modbus 工具测才是有效验证。5. Modbus 工具层实现把解码逻辑做成确定性代码工具层是整个方案的“准确率来源”。建议把 Modbus 通信和数值解码封装成独立模块尽量少依赖 LLM 逻辑。这样后续无论换什么模型工具层的行为都不会变。5.1 通信客户端封装# modbus_tool.py from pymodbus.client import ModbusTcpClient class ModbusTool: def __init__(self, host: str, port: int 502, unit_id: int 1): self.host host self.port port self.unit_id unit_id self.client None def connect(self) - bool: if self.client is None: self.client ModbusTcpClient(self.host, portself.port) return self.client.connect() def close(self): if self.client: self.client.close() self.client None def read_holding_registers(self, address: int, count: int 1): if not self.connect(): raise RuntimeError(Modbus TCP 连接失败) result self.client.read_holding_registers( addressaddress, countcount, slaveself.unit_id ) if result.isError(): raise RuntimeError(f读取保持寄存器失败: {result}) return result.registers def write_single_register(self, address: int, value: int): if not self.connect(): raise RuntimeError(Modbus TCP 连接失败) result self.client.write_register( addressaddress, valuevalue, slaveself.unit_id ) if result.isError(): raise RuntimeError(f写入保持寄存器失败: {result}) return True这个类的目标是把 pymodbus 的底层调用统一隔离。后续如果需要支持 Modbus RTU只需要把初始化部分替换成串口方式# Modbus RTU 初始化示例 from pymodbus.client import ModbusSerialClient rtu_client ModbusSerialClient( port/dev/ttyUSB0, baudrate9600, bytesize8, parityN, stopbits1, timeout3 )从代码结构看工具层只暴露两个核心动作读和写。LLM 不直接接触串口参数、超时设置、寄存器原始字节这些全部由工具层内部处理。5.2 寄存器数值解码字节序是重灾区解码逻辑必须严格按设备手册来。下面是一个支持 uint16、int16、uint32、int32、float32 的通用解码器字节序和字序均支持配置。# decoder.py import struct def decode_registers(registers, data_type: str uint16, byte_order: str big, word_order: str big): 按指定数据类型解析 Modbus 寄存器列表。 registers: 从设备读取到的寄存器列表例如 [0x4195, 0x999A] data_type: uint16/int16/uint32/int32/float32 byte_order: big/little字节序 word_order: big/little字序仅跨寄存器类型有效 if data_type in (uint16, int16): value registers[0] if data_type int16 and value 0x8000: value - 0x10000 return value if len(registers) 2: raise ValueError(32位数据类型至少需要2个寄存器) regs list(registers) if word_order little: regs regs[::-1] raw b.join(r.to_bytes(2, byteorderbyte_order, signedFalse) for r in regs) if data_type uint32: return struct.unpack(I, raw)[0] if data_type int32: return struct.unpack(i, raw)[0] if data_type float32: return struct.unpack(f, raw)[0] raise ValueError(f不支持的数据类型: {data_type})这里的关键是LLM 不需要知道字节序怎么排它会把这个参数按枚举传进来解码器保证同样的原始值在任何一次调用中都返回同样的数值。这就是“确定性”工具层的好处。字节序具体怎么配置最终取决于设备厂商的手册。常见 PLC 厂商默认是 Big-Endian 字序但很多仪表、传感器会提供“字节序可配置”选项实际测试时要先在 Modbus Poll 这类工具里手动读一次对照设备手册确认解码方式。5.3 工具注册与 JSON Schema为了让 LLM 正确生成调用需要把工具信息告知 LLM。以 OpenAI 兼容格式为例工具定义如下[ { type: function, function: { name: read_holding_registers, description: 读取Modbus设备的保持寄存器可指定地址、数量、数据类型、字节序和字序, parameters: { type: object, properties: { address: {type: integer, description: 寄存器起始地址协议地址从0开始}, count: {type: integer, description: 寄存器数量32位以上类型需要2个}, data_type: { type: string, enum: [uint16, int16, uint32, int32, float32] }, byte_order: { type: string, enum: [big, little] }, word_order: { type: string, enum: [big, little] } }, required: [address, count] } } } ]注意 JSON Schema 里的 description 要写得尽量明确这会直接影响 LLM 解析用户问题的准确率。把“地址从 0 开始”“32 位类型需要 2 个寄存器”这类约束写清楚能减少很多错误调用。实际项目中工具定义可能不止一个但原则很简单每个工具只做一件事参数尽量少枚举尽量全。6. LLM Function Calling 接入工具层就绪后接下来把 LLM 接入循环。这里使用 OpenAI 兼容接口。完整流程是用户提问 - 带 tools 调用 LLM - 拿到 tool_calls - 执行工具 - 把工具结果回传 LLM - 输出最终回答。6.1 基础调用代码# assistant.py import json from openai import OpenAI from modbus_tool import ModbusTool from decoder import decode_registers client OpenAI(api_keyYOUR_API_KEY, base_url可选本地模型服务地址) modbus ModbusTool(host192.168.1.100, unit_id1) modbus.connect() tools [ ... ] # 上文的工具定义 def execute_tool(name: str, args: dict): if name read_holding_registers: registers modbus.read_holding_registers( addressint(args[address]), countint(args[count]) ) return decode_registers( registers, data_typeargs.get(data_type, uint16), byte_orderargs.get(byte_order, big), word_orderargs.get(word_order, big) ) if name write_single_register: modbus.write_single_register( addressint(args[address]), valueint(args[value]) ) return 写入成功 raise ValueError(f未知工具: {name}) def chat_with_modbus(user_message: str) - str: messages [ {role: system, content: 你是工业现场数据助手。你可以通过Modbus工具读取设备寄存器回答数值问题。回答时要说明数据来源和单位。涉及写入操作时必须先向用户二次确认。}, {role: user, content: user_message} ] response client.chat.completions.create( modelgpt-4o-mini, # 实际模型按部署环境调整 messagesmessages, toolstools, tool_choiceauto ) msg response.choices[0].message # 如果 LLM 没有调用工具直接返回 if not msg.tool_calls: return msg.content or # 执行工具并回传结果 messages.append(msg) for tool_call in msg.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) tool_result execute_tool(tool_name, tool_args) messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps({value: tool_result}) }) second_response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto ) return second_response.choices[0].message.content这段代码中的关键点工具执行结果用 JSON 字符串回传给 LLMLLM 看到的是value: 25.6而不是原始寄存器字节。第二次调用 LLM 时LLM 会把工具结果组织成自然语言回答。实际模型名、API 地址需要根据你的 LLM 服务调整。如果使用本地模型把 base_url 改成本地推理服务地址即可。一次对话中工具调用可能有多个tool_calls是一个列表所以要用 for 循环逐个执行。6.2 减少错误调用的提示词策略虽然工具层能保证解码准确但 LLM 仍可能生成错误参数。建议在 system prompt 里加入以下约束- 所有寄存器地址都是协议地址从 0 开始。 - 读取 32 位浮点数时 count 必须是 2。 - 如果不确定数据类型请先用 uint16 读取并询问用户设备手册信息。 - 涉及写入操作时必须主动向用户二次确认。 - 如果用户没有提供从站号默认使用 1 号从站。这类约束能让 LLM 在参数歧义时停下来问用户而不是乱猜一个值。经验是与其让 LLM 从设备手册里“推断”没给的信息不如让它把缺失参数列出来反问用户。7. 功能测试与效果验证部署完成后按以下顺序测试功能和稳定性。建议先在模拟器上跑完所有测试再切换到真实设备。7.1 测试环境搭建本机启动一个 Modbus 模拟从站初始化固定寄存器值地址 0float32 25.6实际占 0、1 两个寄存器大端字节序地址 10uint16 100地址 20int16 -50地址 30uint16 400用 Modbus Poll 连接模拟从站确认这些值能被读到。这一步能先排除网络和从站配置问题后续再测 LLM 链路时如果数值不对可以直接判断是工具层参数问题还是解码逻辑问题。7.2 自然语言查询测试测试用例和预期输出测试输入预期工具调用预期回答当前温度是多少read_holding_registers(address0, count2, data_typefloat32)当前温度是 25.6℃地址 10 的原始值是多少read_holding_registers(address10, count1)地址 10 的寄存器值是 100读一下地址 20 的有符号数值read_holding_registers(address20, count1, data_typeint16)地址 20 的值是 -50帮我查一下 3 号从站的电压先尝试读若从站不存在则返回错误提示查询失败检查从站号判断成功的标准工具层返回的数值和设备模拟器里的初始值一致LLM 最终回答的数值正确且单位正确。如果数值差了一个数量级优先检查字节序和字序。我在实际调试中遇到最多的问题就是 float32 的 word_order 配反导致读出来的数值完全不对。7.3 写入操作测试写入是高风险操作测试时应该先连接模拟从站不要直接操作现场设备。# 在 Python 中直接调用写入工具 from modbus_tool import ModbusTool tool ModbusTool(host127.0.0.1, unit_id1) tool.connect() tool.write_single_register(address10, value200) # 再读回来验证 value tool.read_holding_registers(address10, count1) print(value) # 预期输出 200测试过程中需要验证写入成功后再读取同一个寄存器值是否一致。写入值超出 16 位范围时工具层是否能拦截。LLM 在用户没有明确确认时是否拒绝执行写操作。如果设备是只读点位工具层返回错误码时LLM 能否把这个失败信息合理解释给用户。写入操作建议单独配置风险等级。现场有大量只读仪表乱写寄存器可能导致设备参数错乱所以写工具默认不给 LLM 开放或者只在显式授权模式下开放。8. 接口 API 与批量任务为了让这套能力真正可用建议把工具层封装成 HTTP 服务然后由 LLM 应用去调用。这样也方便多个 LLM 应用复用同一套 Modbus 工具。8.1 FastAPI 封装示例# api.py from fastapi import FastAPI from pydantic import BaseModel from modbus_tool import ModbusTool from decoder import decode_registers app FastAPI() modbus ModbusTool(host192.168.1.100, unit_id1) class ReadRequest(BaseModel): address: int count: int 1 data_type: str uint16 byte_order: str big word_order: str big class WriteRequest(BaseModel): address: int value: int app.post(/modbus/read) def read_registers(req: ReadRequest): registers modbus.read_holding_registers(addressreq.address, countreq.count) value decode_registers( registers, data_typereq.data_type, byte_orderreq.byte_order, word_orderreq.word_order ) return {address: req.address, value: value, registers: registers} app.post(/modbus/write) def write_register(req: WriteRequest): modbus.write_single_register(addressreq.address, valuereq.value) return {address: req.address, value: req.value, status: ok}启动 HTTP 服务uvicorn api:app --host 127.0.0.1 --port 8000调用测试curl -X POST http://127.0.0.1:8000/modbus/read \ -H Content-Type: application/json \ -d {address: 0, count: 2, data_type: float32, byte_order: big, word_order: big}返回结果{address: 0, value: 25.6, registers: [16789, 39322]}这个接口只暴露给内网或 LLM 应用服务使用不要开放到公网。Modbus 设备往往在工控网络内权限边界要严格把控。如果需要从外部访问建议增加 Token 鉴权和 IP 白名单并且只开放只读接口写入接口单独走审批流程。8.2 批量轮询任务如果需要定时采集一批点位可以维护一个点位配置文件用 Python 脚本批量执行。# points.yaml points: - name: 温度 address: 0 count: 2 data_type: float32 byte_order: big word_order: big - name: 压力 address: 10 count: 2 data_type: float32 byte_order: big word_order: big - name: 液位 address: 20 count: 1 data_type: uint16 byte_order: big word_order: big批量读取脚本# batch_read.py import yaml from modbus_tool import ModbusTool from decoder import decode_registers with open(points.yaml, r) as f: config yaml.safe_load(f) tool ModbusTool(host192.168.1.100, unit_id1) tool.connect() for point in config[points]: try: regs tool.read_holding_registers(point[address], point[count]) value decode_registers( regs, data_typepoint[data_type], byte_orderpoint.get(byte_order, big), word_orderpoint.get(word_order, big) ) print(f{point[name]}: {value}) except Exception as e: print(f{point[name]}: 读取失败 - {e})批量任务要注意每个设备建议串行读取或限制并发避免从站过载。读取失败时要有重试策略和日志记录。点位配置要做版本管理因为现场设备点位经常调整。批量任务建议写入时间戳和状态字段方便后续做历史趋势分析。9. 资源占用与性能观察这个方案本身对资源占用很低因为它不运行训练任务只是做 LLM 调用和 Modbus 通信。需要观察的性能点如下。LLM 调用延迟是主要瓶颈。一次 Function Calling 通常需要两次 LLM 调用总耗时一般在几百毫秒到几秒和 LLM 服务有关。如果用户只是查询单个点位这个延迟完全可以接受。如果用户连续问几十个点位每次查询都走两轮 LLM 调用延迟会叠加。Modbus 通信延迟。TCP 方式通常在毫秒级但现场网络拥塞或设备响应慢时会明显拉长。批量轮询时如果从站响应慢可以用timeout参数控制单次读取的最长等待时间避免一个点位卡住整个任务。如果使用本地 LLM需要关注显存占用和推理速度。支持 Function Calling 的本地模型建议至少使用 7B 以上规模实际显存以模型量化版本为准。8GB 显存可以跑一些量化后的模型但推理速度和回答质量需要根据实际模型评测。性能观察方法# 观察进程资源占用 top -p $(pgrep -f assistant.py) # 或者使用 Python 直接测单次读取耗时 import time start time.time() tool.read_holding_registers(0, 2) print(time.time() - start)如果发现单次工具调用耗时异常比如超过 1 秒需要检查是网络延迟还是设备响应慢。可以用 Modbus Poll 手动读取同一地址对比如果 Modbus Poll 也慢那就是设备或网络问题和 LLM 链路无关。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Modbus TCP 连接失败IP 端口错误、设备未启动、防火墙拦截用 nc 检查 502 端口看设备日志修正 IP 端口放行防火墙确认设备 Modbus Server 使能能 ping 通但 Modbus 扫描不通502 端口未开放或设备 Modbus 未使能用 Modbus Poll 连接测试检查设备通信设置确认端口和从站号读取返回 Illegal Data Address地址超范围或设备映射不一致查看设备寄存器映射表对比协议地址按设备手册调整地址偏移数值完全不对数据类型/字节序/字序配置错误先用 uint16 读取寄存器原始值对照手册修改 data_type、byte_order 或 word_order 参数读取返回 Illegal Data Value寄存器数量或数据类型不匹配检查 count 是否为 1 或 2调整 count确认数据类型对应的寄存器数量LLM 生成的参数不合法提示词约束不足或模型理解偏差打印 tool_calls 的原始调用参数加强 system prompt增加枚举约束写入没有生效写入值超范围、现场设备禁止写入查看功能码错误码工具层增加范围校验检查设备写使能多次调用后连接断开长连接未管理、设备超时查看日志中的断连时间点每次调用检查连接状态失败时重连LLM 回答单位或语言不正确缺少系统提示词约束查看 system prompt要求回答中带上单位和数据来源说明批量任务中一个点位失败卡住无超时设置或异常未捕获查看日志中的异常栈单点读取加 timeout异常单独捕获并记录后继续11. 最佳实践与使用建议第一工具层和 LLM 逻辑要彻底分离。工具层不依赖任何 LLM 内容只做 Modbus 通信和解码。这样即使换 LLM 模型工具层都不用改。我建议把 Modbus 工具设计成一个独立的 Python 包或微服务单独维护版本LLM 应用只通过接口调用。第二所有涉及写寄存器的操作建议单独封装一个工具函数并且让 LLM 在调用前必须经过用户确认。最好在工具函数内部也加一道“最小值/最大值”校验防止异常参数写到设备上。比如某个寄存器只能写 0 或 1工具层就应该在校验时拦截其他值。第三给工具函数增加超时控制。Modbus 设备在异常状态下可能长时间不响应没有超时的话整个流程会被卡死。pymodbus 的 timeout 参数要设置HTTP API 层也要设置超时。推荐设置 3 到 5 秒具体以设备响应时间为准。第四日志记录非常关键。把每次工具调用的参数、原始寄存器返回值、解码结果、耗时全部记录下来。这样设备点位调整或出现数值异常时能快速定位是参数问题还是设备问题。日志格式建议包含时间戳、操作类型、点位名称、原始值、解码值、错误信息。第五现场使用前一定要在模拟器上完整验证。建议准备一个和现场设备点位一致的点位表先用模拟器跑一遍所有读操作确认解码结果与预期一致再切换真实设备。模拟器里可以人为设置异常点位测试工具层在异常情况下的表现。第六工业数据合规。Modbus 设备通常部署在生产环境读取和写入设备数据前必须确认已获得设备归属方和运维方的授权。不要把生产设备的调试接口暴露 LLM+Function Calling 让 Modbus 寄存器解码更可靠
核心思路LLM 负责说话Modbus 工具负责干活这次我们来聊一个有点具体的工程问题当 LLM 需要处理 Modbus 寄存器时直接让大模型去解码寄存器值结果往往不靠谱。Modbus 寄存器本质上是二进制结构化数据涉及功能码、字节序、数据类型、缩放系数、位映射等一堆规则而大模型擅长的自然语言模式识别在这一场景里恰恰派不上用场。与其反复提示、纠正 LLM 的错误判断不如从架构上绕开这个问题让 LLM 永远不需要解码 Modbus 寄存器把解码和通信逻辑下沉到专用工具层。LLM 只负责理解用户意图、整理参数、生成工具调用真正的寄存器读写和解码全部由工具函数完成。这就是本文要拆解的方案。文章会围绕一条主线展开先用具体例子说明 LLM 在 Modbus 寄存器解码上的短板然后给出一个“LLM Function Calling Modbus 工具层”的架构接着用 Python 代码实现 Modbus 工具层、工具注册、LLM 调用链路再做一轮功能测试和效果验证最后补充 API 封装、批量任务、常见问题和最佳实践。这套思路适合三类读者一是正在做工业数据问答、设备巡检助手等项目的开发者二是想把大模型接到 PLC、传感器、能源仪表等 Modbus 设备上的工程师三是刚开始接触 Function Calling想找一个实际问题练手的 LLM 应用开发者。1. 核心能力速览能力项说明项目定义通过 Function Calling 让 LLM 调用 Modbus 工具避免 LLM 直接解码寄存器核心思路意图理解交给 LLM解码和通信交给工具层技术栈Python、pymodbus、OpenAI 兼容 Function Calling、FastAPI可选支持协议Modbus TCP / Modbus RTU以 pymodbus 支持范围为准主要功能自然语言查询寄存器、写入寄存器、批量读取、异常检测适用场景工业设备数据问答、巡检助手、MES/ERP 数据对接、调试工具硬件要求普通 PC 或服务器即可无需 GPULLM 可走云端 API 或本地模型部署方式命令行脚本、HTTP 服务、定时批量任务是否支持 API支持可将工具封装为 HTTP 接口是否支持批量任务支持可配置点位表批量轮询安全边界工业设备操作需授权写入操作必须增加确认机制这套方案并不试图提升 LLM 对二进制数据的“理解能力”而是从架构层面把 LLM 不擅长的部分彻底拿掉。这也是它和“给 LLM 堆提示词”“用 RAG 喂协议文档”这类思路最本质的区别。2. 为什么 LLM 不适合直接解码 Modbus 寄存器先看一个典型场景。设备手册上写着某个温度传感器的当前值在保持寄存器地址 40001协议地址 0x0000数据类型是 32 位浮点数字节序是 Big-Endian字序是 Little-Endian。如果让 LLM 直接根据寄存器原始值0x4195 0x999A推测它可能会尝试用 ASCII、十六进制或“推理”的方式解读结果大概率是错的。原因不是 LLM 不够聪明而是这个问题本质上是一段确定的二进制解码程序不是文本生成任务。Modbus 寄存器解码的难点集中在几个地方功能码决定了操作类型。读保持寄存器是 0x03读输入寄存器是 0x04写单个寄存器是 0x06写多个寄存器是 0x10。LLM 如果不熟悉 Modbus 协议很容易选错功能码。地址存在协议地址和设备手册地址的映射关系。Modbus 协议地址从 0 开始但设备手册常写成 40001、30001 这样的“PLC 风格地址”换算规则因厂商而异。数据类型不唯一。同一个 16 位寄存器可以解释成无符号整数、有符号整数、高字节位合并的 32 位整数、浮点数、甚至是多个开关位。没有设备手册和字节序信息LLM 无从判断。字节序和字序的处理非常严格。32 位以上数据需要跨多个寄存器寄存器顺序和字节顺序稍有不同结果就是完全错误的数值。设备返回错误码的情况很多。非法地址、非法数据值、从站无响应这些都需要协议层判断LLM 的“生成式猜测”在这里没有任何优势。所以结论不是“LLM 需要更多提示词”而是“解码任务根本不应该交给 LLM”。这也正是标题的意思既然 LLM 不擅长解码 Modbus 寄存器那就让它永远不需要这么做。把确定性交给代码把灵活性留给大模型这才是 LLM 应用落地时更合理的分工。3. 整体架构设计LLM 只做意图理解工具层只做解码通信这套方案的核心是“职责分离”。LLM 只做它擅长的事理解用户的自然语言转换成结构化的工具调用参数。Modbus 工具层做它擅长的事连接设备、读取寄存器、按规则解码、返回确定性的数值。两层通过 Function Calling 机制衔接。用户提问 3号从站当前温度是多少 | v ----------------------- | LLM 应用 | | 识别意图、生成工具调用 | ----------------------- | v 调用 read_holding_registers(address0, count2, data_typefloat32) ----------------------- | Modbus 工具层 | | 读取寄存器、按规则解码 | ----------------------- | v Modbus TCP / RTU ----------------------- | Modbus 从站设备 | -----------------------架构上有几个关键决策工具函数的返回值必须是解码后的、人可读的数值而不是原始字节。这样 LLM 拿到结构化结果后只需要组织语言回答用户。工具注册时要把参数约束放在 JSON Schema 里让 LLM 按枚举生成参数减少非法参数出现。所有写操作必须在工具层做二次保护例如独立的写入函数、必填确认字段、写入值范围检查。LLM 服务本身可以是云端 API也可以是本地模型只要支持 Function Calling 格式即可。这种架构还有一个好处Modbus 工具层可以被任意多个 LLM 应用复用。不管上层用的是 GPT、Claude还是某个开源模型只要它们都能输出标准 Function Calling 格式工具层可以保持不变。这相当于把“看懂自然语言”和“读懂设备协议”两个问题彻底解耦。4. 环境准备与前置条件以下环境准备以最常见的 Python 部署方式为例实际环境可能因操作系统和网络条件有所不同。4.1 Python 与依赖建议使用 Python 3.10 以上版本。需要安装的核心依赖pip install pymodbus openai fastapi uvicorn pyyamlpymodbusModbus TCP/RTU 通信库负责和现场设备通信。openai调用 OpenAI 兼容的 LLM 接口代码里会用到 Function Calling。fastapi uvicorn把工具层封装成 HTTP 服务方便和 LLM 解耦。pyyaml读取点位配置文件方便批量任务管理。如果要用本地 LLM 并支持 Function Calling可以选择支持工具调用的本地推理框架或模型服务接口保持 OpenAI 兼容即可。4.2 Modbus 设备或模拟环境测试时需要有一个 Modbus 从站。如果没有真实设备可以使用 Modbus Slave 模拟器在本地启动一个从站服务配置好寄存器初始值。另一台测试机用 Modbus Poll 也可以但这里我们是自己写代码推荐使用支持命令行或脚本启动的模拟器。Modbus Poll 是日常调试常用的上位机工具它本身也是验证工具层读写结果是否正确的一个参照物。实测环境建议使用局域网内的一台 Modbus TCP 设备或者本机启动模拟从站。文章后续代码以 Modbus TCP 为例Modbus RTU 的区别主要在 client 初始化时使用串口配置协议层面的寄存器读写逻辑完全一致。4.3 网络与端口检查Modbus TCP 默认端口是 502。如果设备在远程服务器或 PLC 网段内需要先确认本机到设备 502 端口能通信。可以用下面的命令检查# 检查 TCP 502 端口是否可达IP 替换为实际设备地址 nc -zv 192.168.1.100 502如果是不通的状态优先排查防火墙、交换机 VLAN、设备自身使能状态。这里有一个常见坑“能 ping 通但 Modbus 扫描不通”。原因通常是 502 端口只对特定 IP 开放或者设备 Modbus Server 功能没有开启。Ping 通只代表网络层通不代表应用层服务可用。所以调试时不要只看 ICMP 结果直接用 Modbus 工具测才是有效验证。5. Modbus 工具层实现把解码逻辑做成确定性代码工具层是整个方案的“准确率来源”。建议把 Modbus 通信和数值解码封装成独立模块尽量少依赖 LLM 逻辑。这样后续无论换什么模型工具层的行为都不会变。5.1 通信客户端封装# modbus_tool.py from pymodbus.client import ModbusTcpClient class ModbusTool: def __init__(self, host: str, port: int 502, unit_id: int 1): self.host host self.port port self.unit_id unit_id self.client None def connect(self) - bool: if self.client is None: self.client ModbusTcpClient(self.host, portself.port) return self.client.connect() def close(self): if self.client: self.client.close() self.client None def read_holding_registers(self, address: int, count: int 1): if not self.connect(): raise RuntimeError(Modbus TCP 连接失败) result self.client.read_holding_registers( addressaddress, countcount, slaveself.unit_id ) if result.isError(): raise RuntimeError(f读取保持寄存器失败: {result}) return result.registers def write_single_register(self, address: int, value: int): if not self.connect(): raise RuntimeError(Modbus TCP 连接失败) result self.client.write_register( addressaddress, valuevalue, slaveself.unit_id ) if result.isError(): raise RuntimeError(f写入保持寄存器失败: {result}) return True这个类的目标是把 pymodbus 的底层调用统一隔离。后续如果需要支持 Modbus RTU只需要把初始化部分替换成串口方式# Modbus RTU 初始化示例 from pymodbus.client import ModbusSerialClient rtu_client ModbusSerialClient( port/dev/ttyUSB0, baudrate9600, bytesize8, parityN, stopbits1, timeout3 )从代码结构看工具层只暴露两个核心动作读和写。LLM 不直接接触串口参数、超时设置、寄存器原始字节这些全部由工具层内部处理。5.2 寄存器数值解码字节序是重灾区解码逻辑必须严格按设备手册来。下面是一个支持 uint16、int16、uint32、int32、float32 的通用解码器字节序和字序均支持配置。# decoder.py import struct def decode_registers(registers, data_type: str uint16, byte_order: str big, word_order: str big): 按指定数据类型解析 Modbus 寄存器列表。 registers: 从设备读取到的寄存器列表例如 [0x4195, 0x999A] data_type: uint16/int16/uint32/int32/float32 byte_order: big/little字节序 word_order: big/little字序仅跨寄存器类型有效 if data_type in (uint16, int16): value registers[0] if data_type int16 and value 0x8000: value - 0x10000 return value if len(registers) 2: raise ValueError(32位数据类型至少需要2个寄存器) regs list(registers) if word_order little: regs regs[::-1] raw b.join(r.to_bytes(2, byteorderbyte_order, signedFalse) for r in regs) if data_type uint32: return struct.unpack(I, raw)[0] if data_type int32: return struct.unpack(i, raw)[0] if data_type float32: return struct.unpack(f, raw)[0] raise ValueError(f不支持的数据类型: {data_type})这里的关键是LLM 不需要知道字节序怎么排它会把这个参数按枚举传进来解码器保证同样的原始值在任何一次调用中都返回同样的数值。这就是“确定性”工具层的好处。字节序具体怎么配置最终取决于设备厂商的手册。常见 PLC 厂商默认是 Big-Endian 字序但很多仪表、传感器会提供“字节序可配置”选项实际测试时要先在 Modbus Poll 这类工具里手动读一次对照设备手册确认解码方式。5.3 工具注册与 JSON Schema为了让 LLM 正确生成调用需要把工具信息告知 LLM。以 OpenAI 兼容格式为例工具定义如下[ { type: function, function: { name: read_holding_registers, description: 读取Modbus设备的保持寄存器可指定地址、数量、数据类型、字节序和字序, parameters: { type: object, properties: { address: {type: integer, description: 寄存器起始地址协议地址从0开始}, count: {type: integer, description: 寄存器数量32位以上类型需要2个}, data_type: { type: string, enum: [uint16, int16, uint32, int32, float32] }, byte_order: { type: string, enum: [big, little] }, word_order: { type: string, enum: [big, little] } }, required: [address, count] } } } ]注意 JSON Schema 里的 description 要写得尽量明确这会直接影响 LLM 解析用户问题的准确率。把“地址从 0 开始”“32 位类型需要 2 个寄存器”这类约束写清楚能减少很多错误调用。实际项目中工具定义可能不止一个但原则很简单每个工具只做一件事参数尽量少枚举尽量全。6. LLM Function Calling 接入工具层就绪后接下来把 LLM 接入循环。这里使用 OpenAI 兼容接口。完整流程是用户提问 - 带 tools 调用 LLM - 拿到 tool_calls - 执行工具 - 把工具结果回传 LLM - 输出最终回答。6.1 基础调用代码# assistant.py import json from openai import OpenAI from modbus_tool import ModbusTool from decoder import decode_registers client OpenAI(api_keyYOUR_API_KEY, base_url可选本地模型服务地址) modbus ModbusTool(host192.168.1.100, unit_id1) modbus.connect() tools [ ... ] # 上文的工具定义 def execute_tool(name: str, args: dict): if name read_holding_registers: registers modbus.read_holding_registers( addressint(args[address]), countint(args[count]) ) return decode_registers( registers, data_typeargs.get(data_type, uint16), byte_orderargs.get(byte_order, big), word_orderargs.get(word_order, big) ) if name write_single_register: modbus.write_single_register( addressint(args[address]), valueint(args[value]) ) return 写入成功 raise ValueError(f未知工具: {name}) def chat_with_modbus(user_message: str) - str: messages [ {role: system, content: 你是工业现场数据助手。你可以通过Modbus工具读取设备寄存器回答数值问题。回答时要说明数据来源和单位。涉及写入操作时必须先向用户二次确认。}, {role: user, content: user_message} ] response client.chat.completions.create( modelgpt-4o-mini, # 实际模型按部署环境调整 messagesmessages, toolstools, tool_choiceauto ) msg response.choices[0].message # 如果 LLM 没有调用工具直接返回 if not msg.tool_calls: return msg.content or # 执行工具并回传结果 messages.append(msg) for tool_call in msg.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) tool_result execute_tool(tool_name, tool_args) messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps({value: tool_result}) }) second_response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto ) return second_response.choices[0].message.content这段代码中的关键点工具执行结果用 JSON 字符串回传给 LLMLLM 看到的是value: 25.6而不是原始寄存器字节。第二次调用 LLM 时LLM 会把工具结果组织成自然语言回答。实际模型名、API 地址需要根据你的 LLM 服务调整。如果使用本地模型把 base_url 改成本地推理服务地址即可。一次对话中工具调用可能有多个tool_calls是一个列表所以要用 for 循环逐个执行。6.2 减少错误调用的提示词策略虽然工具层能保证解码准确但 LLM 仍可能生成错误参数。建议在 system prompt 里加入以下约束- 所有寄存器地址都是协议地址从 0 开始。 - 读取 32 位浮点数时 count 必须是 2。 - 如果不确定数据类型请先用 uint16 读取并询问用户设备手册信息。 - 涉及写入操作时必须主动向用户二次确认。 - 如果用户没有提供从站号默认使用 1 号从站。这类约束能让 LLM 在参数歧义时停下来问用户而不是乱猜一个值。经验是与其让 LLM 从设备手册里“推断”没给的信息不如让它把缺失参数列出来反问用户。7. 功能测试与效果验证部署完成后按以下顺序测试功能和稳定性。建议先在模拟器上跑完所有测试再切换到真实设备。7.1 测试环境搭建本机启动一个 Modbus 模拟从站初始化固定寄存器值地址 0float32 25.6实际占 0、1 两个寄存器大端字节序地址 10uint16 100地址 20int16 -50地址 30uint16 400用 Modbus Poll 连接模拟从站确认这些值能被读到。这一步能先排除网络和从站配置问题后续再测 LLM 链路时如果数值不对可以直接判断是工具层参数问题还是解码逻辑问题。7.2 自然语言查询测试测试用例和预期输出测试输入预期工具调用预期回答当前温度是多少read_holding_registers(address0, count2, data_typefloat32)当前温度是 25.6℃地址 10 的原始值是多少read_holding_registers(address10, count1)地址 10 的寄存器值是 100读一下地址 20 的有符号数值read_holding_registers(address20, count1, data_typeint16)地址 20 的值是 -50帮我查一下 3 号从站的电压先尝试读若从站不存在则返回错误提示查询失败检查从站号判断成功的标准工具层返回的数值和设备模拟器里的初始值一致LLM 最终回答的数值正确且单位正确。如果数值差了一个数量级优先检查字节序和字序。我在实际调试中遇到最多的问题就是 float32 的 word_order 配反导致读出来的数值完全不对。7.3 写入操作测试写入是高风险操作测试时应该先连接模拟从站不要直接操作现场设备。# 在 Python 中直接调用写入工具 from modbus_tool import ModbusTool tool ModbusTool(host127.0.0.1, unit_id1) tool.connect() tool.write_single_register(address10, value200) # 再读回来验证 value tool.read_holding_registers(address10, count1) print(value) # 预期输出 200测试过程中需要验证写入成功后再读取同一个寄存器值是否一致。写入值超出 16 位范围时工具层是否能拦截。LLM 在用户没有明确确认时是否拒绝执行写操作。如果设备是只读点位工具层返回错误码时LLM 能否把这个失败信息合理解释给用户。写入操作建议单独配置风险等级。现场有大量只读仪表乱写寄存器可能导致设备参数错乱所以写工具默认不给 LLM 开放或者只在显式授权模式下开放。8. 接口 API 与批量任务为了让这套能力真正可用建议把工具层封装成 HTTP 服务然后由 LLM 应用去调用。这样也方便多个 LLM 应用复用同一套 Modbus 工具。8.1 FastAPI 封装示例# api.py from fastapi import FastAPI from pydantic import BaseModel from modbus_tool import ModbusTool from decoder import decode_registers app FastAPI() modbus ModbusTool(host192.168.1.100, unit_id1) class ReadRequest(BaseModel): address: int count: int 1 data_type: str uint16 byte_order: str big word_order: str big class WriteRequest(BaseModel): address: int value: int app.post(/modbus/read) def read_registers(req: ReadRequest): registers modbus.read_holding_registers(addressreq.address, countreq.count) value decode_registers( registers, data_typereq.data_type, byte_orderreq.byte_order, word_orderreq.word_order ) return {address: req.address, value: value, registers: registers} app.post(/modbus/write) def write_register(req: WriteRequest): modbus.write_single_register(addressreq.address, valuereq.value) return {address: req.address, value: req.value, status: ok}启动 HTTP 服务uvicorn api:app --host 127.0.0.1 --port 8000调用测试curl -X POST http://127.0.0.1:8000/modbus/read \ -H Content-Type: application/json \ -d {address: 0, count: 2, data_type: float32, byte_order: big, word_order: big}返回结果{address: 0, value: 25.6, registers: [16789, 39322]}这个接口只暴露给内网或 LLM 应用服务使用不要开放到公网。Modbus 设备往往在工控网络内权限边界要严格把控。如果需要从外部访问建议增加 Token 鉴权和 IP 白名单并且只开放只读接口写入接口单独走审批流程。8.2 批量轮询任务如果需要定时采集一批点位可以维护一个点位配置文件用 Python 脚本批量执行。# points.yaml points: - name: 温度 address: 0 count: 2 data_type: float32 byte_order: big word_order: big - name: 压力 address: 10 count: 2 data_type: float32 byte_order: big word_order: big - name: 液位 address: 20 count: 1 data_type: uint16 byte_order: big word_order: big批量读取脚本# batch_read.py import yaml from modbus_tool import ModbusTool from decoder import decode_registers with open(points.yaml, r) as f: config yaml.safe_load(f) tool ModbusTool(host192.168.1.100, unit_id1) tool.connect() for point in config[points]: try: regs tool.read_holding_registers(point[address], point[count]) value decode_registers( regs, data_typepoint[data_type], byte_orderpoint.get(byte_order, big), word_orderpoint.get(word_order, big) ) print(f{point[name]}: {value}) except Exception as e: print(f{point[name]}: 读取失败 - {e})批量任务要注意每个设备建议串行读取或限制并发避免从站过载。读取失败时要有重试策略和日志记录。点位配置要做版本管理因为现场设备点位经常调整。批量任务建议写入时间戳和状态字段方便后续做历史趋势分析。9. 资源占用与性能观察这个方案本身对资源占用很低因为它不运行训练任务只是做 LLM 调用和 Modbus 通信。需要观察的性能点如下。LLM 调用延迟是主要瓶颈。一次 Function Calling 通常需要两次 LLM 调用总耗时一般在几百毫秒到几秒和 LLM 服务有关。如果用户只是查询单个点位这个延迟完全可以接受。如果用户连续问几十个点位每次查询都走两轮 LLM 调用延迟会叠加。Modbus 通信延迟。TCP 方式通常在毫秒级但现场网络拥塞或设备响应慢时会明显拉长。批量轮询时如果从站响应慢可以用timeout参数控制单次读取的最长等待时间避免一个点位卡住整个任务。如果使用本地 LLM需要关注显存占用和推理速度。支持 Function Calling 的本地模型建议至少使用 7B 以上规模实际显存以模型量化版本为准。8GB 显存可以跑一些量化后的模型但推理速度和回答质量需要根据实际模型评测。性能观察方法# 观察进程资源占用 top -p $(pgrep -f assistant.py) # 或者使用 Python 直接测单次读取耗时 import time start time.time() tool.read_holding_registers(0, 2) print(time.time() - start)如果发现单次工具调用耗时异常比如超过 1 秒需要检查是网络延迟还是设备响应慢。可以用 Modbus Poll 手动读取同一地址对比如果 Modbus Poll 也慢那就是设备或网络问题和 LLM 链路无关。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Modbus TCP 连接失败IP 端口错误、设备未启动、防火墙拦截用 nc 检查 502 端口看设备日志修正 IP 端口放行防火墙确认设备 Modbus Server 使能能 ping 通但 Modbus 扫描不通502 端口未开放或设备 Modbus 未使能用 Modbus Poll 连接测试检查设备通信设置确认端口和从站号读取返回 Illegal Data Address地址超范围或设备映射不一致查看设备寄存器映射表对比协议地址按设备手册调整地址偏移数值完全不对数据类型/字节序/字序配置错误先用 uint16 读取寄存器原始值对照手册修改 data_type、byte_order 或 word_order 参数读取返回 Illegal Data Value寄存器数量或数据类型不匹配检查 count 是否为 1 或 2调整 count确认数据类型对应的寄存器数量LLM 生成的参数不合法提示词约束不足或模型理解偏差打印 tool_calls 的原始调用参数加强 system prompt增加枚举约束写入没有生效写入值超范围、现场设备禁止写入查看功能码错误码工具层增加范围校验检查设备写使能多次调用后连接断开长连接未管理、设备超时查看日志中的断连时间点每次调用检查连接状态失败时重连LLM 回答单位或语言不正确缺少系统提示词约束查看 system prompt要求回答中带上单位和数据来源说明批量任务中一个点位失败卡住无超时设置或异常未捕获查看日志中的异常栈单点读取加 timeout异常单独捕获并记录后继续11. 最佳实践与使用建议第一工具层和 LLM 逻辑要彻底分离。工具层不依赖任何 LLM 内容只做 Modbus 通信和解码。这样即使换 LLM 模型工具层都不用改。我建议把 Modbus 工具设计成一个独立的 Python 包或微服务单独维护版本LLM 应用只通过接口调用。第二所有涉及写寄存器的操作建议单独封装一个工具函数并且让 LLM 在调用前必须经过用户确认。最好在工具函数内部也加一道“最小值/最大值”校验防止异常参数写到设备上。比如某个寄存器只能写 0 或 1工具层就应该在校验时拦截其他值。第三给工具函数增加超时控制。Modbus 设备在异常状态下可能长时间不响应没有超时的话整个流程会被卡死。pymodbus 的 timeout 参数要设置HTTP API 层也要设置超时。推荐设置 3 到 5 秒具体以设备响应时间为准。第四日志记录非常关键。把每次工具调用的参数、原始寄存器返回值、解码结果、耗时全部记录下来。这样设备点位调整或出现数值异常时能快速定位是参数问题还是设备问题。日志格式建议包含时间戳、操作类型、点位名称、原始值、解码值、错误信息。第五现场使用前一定要在模拟器上完整验证。建议准备一个和现场设备点位一致的点位表先用模拟器跑一遍所有读操作确认解码结果与预期一致再切换真实设备。模拟器里可以人为设置异常点位测试工具层在异常情况下的表现。第六工业数据合规。Modbus 设备通常部署在生产环境读取和写入设备数据前必须确认已获得设备归属方和运维方的授权。不要把生产设备的调试接口暴露