1. 为什么我要用AI原生思路重写一个Modbus仿真器做过工业自动化上位机开发的人都有一个共同的痛点手头没有PLC没有仪表没有变频器但代码得先跑起来。你总不能每次调试都跑到车间去蹭设备更不可能让甲方把产线停下来等你调通讯。所以几乎每个搞工控软件的人早晚都会自己写一个Modbus仿真器。我以前也写过用Python的pymodbus搭一个server几百行代码能读能写凑合能用。但问题是每次需求一变——比如要模拟一个带异常响应的从站、要模拟寄存器值随时间变化的传感器、要模拟多从站并发——我就得回去改代码改完再测测完再改。这东西本质上还是一个我写死的程序而不是一个能听懂人话的工具。这次我换了个思路既然现在大模型写代码已经足够靠谱为什么不做一个AI原生的Modbus仿真器所谓AI原生不是说在仿真器里塞一个聊天框而是让整个仿真器的配置、行为定义、异常注入、测试用例生成全部由自然语言驱动。你说模拟一个温度传感器地址1保持寄存器40001存温度每秒随机波动在20到30度之间超过28度时线圈00001置位报警它就能直接把这个从站跑起来。这篇文章我会把整个项目的设计思路、核心实现、踩过的坑、以及实际调试中的经验完整拆开讲。适合两类人看一类是正在做工控上位机、SCADA、物联网网关开发需要频繁调试Modbus通讯的工程师另一类是想了解AI原生应用到底怎么落地的人——不是套壳聊天而是真正把模型能力嵌进工具的核心链路里。先说清楚这个仿真器能干什么支持Modbus TCP和Modbus RTU over TCP支持多从站并发支持线圈、离散输入、保持寄存器、输入寄存器四类数据区支持通过自然语言定义数据行为常量、随机、正弦波、斜坡、自定义表达式支持异常码注入支持请求日志和报文抓取。下面逐层拆。2. 整体架构设计与技术选型思路2.1 为什么选Python而不是Go或Rust工控领域很多人第一反应是用C#或者C因为上位机生态成熟。但这个项目的核心诉求是快速迭代AI集成Python的优势太明显了。pymodbus这个库虽然有些历史包袱但它的server端实现足够稳定而且允许你自定义数据块和请求处理逻辑。更重要的是大模型生成Python代码的准确率明显高于生成C或Rust这在AI原生开发里是个决定性因素——你让模型生成一段带生命周期管理的Rust Modbus server它大概率会给你整出借用检查的错误。性能方面Modbus本身就不是高吞吐协议。一个从站每秒几百次请求已经算高频了Python完全扛得住。如果你真要模拟上千个从站那确实该换语言但那种场景也不是仿真器的典型用途。2.2 AI原生到底体现在哪一层这是我最想强调的一点。很多号称AI原生的工具实际上只是加了个对话框你把需求告诉它它给你生成一段配置然后你拿着配置去用传统工具。这不叫原生这叫外挂。我的设计是把AI放在三个关键位置第一配置生成层。用户用自然语言描述从站行为模型输出结构化的数据点定义。这一层的关键是设计好中间表示IR不能让模型直接生成代码否则不可控。第二行为表达式层。数据点的值怎么变我用了一个受限的表达式DSL模型负责把每秒上升0.5然后到100归零翻译成ramp(0.5, 0, 100, resetTrue)这样的表达式。DSL是白名单的模型只能组合已有函数不能凭空造。第三测试用例生成层。这是我觉得最有价值的部分。你告诉模型帮我测试上位机在从站返回异常码06时的重试逻辑它直接生成一串Modbus请求序列注入到仿真器里执行然后比对上位机的响应。注意不要让模型直接操作socket或者直接写寄存器。所有模型输出必须经过一层校验和沙箱。我吃过这个亏早期版本模型生成了一段把所有寄存器清零的初始化代码结果每次重连都触发调试了半天才发现。2.3 模块划分整个项目我分成五个模块协议层基于pymodbus的server处理TCP和RTU over TCP的帧解析。数据模型层定义从站、数据区、数据点的数据结构以及表达式的求值引擎。AI适配层封装模型调用负责自然语言到IR的转换以及IR的校验。调度层管理多个从站的生命周期处理数据点的定时更新。观测层请求日志、报文抓取、异常统计方便调试。这五层之间通过明确的接口通信AI适配层只产出IR不碰协议层。这样即使模型抽风最坏情况也只是配置生成失败不会把整个仿真器搞崩。3. 核心细节解析与实操要点3.1 数据点定义的中间表示设计中间表示是整个项目的骨架。我设计的是一个JSON结构一个从站长这样{ unit_id: 1, mode: tcp, port: 5020, points: [ { area: holding_register, address: 0, type: uint16, behavior: { kind: expression, expr: 20 5 * sin(t / 10), update_interval_ms: 1000 } }, { area: coil, address: 0, behavior: { kind: expression, expr: holding_register[0] 28 ? 1 : 0, update_interval_ms: 1000 } } ] }这里有几个设计决策值得说。为什么用表达式而不是回调函数因为表达式可以被序列化、可以被模型生成、可以被校验。回调函数是代码代码没法安全地让模型生成。表达式引擎我用了Python的ast模块做白名单解析只允许数学运算、比较、三元表达式和预置函数sin、cos、ramp、random、step等。任何属性访问、导入、函数调用白名单之外的一律拒绝。为什么update_interval_ms放在数据点级别因为不同数据点的更新频率差异很大。温度可能1秒更新一次而一个累计产量计数器可能10秒才变一次。如果统一用一个全局tick要么浪费CPU要么精度不够。地址从0还是从1开始这是Modbus的经典坑。协议层面地址是从0开始的但很多PLC文档写的是40001这种1-based的逻辑地址。我在IR里统一用0-based的协议地址然后在AI适配层做转换——用户说40001模型负责转成holding_register区域的地址0。这个转换规则我在prompt里写得很死避免模型自由发挥。3.2 表达式引擎的实现要点表达式引擎是这个项目里技术含量最高的部分。我用了ast.parse加自定义的NodeVisitor核心逻辑是这样import ast import math import random ALLOWED_FUNCS { sin: math.sin, cos: math.cos, abs: abs, min: min, max: max, random: random.random, ramp: None, # 特殊处理 step: None, } class ExprValidator(ast.NodeVisitor): def visit_Call(self, node): if not isinstance(node.func, ast.Name): raise ValueError(只允许简单函数调用) if node.func.id not in ALLOWED_FUNCS: raise ValueError(f函数 {node.func.id} 不在白名单) self.generic_visit(node) def visit_Attribute(self, node): raise ValueError(禁止属性访问) def visit_Import(self, node): raise ValueError(禁止导入)求值的时候我把t当前时间单位秒、holding_register一个可索引的寄存器视图、coil等作为上下文变量注入。这样表达式就能引用其他数据点的当前值实现联动。实操心得ramp和step这类有状态的函数不能每次求值都重新算。我的做法是给每个数据点维护一个状态字典ramp函数从状态里读上次的值和方向。这个状态在从站重启时清零但数据点更新时不重置。这个细节如果不注意斜坡函数会变成每次从起点开始看起来就像没生效。3.3 AI适配层的prompt工程让模型稳定输出符合IR的JSON比想象中难。我试过几种方案第一种是直接给JSON Schema让模型按schema输出。问题是模型经常在数值类型上犯错比如把update_interval_ms写成字符串1000。第二种是给few-shot示例。效果好一些但示例一多token消耗大而且模型会过度模仿示例遇到新场景就懵。最后我采用的是两阶段方案第一阶段让模型输出一个宽松的自然语言结构化描述第二阶段用一个小的校验和修复函数把它转成严格IR。修复函数处理常见的类型错误、地址转换、区域名归一化。这样模型只需要关注意图理解格式问题交给代码兜底。prompt的核心部分大概是这样你是一个Modbus从站配置助手。用户会用自然语言描述从站行为。 你需要输出JSON字段包括unit_id、points数组。 每个point包含area只能是coil/discrete_input/holding_register/input_register、 address0-based协议地址、type、behavior。 behavior的expr只能使用这些函数sin, cos, abs, min, max, random, ramp, step。 可用变量t秒、holding_register[i]、coil[i]、discrete_input[i]、input_register[i]。 逻辑地址转换规则00001-09999对应coil的0-999810001-19999对应discrete_input 30001-39999对应input_register40001-49999对应holding_register。这个prompt我迭代了十几版最大的教训是不要试图让模型一次做太多事。早期我让模型同时处理生成配置和解释配置结果它经常在JSON里塞注释。后来拆成两个独立的调用各自专注稳定性大幅提升。4. 实操过程与核心环节实现4.1 环境搭建与依赖安装先把环境跑起来。我用的是Python 3.11pymodbus选的3.6版本。这里有个坑要提醒pymodbus 3.x和2.x的API差异很大网上很多教程还是2.x的写法直接抄会报错。python -m venv venv source venv/bin/activate # Windows下是 venv\Scripts\activate pip install pymodbus3.6.6 pip install openai # 或者你用的任何模型SDK pip install pydantic # 用于IR校验pymodbus 3.6的server启动方式和2.x完全不同。2.x是StartTcpServer3.x改成了ModbusTcpServer类加serve_forever。我一开始照着旧文档写卡了半小时才发现API变了。4.2 从站启动的核心代码启动一个TCP从站的骨架大概是这样from pymodbus.server import ModbusTcpServer from pymodbus.datastore import ( ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext, ) def build_context(ir_config): # 为每个从站构建数据块 contexts {} for slave in ir_config[slaves]: # 初始化四类数据区大小按最大地址1 coils ModbusSequentialDataBlock(0, [0] * 10000) discretes ModbusSequentialDataBlock(0, [0] * 10000) holdings ModbusSequentialDataBlock(0, [0] * 10000) inputs ModbusSequentialDataBlock(0, [0] * 10000) context ModbusSlaveContext( cocoils, didiscretes, hrholdings, irinputs ) contexts[slave[unit_id]] context return ModbusServerContext(slavescontexts, singleFalse) async def run_server(context, port): server ModbusTcpServer(context, address(0.0.0.0, port)) await server.serve_forever()注意singleFalse这个参数。如果设成True所有unit_id都会路由到同一个context多从站就失效了。这个坑我在测试多从站并发时踩过现象是无论请求哪个unit_id返回的都是同一个从站的数据。4.3 数据点的定时更新数据点的更新我用了一个独立的asyncio任务每个数据点根据自己的update_interval_ms决定是否更新。核心逻辑async def update_loop(context, points): last_update {i: 0 for i in range(len(points))} start_time time.time() while True: now time.time() t now - start_time for i, point in enumerate(points): interval point[behavior][update_interval_ms] / 1000 if now - last_update[i] interval: value evaluate(point[behavior][expr], t, context) write_point(context, point, value) last_update[i] now await asyncio.sleep(0.05) # 50ms的调度精度50ms的调度精度是个折中。再低会浪费CPU再高会导致快速变化的数据点更新不及时。实测下来对于1秒更新间隔的数据点50ms的调度误差完全可接受。注意evaluate函数里如果表达式抛异常比如除零不要让整个更新循环崩掉。我的做法是捕获异常把该数据点标记为错误状态值保持上一次的有效值同时记录一条日志。这样即使某个表达式写错了其他数据点还能正常工作。4.4 异常注入的实现异常注入是仿真器最有价值的功能之一。真实设备在异常情况下会返回异常码比如请求了不存在的寄存器返回02非法数据地址从站忙返回06从站忙。上位机的异常处理逻辑如果没测过上线就是事故。我的实现方式是在请求处理链路上加一个拦截器。pymodbus允许你自定义ModbusSlaveContext的getValues和setValues方法我在这两个方法里检查异常注入规则class InjectableSlaveContext(ModbusSlaveContext): def __init__(self, *args, injection_rulesNone, **kwargs): super().__init__(*args, **kwargs) self.injection_rules injection_rules or [] def getValues(self, fc_as_hex, address, count1): for rule in self.injection_rules: if rule[fc] fc_as_hex and rule[address] address: if rule[mode] exception: raise ModbusException(rule[code]) elif rule[mode] delay: time.sleep(rule[delay_ms] / 1000) return super().getValues(fc_as_hex, address, count)异常注入规则同样可以由AI生成。你说当上位机读取40001时前三次返回从站忙第四次正常返回模型会生成一个带计数器的规则。这个计数器状态我放在规则对象里每次触发递减。4.5 报文抓取与日志调试Modbus最痛苦的就是不知道线上到底发了什么。我在server外面包了一层把每个请求的原始PDU和响应都记下来。pymodbus的日志级别调到DEBUG能看到一些但格式不友好。我自己写了一个framer的wrapper在decode和encode两个环节抓数据。日志格式我设计成一行一条方便grep[2024-01-15 10:23:45.123] unit1 fc03 addr0 count2 req01 03 00 00 00 02 resp01 03 04 00 14 00 1E这样你一眼就能看出请求了什么、返回了什么。如果响应是异常resp会是01 83 02这种83是03加0x8002是异常码。5. 常见问题与排查技巧实录5.1 连接上了但读不到数据这是最高频的问题。排查顺序我总结成一张表现象可能原因排查方法连接成功读返回全0数据点没配置或表达式求值为0检查IR里points是否为空表达式是否正确连接成功读返回异常码02地址超出数据块范围检查数据块大小是否覆盖了请求地址连接成功读超时从站没在更新循环里检查update_loop是否启动是否有异常被吞掉连接被拒绝端口被占用或没监听netstat看端口检查server是否启动成功多从站时数据串了single参数设成了True改成False检查unit_id映射我遇到最隐蔽的一次是读返回全0查了半天发现是表达式里用了holding_register[0]引用自己导致求值时读到的是初始值0然后写回去还是0死循环。这种自引用要在校验阶段就拦住。5.2 表达式求值性能问题如果你的数据点很多比如几百个每个都带复杂表达式50ms的调度周期可能跑不完一轮。我实测过纯Python的ast求值一个简单表达式大概20微秒复杂的三元加函数调用大概100微秒。500个数据点就是50毫秒刚好卡在调度周期上。优化方案有两个一是把表达式编译成Python的code对象缓存起来避免每次重新parse二是对不常变的数据点降低更新频率。我用了第一个方案compile(expr, expr, eval)之后缓存求值速度能提升3到5倍。实操心得不要用eval直接执行用户表达式哪怕你做了ast校验。eval的作用域控制很麻烦容易出安全问题。用compile加受限的eval并且globals传空字典locals只传白名单变量这样最稳妥。5.3 AI生成的配置不准确模型偶尔会把40001理解成地址40001而不是逻辑地址。我的解决办法是在prompt里加一个转换示例并且在修复函数里做二次校验如果holding_register的地址大于9999大概率是逻辑地址没转换自动减去40001。还有一种情况是模型把每秒增加1翻译成t * 1这在t是秒的时候是对的但如果用户想要的是每次读取时增加1那就错了。这种语义歧义我在prompt里明确要求模型区分基于时间的连续变化和基于事件的离散变化前者用t后者用状态函数。5.4 多从站并发的资源竞争多个从站共享一个事件循环如果某个从站的更新循环里有阻塞操作比如time.sleep会拖慢所有从站。我早期在异常注入的delay里用了time.sleep结果一个从站的延迟注入把其他从站全卡住了。后来改成await asyncio.sleep问题解决。这个坑的教训是在异步环境里任何同步阻塞都是灾难。pymodbus的server本身是异步的你的所有扩展代码也必须是异步的。5.5 常见问题速查表问题根因解决表达式报NameError引用了未定义的变量检查变量名拼写确认上下文注入了该变量ramp函数不生效状态没持久化检查状态字典是否在数据点级别维护异常注入只生效一次计数器没重置确认规则的生命周期是否需要按连接重置日志里看不到请求日志级别或framer没挂上检查wrapper是否真的被调用从站重启后数据丢失数据块没持久化如果需要持久化把数据块存到文件或数据库6. 这套方案还能怎么扩展写到这里核心的东西基本讲完了。最后分享几个我后续打算做的扩展方向也给正在做类似项目的朋友一些思路。第一个是场景回放。把真实设备抓到的报文录下来导入仿真器让它按时间轴重放。这样你就能在没有设备的情况下复现线上问题。这个功能对排查偶发通讯故障特别有用。第二个是协议模糊测试。让AI生成大量畸形的Modbus请求测试上位机的健壮性。比如超长PDU、非法功能码、边界地址。很多上位机在这种输入下会直接崩提前发现能省很多事。第三个是与CI集成。把仿真器作为测试环境的一部分每次上位机代码提交自动跑一遍Modbus通讯测试用例。用例同样由AI根据接口文档生成覆盖正常流程和异常流程。我个人在实际操作中的体会是AI原生开发的关键不在于模型多强而在于你把模型的输出约束得多好。模型是个很有创造力但很不靠谱的实习生你得给它设计好工位、划好红线、准备好兜底。这个Modbus仿真器我前后迭代了大概三周其中一半时间花在IR设计和校验上真正写协议代码的时间反而不多。但正是这部分不性感的工作让整个工具从玩具变成了能天天用的东西。最后再分享一个小技巧如果你也在做类似的AI原生工具一定要把每次模型调用的输入输出都存下来。我存了一个月的数据后来发现某些场景下模型的表现有规律性的偏差拿这些数据去微调prompt效果比拍脑袋改好得多。数据是AI原生应用最宝贵的资产别浪费。