#!/usr/bin/env python """ End-to-end MCQ over the FULL video via Gemini 3 Flash on Vertex. Sends `gs://{bucket}/{vid}.mp4` URIs directly. Vertex fetches the bytes server-side, so the local server never downloads the movie. Caching is per-movie (one cache create + N question hits per movie); the TTL keeps the bytes resident for the duration of the inner loop. Usage: bash run.sh # cache on (default) bash run.sh --no-use-cache # inline; re-uploads bytes per call bash run.sh --model gemini-3-pro-preview # different snapshot Requires `gcloud auth application-default login`, `GOOGLE_CLOUD_PROJECT` (or `--project`), and movies at `gs://{CHNB_GCS_BUCKET}/{vid}.mp4`. """ from __future__ import annotations import argparse import os import sys import warnings from pathlib import Path _PROJECT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(_PROJECT)) from google import genai from google.genai import types from google.genai.types import ( Content, CreateCachedContentConfig, GenerateContentConfig, Part, ) from util.backbones import FatalBackboneError, _detect_fatal_gemini from util.benchmark import load_benchmark from util.config import ( DEFAULT_MODELS, DEFAULT_VERTEX_LOCATION, GOOGLE_CLOUD_PROJECT_ENV, ) from util.eval_loop import run_sweep from util.parsers import parse_mcq_json from util.paths import ( BENCHMARK_PATH, GCS_BUCKET, RESULTS_DIR, ensure_dir, results_filename, ) from util.prompts import HEADER_VIDEO_CLIP, mcq_prompt_json EXPERIMENT_NAME = 'gemini3flash_video' SYSTEM_INSTRUCTION = ( 'You answer movie MCQs based on the cached movie video. Watch the video ' 'carefully, then answer the question. Output only valid JSON.' ) def main() -> None: ap = argparse.ArgumentParser() ap.add_argument('--benchmark', default=str(BENCHMARK_PATH)) ap.add_argument('--model', default=DEFAULT_MODELS['gemini']) ap.add_argument('--project', default=os.environ.get(GOOGLE_CLOUD_PROJECT_ENV, ''), help='GCP project id. REQUIRED — set via env or this flag.') ap.add_argument('--location', default=DEFAULT_VERTEX_LOCATION) ap.add_argument('--gcs-bucket', default=GCS_BUCKET, help=f'GCS bucket containing {{vid}}.mp4. Default: {GCS_BUCKET}.') ap.add_argument('--max-output-tokens', type=int, default=24576) ap.add_argument('--temperature', type=float, default=0.0) ap.add_argument('--use-cache', action=argparse.BooleanOptionalAction, default=True, help='Cache the video once per movie (default), then run all ' 'its questions against the cache. Pass --no-use-cache to ' 'inline the video each call (~40x cheaper with cache for ' 'a 60-question/movie benchmark).') ap.add_argument('--cache-ttl-sec', type=int, default=600, help='Cache TTL in seconds. Must comfortably exceed the time ' 'to ask all questions for one movie (~5-10 min typical).') ap.add_argument('--checkpoint-every', type=int, default=50) args = ap.parse_args() if not args.project: sys.exit(f'{GOOGLE_CLOUD_PROJECT_ENV} not set; export it or pass --project.') out_dir = ensure_dir(RESULTS_DIR / EXPERIMENT_NAME) df = load_benchmark(args.benchmark) print(f'Loaded {len(df)} questions across {df["video_file_id"].nunique()} movies') client = genai.Client(vertexai=True, project=args.project, location=args.location) print(f'Vertex Gemini ready: project={args.project}, location={args.location}, ' f'model={args.model}, use_cache={args.use_cache}') try: THINKING_HIGH = types.ThinkingConfig(thinking_level=types.ThinkingLevel.HIGH) except (AttributeError, TypeError): THINKING_HIGH = None warnings.warn( 'ThinkingConfig not available for this model or version of the Python SDK.' ) def gcs_uri(vid: str) -> str: return f'gs://{args.gcs_bucket}/{vid}.mp4' def build_movie_ctx(vid, movie_df): uri = gcs_uri(vid) if not args.use_cache: return {'uri': uri, 'cache_name': None} try: cache = client.caches.create( model=args.model, config=CreateCachedContentConfig( contents=[Content(role='user', parts=[ Part.from_uri(file_uri=uri, mime_type='video/mp4'), ])], system_instruction=SYSTEM_INSTRUCTION, ttl=f'{args.cache_ttl_sec}s', ), ) print(f' cached {vid}: {cache.name}') return {'uri': uri, 'cache_name': cache.name} except Exception as e: print(f' cache create failed for {vid}: {type(e).__name__}: {e!r}') return {'uri': uri, 'cache_name': None} def teardown_movie_ctx(ctx): cache_name = ctx.get('cache_name') if cache_name: try: client.caches.delete(name=cache_name) except Exception as e: print(f' cache delete failed: {type(e).__name__}: {e!r} ' f'(non-fatal; cache will expire via TTL)') def call_fn(row, ctx): prompt = mcq_prompt_json(row, header=HEADER_VIDEO_CLIP) cache_name = ctx.get('cache_name') config_kwargs = { 'temperature': args.temperature, 'max_output_tokens': args.max_output_tokens, 'response_mime_type': 'application/json', } if THINKING_HIGH is not None: config_kwargs['thinking_config'] = THINKING_HIGH try: if cache_name: config_kwargs['cached_content'] = cache_name resp = client.models.generate_content( model=args.model, contents=prompt, config=GenerateContentConfig(**config_kwargs), ) else: config_kwargs['system_instruction'] = SYSTEM_INSTRUCTION resp = client.models.generate_content( model=args.model, contents=[ Part.from_uri(file_uri=ctx['uri'], mime_type='video/mp4'), prompt, ], config=GenerateContentConfig(**config_kwargs), ) except Exception as e: fatal = _detect_fatal_gemini(e) if fatal: raise FatalBackboneError(f'gemini3flash_video({args.model}): {fatal}') from e raise text = (resp.text or '').strip() if not text: finish = (getattr(resp, 'candidates', [None])[0] or None) finish_reason = getattr(finish, 'finish_reason', None) if finish else None warnings.warn( f'gemini3flash_video({args.model}): empty text. ' f'finish_reason={finish_reason!r}. ' f'If MAX_TOKENS, raise --max-output-tokens.', UserWarning, stacklevel=2, ) return parse_mcq_json(text, qid=row['qid']) results_path = out_dir / results_filename(model_id=args.model) run_sweep( df, call_fn=call_fn, build_movie_ctx=build_movie_ctx, teardown_movie_ctx=teardown_movie_ctx, results_path=results_path, experiment_name=EXPERIMENT_NAME, extra_row_fields={ 'model': args.model, 'backbone': 'gemini', 'mode': 'video', 'thinking_level': 'HIGH', 'grounding': 'none', 'cached': bool(args.use_cache), }, checkpoint_every=args.checkpoint_every, ) if __name__ == '__main__': main()