企业官网建设流程全解析

这次我们看一个偏工程的问题：LLM 直接解析 Modbus 寄存器，效果很差；更靠谱的落地方式是让 LLM 永远不需要解析原始寄存器。这篇文章从原因、架构、代码到批量任务，把这条链路拆开讲。

先给结论。Modbus 寄存器是裸数据，没有类型、没有单位、没有语义。LLM 再强，让它猜一个 16 位整数值到底是温度、压力还是状态字，本质上都是在做不确定推测。真正可落地的做法是：用确定性代码按设备点表把寄存器解码成结构化数据，再把这些字段交给 LLM 做诊断、解释和告警分析。这就是标题里说的“make sure they never have to”。

这篇文章会带你完成：

• 理清 LLM 不适合直接解码 Modbus 寄存器的原因
• 设计“采集层 + 点表层 + LLM 分析层”的分层架构
• 用 pymodbus 读取寄存器并按点表解码为 JSON
• 把结构化数据交给本地 LLM 服务做设备诊断
• 封装 FastAPI 接口并支持批量设备分析
• 整理常见排查方法和工程化建议

适合正在做工业数据 + LLM、设备运维助手、物联网平台智能诊断的开发者。如果你只是想在本地快速试一下效果，也可以直接按章节把代码跑通。

1. 核心思路与能力速览

这个方案的核心不是“教 LLM 认识 Modbus 协议”，而是把协议解析从 LLM 的职责里彻底拿掉。Modbus 寄存器解析交给代码，LLM 只负责基于结构化数据做语义分析。

从工程角度看，这个分工有几个明显好处：

• 确定性部分不依赖模型能力。寄存器值是 0x41C8，按 float32 解码就是 25.0，代码不会猜错。
• LLM 的上下文是干净的。它看到的是“temperature=25.0°C，fault_bit=1”，而不是一串十六进制数。
• 排查方便。点表错误、字节序错误可以在转换层直接暴露，不需要去翻 LLM 的生成结果。
• 批量任务稳定。每个设备只需要读取少量寄存器，token 消耗可控，接口调用可重试。

2. 为什么 LLM 直接解码 Modbus 寄存器不靠谱

2.1 Modbus 寄存器是“裸数据”

Modbus 协议本身只定义“寄存器”和“读写操作”，不对寄存器里的内容做类型约束。一个保持寄存器是 16 位，可能是无符号整数、有符号整数、某个 32 位浮点数的高半部分，也可能是 16 个布尔状态位的组合。

换句话说，同样读到0x1A2B，在不同设备、不同点表里代表的意义完全不同。LLM 即使知道 Modbus 协议，也无法从单个值判断它是温度、压力、转速还是状态字。

2.2 语义都在点表里

现场设备的含义主要靠“点表”（register map）定义。点表说明哪个地址对应什么字段，是什么数据类型，单位是什么，缩放因子是多少。

举例：

没有点表，LLM 看到0x41C80000只能猜。有了点表，准确解码只是一个循环加一个struct.unpack的事。

2.3 字节序和位字段让猜测更不可靠

同一个 float32 值，寄存器字序和字节序都可能不同。大端字序、小端字序、字节反转，组合起来就有多种解释。LLM 在上下文里看到原始寄存器列表时，几乎不可能稳定判断当前设备的字节序规则。

位字段也同样麻烦。一个状态寄存器里可能同时包含“运行中”“故障”“远程模式”“报警”多个标志位。LLM 把 16 位整数值直接翻译成自然语言，很容易漏位或错位。

所以，结论很直接：协议解析和字节序转换属于确定性工程问题，应该用代码解决。LLM 的价值应该在“看懂结构化数据后如何判断”这一层。

3. 整体架构：先确定性解析，再交给 LLM

3.1 数据流设计

Modbus 从站设备 ↓ 读取保持寄存器 Modbus 采集客户端 ↓ 寄存器整数列表 点表解码器 ↓ JSON 结构化数据 LLM 分析服务 ↓ 自然语言诊断结果 数据库 / 告警系统 / 工单系统

整个链路里只有一个地方出现原始寄存器：采集层到解码器之间。LLM 永远拿不到原始寄存器，它拿到的是已经带字段名、单位、描述的结构化数据。

3.2 各层职责

采集层负责和 Modbus 设备通信，把寄存器和线圈读取为整数列表。这一层要处理连接超时、从站地址、批量读取长度。

转换层也就是点表解码器。它根据设备点表配置，把整数列表转换成带语义的 JSON。这一层要处理数据类型、字节序、字序、缩放因子和位字段。

分析层是 LLM 服务。它接收结构化 JSON，按照系统提示词进行诊断分析，输出设备状态判断和处理建议。这一层不应该关心 Modbus 协议细节。

4. 环境准备与前置条件

建议使用 Python 3.10 及以上的版本，按项目创建虚拟环境。

安装依赖：

pip install pymodbus fastapi uvicorn requests

如果你的 LLM 服务使用 OpenAI SDK，也可以安装：

pip install openai

依赖说明：

• pymodbus：读取 Modbus TCP 寄存器的库，本文示例按 3.x 版本编写。
• FastAPI / uvicorn：用于封装结构化数据分析和批量任务接口。
• requests：调用本地 LLM 服务的 HTTP 接口。
• Modbus 从站或模拟器：用于测试。可以用真实设备，也可以用 Modbus Slave 工具或 pymodbus 官方 server 示例。

需要在本地启动一个 OpenAI 兼容的 LLM 服务，例如 vLLM、Ollama 或 Xinference 之类的部署工具，然后拿到它的 Base URL 和模型名。后面的示例代码里会频繁用到。

5. 代码实现：寄存器读取 + 点表映射

5.1 可选的 Modbus 从站模拟器

如果你手头没有真实的 Modbus 设备，可以先起一个模拟从站。下面代码在 pymodbus 3.5 附近版本中可运行，不同小版本 API 命名可能有差异，以官方示例为准。

import asyncio from pymodbus.server import StartAsyncTcpServer from pymodbus.datastore import ModbusSlaveContext, ModbusSequentialDataBlock def build_data_block(): # 25.0 的 IEEE 754 表示 0x41C80000 temperature = 0x41C80000 # 约 101.3 的 IEEE 754 表示 0x42CA999A pressure = 0x42CA999A data = [ (temperature >> 16) & 0xFFFF, temperature & 0xFFFF, (pressure >> 16) & 0xFFFF, pressure & 0xFFFF, 0x0003, # 状态字：bit0 上电，bit1 故障 0, 0, 0, 0, 0, ] return ModbusSequentialDataBlock(0, data) async def run_server(): block = build_data_block() context = ModbusSlaveContext( di=block, co=block, hr=block, ir=block, ) await StartAsyncTcpServer(context, address=("127.0.0.1", 5020)) if __name__ == "__main__": asyncio.run(run_server())

运行后，这个模拟从站会在本机 5020 端口提供 10 个保持寄存器。寄存器 0-1 组成 float32 温度 25.0，寄存器 2-3 组成 float32 压力约 101.3，寄存器 4 是状态字 0x0003。

如果你已经有真实设备，跳过这一步，直接连接设备 IP 和端口。

5.2 定义设备点表

点表是整个方案的核心。每个字段要写清楚寄存器地址偏移、寄存器数量、数据类型、字节序、字序、单位、描述和位定义。

[ { "name": "device_temperature", "address": 0, "count": 2, "type": "f", "byteorder": ">", "wordorder": "big", "unit": "°C", "description": "设备当前温度", "scale": 1.0 }, { "name": "device_pressure", "address": 2, "count": 2, "type": "f", "byteorder": ">", "wordorder": "big", "unit": "kPa", "description": "运行压力", "scale": 1.0 }, { "name": "running_status", "address": 4, "count": 1, "type": "H", "byteorder": ">", "unit": "", "description": "运行状态字", "bit_fields": [ {"name": "power_on", "bit": 0, "description": "是否上电"}, {"name": "fault", "bit": 1, "description": "故障标志"} ] } ]

这里用type指定 struct 格式符：f是 float32，H是 uint16，i是 int32，I是 uint32，h是 int16。byteorder用>或<。wordorder表示字序，部分 PLC 厂商把 32 位数据的低字放在前一个寄存器，就需要配置为little。

5.3 读取 Modbus 寄存器

用 pymodbus 客户端读取保持寄存器：

from pymodbus.client import ModbusTcpClient def read_registers(host: str, port: int, start: int, count: int, slave: int = 1): client = ModbusTcpClient(host, port=port, timeout=3) try: if not client.connect(): raise RuntimeError("连接 Modbus 从站失败") result = client.read_holding_registers( address=start, count=count, slave=slave, ) if result.isError(): raise RuntimeError(f"读取寄存器失败: {result}") return result.registers finally: client.close()

读取成功后返回一个整数列表。比如读取 10 个寄存器，会得到类似[16840, 0, 17178, 39322, 3, 0, 0, 0, 0, 0]这样的结果。注意这里地址偏移是协议地址，不是 PLC 面板上显示的 40001，两者之间存在40001 -> 0的偏移关系。

5.4 按点表解码为结构化 JSON

解码器负责把整数列表变成结构化 JSON。下面的代码根据字段配置依次做字序调整、struct 解码、缩放和位字段拆分。

import json import struct from typing import Any, List, Dict def decode_one(regs: List[int], field: Dict[str, Any]): # 将每个寄存器按大端转成 2 字节 raw = b"".join(r.to_bytes(2, byteorder="big", signed=False) for r in regs) # 字序调整，处理部分 PLC 的低字在前问题 if field.get("wordorder") == "little": words = [raw[i:i + 2] for i in range(0, len(raw), 2)] words.reverse() raw = b"".join(words) fmt = field["byteorder"] + field["type"] size = struct.calcsize(fmt) if len(raw) < size: raise ValueError(f"寄存器数量不足，无法按 {fmt} 解码") value = struct.unpack(fmt, raw[:size])[0] if field.get("scale"): value = value * field["scale"] # 位字段拆分，仅用于整数类型的字段 if field.get("bit_fields") and isinstance(value, int): bit_values = {} for bf in field["bit_fields"]: bit_values[bf["name"]] = { "value": (value >> bf["bit"]) & 1, "description": bf.get("description", ""), } value = {"raw": value, "bits": bit_values} return value def decode_registers(registers: List[int], point_map: List[Dict[str, Any]]): decoded = {} for field in point_map: addr = field["address"] count = field["count"] if addr + count > len(registers): raise ValueError(f"字段 {field['name']} 需要 {count} 个寄存器，但读取范围不足") regs = registers[addr:addr + count] value = decode_one(regs, field) decoded[field["name"]] = { "value": value, "unit": field.get("unit", ""), "description": field.get("description", ""), } return decoded

把点表定义成 Python 常量：

POINT_MAP = [ { "name": "device_temperature", "address": 0, "count": 2, "type": "f", "byteorder": ">", "wordorder": "big", "unit": "°C", "description": "设备当前温度", "scale": 1.0, }, { "name": "device_pressure", "address": 2, "count": 2, "type": "f", "byteorder": ">", "wordorder": "big", "unit": "kPa", "description": "运行压力", "scale": 1.0, }, { "name": "running_status", "address": 4, "count": 1, "type": "H", "byteorder": ">", "unit": "", "description": "运行状态字", "bit_fields": [ {"name": "power_on", "bit": 0, "description": "是否上电"}, {"name": "fault", "bit": 1, "description": "故障标志"}, ], }, ]

主流程测试：

if __name__ == "__main__": registers = read_registers("127.0.0.1", 5020, 0, 10) print("原始寄存器:", registers) decoded = decode_registers(registers, POINT_MAP) print(json.dumps(decoded, ensure_ascii=False, indent=2))

预期输出类似：

{ "device_temperature": { "value": 25.0, "unit": "°C", "description": "设备当前温度" }, "device_pressure": { "value": 101.3, "unit": "kPa", "description": "运行压力" }, "running_status": { "value": { "raw": 3, "bits": { "power_on": {"value": 1, "description": "是否上电"}, "fault": {"value": 1, "description": "故障标志"} } }, "unit": "", "description": "运行状态字" } }

到这里，LLM 需要的信息已经全部变成了可读字段。后面的调用不再涉及任何十六进制内容。

6. 让 LLM 基于结构化数据做分析

6.1 Prompt 设计

LLM 的 System Prompt 要明确告诉它：你拿到的是已经解码好的设备数据，字段名、数值、单位和描述都是可信的，不要怀疑这些字段的含义，也不要试图去还原原始寄存器。

用户 Prompt 直接放 JSON 结构化数据即可。

SYSTEM_PROMPT = ( "你是工业设备智能诊断助手。用户会给你一批已经解码好的设备数据，" "字段名、数值、单位和描述都是可信的。你不需要猜测寄存器含义，" "只需要基于这些数据判断设备状态、发现异常并给出处理建议。" "回答要简洁、专业，使用中文。" ) def build_user_prompt(decoded: dict) -> str: return "以下是设备当前运行数据：\n" + json.dumps(decoded, ensure_ascii=False, indent=2)

6.2 调用 LLM 接口

假设你本机部署了一个 OpenAI 兼容的 LLM 服务，可以通过 HTTP 调用。

import requests def analyze_device_data( decoded: dict, base_url: str = "http://127.0.0.1:8000/v1", model: str = "your-local-model", ): payload = { "model": model, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": build_user_prompt(decoded)}, ], "temperature": 0.2, } resp = requests.post( f"{base_url}/chat/completions", json=payload, timeout=60, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]

如果使用的是 OpenAI SDK，也可以写成：

from openai import OpenAI client = OpenAI(base_url="http://127.0.0.1:8000/v1", api_key="EMPTY") resp = client.chat.completions.create( model="your-local-model", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": build_user_prompt(decoded)}, ], temperature=0.2, ) print(resp.choices[0].message.content)

6.3 增加稳定输出格式

如果希望 LLM 的结果能被下游系统直接消费，可以在 Prompt 里要求输出固定 JSON。

SYSTEM_PROMPT_JSON = ( "你是工业设备智能诊断助手。你会收到已经解码好的设备数据，" "请基于数据输出 JSON，包含三个字段：" "status（normal/warning/fault）、reason（判断依据）、action（处理建议）。" "只输出 JSON，不要输出多余说明。" )

这样下游告警系统可以直接解析结果，不需要再做一轮自然语言后处理。

7. 接口 API 与批量任务

7.1 用 FastAPI 封装结构化分析接口

把“寄存器读取 -> 解码 -> LLM 分析”包装成一个 HTTP 接口，方便外部系统调用。

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI(title="Modbus LLM Analysis Gateway", version="0.1.0") class DeviceSnapshot(BaseModel): device_id: str registers: list[int] point_map: list[dict] @app.post("/analyze") def analyze_device(snapshot: DeviceSnapshot): decoded = decode_registers(snapshot.registers, snapshot.point_map) analysis = analyze_device_data(decoded) return { "device_id": snapshot.device_id, "decoded": decoded, "analysis": analysis, }

启动服务：

uvicorn app:app --host 0.0.0.0 --port 8001

这里需要注意：FastAPI 服务把registers和point_map都作为请求参数，调用方负责先拿到寄存器数据。如果希望接口直接读设备，可以把host、port、slave也放到请求模型里，接口内部调用read_registers。

7.2 批量设备分析

批量任务建议采用目录配置的方式。每个设备一个 JSON 文件，包含连接信息和点表。

设备配置示例：

{ "device_id": "pump_01", "host": "127.0.0.1", "port": 5020, "slave": 1, "start": 0, "count": 10, "point_map": [] }

批量处理脚本：

import glob import json import os from concurrent.futures import ThreadPoolExecutor, as_completed def process_device(device_file: str): with open(device_file, "r", encoding="utf-8") as f: cfg = json.load(f) registers = read_registers( cfg["host"], cfg["port"], cfg.get("start", 0), cfg.get("count", 10), slave=cfg.get("slave", 1), ) decoded = decode_registers(registers, cfg["point_map"]) analysis = analyze_device_data(decoded) return { "device_id": cfg.get("device_id", os.path.basename(device_file)), "decoded": decoded, "analysis": analysis, "ok": True, } def run_batch(device_dir: str, max_workers: int = 2): files = glob.glob(os.path.join(device_dir, "*.json")) results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = {pool.submit(process_device, f): f for f in files} for future in as_completed(futures): device_file = futures[future] try: results.append(future.result()) except Exception as exc: results.append({ "device_file": device_file, "error": str(exc), "ok": False, }) return results

批量处理时，max_workers不要开太大。LLM 推理服务通常有并发限制，同时打太多请求很容易把服务打挂。建议先从 1 到 2 开始，观察延迟再逐步增加。

7.3 日志与失败重试

批量任务必须加日志。每个设备的读取时间、解码结果、LLM 响应时间、状态码都要记录下来。失败时要有重试机制，例如对超时的请求重试两次。

import time def analyze_with_retry(decoded: dict, retries: int = 2): for attempt in range(retries): try: return analyze_device_data(decoded) except Exception as exc: if attempt == retries - 1: raise wait = 2 ** attempt time.sleep(wait) raise RuntimeError("analyze failed after retries")

这里的指数退避策略比较基础，但已经能应付大多数瞬时超时问题。

8. 性能与资源占用观察

Modbus 轮询本身非常轻量。一次读取 10 个保持寄存器的 TCP 请求只有几十字节，响应也很快。性能瓶颈基本都集中在 LLM 推理环节。

建议在验证阶段重点观察三件事。

第一，Modbus 采集频率。普通设备状态监视 1 秒到 5 秒轮询一次已经足够，不需要追求毫秒级。如果设备点表很长，建议按功能码分组读取，不要一次读很大的连续区间。

第二，LLM 请求