若该文为原创文章转载请注明原文出处。在B站看到大神使用YOLOV26LVM多模态视频检测所以想偿试一下基于 YOLOv8 与 Qwen3.5 的多模态视频行为分析系统.一、背景随着智慧城市、智能安防、工业巡检等领域的快速发展传统视频监控系统仅能实现“看得见”却难以“看得懂”。海量监控视频依赖人工值守存在效率低、漏报率高、响应滞后等问题。近年来计算机视觉与大型语言模型LLM技术取得突破性进展。YOLO 系列目标检测算法以其高实时性与准确率成为边缘端视觉感知的主流选择而多模态大模型如通义千问 Qwen3.5具备强大的图文理解与推理能力能够对检测到的物体、场景进行深度语义分析。将两者融合可以构建“感知‑理解‑决策”一体化的智能视频分析系统实现从“目标检测”到“行为理解”的跨越。本程序正是在此背景下开发利用 YOLOv8 对视频流进行实时目标检测与跟踪提取物体类别、位置、轨迹等结构化信息并调用 Qwen3.5 多模态 API 对关键帧进行行为分析最终在视频画面上叠加检测框、跟踪 ID 及行为分析文本输出带智能标注的高清视频。系统适用于校园安全、交通路口、工地监管、居家看护等场景可有效降低人工监控负担提升异常行为预警能力。二、方案2.1 系统架构本系统采用异步解耦、分层处理的设计思想整体架构分为三层感知层YOLOv8 目标检测与跟踪模块逐帧处理视频输出带跟踪 ID 的物体检测结果。理解层异步任务队列 Qwen3.5 API 调用对关键帧的结构化检测信息进行语义分析生成场景描述、异常判断、风险等级及处理建议。应用层视频渲染与输出模块将检测框、跟踪 ID 以及分析文本叠加到原始视频上支持实时预览与保存为 1080P 高清视频。系统工作流程如下图所示2.2 核心功能模块模块技术实现功能描述目标检测与跟踪YOLOv8 ByteTrack内置检测视频中的人、车、物体等并为每个目标分配唯一 ID输出边界框、类别、置信度、跟踪 ID。多模态行为分析Qwen3.5‑omni‑flash阿里云百炼 API将检测结果与关键帧图像一同发送给大模型分析场景是否存在异常行为返回 JSON 格式的分析结论场景概括、异常标志、风险等级、详细说明、处理建议。中文文本渲染OpenCV PILPillow解决 OpenCV 原生 putText 不支持中文的问题使用 TrueType 字体如黑体在视频左上角绘制清晰的中文分析文本。视频处理与输出OpenCV 高质量缩放支持输入任意分辨率视频输出 1080P1920×1080高清视频使用 INTER_LANCZOS4 插值与 20 Mbps 比特率设置保证画面清晰。异步调用与频率控制threading Queue为避免 API 调用阻塞视频处理采用后台线程队列每隔 FRAME_SKIP 帧默认 30 帧触发一次分析并限制调用间隔2 秒平衡成本与实时性。2.3 关键技术参数检测模型YOLOv8n可替换为 s/m/l/x 以提升精度多模态模型qwen3.5‑omni‑flash成本低、速度快分析频率每 30 帧调用一次 API约 1 秒一次若视频 30fps输出分辨率1920×1080保持原始宽高比字体大小主文本 32px辅助文本 24px编码格式H.264MP4比特率 20 Mbps2.4 部署与使用环境准备Python 3.9安装依赖pip install ultralytics opencv-python openai python-dotenv Pillow获取阿里云百炼 API Key并在.env文件中配置DASHSCOPE_API_KEY运行方式修改INPUT_VIDEO_PATH为待分析视频路径执行python video_analyzer.py程序自动下载 YOLOv8 模型逐帧处理实时显示预览画面按q可提前终止处理完成后在指定路径生成output_analyzed.mp4参数调优FRAME_SKIP增大可降低 API 调用成本减小可提高分析密度CONF_THRESHOLD调整检测置信度阈值过滤低质量框TARGET_WIDTH可改为 1280 等其他宽度FONT_PATH根据操作系统修改中文字体路径2.5 应用示例校园安全检测儿童在危险区域如马路、水池边玩耍及时预警。工地监管识别工人未戴安全帽、闯入禁区等违规行为。交通路口分析行人闯红灯、车辆违规变道等事件。居家养老监测老人摔倒、长时间静止等异常状况。三、环境安装需要安装以下 Python 依赖库。请在终端或命令提示符中运行以下命令进行安装pip install ultralytics opencv-python openai python-dotenv Pillow依赖说明库用途ultralytics加载并运行 YOLOv8 模型支持目标检测与跟踪opencv-python视频读取、图像处理、绘制检测框、保存输出视频openai调用阿里云百炼平台的 Qwen3.5 API兼容 OpenAI 接口python-dotenv从.env文件中读取DASHSCOPE_API_KEY环境变量Pillow在视频帧上绘制中文字符OpenCV 原生不支持中文四、测试代码#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import cv2 import json import base64 import re import threading from queue import Queue from dotenv import load_dotenv from openai import OpenAI from ultralytics import YOLO import numpy as np from PIL import Image, ImageDraw, ImageFont load_dotenv() QWEN_API_KEY os.getenv(DASHSCOPE_API_KEY, your-api-key-here) QWEN_BASE_URL https://dashscope.aliyuncs.com/compatible-mode/v1 MODEL_NAME qwen3.5-omni-flash INPUT_VIDEO_PATH input_video.mp4 OUTPUT_VIDEO_PATH output_analyzed.mp4 YOLO_MODEL_PATH yolov8n.pt FRAME_SKIP 30 CONF_THRESHOLD 0.5 SHOW_PREVIEW True SAVE_RESULT True # 1080P 宽度 1920 TARGET_WIDTH 1920 FONT_SIZE 32 # 主文本字体大小 FONT_SMALL_SIZE 24 # 辅助文本字体大小 # 中文字体路径Windows 黑体其他系统请自行修改 FONT_PATH C:/Windows/Fonts/simhei.ttf client OpenAI(api_keyQWEN_API_KEY, base_urlQWEN_BASE_URL) def load_yolo_model(model_path): try: print(fLoading YOLOv8 model: {model_path}) model YOLO(model_path) return model except Exception as e: print(f模型加载失败: {e}) if os.path.exists(model_path): os.remove(model_path) model YOLO(model_path) return model model load_yolo_model(YOLO_MODEL_PATH) def frame_to_base64(frame): _, buffer cv2.imencode(.jpg, frame, [cv2.IMWRITE_JPEG_QUALITY, 80]) b64_str base64.b64encode(buffer).decode(utf-8) return fdata:image/jpeg;base64,{b64_str} def extract_json_from_text(text): pattern r(?:json)?\s*(\{.*?\})\s* match re.search(pattern, text, re.DOTALL) if match: return match.group(1) start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: return text[start:end1] return text def analyze_behavior_with_qwen(detections, frame_image_pathNone): detections_summary [] for d in detections: detections_summary.append({ track_id: d[track_id], class: d[class], confidence: d[confidence], bbox: d[bbox] }) prompt f 你是一个专业的视频监控行为分析专家。请基于以下YOLOv8模型检测到的物体信息分析当前场景是否存在异常行为。 检测到的物体列表JSON格式{json.dumps(detections_summary, ensure_asciiFalse, indent2)} 请按以下JSON格式输出你的分析结果不要包含其他任何解释或标记 {{ scene_summary: 一句话概括当前场景, abnormal_behavior: true/false, risk_level: 低/中/高, analysis_details: 详细的行为分析说明, suggestion: 如果有异常请给出处理建议 }} print(\n *50) print(f[API调用] 发送分析请求检测到 {len(detections)} 个物体) print(*50) user_content [{type: text, text: prompt}] if frame_image_path: user_content.insert(0, {type: image_url, image_url: {url: frame_image_path}}) try: response client.chat.completions.create( modelMODEL_NAME, messages[{role: user, content: user_content}], temperature0.3, max_tokens500, ) raw_text response.choices[0].message.content print(f[API返回原始内容] {raw_text}) json_str extract_json_from_text(raw_text) try: result_json json.loads(json_str) print(f[解析结果] 异常行为: {result_json.get(abnormal_behavior)}, 风险等级: {result_json.get(risk_level)}) print(f 场景概括: {result_json.get(scene_summary)}) return result_json except json.JSONDecodeError as e: print(f[JSON解析失败] {e}) return {scene_summary: raw_text[:100], abnormal_behavior: False, risk_level: 未知} except Exception as e: print(f[API调用失败] {e}) return None def draw_chinese_text_opencv(img, text, position, font_path, font_size, color_bgr): 使用 PIL 在 OpenCV 图像上绘制中文文本 try: img_pil Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB)) draw ImageDraw.Draw(img_pil) font ImageFont.truetype(font_path, font_size) color_rgb (color_bgr[2], color_bgr[1], color_bgr[0]) draw.text(position, text, fontfont, fillcolor_rgb) img[:] cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR) except Exception as e: print(f[中文绘制错误] {e}) return img def draw_analysis_on_frame(frame, detections, latest_analysis): # 绘制 YOLO 检测框英文用 OpenCV 原生方法 for det in detections: bbox det[bbox] x1, y1, x2, y2 map(int, bbox) label f{det[class]} {det[confidence]:.2f} ID:{det[track_id]} cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2) cv2.putText(frame, label, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2) # 绘制中文分析结果 if latest_analysis and isinstance(latest_analysis, dict): summary latest_analysis.get(scene_summary, 无描述) abnormal latest_analysis.get(abnormal_behavior) risk latest_analysis.get(risk_level, 未知) status_text f异常: {是 if abnormal else 否} | 风险: {risk} suggestion latest_analysis.get(suggestion, ) short_sug suggestion[:60] (... if len(suggestion) 60 else ) if suggestion else # 绘制三行文本调整位置以适应 1080P frame draw_chinese_text_opencv(frame, summary, (20, 50), FONT_PATH, FONT_SIZE, (0, 0, 255)) frame draw_chinese_text_opencv(frame, status_text, (20, 100), FONT_PATH, FONT_SMALL_SIZE, (0, 0, 255)) if short_sug: frame draw_chinese_text_opencv(frame, short_sug, (20, 140), FONT_PATH, FONT_SMALL_SIZE, (0, 0, 255)) return frame def resize_frame_to_width(frame, target_width): 高质量缩放使用 LANCZOS 插值 h, w frame.shape[:2] if w target_width: return frame ratio target_width / w new_h int(h * ratio) # 使用 INTER_LANCZOS4 获得更清晰的缩放结果 return cv2.resize(frame, (target_width, new_h), interpolationcv2.INTER_LANCZOS4) def main(): if not os.path.exists(INPUT_VIDEO_PATH): print(f错误视频文件不存在 - {INPUT_VIDEO_PATH}) return cap cv2.VideoCapture(INPUT_VIDEO_PATH) original_fps int(cap.get(cv2.CAP_PROP_FPS)) original_width int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) original_height int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) print(f原始视频: {original_width}x{original_height}, {original_fps} fps) output_width TARGET_WIDTH # 读取一帧确定缩放后的高度 ret, sample_frame cap.read() if not ret: print(无法读取视频帧) return cap.set(cv2.CAP_PROP_POS_FRAMES, 0) sample_resized resize_frame_to_width(sample_frame, output_width) output_height sample_resized.shape[0] print(f输出视频: {output_width}x{output_height}, {original_fps} fps) out None if SAVE_RESULT: fourcc cv2.VideoWriter_fourcc(*mp4v) out cv2.VideoWriter(OUTPUT_VIDEO_PATH, fourcc, original_fps, (output_width, output_height)) # 尝试设置更高的比特率以减少模糊部分后端支持 if out.isOpened(): try: out.set(cv2.CAP_PROP_BITRATE, 20000000) # 20 Mbps print(已设置输出比特率为 20 Mbps) except: print(当前后端不支持设置比特率使用默认编码质量) analysis_queue Queue() latest_analysis None analysis_lock threading.Lock() def worker(): nonlocal latest_analysis while True: item analysis_queue.get() if item is None: break frame_id, frame_b64, detections item result analyze_behavior_with_qwen(detections, frame_b64) if result: with analysis_lock: latest_analysis result analysis_queue.task_done() worker_thread threading.Thread(targetworker, daemonTrue) worker_thread.start() frame_idx 0 last_analysis_frame -FRAME_SKIP for result in model.track(sourceINPUT_VIDEO_PATH, streamTrue, persistTrue, confCONF_THRESHOLD): frame_idx 1 annotated_frame result.plot() detections [] if result.boxes is not None: boxes result.boxes.xyxy.cpu().numpy() track_ids result.boxes.id.cpu().numpy() if result.boxes.id is not None else [-1]*len(boxes) classes result.boxes.cls.cpu().numpy() confs result.boxes.conf.cpu().numpy() for box, tid, cls, conf in zip(boxes, track_ids, classes, confs): detections.append({ track_id: int(tid), class: model.names[int(cls)], confidence: float(conf), bbox: box.tolist() }) if (frame_idx - last_analysis_frame) FRAME_SKIP: last_analysis_frame frame_idx frame_b64 frame_to_base64(annotated_frame) analysis_queue.put((frame_idx, frame_b64, detections)) with analysis_lock: current_analysis latest_analysis output_frame draw_analysis_on_frame(annotated_frame, detections, current_analysis) output_frame resize_frame_to_width(output_frame, TARGET_WIDTH) if SHOW_PREVIEW: cv2.imshow(YOLOv8 Qwen3.5 Analysis, output_frame) if cv2.waitKey(1) 0xFF ord(q): break if out: out.write(output_frame) if frame_idx % 100 0: print(f已处理 {frame_idx} 帧) analysis_queue.put(None) analysis_queue.join() cap.release() if out: out.release() cv2.destroyAllWindows() print(f处理完成输出视频: {OUTPUT_VIDEO_PATH} (1080P)) if __name__ __main__: main()输出结果注意QWen3.5模型免费用量是1,000,000不要超过哈超过是要费用的。后面想在RK3568上处理。如有侵权或需要完整代码请及时联系博主。