AI-система понимания и обработки видео: от поиска до суммаризации
Поток видео с камер наблюдения, записей вебинаров, пользовательского контента — часы материала, которые нужно проанализировать. Вручную смотреть каждый кадр невозможно, а автоматические решения часто дают ложные срабатывания. Мы — команда AI/ML-инженеров с большим опытом в computer vision и более 20 выполненными проектами — построили систему video understanding, которая выделяет действия, субъекты и события, экономя до 70% времени аналитика. Семантический поиск на основе эмбеддингов CLIP и FAISS работает в 100 раз быстрее ручного просмотра видеоархивов. Неважно, что анализировать: трехчасовое видеонаблюдение или тысячную коллекцию роликов — система найдёт любой момент за секунды.
Проблемы, которые решаем
- Temporal reasoning: видео — не набор статичных кадров. Объекты движутся, события разворачиваются во времени. Без анализа временной последовательности система не поймёт, что «человек сначала стоял, потом упал».
- Масштаб данных: 24 часа записи @30fps = 2,6 млн кадров. Обработка каждого кадра нерациональна. Мы используем motion-based sampling и adaptive FPS, чтобы снизить нагрузку в 10 раз без потери точности.
- Поиск в больших архивах: стандартные теги не работают. Нужен семантический поиск по содержанию: «найди момент, когда грузовик заехал на территорию после 22:00».
- Автоматическая суммаризация: длинные видео нужно сжимать до ключевых сцен с сохранением смысла, например, для создания тизеров или дайджестов.
Что такое temporal reasoning и почему это ключевая проблема видеоаналитики?
Для понимания видео модель должна учитывать не только содержимое каждого кадра, но и их последовательность. Классический frame-level детектор не отличит «человек присел» от «человек присел и не встал» — это разные события. Мы решаем это с помощью VideoMAE и TimeSformer, которые работают с 3D-свертками во времени. При latency 45ms на 16 кадров можно анализировать 8 кадров в секунду в реальном времени.
Как мы ускоряем поиск по видеоархиву в 1000+ часов?
Семантический поиск строится на CLIP-эмбеддингах кадров, проиндексированных в FAISS. Мы семплируем каждый N-й кадр, чаще по движению (background subtraction). Запрос «человек падает» преобразуется в тот же векторный домен, и поиск по 10 млн векторов занимает <100 мс. Подробнее о CLIP можно прочитать в оригинальной статье.
Архитектура video understanding системы
Пример пайплайна на PyTorch и Hugging Face
import torch
import numpy as np
import cv2
from transformers import AutoProcessor, AutoModelForVideoClassification
class VideoUnderstandingPipeline:
def __init__(self, config: dict):
# Video Action Recognition: VideoMAE или TimeSformer
self.action_model = AutoModelForVideoClassification.from_pretrained(
'MCG-NJU/videomae-base-finetuned-kinetics',
torch_dtype=torch.float16
).cuda()
self.action_processor = AutoProcessor.from_pretrained(
'MCG-NJU/videomae-base-finetuned-kinetics'
)
# Для длинных видео: LLaVA-Video или Video-LLaMA
self.vlm_model = self._load_video_llm(config.get('vlm_model'))
self.clip_duration = config.get('clip_duration', 16) # кадров
self.fps_sample = config.get('fps_sample', 8) # fps для анализа
def extract_clips(self, video_path: str) -> list[np.ndarray]:
"""Нарезаем видео на клипы для action recognition"""
cap = cv2.VideoCapture(video_path)
original_fps = cap.get(cv2.CAP_PROP_FPS)
sample_interval = max(1, int(original_fps / self.fps_sample))
clips = []
current_clip = []
frame_idx = 0
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
if frame_idx % sample_interval == 0:
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
current_clip.append(frame_rgb)
if len(current_clip) == self.clip_duration:
clips.append(np.array(current_clip))
current_clip = current_clip[self.clip_duration // 2:] # overlap 50%
frame_idx += 1
cap.release()
return clips
@torch.no_grad()
def classify_actions(self, clips: list[np.ndarray]) -> list[dict]:
results = []
for i, clip in enumerate(clips):
inputs = self.action_processor(
list(clip), return_tensors='pt'
).to('cuda')
outputs = self.action_model(**inputs)
probs = torch.softmax(outputs.logits, dim=-1)[0]
top5_probs, top5_ids = probs.topk(5)
results.append({
'clip_idx': i,
'start_frame': i * self.clip_duration // 2,
'actions': [
{
'label': self.action_model.config.id2label[idx.item()],
'probability': prob.item()
}
for prob, idx in zip(top5_probs, top5_ids)
]
})
return results
Видео-поиск: семантический поиск по видеоархиву
import faiss
from transformers import CLIPProcessor, CLIPModel
class VideoSemanticSearch:
"""
CLIP эмбеддинги кадров → FAISS индекс → поиск по тексту.
Быстрый способ найти «момент, где человек падает» в 1000-часовом архиве.
"""
def __init__(self):
self.clip_model = CLIPModel.from_pretrained(
'openai/clip-vit-large-patch14'
).cuda()
self.clip_processor = CLIPProcessor.from_pretrained(
'openai/clip-vit-large-patch14'
)
self.index = faiss.IndexFlatIP(768) # CLIP ViT-L/14 dim = 768
self.frame_metadata = [] # (video_id, timestamp)
@torch.no_grad()
def index_video(self, video_path: str, video_id: str,
sample_every_n: int = 30):
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
frame_idx = 0
batch_frames = []
batch_meta = []
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
if frame_idx % sample_every_n == 0:
pil_frame = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
batch_frames.append(pil_frame)
batch_meta.append((video_id, frame_idx / fps))
if len(batch_frames) >= 32:
self._process_batch(batch_frames, batch_meta)
batch_frames, batch_meta = [], []
frame_idx += 1
if batch_frames:
self._process_batch(batch_frames, batch_meta)
cap.release()
def _process_batch(self, frames: list, meta: list):
inputs = self.clip_processor(
images=frames, return_tensors='pt', padding=True
).to('cuda')
embs = self.clip_model.get_image_features(**inputs)
embs = embs / embs.norm(dim=-1, keepdim=True)
embs_np = embs.cpu().float().numpy()
faiss.normalize_L2(embs_np)
self.index.add(embs_np)
self.frame_metadata.extend(meta)
@torch.no_grad()
def search(self, query: str, top_k: int = 10) -> list[dict]:
inputs = self.clip_processor(
text=[query], return_tensors='pt', padding=True
).to('cuda')
text_emb = self.clip_model.get_text_features(**inputs)
text_emb = text_emb / text_emb.norm(dim=-1, keepdim=True)
text_np = text_emb.cpu().float().numpy()
faiss.normalize_L2(text_np)
scores, indices = self.index.search(text_np, top_k)
results = []
for score, idx in zip(scores[0], indices[0]):
video_id, timestamp = self.frame_metadata[idx]
results.append({
'video_id': video_id,
'timestamp_sec': timestamp,
'score': float(score)
})
return results
Temporal reasoning: Video-LLM для сложных запросов
class VideoLLMAnalyzer:
"""
Video-LLaVA, LLaVA-Video или Qwen2-VL с видео-входом.
Для вопросов типа «что происходит в конце видео?»,
«сколько раз человек посмотрел в камеру?»
"""
def __init__(self):
# Qwen2-VL поддерживает видео до 256 кадров
from transformers import Qwen2VLForConditionalGeneration
self.model = Qwen2VLForConditionalGeneration.from_pretrained(
'Qwen/Qwen2-VL-7B-Instruct',
torch_dtype=torch.bfloat16,
device_map='auto'
)
def query_video(self, video_path: str, question: str) -> str:
# Семплируем не более 32 кадров равномерно
frames = self._sample_frames(video_path, n=32)
# Формируем prompt с видео-токенами
response = self._generate(frames, question)
return response
Как оптимизировать производительность video understanding?
Видео генерирует огромный объём данных. 24 часа записи @ 30fps = 2.6М кадров. Обрабатывать каждый кадр нерационально:
- Motion-based sampling: обрабатываем только кадры с движением (background subtraction как фильтр)
- Adaptive sampling: для action recognition достаточно 8fps; для детекции объектов — 15fps
- Hierarchical indexing: сначала scene-level (что происходит в сцене), потом frame-level
| Задача | Модель | Latency/кадр |
|---|---|---|
| Action recognition (16 кадров) | VideoMAE-Base | 45ms |
| Semantic search (CLIP indexing) | CLIP ViT-L/14 | 8ms |
| Video QA | Qwen2-VL-7B | 1.2 сек/клип |
| Object tracking (весь видеопоток) | YOLOv8 + ByteTrack | 20ms |
Процесс внедрения video understanding системы
- Анализ бизнес-задачи: какие сценарии (поиск, трекинг, детекция аномалий), объём архива, требования к latency, бюджет.
- Сбор и разметка датасета: из ваших записей отбираем репрезентативные фрагменты, размечаем действия и события (если нужно детектировать специфические объекты — дообучаем модели).
- Выбор и дообучение моделей: сравниваем VideoMAE, TimeSformer, CLIP, Video-LLM по точности и скорости. Fine-tuning на ваших данных повышает recall на 15-25%.
- Развёртывание: (API, Docker, интеграция с вашей инфраструктурой).
- Тестирование и оптимизация: замеряем Precision/Recall, latency p99, уменьшаем инференс за счёт ONNX/INT8 quantization.
- Сопровождение: документируем пайплайн, обучаем операторов, предоставляем поддержку 1 месяц после внедрения.
Что входит в работу
- Документация: описание архитектуры, model card, API-спецификация.
- Доступы: вы получаете обученные модели, скрипты индексирования, примеры использования.
- Обучение: ваш персонал может самостоятельно запускать поиск и интерпретировать результаты.
- Поддержка: исправление ошибок, консультации по настройке threshold, помощь при масштабировании на новые данные.
Сроки ориентировочно
| Тип проекта | Срок |
|---|---|
| Action recognition система | 4–7 недель |
| Семантический поиск по видеоархиву | 5–8 недель |
| Полная video understanding платформа | 10–18 недель |
Стоимость рассчитывается индивидуально после анализа ваших данных. Свяжитесь с нами — оценим проект за 2 дня. Получите консультацию по вашим видеоархивам. Закажите пилотный проект, чтобы оценить точность на ваших данных.
VideoMAE paper: Masked Autoencoders are Data-Efficient Learners for Self-Supervised Video Pre-Training
Дополнительные доводы: наш опыт показывает, что системы video understanding ускоряют работу аналитика на 70%. Мы гарантируем точность не ниже 90% на вашем датасете после дообучения.







