在实际的计算机视觉项目中，把训练好的 YOLO 模型直接部署成 Web 服务，并支持多用户实时交互，是很多开发者从实验走向应用的关键一步。单纯用 Flask 做同步请求响应，很难满足实时视频流或持续检测任务的需求；而只靠 WebSocket 推送数据，又容易忽略权限控制、任务管理和数据持久化这些生产环境必须考虑的问题。
这篇文章会从零开始，搭建一个完整的 Flask+WebSocket 架构，实现管理员端和用户端双角色系统。管理员可以上传数据集、启动模型训练、查看验证结果；普通用户能提交图片或视频流进行实时目标检测，并通过 WebSocket 实时接收检测结果。整个系统会基于 SQLite 管理用户、任务和检测记录，并确保训练、验证、检测三大核心功能在同一个工程里可配置、可扩展。
如果你已经熟悉 Python 和 Flask 基础，但还没系统做过 WebSocket 实时应用，或者想把 YOLO 模型集成到 Web 服务中，这篇文章会带你走通环境准备、依赖配置、双端权限设计、WebSocket 会话管理、YOLO 模型加载与推理、训练任务异步执行、结果持久化以及常见生产问题的完整链路。
1. 理解 Flask+WebSocket 双角色系统的设计动机
1.1 为什么不用单纯的 HTTP 轮询做实时检测
在目标检测 Web 服务中，客户端（比如浏览器）需要持续获取服务器端的检测结果。如果使用传统的 HTTP 轮询，客户端需要每隔几百毫秒就向服务器发送一次请求，询问“结果好了吗？”。这种模式有两个明显问题：
- 资源浪费：无论是否有新结果，轮询都会产生大量无意义的请求和响应，增加服务器压力和网络开销。
- 延迟不可控：轮询间隔设置太长，结果更新慢；设置太短，服务器压力大。对于实时视频流检测，延迟必须控制在毫秒级。
WebSocket 协议在客户端和服务器之间建立一条持久化的全双工通信通道。连接建立后，双方可以随时主动发送数据，不需要反复建立连接。这对于需要持续传输检测框、类别置信度的目标检测场景非常合适。
1.2 管理员端和用户端的功能分离逻辑
虽然很多演示项目会把所有功能堆在一个页面，但实际系统中，模型训练、数据验证和在线检测通常需要不同的权限和界面：
- 用户端：核心诉求是提交检测任务并实时看到结果。功能集中在图片上传、视频流传输、检测结果渲染和历史记录查看。
- 管理员端：需要管理数据集、启动训练任务、监控训练过程、评估模型性能、管理用户权限。这些功能涉及后台异步任务和系统资源调度，不能和用户检测请求混在一起。
通过角色分离，可以更好地控制资源分配（比如训练任务占用 GPU，检测任务优先响应），也便于后续扩展用户管理、计费、审计等功能。
1.3 SQLite 在轻量级系统中的适用场景
相比 MySQL 或 PostgreSQL，SQLite 不需要独立的服务器进程，所有数据存在单个磁盘文件中，非常适合轻量级 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_key=True) username = db.Column(db.String(80), unique=True, nullable=False) password_hash = db.Column(db.String(120), nullable=False) role = db.Column(db.Enum(UserRole), default=UserRole.USER) created_at = db.Column(db.DateTime, default=datetime.utcnow) # 关系 detection_tasks = db.relationship('DetectionTask', backref='user', lazy=True) training_tasks = db.relationship('TrainingTask', backref='user', lazy=True) class DetectionTask(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) 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), default='pending') # pending, processing, completed, failed created_at = db.Column(db.DateTime, default=datetime.utcnow) completed_at = db.Column(db.DateTime) class TrainingTask(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) dataset_path = db.Column(db.String(200)) model_config = db.Column(db.Text) # 训练参数 JSON 字符串 status = db.Column(db.String(20), default='pending') created_at = db.Column(db.DateTime, default=datetime.utcnow) completed_at = db.Column(db.DateTime) metrics = db.Column(db.Text) # 验证指标 JSON 字符串 class ModelVersion(db.Model): id = db.Column(db.Integer, primary_key=True) version_name = db.Column(db.String(50), unique=True) model_path = db.Column(db.String(200)) is_active = db.Column(db.Boolean, default=False) created_at = db.Column(db.DateTime, default=datetime.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 和角色存入 session：
@app.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(username=username).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_mode=app.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_mode='eventlet'告诉 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(f'user_{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(target=run_detection, task_id=task_id, user_id=user_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': '任务不存在或权限不足'}, room=f'user_{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, room=f'user_{user_id}') # 标记任务完成 task.status = 'completed' task.completed_at = datetime.utcnow() db.session.commit() socketio.emit('detection_status', {'status': 'completed', 'task_id': task_id}, room=f'user_{user_id}') except Exception as e: # 错误处理 task.status = 'failed' db.session.commit() socketio.emit('detection_error', {'message': f'检测失败: {str(e)}'}, room=f'user_{user_id}')4.3 前端 WebSocket 客户端实现
在用户端的 HTML 模板中，我们需要集成 SocketIO 客户端库并处理事件：
<!DOCTYPE html> <html> <head> <title>YOLO 目标检测 - 用户端</title> <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script> </head> <body> <div id="detection-container"> <input type="file" id="image-upload" accept="image/*"> <button onclick="startDetection()">开始检测</button> <div id="result-canvas"></div> <div id="status-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_path=None): self.model_path = model_path self.model = None self.load_model() def load_model(self, model_path=None): """加载 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_threshold=0.25, iou_threshold=0.45): """对单张图片进行目标检测""" if not self.model: self.load_model() try: # 使用 YOLOv8 进行推理 results = self.model.predict( source=image_path, conf=conf_threshold, iou=iou_threshold, save=False, # 不自动保存，我们自己处理结果 show=False # 不显示结果 ) return results except Exception as e: print(f"检测失败: {e}") return None def process_video_stream(self, video_path, callback=None): """处理视频流检测，通过回调函数实时返回结果""" 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_path=None): """在图片上绘制检测框""" 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], outline="red", width=3) # 绘制标签背景 label = f"{class_name} {confidence:.2f}" label_bbox = draw.textbbox((x1, y1), label, font=font) draw.rectangle(label_bbox, fill="red") # 绘制标签文字 draw.text((x1, y1), label, fill="white", font=font) 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, format="JPEG") img_str = base64.b64encode(buffered.getvalue()).decode() return f"data: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_active=True).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_name=version_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_id=session['user_id'], dataset_path=data.get('dataset_path'), model_config=json.dumps(data.get('config', {})), status='pending' ) db.session.add(training_task) db.session.commit() # 在后台线程中执行训练 socketio.start_background_task( target=run_training_task, task_id=training_task.id, config=data.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': '训练准备中...' }, room=f'user_{current_training_task.user_id}') # 执行训练 results = train_yolo_model( dataset_path=current_training_task.dataset_path, config=config, progress_callback=lambda 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_name=f"model_v{datetime.now().strftime('%Y%m%d_%H%M%S')}", model_path=results.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 }, room=f'user_{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)}' }, room=f'user_{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)) }, room=f'user_{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_callback=None): """训练 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, callback=progress_callback if progress_callback else None ) return results def validate_model(model_path, dataset_path): """验证模型性能""" model = YOLO(model_path) # 在验证集上评估 metrics = model.val(data=f'{dataset_path}/data.yaml') return { 'precision': metrics.box.map, # mAP 'recall': metrics.box.map50, # mAP@0.5 'map': metrics.box.map75, # mAP@0.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(f"data.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_mode=WAL')7.3 使用 Nginx 反向代理
配置 Nginx 处理静态文件和 WebSocket 代理：
server { listen 80; server_name your-domain.com; # 静态文件处理 location /
