YOLO11 + WebRTC：低延迟视频检测系统部署

1. 引言：当实时检测遇上低延迟传输

想象一下，你需要监控一个仓库的实时动态，或者分析一场体育比赛的现场画面。传统的视频流处理方案往往存在明显的延迟，等你看到检测结果时，画面里的人可能已经走远了。这种“慢半拍”的体验，在需要即时响应的场景里，比如安防预警、工业质检或者互动直播，是完全无法接受的。

今天，我们就来解决这个痛点。我们将一个强大的目标检测模型——YOLO11，与一个专为实时通信而生的技术——WebRTC，结合起来。YOLO11负责“看得准”，它能快速识别出画面中的物体；WebRTC负责“传得快”，它能以极低的延迟将视频流和检测结果同步推送到你的浏览器上。

这篇文章，我将带你从零开始，一步步搭建一个完整的低延迟视频检测系统。你不需要是深度学习或网络通信的专家，我会用最直白的方式，告诉你每一步该怎么做，并提供可以直接运行的代码。最终，你将拥有一个在浏览器里就能实时看到检测框、几乎没有延迟的演示系统。

2. 环境准备：一键获取YOLO11完整开发环境

工欲善其事，必先利其器。为了省去繁琐的环境配置，我们直接使用一个预置好的YOLO11深度学习镜像。这个镜像里已经包含了运行YOLO11所需的所有依赖，比如PyTorch、Ultralytics库、OpenCV等，开箱即用。

2.1 启动与访问你的YOLO11环境

获取环境后，你有两种主要的方式来使用它：通过网页版的Jupyter Notebook，或者通过更灵活的SSH终端。

方式一：通过Jupyter Notebook（推荐新手） 这是一种图形化的操作方式，特别适合边写代码边看结果。

1. 在镜像管理页面，找到并启动我们提供的“YOLO11完整可运行环境”镜像。
2. 启动成功后，你会看到一个名为“Jupyter”的访问入口，点击它。
3. 系统会打开一个新的浏览器标签页，这就是Jupyter Lab的界面。在这里，你可以创建新的Python笔记本（.ipynb文件），编写和运行代码，就像在本地使用一样方便。

方式二：通过SSH终端（适合进阶用户） 如果你更喜欢在命令行里操作，或者需要运行一些后台脚本，SSH是你的好帮手。

1. 同样在镜像详情页，找到“SSH”访问方式。
2. 你会获得一个命令，格式类似 ssh root@你的实例IP -p 端口号。
3. 打开你电脑上的终端（Windows用户可以用PowerShell或CMD），粘贴并运行这个命令。
4. 首次连接可能需要输入密码（通常会在页面提示），输入后你就进入了远程服务器的命令行环境，可以自由执行各种Linux命令了。

无论选择哪种方式，你现在都已经拥有了一个功能齐全的YOLO11开发环境。接下来，我们进入正题。

3. 快速上手：验证YOLO11基础检测能力

在搭建复杂系统之前，我们先确保YOLO11本身能正常工作。让我们跑一个最简单的训练示例，感受一下它的速度。

首先，我们需要进入项目目录。在Jupyter中新建一个代码单元格，或者在SSH终端里，执行以下命令：

cd ultralytics-8.3.9/

这个目录包含了YOLO11（Ultralytics框架）的所有源代码和示例。接下来，我们运行一个基础的训练脚本。虽然我们最终目标是做推理（检测），但训练过程能最全面地测试环境是否配置正确。

在终端或Notebook的代码单元格中，输入：

python train.py

这个命令会使用一个内置的小型数据集（比如COCO128）启动一个简短的训练流程。你会看到终端开始滚动输出信息，包括加载模型、计算损失、更新权重等。运行完成后，通常会生成一个 runs/train/ 目录，里面保存了训练好的模型权重文件（.pt文件）和训练过程的各种图表。

看到类似上面的输出，并且最终生成了模型文件，就说明你的YOLO11环境完全正常，可以开始进行更酷的实时检测了。

4. 核心构建：YOLO11实时检测服务端

现在，我们来构建系统的“大脑”——一个能够持续接收视频流、进行实时检测并返回结果的服务端程序。我们将使用Python的Flask框架来搭建一个简单的Web服务器，并使用OpenCV来处理视频帧。

4.1 创建Flask应用与视频处理循环

首先，在你的工作目录下创建一个新的Python文件，比如叫 realtime_server.py。我们将分步编写其中的代码。

第一步：导入必要的库

from flask import Flask, Response, jsonify, request
import cv2
from ultralytics import YOLO
import numpy as np
import threading
import time
import logging

# 设置日志，方便查看运行状态
logging.basicConfig(level=logging.INFO)
app = Flask(__name__)

我们导入了Flask用于创建Web服务，cv2 (OpenCV)用于处理图像，YOLO来自Ultralytics库用于加载检测模型，其他是一些辅助库。

第二步：加载YOLO11模型 在Flask应用启动时，我们加载预训练的YOLO11模型。这里我们使用中等尺寸的模型 yolo11m.pt，它在精度和速度之间取得了很好的平衡。你也可以根据需求换成更小的 yolo11n.pt（更快）或更大的 yolo11l.pt（更准）。

# 加载YOLO11模型（确保模型文件路径正确，或会自动下载）
model = YOLO('yolo11m.pt')
logging.info("YOLO11模型加载成功！")

第三步：定义全局变量和检测函数 我们需要一个全局变量来存储最新的检测结果和帧，以便不同的HTTP请求都能获取到。

latest_frame = None
latest_results = None
frame_lock = threading.Lock()

def run_detection():
    """持续从摄像头捕获帧并进行检测的函数"""
    global latest_frame, latest_results
    # 打开摄像头，0通常代表默认摄像头。如果是视频文件，传入文件路径。
    cap = cv2.VideoCapture(0)
    
    if not cap.isOpened():
        logging.error("无法打开摄像头！")
        return
    
    logging.info("开始摄像头捕获与实时检测...")
    while True:
        success, frame = cap.read()
        if not success:
            break
        
        # 使用YOLO11进行检测
        results = model(frame, verbose=False)  # verbose=False关闭冗余输出
        
        # 更新全局变量
        with frame_lock:
            latest_frame = frame.copy()
            latest_results = results
        
        # 控制一下检测频率，避免CPU占用过高，这里大约30FPS
        time.sleep(0.033)
    
    cap.release()

这个函数会在一个独立的线程中运行，不断从摄像头抓取画面，并用YOLO11进行检测，然后把最新的帧和结果存起来。

第四步：启动检测线程和Flask路由 我们需要在Flask应用启动时，就运行上面的检测循环。

# 在应用启动前启动检测线程
detection_thread = threading.Thread(target=run_detection, daemon=True)
detection_thread.start()

@app.route('/')
def index():
    """提供一个简单的HTML页面作为客户端"""
    return '''
    <html>
      <head><title>YOLO11实时检测</title></head>
      <body>
        <h2>YOLO11实时视频检测</h2>
        <img id="videoFeed" src="/video_feed" width="640" height="480">
        <br>
        <div id="results"></div>
        <script>
            // 可选：通过JavaScript定期获取检测结果文本
            setInterval(() => {
                fetch('/get_results')
                    .then(r => r.json())
                    .then(data => {
                        document.getElementById('results').innerHTML = JSON.stringify(data, null, 2);
                    });
            }, 500);
        </script>
      </body>
    </html>
    '''

@app.route('/video_feed')
def video_feed():
    """生成携带检测框的视频流（MJPEG格式）"""
    def generate():
        while True:
            with frame_lock:
                if latest_frame is None:
                    continue
                frame_with_boxes = latest_frame.copy()
                if latest_results:
                    # 将检测结果绘制到帧上
                    annotated_frame = latest_results[0].plot()
                    frame_with_boxes = annotated_frame
            
            # 将帧编码为JPEG图片
            ret, jpeg = cv2.imencode('.jpg', frame_with_boxes)
            if not ret:
                continue
            
            # 按照MJPEG流格式输出
            yield (b'--frame\r\n'
                   b'Content-Type: image/jpeg\r\n\r\n' + jpeg.tobytes() + b'\r\n')
            time.sleep(0.033) # 控制流帧率
    
    return Response(generate(),
                    mimetype='multipart/x-mixed-replace; boundary=frame')

@app.route('/get_results')
def get_results():
    """获取最新的检测结果（JSON格式）"""
    with frame_lock:
        if latest_results:
            # 提取检测到的物体信息：类别、置信度、坐标
            result_list = []
            for box in latest_results[0].boxes:
                cls_id = int(box.cls[0])
                conf = float(box.conf[0])
                xyxy = box.xyxy[0].tolist()
                result_list.append({
                    'class': model.names[cls_id],
                    'confidence': round(conf, 2),
                    'bbox': [round(x, 1) for x in xyxy]
                })
            return jsonify({'detections': result_list})
        return jsonify({'detections': []})

if __name__ == '__main__':
    # 运行Flask服务器，host='0.0.0.0'允许外部访问
    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)

4.2 运行与测试服务端

保存好 realtime_server.py 文件后，在终端中运行它：

python realtime_server.py

如果一切顺利，你会看到日志输出“模型加载成功”和“开始摄像头捕获”。现在，打开你的浏览器，访问 http://你的服务器IP地址:5000（如果就在本机运行，可以访问 http://127.0.0.1:5000）。

你应该能看到一个网页，中间显示着你摄像头的实时画面，并且YOLO11检测出的物体（比如人、杯子、键盘）都被画上了框。页面下方还会以JSON格式显示检测到的物体列表和置信度。

至此，一个基础的实时检测服务端就完成了。但它使用的是HTTP轮询或MJPEG流，延迟通常在几百毫秒，还不够“实时”。接下来，我们引入WebRTC来大幅降低延迟。

5. 实现超低延迟：集成WebRTC技术

WebRTC（Web Real-Time Communication）是一种支持网页浏览器进行实时音视频通信的技术。它的核心优势是点对点（P2P）传输和极低的延迟，非常适合我们的实时检测场景。

我们将使用一个名为 aiortc 的Python库，它让我们能在服务端轻松构建WebRTC应用。同时，我们需要一个简单的信令服务器，来帮助客户端和服务端建立连接。

5.1 搭建信令服务器与WebRTC服务端

首先，安装必要的库。在终端中执行：

pip install aiortc opencv-python-headless aiohttp

接下来，我们创建一个新的文件 webrtc_server.py。这个文件会比之前的复杂一些，因为它要处理WebRTC的连接、协商和媒体流传输。

import asyncio
import json
import logging
from aiohttp import web
from aiortc import RTCPeerConnection, RTCSessionDescription, VideoStreamTrack
from aiortc.contrib.media import MediaBlackhole, MediaPlayer, MediaRecorder
from av import VideoFrame
import cv2
from ultralytics import YOLO
import numpy as np

# 加载YOLO模型
model = YOLO('yolo11m.pt')
logging.basicConfig(level=logging.INFO)

# 存储活跃的WebRTC连接
pcs = set()

class YOLOVideoStreamTrack(VideoStreamTrack):
    """
    一个自定义的视频流轨道，每一帧都经过YOLO11处理。
    """
    kind = "video"

    def __init__(self):
        super().__init__()
        self.cap = cv2.VideoCapture(0)  # 捕获摄像头
        if not self.cap.isOpened():
            raise RuntimeError("无法打开摄像头")

    async def recv(self):
        """
        获取下一帧视频，并用YOLO处理。
        这是WebRTC媒体流的核心方法。
        """
        pts, time_base = await self.next_timestamp()

        # 从摄像头读取一帧
        ret, frame = self.cap.read()
        if not ret:
            # 如果读取失败，返回一个黑帧
            frame = np.zeros((480, 640, 3), dtype=np.uint8)
        else:
            # 使用YOLO11进行检测
            results = model(frame, verbose=False)
            # 将检测结果绘制到帧上
            if results and results[0].boxes is not None:
                frame = results[0].plot()

        # 将OpenCV的BGR格式转换为RGB，并创建AV VideoFrame
        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        video_frame = VideoFrame.from_ndarray(frame_rgb, format='rgb24')
        video_frame.pts = pts
        video_frame.time_base = time_base
        return video_frame

    def release(self):
        """释放摄像头资源"""
        if self.cap:
            self.cap.release()

async def index(request):
    """提供一个测试用的HTML页面"""
    with open('webrtc_client.html', 'r') as f:
        content = f.read()
    return web.Response(text=content, content_type='text/html')

async def offer(request):
    """处理客户端发来的SDP Offer，并返回Answer"""
    params = await request.json()
    offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])

    # 创建一个新的RTCPeerConnection
    pc = RTCPeerConnection()
    pcs.add(pc)

    # 准备本地媒体（我们的YOLO视频轨道）
    video_track = YOLOVideoStreamTrack()

    @pc.on("connectionstatechange")
    async def on_connectionstatechange():
        logging.info(f"连接状态变为: {pc.connectionState}")
        if pc.connectionState == "failed" or pc.connectionState == "closed":
            # 连接关闭或失败时，清理资源
            video_track.release()
            await pc.close()
            pcs.discard(pc)

    # 将视频轨道添加到连接中
    pc.addTrack(video_track)

    # 设置远程描述（客户端的Offer）
    await pc.setRemoteDescription(offer)

    # 创建本地Answer
    answer = await pc.createAnswer()
    await pc.setLocalDescription(answer)

    # 将Answer返回给客户端
    return web.Response(
        content_type="application/json",
        text=json.dumps({
            "sdp": pc.localDescription.sdp,
            "type": pc.localDescription.type
        }),
    )

async def on_shutdown(app):
    """关闭时，清理所有WebRTC连接"""
    coros = [pc.close() for pc in pcs]
    await asyncio.gather(*coros)
    pcs.clear()

if __name__ == "__main__":
    # 创建Aiohttp应用
    app = web.Application()
    app.router.add_get('/', index)
    app.router.add_post('/offer', offer)
    app.on_shutdown.append(on_shutdown)
    
    # 运行服务器
    web.run_app(app, host='0.0.0.0', port=8080)

这个服务器做了几件关键的事：

1. 创建了一个自定义的 YOLOVideoStreamTrack 类，它继承自WebRTC的视频轨道。在这个类的 recv() 方法里，我们不断从摄像头抓帧，并用YOLO11处理，然后将处理后的帧返回。
2. 提供了一个 /offer 的接口。当浏览器（客户端）想要建立连接时，会发送一个SDP Offer到这里。服务器收到后，会创建一个对等的连接（RTCPeerConnection），将我们的视频轨道附加上去，然后生成一个SDP Answer返回给浏览器。
3. 这样，一个点对点的、极低延迟的视频流通道就建立起来了。视频数据不再通过服务器中转，而是直接在浏览器和我们的Python检测程序之间传输。

5.2 创建WebRTC客户端页面

我们需要一个简单的HTML页面作为客户端。在与 webrtc_server.py 相同的目录下，创建 webrtc_client.html 文件。

<!DOCTYPE html>
<html>
<head>
    <title>YOLO11 + WebRTC 实时检测</title>
    <style>
        body { font-family: Arial; margin: 20px; }
        #video { border: 2px solid #ccc; background: #000; }
        #status { margin-top: 10px; color: #666; }
    </style>
</head>
<body>
    <h2>YOLO11 + WebRTC 低延迟检测演示</h2>
    <p>点击按钮开始。视频流将直接从服务器摄像头获取，并经过YOLO11实时处理。</p>
    
    <button id="startButton">开始检测</button>
    <button id="stopButton" disabled>停止</button>
    
    <br><br>
    <video id="video" autoplay playsinline width="640" height="480"></video>
    
    <div id="status">状态：等待连接...</div>

    <script>
        const videoElement = document.getElementById('video');
        const startButton = document.getElementById('startButton');
        const stopButton = document.getElementById('stopButton');
        const statusDiv = document.getElementById('status');
        
        let pc = null; // WebRTC连接对象
        
        // 信令服务器地址（假设与HTML页面同源，否则需处理CORS）
        const signalingServer = 'http://' + window.location.hostname + ':8080';
        
        startButton.onclick = async () => {
            startButton.disabled = true;
            stopButton.disabled = false;
            statusDiv.textContent = '状态：正在建立连接...';
            
            try {
                // 1. 创建RTCPeerConnection
                const configuration = { iceServers: [{ urls: 'stun:stun.l.google.com:19302' }] };
                pc = new RTCPeerConnection(configuration);
                
                // 2. 监听远程视频流，并显示在video标签上
                pc.ontrack = (event) => {
                    console.log('收到远程视频轨道');
                    videoElement.srcObject = event.streams[0];
                };
                
                pc.oniceconnectionstatechange = () => {
                    statusDiv.textContent = `ICE连接状态: ${pc.iceConnectionState}`;
                };
                
                // 3. 创建Offer
                const offer = await pc.createOffer();
                await pc.setLocalDescription(offer);
                
                // 4. 将Offer发送给我们的信令服务器
                const response = await fetch(signalingServer + '/offer', {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({
                        sdp: pc.localDescription.sdp,
                        type: pc.localDescription.type
                    })
                });
                
                // 5. 接收服务器的Answer，并设置为远程描述
                const answer = await response.json();
                await pc.setRemoteDescription(new RTCSessionDescription(answer));
                
                statusDiv.textContent = '状态：连接已建立，开始传输视频！';
                
            } catch (error) {
                console.error('连接失败:', error);
                statusDiv.textContent = '状态：连接失败 - ' + error.message;
                startButton.disabled = false;
                stopButton.disabled = true;
            }
        };
        
        stopButton.onclick = () => {
            if (pc) {
                pc.close();
                pc = null;
            }
            videoElement.srcObject = null;
            statusDiv.textContent = '状态：连接已关闭。';
            startButton.disabled = false;
            stopButton.disabled = true;
        };
    </script>
</body>
</html>

这个HTML页面包含了一个视频播放窗口和两个按钮。点击“开始检测”，浏览器会尝试与我们的Python WebRTC服务器建立连接。一旦成功，经过YOLO11处理的实时视频流就会直接显示在网页上，延迟可以低至几十毫秒。

6. 部署与运行你的低延迟检测系统

所有代码都已就绪，让我们把它们跑起来，看看最终效果。

第一步：启动WebRTC信令与检测服务器 在你的终端中，运行：

python webrtc_server.py

服务器将在 http://0.0.0.0:8080 启动。你会看到日志输出，表明服务器正在运行。

第二步：在浏览器中访问客户端 确保你的开发环境（或服务器）的8080端口可以被访问。打开浏览器，访问 http://你的服务器IP:8080。

第三步：建立连接并观看实时检测

1. 在打开的网页中，点击“开始检测”按钮。
2. 浏览器会弹出权限请求，询问是否允许使用摄像头（实际上我们用的是服务器的摄像头）。这里需要特别注意： 我们的服务器代码 (YOLOVideoStreamTrack) 访问的是服务器本地的摄像头（cv2.VideoCapture(0)）。如果你在远程服务器上运行，且服务器没有物理摄像头，这里会失败。
3. 解决方案：你可以修改服务器代码，将 cv2.VideoCapture(0) 替换为一个视频文件路径（如 ‘test.mp4’）来测试，或者使用RTSP等网络视频流地址（如 ‘rtsp://username:password@ip:port/stream’）。
4. 如果一切顺利（比如使用了视频文件），点击“开始”后，稍等片刻，视频画面就会出现在网页中，并且物体已经被YOLO11实时标记出来。

体验低延迟：试着在摄像头前快速移动一个物体（比如你的手或一个杯子）。你会发现，网页上检测框的跟随几乎感觉不到延迟，这与之前HTTP流的方式有质的提升。这就是WebRTC的威力。

7. 总结与展望

通过这篇文章，我们完成了一个从理论到实践的完整项目：将顶尖的目标检测算法YOLO11与实时通信技术WebRTC相结合，构建了一个超低延迟的视频检测系统。

我们来回顾一下核心步骤：

1. 环境准备：利用预置镜像，快速获得一个包含YOLO11所有依赖的、开箱即用的开发环境。
2. 模型验证：运行简单的训练脚本，确保YOLO11基础功能正常。
3. 构建服务端：使用Flask和OpenCV搭建了一个能进行实时检测并输出MJPEG视频流的后台服务。
4. 集成WebRTC：引入 aiortc 库，创建了一个自定义的视频轨道，将YOLO11检测过程嵌入到WebRTC媒体流中，实现了点对点的极低延迟传输。
5. 创建客户端：编写了一个HTML页面，通过简单的JavaScript代码与我们的WebRTC服务器握手，并接收、播放处理后的视频流。

这个系统的价值在于：

• 低延迟：WebRTC技术将延迟从数百毫秒降低到百毫秒以内，满足了真正“实时”应用的需求。
• 易部署：客户端只需一个现代浏览器，无需安装任何插件或软件。
• 高灵活：服务端可以部署在云端、边缘设备或本地服务器，适应不同场景。

你可以在此基础上继续探索：

• 更换视频源：尝试接入网络摄像头（RTSP）、视频文件或屏幕捕捉。
• 优化性能：调整YOLO11模型大小（如使用 yolo11n.pt）、图像分辨率或检测频率，在速度和精度间找到最佳平衡。
• 丰富功能：在视频流上叠加更多信息（如计数、轨迹跟踪），或将检测结果通过WebSocket发送到客户端进行更复杂的交互。
• 应用到实际场景：将其用于智能门禁、生产线瑕疵检测、直播间互动特效等具体项目。

希望这个项目能成为你探索实时AI应用的一个起点。动手尝试，修改代码，让它解决你遇到的实际问题吧。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。