工业安全系统最难的并不是“识别出画面里有人”，而是要在网络带宽、模型延迟和误报率的共同约束下，及时判断某个现场事件是否真的危险。Tata Elxsi 构建的 IRIS 平台采用一条清晰的边云协同链路：在边缘过滤摄像头视频，通过 Amazon Kinesis 传输元数据，使用 Amazon SageMaker AI 运行计算机视觉模型，再把多次检测结果关联为高置信度告警，将发现不安全状况的时间从分钟缩短到秒级。
为什么不把所有视频直接送进云端
工业现场可能同时部署数十甚至数百路摄像头。若持续上传原始高清视频，系统很快会遇到三个问题：
- 带宽成本高：大量画面没有安全事件，但仍会占用上行链路。
- 端到端延迟不稳定：视频上传、解码和排队都会拉长告警时间。
- 隐私与合规压力大：原始视频包含人员、设备和生产流程等敏感信息。
IRIS 的关键选择是在边缘先过滤视频。边缘节点可以完成抽帧、运动检测、区域裁剪或轻量级初筛，只把值得分析的帧、对象特征或事件元数据发送到云端。这样一来，Kinesis 承担的是连续事件流，而不是所有摄像头的完整视频流。
一条典型元数据可以设计为：
{
"camera_id": "line-a-camera-07",
"site_id": "factory-01",
"captured_at": "2025-03-08T10:15:23.184Z",
"frame_uri": "s3://safety-frames/factory-01/frame-001.jpg",
"zone": "forklift-lane",
"edge_signals": {
"motion": true,
"person_count": 1
}
}
这里的 frame_uri 是一种可选实践：边缘端可把选中的帧写入受控对象存储，再通过事件传递位置；如果边缘端已经生成足够可靠的特征，也可以只发送元数据。实际方案应根据延迟、合规要求和模型输入方式决定。
Kinesis、SageMaker AI 与事件关联各自解决什么问题
这套架构不是简单地把模型接到摄像头后面，而是把实时处理拆成三个职责明确的阶段。
Kinesis：吸收摄像头事件流
Amazon Kinesis 位于边缘设备和云端处理服务之间，可以缓冲突发事件，并让生产者与消费者解耦。分区键通常应选择 camera_id 或其哈希值，以尽量保持同一摄像头事件的顺序。
不过，如果少数摄像头产生的事件量远高于其他设备，直接使用摄像头 ID 可能形成热分区。此时可以使用 camera_id + 时间窗口 作为分区键，同时在下游按摄像头和事件时间重新排序。
SageMaker AI：执行视觉推理
消费者从流中取得事件后，可以读取对应帧或特征，并调用部署在 SageMaker AI 上的计算机视觉模型。模型输出可能包括：
- 未佩戴安全帽或防护装备；
- 人员进入受限区域；
- 人员与叉车、机械臂等设备距离过近；
- 跌倒、烟雾或其他异常状态。
具体风险类型取决于训练数据和工厂场景。模型置信度也不应直接等同于业务风险等级：一次置信度为 0.91 的检测，仍可能是光照变化、遮挡或反光造成的误判。
事件关联：把“模型命中”变成“可信告警”
IRIS 会关联检测结果，以生成高置信度告警。这一步非常重要，因为工业告警不能只依赖单帧判断。可以这样实践：只有同一摄像头、同一区域、同一种风险在短时间窗口内连续出现，或者同时满足多个条件时，才触发通知。
例如，“人员位于叉车通道”本身可能只是普通作业；如果同一时间还检测到叉车接近，并且距离持续缩短，风险等级才应升级。
可改造的最小事件生产示例
下面的示例模拟边缘节点向 Kinesis Data Streams 发送筛选后的摄像头元数据。它并非 IRIS 的原始实现，而是基于上述架构可以采用的最小参考实现。
运行前需要安装 AWS CLI 和 Python 依赖，并配置具备 Kinesis 写权限的 AWS 凭证：
python -m pip install boto3
aws kinesis create-stream \
--stream-name industrial-safety-events \
--shard-count 1 \
--region us-east-1
aws kinesis wait stream-exists \
--stream-name industrial-safety-events \
--region us-east-1
保存下面的代码为 edge_publish.py。运行时可按实际环境修改区域、流名称、摄像头编号和帧地址：
import json
import os
import time
import uuid
from datetime import datetime, timezone
import boto3
REGION = os.getenv("AWS_REGION", "us-east-1")
STREAM_NAME = os.getenv("KINESIS_STREAM", "industrial-safety-events")
CAMERA_ID = os.getenv("CAMERA_ID", "line-a-camera-07")
kinesis = boto3.client("kinesis", region_name=REGION)
def publish_candidate_event(person_count: int, zone: str) -> None:
event = {
"event_id": str(uuid.uuid4()),
"camera_id": CAMERA_ID,
"site_id": "factory-01",
"captured_at": datetime.now(timezone.utc).isoformat(),
"frame_uri": f"s3://safety-frames/factory-01/{uuid.uuid4()}.jpg",
"zone": zone,
"edge_signals": {
"motion": True,
"person_count": person_count,
},
}
response = kinesis.put_record(
StreamName=STREAM_NAME,
PartitionKey=CAMERA_ID,
Data=json.dumps(event).encode("utf-8"),
)
print(
json.dumps(
{
"event_id": event["event_id"],
"shard_id": response["ShardId"],
"sequence_number": response["SequenceNumber"],
},
indent=2,
)
)
if __name__ == "__main__":
# 在真实边缘节点上，只在运动检测或轻量模型命中后调用。
while True:
publish_candidate_event(person_count=1, zone="forklift-lane")
time.sleep(5)
启动模拟生产者：
AWS_REGION=us-east-1 \
KINESIS_STREAM=industrial-safety-events \
CAMERA_ID=line-a-camera-07 \
python edge_publish.py
生产环境还应加入本地磁盘队列、指数退避重试、事件唯一 ID 和断网恢复机制。仅调用 put_record 并不意味着事件已经完成端到端处理，下游还需要处理重复消息和失败重放。
告警策略应围绕业务风险，而不是单次置信度
一个可落地的关联规则可以定义为：在 10 秒窗口内，同一摄像头至少出现 3 次“未佩戴安全帽”检测，且平均置信度不低于 0.85，才生成告警。对于人员进入机械臂活动区域等高危事件，则可以降低重复次数，但要求区域和设备状态同时匹配。
建议为每条告警保留以下解释信息：
- 触发规则及阈值；
- 参与关联的事件 ID；
- 模型版本和置信度；
- 摄像头、区域及事件时间；
- 可供授权人员复核的证据帧；
- 告警是否被确认、忽略或标记为误报。
这些数据既能支持审计，也能形成模型再训练和阈值调整所需的反馈闭环。
上线前需要验证的边界
秒级检测不等于可以跳过工程治理。采用类似 IRIS 的架构时，应重点检查：
- 测量完整延迟：从摄像头采集时间开始，而不是只统计模型推理耗时。
- 控制误报与漏报：按摄像头、班次、光照和风险类型分别评估。
- 设计降级路径：边缘断网时缓存事件，模型不可用时进入人工复核或备用规则。
- 保护敏感数据：限制原始帧保留时间，使用传输与静态加密，并按角色控制访问。
- 避免重复告警：为事件设置幂等键和冷却窗口，防止同一事故反复通知。
- 保留人工责任链：系统负责快速发现和升级风险，不应在缺乏额外安全机制时直接控制高危设备。
这套方案的价值不只来自某个视觉模型，而是来自边缘筛选、流式传输、云端推理和事件关联之间的配合。只有把延迟、置信度、可解释性和故障处理放在同一条链路中设计，工业视觉系统才能真正把“模型看见了什么”转化为现场可以及时采取行动的安全告警。