Как вы проектируете систему для real-time video understanding (поток с камер)?

Краткий тезис

Проектирование системы real-time video understanding требует баланса между качеством анализа и задержкой (latency). Ключевые компоненты: агрессивный frame sampling (1–5 кадров в секунду), лёгкий vision encoder (например, ViT-S/16 или EfficientNet), temporal modeling (VideoCoCa, TimeSformer или 3D CNN) и LLM с memory для учёта истории кадров. Оптимизация включает квантование, прунинг, аппаратное ускорение (TensorRT, DeepStream) и асинхронный пайплайн. Система должна работать на edge-устройствах (Jetson, Raspberry Pi) или в облаке с гарантией latency < 100 мс.

1. Термин: Real-time video understanding

Real-time video understanding — это задача анализа непрерывного видеопотока (с камеры, файла или сети) с минимальной задержкой, чтобы система могла реагировать на события по мере их возникновения. В отличие от batch-обработки, здесь критичны latency (время от кадра до вывода) и throughput (количество обработанных кадров в секунду).

Основные вызовы

• Огромный объём данных (30 FPS → 1080p → ~150 МБ/с)
• Необходимость учёта временной динамики (движение, действия)
• Ограниченные вычислительные ресурсы на edge
• Требование к детерминированной задержке (не выше порога, например, 50–200 мс)

2. Frame sampling (сэмплирование кадров)

Первый этап — решить, какие кадры отправлять на анализ. Обрабатывать каждый кадр (30 FPS) нереально для LLM-моделей.

Стратегии сэмплирования

Рекомендация Для real-time использовать uniform sampling с 1–5 FPS как базовый, а для сцен с быстрым движением — динамически повышать до 10 FPS через motion trigger.

Пример кода (адаптивный сэмплер):

import cv2
import numpy as np

class AdaptiveFrameSampler:
    def __init__(self, base_fps=5, motion_threshold=30.0):
        self.base_fps = base_fps
        self.motion_threshold = motion_threshold
        self.prev_frame = None
        self.frame_count = 0
        self.target_fps = base_fps

    def should_sample(self, frame):
        if self.prev_frame is None:
            self.prev_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            return True

        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        diff = cv2.absdiff(self.prev_frame, gray)
        motion = np.mean(diff)

        # Адаптивное изменение FPS
        if motion > self.motion_threshold:
            self.target_fps = min(10, self.base_fps * 2)
        else:
            self.target_fps = self.base_fps

        self.prev_frame = gray
        self.frame_count += 1
        interval = int(30 / self.target_fps)  # 30 FPS исходный
        if self.frame_count % interval == 0:
            return True
        return False

3. Vision encoder (кодировщик изображений)

После сэмплирования каждый кадр нужно преобразовать в векторное представление (embedding). Для real-time выбирают лёгкие архитектуры.

Популярные vision encoder'ы

Выбор Для edge-устройств — MobileNetV3 или EfficientNet-B0. Для облака с GPU — ViT-S/16 или ConvNeXt-Tiny.

Важно Encoder должен быть предобучен на крупном датасете (ImageNet-21k, CLIP) и затем дообучен (fine-tuned) на доменных данных (например, видео с камер наблюдения).

Пример использования CLIP vision encoder (лёгкий вариант):

import torch
from transformers import CLIPVisionModel, CLIPProcessor

model = CLIPVisionModel.from_pretrained("openai/clip-vit-base-patch16")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch16")

def encode_frame(frame_pil):
    inputs = processor(images=frame_pil, return_tensors="pt")
    with torch.no_grad():
        outputs = model(**inputs)
    return outputs.pooler_output  # [1, 512]

4. Temporal modeling (моделирование временных связей)

Один кадр не даёт полной картины — нужно понимать динамику: движение, действия, изменение сцены. Для этого применяют temporal modeling.

Основные подходы

Рекомендация для real-time Использовать TSM (Temporal Shift Module) или VideoCoCa (если есть GPU). TSM встраивается в любой 2D CNN и добавляет временной контекст с минимальным overhead.

VideoCoCa (Video Contrastive Captioner) — гибридная модель, которая объединяет пространственные и временные эмбеддинги через cross-attention. Позволяет обрабатывать поток кадров как последовательность токенов.

Пример TSM на PyTorch (упрощённо):

class TemporalShift(nn.Module):
    def __init__(self, net, n_segment=8, fold_div=8):
        super().__init__()
        self.net = net
        self.n_segment = n_segment
        self.fold_div = fold_div

    def forward(self, x):
        # x: [B * T, C, H, W] — батч из T кадров
        b, c, h, w = x.size()
        t = self.n_segment
        x = x.view(-1, t, c, h, w)  # [B, T, C, H, W]
        fold = c // self.fold_div
        out = torch.zeros_like(x)
        out[:, :-1, :fold] = x[:, 1:, :fold]   # сдвиг вперёд
        out[:, 1:, fold:2*fold] = x[:, :-1, fold:2*fold]  # сдвиг назад
        out[:, :, 2*fold:] = x[:, :, 2*fold:]  # остальное без изменений
        out = out.view(b, c, h, w)
        return self.net(out)

5. LLM с memory (память прошлых кадров)

После получения эмбеддингов кадров (с временным контекстом) их нужно передать в LLM для генерации ответа (например, описание сцены, детекция аномалий). Чтобы LLM «помнила» предыдущие кадры, используют memory.

Типы памяти

Рекомендация Для real-time — sliding window (последние 4–8 кадров) + compressed memory через lightweight encoder (например, Perceiver). Это даёт контекст без линейного роста latency.

Пример архитектуры с memory

Кадр t  →  Vision Encoder  →  Temporal Model  →  Embedding t
                                                      ↓
                                              Memory Buffer (FIFO, 8 slots)
                                                      ↓
                                              LLM Prompt: [embed_t-7, ..., embed_t, query]
                                                      ↓
                                              Ответ LLM

Код (псевдо):

class VideoLLM:
    def __init__(self, encoder, temporal_model, llm, memory_size=8):
        self.encoder = encoder
        self.temporal_model = temporal_model
        self.llm = llm
        self.memory = deque(maxlen=memory_size)

    def process_frame(self, frame):
        emb = self.encoder(frame)
        emb = self.temporal_model(emb, self.memory)  # учитывает прошлые
        self.memory.append(emb)
        prompt = self.build_prompt(self.memory, query="What is happening?")
        response = self.llm.generate(prompt)
        return response

6. Оптимизация latency (задержки)

Главная метрика — end-to-end latency от захвата кадра до вывода LLM. Цель: < 100 мс для интерактивных систем, < 200 мс для мониторинга.

Методы оптимизации

Пример асинхронного пайплайна (Python asyncio):

import asyncio
import cv2

async def capture_loop(queue):
    cap = cv2.VideoCapture(0)
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        await queue.put(frame)
        await asyncio.sleep(1/30)  # 30 FPS

async def process_loop(queue, model):
    while True:
        frame = await queue.get()
        result = await model.process_frame(frame)
        # отправить результат на вывод

async def main():
    queue = asyncio.Queue(maxsize=10)
    model = VideoLLM(...)
    await asyncio.gather(capture_loop(queue), process_loop(queue, model))

7. Оценка качества real-time video understanding

Метрики делятся на offline (на датасете) и online (в реальном времени).

Offline метрики:

• Accuracy / F1 для классификации действий (например, UCF101, Kinetics)
• mAP для детекции событий (ActivityNet)
• CIDEr / BLEU для описания видео (VATEX, MSVD)
• Latency (среднее, p99) на целевой платформе

Online метрики

• End-to-end latency (захват → ответ)
• Throughput (кадров/сек)
• Jitter (вариация задержки)
• User satisfaction (опросы, A/B тесты)

Инструменты

• DeepStream SDK (NVIDIA) — готовые метрики pipeline
• MLPerf Inference — бенчмарк для edge
• Prometheus + Grafana — мониторинг в production

8. Пет-проект для закрепления

Задача Разработать real-time систему, которая определяет, когда человек входит в комнату (детекция движения + классификация действия «вход»).

Инструменты

• Python, OpenCV, PyTorch
• Предобученный MobileNetV3 (encoder) + TSM (temporal)
• LLM (например, TinyLlama 1.1B) для генерации текстового оповещения
• FAISS для хранения эталонных эмбеддингов (опционально)
• TensorRT для оптимизации

Шаги:

1. Собрать датасет: 10 видео «человек входит», 10 «пустая комната».
2. Реализовать адаптивный frame sampler (1–5 FPS).
3. Загрузить MobileNetV3 + TSM, дообучить на бинарную классификацию (вход/не вход).
4. Интегрировать TinyLlama: при детекции «вход» генерировать описание («Человек вошёл в 14:23:15»).
5. Оптимизировать через TensorRT: квантование INT8, асинхронный пайплайн.
6. Замерить latency на Jetson Nano или CPU.

Ожидаемый результат

• Работает в реальном времени (≥10 FPS, latency < 150 мс)
• Точность детекции > 90%
• LLM генерирует осмысленные оповещения

9. Связь с другими вопросами

10. Навигация

• Предыдущий: 369
• Следующий: 371
• Индекс: 00. Индекс разборов

Навигация

• Предыдущий: 369
• Следующий: 371
• Индекс: 00. Индекс разборов