Indexación de videos largos con la API de Gemini: marcas de tiempo, elementos visuales y auditoría de brechas
Un flujo de trabajo de ingeniería para indexar grabaciones de video largas con la API de Gemini: carga mediante la Files API, generación de borradores agénticos en la Interactions API, validación programática de filas estructuradas e inspección dirigida de brechas de cobertura.
Índice
Las grabaciones de video de larga duración —charlas técnicas, talleres, sesiones de revisión de arquitectura y screencasts— concentran una alta densidad de información distribuida entre el audio hablado y los elementos visuales en pantalla. Los resúmenes generales estándar solo capturan temas amplios, lo que obliga a los ingenieros a recorrer manualmente el metraje cada vez que necesitan un comando de terminal o un parámetro de configuración específico.
Los modelos multimodales de Gemini pueden analizar transmisiones de audio y video de forma simultánea, generando índices estructurados de eventos con marcas de tiempo. No obstante, la salida directa de una primera pasada de inferencia del modelo es únicamente un conjunto preliminar de candidatos y no un esquema de referencia listo para producción. Construir un índice fiable exige una canalización rigurosa: subir el archivo una sola vez mediante la Files API y reutilizar su URI, extraer intervalos preliminares, validar estrictamente el esquema de salida, auditar brechas de cobertura sospechosas y llevar a cabo una calibración dirigida.
Arquitectura: métodos de entrega y modos de procesamiento
En la documentación actual de Google Video Understanding, las implementaciones principales giran en torno a la Interactions API y la biblioteca google-genai. Aunque el método clásico generate_content sigue siendo compatible por motivos de retrocompatibilidad, la Interactions API ofrece un control más claro sobre los parámetros de procesamiento multimodal.
1. Métodos de entrega de video
- Files API (recomendada para videos largos): La opción más adecuada para grabaciones que van desde varios minutos hasta varias horas. El archivo se sube una sola vez, se indexa en el servidor y se referencia mediante su URI a lo largo de solicitudes sucesivas sin volver a transmitir los bytes sin procesar.
- Google Cloud Storage (GCS): Ideal para archivos de video existentes que ya se encuentran alojados en la infraestructura de Google Cloud.
- Datos insertados (Inline Data): Transmisión directa de bytes sin procesar dentro del cuerpo de la solicitud. La documentación detalla restricciones variables de carga útil según el entorno; para un manejo estable de material de una hora de duración, conviene recurrir a la Files API o Cloud Storage, evitando transmitir video sin procesar en línea.
2. Modos de procesamiento: estático vs. agéntico
- Procesamiento estático (Static Processing):
De forma predeterminada, el modelo toma muestras de fotogramas a una tasa discreta de 1 fotograma por segundo (1 FPS). Cada segundo se convierte en tokens, lo que acumula una carga considerable de contexto a lo largo de un período de 60 minutos. Tenga en cuenta lo siguiente: el muestreo a 1 FPS puede omitir eventos visuales breves (como cambios de ventana en fracciones de segundo o mensajes emergentes fugaces) y no garantiza la captura de cada microinteracción. - Comprensión agéntica de video (Agentic Video Understanding):
El modelo navega por el video de manera dinámica, solicitando fotogramas específicos y segmentos de audio según sea necesario. Esto reduce de forma drástica el volumen de tokens de contexto procesados.
Limitación: La heurística agéntica depende de pistas de audio y disparadores semánticos. Si un ponente realiza acciones en pantalla en silencio (como escribir un comando o examinar un diagrama) sin hablar, la heurística puede tratar dicho intervalo como fondo inactivo y omitir la solicitud de fotogramas visuales detallados.
Paso 1. Carga de video y sondeo de estado
Los archivos subidos a la Files API no están listos para la inferencia de inmediato; el servidor debe desempaquetar el contenedor e indexar las pistas audiovisuales. Su aplicación debe sondear el recurso hasta que su estado cambie a 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("Видео готово к обработке.")Paso 2. Solicitud de un borrador de índice mediante la Interactions API
Para grabaciones largas, invoque client.interactions.create con "processing": "agentic". El prompt exige una estructura tabular estricta: intervalo de marcas de tiempo MM:SS - MM:SS, tipo de evento (visual, speech, hybrid), una afirmación breve de lo hablado (CLAIM) y la acción en pantalla (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_textSobre los pasos de navegación agéntica:
La presencia de entradasprocessing_calldentro deinteraction.stepsconfirma que el modelo navegó dinámicamente por la línea de tiempo del video en lugar de procesar una transmisión continua. Sin embargo, ver estas llamadas solo confirma la actividad de navegación: no garantiza la exhaustividad de los resultados en todos los eventos relevantes de la línea de tiempo.
Paso 3. Ejemplo de estructura de salida (borrador hipotético)
A continuación se muestra un fragmento hipotético ilustrativo de la salida del modelo que demuestra el esquema de datos esperado:
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-кода на экране, завершение созвонаPaso 4. Validación del marcado y búsqueda local
Las respuestas sin procesar del LLM no se pueden considerar conjuntos de datos estructurados fiables sin una verificación previa. Un analizador resistente no debe descartar en silencio los registros corruptos; en su lugar, debe aislarlos para su corrección manual. El validador comprueba: exactamente cinco campos, tipos de eventos válidos (visual, speech, hybrid) y marcas de tiempo MM:SS bien formadas dentro de los límites de duración del video.
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)}")Paso 5. Auditoría de cobertura e identificación de vacíos
Antes de publicar el índice, audite la línea de tiempo para detectar puntos ciegos:
- Inspeccionar zonas de referencia:
Revise manualmente el inicio del video (introducción y diapositivas de título), el punto intermedio (donde suelen concentrarse las demostraciones en vivo o los debates de arquitectura) y la conclusión (preguntas, respuestas y notas de cierre). - Analizar los intervalos entre marcas de tiempo:
No existe un umbral universal para el espaciado aceptable entre las entradas del índice; la tolerancia depende del caso de uso. En un screencast denso, una brecha de 90 segundos podría significar la pérdida de un paso de configuración. En una conferencia general, una única tesis que abarque 5 minutos puede ser completamente razonable. Si un intervalo parece incoherente con el ritmo de la presentación, márquelo como sospechoso. - Inspeccionar eventos visuales silenciosos:
Si el ponente realizó una demostración en pantalla sin narración, el modo agéntico podría haber pasado por alto ese segmento. Derive dichas ventanas a una revisión estática dirigida.
Paso 6. Inspección dirigida de zonas sospechosas mediante clips estáticos
Reevaluar un segmento sospechoso no requiere procesar nuevamente todo el video de 60 minutos. El recorte de clips restringe el análisis en modo estático a límites precisos en segundos (start_offset y end_offset).
Limitación del método:
El modo estático a 1 FPS proporciona una cuadrícula de muestreo fija que ayuda a descubrir acciones omitidas durante la pasada agéntica general. Sin embargo, no garantiza una exhaustividad absoluta: los cambios visuales que ocurran en menos de un segundo pueden seguir quedando entre los fotogramas muestreados.
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)Integre las entradas recién extraídas en el índice validado, ya sea de forma manual o a través de una etapa de revisión semiautomatizada.
Una vez que haya resuelto todas las entradas en errors y verificado las marcas de tiempo con respecto a la grabación original, exporte el índice final. El siguiente fragmento de código detiene deliberadamente la ejecución si el analizador reporta errores pendientes; asegúrese de que valid_index contenga sus correcciones verificadas antes de ejecutarlo.
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)El archivo CSV resultante permite a los usuarios buscar afirmaciones o acciones en pantalla específicas y saltar directamente a los segmentos pertinentes en la grabación de origen. Seguirá siendo un borrador hasta que un editor humano confirme tanto qué eventos se seleccionaron como sus límites de evento, no meramente los límites de muestreo, frente a la grabación original.
Resolución de problemas y casos extremos
- Estado
FAILEDdurante la carga en la Files API:
Evite diagnosticar el problema sin los datos que proporciona la API. Los fallos pueden deberse a formatos de contenedor no compatibles, encabezados de archivo dañados o errores temporales de infraestructura. Inspeccione el atributofile.errormediante el SDK, verifique la reproducción local conffprobe, estandarice las pistas medianteffmpeg(-c:v libx264 -c:a aac) si es necesario y vuelva a intentarlo. - Error
401 Unauthorizedo caídas de red:
Un error 401 indica explícitamente un fallo de autenticación (una clave no válida o ausente, o una variable de entornoGEMINI_API_KEYno configurada), no una sesión de procesamiento caducada. Para evitar tiempos de espera en las conexiones HTTP del cliente durante llamadas prolongadas, habilite la transmisión mediantestream=True. - Marcas de tiempo que superan la duración total:
Esta desviación puede ocurrir a medida que aumenta la complejidad del contexto. Contarréstela con instrucciones estrictas en el prompt y validación programática (e_sec > max_duration_sec) en su analizador. - Desviación estructural del TSV:
Cuando el formato se rompa, dirija las líneas mal formadas amalformed_rowsy proporcione 1 o 2 líneas de referencia de ejemplo (few-shot) en el prompt del sistema.
Flujo de trabajo de verificación previo a la publicación
- Verificar la preparación del recurso: confirme que el archivo haya alcanzado el estado
ACTIVEen la Files API. - Generar el borrador inicial: cree el índice de referencia utilizando el procesamiento
agentica través de la Interactions API. - Ejecutar la validación programática: asegúrese de que todas las filas contengan los cinco campos obligatorios, tipos válidos y el formato
MM:SS. Aísle las líneas no válidas para su corrección. - Auditar la cobertura: inspeccione las zonas de referencia (inicio, punto medio y final) y evalúe la densidad de las marcas de tiempo frente al ritmo de la charla.
- Realizar una inspección dirigida: reexamine las brechas sospechosas o las secciones de pantalla en silencio mediante clips estáticos (
start_offset/end_offset). - Efectuar una calibración manual puntual: compruebe la hora de inicio real de cada evento frente al audio o video de origen relevante según lo requiera el caso de uso; no exija un cambio visual coincidente, ya que puede ocurrir un evento solo de audio.
Para conocer los esquemas de solicitud, los modos de procesamiento y las restricciones de muestreo, consulte la documentación oficial de Gemini Video Understanding.
