在实际的计算机视觉项目中把训练好的 YOLO 模型直接部署成 Web 服务并支持多用户实时交互是很多开发者从实验走向应用的关键一步。单纯用 Flask 做同步请求响应很难满足实时视频流或持续检测任务的需求而只靠 WebSocket 推送数据又容易忽略权限控制、任务管理和数据持久化这些生产环境必须考虑的问题。这篇文章会从零开始搭建一个完整的 FlaskWebSocket 架构实现管理员端和用户端双角色系统。管理员可以上传数据集、启动模型训练、查看验证结果普通用户能提交图片或视频流进行实时目标检测并通过 WebSocket 实时接收检测结果。整个系统会基于 SQLite 管理用户、任务和检测记录并确保训练、验证、检测三大核心功能在同一个工程里可配置、可扩展。如果你已经熟悉 Python 和 Flask 基础但还没系统做过 WebSocket 实时应用或者想把 YOLO 模型集成到 Web 服务中这篇文章会带你走通环境准备、依赖配置、双端权限设计、WebSocket 会话管理、YOLO 模型加载与推理、训练任务异步执行、结果持久化以及常见生产问题的完整链路。1. 理解 FlaskWebSocket 双角色系统的设计动机1.1 为什么不用单纯的 HTTP 轮询做实时检测在目标检测 Web 服务中客户端比如浏览器需要持续获取服务器端的检测结果。如果使用传统的 HTTP 轮询客户端需要每隔几百毫秒就向服务器发送一次请求询问“结果好了吗”。这种模式有两个明显问题资源浪费无论是否有新结果轮询都会产生大量无意义的请求和响应增加服务器压力和网络开销。延迟不可控轮询间隔设置太长结果更新慢设置太短服务器压力大。对于实时视频流检测延迟必须控制在毫秒级。WebSocket 协议在客户端和服务器之间建立一条持久化的全双工通信通道。连接建立后双方可以随时主动发送数据不需要反复建立连接。这对于需要持续传输检测框、类别置信度的目标检测场景非常合适。1.2 管理员端和用户端的功能分离逻辑虽然很多演示项目会把所有功能堆在一个页面但实际系统中模型训练、数据验证和在线检测通常需要不同的权限和界面用户端核心诉求是提交检测任务并实时看到结果。功能集中在图片上传、视频流传输、检测结果渲染和历史记录查看。管理员端需要管理数据集、启动训练任务、监控训练过程、评估模型性能、管理用户权限。这些功能涉及后台异步任务和系统资源调度不能和用户检测请求混在一起。通过角色分离可以更好地控制资源分配比如训练任务占用 GPU检测任务优先响应也便于后续扩展用户管理、计费、审计等功能。1.3 SQLite 在轻量级系统中的适用场景相比 MySQL 或 PostgreSQLSQLite 不需要独立的服务器进程所有数据存在单个磁盘文件中非常适合轻量级 Web 应用、嵌入式设备或原型开发。在这个系统中我们用 SQLite 管理用户表区分管理员和普通用户检测任务表记录用户提交的检测请求和结果训练任务表记录管理员发起的训练配置和状态模型版本表记录不同版本的 YOLO 模型文件和性能指标对于并发不高比如同时在线用户小于100、数据量不大GB级别以内的场景SQLite 完全能满足需求而且部署简单。如果后续需要扩展可以平滑迁移到其他数据库。2. 环境准备与项目结构设计2.1 基础环境与 Python 依赖确保你的 Python 版本在 3.8 以上YOLOv8 对 3.7 支持不完善。建议使用虚拟环境隔离项目依赖# 创建并激活虚拟环境Windows 用 venv 或 conda python -m venv yolo_websocket_env source yolo_websocket_env/bin/activate # Linux/Mac # 或者 .\yolo_websocket_env\Scripts\activate # Windows # 安装核心依赖 pip install flask flask-socketio eventlet pip install ultralytics # 包含 YOLOv8 pip install opencv-python pillow pip install sqlalchemy # ORM 支持这里选择eventlet作为 WebSocket 的异步服务器因为它对 Flask-SocketIO 支持较好能处理大量并发连接。如果你的环境有兼容性问题也可以尝试gevent。2.2 项目目录结构一个清晰的项目结构能避免后续配置混乱。建议按功能模块划分目录yolo_websocket_system/ ├── app.py # Flask 应用入口 ├── config.py # 配置文件数据库路径、模型路径等 ├── requirements.txt # 依赖列表 ├── static/ # 静态资源CSS, JS, 上传的图片等 │ ├── css/ │ ├── js/ │ └── uploads/ # 用户上传的待检测图片/视频 ├── templates/ # HTML 模板 │ ├── base.html # 基础模板 │ ├── user_dashboard.html # 用户端界面 │ └── admin_dashboard.html # 管理员端界面 ├── models/ # 数据模型SQLAlchemy ORM │ └── models.py ├── utils/ # 工具函数 │ ├── yolo_helper.py # YOLO 模型加载和推理封装 │ └── auth.py # 认证装饰器 ├── tasks/ # 后台任务 │ ├── training.py # 训练任务逻辑 │ └── validation.py # 验证任务逻辑 └── instance/ # 实例文件夹SQLite 数据库存放 └── app.db2.3 关键配置文件说明在config.py中定义不同环境的配置import os basedir os.path.abspath(os.path.dirname(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY) or your-secret-key-here SQLALCHEMY_DATABASE_URI os.environ.get(DATABASE_URL) or \ sqlite:/// os.path.join(basedir, instance, app.db) SQLALCHEMY_TRACK_MODIFICATIONS False # 文件上传配置 MAX_CONTENT_LENGTH 16 * 1024 * 1024 # 16MB 限制 UPLOAD_FOLDER os.path.join(basedir, static, uploads) # YOLO 模型配置 YOLO_MODEL_PATH os.path.join(basedir, models, yolov8n.pt) # 默认模型 YOLO_TRAINING_DIR os.path.join(basedir, training_data) # WebSocket 异步模式 ASYNC_MODE eventlet注意生产环境中SECRET_KEY必须通过环境变量设置不要硬编码在代码中。SQLite 数据库路径放在 instance 文件夹是 Flask 的约定便于不同实例隔离数据。3. 数据库模型与用户认证设计3.1 使用 SQLAlchemy 定义数据表在models/models.py中我们用 SQLAlchemy ORM 定义四个核心表from flask_sqlalchemy import SQLAlchemy from datetime import datetime import enum db SQLAlchemy() class UserRole(enum.Enum): ADMIN admin USER user class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) password_hash db.Column(db.String(120), nullableFalse) role db.Column(db.Enum(UserRole), defaultUserRole.USER) created_at db.Column(db.DateTime, defaultdatetime.utcnow) # 关系 detection_tasks db.relationship(DetectionTask, backrefuser, lazyTrue) training_tasks db.relationship(TrainingTask, backrefuser, lazyTrue) class DetectionTask(db.Model): id db.Column(db.Integer, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(user.id), nullableFalse) input_type db.Column(db.String(20)) # image 或 video input_path db.Column(db.String(200)) # 上传文件保存路径 result_path db.Column(db.String(200)) # 检测结果文件路径 status db.Column(db.String(20), defaultpending) # pending, processing, completed, failed created_at db.Column(db.DateTime, defaultdatetime.utcnow) completed_at db.Column(db.DateTime) class TrainingTask(db.Model): id db.Column(db.Integer, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(user.id), nullableFalse) dataset_path db.Column(db.String(200)) model_config db.Column(db.Text) # 训练参数 JSON 字符串 status db.Column(db.String(20), defaultpending) created_at db.Column(db.DateTime, defaultdatetime.utcnow) completed_at db.Column(db.DateTime) metrics db.Column(db.Text) # 验证指标 JSON 字符串 class ModelVersion(db.Model): id db.Column(db.Integer, primary_keyTrue) version_name db.Column(db.String(50), uniqueTrue) model_path db.Column(db.String(200)) is_active db.Column(db.Boolean, defaultFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow)这种设计允许一个用户有多个检测任务和训练任务同时通过ModelVersion表管理不同版本的 YOLO 模型。is_active字段标记当前正在使用的模型版本。3.2 实现基于会话的登录认证在utils/auth.py中我们编写认证装饰器区分管理员和普通用户权限from functools import wraps from flask import session, redirect, url_for, flash from models.models import UserRole def login_required(f): wraps(f) def decorated_function(*args, **kwargs): if user_id not in session: flash(请先登录) return redirect(url_for(login)) return f(*args, **kwargs) return decorated_function def admin_required(f): wraps(f) def decorated_function(*args, **kwargs): if user_id not in session: flash(请先登录) return redirect(url_for(login)) if session.get(user_role) ! UserRole.ADMIN.value: flash(需要管理员权限) return redirect(url_for(user_dashboard)) return f(*args, **kwargs) return decorated_function在登录路由中验证用户凭证后将用户 ID 和角色存入 sessionapp.route(/login, methods[GET, POST]) def login(): if request.method POST: username request.form.get(username) password request.form.get(password) user User.query.filter_by(usernameusername).first() if user and check_password_hash(user.password_hash, password): session[user_id] user.id session[user_role] user.role.value session[username] user.username if user.role UserRole.ADMIN: return redirect(url_for(admin_dashboard)) else: return redirect(url_for(user_dashboard)) else: flash(用户名或密码错误) return render_template(login.html)注意实际项目中密码必须哈希存储这里使用check_password_hash和generate_password_hash来自 Werkzeug处理密码安全。4. WebSocket 集成与实时通信机制4.1 初始化 Flask-SocketIO在app.py中我们需要正确初始化 Flask-SocketIO并配置异步模式from flask import Flask, render_template, session, request from flask_socketio import SocketIO, emit, join_room, leave_room from config import Config app Flask(__name__) app.config.from_object(Config) # 初始化数据库 from models.models import db db.init_app(app) # 初始化 SocketIO设置 cors_allowed_origins 允许跨域如果需要 socketio SocketIO(app, async_modeapp.config[ASYNC_MODE], cors_allowed_origins*) # 创建应用上下文并初始化数据库表 with app.app_context(): db.create_all() app.route(/) def index(): if user_id in session: if session.get(user_role) admin: return redirect(url_for(admin_dashboard)) else: return redirect(url_for(user_dashboard)) return redirect(url_for(login))关键参数async_modeeventlet告诉 SocketIO 使用 eventlet 作为异步后端这对于处理并发 WebSocket 连接至关重要。如果环境中没有安装 eventlet会回退到其他模式但性能可能受影响。4.2 设计 WebSocket 事件处理逻辑WebSocket 通信基于事件驱动。我们需要定义客户端和服务器之间的事件类型# 客户端发送给服务器的事件 socketio.on(connect) def handle_connect(): 客户端连接时自动触发 user_id session.get(user_id) if user_id: join_room(fuser_{user_id}) # 将用户加入其专属房间 emit(connection_status, {status: connected, message: WebSocket 连接成功}) else: emit(connection_status, {status: unauthorized}) return False # 拒绝未认证的连接 socketio.on(disconnect) def handle_disconnect(): 客户端断开连接时触发 print(Client disconnected) socketio.on(start_detection) def handle_start_detection(data): 处理开始检测的请求 task_id data.get(task_id) user_id session.get(user_id) if not user_id: emit(detection_error, {message: 未认证用户}) return # 将任务ID与当前SocketIO会话关联 session[current_task_id] task_id # 通知客户端开始处理 emit(detection_status, {status: processing, task_id: task_id}) # 异步执行检测任务 socketio.start_background_task(targetrun_detection, task_idtask_id, user_iduser_id) def run_detection(task_id, user_id): 在后台线程中运行检测任务 try: # 获取任务信息 task DetectionTask.query.get(task_id) if not task or task.user_id ! user_id: socketio.emit(detection_error, {message: 任务不存在或权限不足}, roomfuser_{user_id}) return # 更新任务状态 task.status processing db.session.commit() # 执行 YOLO 检测 results yolo_helper.detect_image(task.input_path) # 实时推送检测进度和结果 for i, result in enumerate(results): # 构造检测结果数据 detection_data { frame_index: i, boxes: result.boxes.xyxy.tolist() if result.boxes else [], confidences: result.boxes.conf.tolist() if result.boxes else [], class_ids: result.boxes.cls.tolist() if result.boxes else [], class_names: [result.names[int(cls_id)] for cls_id in result.boxes.cls] if result.boxes else [] } # 发送给当前用户专属房间 socketio.emit(detection_progress, detection_data, roomfuser_{user_id}) # 标记任务完成 task.status completed task.completed_at datetime.utcnow() db.session.commit() socketio.emit(detection_status, {status: completed, task_id: task_id}, roomfuser_{user_id}) except Exception as e: # 错误处理 task.status failed db.session.commit() socketio.emit(detection_error, {message: f检测失败: {str(e)}}, roomfuser_{user_id})4.3 前端 WebSocket 客户端实现在用户端的 HTML 模板中我们需要集成 SocketIO 客户端库并处理事件!DOCTYPE html html head titleYOLO 目标检测 - 用户端/title script srchttps://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js/script /head body div iddetection-container input typefile idimage-upload acceptimage/* button onclickstartDetection()开始检测/button div idresult-canvas/div div idstatus-message/div /div script // 连接 WebSocket 服务器 const socket io(); // 监听连接状态 socket.on(connection_status, function(data) { if (data.status connected) { console.log(WebSocket 连接成功); } else { console.error(连接失败:, data.message); } }); // 监听检测进度 socket.on(detection_progress, function(data) { // 实时更新检测结果到画布 updateDetectionResults(data); }); // 监听检测状态 socket.on(detection_status, function(data) { document.getElementById(status-message).innerText 任务 ${data.task_id} 状态: ${data.status}; }); // 监听错误 socket.on(detection_error, function(data) { console.error(检测错误:, data.message); alert(检测失败: data.message); }); function startDetection() { const fileInput document.getElementById(image-upload); if (!fileInput.files.length) { alert(请选择图片); return; } const formData new FormData(); formData.append(image, fileInput.files[0]); // 先通过 HTTP 上传文件并创建检测任务 fetch(/api/detection/tasks, { method: POST, body: formData }) .then(response response.json()) .then(data { if (data.success) { // 通过 WebSocket 开始实时检测 socket.emit(start_detection, {task_id: data.task_id}); } else { alert(任务创建失败: data.message); } }); } function updateDetectionResults(data) { // 使用 Canvas 或 DOM 元素渲染检测框和标签 // 这里简化实现实际需要绘制边界框和类别标签 const canvas document.getElementById(result-canvas); // ... 绘制逻辑 } /script /body /html这种设计实现了文件上传用 HTTP、实时进度用 WebSocket 的混合架构既保证了文件传输的可靠性又获得了实时更新的优势。5. YOLO 模型集成与推理优化5.1 封装 YOLO 模型工具类在utils/yolo_helper.py中我们创建一个专门处理 YOLO 模型的类from ultralytics import YOLO import cv2 import os from datetime import datetime class YOLOHelper: def __init__(self, model_pathNone): self.model_path model_path self.model None self.load_model() def load_model(self, model_pathNone): 加载 YOLO 模型 if model_path: self.model_path model_path if not self.model_path or not os.path.exists(self.model_path): # 如果指定路径不存在尝试下载预训练模型 self.model_path yolov8n.pt # 使用 Ultralytics 提供的默认模型 try: self.model YOLO(self.model_path) print(f模型加载成功: {self.model_path}) except Exception as e: print(f模型加载失败: {e}) raise def detect_image(self, image_path, conf_threshold0.25, iou_threshold0.45): 对单张图片进行目标检测 if not self.model: self.load_model() try: # 使用 YOLOv8 进行推理 results self.model.predict( sourceimage_path, confconf_threshold, iouiou_threshold, saveFalse, # 不自动保存我们自己处理结果 showFalse # 不显示结果 ) return results except Exception as e: print(f检测失败: {e}) return None def process_video_stream(self, video_path, callbackNone): 处理视频流检测通过回调函数实时返回结果 cap cv2.VideoCapture(video_path) if not cap.isOpened(): raise ValueError(无法打开视频文件) frame_count 0 while True: ret, frame cap.read() if not ret: break # 使用 YOLO 检测当前帧 results self.model(frame) # 如果有回调函数传递检测结果 if callback: callback(frame_count, results[0]) frame_count 1 cap.release() def get_model_info(self): 获取模型信息 if not self.model: return None return { model_name: os.path.basename(self.model_path), input_size: self.model.model.args.get(imgsz, 640), classes: self.model.names if hasattr(self.model, names) else {} } # 创建全局实例 yolo_helper YOLOHelper()5.2 处理检测结果的可视化与序列化检测结果需要既能实时推送到前端又能保存到数据库。我们添加结果处理功能import json import base64 from PIL import Image, ImageDraw, ImageFont class DetectionResultProcessor: staticmethod def results_to_json(results): 将 YOLO 检测结果转换为 JSON 格式 if not results or len(results) 0: return [] detections [] for result in results: if result.boxes is not None: for i, box in enumerate(result.boxes): detection { bbox: box.xyxy[0].tolist(), # [x1, y1, x2, y2] confidence: float(box.conf[0]), class_id: int(box.cls[0]), class_name: result.names[int(box.cls[0])] } detections.append(detection) return detections staticmethod def draw_boxes_on_image(image_path, results, output_pathNone): 在图片上绘制检测框 image Image.open(image_path) draw ImageDraw.Draw(image) # 尝试加载字体如果系统支持 try: font ImageFont.truetype(Arial.ttf, 20) except: font ImageFont.load_default() for result in results: if result.boxes is not None: for box in result.boxes: # 提取坐标和类别 x1, y1, x2, y2 box.xyxy[0].tolist() confidence float(box.conf[0]) class_id int(box.cls[0]) class_name result.names[class_id] # 绘制边界框 draw.rectangle([x1, y1, x2, y2], outlinered, width3) # 绘制标签背景 label f{class_name} {confidence:.2f} label_bbox draw.textbbox((x1, y1), label, fontfont) draw.rectangle(label_bbox, fillred) # 绘制标签文字 draw.text((x1, y1), label, fillwhite, fontfont) if output_path: image.save(output_path) return image staticmethod def image_to_base64(image): 将 PIL Image 转换为 base64 字符串用于前端显示 from io import BytesIO buffered BytesIO() image.save(buffered, formatJPEG) img_str base64.b64encode(buffered.getvalue()).decode() return fdata:image/jpeg;base64,{img_str}5.3 模型热切换与版本管理在生产环境中可能需要在不重启服务的情况下切换模型版本class ModelManager: def __init__(self, db_session): self.db db_session self.current_model None self.model_versions {} def load_active_model(self): 从数据库加载当前激活的模型 active_version ModelVersion.query.filter_by(is_activeTrue).first() if not active_version: # 如果没有激活的模型使用默认模型 self.current_model YOLOHelper() return self.current_model if active_version.model_path in self.model_versions: # 如果模型已加载直接使用 self.current_model self.model_versions[active_version.model_path] else: # 加载新模型 self.current_model YOLOHelper(active_version.model_path) self.model_versions[active_version.model_path] self.current_model return self.current_model def switch_model(self, version_name): 切换当前使用的模型版本 new_version ModelVersion.query.filter_by(version_nameversion_name).first() if not new_version: raise ValueError(f模型版本不存在: {version_name}) # 取消当前激活的模型 ModelVersion.query.update({ModelVersion.is_active: False}) # 激活新模型 new_version.is_active True self.db.session.commit() # 重新加载模型 return self.load_active_model()6. 训练与验证功能的后台任务实现6.1 使用 Flask 后台线程处理训练任务模型训练是耗时操作不能在 Web 请求中同步执行。我们使用 Flask-SocketIO 的后台任务功能from threading import Lock # 用于同步训练任务的状态 training_lock Lock() current_training_task None socketio.on(start_training) admin_required # 只有管理员能触发训练 def handle_start_training(data): 处理训练任务请求 global current_training_task if current_training_task and current_training_task.status processing: emit(training_error, {message: 已有训练任务在进行中}) return # 创建训练任务记录 training_task TrainingTask( user_idsession[user_id], dataset_pathdata.get(dataset_path), model_configjson.dumps(data.get(config, {})), statuspending ) db.session.add(training_task) db.session.commit() # 在后台线程中执行训练 socketio.start_background_task( targetrun_training_task, task_idtraining_task.id, configdata.get(config, {}) ) emit(training_status, { status: started, task_id: training_task.id, message: 训练任务已开始 }) def run_training_task(task_id, config): 执行训练任务的核心逻辑 global current_training_task with training_lock: current_training_task TrainingTask.query.get(task_id) if not current_training_task: return try: current_training_task.status processing db.session.commit() # 通知客户端训练开始 socketio.emit(training_progress, { task_id: task_id, epoch: 0, progress: 0, message: 训练准备中... }, roomfuser_{current_training_task.user_id}) # 执行训练 results train_yolo_model( dataset_pathcurrent_training_task.dataset_path, configconfig, progress_callbacklambda epoch, metrics: update_training_progress(task_id, epoch, metrics) ) # 保存训练结果 current_training_task.metrics json.dumps(results.metrics) current_training_task.status completed current_training_task.completed_at datetime.utcnow() db.session.commit() # 创建新的模型版本记录 new_version ModelVersion( version_namefmodel_v{datetime.now().strftime(%Y%m%d_%H%M%S)}, model_pathresults.save_dir ) db.session.add(new_version) db.session.commit() socketio.emit(training_complete, { task_id: task_id, metrics: results.metrics, model_version: new_version.version_name }, roomfuser_{current_training_task.user_id}) except Exception as e: current_training_task.status failed db.session.commit() socketio.emit(training_error, { task_id: task_id, message: f训练失败: {str(e)} }, roomfuser_{current_training_task.user_id}) finally: current_training_task None def update_training_progress(task_id, epoch, metrics): 更新训练进度并推送到客户端 socketio.emit(training_progress, { task_id: task_id, epoch: epoch, metrics: metrics, progress: min(100, epoch * 100 // metrics.get(epochs, 100)) }, roomfuser_{current_training_task.user_id})6.2 YOLO 模型训练封装在tasks/training.py中实现具体的训练逻辑from ultralytics import YOLO import yaml def train_yolo_model(dataset_path, config, progress_callbackNone): 训练 YOLO 模型 # 准备训练配置 training_config { data: f{dataset_path}/data.yaml, epochs: config.get(epochs, 100), imgsz: config.get(image_size, 640), batch: config.get(batch_size, 16), patience: config.get(patience, 50), save: True, exist_ok: True, pretrained: True } # 加载模型从预训练模型开始 model YOLO(yolov8n.pt) # 开始训练带有进度回调 results model.train( **training_config, callbackprogress_callback if progress_callback else None ) return results def validate_model(model_path, dataset_path): 验证模型性能 model YOLO(model_path) # 在验证集上评估 metrics model.val(dataf{dataset_path}/data.yaml) return { precision: metrics.box.map, # mAP recall: metrics.box.map50, # mAP0.5 map: metrics.box.map75, # mAP0.75 speed: metrics.speed # 推理速度 }6.3 数据集格式验证与预处理训练前需要验证数据集格式是否正确import os import yaml def validate_dataset_structure(dataset_path): 验证 YOLO 数据集格式 required_files [data.yaml] required_dirs [images, labels] for file in required_files: if not os.path.exists(os.path.join(dataset_path, file)): raise ValueError(f缺少必要文件: {file}) for dir_name in required_dirs: dir_path os.path.join(dataset_path, dir_name) if not os.path.exists(dir_path): raise ValueError(f缺少必要目录: {dir_name}) # 检查 train/val 子目录 for split in [train, val]: split_dir os.path.join(dir_path, split) if not os.path.exists(split_dir): raise ValueError(f缺少分割目录: {split}) # 检查是否有文件 if split train and len(os.listdir(split_dir)) 0: raise ValueError(训练集不能为空) # 验证 data.yaml 内容 with open(os.path.join(dataset_path, data.yaml), r) as f: data_config yaml.safe_load(f) required_keys [names, nc] for key in required_keys: if key not in data_config: raise ValueError(fdata.yaml 缺少必要键: {key}) return True def create_data_yaml(dataset_path, class_names): 创建标准的 data.yaml 文件 data_config { path: dataset_path, train: images/train, val: images/val, nc: len(class_names), names: class_names } with open(os.path.join(dataset_path, data.yaml), w) as f: yaml.dump(data_config, f) return data_config7. 系统部署与生产环境配置7.1 使用 Gunicorn 部署 Flask 应用对于生产环境不建议直接使用 Flask 开发服务器。可以使用 Gunicorn 配合 eventlet# 安装 Gunicorn pip install gunicorn # 使用 eventlet worker 启动支持 WebSocket gunicorn --bind 0.0.0.0:8000 --worker-class eventlet -w 4 app:app创建gunicorn_config.py配置文件# gunicorn_config.py bind 0.0.0.0:8000 workers 4 worker_class eventlet worker_connections 1000 timeout 120 keepalive 27.2 生产环境数据库优化SQLite 在生产环境中需要注意以下配置class ProductionConfig(Config): # 使用更安全的数据库路径 SQLALCHEMY_DATABASE_URI sqlite:////var/lib/yolo-system/app.db # 生产环境配置 SQLALCHEMY_ENGINE_OPTIONS { pool_recycle: 300, pool_pre_ping: True } # 启用 WAL 模式提高并发性能 staticmethod def init_app(app): import sqlite3 from flask import g app.before_request def enable_wal(): if hasattr(g, db_connection): g.db_connection.execute(PRAGMA journal_modeWAL)7.3 使用 Nginx 反向代理配置 Nginx 处理静态文件和 WebSocket 代理server { listen 80; server_name your-domain.com; # 静态文件处理 location /
Flask+WebSocket双角色YOLO检测系统：从模型训练到实时部署
在实际的计算机视觉项目中把训练好的 YOLO 模型直接部署成 Web 服务并支持多用户实时交互是很多开发者从实验走向应用的关键一步。单纯用 Flask 做同步请求响应很难满足实时视频流或持续检测任务的需求而只靠 WebSocket 推送数据又容易忽略权限控制、任务管理和数据持久化这些生产环境必须考虑的问题。这篇文章会从零开始搭建一个完整的 FlaskWebSocket 架构实现管理员端和用户端双角色系统。管理员可以上传数据集、启动模型训练、查看验证结果普通用户能提交图片或视频流进行实时目标检测并通过 WebSocket 实时接收检测结果。整个系统会基于 SQLite 管理用户、任务和检测记录并确保训练、验证、检测三大核心功能在同一个工程里可配置、可扩展。如果你已经熟悉 Python 和 Flask 基础但还没系统做过 WebSocket 实时应用或者想把 YOLO 模型集成到 Web 服务中这篇文章会带你走通环境准备、依赖配置、双端权限设计、WebSocket 会话管理、YOLO 模型加载与推理、训练任务异步执行、结果持久化以及常见生产问题的完整链路。1. 理解 FlaskWebSocket 双角色系统的设计动机1.1 为什么不用单纯的 HTTP 轮询做实时检测在目标检测 Web 服务中客户端比如浏览器需要持续获取服务器端的检测结果。如果使用传统的 HTTP 轮询客户端需要每隔几百毫秒就向服务器发送一次请求询问“结果好了吗”。这种模式有两个明显问题资源浪费无论是否有新结果轮询都会产生大量无意义的请求和响应增加服务器压力和网络开销。延迟不可控轮询间隔设置太长结果更新慢设置太短服务器压力大。对于实时视频流检测延迟必须控制在毫秒级。WebSocket 协议在客户端和服务器之间建立一条持久化的全双工通信通道。连接建立后双方可以随时主动发送数据不需要反复建立连接。这对于需要持续传输检测框、类别置信度的目标检测场景非常合适。1.2 管理员端和用户端的功能分离逻辑虽然很多演示项目会把所有功能堆在一个页面但实际系统中模型训练、数据验证和在线检测通常需要不同的权限和界面用户端核心诉求是提交检测任务并实时看到结果。功能集中在图片上传、视频流传输、检测结果渲染和历史记录查看。管理员端需要管理数据集、启动训练任务、监控训练过程、评估模型性能、管理用户权限。这些功能涉及后台异步任务和系统资源调度不能和用户检测请求混在一起。通过角色分离可以更好地控制资源分配比如训练任务占用 GPU检测任务优先响应也便于后续扩展用户管理、计费、审计等功能。1.3 SQLite 在轻量级系统中的适用场景相比 MySQL 或 PostgreSQLSQLite 不需要独立的服务器进程所有数据存在单个磁盘文件中非常适合轻量级 Web 应用、嵌入式设备或原型开发。在这个系统中我们用 SQLite 管理用户表区分管理员和普通用户检测任务表记录用户提交的检测请求和结果训练任务表记录管理员发起的训练配置和状态模型版本表记录不同版本的 YOLO 模型文件和性能指标对于并发不高比如同时在线用户小于100、数据量不大GB级别以内的场景SQLite 完全能满足需求而且部署简单。如果后续需要扩展可以平滑迁移到其他数据库。2. 环境准备与项目结构设计2.1 基础环境与 Python 依赖确保你的 Python 版本在 3.8 以上YOLOv8 对 3.7 支持不完善。建议使用虚拟环境隔离项目依赖# 创建并激活虚拟环境Windows 用 venv 或 conda python -m venv yolo_websocket_env source yolo_websocket_env/bin/activate # Linux/Mac # 或者 .\yolo_websocket_env\Scripts\activate # Windows # 安装核心依赖 pip install flask flask-socketio eventlet pip install ultralytics # 包含 YOLOv8 pip install opencv-python pillow pip install sqlalchemy # ORM 支持这里选择eventlet作为 WebSocket 的异步服务器因为它对 Flask-SocketIO 支持较好能处理大量并发连接。如果你的环境有兼容性问题也可以尝试gevent。2.2 项目目录结构一个清晰的项目结构能避免后续配置混乱。建议按功能模块划分目录yolo_websocket_system/ ├── app.py # Flask 应用入口 ├── config.py # 配置文件数据库路径、模型路径等 ├── requirements.txt # 依赖列表 ├── static/ # 静态资源CSS, JS, 上传的图片等 │ ├── css/ │ ├── js/ │ └── uploads/ # 用户上传的待检测图片/视频 ├── templates/ # HTML 模板 │ ├── base.html # 基础模板 │ ├── user_dashboard.html # 用户端界面 │ └── admin_dashboard.html # 管理员端界面 ├── models/ # 数据模型SQLAlchemy ORM │ └── models.py ├── utils/ # 工具函数 │ ├── yolo_helper.py # YOLO 模型加载和推理封装 │ └── auth.py # 认证装饰器 ├── tasks/ # 后台任务 │ ├── training.py # 训练任务逻辑 │ └── validation.py # 验证任务逻辑 └── instance/ # 实例文件夹SQLite 数据库存放 └── app.db2.3 关键配置文件说明在config.py中定义不同环境的配置import os basedir os.path.abspath(os.path.dirname(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY) or your-secret-key-here SQLALCHEMY_DATABASE_URI os.environ.get(DATABASE_URL) or \ sqlite:/// os.path.join(basedir, instance, app.db) SQLALCHEMY_TRACK_MODIFICATIONS False # 文件上传配置 MAX_CONTENT_LENGTH 16 * 1024 * 1024 # 16MB 限制 UPLOAD_FOLDER os.path.join(basedir, static, uploads) # YOLO 模型配置 YOLO_MODEL_PATH os.path.join(basedir, models, yolov8n.pt) # 默认模型 YOLO_TRAINING_DIR os.path.join(basedir, training_data) # WebSocket 异步模式 ASYNC_MODE eventlet注意生产环境中SECRET_KEY必须通过环境变量设置不要硬编码在代码中。SQLite 数据库路径放在 instance 文件夹是 Flask 的约定便于不同实例隔离数据。3. 数据库模型与用户认证设计3.1 使用 SQLAlchemy 定义数据表在models/models.py中我们用 SQLAlchemy ORM 定义四个核心表from flask_sqlalchemy import SQLAlchemy from datetime import datetime import enum db SQLAlchemy() class UserRole(enum.Enum): ADMIN admin USER user class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) password_hash db.Column(db.String(120), nullableFalse) role db.Column(db.Enum(UserRole), defaultUserRole.USER) created_at db.Column(db.DateTime, defaultdatetime.utcnow) # 关系 detection_tasks db.relationship(DetectionTask, backrefuser, lazyTrue) training_tasks db.relationship(TrainingTask, backrefuser, lazyTrue) class DetectionTask(db.Model): id db.Column(db.Integer, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(user.id), nullableFalse) input_type db.Column(db.String(20)) # image 或 video input_path db.Column(db.String(200)) # 上传文件保存路径 result_path db.Column(db.String(200)) # 检测结果文件路径 status db.Column(db.String(20), defaultpending) # pending, processing, completed, failed created_at db.Column(db.DateTime, defaultdatetime.utcnow) completed_at db.Column(db.DateTime) class TrainingTask(db.Model): id db.Column(db.Integer, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(user.id), nullableFalse) dataset_path db.Column(db.String(200)) model_config db.Column(db.Text) # 训练参数 JSON 字符串 status db.Column(db.String(20), defaultpending) created_at db.Column(db.DateTime, defaultdatetime.utcnow) completed_at db.Column(db.DateTime) metrics db.Column(db.Text) # 验证指标 JSON 字符串 class ModelVersion(db.Model): id db.Column(db.Integer, primary_keyTrue) version_name db.Column(db.String(50), uniqueTrue) model_path db.Column(db.String(200)) is_active db.Column(db.Boolean, defaultFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow)这种设计允许一个用户有多个检测任务和训练任务同时通过ModelVersion表管理不同版本的 YOLO 模型。is_active字段标记当前正在使用的模型版本。3.2 实现基于会话的登录认证在utils/auth.py中我们编写认证装饰器区分管理员和普通用户权限from functools import wraps from flask import session, redirect, url_for, flash from models.models import UserRole def login_required(f): wraps(f) def decorated_function(*args, **kwargs): if user_id not in session: flash(请先登录) return redirect(url_for(login)) return f(*args, **kwargs) return decorated_function def admin_required(f): wraps(f) def decorated_function(*args, **kwargs): if user_id not in session: flash(请先登录) return redirect(url_for(login)) if session.get(user_role) ! UserRole.ADMIN.value: flash(需要管理员权限) return redirect(url_for(user_dashboard)) return f(*args, **kwargs) return decorated_function在登录路由中验证用户凭证后将用户 ID 和角色存入 sessionapp.route(/login, methods[GET, POST]) def login(): if request.method POST: username request.form.get(username) password request.form.get(password) user User.query.filter_by(usernameusername).first() if user and check_password_hash(user.password_hash, password): session[user_id] user.id session[user_role] user.role.value session[username] user.username if user.role UserRole.ADMIN: return redirect(url_for(admin_dashboard)) else: return redirect(url_for(user_dashboard)) else: flash(用户名或密码错误) return render_template(login.html)注意实际项目中密码必须哈希存储这里使用check_password_hash和generate_password_hash来自 Werkzeug处理密码安全。4. WebSocket 集成与实时通信机制4.1 初始化 Flask-SocketIO在app.py中我们需要正确初始化 Flask-SocketIO并配置异步模式from flask import Flask, render_template, session, request from flask_socketio import SocketIO, emit, join_room, leave_room from config import Config app Flask(__name__) app.config.from_object(Config) # 初始化数据库 from models.models import db db.init_app(app) # 初始化 SocketIO设置 cors_allowed_origins 允许跨域如果需要 socketio SocketIO(app, async_modeapp.config[ASYNC_MODE], cors_allowed_origins*) # 创建应用上下文并初始化数据库表 with app.app_context(): db.create_all() app.route(/) def index(): if user_id in session: if session.get(user_role) admin: return redirect(url_for(admin_dashboard)) else: return redirect(url_for(user_dashboard)) return redirect(url_for(login))关键参数async_modeeventlet告诉 SocketIO 使用 eventlet 作为异步后端这对于处理并发 WebSocket 连接至关重要。如果环境中没有安装 eventlet会回退到其他模式但性能可能受影响。4.2 设计 WebSocket 事件处理逻辑WebSocket 通信基于事件驱动。我们需要定义客户端和服务器之间的事件类型# 客户端发送给服务器的事件 socketio.on(connect) def handle_connect(): 客户端连接时自动触发 user_id session.get(user_id) if user_id: join_room(fuser_{user_id}) # 将用户加入其专属房间 emit(connection_status, {status: connected, message: WebSocket 连接成功}) else: emit(connection_status, {status: unauthorized}) return False # 拒绝未认证的连接 socketio.on(disconnect) def handle_disconnect(): 客户端断开连接时触发 print(Client disconnected) socketio.on(start_detection) def handle_start_detection(data): 处理开始检测的请求 task_id data.get(task_id) user_id session.get(user_id) if not user_id: emit(detection_error, {message: 未认证用户}) return # 将任务ID与当前SocketIO会话关联 session[current_task_id] task_id # 通知客户端开始处理 emit(detection_status, {status: processing, task_id: task_id}) # 异步执行检测任务 socketio.start_background_task(targetrun_detection, task_idtask_id, user_iduser_id) def run_detection(task_id, user_id): 在后台线程中运行检测任务 try: # 获取任务信息 task DetectionTask.query.get(task_id) if not task or task.user_id ! user_id: socketio.emit(detection_error, {message: 任务不存在或权限不足}, roomfuser_{user_id}) return # 更新任务状态 task.status processing db.session.commit() # 执行 YOLO 检测 results yolo_helper.detect_image(task.input_path) # 实时推送检测进度和结果 for i, result in enumerate(results): # 构造检测结果数据 detection_data { frame_index: i, boxes: result.boxes.xyxy.tolist() if result.boxes else [], confidences: result.boxes.conf.tolist() if result.boxes else [], class_ids: result.boxes.cls.tolist() if result.boxes else [], class_names: [result.names[int(cls_id)] for cls_id in result.boxes.cls] if result.boxes else [] } # 发送给当前用户专属房间 socketio.emit(detection_progress, detection_data, roomfuser_{user_id}) # 标记任务完成 task.status completed task.completed_at datetime.utcnow() db.session.commit() socketio.emit(detection_status, {status: completed, task_id: task_id}, roomfuser_{user_id}) except Exception as e: # 错误处理 task.status failed db.session.commit() socketio.emit(detection_error, {message: f检测失败: {str(e)}}, roomfuser_{user_id})4.3 前端 WebSocket 客户端实现在用户端的 HTML 模板中我们需要集成 SocketIO 客户端库并处理事件!DOCTYPE html html head titleYOLO 目标检测 - 用户端/title script srchttps://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js/script /head body div iddetection-container input typefile idimage-upload acceptimage/* button onclickstartDetection()开始检测/button div idresult-canvas/div div idstatus-message/div /div script // 连接 WebSocket 服务器 const socket io(); // 监听连接状态 socket.on(connection_status, function(data) { if (data.status connected) { console.log(WebSocket 连接成功); } else { console.error(连接失败:, data.message); } }); // 监听检测进度 socket.on(detection_progress, function(data) { // 实时更新检测结果到画布 updateDetectionResults(data); }); // 监听检测状态 socket.on(detection_status, function(data) { document.getElementById(status-message).innerText 任务 ${data.task_id} 状态: ${data.status}; }); // 监听错误 socket.on(detection_error, function(data) { console.error(检测错误:, data.message); alert(检测失败: data.message); }); function startDetection() { const fileInput document.getElementById(image-upload); if (!fileInput.files.length) { alert(请选择图片); return; } const formData new FormData(); formData.append(image, fileInput.files[0]); // 先通过 HTTP 上传文件并创建检测任务 fetch(/api/detection/tasks, { method: POST, body: formData }) .then(response response.json()) .then(data { if (data.success) { // 通过 WebSocket 开始实时检测 socket.emit(start_detection, {task_id: data.task_id}); } else { alert(任务创建失败: data.message); } }); } function updateDetectionResults(data) { // 使用 Canvas 或 DOM 元素渲染检测框和标签 // 这里简化实现实际需要绘制边界框和类别标签 const canvas document.getElementById(result-canvas); // ... 绘制逻辑 } /script /body /html这种设计实现了文件上传用 HTTP、实时进度用 WebSocket 的混合架构既保证了文件传输的可靠性又获得了实时更新的优势。5. YOLO 模型集成与推理优化5.1 封装 YOLO 模型工具类在utils/yolo_helper.py中我们创建一个专门处理 YOLO 模型的类from ultralytics import YOLO import cv2 import os from datetime import datetime class YOLOHelper: def __init__(self, model_pathNone): self.model_path model_path self.model None self.load_model() def load_model(self, model_pathNone): 加载 YOLO 模型 if model_path: self.model_path model_path if not self.model_path or not os.path.exists(self.model_path): # 如果指定路径不存在尝试下载预训练模型 self.model_path yolov8n.pt # 使用 Ultralytics 提供的默认模型 try: self.model YOLO(self.model_path) print(f模型加载成功: {self.model_path}) except Exception as e: print(f模型加载失败: {e}) raise def detect_image(self, image_path, conf_threshold0.25, iou_threshold0.45): 对单张图片进行目标检测 if not self.model: self.load_model() try: # 使用 YOLOv8 进行推理 results self.model.predict( sourceimage_path, confconf_threshold, iouiou_threshold, saveFalse, # 不自动保存我们自己处理结果 showFalse # 不显示结果 ) return results except Exception as e: print(f检测失败: {e}) return None def process_video_stream(self, video_path, callbackNone): 处理视频流检测通过回调函数实时返回结果 cap cv2.VideoCapture(video_path) if not cap.isOpened(): raise ValueError(无法打开视频文件) frame_count 0 while True: ret, frame cap.read() if not ret: break # 使用 YOLO 检测当前帧 results self.model(frame) # 如果有回调函数传递检测结果 if callback: callback(frame_count, results[0]) frame_count 1 cap.release() def get_model_info(self): 获取模型信息 if not self.model: return None return { model_name: os.path.basename(self.model_path), input_size: self.model.model.args.get(imgsz, 640), classes: self.model.names if hasattr(self.model, names) else {} } # 创建全局实例 yolo_helper YOLOHelper()5.2 处理检测结果的可视化与序列化检测结果需要既能实时推送到前端又能保存到数据库。我们添加结果处理功能import json import base64 from PIL import Image, ImageDraw, ImageFont class DetectionResultProcessor: staticmethod def results_to_json(results): 将 YOLO 检测结果转换为 JSON 格式 if not results or len(results) 0: return [] detections [] for result in results: if result.boxes is not None: for i, box in enumerate(result.boxes): detection { bbox: box.xyxy[0].tolist(), # [x1, y1, x2, y2] confidence: float(box.conf[0]), class_id: int(box.cls[0]), class_name: result.names[int(box.cls[0])] } detections.append(detection) return detections staticmethod def draw_boxes_on_image(image_path, results, output_pathNone): 在图片上绘制检测框 image Image.open(image_path) draw ImageDraw.Draw(image) # 尝试加载字体如果系统支持 try: font ImageFont.truetype(Arial.ttf, 20) except: font ImageFont.load_default() for result in results: if result.boxes is not None: for box in result.boxes: # 提取坐标和类别 x1, y1, x2, y2 box.xyxy[0].tolist() confidence float(box.conf[0]) class_id int(box.cls[0]) class_name result.names[class_id] # 绘制边界框 draw.rectangle([x1, y1, x2, y2], outlinered, width3) # 绘制标签背景 label f{class_name} {confidence:.2f} label_bbox draw.textbbox((x1, y1), label, fontfont) draw.rectangle(label_bbox, fillred) # 绘制标签文字 draw.text((x1, y1), label, fillwhite, fontfont) if output_path: image.save(output_path) return image staticmethod def image_to_base64(image): 将 PIL Image 转换为 base64 字符串用于前端显示 from io import BytesIO buffered BytesIO() image.save(buffered, formatJPEG) img_str base64.b64encode(buffered.getvalue()).decode() return fdata:image/jpeg;base64,{img_str}5.3 模型热切换与版本管理在生产环境中可能需要在不重启服务的情况下切换模型版本class ModelManager: def __init__(self, db_session): self.db db_session self.current_model None self.model_versions {} def load_active_model(self): 从数据库加载当前激活的模型 active_version ModelVersion.query.filter_by(is_activeTrue).first() if not active_version: # 如果没有激活的模型使用默认模型 self.current_model YOLOHelper() return self.current_model if active_version.model_path in self.model_versions: # 如果模型已加载直接使用 self.current_model self.model_versions[active_version.model_path] else: # 加载新模型 self.current_model YOLOHelper(active_version.model_path) self.model_versions[active_version.model_path] self.current_model return self.current_model def switch_model(self, version_name): 切换当前使用的模型版本 new_version ModelVersion.query.filter_by(version_nameversion_name).first() if not new_version: raise ValueError(f模型版本不存在: {version_name}) # 取消当前激活的模型 ModelVersion.query.update({ModelVersion.is_active: False}) # 激活新模型 new_version.is_active True self.db.session.commit() # 重新加载模型 return self.load_active_model()6. 训练与验证功能的后台任务实现6.1 使用 Flask 后台线程处理训练任务模型训练是耗时操作不能在 Web 请求中同步执行。我们使用 Flask-SocketIO 的后台任务功能from threading import Lock # 用于同步训练任务的状态 training_lock Lock() current_training_task None socketio.on(start_training) admin_required # 只有管理员能触发训练 def handle_start_training(data): 处理训练任务请求 global current_training_task if current_training_task and current_training_task.status processing: emit(training_error, {message: 已有训练任务在进行中}) return # 创建训练任务记录 training_task TrainingTask( user_idsession[user_id], dataset_pathdata.get(dataset_path), model_configjson.dumps(data.get(config, {})), statuspending ) db.session.add(training_task) db.session.commit() # 在后台线程中执行训练 socketio.start_background_task( targetrun_training_task, task_idtraining_task.id, configdata.get(config, {}) ) emit(training_status, { status: started, task_id: training_task.id, message: 训练任务已开始 }) def run_training_task(task_id, config): 执行训练任务的核心逻辑 global current_training_task with training_lock: current_training_task TrainingTask.query.get(task_id) if not current_training_task: return try: current_training_task.status processing db.session.commit() # 通知客户端训练开始 socketio.emit(training_progress, { task_id: task_id, epoch: 0, progress: 0, message: 训练准备中... }, roomfuser_{current_training_task.user_id}) # 执行训练 results train_yolo_model( dataset_pathcurrent_training_task.dataset_path, configconfig, progress_callbacklambda epoch, metrics: update_training_progress(task_id, epoch, metrics) ) # 保存训练结果 current_training_task.metrics json.dumps(results.metrics) current_training_task.status completed current_training_task.completed_at datetime.utcnow() db.session.commit() # 创建新的模型版本记录 new_version ModelVersion( version_namefmodel_v{datetime.now().strftime(%Y%m%d_%H%M%S)}, model_pathresults.save_dir ) db.session.add(new_version) db.session.commit() socketio.emit(training_complete, { task_id: task_id, metrics: results.metrics, model_version: new_version.version_name }, roomfuser_{current_training_task.user_id}) except Exception as e: current_training_task.status failed db.session.commit() socketio.emit(training_error, { task_id: task_id, message: f训练失败: {str(e)} }, roomfuser_{current_training_task.user_id}) finally: current_training_task None def update_training_progress(task_id, epoch, metrics): 更新训练进度并推送到客户端 socketio.emit(training_progress, { task_id: task_id, epoch: epoch, metrics: metrics, progress: min(100, epoch * 100 // metrics.get(epochs, 100)) }, roomfuser_{current_training_task.user_id})6.2 YOLO 模型训练封装在tasks/training.py中实现具体的训练逻辑from ultralytics import YOLO import yaml def train_yolo_model(dataset_path, config, progress_callbackNone): 训练 YOLO 模型 # 准备训练配置 training_config { data: f{dataset_path}/data.yaml, epochs: config.get(epochs, 100), imgsz: config.get(image_size, 640), batch: config.get(batch_size, 16), patience: config.get(patience, 50), save: True, exist_ok: True, pretrained: True } # 加载模型从预训练模型开始 model YOLO(yolov8n.pt) # 开始训练带有进度回调 results model.train( **training_config, callbackprogress_callback if progress_callback else None ) return results def validate_model(model_path, dataset_path): 验证模型性能 model YOLO(model_path) # 在验证集上评估 metrics model.val(dataf{dataset_path}/data.yaml) return { precision: metrics.box.map, # mAP recall: metrics.box.map50, # mAP0.5 map: metrics.box.map75, # mAP0.75 speed: metrics.speed # 推理速度 }6.3 数据集格式验证与预处理训练前需要验证数据集格式是否正确import os import yaml def validate_dataset_structure(dataset_path): 验证 YOLO 数据集格式 required_files [data.yaml] required_dirs [images, labels] for file in required_files: if not os.path.exists(os.path.join(dataset_path, file)): raise ValueError(f缺少必要文件: {file}) for dir_name in required_dirs: dir_path os.path.join(dataset_path, dir_name) if not os.path.exists(dir_path): raise ValueError(f缺少必要目录: {dir_name}) # 检查 train/val 子目录 for split in [train, val]: split_dir os.path.join(dir_path, split) if not os.path.exists(split_dir): raise ValueError(f缺少分割目录: {split}) # 检查是否有文件 if split train and len(os.listdir(split_dir)) 0: raise ValueError(训练集不能为空) # 验证 data.yaml 内容 with open(os.path.join(dataset_path, data.yaml), r) as f: data_config yaml.safe_load(f) required_keys [names, nc] for key in required_keys: if key not in data_config: raise ValueError(fdata.yaml 缺少必要键: {key}) return True def create_data_yaml(dataset_path, class_names): 创建标准的 data.yaml 文件 data_config { path: dataset_path, train: images/train, val: images/val, nc: len(class_names), names: class_names } with open(os.path.join(dataset_path, data.yaml), w) as f: yaml.dump(data_config, f) return data_config7. 系统部署与生产环境配置7.1 使用 Gunicorn 部署 Flask 应用对于生产环境不建议直接使用 Flask 开发服务器。可以使用 Gunicorn 配合 eventlet# 安装 Gunicorn pip install gunicorn # 使用 eventlet worker 启动支持 WebSocket gunicorn --bind 0.0.0.0:8000 --worker-class eventlet -w 4 app:app创建gunicorn_config.py配置文件# gunicorn_config.py bind 0.0.0.0:8000 workers 4 worker_class eventlet worker_connections 1000 timeout 120 keepalive 27.2 生产环境数据库优化SQLite 在生产环境中需要注意以下配置class ProductionConfig(Config): # 使用更安全的数据库路径 SQLALCHEMY_DATABASE_URI sqlite:////var/lib/yolo-system/app.db # 生产环境配置 SQLALCHEMY_ENGINE_OPTIONS { pool_recycle: 300, pool_pre_ping: True } # 启用 WAL 模式提高并发性能 staticmethod def init_app(app): import sqlite3 from flask import g app.before_request def enable_wal(): if hasattr(g, db_connection): g.db_connection.execute(PRAGMA journal_modeWAL)7.3 使用 Nginx 反向代理配置 Nginx 处理静态文件和 WebSocket 代理server { listen 80; server_name your-domain.com; # 静态文件处理 location /
