当讨论“天网”时技术人员真正关心的问题通常是一套自动化监控预警系统如何从视频流采集开始一步步完成目标识别、告警通知和数据沉淀。科幻作品里的天网是一个拥有自主意识的全球控制系统而在工程实践中我们所说的天网更接近一套由摄像头、视频解码、AI模型、消息中间件和可视化看板组成的智能安防监控架构。这类系统已经在工厂安全生产、园区周界防护、仓库消防预警等场景中大量落地。本文假设读者已经掌握基本的Python语法了解HTTP接口和数据库基础知识但不要求读者有算法背景。读完后你会理解监控预警系统的核心链路能搭建一个最小可运行项目并在遇到视频流拉取失败、模型推理卡顿、告警重复等问题时知道从哪里排查。1. 把“天网”拉回工程语境智能监控预警系统到底在做什么1.1 从科幻名词到工程架构在科幻设定中天网是一个可以自主感知、决策并执行动作的人工智能系统它有摄像头、传感器、武器系统和数据中心。如果把“自主决策”和“执行动作”去掉只保留“感知、识别、告警、留痕”这四个能力天网式系统就变成了一套企业级安防监控平台。这类平台通常不具备自主意识也不会直接控制物理设备。它做的事情是从摄像头不断读取画面用AI模型判断画面中是否存在目标对象一旦目标触发规则就生成一条告警记录并通知相关人员。整个过程是确定性的、可配置的也是可关闭的。理解这一点就不会把工程问题神秘化。在项目里将新系统称为“天网”更多是一种类比它像天网一样拥有覆盖全场的“眼睛”但它的行为边界由工程师定义。实际项目中摄像头只负责采集画面AI模型只负责给出置信度告警规则由业务人员配置管理人员可以通过开关随时干预。1.2 落地场景与业务价值最典型的落地场景包括几类工厂安全帽检测在作业区域检测未佩戴安全帽的人员触发告警减少安全事故。园区周界防护在非开放时段检测人员靠近或跨越围栏。仓库烟火预警检测烟雾和明火第一时间通知消防值班人员。停车区域管理识别占用消防通道的车辆。生产线行为规范检测操作人员是否进入危险区域。这些场景的共同特点是监控点位很多人眼无法长时间盯屏事件出现时间短错过之后很难回溯传统摄像头只能录像不能主动发现问题。智能监控预警系统把“事后查录像”变成“事中告警、事后可查”这是它最大的业务价值。从技术角度看这类系统的核心挑战不在单一环节而在链路协同视频采集要稳定AI推理要在可接受的延迟内完成告警要准确且不刷屏数据要能回溯界面要让值班人员快速理解发生了什么。1.3 系统中的关键角色一套完整的监控预警系统可以拆成六个角色每个角色的实现方式可以不同但这个分工在大多数项目里一致。角色职责常见实现视频源输出连续画面IPC摄像头、NVR、本地视频文件流媒体服务统一接入并转发视频流RTSP/RTMP服务、FFmpeg推理服务对视频帧执行AI识别YOLO、OpenCV、TensorRT告警服务判断规则、生成告警、发送通知Python后台任务、Webhook、消息队列数据存储保存告警、事件和图片证据MySQL、PostgreSQL、对象存储可视化看板展示实时状态和历史查询Web前后端、大屏系统1.4 为什么不能只靠普通录像普通数字录像系统只做两件事存储码流和回放画面。它不判断画面中是否存在异常也不主动通知任何人。这意味着异常事件发生后负责人必须等操作人员上报然后人工翻查录像效率很低。而智能监控预警系统在画面采集到后立即做一次推理判断把结果写入结构化数据同时触发通知。排查问题时可以直接检索告警时间、设备和目标类型不再依赖肉眼回放。1.5 技术路线选择实现这套系统常见有三种路线。第一种是使用厂商私有协议和配套平台部署简单但扩展性差难以接入异构设备。第二种是使用开源流媒体服务和自研AI推理模块灵活度高但需要团队维护多个组件。第三种是直接用云服务成本低、免运维但视频数据出域可能不适合部分数据敏感场景。本文采用第二种路线原因是它最适合理解技术链路也方便在开发环境随意调试。2. 环境准备硬件、依赖和项目结构先对齐2.1 开发环境与依赖列表搭建最小系统不需要很高配置。建议开发机至少具备4核CPU和8GB内存如果使用GPU推理显存建议6GB以上。操作系统推荐Ubuntu 20.04或Windows 10以上版本。运行环境建议使用Python 3.10因为较新的推理库对旧版本支持逐渐减弱。主要依赖库如下表。这里的版本是常见组合实际落地前要结合自己的操作系统和官方文档确认。软件/库用途建议版本范围Python编写服务和推理逻辑3.10OpenCV-Python读取视频流、处理图像4.8.x 或 4.9.xPyTorch加载并运行YOLO模型2.xUltralytics YOLO目标检测封装库8.xFastAPI提供HTTP接口0.104SQLAlchemy操作数据库2.xMySQL / SQLite保存告警数据MySQL 8.x 或内置SQLite这些库并不全是唯一选择。OpenCV可以换成FFmpeg命令行工具PyTorch可以换成ONNX RuntimeFastAPI也可以换成Flask。判断标准是团队熟悉度与部署环境兼容性。2.2 模拟视频源与流媒体服务实际项目中视频来自IPC摄像头地址通常是rtsp://user:passwordip:554/stream1。开发阶段如果没有摄像头可以用本地视频文件作为输入也可以用OpenCV把一段循环视频模拟成RTSP流。最简单的开发方式是不搭建流媒体服务直接用OpenCV读取本地视频import cv2 video_path ./demo.mp4 cap cv2.VideoCapture(video_path) while cap.isOpened(): ret, frame cap.read() if not ret: cap.set(cv2.CAP_PROP_POS_FRAMES, 0) continue cv2.imshow(frame, frame) if cv2.waitKey(1) 0xFF ord(q): break cap.release() cv2.destroyAllWindows()这段代码循环读取同一个视频文件模拟持续的视频流。对于验证AI推理和告警链路这种方式足够。要注意本地文件的读取速度会远超实时帧率如果代码里不添加按帧率控制的等待时间推理服务会以极高速度处理造成CPU占满。如果必须验证RTSP协议可以使用FFmpeg搭建本地流媒体服务ffmpeg -re -stream_loop -1 -i demo.mp4 -rtsp_transport tcp -f rtsp rtsp://127.0.0.1:8554/live然后OpenCV读取rtsp://127.0.0.1:8554/live。-stream_loop -1表示无限循环-re表示按原始帧率读取避免发送过快。2.3 项目目录结构为了让后续实现不混乱建议按下面的目录组织工程。这个结构不是唯一答案但功能边界清晰适合逐步扩展。smart-surveillance/ ├── app/ │ ├── main.py # FastAPI入口负责接口和静态页面 │ ├── detector.py # 模型加载和帧检测封装 │ ├── stream.py # 视频流读取与帧管理 │ ├── alert.py # 告警规则判断与通知 │ ├── models.py # 数据库模型 │ └── config.py # 配置项 ├── models/ │ └── yolov8n.pt # 目标检测权重文件 ├── data/ │ └── demo.mp4 # 测试视频 ├── static/ │ └── index.html # 可视化看板 ├── requirements.txt └── README.md这个结构中stream.py只负责提供视频帧detector.py只负责推理alert.py只负责告警逻辑main.py负责把组件组织成服务。职责分离后想要替换摄像头或换模型不会牵连整个项目。3. 核心实现从视频流采集到AI告警的完整链路3.1 视频流接入与帧提取在stream.py中封装一个视频帧读取器。关键点是读取视频流不能阻塞主逻辑因此通常放到后台线程里持续读取并且只保留最新帧避免内存不断上涨。import cv2 import threading import time class StreamReader: def __init__(self, source: str, interval: float 0.03): self.cap cv2.VideoCapture(source) self.interval interval self.lock threading.Lock() self.frame None self.running True threading.Thread(targetself._loop, daemonTrue).start() def _loop(self): while self.running: ret, frame self.cap.read() if not ret: self.cap.set(cv2.CAP_PROP_POS_FRAMES, 0) continue with self.lock: self.frame frame time.sleep(self.interval) def get_frame(self): with self.lock: return self.frame.copy() if self.frame is not None else None def release(self): self.running False self.cap.release()为什么要用锁因为推理服务在另一个线程里调用get_frame如果直接读写self.frame可能会出现读取到不完整图像的情况。interval控制处理速度一般结合视频帧率调整例如25FPS对应0.04秒。实际项目中还需要考虑摄像头断线重连。可以在_loop里检测cap.isOpened()如果断开就cap.open(source)重试并增加重试间隔避免日志刷屏。3.2 AI目标检测模型接入在detector.py中使用YOLO模型。以Ultralytics YOLO为例加载模型并封装一个predict方法from ultralytics import YOLO class Detector: def __init__(self, model_path: str): self.model YOLO(model_path) def predict(self, frame): results self.model.predict(frame, verboseFalse) detections [] for r in results: for box in r.boxes: detections.append({ class_id: int(box.cls[0]), class_name: r.names[int(box.cls[0])], confidence: float(box.conf[0]), x: int(box.xyxy[0][0]), y: int(box.xyxy[0][1]), w: int(box.xyxy[0][2] - box.xyxy[0][0]), h: int(box.xyxy[0][3] - box.xyxy[0][1]), }) return detections这里返回的目标框坐标格式是x, y, w, h方便后续在画面上绘制矩形。confidence是置信度取值0到1。置信度越低漏报越少但误报越多置信度越高误报越少但漏报可能增加。在开发环境模型文件yolov8n.pt可以从Ultralytics官方仓库下载也可以换成自己训练的权重。初次加载模型时会下载权重网络环境不同可能影响耗时。如果使用GPU推理需要在安装PyTorch时选择对应CUDA版本否则CPU推理也能运行只是帧率较低。3.3 告警触发与消息通知告警逻辑不应该简单写成“检测到目标就发送消息”。如果目标连续出现在几十帧中系统会发送大量重复告警。常见做法是增加冷却时间和目标区域过滤。alert.py的核心逻辑如下import time import requests class AlertManager: def __init__(self, webhook_url: str, cooldown: int 30): self.webhook_url webhook_url self.cooldown cooldown self.last_alert_time 0 def should_alert(self, detections, enabled_classes) - bool: if not detections: return False now time.time() if now - self.last_alert_time self.cooldown: return False for det in detections: if det[class_name] in enabled_classes and det[confidence] 0.5: self.last_alert_time now return True return False def send(self, message: dict): if self.webhook_url: requests.post(self.webhook_url, jsonmessage, timeout3) print(alert:, message)冷却时间cooldown的单位是秒表示同一事件在冷却时间内只通知一次。对于安全帽检测冷却时间可以设为10到30秒对于烟火检测可以适当缩短。enabled_classes用于过滤需要触发告警的目标类别避免“人”这种常见目标频繁告警。发送消息时可以包含告警时间、设备名称、目标类别、置信度、现场截图地址等信息。截图可以在告警时把当前帧保存为图片并上传到对象存储或本地目录方便值班人员查看。def build_message(det, image_path, device_id): return { msgtype: text, text: { content: f【安防告警】设备:{device_id} 检测到{det[class_name]} 置信度:{det[confidence]:.2f} 截图:{image_path} } }如果是企业微信或钉钉机器人可以直接使用Webhook地址如果是自研平台可以调用自己的告警接口。生产环境中不建议在告警逻辑里同步发送HTTP请求因为网络延迟会占用推理线程。更好的方式是把告警写入消息队列由独立消费者负责通知。3.4 告警数据落库与查询告警数据需要长期保存。为了方便演示这里使用SQLite通过SQLAlchemy定义一张告警表。CREATE TABLE alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id VARCHAR(64) NOT NULL, class_name VARCHAR(64) NOT NULL, confidence REAL NOT NULL, image_path VARCHAR(255), created_at DATETIME DEFAULT CURRENT_TIMESTAMP );这张表结构简单但已经能满足两个核心需求按照设备和时间筛选告警以及查看告警对应的截图路径。实际项目中可以增加video_segment_url字段来保存告警前后几秒的视频片段。使用SQLAlchemy实现时会这样定义模型from sqlalchemy import Column, Integer, String, Float, DateTime from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base declarative_base() class Alert(Base): __tablename__ alerts id Column(Integer, primary_keyTrue, autoincrementTrue) device_id Column(String(64), nullableFalse) class_name Column(String(64), nullableFalse) confidence Column(Float, nullableFalse) image_path Column(String(255)) created_at Column(DateTime, defaultdatetime.now)保存告警的代码只需要构造Alert对象并提交会话。这里要注意不要把数据库提交放在高频率的推理循环里否则大量写操作会阻塞主链路。建议在告警冷却时间内只写一次或者把写入操作放到独立线程池。3.5 主流程组装main.py中需要把上面的模块串起来。下面是一个最小主循环示例import os from app.stream import StreamReader from app.detector import Detector from app.alert import AlertManager from app.models import Alert from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker SOURCE os.getenv(STREAM_SOURCE, ./data/demo.mp4) MODEL_PATH os.getenv(MODEL_PATH, ./models/yolov8n.pt) WEBHOOK_URL os.getenv(ALERT_WEBHOOK_URL, ) DEVICE_ID os.getenv(DEVICE_ID, gate-01) stream StreamReader(SOURCE, interval0.05) detector Detector(MODEL_PATH) alert_manager AlertManager(WEBHOOK_URL, cooldown30) engine create_engine(sqlite:///alerts.db) Base.metadata.create_all(engine) Session sessionmaker(bindengine) while True: frame stream.get_frame() if frame is None: continue detections detector.predict(frame) enabled_classes {person, car} if alert_manager.should_alert(detections, enabled_classes): alert Alert( device_idDEVICE_ID, class_namedetections[0][class_name], confidencedetections[0][confidence], image_path/tmp/alert.jpg, ) session Session() session.add(alert) session.commit() session.close() alert_manager.send({text: f设备{DEVICE_ID}出现目标})这个循环展示了完整链路取帧、推理、判断告警、落库、通知。实际项目里还可以把循环放到独立线程避免阻塞FastAPI的请求处理。4. 可视化看板把监控状态变成可读界面4.1 后端接口设计为了让前端展示告警列表后端需要提供查询接口。FastAPI示例from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine app FastAPI() engine create_engine(sqlite:///alerts.db) Session sessionmaker(bindengine) app.get(/api/alerts) def list_alerts(limit: int 50): session Session() rows session.query(Alert).order_by(Alert.created_at.desc()).limit(limit).all() data [ { id: r.id, device_id: r.device_id, class_name: r.class_name, confidence: r.confidence, image_path: r.image_path, created_at: r.created_at.strftime(%Y-%m-%d %H:%M:%S), } for r in rows ] session.close() return {code: 0, data: data} app.mount(/, StaticFiles(directorystatic, htmlTrue), namestatic)这个接口返回最近50条告警前端可以轮询调用。code字段用于约定业务状态0表示正常。实际项目中还要支持分页、按设备过滤、按时间范围查询。4.2 前端页面与实时刷新静态页面index.html不需要复杂框架。一个最简单的看板逻辑是用setInterval每5秒请求一次告警列表然后渲染表格。!DOCTYPE html html langzh-CN head meta charsetUTF-8 title监控告警看板/title /head body h2实时告警列表/h2 table idalertTable border1 thead trthID/thth设备/thth目标/thth置信度/thth时间/thth截图/th/tr /thead tbody/tbody /table script function loadAlerts() { fetch(/api/alerts?limit20) .then(res res.json()) .then(res { const tbody document.querySelector(#alertTable tbody); tbody.innerHTML ; res.data.forEach(row { const tr document.createElement(tr); tr.innerHTML td${row.id}/tdtd${row.device_id}/tdtd${row.class_name}/td td${row.confidence.toFixed(2)}/tdtd${row.created_at}/td td${row.image_path ? a href${row.image_path} target_blank查看/a : }/td; tbody.appendChild(tr); }); }); } setInterval(loadAlerts, 5000); loadAlerts(); /script /body /html实际生产环境可以引入Vue或React增加画面轮播、统计图表、告警声音提醒等功能。但最基本的看板只需要一个接口加一个表格重点是让数据流先跑通。前端轮询间隔不建议短于2秒否则会频繁请求接口。如果需要更高实时性可以改用WebSocket或SSE由后端在告警写入时主动推送事件。4.3 运行验证从启动到告警闭环当所有代码就绪后按以下顺序启动并验证。第一步确认依赖已安装pip install -r requirements.txt第二步启动FastAPI服务uvicorn app.main:app --host 0.0.0.0 --port 8000第三步在另一个终端启动推理任务。如果main.py中已经写了入口可以直接运行python -m app.main如果一切正常控制台会输出视频帧读取日志和告警日志。在检测到目标且达到告警条件时alerts.db中会新增记录。打开浏览器访问http://127.0.0.1:8000看板会显示告警列表。需要重点验证的不只是“程序能启动”还包括视频源是否稳定读取。检测结果中的目标类别是否符合预期。告警是否触发是否受到冷却时间控制。数据库是否写入告警记录。前端能否正确展示并刷新。如果某个环节没有出现按照后文排查路径逐层检查。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。等到生产环境再补验证成本会高得多。5. 常见问题排查视频流、检测、告警和性能5.1 视频流拉取失败现象程序启动后没有日志输出cap.read()一直返回False或者控制台报错无法打开摄像头。问题现象常见原因检查方式处理建议本地视频文件无法打开路径错误或文件损坏检查文件是否存在用播放器打开修正路径使用可正常播放的MP4RTSP地址无法连接设备未开机、网络不通、账号密码错误ping设备IP用VLC测试RTSP地址使用VLC确认地址可播放后再填入代码读取几秒后停止
当讨论“天网”时技术人员真正关心的问题通常是一套自动化监控预警系统如何从视频流采集开始一步步完成目标识别、告警通知和数据沉淀。科幻作品里的天网是一个拥有自主意识的全球控制系统而在工程实践中我们所说的天网更接近一套由摄像头、视频解码、AI模型、消息中间件和可视化看板组成的智能安防监控架构。这类系统已经在工厂安全生产、园区周界防护、仓库消防预警等场景中大量落地。本文假设读者已经掌握基本的Python语法了解HTTP接口和数据库基础知识但不要求读者有算法背景。读完后你会理解监控预警系统的核心链路能搭建一个最小可运行项目并在遇到视频流拉取失败、模型推理卡顿、告警重复等问题时知道从哪里排查。1. 把“天网”拉回工程语境智能监控预警系统到底在做什么1.1 从科幻名词到工程架构在科幻设定中天网是一个可以自主感知、决策并执行动作的人工智能系统它有摄像头、传感器、武器系统和数据中心。如果把“自主决策”和“执行动作”去掉只保留“感知、识别、告警、留痕”这四个能力天网式系统就变成了一套企业级安防监控平台。这类平台通常不具备自主意识也不会直接控制物理设备。它做的事情是从摄像头不断读取画面用AI模型判断画面中是否存在目标对象一旦目标触发规则就生成一条告警记录并通知相关人员。整个过程是确定性的、可配置的也是可关闭的。理解这一点就不会把工程问题神秘化。在项目里将新系统称为“天网”更多是一种类比它像天网一样拥有覆盖全场的“眼睛”但它的行为边界由工程师定义。实际项目中摄像头只负责采集画面AI模型只负责给出置信度告警规则由业务人员配置管理人员可以通过开关随时干预。1.2 落地场景与业务价值最典型的落地场景包括几类工厂安全帽检测在作业区域检测未佩戴安全帽的人员触发告警减少安全事故。园区周界防护在非开放时段检测人员靠近或跨越围栏。仓库烟火预警检测烟雾和明火第一时间通知消防值班人员。停车区域管理识别占用消防通道的车辆。生产线行为规范检测操作人员是否进入危险区域。这些场景的共同特点是监控点位很多人眼无法长时间盯屏事件出现时间短错过之后很难回溯传统摄像头只能录像不能主动发现问题。智能监控预警系统把“事后查录像”变成“事中告警、事后可查”这是它最大的业务价值。从技术角度看这类系统的核心挑战不在单一环节而在链路协同视频采集要稳定AI推理要在可接受的延迟内完成告警要准确且不刷屏数据要能回溯界面要让值班人员快速理解发生了什么。1.3 系统中的关键角色一套完整的监控预警系统可以拆成六个角色每个角色的实现方式可以不同但这个分工在大多数项目里一致。角色职责常见实现视频源输出连续画面IPC摄像头、NVR、本地视频文件流媒体服务统一接入并转发视频流RTSP/RTMP服务、FFmpeg推理服务对视频帧执行AI识别YOLO、OpenCV、TensorRT告警服务判断规则、生成告警、发送通知Python后台任务、Webhook、消息队列数据存储保存告警、事件和图片证据MySQL、PostgreSQL、对象存储可视化看板展示实时状态和历史查询Web前后端、大屏系统1.4 为什么不能只靠普通录像普通数字录像系统只做两件事存储码流和回放画面。它不判断画面中是否存在异常也不主动通知任何人。这意味着异常事件发生后负责人必须等操作人员上报然后人工翻查录像效率很低。而智能监控预警系统在画面采集到后立即做一次推理判断把结果写入结构化数据同时触发通知。排查问题时可以直接检索告警时间、设备和目标类型不再依赖肉眼回放。1.5 技术路线选择实现这套系统常见有三种路线。第一种是使用厂商私有协议和配套平台部署简单但扩展性差难以接入异构设备。第二种是使用开源流媒体服务和自研AI推理模块灵活度高但需要团队维护多个组件。第三种是直接用云服务成本低、免运维但视频数据出域可能不适合部分数据敏感场景。本文采用第二种路线原因是它最适合理解技术链路也方便在开发环境随意调试。2. 环境准备硬件、依赖和项目结构先对齐2.1 开发环境与依赖列表搭建最小系统不需要很高配置。建议开发机至少具备4核CPU和8GB内存如果使用GPU推理显存建议6GB以上。操作系统推荐Ubuntu 20.04或Windows 10以上版本。运行环境建议使用Python 3.10因为较新的推理库对旧版本支持逐渐减弱。主要依赖库如下表。这里的版本是常见组合实际落地前要结合自己的操作系统和官方文档确认。软件/库用途建议版本范围Python编写服务和推理逻辑3.10OpenCV-Python读取视频流、处理图像4.8.x 或 4.9.xPyTorch加载并运行YOLO模型2.xUltralytics YOLO目标检测封装库8.xFastAPI提供HTTP接口0.104SQLAlchemy操作数据库2.xMySQL / SQLite保存告警数据MySQL 8.x 或内置SQLite这些库并不全是唯一选择。OpenCV可以换成FFmpeg命令行工具PyTorch可以换成ONNX RuntimeFastAPI也可以换成Flask。判断标准是团队熟悉度与部署环境兼容性。2.2 模拟视频源与流媒体服务实际项目中视频来自IPC摄像头地址通常是rtsp://user:passwordip:554/stream1。开发阶段如果没有摄像头可以用本地视频文件作为输入也可以用OpenCV把一段循环视频模拟成RTSP流。最简单的开发方式是不搭建流媒体服务直接用OpenCV读取本地视频import cv2 video_path ./demo.mp4 cap cv2.VideoCapture(video_path) while cap.isOpened(): ret, frame cap.read() if not ret: cap.set(cv2.CAP_PROP_POS_FRAMES, 0) continue cv2.imshow(frame, frame) if cv2.waitKey(1) 0xFF ord(q): break cap.release() cv2.destroyAllWindows()这段代码循环读取同一个视频文件模拟持续的视频流。对于验证AI推理和告警链路这种方式足够。要注意本地文件的读取速度会远超实时帧率如果代码里不添加按帧率控制的等待时间推理服务会以极高速度处理造成CPU占满。如果必须验证RTSP协议可以使用FFmpeg搭建本地流媒体服务ffmpeg -re -stream_loop -1 -i demo.mp4 -rtsp_transport tcp -f rtsp rtsp://127.0.0.1:8554/live然后OpenCV读取rtsp://127.0.0.1:8554/live。-stream_loop -1表示无限循环-re表示按原始帧率读取避免发送过快。2.3 项目目录结构为了让后续实现不混乱建议按下面的目录组织工程。这个结构不是唯一答案但功能边界清晰适合逐步扩展。smart-surveillance/ ├── app/ │ ├── main.py # FastAPI入口负责接口和静态页面 │ ├── detector.py # 模型加载和帧检测封装 │ ├── stream.py # 视频流读取与帧管理 │ ├── alert.py # 告警规则判断与通知 │ ├── models.py # 数据库模型 │ └── config.py # 配置项 ├── models/ │ └── yolov8n.pt # 目标检测权重文件 ├── data/ │ └── demo.mp4 # 测试视频 ├── static/ │ └── index.html # 可视化看板 ├── requirements.txt └── README.md这个结构中stream.py只负责提供视频帧detector.py只负责推理alert.py只负责告警逻辑main.py负责把组件组织成服务。职责分离后想要替换摄像头或换模型不会牵连整个项目。3. 核心实现从视频流采集到AI告警的完整链路3.1 视频流接入与帧提取在stream.py中封装一个视频帧读取器。关键点是读取视频流不能阻塞主逻辑因此通常放到后台线程里持续读取并且只保留最新帧避免内存不断上涨。import cv2 import threading import time class StreamReader: def __init__(self, source: str, interval: float 0.03): self.cap cv2.VideoCapture(source) self.interval interval self.lock threading.Lock() self.frame None self.running True threading.Thread(targetself._loop, daemonTrue).start() def _loop(self): while self.running: ret, frame self.cap.read() if not ret: self.cap.set(cv2.CAP_PROP_POS_FRAMES, 0) continue with self.lock: self.frame frame time.sleep(self.interval) def get_frame(self): with self.lock: return self.frame.copy() if self.frame is not None else None def release(self): self.running False self.cap.release()为什么要用锁因为推理服务在另一个线程里调用get_frame如果直接读写self.frame可能会出现读取到不完整图像的情况。interval控制处理速度一般结合视频帧率调整例如25FPS对应0.04秒。实际项目中还需要考虑摄像头断线重连。可以在_loop里检测cap.isOpened()如果断开就cap.open(source)重试并增加重试间隔避免日志刷屏。3.2 AI目标检测模型接入在detector.py中使用YOLO模型。以Ultralytics YOLO为例加载模型并封装一个predict方法from ultralytics import YOLO class Detector: def __init__(self, model_path: str): self.model YOLO(model_path) def predict(self, frame): results self.model.predict(frame, verboseFalse) detections [] for r in results: for box in r.boxes: detections.append({ class_id: int(box.cls[0]), class_name: r.names[int(box.cls[0])], confidence: float(box.conf[0]), x: int(box.xyxy[0][0]), y: int(box.xyxy[0][1]), w: int(box.xyxy[0][2] - box.xyxy[0][0]), h: int(box.xyxy[0][3] - box.xyxy[0][1]), }) return detections这里返回的目标框坐标格式是x, y, w, h方便后续在画面上绘制矩形。confidence是置信度取值0到1。置信度越低漏报越少但误报越多置信度越高误报越少但漏报可能增加。在开发环境模型文件yolov8n.pt可以从Ultralytics官方仓库下载也可以换成自己训练的权重。初次加载模型时会下载权重网络环境不同可能影响耗时。如果使用GPU推理需要在安装PyTorch时选择对应CUDA版本否则CPU推理也能运行只是帧率较低。3.3 告警触发与消息通知告警逻辑不应该简单写成“检测到目标就发送消息”。如果目标连续出现在几十帧中系统会发送大量重复告警。常见做法是增加冷却时间和目标区域过滤。alert.py的核心逻辑如下import time import requests class AlertManager: def __init__(self, webhook_url: str, cooldown: int 30): self.webhook_url webhook_url self.cooldown cooldown self.last_alert_time 0 def should_alert(self, detections, enabled_classes) - bool: if not detections: return False now time.time() if now - self.last_alert_time self.cooldown: return False for det in detections: if det[class_name] in enabled_classes and det[confidence] 0.5: self.last_alert_time now return True return False def send(self, message: dict): if self.webhook_url: requests.post(self.webhook_url, jsonmessage, timeout3) print(alert:, message)冷却时间cooldown的单位是秒表示同一事件在冷却时间内只通知一次。对于安全帽检测冷却时间可以设为10到30秒对于烟火检测可以适当缩短。enabled_classes用于过滤需要触发告警的目标类别避免“人”这种常见目标频繁告警。发送消息时可以包含告警时间、设备名称、目标类别、置信度、现场截图地址等信息。截图可以在告警时把当前帧保存为图片并上传到对象存储或本地目录方便值班人员查看。def build_message(det, image_path, device_id): return { msgtype: text, text: { content: f【安防告警】设备:{device_id} 检测到{det[class_name]} 置信度:{det[confidence]:.2f} 截图:{image_path} } }如果是企业微信或钉钉机器人可以直接使用Webhook地址如果是自研平台可以调用自己的告警接口。生产环境中不建议在告警逻辑里同步发送HTTP请求因为网络延迟会占用推理线程。更好的方式是把告警写入消息队列由独立消费者负责通知。3.4 告警数据落库与查询告警数据需要长期保存。为了方便演示这里使用SQLite通过SQLAlchemy定义一张告警表。CREATE TABLE alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id VARCHAR(64) NOT NULL, class_name VARCHAR(64) NOT NULL, confidence REAL NOT NULL, image_path VARCHAR(255), created_at DATETIME DEFAULT CURRENT_TIMESTAMP );这张表结构简单但已经能满足两个核心需求按照设备和时间筛选告警以及查看告警对应的截图路径。实际项目中可以增加video_segment_url字段来保存告警前后几秒的视频片段。使用SQLAlchemy实现时会这样定义模型from sqlalchemy import Column, Integer, String, Float, DateTime from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base declarative_base() class Alert(Base): __tablename__ alerts id Column(Integer, primary_keyTrue, autoincrementTrue) device_id Column(String(64), nullableFalse) class_name Column(String(64), nullableFalse) confidence Column(Float, nullableFalse) image_path Column(String(255)) created_at Column(DateTime, defaultdatetime.now)保存告警的代码只需要构造Alert对象并提交会话。这里要注意不要把数据库提交放在高频率的推理循环里否则大量写操作会阻塞主链路。建议在告警冷却时间内只写一次或者把写入操作放到独立线程池。3.5 主流程组装main.py中需要把上面的模块串起来。下面是一个最小主循环示例import os from app.stream import StreamReader from app.detector import Detector from app.alert import AlertManager from app.models import Alert from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker SOURCE os.getenv(STREAM_SOURCE, ./data/demo.mp4) MODEL_PATH os.getenv(MODEL_PATH, ./models/yolov8n.pt) WEBHOOK_URL os.getenv(ALERT_WEBHOOK_URL, ) DEVICE_ID os.getenv(DEVICE_ID, gate-01) stream StreamReader(SOURCE, interval0.05) detector Detector(MODEL_PATH) alert_manager AlertManager(WEBHOOK_URL, cooldown30) engine create_engine(sqlite:///alerts.db) Base.metadata.create_all(engine) Session sessionmaker(bindengine) while True: frame stream.get_frame() if frame is None: continue detections detector.predict(frame) enabled_classes {person, car} if alert_manager.should_alert(detections, enabled_classes): alert Alert( device_idDEVICE_ID, class_namedetections[0][class_name], confidencedetections[0][confidence], image_path/tmp/alert.jpg, ) session Session() session.add(alert) session.commit() session.close() alert_manager.send({text: f设备{DEVICE_ID}出现目标})这个循环展示了完整链路取帧、推理、判断告警、落库、通知。实际项目里还可以把循环放到独立线程避免阻塞FastAPI的请求处理。4. 可视化看板把监控状态变成可读界面4.1 后端接口设计为了让前端展示告警列表后端需要提供查询接口。FastAPI示例from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine app FastAPI() engine create_engine(sqlite:///alerts.db) Session sessionmaker(bindengine) app.get(/api/alerts) def list_alerts(limit: int 50): session Session() rows session.query(Alert).order_by(Alert.created_at.desc()).limit(limit).all() data [ { id: r.id, device_id: r.device_id, class_name: r.class_name, confidence: r.confidence, image_path: r.image_path, created_at: r.created_at.strftime(%Y-%m-%d %H:%M:%S), } for r in rows ] session.close() return {code: 0, data: data} app.mount(/, StaticFiles(directorystatic, htmlTrue), namestatic)这个接口返回最近50条告警前端可以轮询调用。code字段用于约定业务状态0表示正常。实际项目中还要支持分页、按设备过滤、按时间范围查询。4.2 前端页面与实时刷新静态页面index.html不需要复杂框架。一个最简单的看板逻辑是用setInterval每5秒请求一次告警列表然后渲染表格。!DOCTYPE html html langzh-CN head meta charsetUTF-8 title监控告警看板/title /head body h2实时告警列表/h2 table idalertTable border1 thead trthID/thth设备/thth目标/thth置信度/thth时间/thth截图/th/tr /thead tbody/tbody /table script function loadAlerts() { fetch(/api/alerts?limit20) .then(res res.json()) .then(res { const tbody document.querySelector(#alertTable tbody); tbody.innerHTML ; res.data.forEach(row { const tr document.createElement(tr); tr.innerHTML td${row.id}/tdtd${row.device_id}/tdtd${row.class_name}/td td${row.confidence.toFixed(2)}/tdtd${row.created_at}/td td${row.image_path ? a href${row.image_path} target_blank查看/a : }/td; tbody.appendChild(tr); }); }); } setInterval(loadAlerts, 5000); loadAlerts(); /script /body /html实际生产环境可以引入Vue或React增加画面轮播、统计图表、告警声音提醒等功能。但最基本的看板只需要一个接口加一个表格重点是让数据流先跑通。前端轮询间隔不建议短于2秒否则会频繁请求接口。如果需要更高实时性可以改用WebSocket或SSE由后端在告警写入时主动推送事件。4.3 运行验证从启动到告警闭环当所有代码就绪后按以下顺序启动并验证。第一步确认依赖已安装pip install -r requirements.txt第二步启动FastAPI服务uvicorn app.main:app --host 0.0.0.0 --port 8000第三步在另一个终端启动推理任务。如果main.py中已经写了入口可以直接运行python -m app.main如果一切正常控制台会输出视频帧读取日志和告警日志。在检测到目标且达到告警条件时alerts.db中会新增记录。打开浏览器访问http://127.0.0.1:8000看板会显示告警列表。需要重点验证的不只是“程序能启动”还包括视频源是否稳定读取。检测结果中的目标类别是否符合预期。告警是否触发是否受到冷却时间控制。数据库是否写入告警记录。前端能否正确展示并刷新。如果某个环节没有出现按照后文排查路径逐层检查。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。等到生产环境再补验证成本会高得多。5. 常见问题排查视频流、检测、告警和性能5.1 视频流拉取失败现象程序启动后没有日志输出cap.read()一直返回False或者控制台报错无法打开摄像头。问题现象常见原因检查方式处理建议本地视频文件无法打开路径错误或文件损坏检查文件是否存在用播放器打开修正路径使用可正常播放的MP4RTSP地址无法连接设备未开机、网络不通、账号密码错误ping设备IP用VLC测试RTSP地址使用VLC确认地址可播放后再填入代码读取几秒后停止
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
