Gemini APIによる長尺動画のインデックス作成：タイムスタンプ、映像解析、ギャップ監査

技術講演、ワークショップ、アーキテクチャレビューの会議、スクリーンキャストといった長尺の動画記録には、登壇者の音声と画面上の視覚情報の双方にわたって高密度な情報が含まれています。しかし、標準的な大まかな要約では全体のトピックしか把握できず、特定のターミナルコマンドや設定パラメータを確認したい場合、エンジニアは動画のシークバーを手動で操作して該当箇所を探し出さざるを得ません。

Geminiのマルチモーダルモデルは、音声と映像のストリームを同時に解析し、タイムスタンプが付与された構造化イベントインデックスを生成できます。ただし、モデルの初回推論で得られる生の出力はあくまで**下書き候補の集合（ドラフト候補群）**であり、そのまま本番環境で参照できる信頼性の高い目次ではありません。実用に耐えうるインデックスを構築するには、Files APIによる一度のアップロードとURIの再利用、下書き区間の抽出、出力スキーマの厳格な検証、不自然なカバレッジギャップの監査、そしてピンポイントのキャリブレーションを行う規律あるパイプラインが不可欠です。

---

アーキテクチャ：動画の配信方法と処理モード

現在のGoogle Video Understandingドキュメントにおいて、主要な実装はInteractions APIおよびgoogle-genaiライブラリを中心として構成されています。従来のgenerate_contentメソッドも下位互換性のために引き続きサポートされていますが、Interactions APIのほうがマルチモーダル処理パラメータをより明確に制御できます。

1. 動画の配信方法

• Files API（長尺動画に推奨）：数分から数時間に及ぶ動画ファイルに最適です。ファイルは一度アップロードされればサーバー上でインデックス化され、生のバイト列を再送信することなく、URI参照によって反復リクエストに利用できます。
• Google Cloud Storage (GCS)：すでにGoogle Cloudのインフラストラクチャ上に配置されている既存の動画アーカイブに適しています。
• インラインデータ（Inline Data）：リクエストのペイロード内で生のバイト列を直接送信します。ドキュメントには環境ごとのさまざまなペイロード制限が記載されていますが、1時間のメディアを安定して扱うには、生の動画をインラインでストリーミングするのを避け、Files APIまたはCloud Storageを利用するのが現実的です。

2. 処理モード：Static vs. Agentic

• 静的処理（Static Processing）：
  デフォルトでは、モデルは1秒あたり1フレーム（1 FPS）の離散レートでフレームをサンプリングします。1秒ごとにトークンへと変換されるため、60分の動画では膨大なコンテキスト負荷が蓄積されます。1 FPSのサンプリングは瞬間的な視覚的イベントを見逃す可能性があり（1秒未満のウィンドウ切り替えや短いツールチップなど）、すべての微小なインタラクションの捕捉を保証できるわけではない点に留意してください。
• エージェンティック動画理解（Agentic Video Understanding）：
  モデルが動画内を動的にナビゲートし、必要に応じて特定のフレームや音声セグメントを取得します。これにより、処理されるコンテキストトークンの量が大幅に削減されます。
  制約事項： エージェンティックのヒューリスティックは、音声の手がかりや意味的なトリガーに依存しています。もし登壇者が何も話さずに画面上で操作（コマンドの入力やダイアグラムの確認など）を行った場合、ヒューリスティックはその区間をアイドルの背景と見なし、詳細な映像フレームのリクエストをスキップしてしまう可能性があります。

---

ステップ1：動画のアップロードとステータスのポーリング

Files APIにアップロードされたファイルは、すぐに推論に使用できるわけではありません。サーバー側でコンテナを展開し、映像・音声トラックをインデックス化する必要があります。そのため、アプリケーション側でリソースの状態が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による下書きインデックスのリクエスト

長尺の録画に対しては、"processing": "agentic"を指定してclient.interactions.createを呼び出します。プロンプトでは厳格な表形式の出力を指定します。具体的には、タイムスタンプ範囲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

エージェンティックナビゲーションのステップについて：
interaction.steps内にprocessing_callのエントリが存在することは、モデルが動画を連続ストリームとして読み込むのではなく、タイムラインを動的にナビゲートしたことを裏付けます。ただし、これらの呼び出しが確認できたとしても、それはナビゲーション動作が行われたことの証明に過ぎず、タイムライン上のすべての関連イベントにおける出力の完全性を保証するものではありません。

---

ステップ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の生の出力は、検証なしに構造化データセットとして信頼することはできません。堅牢なパーサーは、破損したレコードを暗黙的に破棄するのではなく、手動編集のためにそれらを隔離する必要があります。バリデータは、正確に5つのフィールドが存在すること、有効なイベントタイプ（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：カバレッジ監査とギャップの特定

インデックスを公開する前に、タイムラインを監査して「ブラインドスポット（死角）」がないか確認します。

1. 基準となるゾーンのスポットチェック：
   動画の冒頭（導入部やタイトルスライド）、中間地点（ライブデモやアーキテクチャの議論がピークに達しやすい箇所）、そして終盤（質疑応答や締めくくりの挨拶）を手動で確認します。
2. タイムスタンプの間隔分析：
   インデックス項目間の許容間隔に一律のしきい値は存在せず、許容範囲はユースケースによって異なります。情報密度の高いスクリーンキャストでは、90秒のギャップは設定手順の抜け落ちを意味する可能性があります。一方で概説講義であれば、1つの論点が5分間に及ぶことも十分に妥当です。間隔がプレゼンテーションのペースと不整合に見える場合は、疑わしい区間としてフラグを立てます。
3. 音声のない視覚的イベントの検査：
   登壇者がナレーションなしで画面操作を行った場合、エージェンティックモードはそのセグメントを大まかに見過ごしてしまった可能性があります。そのような時間枠は、ピンポイントの静的レビューへとルーティングします。

---

ステップ6：静的クリップによる疑わしいゾーンのピンポイント検査

疑わしいセグメントを再評価するために、60分の動画全体を再実行する必要はありません。クリップのスライシング機能により、静的モードでの解析を秒単位の正確な境界（start_offsetおよびend_offset）に限定できます。

手法の制約事項：
1 FPSでの静的モードは固定のサンプリンググリッドを提供し、全体的なエージェンティックパスで見逃されたアクションを発見するのに役立ちます。ただし、絶対的な再現率（網羅性）を保証するものではありません。1秒未満で発生した視覚的な変化は、サンプリングフレームの間に抜け落ちる可能性があります。

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を使用することで、ユーザーは特定の主張や画面アクションを検索し、元の録画の該当セグメントへ直接ジャンプできるようになります。ただし、編集担当者が元の録画と照らし合わせて、単なるサンプリング境界だけでなく、どのイベントが選択されたかとそのイベント境界の双方を確認するまでは、あくまで下書き（ドラフト）の位置付けとなります。

---

トラブルシューティングとエッジケース

• Files APIアップロード時のFAILEDステータス：
  APIからの診断情報なしに原因を判断することは避けてください。失敗の原因としては、サポートされていないコンテナ形式、壊れたファイルヘッダー、あるいは一時的なインフラエラーなどが考えられます。SDKを介してfile.error属性を確認し、ffprobeでローカル再生を検証した上で、必要に応じてffmpeg（-c:v libx264 -c:a aac）でストリームを標準化して再試行してください。
• 401 Unauthorizedまたはネットワーク切断：
  401エラーは認証の失敗（無効または欠落しているキー、あるいは未設定のGEMINI_API_KEY環境変数）を一意に示すものであり、処理セッションの期限切れではありません。長時間の呼び出し中にクライアントのHTTP接続タイムアウトを防ぐには、stream=Trueでストリーミングを有効にしてください。
• 合計時間を超えるタイムスタンプ：
  コンテキストの複雑さが増すにつれて、このようなタイムスタンプのドリフトが発生することがあります。プロンプトでの厳格な指示と、パーサーにおけるプログラム的な検証（e_sec > max_duration_sec）によって対処します。
• TSVの構造的ドリフト：
  フォーマットが崩れた場合は、不正な行をmalformed_rowsにルーティングし、システムプロンプトに1〜2行のfew-shot参照例を追加してください。

---

公開前の検証ワークフロー

1. アセットの準備完了を確認：Files APIでファイルがACTIVEステータスに達したことを確認します。
2. 初期ドラフトの生成：Interactions API経由でagentic処理を使用し、ベースラインとなるインデックスを作成します。
3. プログラムによる検証の実行：すべての行に必須の5フィールド、有効なタイプ、およびMM:SS形式が含まれていることを確認します。修正が必要な不正行を隔離します。
4. カバレッジの監査：基準ゾーン（冒頭、中間、終盤）を検査し、講演のペースに照らし合わせてタイムスタンプの密度を評価します。
5. ピンポイント検査の実施：静的クリップ（start_offset/end_offset）を使用して、疑わしいギャップや無言の画面操作セクションを再検証します。
6. 手動によるスポットキャリブレーションの実施：動画プレーヤーでランダムなタイムスタンプをシークし、ユースケースの要件に応じて、関連するソースの音声または映像と各イベントの実際の開始時刻を照合します。音声のみのイベントも発生し得るため、画面の変化との一致を求めないでください。

リクエストスキーマ、処理モード、およびサンプリング制約の詳細については、Gemini Video Understandingの公式ドキュメントを参照してください。