Средний

7 мин

16K

Случалось мне работать с CV: запускаешь сорокаминутное видео, YOLO честно находит людей, машины, собак. На двадцатой минуте падает сеть или, что хуже, камера наблюдения выходит из строя. Перезапускаешь. Модель снова смотрит те же кадры, снова инференс, трекинг ID, пошла пахота GPU…

Так продолжаться не может — подключаю кеширование.

Сегодня разбираемся, как совместить YOLO и кэширование Redis с трекингом объектов так, чтобы каждый кадр считался ровно один раз и чтобы информация не терялась. В конце будут готовые сниппеты, которые можно сразу скопировать и запустить.

Почему нельзя «просто детектить»?

Любой, кто запускал model.track() на видео из десяти тысяч кадров, знает, что модель будет мусолить каждый пиксель, даже если в кадре пустая стена. А если стена неподвижна десять секунд, триста одинаковых кадров получат триста одинаковых вызовов нейросети. Умножаем на цену GPU и стараемся не заплакать.

Вторая проблема: трекинг объектов (ByteTrack, BoT-SORT, DeepSORT) держит своё состояние в оперативной памяти. Выключили свет/сломали камеру или, в конце концов, краш VS-code — потерялись ID всех, кто был в кадре. При перезапуске объект получает новый id, и смысл видеоаналитики обесценивается.

Третья проблема: повторное распознавание. Одна и та же машина паркуется каждое утро, но модель каждый раз с нуля извлекает её признаки, читает номер, определяет цвет. Ненужные телодвижения, не правда ли?

Решение описанных проблем достаточно тривиально: нужен кэш. Простые решения (а-ля пикл-файл и другие локальные кэши) не удовлетворяли моим требованиям архитектуры (прежде всего масштабируемость, модульность), поэтому я использовал Redis, который можно развернуть на отдельном порту.

Кэшируем детекции: Redis

Рассмотрим на простом примере: Redis запущен в докере

Ключ — MD5-хэш изображения. Если кадр поменялся, хэш будет другой.

import hashlib
import pickle
import cv2
from ultralytics import YOLO
import redis

r = redis.Redis(host='127.0.0.1', port=6379, db=0, socket_connect_timeout=2)
model = YOLO("yolo11n.pt")

def frame_hash(frame: cv2.Mat) -> str:
    return hashlib.md5(frame.tobytes()).hexdigest()

def get_detections(frame):
    key = f"yolo_det:{frame_hash(frame)}"
    cached = r.get(key)
    if cached:
        return pickle.loads(cached)

    results = model(frame, verbose=False)[0]
    detections = []
    if results.boxes is not None:
        for box in results.boxes:
            x1, y1, x2, y2 = box.xyxy[0].tolist()
            detections.append({
                "box": [x1, y1, x2, y2],
                "confidence": float(box.conf[0]),
                "class": int(box.cls[0])
            })
    r.setex(key, 1800, pickle.dumps(detections))
    return detections

Что здесь происходит:

• frame_hash вычисляет MD5-хэш картинки.

• При отсутствии кэша model(frame).

• Результат сериализуется через pickle и сохраняется в Redis.

В сущности имеем следующее: первый вызов — порядка 30–50 мс на хорошем GPU. Повторный вызов того же кадра из Redis — около 0.5–1 мс. Приятно? Приятно.

Устойчивый трекинг

Кэширование кадров — отлично. Но что, если вы отслеживаете объекты и вам важно, чтобы ID не скакал после перезапуска системы? (База видеоаналитики)

YOLO изначально даёт трекинг: model.track(frame, persist=True, tracker="bytetrack.yaml"). Проблема в том, что внутреннее состояние трекера живёт в оперативной памяти процесса. При любом падении вы потеряете данные трекинга.

Здесь Redis снова выручает: сохраняем не только детекции, но и сами треки — последовательность bounding boxes с ID для каждого кадра.

import json
import cv2
from ultralytics import YOLO
import redis

r = redis.Redis(host='127.0.0.1', port=6379, db=0)
model = YOLO("yolo11n.pt")

VIDEO_PATH = "parking.mp4"
cache_key = f"track:{VIDEO_PATH}"

cached_tracks = r.get(cache_key)
if cached_tracks:
    tracks = json.loads(cached_tracks)
    print("Треки загружены из Redis.")
else:
    print("Кэш пуст, запускаем полный трекинг.")
    cap = cv2.VideoCapture(VIDEO_PATH)
    tracks = []
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        results = model.track(frame, persist=True, tracker="bytetrack.yaml",
                              verbose=False)
        frame_data = {"frame_idx": len(tracks), "objects": []}
        boxes = results[0].boxes
        if boxes is not None and boxes.id is not None:
            for box, track_id in zip(boxes.xyxy.cpu().numpy(),
                                     boxes.id.cpu().numpy()):
                x1, y1, x2, y2 = map(float, box)
                frame_data["objects"].append({
                    "box": [x1, y1, x2, y2],
                    "id": int(track_id)
                })
        tracks.append(frame_data)
    cap.release()
    r.set(cache_key, json.dumps(tracks))
    print(f"Готово, {len(tracks)} кадров записаны в Redis.")

Теперь систему можно ронять в любой момент) При перезапуске все треки восстановятся из Redis’а и объекты сохранят свои ID.

Эмбеддинги

А что, если кэшировать не просто детекции, а признаки объекта? Векторное представление (embedding) — это сжатая «личность» машины, человека или товара. Извлекли один раз — и потом просто сравниваем новый кадр с сохранённым вектором. Без повторного распознавания.

Какие модели используют для эмбеддингов в CV?

Задача «превратить картинку в вектор» называется извлечением признаков (feature extraction). Под капотом используется свёрточная нейросеть (CNN) или Vision Transformer. На выходе получается вектор, как правильно, имеющий размерность 512, 768 или 2048, который описывает содержимое изображения. Дальше такой вектор можно сравнивать с другими через стандартное косинусное расстояние.

Пример подобных моделей:

• CLIP от OpenAI. Понимает связь между текстом и картинками, но даже если использовать только визуальную часть — качество эмбеддингов очень высокое.

• Модели:
  openai/clip-vit-base-patch16,
  openai/clip-vit-large-patch14.

• Плюсы: отлично работает изначально, не нужно дообучать.

• Минусы: имеет зависимость в виде либы transformers, весит прилично.

• TIMM (EfficientNet, ConvNeXt и подобные). Библиотека timm — это швейцарский нож для CV. Внутри — предобученные модели, выбирай не хочу. Для эмбеддингов чаще всего берут efficientnet_b0 (1280 признаков) или convnext_tiny (768 признаков).

• Плюсы: легковесные, быстрые, отличный выбор для поиска похожих изображений.

• Минусы: ввиду разнообразия содержимого нет конкретики по решению. Придётся выбирать и сравнивать.

• Torchvision (ResNet50, MobileNetV3) Классика, которая идёт в комплекте с PyTorch. ResNet50 (2048 признаков) — проверенное временем решение. MobileNetV3-Large (960 признаков) — шустрая темка для слабого GPU.

• Плюсы: всегда под рукой, не надо ставить лишние зависимости.

• Минусы: для задач распознавания, требующих внимания к большому числу признаков и деталей (например, отличить двух похожих людей) может уступать CLIP.

Рассмотрим несколько гипотетических кейсов:

Парковка или гараж с распознаванием номеров, КПП/въезд на объект. Машина въезжает впервые. YOLO находит авто и вырезает номерную пластину. OCR-модель считывает госномер «А123БТ777». Этот номер — уникальный идентификатор. В Redis по ключу plate:A123BT777 сохраняется эмбеддинг внешнего вида автомобиля (цвет, форма, особые приметы) — вектор на 512 или 1024 числа. Плюс дата первого появления, марка, модель.

При следующем въезде YOLO снова детектит машину, OCR читает номер. Система находит совпадение в Redis: «Это же тот самый рено логан чёрного цвета XX века, я его знаю». Подтягивается его эмбеддинг, сравнивается с текущим изображением для идентификации, и открывается шлагбаум. OCR для номера вызывается, но модель/алгоритм распознавания самой машины со всеми её признаками — нет.

Конвейерная проверка качества. Идут сотни одинаковых плат. Первая — эталонная — детектится YOLO, её разметка (bbox’ы компонентов) сохраняются в Redis как pcb_template:model_X. Для всех последующих плат модель включается только при отклонении от этого эталона по гистограмме или площади.

Фотоловушки в заповеднике. Камера снимает медведя, YOLO вырезает его морду, модель распознавания извлекает эмбеддинг — уникальные особенности шкуры, форму ушей, шрамы. Этот эмбеддинг кэшируется в Redis. Когда через три дня тот же медведь приходит к водопою, система не создаёт новую запись в журнале учёта, а говорит: «Повторная встреча, особь №67».

Общий код такого кэширования:

import numpy as np
import redis
from redis.commands.search.field import VectorField, TagField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from PIL import Image
import torch
from transformers import CLIPModel, CLIPProcessor
import cv2


class CLIPEmbedder:
    def __init__(self, model_name="openai/clip-vit-base-patch16"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model = CLIPModel.from_pretrained(model_name).to(self.device)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.model.eval()

    def extract(self, image: Image.Image) -> np.ndarray:
        inputs = self.processor(images=image, return_tensors="pt").to(self.device)
        with torch.no_grad():
            features = self.model.get_image_features(**inputs)
        features = features / features.norm(dim=-1, keepdim=True)
        return features[0].cpu().numpy()


r = redis.Redis(host='127.0.0.1', port=6379, db=0)
embedder = CLIPEmbedder()

# Параметры индекса
INDEX_NAME = "idx:objects"
VECTOR_DIM = 512  
DISTANCE_METRIC = "COSINE"  
PREFIX = "object:"         


def create_index():
    """Создаёт векторный индекс, если он ещё не существует."""
    try:
        r.ft(INDEX_NAME).info()
        print("Индекс уже существует.")
    except:
        print("Создаём индекс...")
        schema = (
            VectorField(
                "embedding",
                "COSINE",  
                {
                    "TYPE": "FLOAT32",
                    "DIM": VECTOR_DIM,
                    "DISTANCE_METRIC": DISTANCE_METRIC,
                },
            ),
            TagField("id"),  
        )
        definition = IndexDefinition(prefix=[PREFIX], index_type=IndexType.HASH)
        r.ft(INDEX_NAME).create_index(schema, definition=definition)
        print("Индекс создан.")


create_index()


def get_or_identify(crop: np.ndarray) -> str:
    # Преобразуем BGR (OpenCV) → RGB 
    crop_rgb = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB)
    pil_image = Image.fromarray(crop_rgb)

    emb = embedder.extract(pil_image)
    emb_bytes = emb.astype(np.float32).tobytes()
    query = (
        Query("*=>[KNN 1 @embedding $vec]")
        .return_field("__embedding_score")  # расстояние (1 - cosine)
        .return_field("id")
        .dialect(2) 
    )
    results = r.ft(INDEX_NAME).search(query, query_params={"vec": emb_bytes})

    if results.docs:
        distance = float(results.docs[0].__embedding_score)
        similarity = 1.0 - distance
        if similarity > 0.8:
            object_id = results.docs[0].id.split(":")[1]
            return object_id

  
    new_id = str(r.incr("next_object_id"))
    object_key = f"object:{new_id}"
    r.hset(
        object_key,
        mapping={
            "id": new_id,
            "embedding": emb_bytes,
        },
    )
    return new_id

Заключение

Итого после проделанных махинаций, имеем:

• Каждый кадр считается один раз. Даже если скрипт упал, Redis помнит всё, что уже было посчитано.

• Трекинг больше не боится перезагрузок. Потому что состояние хранится в базе данных, а не в оперативной памяти.

• Повторные объекты идентифицируются мгновенно. Машины, люди, платы, медведы — им не нужно повторное распознавание.

Спасибо за прочтение! Буду рад обратной связи!

Какие методы для кэширования объектов в системах видеоаналитики используете вы?

© 2026 ООО «МТ ФИНАНС»