Мультимодальный RAG с Gemini Embedding 2: туториал по работе с видео и изображениями

Текст умер. Да здравствуют картинки и видео

Вы построили RAG для документов. Теперь клиент присылает папку с видео-инструкциями, коллажами из Figma и скриншотами ошибок. И просит "сделать поиск как в Google, но по нашему контенту".

Традиционные эмбеддинг-модели смотрят на видео как на набор субтитров. Gemini Embedding 2 (релиз конца 2025) видит движение, объекты, контекст. Она помещает кадр из видео, скриншот интерфейса и текстовое описание в одно векторное пространство. И находит связи, которые человек не заметит.

Зачем это нужно? (Кроме того, чтобы впечатлить босса)

Представьте:

• Дизайнер ищет "темная тема, кнопка слева, анимация появления". Система находит макет в Figma, соответствующий фрагмент видео-презентации и описание в гайдлайне.
• Поддержка ищет "ошибка синего экрана с кодом 0x0000001E". Находит скриншот из логов, момент из записи экрана пользователя и статью в базе знаний.
• Маркетолог спрашивает "где в наших роликах появляется наш продукт". Получает таймкоды всех упоминаний, даже если в аудио о нем не сказано ни слова.

Это не будущее. Это сегодня. Стоимость эмбеддинга одного кадра - около $0.0001. Поиск по миллиону векторов - доли секунды.

1Готовим инструменты: что нам понадобится

Не будем изобретать колесо. Возьмем проверенный стек:

Первое, что нужно сделать - получить API ключ в Google AI Studio. Бесплатная квота - 1500 запросов в минуту, для начала хватит.

Второе - создать проект в Supabase. Включаем расширение pgvector в настройках базы. Запоминаем connection string.

# Установим все сразу. Убедитесь, что Python 3.10+ установлен.
pip install google-generativeai supabase pillow moviepy opencv-python numpy

2Ломаем видео на кадры: где здесь семантика?

Самая частая ошибка - пытаться скормить модели все видео целиком. Gemini Embedding 2 принимает до 16 изображений за запрос (на март 2026). Значит, нужно выбрать репрезентативные кадры.

Плохой подход: брать каждый 10-й кадр. Получите 300 одинаковых эмбеддингов для статичной сцены.

Хороший подход: детектировать смену сцены. OpenCV делает это из коробки.

import cv2
from pathlib import Path

class VideoProcessor:
    def __init__(self, threshold=30.0):
        self.threshold = threshold  # Порог для определения смены сцены

    def extract_key_frames(self, video_path, output_dir, interval_sec=2):
        \"\"\"Извлекаем ключевые кадры по смене сцены + раз в N секунд для надежности\"\"\"
        cap = cv2.VideoCapture(str(video_path))
        fps = cap.get(cv2.CAP_PROP_FPS)
        frame_count = 0
        prev_frame = None
        frames_saved = []

        Path(output_dir).mkdir(parents=True, exist_ok=True)

        while True:
            ret, frame = cap.read()
            if not ret:
                break

            # Сохраняем первый кадр всегда
            if prev_frame is None:
                self._save_frame(frame, output_dir, frame_count, frames_saved)
                prev_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
                frame_count += 1
                continue

            # Детектор смены сцены
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            diff = cv2.absdiff(gray, prev_frame)
            non_zero_count = np.count_nonzero(diff > self.threshold)

            # Или если прошло N секунд
            time_passed = frame_count / fps
            if non_zero_count > 1000 or time_passed % interval_sec < 1/fps:
                self._save_frame(frame, output_dir, frame_count, frames_saved)
                prev_frame = gray

            frame_count += 1

        cap.release()
        return frames_saved

    def _save_frame(self, frame, output_dir, frame_id, frames_list):
        filename = f\"frame_{frame_id:06d}.jpg\"
        filepath = Path(output_dir) / filename
        cv2.imwrite(str(filepath), frame)
        frames_list.append(str(filepath))
        return filepath

Этот код сохраняет кадр при значительном изменении изображения или каждые 2 секунды. На 5-минутном видео получится ~150 кадров вместо 9000. Экономия на эмбеддингах - 98%.

3Магия единого пространства: от пикселей к векторам

Вот где начинается самое интересное. Gemini Embedding 2 принимает список файлов или URL-адресов и возвращает векторы одинаковой размерности (1024, если использовать embedding-002). Независимо от того, что на входе: JPEG, PNG, текст, PDF.

import google.generativeai as genai
from pathlib import Path
import time

class GeminiEmbedder:
    def __init__(self, api_key, model_name=\"embedding-002\"):
        genai.configure(api_key=api_key)
        self.model = model_name
        # На март 2026 доступны embedding-001 (512d) и embedding-002 (1024d)
        # Вторая точнее, но дороже. Для мультимодального поиска нужна именно 002

    def embed_files(self, file_paths, batch_size=16):
        \"\"\"Создает эмбеддинги для списка файлов. API принимает до 16 файлов за запрос.\"\"\"
        all_embeddings = []

        for i in range(0, len(file_paths), batch_size):
            batch = file_paths[i:i+batch_size]
            print(f\"Обрабатываю batch {i//batch_size + 1}: {len(batch)} файлов\")

            try:
                # Ключевой вызов - embed_files вместо embed_content
                result = genai.embed_files(
                    model=self.model,
                    files=batch,
                    task_type=\"retrieval_document\",  # Важно для RAG!
                )
                all_embeddings.extend(result.embeddings)
            except Exception as e:
                print(f\"Ошибка с batch {i}: {e}\")
                # Заполняем нулями для сохранения порядка
                all_embeddings.extend([None] * len(batch))

            time.sleep(0.1)  # Уважаем rate limit

        return all_embeddings

    def embed_text(self, texts):
        \"\"\"Текстовые эмбеддинги - для поисковых запросов\"\"\"
        result = genai.embed_content(
            model=self.model,
            content=texts,
            task_type=\"retrieval_query\",  # Другой task_type для запросов!
        )
        return result.embeddings

Обратите внимание на параметр task_type. Для документов (кадров видео, картинок) используем \"retrieval_document\". Для поисковых запросов - \"retrieval_query\". Модель оптимизирует векторы под задачу. Если перепутать, качество поиска упадет на 20-30%.

4Собираем пазл: Supabase как мультимодальный мозг

Хранить векторы и метаданные нужно правильно. Простая таблица не подойдет - у нас есть и бинарники, и векторы, и текстовые описания.

Создаем в Supabase таблицу:

-- Включаем расширение если еще не включено
create extension if not exists vector;

-- Основная таблица для мультимодальных документов
create table multimodal_documents (
    id bigint generated by default as identity primary key,
    content_type text not null check (content_type in ('video_frame', 'image', 'text')),
    source_file text,  -- Исходный файл (video.mp4)
    file_path text,    -- Путь к конкретному кадру/изображению
    storage_url text,  -- URL в Supabase Storage
    description text,  -- Текстовое описание (можно генерировать Gemini)
    embedding vector(1024),  -- Вектор от embedding-002
    timestamp_sec float,     -- Для видео: секунда в исходном файле
    created_at timestamp with time zone default now()
);

-- Индекс для поиска по косинусному расстоянию
create index on multimodal_documents 
using ivfflat (embedding vector_cosine_ops)
with (lists = 100);  -- Для ~1M векторов

Теперь Python-код для сохранения всего в кучу:

from supabase import create_client, Client
import numpy as np

class MultimodalStore:
    def __init__(self, supabase_url, supabase_key):
        self.supabase: Client = create_client(supabase_url, supabase_key)

    def upload_to_storage(self, file_path, bucket=\"multimodal\"):
        \"\"\"Загружает файл в Supabase Storage, возвращает public URL\"\"\"
        file_name = Path(file_path).name
        with open(file_path, 'rb') as f:
            self.supabase.storage.from_(bucket).upload(file_name, f)
        
        # Получаем публичный URL
        return self.supabase.storage.from_(bucket).get_public_url(file_name)

    def insert_document(self, doc_data):
        \"\"\"Вставляет документ с вектором в базу\"\"\"
        # Вектор нужно преобразовать в список для Postgres
        if doc_data.get('embedding') is not None:
            doc_data['embedding'] = doc_data['embedding'].tolist() if \
                hasattr(doc_data['embedding'], 'tolist') else doc_data['embedding']

        response = self.supabase.table('multimodal_documents').insert(doc_data).execute()
        return response.data[0] if response.data else None

    def search_similar(self, query_embedding, limit=10, content_type=None):
        \"\"\"Ищет похожие документы по косинусному расстоянию\"\"\"
        query = self.supabase.rpc('match_documents', {
            'query_embedding': query_embedding.tolist(),
            'match_threshold': 0.7,  # Минимальное сходство
            'match_count': limit
        })

        if content_type:
            query = query.eq('content_type', content_type)

        result = query.execute()
        return result.data

Функцию match_documents нужно создать в Supabase как stored procedure. Иначе поиск будет медленным.

create or replace function match_documents(
  query_embedding vector(1024),
  match_threshold float,
  match_count int
)
returns table (
  id bigint,
  content_type text,
  source_file text,
  storage_url text,
  description text,
  timestamp_sec float,
  similarity float
)
language sql stable
as $$
  select
    id,
    content_type,
    source_file,
    storage_url,
    description,
    timestamp_sec,
    1 - (embedding <=> query_embedding) as similarity
  from multimodal_documents
  where 1 - (embedding <=> query_embedding) > match_threshold
  order by similarity desc
  limit match_count;
$$;

5Собираем все вместе: полный пайплайн от видео до поиска

Теперь склеиваем все компоненты. Представьте, у вас есть папка content/ с видео, картинками и текстовыми файлами.

def process_multimodal_content(content_dir, embedder, store):
    \"\"\"Основной пайплайн обработки мультимодального контента\"\"\"
    all_files = []
    file_metadata = []  # Для хранения метаданных до вставки в БД

    # Обрабатываем видео
    for video_file in Path(content_dir).glob(\"*.mp4\"):
        print(f\"Обрабатываю видео: {video_file.name}\")
        frames = VideoProcessor().extract_key_frames(
            video_file, 
            output_dir=f\"temp_frames/{video_file.stem}\"
        )
        
        for frame_path in frames:
            # Загружаем кадр в storage
            storage_url = store.upload_to_storage(frame_path)
            all_files.append(frame_path)
            file_metadata.append({
                'path': frame_path,
                'type': 'video_frame',
                'source': str(video_file.name),
                'storage_url': storage_url,
                'timestamp': extract_timestamp_from_filename(frame_path)  # Нужно реализовать
            })

    # Обрабатываем изображения
    for img_file in Path(content_dir).glob(\"*.{jpg,png}\"):
        storage_url = store.upload_to_storage(img_file)
        all_files.append(img_file)
        file_metadata.append({
            'path': img_file,
            'type': 'image',
            'source': 'standalone',
            'storage_url': storage_url,
            'timestamp': None
        })

    # Создаем эмбеддинги батчами
    print(f\"Создаю эмбеддинги для {len(all_files)} файлов...\")
    embeddings = embedder.embed_files(all_files)

    # Сохраняем в базу
    for i, (file_path, metadata) in enumerate(zip(all_files, file_metadata)):
        if embeddings[i] is None:
            continue

        doc_data = {
            'content_type': metadata['type'],
            'source_file': metadata['source'],
            'file_path': str(file_path),
            'storage_url': metadata['storage_url'],
            'description': generate_description(file_path),  # Можно использовать Gemini Pro
            'embedding': embeddings[i],
            'timestamp_sec': metadata['timestamp']
        }
        
        store.insert_document(doc_data)
        print(f\"Сохранен документ {i+1}/{len(all_files)}\")

    print(\"Готово! Контент индексирован.\")

# Использование
embedder = GeminiEmbedder(api_key=\"ВАШ_КЛЮЧ\")
store = MultimodalStore(
    supabase_url=\"ВАШ_URL\",
    supabase_key=\"ВАШ_КЛЮЧ\"
)

process_multimodal_content(\"./content\", embedder, store)

После запуска этого кода (он может работать несколько часов для больших архивов) у вас будет полностью индексированная мультимодальная база.

6Ищем: как задавать вопросы системе

Самая интересная часть. Поиск работает с любыми входными данными:

def multimodal_search(query, embedder, store, search_type=\"auto\"):
    \"\"\"Умный поиск по мультимодальной базе\"\"\"
    
    # Определяем тип запроса
    if search_type == \"auto\":
        # Если запрос - путь к файлу, это изображение/видео
        if Path(query).exists():
            # Эмбеддим файл
            query_embedding = embedder.embed_files([query])[0]
        else:
            # Иначе считаем текстом
            query_embedding = embedder.embed_text([query])[0]
    elif search_type == \"text\":
        query_embedding = embedder.embed_text([query])[0]
    elif search_type == \"image\":
        query_embedding = embedder.embed_files([query])[0]
    
    # Ищем похожие
    results = store.search_similar(query_embedding, limit=5)
    
    # Группируем по типу контента
    grouped = {'video_frames': [], 'images': [], 'texts': []}
    for r in results:
        if r['content_type'] == 'video_frame':
            grouped['video_frames'].append(r)
        elif r['content_type'] == 'image':
            grouped['images'].append(r)
        else:
            grouped['texts'].append(r)
    
    return grouped

# Примеры использования:
# 1. Текстовый запрос
results = multimodal_search(\"ошибка загрузки модуля\", embedder, store)
print(\"Найдены кадры видео:\", [r['storage_url'] for r in results['video_frames']])

# 2. Поиск по скриншоту
results = multimodal_search(\"screenshot_error.png\", embedder, store)

# 3. Смешанный запрос: найти похожие на картинку И по тексту
image_results = multimodal_search(\"ui_design.png\", embedder, store)
text_results = multimodal_search(\"темная тема кнопка навигации\", embedder, store)

Где споткнетесь: частые ошибки и их решение

Я собрал список проблем, с которыми столкнулся сам и которые задают в чатах:

• Ошибка: \"Invalid input type\" при вызове embed_files. Проверьте, что файлы существуют и это изображения (JPEG, PNG) или PDF. Gemini Embedding 2 на март 2026 не принимает видеофайлы напрямую - только кадры.
• Поиск возвращает мусор, хотя векторы созданы. Скорее всего, перепутали task_type. Для документов - retrieval_document, для запросов - retrieval_query. Если все документы эмбеддили как query, они будут плохо сравниваться между собой.
• Supabase ругается на размер вектора. embedding-002 возвращает 1024 измерения. В таблице должно быть vector(1024), не vector(512).
• Слишком много кадров из видео, счет за API зашкаливает. Увеличьте threshold в детекторе смены сцены. Или используйте фиксированный интервал в 3-5 секунд вместо 2.
• Не могу найти по тексту то, что видно на картинке. Добавьте текстовые описания. Используйте Gemini Pro Vision, чтобы сгенерировать описание для каждого кадра, и сохраните его в поле description. Тогда поиск по тексту будет искать и в этих описаниях.

Если нужно быстро развернуть простой видео-RAG без всей этой сложности, есть вариант попроще: Локальный RAG для видео. Но там не будет мультимодальности.

Куда это развивать? (Неочевидные применения)

Стандартное применение - поиск по базе знаний. Скучно. Вот что можно сделать еще:

1. Детектор плагиата для дизайнов. Загружаете скриншоты конкурентов - система находит похожие элементы в ваших макетах. Юридический отдел будет в восторге.
2. Автоматическая раскадровка. Ищете \"герой открывает дверь, крупный план\" - система находит все подобные сцены в архиве видео. Режиссеры монтажа экономят недели.
3. Поиск продукта по фото в соцсетях. Пользователь сфотографировал чью-то куртку - система находит ее в вашем каталоге. Даже если на фото только часть куртки.
4. Валидация контента. Автоматически находите кадры с логотипами конкурентов или неподходящим контентом. Можно комбинировать с детектором AI-генерации.

Самое интересное - когда соединяете мультимодальный RAG с генерацией. Нашли похожие кадры → сгенерировали на их основе новое видео через Veo 3.1. Получается бесконечный цикл креатива.

Главный совет: не храните векторы рядом с бинарниками. Supabase Storage может стоить дороже S3. Лучше хранить файлы в S3-совместимом хранилище, а в Supabase держать только метаданные и векторы. Или использовать AI Bridge для автоматизации переноса данных между системами.

Музыка мультимодального поиска только начинается. Текст был королем. Картинки были принцами. Теперь все данные равны в векторах. И это меняет правила игры.

Подписаться на канал