1. 这句话不是噱头而是工程师之间心照不宣的底层共识“PLC 逻辑和 AI Agent 其实是同一种东西”——第一次听到这句话时我正在调试一台西门子 S7-1200 控制的灌装线现场变频器抖动导致编码器反馈跳变梯形图里三个互锁条件反复触发我一边用 TIA Portal 在线监控 DB 块数据一边在手机上刷到某技术群有人发了这张图左边是博途里画出的标准三段式状态机启动→运行→停机右边是 LangGraph 里用StateGraph定义的 Agent 工作流analyze → plan → execute → reflect。两者的节点形状、箭头走向、状态守卫条件Guard Condition的写法几乎一模一样。这不是巧合也不是强行类比。这是控制工程与智能系统在抽象层面上的自然收敛。PLC 不是“老古董”AI Agent 也不是“新玄学”。它们共享同一套建模语言离散事件驱动 确定性状态迁移 周期性环境感知。你写的每一段梯形图本质上都在定义一个有限状态机FSM你配置的每一个 LangChain Tool本质上都在封装一个可调用的确定性函数模块你设置的 PLC 扫描周期比如 10ms和你给 LLM 设置的max_tokens或timeout5s解决的是同一个问题如何在有限资源下对动态世界做出及时、可控、可验证的响应。这个观点对谁最有价值第一类是刚入行的自动化工程师还在背“常开触点”“上升沿触发”符号却看不懂为什么老师傅总说“程序要像流水线一样有节拍”第二类是转行做 AI 工程师的开发者能调通 LangChain但一遇到 Agent 在真实产线里“卡住不动”就抓瞎第三类是工厂里的设备主管既被“AI 赋能”的PPT轰炸又被“PLC 突然停机”的报警单压垮急需一条能打通两边的认知桥梁。这篇文章不讲理论推导只讲我在汇川 H3U PLC 上跑通 PID 温控逻辑、又在同一台工控机上用 FastAPI 搭起 LangGraph Agent 的实操路径——从硬件 IO 点映射到 Agent State Schema从梯形图的扫描周期对齐到 Agent 的invoke()调用节奏从 PLC 的强制输出Force Output调试法迁移到 Agent 的step_by_step回溯机制。所有代码、配置、接线图、状态迁移表都来自我去年在东莞一家注塑厂的真实项目。2. 核心设计思想把“状态”作为唯一通用接口2.1 为什么状态机是两者共同的DNA先抛开术语。想象一个最简单的场景十字路口红绿灯。PLC 实现用三个定时器T0 红灯 30s、T1 黄灯 3s、T2 绿灯 27s配合一个步进寄存器M100 启动、M101 运行、M102 停止梯形图里每个支路都带自锁和互锁扫描周期一到CPU 就检查当前 M100 是否为 1是则置位 Q0.0红灯同时复位 Q0.1绿灯……整个过程不依赖任何外部计算只靠内部标志位和时间条件驱动。AI Agent 实现定义一个TrafficLightState类含current_phase: Literal[red, yellow, green]、elapsed_time: int、next_phase_at: int字段Agent 的run()方法每次被调用就检查elapsed_time next_phase_at满足则更新current_phase并重置计时器再调用update_lights()函数控制 GPIO 输出。表面看一个是继电器逻辑一个是 Python 类方法。但深挖一层两者都维护一个显式的、结构化的状态容器PLC 的 M 区/DB 块Agent 的 State 对象两者都通过预定义的转移规则梯形图支路 / Agent 的add_conditional_edges决定下一状态两者都依赖周期性触发PLC 扫描周期 / Agent 的调度器或 HTTP 轮询来推进状态演进两者都要求状态迁移必须是确定性的——同一输入、同一当前状态必须产生同一输出和下一状态否则系统不可控。这就是为什么“状态机”不是一种实现技巧而是控制系统的第一性原理。PLC 的梯形图、指令表IL、功能块图FBD本质都是状态机的不同图形化语法糖而 LangGraph 的StateGraph、AutoGen 的GroupChatManager、甚至 Spring AI 的AgentExecutor也都是在用不同编程范式封装状态机。我见过太多人把 AI Agent 当成“会自己思考的黑盒”结果部署后发现它在产线上反复执行错误动作——根本原因就是没把它当成一个需要严格定义状态边界的 PLC 来对待。2.2 扫描周期 vs Agent 调用节奏时间维度的硬约束PLC 的扫描周期Scan Cycle常被误解为“CPU 多快”其实它定义的是系统对物理世界的采样-决策-执行闭环的最大延迟。以台达 DVP-ES3 为例典型扫描周期 10ms意味着每 10msPLC 读取一次所有输入端子I0.0~I0.7的电平执行一遍用户程序梯形图根据当前输入和内部状态计算出所有输出端子Q0.0~Q0.7的新值将计算结果写入输出锁存器驱动外部继电器或变频器。这个 10ms 是铁律。你不能指望它“更快一点去处理紧急停止信号”因为整个架构就是围绕这个周期设计的——中断Interrupt只是特例且需硬件支持。AI Agent 的“扫描周期”在哪里不在代码里而在你的调度层。常见错误是用while True: agent.invoke(state); time.sleep(0.1)—— 这相当于把 PLC 扫描周期设成 100ms但实际执行可能因 LLM 推理波动在 200ms~2s 之间完全失控用 Webhook 触发 Agent —— 这相当于把 PLC 改成“有人按按钮才扫描”无法应对连续过程如温度 PID 调节。正确做法是主动控制节奏。我在东莞项目中用 FastAPI 写了一个POST /agent/tick接口其核心逻辑是app.post(/agent/tick) def tick_agent(): # 1. 从 Redis 读取最新传感器数据模拟 PLC 输入扫描 sensor_data redis.hgetall(sensor:latest) # 2. 构建当前状态对应 PLC 的 DB 块 current_state { temperature: float(sensor_data.get(temp, 25.0)), setpoint: get_setpoint_from_hmi(), # 从 HMI 读设定值 motor_speed: int(sensor_data.get(speed, 0)), current_phase: redis.get(state:phase) or idle } # 3. 调用 Agent模拟 PLC 执行用户程序 result agent.invoke(current_state) # 4. 将输出写入硬件模拟 PLC 输出刷新 if result.get(output_action) start_motor: write_to_plc_output(Q0.0, 1) # 启动电机 elif result.get(output_action) adjust_pid: send_pid_params_to_plc(result[pid_params]) # 下发 PID 参数 return {status: ok, next_tick_in_ms: 50} # 强制 50ms 周期关键点在于tick接口本身不耗时Redis 读写 1msLLM 推理在后台异步完成前端HMI 或 SCADA按固定间隔如 50ms轮询此接口确保节奏稳定所有传感器数据通过 Redis 缓存避免每次调用都直连硬件——这就像 PLC 的输入映像区Input Image Table保证一个扫描周期内输入值不变。提示不要试图让 LLM 直接控制 IO。我的方案中Agent 只负责“决策”输出action和params真正的“执行”由 FastAPI 的write_to_plc_output()函数完成该函数通过 Modbus TCP 协议与 PLC 通信。这完全复刻了 PLC 的“程序逻辑层”与“IO 驱动层”分离的设计哲学。2.3 梯形图符号与 Agent 工具链从物理触点到数字函数梯形图的符号体系是工业界百年沉淀的领域特定语言DSL。我们来逐个解码它与 AI Agent 工具Tool的对应关系梯形图符号物理含义对应 Agent Tool关键特性常开触点输入端子 I0.0 为高电平时闭合常闭触点/输入端子 I0.0 为低电平时闭合定时器TON通电延时到达设定值后置位wait_for_seconds(seconds5.0)阻塞式等待超时返回 True计数器CTU加计数达到设定值后置位increment_counter(namebatch_count, target100)状态持久化原子操作置位线圈S置位指定输出位自锁set_output(pinQ0.0, value1)写入硬件改变物理状态复位线圈R复位指定输出位reset_output(pinQ0.0)写入硬件恢复初始状态看到这里你应该明白所谓“AI Agent 搭建”本质就是用 Python 函数重新实现一套符合你产线需求的“梯形图符号库”。我在项目中封装了 12 个核心 Tool全部基于 Modbus TCP 协议覆盖汇川 H3U PLC 的所有常用功能# 示例封装一个“带互锁的电机启停”Tool tool def motor_control( action: str start, interlock_check: bool True ) - str: 控制电机启停内置硬件互锁逻辑 - action: start or stop - interlock_check: 是否检查急停、过载等互锁条件 if interlock_check: # 读取所有互锁输入点I0.1 急停、I0.2 过载、I0.3 门禁 interlocks read_modbus_coils([1, 2, 3]) # 地址从0开始 if any(interlocks): return Interlock active, cannot execute if action start: write_modbus_coil(0, 1) # Q0.0 1 return Motor started else: write_modbus_coil(0, 0) # Q0.0 0 return Motor stopped这个函数就是梯形图里那个经典的“启保停”电路Start-Keep-Stop的数字孪生。它把物理世界的互锁条件急停按钮按下即 I0.10、执行动作Q0.0 输出、状态反馈返回字符串全部封装在一个可测试、可复用的单元里。当你用 LangGraph 把motor_control、pid_tune、alarm_log这些 Tool 连接起来时你画的不是流程图而是一张数字时代的梯形图。3. 实操拆解从 PLC 梯形图到 AI Agent 的完整映射3.1 案例背景注塑机温控系统升级原始系统汇川 H3U PLC K型热电偶 SSR 固态继电器用传统 PID 指令PID_Compact控制料筒温度。问题温度波动大±5℃尤其在加料阶段PID 参数固定无法适应不同塑料材质ABS/PC/PP的热容差异故障报警仅靠指示灯无历史记录维修靠经验。升级目标保留原有 PLC 硬件新增一台工控机运行 AI Agent实现根据实时温度曲线自动优化 PID 参数识别异常升温模式如加热棒短路提前报警生成每批次温度日志供质量追溯。关键约束不能改动 PLC 程序客户拒绝停机重刷Agent 必须与现有 HMI威纶通 MT8071iE共存所有通信走 Modbus TCP端口号 502PLC 默认。3.2 硬件层对接让 Agent “看见” PLC 的世界第一步明确 PLC 的数据映射。汇川 H3U 的 Modbus 地址规划如下需在 PLC 程序中预先分配功能Modbus 地址数据类型说明当前温度40001FLOAT读取K型热电偶经 PLC AD 转换后值设定温度40003FLOAT写入HMI 修改后同步至此加热输出百分比40005INT16读取PID 指令输出值0~100当前材质代码40007INT16读取HMI 选择 ABS1, PC2, PP3PID_P 参数40009FLOAT写入Agent 下发的 P 值PID_I 参数40011FLOAT写入Agent 下发的 I 值PID_D 参数40013FLOAT写入Agent 下发的 D 值报警代码40015INT16写入0正常1超温2升温过慢3传感器断线注意Modbus 地址从 1 开始编号但实际通信时40001 对应功能码 03读保持寄存器的地址 0。很多初学者在这里栽跟头——用pymodbus读 40001 时address0不是address40001。我在工控机上用 Python 实现了一个PlcInterface类封装所有读写操作from pymodbus.client import ModbusTcpClient from pymodbus.payload import BinaryPayloadDecoder from pymodbus.constants import Endian class PlcInterface: def __init__(self, host192.168.1.10, port502): self.client ModbusTcpClient(host, port) self.client.connect() def read_float(self, address: int) - float: 读取 FLOAT 类型占2个寄存器 result self.client.read_holding_registers(address, 2) decoder BinaryPayloadDecoder.fromRegisters( result.registers, byteorderEndian.Big, wordorderEndian.Little ) return decoder.decode_32bit_float() def write_float(self, address: int, value: float): 写入 FLOAT 类型 builder BinaryPayloadBuilder(byteorderEndian.Big, wordorderEndian.Little) builder.add_32bit_float(value) payload builder.to_registers() self.client.write_registers(address, payload) def read_int16(self, address: int) - int: 读取 INT16 类型占1个寄存器 result self.client.read_holding_registers(address, 1) return result.registers[0] def write_int16(self, address: int, value: int): 写入 INT16 类型 self.client.write_register(address, value) # 实例化全局复用 plc PlcInterface(192.168.1.10)这个类就是 Agent 的“眼睛和手”。它不关心 PLC 里怎么算 PID只负责准确读取40001的温度值、写入40009的新 P 值。这种解耦正是工业系统可靠性的基石——PLC 专注实时控制Agent 专注智能决策。3.3 状态机设计定义 Agent 的“梯形图逻辑”我们不再写梯形图而是用 LangGraph 定义一个TemperatureControlStatefrom typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver import operator class TemperatureControlState(TypedDict): current_temp: float setpoint: float material_code: int pid_p: float pid_i: float pid_d: float alarm_code: int history: Annotated[List[dict], operator.add] # 存储最近100个采样点 # 定义节点函数 def read_sensors(state: TemperatureControlState) - TemperatureControlState: 读取 PLC 传感器数据模拟 PLC 输入扫描 state[current_temp] plc.read_float(40001) state[setpoint] plc.read_float(40003) state[material_code] plc.read_int16(40007) state[alarm_code] plc.read_int16(40015) return state def check_stability(state: TemperatureControlState) - str: 判断当前温度是否稳定类似梯形图中的定时器比较 # 获取最近10个温度点 recent_temps [h[temp] for h in state[history][-10:]] if len(recent_temps) 10: return tune_pid # 初始阶段直接调参 # 计算标准差小于0.3℃视为稳定 import numpy as np std_dev np.std(recent_temps) if std_dev 0.3: return monitor # 稳定进入监控 else: return tune_pid # 不稳定重新调参 def tune_pid(state: TemperatureControlState) - TemperatureControlState: 根据材质和当前偏差优化 PID 参数核心智能 # 简化版规则引擎实际可用轻量级ML模型 base_p {1: 2.5, 2: 3.0, 3: 2.0}[state[material_code]] # ABS/PC/PP 基础P值 error state[setpoint] - state[current_temp] # 误差大时增强P误差小时增强I if abs(error) 5.0: state[pid_p] base_p * 1.5 state[pid_i] 0.5 else: state[pid_p] base_p state[pid_i] 1.2 # 写入 PLC plc.write_float(40009, state[pid_p]) plc.write_float(40011, state[pid_i]) plc.write_float(40013, state[pid_d]) # 记录日志 state[history].append({ timestamp: time.time(), temp: state[current_temp], action: tuned_pid, params: {p: state[pid_p], i: state[pid_i]} }) return state def monitor_anomalies(state: TemperatureControlState) - TemperatureControlState: 检测异常升温模式如加热棒短路温度线性飙升 if len(state[history]) 5: return state recent state[history][-5:] temps [r[temp] for r in recent] times [r[timestamp] for r in recent] # 线性拟合斜率℃/s from scipy import stats slope, _, _, _, _ stats.linregress(times, temps) if slope 2.0: # 每秒升2℃判定为短路 state[alarm_code] 1 plc.write_int16(40015, 1) # 发送微信报警略 return state # 构建状态图 workflow StateGraph(TemperatureControlState) workflow.add_node(read_sensors, read_sensors) workflow.add_node(tune_pid, tune_pid) workflow.add_node(monitor_anomalies, monitor_anomalies) workflow.add_edge(START, read_sensors) workflow.add_conditional_edges( read_sensors, check_stability, { tune_pid: tune_pid, monitor: monitor_anomalies } ) workflow.add_edge(tune_pid, read_sensors) # 循环回读 workflow.add_edge(monitor_anomalies, read_sensors) # 添加内存检查点支持断点续跑 checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer)这段代码就是一张完整的“数字梯形图”。它包含输入扫描节点read_sensors对应 PLC 的输入映像区刷新状态判断分支check_stability对应梯形图中的比较指令CMP和跳转执行节点tune_pid对应梯形图中的 PID 指令块异常处理节点monitor_anomalies对应梯形图中的故障诊断支路循环边add_edge(tune_pid, read_sensors)对应 PLC 的无限扫描循环。实操心得LangGraph 的add_conditional_edges就是梯形图里的“分支触点”。我最初用if-else写结果状态流转混乱。改成条件边后整个逻辑变得像看梯形图一样清晰——每个节点只做一件事转移规则写在边上一目了然。3.4 部署与联调让 Agent 在真实产线上“呼吸”部署不是把代码扔进服务器就完事。真实产线要求零停机PLC 程序不能动Agent 必须兼容现有 Modbus 地址抗干扰工控机网口可能被电磁干扰Modbus 通信需重试机制可观测运维人员要能快速看懂 Agent 在干什么。我的部署方案网络隔离工控机与 PLC 用独立网段192.168.1.0/24不接入企业内网避免 IT 安全策略干扰通信加固修改PlcInterface加入指数退避重试import time import random def read_float_with_retry(self, address: int, max_retries3) - float: for i in range(max_retries): try: return self.read_float(address) except Exception as e: if i max_retries - 1: raise e wait_time (2 ** i) random.uniform(0, 1) # 1s, 3s, 7s time.sleep(wait_time) return 0.0HMI 集成威纶通 HMI 支持 Modbus TCP 主站我新增一个“AI 模式”页面显示当前 Agent 状态Running/Idle/Error最近一次 PID 调参时间当前报警代码用图标直观显示一个“手动触发调参”按钮对应调用/agent/tick接口。日志与告警所有tune_pid和monitor_anomalies的动作都写入本地 SQLite 数据库并同步到企业微信机器人。运维人员手机收到“【注塑A线】AI温控于 14:22:05 自动将 P 值从 2.5 调整为 3.75因检测到 PC 材质升温缓慢”。联调时踩过的坑坑1PLC 的 Modbus 响应超时设为 1s但工控机网络延迟偶尔达 1.2s导致 Agent 卡死。解决在pymodbus客户端初始化时timeout2并捕获ModbusIOException。坑2HMI 修改设定值后PLC 程序未及时更新40003Agent 读到旧值。解决在 PLC 程序中HMI 写入40003后用一个上升沿触发脉冲确保值已稳定。坑3Agent 连续调参PLC PID 指令来不及响应出现震荡。解决在tune_pid节点中加入防抖逻辑——两次调参间隔至少 30 秒。这些细节教科书不会写但决定了项目成败。它们和你调试梯形图时反复修改定时器时间、调整互锁条件一样是工程师的肌肉记忆。4. 常见问题与实战排查指南4.1 “Agent 跑着跑着就停了”——状态机死锁排查现象Agent 在read_sensors节点后不再流转日志停在“Reading sensors...”但 PLC 数据正常。排查步骤确认检查点Checkpointer是否生效LangGraph 默认不保存状态若 Agent 进程重启会丢失上下文。检查MemorySaver是否传入compile()并在invoke()时指定config{configurable: {thread_id: temp_control}}。检查条件边返回值是否匹配check_stability函数必须返回字符串tune_pid或monitor且必须与add_conditional_edges中的字典键完全一致大小写敏感。我曾因返回TUNE_PID大写导致分支失效。验证节点函数是否抛出未捕获异常tune_pid中调用scipy.stats.linregress若recent_temps长度不足会报错。用try-except包裹所有节点函数并在日志中打印完整 traceback。提示在开发阶段在每个节点函数开头加print(f[{node_name}] Entered with state: {state})是最朴素有效的调试法。它比任何 IDE 断点都可靠因为 LangGraph 的异步调度会让断点失效。4.2 “PLC 数据读出来是乱码”——Modbus 地址与字节序陷阱现象plc.read_float(40001)返回1.23e-38或负数明显不是温度值。根本原因字节序Endianness不匹配。汇川 PLC 默认使用 Big-Endian高位字在前但 x86 工控机是 Little-Endian。pymodbus的BinaryPayloadDecoder必须显式指定# 错误默认字节序读出来是乱码 decoder BinaryPayloadDecoder.fromRegisters(registers, Endian.Little) # 正确汇川 PLC 用 Big-Endian decoder BinaryPayloadDecoder.fromRegisters( registers, byteorderEndian.Big, # 寄存器字节序 wordorderEndian.Little # 寄存器内字序汇川特殊 )验证方法用 Modbus Poll 工具连接 PLC读取 40001看工具显示的值是否与 HMI 一致。若一致说明 PLC 地址正确再对比pymodbus读出的值若不一致90% 是字节序问题。4.3 “Agent 调参后温度更不稳定了”——PID 参数下发时机错位现象Agent 计算出新 PID 值并写入40009/40011/40013但 PLC 的 PID_Compact 指令未立即生效导致新旧参数混用。原因PLC 的 PID 指令有“使能端”EN和“更新端”UPDATE。传统梯形图中UPDATE通常接一个脉冲如 TON 定时器的 Q 点只有脉冲上升沿时PID 才读取新参数。解决方案在 PLC 程序中为 Agent 新增一个“参数更新”标志位如 M1000Agent 写完 PID 参数后再写M10001100ms 后写M10000模拟脉冲PID_Compact 指令的UPDATE端接M1000。这样Agent 的“写参数”和“触发更新”成为原子操作彻底避免错位。4.4 “并发扛不住10个设备就卡死”——Agent 的横向扩展实践热搜词里有“ai agent 怎么扛并发”这确实是痛点。但别急着上 Kubernetes。先问你的场景真的需要高并发吗在注塑厂12 台注塑机每台 Agent 按 50ms 周期调用单台 QPS20。12 台共 240 QPS一台 4 核工控机完全能扛。真正瓶颈是 Modbus TCP 连接数——pymodbus默认每个客户端独占一个 socket12 个 PLC 就要 12 个连接。优化方案连接池用pymodbus的ModbusTcpClient结合concurrent.futures.ThreadPoolExecutor复用连接批量读取不用read_float(40001)单次读改用read_holding_registers(40001, 10)一次读 10 个地址再解码边缘缓存在工控机内存中缓存传感器数据Agent 读取时优先用缓存每 100ms 主动刷新一次。我实测单台工控机管理 20 台 PLCCPU 占用率峰值 45%远低于 80% 阈值。所谓“并发问题”80% 是过早优化。先确保单实例稳定再谈扩展。4.5 “怎么向老板证明 AI Agent 有效”——效果量化与 ROI 计算老板不关心技术只关心省了多少钱提了多高效率降了多少故障我的量化方法温度波动降低用 PLC 的40001历史数据计算上线前后 7 天的温度标准差。原系统 ±4.2℃上线后 ±1.8℃波动减少 57%。故障停机减少统计“加热异常”报警次数。原系统平均每天 3.2 次AI 系统上线后 0.3 次下降 91%。人工干预减少记录工程师手动调 PID 的次数。原系统每周 12 次现系统每周 0.5 次释放人力约 8 小时/周。ROI 计算简化年节省人工成本8 小时 × 52 周 × 150 元/小时 62,400 元年减少废品损失按每班次减少 1 次停机每次损失 2,000 元年节约 312,000 元项目总投入工控机 8,000 元 开发 3 人×2 周×20,000 元 128,000 元投资回收期 6 个月。注意所有数据必须来自 PLC 原始寄存器而非 HMI 显示值。我用脚本每天凌晨自动导出40001的 24 小时数据到 CSV用 Pandas 计算指标。这才是工程师该交的答卷。5. 从 PLC 工程师到 AI 工程师一条被忽视的平滑路径写完这篇我关掉电脑走到车间。那台汇川 H3U PLC 还在嗡嗡运行散热风扇声和十年前一模一样。但它的旁边多了一台工控机屏幕上滚动着 LangGraph 的状态日志。没有颠覆只有延伸——就像当年 PLC 替代继电器柜时老师傅们也是先学会看