Реализация AI-обработки видео Video Understanding

AI-система понимания и обработки видео

Понимание видео — задача сложнее, чем обработка отдельных кадров. Видео — это пространственно-временной объём данных: объекты движутся, сцены меняются, события разворачиваются во времени. «Что происходит на этом видео?» требует temporal reasoning — понимания последовательности событий.

Архитектура video understanding системы

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

Производительность и оптимизация

Видео генерирует огромный объём данных. 24 часа записи @ 30fps = 2.6М кадров. Обрабатывать каждый кадр нерационально:

• Motion-based sampling: обрабатываем только кадры с движением (background subtraction как фильтр)
• Adaptive sampling: для action recognition достаточно 8fps; для детекции объектов — 15fps
• Hierarchical indexing: сначала scene-level (что происходит в сцене), потом frame-level