diff --git a/src/integration/gemini_agentic_video.py b/src/integration/gemini_agentic_video.py new file mode 100644 index 000000000..277fb0005 --- /dev/null +++ b/src/integration/gemini_agentic_video.py @@ -0,0 +1,113 @@ +"""Gemini agentic video-understanding adapter. + +This module isolates the Gemini Interactions API from EventRelay's existing +``generateContent`` integration. It enables targeted, server-side inspection +of transcripts, frames, and audio without changing the production path until +benchmark evidence supports promotion. +""" + +from __future__ import annotations + +import asyncio +import os +from dataclasses import dataclass +from typing import Any, Literal, Sequence + +ProcessingMode = Literal["agentic", "static"] + + +@dataclass(frozen=True) +class VideoInput: + """One video reference and its independently selected processing mode.""" + + uri: str + processing: ProcessingMode = "agentic" + mime_type: str | None = None + + +@dataclass(frozen=True) +class AgenticVideoReceipt: + """Stable execution receipt retained by EventRelay after analysis.""" + + output_text: str + total_tokens: int | None + model: str + sources: tuple[str, ...] + processing_modes: tuple[ProcessingMode, ...] + + +class GeminiAgenticVideoService: + """Run Gemini's Think -> Act -> Observe video-analysis loop.""" + + DEFAULT_MODEL = "gemini-3.7-flash" + + def __init__( + self, + api_key: str | None = None, + *, + client: Any | None = None, + model: str | None = None, + ) -> None: + self.model = model or os.getenv( + "GEMINI_AGENTIC_VIDEO_MODEL", self.DEFAULT_MODEL + ) + if client is not None: + self._client = client + return + + from google import genai + + resolved_key = api_key or os.getenv("GEMINI_API_KEY") + self._client = ( + genai.Client(api_key=resolved_key) if resolved_key else genai.Client() + ) + + @staticmethod + def build_input(videos: Sequence[VideoInput], prompt: str) -> list[dict[str, str]]: + """Build the documented Interactions API input without materializing media.""" + if not videos: + raise ValueError("At least one video reference is required") + if not prompt.strip(): + raise ValueError("A non-empty analysis prompt is required") + + items: list[dict[str, str]] = [] + for video in videos: + if not video.uri.strip(): + raise ValueError("Video URI must not be empty") + item = { + "type": "video", + "uri": video.uri, + "processing": video.processing, + } + if video.mime_type: + item["mime_type"] = video.mime_type + items.append(item) + items.append({"type": "text", "text": prompt}) + return items + + async def analyze( + self, + videos: Sequence[VideoInput], + prompt: str, + *, + model: str | None = None, + ) -> AgenticVideoReceipt: + """Analyze referenced media and return a durable, comparable receipt.""" + selected_model = model or self.model + request_input = self.build_input(videos, prompt) + response = await asyncio.to_thread( + self._client.interactions.create, + model=selected_model, + input=request_input, + ) + + usage = getattr(response, "usage", None) + total_tokens = getattr(usage, "total_tokens", None) + return AgenticVideoReceipt( + output_text=str(getattr(response, "output_text", "")), + total_tokens=int(total_tokens) if total_tokens is not None else None, + model=selected_model, + sources=tuple(video.uri for video in videos), + processing_modes=tuple(video.processing for video in videos), + ) + diff --git a/tests/unit/test_gemini_agentic_video.py b/tests/unit/test_gemini_agentic_video.py new file mode 100644 index 000000000..3859366ac --- /dev/null +++ b/tests/unit/test_gemini_agentic_video.py @@ -0,0 +1,74 @@ +from types import SimpleNamespace + +import pytest + +from src.integration.gemini_agentic_video import ( + GeminiAgenticVideoService, + VideoInput, +) + + +class FakeInteractions: + def __init__(self) -> None: + self.request = None + + def create(self, **kwargs): + self.request = kwargs + return SimpleNamespace( + output_text="grounded result", + usage=SimpleNamespace(total_tokens=321), + ) + + +@pytest.mark.asyncio +async def test_agentic_youtube_request_returns_execution_receipt(): + interactions = FakeInteractions() + client = SimpleNamespace(interactions=interactions) + service = GeminiAgenticVideoService(client=client) + + receipt = await service.analyze( + [VideoInput("https://youtu.be/auJzb1D-fag")], + "Find the implementation steps and their timestamps.", + ) + + assert interactions.request == { + "model": "gemini-3.7-flash", + "input": [ + { + "type": "video", + "uri": "https://youtu.be/auJzb1D-fag", + "processing": "agentic", + }, + { + "type": "text", + "text": "Find the implementation steps and their timestamps.", + }, + ], + } + assert receipt.output_text == "grounded result" + assert receipt.total_tokens == 321 + assert receipt.sources == ("https://youtu.be/auJzb1D-fag",) + assert receipt.processing_modes == ("agentic",) + + +def test_mixed_mode_keeps_each_video_processing_policy(): + request_input = GeminiAgenticVideoService.build_input( + [ + VideoInput("gs://bucket/reference.mp4", "agentic", "video/mp4"), + VideoInput("gs://bucket/clip.mp4", "static", "video/mp4"), + ], + "Locate the clip in the reference recording.", + ) + + assert request_input[0]["processing"] == "agentic" + assert request_input[1]["processing"] == "static" + assert request_input[0]["mime_type"] == "video/mp4" + + +@pytest.mark.parametrize( + ("videos", "prompt"), + [([], "question"), ([VideoInput("")], "question"), ([VideoInput("x")], " ")], +) +def test_invalid_requests_fail_before_calling_provider(videos, prompt): + with pytest.raises(ValueError): + GeminiAgenticVideoService.build_input(videos, prompt)