别再手动巡检了用YOLO26Flask30分钟给工地装个‘AI安全员’附完整代码想象一下清晨六点工地刚刚苏醒塔吊开始转动工人们陆续进场。作为项目经理你最揪心的是什么是某个角落的工人因为匆忙忘了戴安全帽而巡检员半小时后才可能走到那里。传统的安全管理依赖人力存在盲区、滞后和疲劳问题。但现在技术给了我们一双永不疲倦的“眼睛”。这篇文章不是写给研究员的论文而是给每一位希望用技术解决实际业务痛点的项目经理、技术负责人和行动派工程师的实战手册。我们将绕过繁琐的理论和复杂的环境配置直接聚焦于如何用最新的YOLO26目标检测模型和轻量级的Flask框架在极短的时间内构建一个能真正“上岗”的智能安全监控原型系统。这个系统能实时分析监控画面自动识别未佩戴安全帽的行为并即时推送告警把安全隐患扼杀在萌芽状态。我们追求的不是学术精度上的小数点后几位而是开箱即用、快速集成和稳定运行。1. 从业务痛点到技术方案为什么是YOLO26Flask在建筑、电力、矿山等高危行业安全帽是生命的“最后一道防线”。但人工监督成本高、覆盖难事后查录像更是效率低下。我们需要的是一个能7x24小时在线的“AI安全员”。这个角色需要具备几个核心能力看得准在各种光照、遮挡下都能识别、反应快实时或近实时处理、好集成能轻松对接现有监控平台或通讯工具、易部署技术团队能快速上手无需深厚AI背景。YOLO26作为YOLO系列的最新成员在速度和精度的平衡上做出了显著改进。对于安全帽识别这个具体场景它的优势非常突出小目标检测能力强工地监控中远处的工人头部在画面中可能只占几十个像素。YOLO26通过改进的网络结构和训练策略提升了对这类小目标的捕捉能力。兼顾姿态估计这是其一大亮点。它不仅能框出“安全帽”还能通过检测头部关键点初步判断佩戴是否规范例如帽子是否戴歪、仅挂在后脑勺这比简单的“有/无”二分类提供了更深一层的安全洞察。推理效率高在主流GPU上能达到很高的帧率满足多路摄像头实时分析的需求。而Flask作为一个轻量级的Python Web框架则是连接AI模型与业务应用的“桥梁”。它的价值在于极简封装用十几行代码就能将模型封装成HTTP API供其他系统如门禁、大屏、移动APP调用。灵活轻便不依赖重型应用服务器资源占用少非常适合在边缘计算设备如工控机、NVIDIA Jetson上部署。快速原型让工程师能专注于业务逻辑如告警规则、结果推送而非复杂的Web开发。这个组合恰好满足了“快速验证、低成本试错、平滑集成”的工程化需求。2. 30分钟极速部署让你的“AI安全员”先跑起来我们信奉“Seeing is Believing”。在深入细节之前先让整个系统动起来建立直观感受。这里假设你已经有一台安装了Python和CUDA的Linux服务器或高性能PCWindows下步骤类似。2.1 环境准备一步到位的依赖安装为了避免陷入“依赖地狱”我们使用requirements.txt文件来管理所有包。创建一个项目目录并在其中创建该文件# requirements.txt torch1.10.0, 2.0.0 torchvision0.11.0, 0.16.0 ultralytics8.4.2 # 包含YOLO26 flask2.0.0 opencv-python-headless4.5.0 # 服务器环境用headless版本 pillow9.0.0 requests2.28.0 # 用于后续告警推送 numpy1.21.0在终端中使用以下命令创建虚拟环境并安装依赖强烈推荐使用虚拟环境以隔离项目# 创建并激活虚拟环境以venv为例 python -m venv yolo26_env source yolo26_env/bin/activate # Linux/macOS # 对于Windows: yolo26_env\Scripts\activate # 安装依赖 pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple注意ultralytics包会自动处理YOLO模型的下载。首次运行时会从官方仓库下载yolo26n.pt等预训练权重请确保网络通畅。2.2 核心脚本五步搭建Flask推理服务现在创建我们的主应用文件app.py。这个文件将完成模型加载、图像接收、推理和结果返回的全流程。# app.py import os import cv2 import numpy as np from flask import Flask, request, jsonify, render_template from ultralytics import YOLO from datetime import datetime import logging # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app Flask(__name__) # 全局加载模型服务启动时加载一次 # 使用YOLO26n的预训练模型它足够轻量且准确 # 后续可以替换为你自己训练的安全帽专用模型 MODEL_PATH yolo26n.pt # 或 path/to/your/custom_model.pt model YOLO(MODEL_PATH) logger.info(f模型 {MODEL_PATH} 加载成功。) # 定义我们关心的类别COCO数据集中person0 # 注意预训练模型没有‘helmet’类这里我们先演示人员检测下一节会替换为定制模型。 TARGET_CLASS_NAMES [person] # 后续替换为 [helmet, person] TARGET_CLASS_IDS [0] # 后续根据自定义数据调整 app.route(/) def index(): 提供一个简单的测试页面 return render_template(index.html) # 需要创建 templates/index.html app.route(/health) def health_check(): 健康检查端点用于运维 return jsonify({status: healthy, model: MODEL_PATH}) app.route(/detect, methods[POST]) def detect(): 核心检测API。 接收表单中的图片文件字段名‘image’或图片URL字段名‘url’。 返回检测到的目标信息。 start_time datetime.now() results [] try: if image in request.files: # 方式一通过文件上传 file request.files[image] img_bytes file.read() img_np np.frombuffer(img_bytes, np.uint8) img cv2.imdecode(img_np, cv2.IMREAD_COLOR) source_desc fuploaded file: {file.filename} elif url in request.form: # 方式二通过图片URL需要安装requests已在依赖中 import requests img_url request.form[url] resp requests.get(img_url, streamTrue, timeout10) resp.raise_for_status() img_array np.asarray(bytearray(resp.content), dtypenp.uint8) img cv2.imdecode(img_array, cv2.IMREAD_COLOR) source_desc fimage url: {img_url} else: return jsonify({error: 请提供‘image’文件或‘url’参数}), 400 if img is None: return jsonify({error: 无法解码图片}), 400 # 执行YOLO推理 # conf: 置信度阈值过滤掉不可信的检测框 # iou: NMS的IoU阈值用于合并重叠框 inference_results model.predict( sourceimg, conf0.5, iou0.6, classesTARGET_CLASS_IDS, # 只检测我们感兴趣的类别 verboseFalse # 不输出详细日志到控制台 ) # 解析结果 for r in inference_results: if r.boxes is not None: boxes r.boxes.xyxy.cpu().numpy() # 边界框 [x1, y1, x2, y2] confidences r.boxes.conf.cpu().numpy() class_ids r.boxes.cls.cpu().numpy().astype(int) for box, conf, cls_id in zip(boxes, confidences, class_ids): # 将numpy类型转换为Python原生类型便于JSON序列化 result { bbox: [int(box[0]), int(box[1]), int(box[2]), int(box[3])], confidence: float(conf), class_id: int(cls_id), class_name: model.names[int(cls_id)] } results.append(result) # 可选在图片上绘制检测框并保存用于调试 if results and request.args.get(save_image, false).lower() true: output_img r.plot() # ultralytics提供的便捷绘图方法 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) output_path fdetections/detected_{timestamp}.jpg os.makedirs(detections, exist_okTrue) cv2.imwrite(output_path, output_img) logger.info(f结果图片已保存至: {output_path}) process_time (datetime.now() - start_time).total_seconds() logger.info(f检测完成处理{source_desc}耗时{process_time:.3f}秒发现{len(results)}个目标。) return jsonify({ success: True, detections: results, count: len(results), process_time_seconds: process_time }) except Exception as e: logger.error(f检测过程中发生错误: {e}, exc_infoTrue) return jsonify({success: False, error: str(e)}), 500 if __name__ __main__: # 启动Flask开发服务器 # host0.0.0.0 允许外部网络访问生产环境应使用NginxGunicorn等WSGI服务器 app.run(host0.0.0.0, port5000, debugFalse) # 生产环境务必设置 debugFalse同时创建一个简单的templates/index.html用于测试!DOCTYPE html html head titleYOLO26安全帽检测测试页/title /head body h2上传图片测试安全帽检测API/h2 form action/detect methodpost enctypemultipart/form-data input typefile nameimage acceptimage/* brbr input typesubmit value开始检测 /form hr h3或通过图片URL测试/h3 form action/detect methodpost input typetext nameurl size50 placeholder输入图片URL例如: https://example.com/site.jpg brbr input typesubmit value检测URL图片 /form p查看 a href/health服务健康状态/a/p /body /html2.3 启动与测试在项目根目录下执行python app.py看到类似* Running on http://0.0.0.0:5000的输出后服务就启动了。现在打开浏览器访问http://你的服务器IP:5000上传一张包含人物的工地图片。点击提交后页面会返回一个JSON响应里面包含了检测到的每个目标的坐标、置信度和类别。恭喜你的第一个“AI安全员”后端服务已经上线了。它已经可以接收图片并返回检测结果。但这只是个开始它现在还只能识别人不认识安全帽。接下来我们要赋予它真正的“专业能力”。3. 从“识人”到“识帽”定制专属安全帽检测模型预训练的YOLO26模型是在COCO等通用数据集上训练的并不认识“安全帽”这个特定类别。我们必须用自己的工地图片训练它这个过程称为“微调”。3.1 数据准备给AI准备“教材”高质量的数据是模型好坏的基石。你需要收集和标注工地现场的图片。数据收集建议来源多样包含白天、黄昏、夜间、晴天、阴雨等不同光照包含近景、中景、远景包含工人戴帽、未戴帽、戴帽不规范等场景。数量要求一个可用的模型至少需要500-1000张标注图片。理想情况是2000张以上并按8:1:1划分训练集、验证集和测试集。标注工具推荐使用Roboflow、CVAT或LabelImg。它们都支持导出YOLO格式。YOLO格式数据集结构your_helmet_dataset/ ├── images/ │ ├── train/ # 训练集图片 │ │ ├── img_001.jpg │ │ └── ... │ └── val/ # 验证集图片 │ ├── img_101.jpg │ └── ... ├── labels/ │ ├── train/ # 训练集标签与图片同名扩展名为.txt │ │ ├── img_001.txt │ │ └── ... │ └── val/ # 验证集标签 │ ├── img_101.txt │ └── ... └── dataset.yaml # 数据集配置文件每个.txt标签文件的内容格式为class_id x_center y_center width height坐标是归一化后的0到1之间。例如0 0.45 0.32 0.1 0.15表示类别0安全帽的中心点位于图片宽度的45%、高度的32%处其宽度和高度分别占图片宽高的10%和15%。关键的dataset.yaml文件# dataset.yaml path: /absolute/path/to/your_helmet_dataset # 数据集的根目录绝对路径 train: images/train # 训练集路径相对于path val: images/val # 验证集路径相对于path # 类别信息 nc: 2 # 类别数量 names: [helmet, person] # 类别名称顺序对应 class_id 0 和 13.2 模型训练让AI学习“安全规范”准备好数据后创建训练脚本train.py# train.py from ultralytics import YOLO import os def main(): # 1. 加载一个YOLO26模型架构这里以yolo26n为例平衡速度与精度 # 使用‘.yaml’文件初始化模型结构而不是预训练权重‘.pt’ model YOLO(yolo26n.yaml) # 这会创建一个随机初始化的模型 # 2. 开始训练 results model.train( datapath/to/your/dataset.yaml, # 上一步创建的配置文件路径 epochs100, # 训练轮数根据数据集大小调整 patience30, # 早停耐心值如果30轮验证指标没提升就停止 batch32, # 批次大小根据GPU显存调整16, 32, 64 imgsz640, # 输入图片尺寸 workers4, # 数据加载线程数 device0, # 使用GPU 0如果是CPU则设为‘cpu’ projecthelmet_detection_project, # 项目文件夹名称 nameexp1, # 实验名称 exist_okTrue, # 允许覆盖已存在的实验目录 pretrainedTrue, # 从官方预训练权重开始微调这是关键 optimizerAdamW, # 优化器 lr00.01, # 初始学习率 lrf0.01, # 最终学习率因子 (lr0 * lrf) momentum0.937, # SGD动量 weight_decay0.0005, # 权重衰减 warmup_epochs3.0, # 学习率预热轮数 box7.5, # 边界框损失权重 cls0.5, # 分类损失权重 dfl1.5, # DFL损失权重 pose12.0, # 姿态损失权重如果做姿态估计 kobj2.0, # 关键点对象性损失权重 fl_gamma0.0, # Focal loss gamma label_smoothing0.0, # 标签平滑 nbs64, # 名义批次大小 overlap_maskTrue, scale0.5, dropout0.0, valTrue, # 训练中验证 saveTrue, # 保存检查点 save_period-1, deterministicFalse, single_clsFalse, rectFalse, cos_lrFalse, close_mosaic10, resumeFalse, ampTrue, # 自动混合精度训练节省显存加速训练 ) print(训练完成) if __name__ __main__: main()运行训练python train.py训练过程会在helmet_detection_project/exp1目录下生成大量日志和结果。最重要的输出是weights/best.pt这是验证集上表现最好的模型我们将用它来更新我们的Flask服务。3.3 更新服务换上“专业”模型训练完成后修改app.py中的MODEL_PATH指向你新训练好的模型MODEL_PATH helmet_detection_project/exp1/weights/best.pt同时更新类别信息TARGET_CLASS_NAMES [helmet, person] # 现在模型认识这两个类了 # 你需要根据训练时 dataset.yaml 中 ‘names’ 的顺序来确定ID # 假设 ‘names‘ 是 [‘helmet‘, ‘person‘]那么 helmet的ID是0person是1。 TARGET_CLASS_IDS [0, 1] # 我们同时检测安全帽和人用于后续关联分析重启Flask服务现在你的API就能准确识别图片中的安全帽和人员了。4. 从单张图片到实时视频流处理摄像头RTSP流工地监控系统通常通过RTSP协议输出视频流。我们需要让Flask服务能处理这种持续的流数据。这里介绍两种方案。4.1 方案一服务端拉流分析推荐用于集中式分析在Flask服务中创建一个后台线程持续拉取RTSP流并进行分析。创建一个新文件stream_processor.py# stream_processor.py import cv2 import threading import time from queue import Queue from ultralytics import YOLO import logging logger logging.getLogger(__name__) class RTSPStreamProcessor: def __init__(self, rtsp_url, model_path, frame_queue_maxsize10): self.rtsp_url rtsp_url self.model YOLO(model_path) self.cap None self.frame_queue Queue(maxsizeframe_queue_maxsize) self.is_running False self.process_thread None self.detection_thread None self.last_results [] # 存储最新的检测结果 def start(self): 启动拉流和检测线程 self.cap cv2.VideoCapture(self.rtsp_url) if not self.cap.isOpened(): logger.error(f无法打开RTSP流: {self.rtsp_url}) return False self.is_running True # 线程1拉流并放入队列 self.process_thread threading.Thread(targetself._capture_frames, daemonTrue) # 线程2从队列取帧进行检测 self.detection_thread threading.Thread(targetself._detect_frames, daemonTrue) self.process_thread.start() self.detection_thread.start() logger.info(fRTSP流处理器已启动: {self.rtsp_url}) return True def _capture_frames(self): 持续从RTSP拉取帧 while self.is_running: ret, frame self.cap.read() if not ret: logger.warning(从RTSP流读取帧失败尝试重连...) time.sleep(2) self._reconnect() continue # 如果队列满了丢弃最旧的帧保证实时性 if self.frame_queue.full(): try: self.frame_queue.get_nowait() except: pass self.frame_queue.put(frame.copy()) # 使用copy避免引用问题 time.sleep(0.03) # 控制拉流频率约30FPS def _detect_frames(self): 持续从队列中取帧进行检测 while self.is_running: if not self.frame_queue.empty(): frame self.frame_queue.get() # 使用YOLO进行推理 results self.model.predict( sourceframe, conf0.5, iou0.6, classes[0, 1], # helmet和person verboseFalse, streamFalse # 单帧模式 ) # 解析并存储结果 detections [] for r in results: if r.boxes is not None: for box, conf, cls_id in zip(r.boxes.xyxy, r.boxes.conf, r.boxes.cls): detections.append({ bbox: box.cpu().numpy().astype(int).tolist(), confidence: float(conf), class_id: int(cls_id), class_name: self.model.names[int(cls_id)] }) self.last_results detections # 这里可以添加告警逻辑例如检测到未戴安全帽的人 self._check_alerts(detections) else: time.sleep(0.01) # 队列空时短暂休眠 def _check_alerts(self, detections): 简单的告警检查逻辑如果检测到‘person’但没有对应的‘helmet’在附近则触发告警 person_boxes [d for d in detections if d[class_name] person] helmet_boxes [d for d in detections if d[class_name] helmet] # 简单的空间关联判断实际应用需要更复杂的逻辑如IoU或距离阈值 for p in person_boxes: p_center [(p[bbox][0]p[bbox][2])/2, (p[bbox][1]p[bbox][3])/2] has_helmet_nearby False for h in helmet_boxes: h_center [(h[bbox][0]h[bbox][2])/2, (h[bbox][1]h[bbox][3])/2] # 计算欧氏距离简化示例 distance ((p_center[0]-h_center[0])**2 (p_center[1]-h_center[1])**2)**0.5 if distance 50: # 像素距离阈值 has_helmet_nearby True break if not has_helmet_nearby: logger.warning(f⚠️ 告警发现未佩戴安全帽的人员位置: {p[bbox]}) # 此处应调用告警推送函数见下一节 def _reconnect(self): RTSP断线重连 if self.cap: self.cap.release() self.cap cv2.VideoCapture(self.rtsp_url) def get_latest_results(self): 获取最新的检测结果供Flask API调用 return self.last_results.copy() def stop(self): 停止处理器 self.is_running False if self.cap: self.cap.release() if self.process_thread: self.process_thread.join(timeout2) if self.detection_thread: self.detection_thread.join(timeout2) logger.info(RTSP流处理器已停止) # 全局处理器实例 stream_processor None def init_stream_processor(rtsp_url, model_path): 初始化并启动流处理器 global stream_processor stream_processor RTSPStreamProcessor(rtsp_url, model_path) return stream_processor.start()然后在app.py中新增一个API端点用于获取实时流的分析结果# 在app.py中添加 from stream_processor import stream_processor, init_stream_processor # 在合适的地方如app启动前初始化流处理器 # 例如可以放在一个配置文件中或者通过环境变量传入RTSP_URL RTSP_URL rtsp://admin:password192.168.1.100:554/stream1 init_stream_processor(RTSP_URL, MODEL_PATH) app.route(/stream/status) def stream_status(): 获取视频流分析状态和最新结果 if stream_processor and stream_processor.is_running: latest_dets stream_processor.get_latest_results() # 统计未戴安全帽的人数 unsafe_count 0 # ... (这里需要实现更精确的人员-安全帽关联统计逻辑) return jsonify({ status: running, latest_detections: latest_dets, unsafe_person_count: unsafe_count }) else: return jsonify({status: stopped}), 5034.2 方案二客户端推流分析适用于边缘计算另一种架构是将模型部署在靠近摄像头的边缘设备如工控机、Jetson上设备拉取RTSP流进行分析只将结构化的告警结果JSON推送到中心服务器。这种方式减轻了中心服务器的压力也更符合边缘计算的理念。你可以将我们之前写的stream_processor.py稍作修改作为一个独立的边缘服务运行并通过HTTP POST将告警信息发送到中心告警平台。5. 系统集成与告警推送让结果产生价值检测到异常不是终点触发行动才是。我们需要将识别结果集成到现有的工作流中。5.1 告警规则引擎在stream_processor.py的_check_alerts方法中我们实现了一个简单的距离判断。在实际项目中你需要一个更健壮的告警规则引擎。考虑以下因素区域入侵检测只对特定区域如高空作业区、危险品存放区进行安全帽检测。持续时长人员未戴安全帽持续超过N秒才触发告警避免瞬时误报。人员跟踪结合目标跟踪算法如ByteTrack对同一人员的违规行为进行去重避免一秒内上报数十次告警。置信度过滤对低置信度的检测框进行特殊处理或忽略。5.2 集成企业微信/钉钉机器人这是最直接有效的告警通知方式。以企业微信机器人为例首先在app.py同目录下创建notifier.py# notifier.py import requests import json import logging from datetime import datetime logger logging.getLogger(__name__) class WeComNotifier: def __init__(self, webhook_url): self.webhook_url webhook_url def send_alert(self, image_urlNone, location未知区域, person_count1): 发送告警消息到企业微信群 current_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) # 构建Markdown消息 title 安全帽佩戴告警 content f**发现未佩戴安全帽的人员** **时间:** {current_time} **位置:** {location} **涉及人数:** {person_count}人 请现场管理人员立即核实并处理。 # 企业微信机器人支持的消息格式 payload { msgtype: markdown, markdown: { content: content } } # 如果提供了图片URL可以以图文形式发送部分机器人支持 if image_url: payload { msgtype: news, news: { articles: [ { title: title, description: f时间:{current_time} 位置:{location}, url: image_url, picurl: image_url } ] } } try: headers {Content-Type: application/json} response requests.post( self.webhook_url, datajson.dumps(payload), headersheaders, timeout5 ) response.raise_for_status() if response.json().get(errcode) 0: logger.info(f企业微信告警发送成功: {location}) return True else: logger.error(f企业微信返回错误: {response.json()}) return False except Exception as e: logger.error(f发送企业微信告警失败: {e}) return False # 钉钉机器人类似只是消息格式不同 class DingTalkNotifier: def __init__(self, webhook_url, secretNone): self.webhook_url webhook_url self.secret secret def _sign(self, timestamp): 钉钉签名计算如果机器人设置了加签 import hmac import hashlib import base64 if not self.secret: return None string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256 ).digest() return base64.b64encode(hmac_code).decode(utf-8) def send_alert(self, location未知区域, person_count1): import time timestamp str(round(time.time() * 1000)) sign self._sign(timestamp) url self.webhook_url if sign: url ftimestamp{timestamp}sign{sign} content f安全帽佩戴告警\n时间: {datetime.now().strftime(%H:%M:%S)}\n位置: {location}\n未戴安全帽人数: {person_count} payload { msgtype: text, text: { content: content } } try: response requests.post(url, jsonpayload, timeout5) response.raise_for_status() logger.info(f钉钉告警发送成功: {location}) return True except Exception as e: logger.error(f发送钉钉告警失败: {e}) return False # 全局通知器实例 wecom_notifier WeComNotifier(你的企业微信机器人Webhook URL) dingtalk_notifier DingTalkNotifier(你的钉钉机器人Webhook URL, 你的加签密钥) def send_alert(location, person_count, platformall): 发送告警到指定平台 success True if platform in [all, wecom]: success wecom_notifier.send_alert(locationlocation, person_countperson_count) if platform in [all, dingtalk]: success dingtalk_notifier.send_alert(locationlocation, person_countperson_count) return success然后在stream_processor.py的_check_alerts方法中触发告警时调用from notifier import send_alert # 在 _check_alerts 方法的告警条件内 if not has_helmet_nearby: logger.warning(f⚠️ 告警发现未佩戴安全帽的人员位置: {p[bbox]}) # 发送告警假设我们有一个摄像头位置信息 send_alert(location3号塔吊东侧作业区, person_count1, platformwecom)5.3 与现有系统集成API与数据库你的Flask API可以很容易地被其他系统调用。大屏展示前端页面定期轮询/stream/status接口获取最新的检测统计和未戴安全帽人数用图表或列表展示在大屏上。门禁联动门禁系统在人员刷卡时抓拍一张照片并调用/detectAPI。如果检测到未戴安全帽可以发出语音提示或延迟开门。工单系统当告警触发时除了发送消息还可以通过调用工单系统如Jira、自研系统的API自动创建一条巡检工单指派给相应的安全员。数据持久化将每一次检测结果时间、摄像头位置、违规类型、图片快照存入数据库如MySQL、PostgreSQL或时序数据库InfluxDB用于后续的统计分析、报表生成和事故追溯。# 示例简单的SQLite日志记录 import sqlite3 from datetime import datetime def log_violation_to_db(camera_id, violation_type, bbox, confidence, image_path): conn sqlite3.connect(safety_violations.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS violations ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, camera_id TEXT, violation_type TEXT, bbox TEXT, confidence REAL, image_path TEXT ) ) cursor.execute( INSERT INTO violations (camera_id, violation_type, bbox, confidence, image_path) VALUES (?, ?, ?, ?, ?) , (camera_id, violation_type, str(bbox), confidence, image_path)) conn.commit() conn.close()将这套系统部署起来你会发现它开始像一个真正的“AI安全员”一样工作不知疲倦地凝视着监控画面精准地发现违规行为并第一时间通知到人。技术的价值就在于将人从重复、枯燥且要求高度专注的巡检工作中解放出来让他们能去处理更需要人类判断和沟通的复杂问题。