Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ dependencies = [
"PyYAML>=6.0.0",
"requests>=2.31.0",
"Pillow>=10.0.0",
"google-genai>=1.0.0",
"google-genai>=2.21.0",
"google-generativeai>=0.3.0",
"opentelemetry-distro>=0.40b0",
"opentelemetry-exporter-otlp>=1.20.0",
Expand Down
2 changes: 1 addition & 1 deletion requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ psutil>=5.9.0
cachetools>=5.0.0

# Google AI & Cloud
google-genai>=1.0.0
google-genai>=2.21.0
google-generativeai>=0.3.0
google-cloud-aiplatform>=1.133.0
google-cloud-vision>=3.7.0
Expand Down
149 changes: 149 additions & 0 deletions src/integration/gemini_agentic_video.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""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
from urllib.parse import urlsplit

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 GeminiAgenticVideoService._is_supported_uri(video.uri):
raise ValueError("Video URI must be a supported YouTube or file URI")
item = {
"type": "video",
"uri": video.uri,
Comment on lines +75 to +80

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@copilot Fix the code for all comments in this review comment.

When a review comment includes a suggested change, apply the suggestion exactly.

Do not make changes beyond what is described in the linked review comment.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in commit 18c876a. URI parsing now rejects malformed or unsupported references before request construction, and regression coverage includes malformed URIs with a nonblank prompt.

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@copilot Fix the code for all comments in this review comment.

When a review comment includes a suggested change, apply the suggestion exactly.

Do not make changes beyond what is described in the linked review comment.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Implemented in commit d021557: URI validation remains covered alongside the refreshed SDK 2.21.0 adapter test; targeted tests pass.

"processing": video.processing,
}
if video.mime_type:
item["mime_type"] = video.mime_type
items.append(item)
items.append({"type": "text", "text": prompt})
return items

@staticmethod
def _is_supported_uri(uri: str) -> bool:
if not uri.strip() or uri != uri.strip():
return False
try:
parsed = urlsplit(uri)
hostname = (parsed.hostname or "").lower()
except ValueError:
return False

if parsed.scheme == "https":
return (
bool(parsed.path)
and (
hostname == "youtu.be"
or hostname == "youtube.com"
or hostname.endswith(".youtube.com")
)
)
if parsed.scheme == "gs":
return bool(parsed.netloc and parsed.path)
if parsed.scheme == "file":
return bool(parsed.path)
return False

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=self._response_text(response),
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),
)

@staticmethod
def _response_text(response: Any) -> str:
output_text = getattr(response, "output_text", None)
if output_text is not None:
return str(output_text)
return "".join(
str(getattr(output, "text", ""))
for output in (getattr(response, "outputs", None) or [])
if getattr(output, "text", None) is not None
)
93 changes: 93 additions & 0 deletions tests/unit/test_gemini_agentic_video.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
from types import SimpleNamespace

import pytest
from google.genai._gaos.types.interactions.interaction import Interaction
from google.genai._gaos.types.interactions.modeloutputstep import ModelOutputStep
from google.genai._gaos.types.interactions.textcontent import TextContent
from google.genai._gaos.types.interactions.usage import Usage

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 Interaction(
id="interaction-123",
created="2026-09-09T20:00:00Z",
status="completed",
updated="2026-09-09T20:00:01Z",
steps=[
ModelOutputStep(content=[TextContent(text="grounded result")])
],
usage=Usage(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(
"uri",
["x", "https://example.com/video.mp4", "ftp://youtu.be/auJzb1D-fag"],
)
def test_malformed_video_uri_fails_before_calling_provider(uri):
with pytest.raises(ValueError):
GeminiAgenticVideoService.build_input([VideoInput(uri)], "question")


@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)
8 changes: 4 additions & 4 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading