1. 从一次现场联调说起:为什么 SCADA 需要 AI Agent Harness
AI Agent Harness 与 SCADA 系统集成,说白了就是给传统 SCADA 装一个能自主感知、推理、决策、执行的“调度中枢”。SCADA 负责采、传、存、显、控、审,Harness 负责把多个 Agent 组装成团队,统一管理工具调用、安全约束和知识库。适合工业物联网运维工程师、SCADA 开发者,以及正在做 AI Agent 落地的人。
我在一个化工聚合釜的仿真环境里做过一次端到端联调:Ignition SCADA 通过 OPC UA 采集温度、搅拌转速、阀门开度,Harness 侧监听告警 Webhook,触发根因分析 Agent,再把决策写回 SCADA 点位。整个过程卡住最久的不是模型,而是 config.toml 里的通道映射和鉴权字段。这篇就把这套配置骨架和联调脚本完整拆开,你复制后改几个地址就能跑通一次读写点位与告警回传。
2. TaoToken 前置准备:拿到可用的 API Key 与接入地址
Harness 里的推理节点需要调用大模型,我习惯把模型访问统一收敛到 TaoToken 上,这样切换模型时不用改业务代码。你需要先拿到 API Key,再确认接入地址。
2.1 注册与获取 API Key
打开官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,完成注册后进入控制台。在左侧找到 API Keys 页面,新建一个 Key,复制保存。这个 Key 只显示一次,丢了只能重建。
控制台地址:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
API Keys 页面:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
2.2 确认接入地址与模型
API 基础地址是 https://taotoken.net/api ,不带任何查询参数。它兼容 OpenAI 风格的接口,所以 Harness 里可以直接用 openai 的 SDK,把 base_url 指过来即可。模型名按你控制台里可用的填,比如 gpt-4o 或 claude 系列。想先验证模型通不通,可以用模型对话页面直接发一条消息:
模型对话:https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
如果你后面要做长期编码或 Agent 常驻调度,可以看 Coding Plan,它更适合持续性的 Agent 任务:
Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
注意:API Key 不要写进前端代码或提交到 Git。联调阶段可以放环境变量,生产环境建议走密钥管理服务。
3. config.toml 骨架:通道、鉴权、采集点映射
Harness 的配置文件我按三块拆:SCADA 通道、模型鉴权、采集点映射。下面这份骨架可以直接用,字段名按你实际 Harness 版本微调。
3.1 完整 config.toml 示例
# config.toml - AI Agent Harness 与 SCADA 集成骨架 [harness] name = "scada-agent-harness" log_level = "info" # 告警监听端口,SCADA Webhook 往这里推 webhook_port = 8787 # Agent 最大并发,工业现场别开太大 max_concurrent_agents = 4 [scada] # SCADA 侧 OPC UA 端点 endpoint = "opc.tcp://192.168.10.21:4840" # 会话超时,单位毫秒 session_timeout_ms = 30000 # 安全策略,现场按实际选 security_policy = "Basic256Sha256" security_mode = "SignAndEncrypt" # 应用证书路径 cert_path = "./certs/harness_client.der" key_path = "./certs/harness_client.pem" [scada.auth] # SCADA 用户名密码鉴权 username = "harness_agent" password = "${SCADA_PASSWORD}" # 从环境变量读取 # 是否允许写入,联调阶段先 false,验证读取后再开 write_enabled = false [llm] # TaoToken 接入地址,不带查询参数 base_url = "https://taotoken.net/api" api_key = "${TAOTOKEN_API_KEY}" model = "gpt-4o" timeout_s = 60 max_retries = 2 [points] # 采集点映射:SCADA 节点 ID -> Agent 内部别名 [points.temperature] node_id = "ns=2;s=Reactor2.T201" alias = "reactor2_temp" data_type = "float" unit = "degC" # 是否参与告警判断 watch = true [points.stirrer_speed] node_id = "ns=2;s=Reactor2.S201" alias = "reactor2_stirrer" data_type = "float" unit = "rpm" watch = true [points.cooling_valve] node_id = "ns=2;s=Reactor2.V204" alias = "reactor2_cooling_valve" data_type = "float" unit = "%" watch = true # 允许 Agent 写回的点位 writable = true [alarm] # 告警回传地址,Harness 处理完往 SCADA 推 callback_url = "http://192.168.10.21:8088/api/alarm/ack" # 告警阈值 temp_high = 85.0 temp_deviation = 0.8 deviation_duration_s = 123.2 关键字段说明
通道部分重点是 endpoint 和 security_policy。工业现场如果用的是 SignAndEncrypt,证书必须双向信任,否则握手直接失败。鉴权部分我把密码和 API Key 都走环境变量,避免明文落盘。采集点映射是联调最容易出错的地方,node_id 必须和 SCADA 侧完全一致,大小写、分号、命名空间都不能差。
| 配置块 | 关键字段 | 作用 | 常见坑 |
|---|---|---|---|
| scada | endpoint | OPC UA 服务地址 | 端口写错或防火墙拦截 |
| scada.auth | write_enabled | 控制是否允许写回 | 联调初期误开导致误写 |
| llm | base_url | 模型接入地址 | 多加了斜杠或查询参数 |
| points | node_id | SCADA 节点标识 | 命名空间索引对不上 |
| alarm | callback_url | 告警回传地址 | SCADA 侧接口未开跨域 |
3.3 CC Switch 切换步骤
CC Switch 用来在多个配置档之间切换,比如现场调试档和仿真档。操作顺序是:先停掉当前 Harness 进程,再执行切换命令,最后重新加载配置。
# 查看当前配置档 cc-switch list # 切换到仿真档 cc-switch use sim-scada # 确认切换结果 cc-switch current # 重新加载 Harness 配置 systemctl restart scada-harness切换后一定要用 cc-switch current 确认,我有一次没确认,结果 Harness 还在读旧档,排查了半小时才发现是切换没生效。
4. 最小联调脚本:验证读写点位与告警回传
配置就绪后,用一个最小脚本验证三件事:能不能读到点位、能不能写回点位、告警能不能回传。
4.1 读取点位验证
import asyncio from asyncua import Client async def read_points(): async with Client(url="opc.tcp://192.168.10.21:4840") as client: node = client.get_node("ns=2;s=Reactor2.T201") value = await node.read_value() print(f"T201 当前温度: {value} degC") return value if __name__ == "__main__": asyncio.run(read_points())跑通后你会看到类似T201 当前温度: 82.4 degC的输出。如果报 BadNodeIdUnknown,就是 node_id 写错了,回 SCADA 侧复制一遍。
4.2 写入点位验证
写入前先把 config.toml 里的 write_enabled 改成 true,并且确认目标点位 writable = true。
import asyncio from asyncua import Client async def write_point(): async with Client(url="opc.tcp://192.168.10.21:4840") as client: node = client.get_node("ns=2;s=Reactor2.V204") await node.write_value(15.0) readback = await node.read_value() print(f"V204 写回后读值: {readback} %") if __name__ == "__main__": asyncio.run(write_point())写回后立刻读一次,确认值真的落到了 SCADA 侧。如果读回来还是旧值,多半是 SCADA 侧点位被锁定或权限不足。
4.3 告警回传验证
用一个本地 HTTP 服务模拟 SCADA 的告警接收端,再让 Harness 往它推一条告警。
from fastapi import FastAPI, Request import uvicorn app = FastAPI() @app.post("/api/alarm/ack") async def alarm_ack(request: Request): body = await request.json() print(f"收到告警回传: {body}") return {"status": "ok"} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8088)启动后,在 Harness 侧触发一次告警,观察这个服务是否打印出回传内容。成功的话你会看到包含点位别名、当前值、时间戳的 JSON。
4.4 端到端串联
把上面三步串起来:脚本先读温度,超过阈值就调用 Harness 的推理接口,拿到决策后写回阀门开度,最后把处理结果回传到告警接口。跑通一次,就完成了一次完整的端到端集成验证。
5. 本篇常见错排查
联调阶段报错集中在几个地方,我按出现频率排一下。
5.1 OPC UA 连接被拒
报错通常是BadConnectionClosed或BadSecurityChecksFailed。先确认 endpoint 地址和端口,再检查证书是否互信。如果现场用的是 Basic256Sha256,客户端证书必须导入 SCADA 的信任列表,反之亦然。防火墙也要放行 4840 端口。
5.2 模型调用返回 401
401 基本都是 API Key 问题。确认环境变量 TAOTOKEN_API_KEY 是否真的注入到了进程里,可以用printenv | grep TAOTOKEN检查。另外确认 base_url 是 https://taotoken.net/api ,不要多加斜杠或路径。
5.3 点位映射对不上
BadNodeIdUnknown说明 node_id 不存在。OPC UA 的节点 ID 格式是ns=索引;s=标识符,命名空间索引在不同 SCADA 上可能不同。最稳的办法是在 SCADA 侧直接复制节点 ID,别手敲。
5.4 告警回传超时
如果 Harness 日志里出现 callback timeout,先确认 callback_url 能从 Harness 所在机器访问通。工业网络常有网段隔离,Harness 和 SCADA 不在同一网段时,需要开路由或做端口映射。
5.5 CC Switch 切换后配置未生效
切换后必须重启 Harness 进程,否则它还在用内存里的旧配置。用 cc-switch current 确认当前档,再 systemctl restart。
提示:联调阶段建议把 write_enabled 保持 false,只验证读取和告警回传。确认链路稳定后再开写入,避免误操作影响现场设备。
6. 下一步:把联调脚本接进你的 Agent 流程
到这里,config.toml 骨架、CC Switch 切换、读写点位和告警回传都跑通了。接下来你可以把第 4 节的脚本封装成 Harness 的一个工具节点,让 Agent 在告警触发时自动调用。模型侧继续用 TaoToken 的接入地址,切换模型时只改 config.toml 里的 model 字段即可。
如果你要验证不同模型在根因分析上的表现,可以直接在模型对话页面里试:
https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
如果要把这套 Harness 做成长期常驻的编码或调度 Agent,Coding Plan 更合适:
https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
接入文档里有更完整的字段说明和示例:
https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
我自己的习惯是先把读取和告警回传跑稳,再开写入,最后才让 Agent 自主决策。工业现场不比普通应用,每一步都留好回退路径,比追求一次跑通更重要。