Индексация длинных видео через Gemini API: таймкоды, визуальный ряд и аудит пропусков
Инженерный подход к созданию структурированного индекса длинных видеозаписей через Gemini API: загрузка через Files API, агентный черновик в Interactions API, программная валидация строк и точечный досмотр пропусков.
Содержание
Длинные видеозаписи — технические доклады, воркшопы, архитектурные созвоны и скринкасты — содержат плотный поток информации, распределенный между голосом спикера и демонстрацией экрана. Обычное текстовое резюме фиксирует лишь верхнеуровневый контекст, из-за чего для поиска конкретной команды в терминале или параметра конфигурации приходится пересматривать видео вручную.
Мультимодальные модели Gemini способны анализировать звук и изображение синхронно, формируя структурированный индекс событий с временными метками. При этом результат первичного прогона нейросети — это черновой набор кандидатов, а не готовое эталонное оглавление. Для получения надежного индекса требуется прозрачный пайплайн: загрузка файла, извлечение черновых интервалов, строгая валидация структуры, поиск подозрительных пропусков и финальная ручная калибровка.
Архитектура: способы передачи и режимы обработки
В актуальной документации Google Video Understanding основные примеры построены вокруг интерфейса Interactions API и библиотеки google-genai. При этом классический метод generate_content не упразднен и сохраняет обратную совместимость, однако для комплексных мультимодальных сценариев Interactions API предлагает более прозрачное управление параметрами обработки.
1. Способы передачи видео
- Files API (рекомендуемый для длинных видео): Оптимален для файлов продолжительностью от нескольких минут до нескольких часов. Файл загружается один раз, проходит предварительную обработку на сервере и становится доступен по URI для серии повторных запросов без повторной передачи байтов.
- Google Cloud Storage (GCS): Подходит для видеоархивов, уже размещенных в инфраструктуре Google Cloud.
- Встраивание в запрос (Inline Data): Передача данных непосредственно в теле вызова. Документация приводит различные пороговые значения для допустимого объема inline-данных в зависимости от платформы и типа запроса, поэтому для стабильной работы с часовыми записями следует опираться на Files API или Cloud Storage, избегая передачи сырого потока в теле запроса.
2. Режимы обработки: Static vs Agentic
- Статический режим (Static Processing):
По умолчанию модель производит дискретную выборку кадров с частотой 1 кадр в секунду (1 FPS). Каждая секунда трансформируется в токены, создавая значительную контекстную нагрузку на 60-минутном отрезке. Важно учитывать: выборка 1 FPS может пропускать кратковременные визуальные события (субсекундные переключения окон, короткие всплывающие подсказки) и не гарантирует фиксацию абсолютно каждой детали. - Агентный режим (Agentic Video Understanding):
Модель динамически перемещается по видеоряду, запрашивая кадры и фрагменты звуковой дорожки по мере необходимости. Это сокращает объем обрабатываемого контекста.
Ограничение: Агентный алгоритм ориентируется на смысловые триггеры и звук. Если спикер замолчал и без комментариев выполнил действия на экране (ввел команду, открыл схему), эвристика может счесть этот отрезок фоновым и не запросить детальные кадры.
Шаг 1. Загрузка видео и опрос статуса
Файл, переданный в Files API, не готов к генерации немедленно: сервер производит распаковку контейнера и индексацию дорожек. Требуется цикл опроса (polling) до перехода ресурса в состояние ACTIVE.
import time
from google import genai
client = genai.Client()
video_path = "tech_workshop_60min.mp4"
video_file = client.files.upload(file=video_path)
while not video_file.state or video_file.state.name != "ACTIVE":
if video_file.state and video_file.state.name == "FAILED":
error_info = getattr(video_file, "error", None)
raise RuntimeError(
f"Файл перешел в статус FAILED. Детали: {error_info}. "
"Проверьте кодеки, целостность контейнера или повторите попытку."
)
time.sleep(5)
video_file = client.files.get(name=video_file.name)
print("Видео готово к обработке.")Шаг 2. Запрос чернового индекса через Interactions API
Для длинной записи вызывается метод client.interactions.create с указанием параметра "processing": "agentic". В промпте задается строгая структура строк: временной диапазон MM:SS - MM:SS, тип события (visual, speech, hybrid), краткий тезис речи (CLAIM) и действие на экране (ACTION).
index_prompt = """
Ты — инструмент технической индексации видеозаписей.
Сформируй хронологический индекс событий для всей 60-минутной записи от 00:00 до 60:00.
Требования к структуре ответа:
1. Выведи результат построчно в формате TSV с разделителем |.
2. Каждая строка должна содержать ровно 5 полей:
START_TIME (MM:SS) | END_TIME (MM:SS) | TYPE (visual/speech/hybrid) | CLAIM | ACTION
3. Не объединяй слишком длинные интервалы в одну запись; фиксируй смену слайдов, терминал, ошибки и выводы спикера.
4. Если речи не было, в поле CLAIM укажи NONE. Если на экране не было динамики, в ACTION укажи STATIC.
5. Выводи исключительно строки данных без вводных слов и пояснений.
"""
interaction = client.interactions.create(
model="gemini-3.8-flash",
input=[
{
"type": "video",
"uri": video_file.uri,
"mime_type": video_file.mime_type,
"processing": "agentic"
},
{
"type": "text",
"text": index_prompt
}
]
)
raw_index_output = interaction.output_textО шагах агентной навигации:
Присутствие элементовprocessing_callв объектеinteraction.stepsсвидетельствует о том, что модель действительно использовала динамическую навигацию по таймлайну, а не читала видео непрерывным потоком. Однако наличие таких вызовов подтверждает лишь сам факт работы навигации, но не гарантирует полноту выборки (output completeness) всех значимых событий.
Шаг 3. Пример структуры вывода (гипотетический драфт)
Ниже приведен гипотетический фрагмент ответа модели для иллюстрации структуры данных:
00:00 | 02:15 | hybrid | Вводное слово, обзор повестки миграции на шардированный кластер | Титульный слайд доклада, окно спикера
02:16 | 05:40 | visual | NONE | Переключение на схему архитектуры сервиса заказов в Miro
05:41 | 09:12 | speech | Пояснение причин отказа от распределенных транзакций в пользу Saga | Статичная схема Miro, курсор неподвижен
27:30 | 29:10 | visual | NONE | Открытие консоли, запуск сценария развертывания реплик БД
29:11 | 32:45 | hybrid | Разбор сценария split-brain при потере сетевой связности | Вывод логов etcd в терминале, подсветка таймаутов
54:10 | 57:25 | speech | Ответ на вопрос о допустимой задержке репликации данных | Финальный слайд с контактами спикера
57:26 | 60:00 | hybrid | Подведение итогов и демонстрация ссылки на репозиторий | Показ QR-кода на экране, завершение созвонаШаг 4. Валидация разметки и локальный поиск
Сырой вывод LLM нельзя считать надежной базой данных без валидации. Парсер не должен молча отбрасывать некорректные строки: их необходимо сохранять отдельно для ручной корректировки. Проверке подлежат: наличие ровно пяти полей, корректность типа события (visual, speech, hybrid) и соответствие формата времени диапазону MM:SS.
import csv
import re
from typing import List, Dict, Tuple
TIME_PATTERN = re.compile(r"^(\d{2}):([0-5]\d)$")
ALLOWED_TYPES = {"visual", "speech", "hybrid"}
def time_to_seconds(t_str: str) -> int:
match = TIME_PATTERN.match(t_str)
if not match:
raise ValueError(f"Некорректный формат времени: {t_str}")
m, s = map(int, match.groups())
return m * 60 + s
def parse_and_validate_tsv(
raw_text: str,
max_duration_sec: int = 3600
) -> Tuple[List[Dict[str, str]], List[Dict[str, str]]]:
valid_rows = []
malformed_rows = []
lines = [line.strip() for line in raw_text.strip().splitlines() if line.strip()]
for idx, line in enumerate(lines, start=1):
cols = [c.strip() for c in line.split("|")]
if len(cols) != 5:
malformed_rows.append({
"line": idx,
"content": line,
"error": f"Ожидалось 5 полей, получено {len(cols)}"
})
continue
start_str, end_str, event_type, claim, action = cols
if event_type.lower() not in ALLOWED_TYPES:
malformed_rows.append({
"line": idx,
"content": line,
"error": f"Недопустимый тип события: {event_type}"
})
continue
try:
s_sec = time_to_seconds(start_str)
e_sec = time_to_seconds(end_str)
except ValueError as err:
malformed_rows.append({
"line": idx,
"content": line,
"error": str(err)
})
continue
if s_sec > e_sec:
malformed_rows.append({
"line": idx,
"content": line,
"error": f"Время начала ({start_str}) больше времени окончания ({end_str})"
})
continue
if e_sec > max_duration_sec:
malformed_rows.append({
"line": idx,
"content": line,
"error": f"Таймкод {end_str} выходит за хронометраж ({max_duration_sec} сек)"
})
continue
valid_rows.append({
"start": start_str,
"end": end_str,
"start_sec": s_sec,
"end_sec": e_sec,
"type": event_type.lower(),
"claim": claim,
"action": action
})
return valid_rows, malformed_rows
def search_index(
rows: List[Dict[str, str]],
keyword: str = None,
event_type: str = None
) -> List[Dict[str, str]]:
results = []
for r in rows:
if event_type and r["type"] != event_type.lower():
continue
if keyword:
kw = keyword.lower()
if kw not in r["claim"].lower() and kw not in r["action"].lower():
continue
results.append(r)
return results
valid_index, errors = parse_and_validate_tsv(raw_index_output, max_duration_sec=3600)
if errors:
print(f"Обнаружено некорректных строк: {len(errors)}. Требуется ручная правка:")
for err in errors:
print(f"Строка {err['line']}: {err['error']} -> {err['content']}")
else:
print(f"Все строки валидны. Записей в индексе: {len(valid_index)}")Шаг 5. Аудит покрытия и поиск пропусков
Перед публикацией индекс проходит проверку на наличие «слепых зон»:
- Выборочный аудит контрольных зон:
Проверьте вручную начало видео (вводная часть и первые слайды), экватор (середина записи, где обычно сосредоточена демонстрация кода или дискуссия) и финальный блок (ответы на вопросы). - Анализ временных разрывов:
Универсального допустимого интервала между метками не существует: допустимая погрешность и допустимый разрыв определяются задачей. Для динамичного скринкаста пропуск в 90 секунд может означать потерю важной настройки, а для обзорной лекции один тезис на 5 минут является нормой. Если обнаружен разрыв, не соответствующий динамике выступления, этот отрезок помечается как подозрительный. - Поиск «тихих» визуальных событий:
Если спикер выполнял демонстрацию без звуковых комментариев, агентный режим мог пропустить этот отрезок. Такие интервалы направляются на точечный досмотр.
Шаг 6. Точечный досмотр подозрительных зон через Static Clip
Для перепроверки подозрительного интервала нет необходимости повторно прогонять весь часовой файл. Механизм нарезки клипов позволяет ограничить анализ точными смещениями в секундах (start_offset и end_offset) в статическом режиме.
Ограничение метода:
Статический режим с частотой 1 FPS обеспечивает регулярную сетку выборки, помогая выявить пропущенные на общем плане действия. Однако он не гарантирует абсолютную точность и полноту (recall): любые изменения, произошедшие быстрее одной секунды, могут не попасть в кадр.
start_sec = 1200
end_sec = 1380
targeted_inspection = client.interactions.create(
model="gemini-3.8-flash",
input=[
{
"type": "video",
"uri": video_file.uri,
"mime_type": video_file.mime_type,
"processing": {
"type": "static",
"start_offset": start_sec,
"end_offset": end_sec,
"fps": 1.0
}
},
{
"type": "text",
"text": (
"Хронологически опиши изменения на экране. "
"Зафиксируй команды в терминале, смену окон и системные сообщения."
)
}
]
)
print("Результат точечного досмотра отрезка:")
print(targeted_inspection.output_text)Полученные уточнения интегрируются в проверенный массив строк вручную или в рамках полуавтоматической сборки.
После исправления строк из errors и проверки найденных таймкодов в исходном видео сохраните итоговый индекс. Пример ниже намеренно останавливает экспорт, пока парсер сообщает об ошибках; перед запуском замените valid_index на массив с уже внесёнными и проверенными исправлениями.
if errors:
raise ValueError("Исправьте строки из errors и повторите проверку перед экспортом")
with open("video_index.csv", "w", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(
output_file,
fieldnames=["start", "end", "type", "claim", "action"],
extrasaction="ignore",
)
writer.writeheader()
writer.writerows(valid_index)CSV позволяет искать тезис или действие и переходить к соответствующему отрезку исходной записи. Он остаётся черновиком, пока ответственный редактор не подтвердит выборку и границы событий.
Диагностика сбоев и граничные случаи
- Статус
FAILEDпри загрузке в Files API:
Не следует делать категоричных выводов о причине сбоя без данных от API. Ошибка может быть вызвана неподдерживаемым контейнером, поврежденным заголовком или временной проблемой инфраструктуры. Проверьте атрибутfile.errorчерез SDK, протестируйте воспроизведение локального файла черезffprobe, при необходимости выполните стандартизацию черезffmpeg(-c:v libx264 -c:a aac) и повторите попытку. - Ошибка
401 Unauthorizedили сетевые сбои:
Ошибка 401 однозначно указывает на проблемы с аутентификацией (недействительный или отсутствующий API-ключ, проблемы с переменной окруженияGEMINI_API_KEY), а не на длительность сессии. Для предотвращения разрыва клиентских HTTP-соединений при долгих операциях используйте потоковый режимstream=True. - Выход таймкодов за границы длительности:
Может возникать при усложнении контекста. Парируется строгой инструкцией в промпте и программной проверкойe_sec > max_duration_secв валидаторе. - Нарушение структуры TSV:
При дрейфе формата вывод передается парсеру с фильтрациейmalformed_rows, а в системную инструкцию добавляются 1–2 примера эталонных строк (few-shot).
Рабочий процесс верификации перед публикацией
- Подтверждение готовности ресурса: файл переведен в статус
ACTIVEв Files API. - Первичный драфт: сформирован в режиме
agenticчерез Interactions API. - Программный контроль структуры: строки проверены валидатором на 5 обязательных полей, валидность типов и формат
MM:SS. Некорректные строки изолированы для исправления. - Аудит покрытия: проверены контрольные зоны (начало, середина, конец), а интервалы сопоставлены с допустимой плотностью под конкретную задачу.
- Точечный досмотр: подозрительные зоны с потенциальными визуальными пропусками перепроверены через статический срез (
start_offset/end_offset). - Выборочная ручная калибровка: оператор проверяет несколько случайных таймкодов в видеоплеере, сопоставляя фактическое начало события с требованиями сценария.
Описание режимов, структуры запроса и ограничений выборки: официальная документация Gemini Video Understanding.
