2026 上海开源软件应用创新大赛已经启动报名，并设置了 AI+工业软件、智算云、开源 AI 工具三大赛道。其中，AI+工业软件关注的不只是一个能在工厂里演示的 AI 应用，而是一个制造业反复追问的问题：AI 什么时候能够稳定地跑在产线上？
这句话里的重点不是“AI”，而是“稳定”。一套工业系统要面对设备差异、脏数据、网络中断、模型漂移和责任追溯。模型给出一次漂亮结果并不难，难的是把结果变成可约束、可回退、可审计的生产动作。
工业场景不缺模型，缺的是可执行边界
聊天应用答错一次，用户通常可以重新提问；产线系统判断错误，可能造成误停机、漏检、原料浪费，甚至带来安全风险。因此，工业 AI 不能只输出一个分类标签或自然语言建议，它还需要回答几个工程问题：
- 输入来自哪台设备，采样时间和单位是否可信？
- 模型版本、特征版本和阈值能否追溯？
- 低置信度结果交给谁处理？
- AI 服务超时或断网时，设备执行什么默认策略？
- 决策是否经过业务规则、安全联锁和权限校验？
这意味着参赛项目即使还不能直接进入真实工厂，也应该展示一条完整链路：采集数据、校验数据、执行推理、应用规则、输出动作、记录审计，并在异常情况下回退。
一个缺少这些环节的视觉检测模型，本质上仍是算法样品；一个能够明确拒绝不可信输入、保留人工复核入口并记录每次决策的系统，才开始接近工业软件。
“能干活”要落在三个接口上
1. 与设备和现有系统连接
工业现场很少从零开始。项目通常需要接入 PLC、SCADA、MES、QMS、ERP 或设备厂商协议。与其把所有精力放在重新设计界面上，不如先定义稳定的数据契约：设备编号、时间戳、测点、单位、质量标志和批次号都应该有明确语义。
接口也不必一开始就覆盖全部协议。可以用 MQTT、OPC UA、HTTP 或文件交换完成最小闭环，再把协议适配层与模型服务隔离。这样更换设备或模型时，不必重写整套系统。
2. 把模型结果翻译成生产动作
“异常概率为 0.87”不是生产动作。工业软件需要把它翻译成“继续运行”“降速”“转人工复核”或“触发停机申请”，并说明阈值和约束来自哪里。
模型适合发现模式，确定性规则适合守住边界。两者组合通常比让模型直接控制设备更容易验证：模型提供风险评分，规则引擎结合设备状态、工艺窗口和安全条件决定是否执行。
3. 用可观测性证明稳定
工业 AI 的指标不能只有准确率。实际项目还应记录端到端延迟、拒绝率、人工复核率、误报与漏报、设备分布差异，以及不同模型版本的表现。
如果系统无法回答“昨晚 02:13 为什么判定 7 号设备需要停机”，就很难进入持续运行阶段。日志、指标和版本信息不是上线后的补充，而是产品的一部分。
可以这样实践：先做一个带安全闸门的决策服务
下面是一个可复制运行的最小骨架。这里明确做两个假设：传感器已经通过网关转换为 JSON；示例中的风险计算只是占位逻辑，真实项目应替换为经过验证的模型或规则。代码重点展示输入校验、置信度门槛、安全回退和审计记录。
将以下内容保存为 decision_gateway.py：
#!/usr/bin/env python3
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
AUDIT_FILE = Path('decisions.jsonl')
MODEL_VERSION = 'demo-risk-model-0.1'
def risk_score(temperature: float, vibration: float) -> float:
# 占位算法：实际使用时替换为经过验证的模型推理。
temp_risk = max(0.0, min(1.0, (temperature - 70.0) / 30.0))
vibration_risk = max(0.0, min(1.0, (vibration - 4.0) / 6.0))
return round(0.4 * temp_risk + 0.6 * vibration_risk, 4)
def decide(event: dict) -> dict:
required = {'device_id', 'timestamp', 'temperature_c', 'vibration_mm_s'}
missing = sorted(required - event.keys())
if missing:
return {'action': 'manual_review', 'reason': f'missing fields: {missing}'}
try:
temperature = float(event['temperature_c'])
vibration = float(event['vibration_mm_s'])
except (TypeError, ValueError):
return {'action': 'manual_review', 'reason': 'invalid numeric value'}
# 超出传感器合理量程时拒绝自动决策。
if not (-20 <= temperature <= 180 and 0 <= vibration <= 50):
return {'action': 'manual_review', 'reason': 'sensor value out of range'}
score = risk_score(temperature, vibration)
if score >= 0.85:
action = 'request_controlled_stop'
elif score >= 0.55:
action = 'reduce_speed_and_inspect'
else:
action = 'continue'
return {'action': action, 'risk_score': score, 'reason': 'policy_v1'}
def process(event: dict) -> dict:
event_id = hashlib.sha256(
json.dumps(event, sort_keys=True).encode('utf-8')
).hexdigest()[:16]
result = {
'event_id': event_id,
'decided_at': datetime.now(timezone.utc).isoformat(),
'model_version': MODEL_VERSION,
**decide(event),
}
with AUDIT_FILE.open('a', encoding='utf-8') as audit:
audit.write(json.dumps({'input': event, 'output': result}) + '\n')
return result
if __name__ == '__main__':
event = json.load(sys.stdin)
print(json.dumps(process(event), indent=2))
运行时输入一条模拟设备事件：
printf '%s' '{"device_id":"press-07","timestamp":"2026-03-18T02:13:00Z","temperature_c":96,"vibration_mm_s":9.2}' \
| python3 decision_gateway.py
cat decisions.jsonl
这个示例没有直接向 PLC 下发停机指令，而是输出 request_controlled_stop。这是一个有意保留的边界：在完成安全认证、联锁验证和现场验收之前，AI 应提交动作请求，由确定性控制系统或授权人员执行最终操作。
后续可以把该脚本改造成 HTTP 或 MQTT 服务，并逐步补上设备身份认证、消息去重、超时策略、指标采集、模型签名和灰度发布。改造过程中应保持输入契约和审计格式稳定，让模型迭代不影响上游设备与下游 MES。
参赛项目应该证明什么
AI+工业软件项目不必假装已经解决整个工厂的问题，但需要清楚展示它解决了哪一个生产环节，以及失败时系统会怎样表现。准备方案时可以检查以下内容：
- 是否选择了可量化的任务，例如缺陷复核、预测性维护、工艺参数推荐或排产辅助？
- 是否给出了端到端流程，而不只是模型准确率和演示界面？
- 是否提供低置信度、数据缺失、服务超时和网络中断的处理策略？
- 是否记录数据来源、模型版本、规则版本、操作者和最终动作？
- 是否能在影子模式中运行，先给建议而不控制设备，并与真实结果对比？
- 开源部分是否包含可运行代码、样例数据、部署说明和明确许可证？
工业 AI 的价值不会由一次流畅对话证明，而要由成千上万次受约束的正确执行证明。对参赛团队而言，最有说服力的作品通常不是功能最多的系统，而是把一个具体任务做成了可连接、可回退、可观测、可持续迭代的工程闭环。