chenmingxuan/Artemis-Eapil
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

merge(video): integrate video analyzer module split

Browse Source
This commit is contained in:
somew1nd committed 2026-09-01 21:46:40 -07:00
2 parents dc619679ea 7cc20bbddc
commit 582f112287
8 files changed
+3200 -2355
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

artemis/agents/video_analyzer/audio_native.py
+610
View File
@@ -0,0 +1,610 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Native Gemini execution path for audio-only segment analysis.
Extracted from ``video_analyzer.py`` as a pure structural split. Patchable
collaborators are looked up late through the facade module so
``mock.patch("artemis.agents.video_analyzer.video_analyzer.<name>")`` targets
keep working.
"""
import asyncio
from pathlib import Path
from google.genai import types
from artemis.agents.video_analyzer import video_analyzer as _va
from artemis.agents.video_analyzer.reliability import classify_video_failure
from artemis.constants import SAFETY_SETTINGS_BLOCK_NONE
from artemis.data_engine.trace import TraceSpan
from artemis.llm.reliability import retry_policy_for
from artemis.utils.logger import get_logger
logger = get_logger(__name__)
class _AudioMedia:
"""Mutable holder for per-attempt audio artifacts shared with failure handling."""
__slots__ = ("audio_path", "actual_start", "prompt_with_context")
def __init__(self):
self.audio_path: Path | None = None
self.actual_start: float | None = None
self.prompt_with_context: str | None = None
async def exec_analyze_audio_only(
analyzer, start_time: float, end_time: float | None, specific_query: str
) -> str:
"""Analyzes only the audio track of a segment, with claim/lease and retries."""
if not isinstance(start_time, (int, float)):
return "Error: start_time must be a numeric recording-relative timestamp."
if not str(specific_query or "").strip():
return "Error: specific_query must describe the audio evidence to find."
max_retries = analyzer.native_max_retries
current_start = round(start_time, 1) if isinstance(start_time, (int, float)) else start_time
current_end = round(end_time, 1) if isinstance(end_time, (int, float)) else end_time
if current_end is None:
controller = _va.get_controller(analyzer.ctx)
async with _va.TRANSCODE_SEMAPHORE:
metadata = await controller.extract_segment_metadata(current_start, None)
duration = getattr(metadata, "duration_seconds", None)
actual_start = getattr(metadata, "actual_start_relative_time", current_start)
if not metadata.success or not isinstance(duration, (int, float)) or duration <= 0:
return f"Error fetching audio segment metadata: {metadata.message}"
current_start = round(float(actual_start), 1)
current_end = round(current_start + float(duration), 1)
if metadata.video_path:
metadata_path = Path(metadata.video_path)
analyzer.local_files_to_cleanup.add(metadata_path)
analyzer.local_dirs_to_cleanup.add(metadata_path.parent)
if current_end <= current_start:
return "Error: end_time must be greater than start_time."
device_id = getattr(getattr(analyzer.ctx, "device", None), "device_id", None)
active_session = _va.get_active_session(device_id) if device_id else None
claim = analyzer.blackboard.claim_segment(
current_start,
current_end,
specific_query,
modality="audio",
model_name=analyzer.model_name,
source_generation=(
active_session.generation if active_session is not None else None
),
)
if claim.state == "cached":
result = (
f"[from {current_start:.1f}s to {current_end:.1f}s] Audio Summary: "
f"{claim.summary or 'No audio summary provided.'}"
)
if claim.analysis:
result += f" Analysis: {claim.analysis}"
return "CACHED AUDIO ANALYSIS: " + result
if claim.state == "in_progress":
return (
f"[from {current_start:.1f}s to {current_end:.1f}s] "
"Audio analysis is already in progress; do not duplicate it."
)
lease_owner = claim.lease_owner
if lease_owner is None:
raise RuntimeError("Video blackboard returned an audio claim without a lease owner")
media = _AudioMedia()
for attempt in range(max_retries + 1):
try:
logger.info(
f"Attempt {attempt + 1}/{max_retries + 1} for"
f" analyze_audio_only: start={current_start},"
f" end={current_end}"
)
result = await _prepare_audio_media(analyzer, media, current_start, current_end)
media.prompt_with_context = _build_audio_prompt(
analyzer, result, media.actual_start, start_time, end_time, specific_query
)
if not analyzer.use_native_gemini:
audio_result = await analyzer._exec_analyze_audio_universal(
audio_path=media.audio_path,
start_time=current_start,
end_time=current_end,
actual_start=media.actual_start,
prompt_with_context=media.prompt_with_context,
specific_query=specific_query,
)
summary, analysis = analyzer._parse_chunk_result(audio_result)
analyzer.blackboard.complete_segment(
current_start,
current_end,
specific_query,
lease_owner,
summary,
analysis,
modality="audio",
)
return audio_result
return await _run_native_audio_conversation(
analyzer,
media,
current_start,
current_end,
start_time,
end_time,
specific_query,
lease_owner,
)
except Exception as e:
# Same contract as the video sub-agent loop: classification
# owns the retry decision and every retry restarts the attempt
# from scratch.
failure = classify_video_failure(e)
attempts_so_far = attempt + 1
if not failure.retryable:
logger.warning(
f"Audio non-retryable failure [{failure.category.value}]"
f" on attempt {attempts_so_far}: {e}"
)
return await _finish_failed_audio(
analyzer,
media,
current_start,
current_end,
specific_query,
lease_owner,
e,
retryable=False,
)
policy = retry_policy_for(failure.category)
allowed_attempts = min(max_retries + 1, policy.max_attempts)
if attempts_so_far >= allowed_attempts:
logger.warning(
f"Audio analysis exhausted {attempts_so_far} attempt(s)"
f" [{failure.category.value}]: {e}"
)
return await _finish_failed_audio(
analyzer,
media,
current_start,
current_end,
specific_query,
lease_owner,
e,
)
delay = policy.delay_for(attempts_so_far)
logger.warning(
f"Audio retryable failure [{failure.category.value}]"
f" ({attempts_so_far}/{allowed_attempts}): {e}; retrying"
f" from scratch in {delay:.2f}s"
)
_va._record_llm_retry(
str(e),
delay,
attempt=attempts_so_far,
max_retries=allowed_attempts,
source="video_audio_agent",
)
await asyncio.sleep(delay)
async def _finish_failed_audio(
analyzer,
media: _AudioMedia,
current_start: float,
current_end: float,
specific_query: str,
lease_owner: str,
error: Exception,
*,
retryable: bool | None = None,
):
"""Records terminal audio failure, attempting the universal fallback first."""
failure = classify_video_failure(error)
if (
analyzer.use_native_gemini
and failure.should_fallback
and media.audio_path is not None
and media.actual_start is not None
and media.prompt_with_context is not None
):
try:
logger.warning(
"Native audio analysis failed"
f" [{failure.category.value}]; using universal fallback"
)
_va._record_llm_event(
"llm_fallback",
{
"reason": "native_audio_analysis_failed",
"category": failure.category.value,
"error": str(error)[:500],
},
)
fallback_result = await analyzer._exec_analyze_audio_universal(
audio_path=media.audio_path,
start_time=current_start,
end_time=current_end,
actual_start=media.actual_start,
prompt_with_context=media.prompt_with_context,
specific_query=specific_query,
force_fallback=True,
)
summary, analysis = analyzer._parse_chunk_result(fallback_result)
analyzer.blackboard.complete_segment(
current_start,
current_end,
specific_query,
lease_owner,
summary,
analysis,
modality="audio",
)
return fallback_result
except Exception as fallback_error:
error = fallback_error
failure = classify_video_failure(error)
retryable = None
analyzer.blackboard.fail_segment(
current_start,
current_end,
specific_query,
lease_owner,
str(error),
modality="audio",
retryable=failure.retryable if retryable is None else retryable,
error_category=failure.category.value,
)
raise error
async def _prepare_audio_media(
analyzer, media: _AudioMedia, current_start: float, current_end: float
):
"""Extracts the segment and its audio track, tracking artifacts for cleanup."""
controller = _va.get_controller(analyzer.ctx)
async with _va.TRANSCODE_SEMAPHORE:
result = await controller.extract_segment_metadata(current_start, current_end)
if not result.success or not result.video_path:
raise Exception(f"Failed to get video segment: {result.message}")
path = Path(result.video_path)
analyzer.local_files_to_cleanup.add(path)
analyzer.local_dirs_to_cleanup.add(path.parent)
async with _va.TRANSCODE_SEMAPHORE:
audio_path = await _va.extract_audio_from_video(path)
analyzer.local_files_to_cleanup.add(audio_path)
analyzer.local_dirs_to_cleanup.add(audio_path.parent)
media.audio_path = audio_path
actual_start = getattr(result, "actual_start_relative_time", current_start)
if not isinstance(actual_start, (int, float)):
actual_start = current_start
media.actual_start = actual_start
return result
def _build_audio_prompt(
analyzer,
result,
actual_start: float,
start_time: float,
end_time: float | None,
specific_query: str,
) -> str:
"""Assembles the audio sub-agent prompt with timing context, ledger, and warnings."""
warnings = analyzer.get_overlapping_warnings(start_time, end_time)
warning_block = ""
if warnings:
lines = [
"WARNING: The following queries were already tried in this timeframe:\n"
]
for w in warnings:
lines.append(f"Searched for: {w['target']} -> {w['summary']}")
warning_block = "\n".join(lines) + "\n\n"
duration_secs = getattr(result, "duration_seconds", None)
actual_end = (
actual_start + duration_secs
if isinstance(duration_secs, (int, float))
else None
)
end_str = f" to {actual_end:.1f}s" if actual_end is not None else ""
warning_val = getattr(result, "warning", None)
truncation_note = (
f" (NOTE: {warning_val})"
if isinstance(warning_val, str) and warning_val
else ""
)
prompt_with_context = (
"IMPORTANT CONTEXT: This audio segment corresponds to"
" the test's relative time from"
f" {actual_start:.1f}s{end_str}{truncation_note}. When"
" you report timestamps, remember that your 00:00 is"
f" exactly {actual_start:.1f}s in the test relative"
" time. Please calculate and output all timestamps as"
" the exact test relative time (e.g., if you hear"
" something at 00:05, report"
f" {actual_start + 5:.1f}s).\n\n"
)
if analyzer.enable_ledger:
prompt_with_context += (
"BLACKBOARD"
f" LEDGER:\n{analyzer.blackboard_ledger}\nDirective:"
" Use the Blackboard to understand what happened"
" outside your assigned timeframe. Do not"
" duplicate findings already on the board.\n\n"
)
if warning_block:
prompt_with_context += (
f"{warning_block}Directive: Do not repeat the"
" failed searches listed above. Shift your"
" attention to the current task.\n\n"
)
prompt_with_context += f"{specific_query}"
return prompt_with_context
async def _run_native_audio_conversation(
analyzer,
media: _AudioMedia,
current_start: float,
current_end: float,
start_time: float,
end_time: float | None,
specific_query: str,
lease_owner: str,
) -> str:
"""Uploads the audio and drives the native Gemini audio agent to a committed answer."""
file = None
try:
with TraceSpan(name="upload_audio_to_gemini") as span:
file = await analyzer.upload_and_poll_file(media.audio_path)
span.result = f"Uploaded {file.name}"
logger.info(
f"Invoking Gemini for audio-only task with model {analyzer.model_name}..."
)
sub_agent_contents = [
types.Content(
role="user",
parts=[
types.Part.from_uri(file_uri=file.uri, mime_type=file.mime_type)
if hasattr(file, "uri")
else file,
types.Part.from_text(text=media.prompt_with_context),
],
)
]
(
final_summary,
final_analysis,
final_confidence_score,
) = await _drive_audio_agent_loop(analyzer, sub_agent_contents)
end_val = end_time if end_time is not None else "unknown"
analyzer._record_blackboard_entry(
{
"start": start_time,
"end": end_val,
"target": specific_query,
"summary": final_summary,
"confidence_score": final_confidence_score,
},
modality="audio",
)
end_str = f"{end_time:.1f}s" if end_time is not None else "unknown"
audio_result = (
f"[from {start_time:.1f}s to {end_str}] Summary:"
f" {final_summary} Analysis: {final_analysis}"
)
analyzer.blackboard.complete_segment(
current_start,
current_end,
specific_query,
lease_owner,
final_summary,
final_analysis,
modality="audio",
)
return audio_result
finally:
if file:
logger.info(f"Cleaning up cloud file {file.name}...")
try:
await asyncio.wait_for(
analyzer.client.aio.files.delete(name=file.name),
timeout=30,
)
analyzer.cloud_files_to_cleanup.discard(file.name)
except Exception as ce:
logger.error(f"Failed to delete cloud file {file.name}: {ce}")
async def _drive_audio_agent_loop(
analyzer, sub_agent_contents: list
) -> tuple[str, str, float]:
"""Iterates the audio agent conversation until submit_answer is accepted."""
sub_max_iterations = 2
sub_iterations = 0
final_confidence_score = 0.0
final_summary = "No summary provided."
final_analysis = "No analysis provided."
final_full_text = ""
while sub_iterations < sub_max_iterations:
sub_iterations += 1
if sub_iterations == sub_max_iterations - 1:
sub_agent_contents.append(
types.Content(
role="user",
parts=[
types.Part.from_text(
text=(
"[WARNING] You must call the"
" submit_answer tool now, time"
" is running out."
)
)
],
)
)
with TraceSpan(name="gemini_generate_audio_content") as span:
response = await asyncio.wait_for(
analyzer.client.aio.models.generate_content(
model=analyzer.model_name,
contents=sub_agent_contents,
config=types.GenerateContentConfig(
system_instruction=analyzer.audio_system_prompt,
tools=[
types.Tool(
function_declarations=[analyzer.submit_answer_tool]
)
],
safety_settings=SAFETY_SETTINGS_BLOCK_NONE,
),
),
timeout=analyzer.model_call_timeout_seconds,
)
full_text = response.text or ""
final_full_text += full_text + "\n"
function_calls = response.function_calls or []
if function_calls:
(
answered,
final_summary,
final_analysis,
final_confidence_score,
) = _handle_audio_function_calls(
sub_agent_contents,
response,
function_calls,
final_summary,
final_analysis,
final_confidence_score,
)
if answered is True:
break
else:
sub_agent_contents.append(
types.Content(
role="model",
parts=[types.Part.from_text(text=full_text)],
)
)
return final_summary, final_analysis, final_confidence_score
def _handle_audio_function_calls(
sub_agent_contents: list,
response,
function_calls: list,
final_summary: str,
final_analysis: str,
final_confidence_score: float,
):
"""Processes audio-agent function calls; returns (answered, summary, analysis, score)."""
answered = False
for fc in function_calls:
if (
fc.name.split(":")[-1] if ":" in fc.name else fc.name
) == "submit_answer":
args = fc.args
if (
"confidence_score" not in args
or not isinstance(
args["confidence_score"],
(int, float),
)
or not (0.0 <= float(args["confidence_score"]) <= 1.0)
):
error_msg = (
"Error: 'confidence_score' is"
" missing or out of bounds. It must"
" be a float between 0.0 and 1.0"
" inclusive. Fix the payload and"
" call submit_answer again."
)
sub_agent_contents.append(
types.Content(
role="model",
parts=response.candidates[0].content.parts,
)
)
sub_agent_contents.append(
types.Content(
role="user",
parts=[
types.Part.from_function_response(
name=fc.name,
response={"error": error_msg},
)
],
)
)
answered = "error"
break
final_summary = (
args.get("summary", "No summary provided.")
.strip()
.replace("\n", " ")
)
final_analysis = (
args.get("analysis", "No analysis provided.")
.strip()
.replace("\n", " ")
)
final_confidence_score = float(args["confidence_score"])
answered = True
break
if answered is True or answered == "error":
return answered, final_summary, final_analysis, final_confidence_score
sub_agent_contents.append(
types.Content(
role="model",
parts=response.candidates[0].content.parts,
)
)
sub_agent_contents.append(
types.Content(
role="user",
parts=[
types.Part.from_function_response(
name=function_calls[0].name,
response={
"error": (
"Tool not recognized."
" Please use"
" submit_answer."
)
},
)
],
)
)
return answered, final_summary, final_analysis, final_confidence_score
artemis/agents/video_analyzer/chunk_conversation.py
+534
View File
@@ -0,0 +1,534 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Native Gemini sub-agent conversation and submit_answer handling for chunks.
Extracted from ``video_analyzer.py`` as a pure structural split. Patchable
collaborators are looked up late through the facade module so
``mock.patch("artemis.agents.video_analyzer.video_analyzer.<name>")`` targets
keep working.
"""
import asyncio
import glob
from pathlib import Path
import re
from google.genai import types
from artemis.agents.video_analyzer import video_analyzer as _va
from artemis.constants import SAFETY_SETTINGS_BLOCK_NONE
from artemis.data_engine.trace import CURRENT_TRACE_ID, TraceSpan
from artemis.utils.logger import get_logger
logger = get_logger(__name__)
class _ChunkMedia:
"""Mutable holder for per-attempt media artifacts shared with failure handling."""
__slots__ = ("path", "compressed_path", "actual_start", "prompt_with_context")
def __init__(self):
self.path: Path | None = None
self.compressed_path: Path | None = None
self.actual_start: float | None = None
self.prompt_with_context: str | None = None
async def _run_native_chunk_conversation(
analyzer,
media: _ChunkMedia,
current_start: float,
current_end: float,
start_time: float,
end_time: float | None,
specific_query: str,
lease_owner: str,
) -> str:
"""Uploads the chunk and drives the native Gemini sub-agent to a committed answer."""
file = None
try:
with TraceSpan(name="upload_video_to_gemini") as span:
file = await analyzer.upload_and_poll_file(media.compressed_path)
span.result = f"Uploaded {file.name}"
logger.info(
"Invoking Gemini for sub-agent task with model"
f" {analyzer.model_name} (streaming)..."
)
trace_id = CURRENT_TRACE_ID.get()
if analyzer.ctx.data_engine and trace_id:
analyzer.ctx.data_engine.record_trace(
type="llm_call",
name="video_sub_agent",
payload={
"contents": [
f"file://{file.name}",
media.prompt_with_context,
],
"system_instruction": analyzer.sub_system_prompt,
},
parent_trace_id=trace_id,
)
sub_agent_contents = [
types.Content(
role="user",
parts=[
types.Part.from_uri(file_uri=file.uri, mime_type=file.mime_type)
if hasattr(file, "uri")
else file,
types.Part.from_text(text=media.prompt_with_context),
],
)
]
final_summary, final_analysis = await _drive_sub_agent_loop(
analyzer,
sub_agent_contents,
trace_id,
media,
start_time,
end_time,
specific_query,
)
end_str = f"{current_end:.1f}s" if current_end is not None else "unknown"
chunk_result = (
f"[from {current_start:.1f}s to {end_str}] Summary:"
f" {final_summary} Analysis: {final_analysis}"
)
analyzer.blackboard.complete_segment(
current_start,
current_end,
specific_query,
lease_owner,
final_summary,
final_analysis,
)
return chunk_result
finally:
if file:
logger.info(f"Cleaning up cloud file {file.name}...")
try:
await asyncio.wait_for(
analyzer.client.aio.files.delete(name=file.name),
timeout=30,
)
analyzer.cloud_files_to_cleanup.discard(file.name)
except Exception as ce:
logger.error(f"Failed to delete cloud file {file.name}: {ce}")
async def _drive_sub_agent_loop(
analyzer,
sub_agent_contents: list,
trace_id,
media: _ChunkMedia,
start_time: float,
end_time: float | None,
specific_query: str,
) -> tuple[str, str]:
"""Iterates the sub-agent conversation until submit_answer is accepted."""
sub_max_iterations = 2
sub_iterations = 0
final_analysis = "No analysis provided."
final_summary = "No summary provided."
final_full_text = ""
while sub_iterations < sub_max_iterations:
sub_iterations += 1
if sub_iterations == sub_max_iterations - 1:
sub_agent_contents.append(
types.Content(
role="user",
parts=[
types.Part.from_text(
text=(
"[WARNING] You must call the"
" submit_answer tool now, time"
" is running out."
)
)
],
)
)
full_text, function_calls, original_parts = await _stream_sub_agent_turn(
analyzer, sub_agent_contents, trace_id
)
final_full_text += full_text + "\n"
if function_calls:
answered, final_summary, final_analysis = await _handle_sub_agent_calls(
analyzer,
sub_agent_contents,
function_calls,
original_parts,
full_text,
media,
start_time,
end_time,
specific_query,
final_summary,
final_analysis,
)
if answered is True:
break
else:
sub_agent_contents.append(
types.Content(
role="model",
parts=[types.Part.from_text(text=full_text)],
)
)
return final_summary, final_analysis
async def _stream_sub_agent_turn(
analyzer, sub_agent_contents: list, trace_id
) -> tuple[str, list, list]:
"""Starts one streamed Gemini turn and consumes it fully."""
with TraceSpan(name="gemini_stream_content_main") as span:
stream = await asyncio.wait_for(
analyzer.client.aio.models.generate_content_stream(
model=analyzer.model_name,
contents=sub_agent_contents,
config=types.GenerateContentConfig(
system_instruction=analyzer.sub_system_prompt,
tools=[
types.Tool(
function_declarations=[analyzer.submit_answer_tool]
)
],
safety_settings=SAFETY_SETTINGS_BLOCK_NONE,
),
),
timeout=60,
)
full_text, function_calls, original_parts = await asyncio.wait_for(
_read_sub_agent_stream(analyzer, span, stream, trace_id),
timeout=analyzer.model_call_timeout_seconds,
)
return full_text, function_calls, original_parts
async def _read_sub_agent_stream(analyzer, span, target_stream, current_trace_id):
"""Consumes a streamed response, mirroring text/thoughts into the trace."""
text = ""
function_calls = []
accumulated_parts = []
async for chunk in target_stream:
if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
span.payload["usage_metadata"] = {
"prompt_token_count": getattr(
chunk.usage_metadata,
"prompt_token_count",
0,
),
"candidates_token_count": getattr(
chunk.usage_metadata,
"candidates_token_count",
0,
),
"total_token_count": getattr(
chunk.usage_metadata,
"total_token_count",
0,
),
}
chunk_text = chunk.text or ""
if chunk_text:
text += chunk_text
if analyzer.ctx.data_engine and current_trace_id:
analyzer.ctx.data_engine.stream_output(
current_trace_id, chunk_text
)
if chunk.function_calls:
function_calls.extend(chunk.function_calls)
candidates = getattr(chunk, "candidates", None)
if (
candidates
and isinstance(candidates, list)
and len(candidates) > 0
):
content = getattr(candidates[0], "content", None)
parts = getattr(content, "parts", None) if content else None
if parts:
for part in parts:
accumulated_parts.append(part)
if getattr(part, "thought", False) and part.text:
if analyzer.ctx.data_engine and current_trace_id:
analyzer.ctx.data_engine.stream_output(
current_trace_id,
part.text,
is_thinking=True,
)
else:
if chunk.function_calls:
for fc in chunk.function_calls:
accumulated_parts.append(types.Part(function_call=fc))
return text, function_calls, accumulated_parts
async def _handle_sub_agent_calls(
analyzer,
sub_agent_contents: list,
function_calls: list,
original_parts: list,
full_text: str,
media: _ChunkMedia,
start_time: float,
end_time: float | None,
specific_query: str,
final_summary: str,
final_analysis: str,
):
"""Processes function calls from one turn; returns (answered, summary, analysis)."""
answered = False
for fc in function_calls:
if (
fc.name.split(":")[-1] if ":" in fc.name else fc.name
) == "submit_answer":
args = fc.args
timeline_events = args.get("timeline_events", [])
final_summary = (
args.get("summary", "No summary provided.")
.strip()
.replace("\n", " ")
)
final_analysis = (
args.get("analysis", "No analysis provided.")
.strip()
.replace("\n", " ")
)
error_msg = _validate_timeline_events(timeline_events, start_time, end_time)
if error_msg is not None:
_append_rejection(
sub_agent_contents,
original_parts,
function_calls,
full_text,
fc.name,
error_msg,
)
answered = "error"
break
await _persist_timeline_events(
analyzer, timeline_events, media, start_time, end_time, specific_query
)
answered = True
break
if answered is True or answered == "error":
return answered, final_summary, final_analysis
_append_rejection(
sub_agent_contents,
original_parts,
function_calls,
full_text,
function_calls[0].name,
(
"Tool not recognized."
" Please use"
" submit_answer."
),
)
return answered, final_summary, final_analysis
def _validate_timeline_events(
timeline_events: list, start_time: float, end_time: float | None
) -> str | None:
"""Returns the first validation error message for the submitted events, if any."""
for event in timeline_events:
if (
"confidence_score" not in event
or not isinstance(
event["confidence_score"],
(int, float),
)
or isinstance(event["confidence_score"], bool)
or not (0.0 <= float(event["confidence_score"]) <= 1.0)
):
return (
"Error: 'confidence_score' is"
" missing or out of bounds in"
" one of the timeline_events."
" It must be a float between"
" 0.0 and 1.0 inclusive. Fix"
" the payload and call"
" submit_answer again."
)
if (
"verification_timestamp_secs" not in event
or not isinstance(
event["verification_timestamp_secs"],
(int, float),
)
or isinstance(
event["verification_timestamp_secs"], bool
)
):
return (
"Error:"
" 'verification_timestamp_secs'"
" is missing or invalid in one"
" of the timeline_events. Fix"
" the payload and call"
" submit_answer again."
)
verification_ts = float(
event.get(
"verification_timestamp_secs",
start_time,
)
)
confidence = float(event.get("confidence_score", 0.0)) # noqa: F841
if verification_ts < start_time or (
end_time is not None and verification_ts > end_time
):
return (
"Error:"
" 'verification_timestamp_secs'"
f" {verification_ts} must fall"
" within the start_time"
f" ({start_time}) and end_time"
f" ({end_time}) boundaries. Fix"
" the payload and call"
" submit_answer again."
)
return None
def _append_rejection(
sub_agent_contents: list,
original_parts: list,
function_calls: list,
full_text: str,
response_name: str,
error_msg: str,
) -> None:
"""Appends the model turn plus an error function response to the conversation."""
sub_agent_contents.append(
types.Content(
role="model",
parts=original_parts
if original_parts
else [
types.Part(function_call=fc)
for fc in function_calls
]
+ (
[types.Part.from_text(text=full_text)]
if full_text
else []
),
)
)
sub_agent_contents.append(
types.Content(
role="user",
parts=[
types.Part.from_function_response(
name=response_name,
response={"error": (error_msg)},
)
],
)
)
async def _persist_timeline_events(
analyzer,
timeline_events: list,
media: _ChunkMedia,
start_time: float,
end_time: float | None,
specific_query: str,
) -> None:
"""Captures verification screenshots and records validated events on the blackboard."""
for event in timeline_events:
verification_ts = float(
event.get(
"verification_timestamp_secs",
start_time,
)
)
confidence = float(event.get("confidence_score", 0.0))
relative_offset = max(0.0, verification_ts - media.actual_start)
out_dir = _va.get_temp_dir("video_analyzer")
out_dir.mkdir(parents=True, exist_ok=True)
existing_imgs = glob.glob(f"{out_dir}/img_*.jpg")
max_idx = -1
for img in existing_imgs:
match = re.search(r"img_(\d+)\.jpg$", img)
if match:
max_idx = max(max_idx, int(match.group(1)))
next_idx = max_idx + 1
screenshot_output_path = out_dir / f"img_{next_idx}.jpg"
while screenshot_output_path.exists():
next_idx += 1
screenshot_output_path = out_dir / f"img_{next_idx}.jpg"
screenshot_output_path.touch()
analyzer.local_files_to_cleanup.add(screenshot_output_path)
ffmpeg_cmd = [
"ffmpeg",
"-ss",
str(relative_offset),
"-i",
str(media.path),
"-vframes",
"1",
"-q:v",
"2",
"-y",
str(screenshot_output_path),
]
try:
proc = await asyncio.create_subprocess_exec(
*ffmpeg_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
await proc.communicate()
except Exception as e:
logger.warning(f"Failed to extract frame: {e}")
screenshot_path_str = str(screenshot_output_path)
entry = {
"start": event.get("start_time", start_time),
"end": event.get("end_time", end_time),
"target": specific_query,
"summary": event.get("transcription", ""),
"confidence_score": confidence,
}
if Path(screenshot_output_path).exists():
entry["screenshot"] = str(screenshot_path_str)
analyzer._record_blackboard_entry(entry)
artemis/agents/video_analyzer/chunk_native.py
+376
View File
@@ -0,0 +1,376 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Native Gemini execution path for a single video chunk sub-agent.
Extracted from ``video_analyzer.py`` as a pure structural split. Patchable
collaborators (``get_controller``, ``compress_video_for_api``, semaphores,
LLM event recorders, ...) are looked up late through the facade module so
``mock.patch("artemis.agents.video_analyzer.video_analyzer.<name>")`` targets
keep working. The sub-agent conversation itself lives in
``chunk_conversation``.
"""
import asyncio
from pathlib import Path
from artemis.agents.video_analyzer import video_analyzer as _va
from artemis.agents.video_analyzer.chunk_conversation import (
_ChunkMedia,
_run_native_chunk_conversation,
)
from artemis.agents.video_analyzer.reliability import classify_video_failure
from artemis.llm.reliability import retry_policy_for
from artemis.utils.logger import get_logger
logger = get_logger(__name__)
async def exec_single_chunk(
analyzer, start_time: float, end_time: float | None, specific_query: str
) -> str:
"""Analyzes one closed chunk interval, with claim/lease and retry handling."""
max_retries = analyzer.native_max_retries
current_start = round(start_time, 1) if isinstance(start_time, (int, float)) else start_time
current_end = round(end_time, 1) if isinstance(end_time, (int, float)) else end_time
if not isinstance(current_start, (int, float)) or not isinstance(
current_end, (int, float)
):
raise ValueError("Child video analysis requires a closed numeric interval")
device_id = getattr(getattr(analyzer.ctx, "device", None), "device_id", None)
active_session = _va.get_active_session(device_id) if device_id else None
claim = analyzer.blackboard.claim_segment(
current_start,
current_end,
specific_query,
model_name=analyzer.model_name,
source_generation=(
active_session.generation if active_session is not None else None
),
)
if claim.state == "cached":
cached = (
f"[from {current_start:.1f}s to {current_end:.1f}s] Summary: "
f"{claim.summary or 'No summary provided.'}"
)
if claim.analysis:
cached += f" Analysis: {claim.analysis}"
return cached
if claim.state == "in_progress":
return (
f"[from {current_start:.1f}s to {current_end:.1f}s] "
"Analysis is already in progress in another video agent; do not duplicate it."
)
lease_owner = claim.lease_owner
if lease_owner is None:
raise RuntimeError("Video blackboard returned a claim without a lease owner")
media = _ChunkMedia()
for attempt in range(max_retries + 1):
try:
logger.info(
f"Attempt {attempt + 1}/{max_retries + 1} for sub-agent:"
f" start={current_start}, end={current_end}"
)
result, slowdown_factor = await _prepare_chunk_media(
analyzer, media, current_start, current_end
)
media.prompt_with_context = _build_chunk_prompt(
analyzer,
result,
media.actual_start,
slowdown_factor,
start_time,
end_time,
specific_query,
)
if not analyzer.use_native_gemini:
chunk_result = await analyzer._exec_single_chunk_universal(
compressed_path=media.compressed_path,
raw_path=media.path,
start_time=current_start,
end_time=current_end,
actual_start=media.actual_start,
prompt_with_context=media.prompt_with_context,
specific_query=specific_query,
)
summary, analysis = analyzer._parse_chunk_result(chunk_result)
analyzer.blackboard.complete_segment(
current_start,
current_end,
specific_query,
lease_owner,
summary,
analysis,
)
return chunk_result
return await _run_native_chunk_conversation(
analyzer,
media,
current_start,
current_end,
start_time,
end_time,
specific_query,
lease_owner,
)
except Exception as e:
# Retry decisions are owned by the shared classification layer.
# Every attempt restarts from scratch (segment re-extracted,
# file re-uploaded, stream re-consumed); partially accumulated
# stream output from the failed attempt is discarded, never
# surfaced as a complete answer.
failure = classify_video_failure(e)
attempts_so_far = attempt + 1
if not failure.retryable:
logger.warning(
f"Sub-agent non-retryable failure [{failure.category.value}]"
f" on attempt {attempts_so_far}: {e}"
)
return await _finish_failed_chunk(
analyzer,
media,
current_start,
current_end,
specific_query,
lease_owner,
e,
retryable=False,
)
policy = retry_policy_for(failure.category)
allowed_attempts = min(max_retries + 1, policy.max_attempts)
if attempts_so_far >= allowed_attempts:
logger.warning(
f"Sub-agent exhausted {attempts_so_far} attempt(s)"
f" [{failure.category.value}]: {e}"
)
return await _finish_failed_chunk(
analyzer,
media,
current_start,
current_end,
specific_query,
lease_owner,
e,
)
delay = policy.delay_for(attempts_so_far)
logger.warning(
f"Sub-agent retryable failure [{failure.category.value}]"
f" ({attempts_so_far}/{allowed_attempts}): {e}; retrying"
f" from scratch in {delay:.2f}s (partial stream output"
" from the failed attempt is discarded)"
)
_va._record_llm_retry(
str(e),
delay,
attempt=attempts_so_far,
max_retries=allowed_attempts,
source="video_sub_agent",
)
await asyncio.sleep(delay)
raise RuntimeError(
f"Video sub-agent exhausted retries for {current_start:.1f}s-{current_end:.1f}s"
)
async def _finish_failed_chunk(
analyzer,
media: _ChunkMedia,
current_start: float,
current_end: float,
specific_query: str,
lease_owner: str,
error: Exception,
*,
retryable: bool | None = None,
):
"""Records terminal failure, attempting the universal fallback first when eligible."""
failure = classify_video_failure(error)
if (
analyzer.use_native_gemini
and failure.should_fallback
and media.path is not None
and media.compressed_path is not None
and media.actual_start is not None
and media.prompt_with_context is not None
):
try:
logger.warning(
"Native video chunk exhausted retries; using universal fallback"
)
_va._record_llm_event(
"llm_fallback",
{
"reason": "native_video_chunk_failed",
"category": failure.category.value,
"error": str(error)[:500],
},
)
fallback_result = await analyzer._exec_single_chunk_universal(
compressed_path=media.compressed_path,
raw_path=media.path,
start_time=current_start,
end_time=current_end,
actual_start=media.actual_start,
prompt_with_context=media.prompt_with_context,
specific_query=specific_query,
force_fallback=True,
)
summary, analysis = analyzer._parse_chunk_result(fallback_result)
analyzer.blackboard.complete_segment(
current_start,
current_end,
specific_query,
lease_owner,
summary,
analysis,
)
return fallback_result
except Exception as fallback_error:
error = fallback_error
failure = classify_video_failure(error)
retryable = None
analyzer.blackboard.fail_segment(
current_start,
current_end,
specific_query,
lease_owner,
str(error),
retryable=failure.retryable if retryable is None else retryable,
error_category=failure.category.value,
)
raise error
async def _prepare_chunk_media(
analyzer, media: _ChunkMedia, current_start: float, current_end: float
):
"""Extracts and compresses the chunk segment, tracking artifacts for cleanup."""
controller = _va.get_controller(analyzer.ctx)
async with _va.TRANSCODE_SEMAPHORE:
result = await controller.extract_segment_metadata(current_start, current_end)
if not result.success or not result.video_path:
raise Exception(f"Failed to get video segment: {result.message}")
path = Path(result.video_path)
analyzer.local_files_to_cleanup.add(path)
analyzer.local_dirs_to_cleanup.add(path.parent)
media.path = path
actual_start = getattr(result, "actual_start_relative_time", current_start)
if not isinstance(actual_start, (int, float)):
actual_start = current_start
media.actual_start = actual_start
slowdown_factor = 1.0
if (
current_end is not None
and (current_end - current_start) <= _va.SLOWDOWN_THRESHOLD_SECONDS
):
slowdown_factor = _va.SLOWDOWN_FACTOR
async with _va.TRANSCODE_SEMAPHORE:
compressed_path = await _va.compress_video_for_api(
path,
force_compress=True,
start_offset_seconds=actual_start,
slowdown_factor=slowdown_factor,
)
if compressed_path != path:
analyzer.local_files_to_cleanup.add(compressed_path)
analyzer.local_dirs_to_cleanup.add(compressed_path.parent)
media.compressed_path = compressed_path
actual_start = getattr(result, "actual_start_relative_time", current_start)
if not isinstance(actual_start, (int, float)):
actual_start = current_start
media.actual_start = actual_start
return result, slowdown_factor
def _build_chunk_prompt(
analyzer,
result,
actual_start: float,
slowdown_factor: float,
start_time: float,
end_time: float | None,
specific_query: str,
) -> str:
"""Assembles the sub-agent prompt with timing context, ledger, and warnings."""
warnings = analyzer.get_overlapping_warnings(start_time, end_time)
warning_block = ""
if warnings:
lines = [
"WARNING: The following queries were already tried in this timeframe:\n"
]
for w in warnings:
lines.append(f"Searched for: {w['target']} -> {w['summary']}")
warning_block = "\n".join(lines) + "\n\n"
duration_secs = getattr(result, "duration_seconds", None)
actual_end = (
actual_start + duration_secs
if isinstance(duration_secs, (int, float))
else None
)
end_str = f" to {actual_end:.1f}s" if actual_end is not None else ""
warning_val = getattr(result, "warning", None)
truncation_note = (
f" (NOTE: {warning_val})"
if isinstance(warning_val, str) and warning_val
else ""
)
slowdown_note = ""
if slowdown_factor != 1.0:
slowdown_note = (
"WARNING: This video is slowed down to capture fast micro-actions.\n\n"
)
prompt_with_context = (
"IMPORTANT CONTEXT: This video segment corresponds to"
" the test's relative time from"
f" {actual_start:.1f}s{end_str}{truncation_note}. A"
" timestamp showing the exact test relative time"
" (formatted as '<seconds> s') is burned in the"
" top-right corner of the video. Please watch this"
" burned-in timestamp and use its value to report all"
" timestamps (e.g., if the burned timestamp shows '90"
f" s', report 90.0s).\n\n{slowdown_note}"
)
if analyzer.enable_ledger:
prompt_with_context += (
"BLACKBOARD"
f" LEDGER:\n{analyzer.blackboard_ledger}\nDirective:"
" Use the Blackboard to understand what happened"
" outside your assigned timeframe. Do not"
" duplicate findings already on the board.\n\n"
)
if warning_block:
prompt_with_context += (
f"{warning_block}Directive: Avoid repeating"
" searches that have already failed. Try to"
" approach the problem from a different angle or"
" use a different search strategy.\n\n"
)
prompt_with_context += f"{specific_query}"
return prompt_with_context
artemis/agents/video_analyzer/gemini_files.py
+154
View File
@@ -0,0 +1,154 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Gemini File API helpers: upload/poll lifecycle and abandoned-file cleanup."""
import asyncio
from datetime import datetime
from pathlib import Path
import time
from artemis.utils.logger import get_logger
try:
from datetime import UTC
except ImportError:
UTC = UTC
logger = get_logger(__name__)
_LAST_CLEANUP_TIME = 0.0
_CLEANUP_LOCK = asyncio.Lock()
async def cleanup_abandoned_gemini_files(client) -> None:
"""Scan and delete remaining cloud video files whose creation time has elapsed standard TTL."""
global _LAST_CLEANUP_TIME
if _CLEANUP_LOCK.locked():
return
async with _CLEANUP_LOCK:
now_ts = time.time()
if now_ts - _LAST_CLEANUP_TIME < 3600:
return
_LAST_CLEANUP_TIME = now_ts
try:
pager = await asyncio.wait_for(client.aio.files.list(), timeout=30)
files = []
async for f in pager:
files.append(f)
now = datetime.now(UTC)
files_to_delete = []
for f in files:
display_name = getattr(f, "display_name", "") or ""
if (
display_name.startswith("compressed_")
or display_name.startswith("audio_")
or "artemis" in display_name
):
create_time = getattr(f, "created_at", None) or getattr(f, "create_time", None)
if not create_time:
continue
if isinstance(create_time, str):
try:
parsed_time = datetime.fromisoformat(create_time.replace("Z", "+00:00"))
except Exception:
continue
elif isinstance(create_time, datetime):
parsed_time = create_time
else:
continue
if parsed_time.tzinfo is None:
parsed_time = parsed_time.replace(tzinfo=UTC)
age_seconds = (now - parsed_time).total_seconds()
if age_seconds > 7200:
files_to_delete.append(f)
logger.info(
f"Marked expired cloud asset for deletion: {f.name}"
f" ({display_name}), Age: {age_seconds / 3600:.1f}h"
)
if files_to_delete:
logger.info(f"Purging {len(files_to_delete)} expired cloud assets in parallel...")
tasks = [
asyncio.wait_for(client.aio.files.delete(name=f.name), timeout=30)
for f in files_to_delete
]
await asyncio.gather(*tasks, return_exceptions=True)
except Exception as e:
logger.warning(f"Routine cloud maintenance skipped: {e}")
async def upload_and_poll_file(
client, compressed_path: Path, cloud_files_to_cleanup: set
) -> any:
"""Upload a media file to the Gemini File API and poll until it is ACTIVE."""
logger.info(f"Uploading {compressed_path} to Gemini File API...")
file_size_mb = compressed_path.stat().st_size / (1024 * 1024)
upload_timeout = max(30.0, min(120.0, file_size_mb * 2.0))
file = await asyncio.wait_for(
client.aio.files.upload(file=compressed_path),
timeout=upload_timeout,
)
cloud_files_to_cleanup.add(file.name)
max_wait = max(60, min(180, 60 + int(file_size_mb * 2)))
wait_interval = 0.5
start_wait = time.time()
retry_count = 0
max_retries = 3
while True:
try:
f_state = await asyncio.wait_for(
client.aio.files.get(name=file.name), timeout=20
)
retry_count = 0
except Exception as poll_error:
retry_count += 1
logger.warning(
f"Temporary issue polling file {file.name} (attempt"
f" {retry_count}/{max_retries}): {poll_error}"
)
if retry_count > max_retries:
raise RuntimeError(
f"Failed to poll Gemini File API after {max_retries} attempts: {poll_error}"
)
await asyncio.sleep(2.0)
continue
if f_state.state.name == "ACTIVE":
break
elif f_state.state.name == "FAILED":
raise RuntimeError(f"Gemini File API processing failed for {file.name}")
if time.time() - start_wait > max_wait:
raise TimeoutError(
f"Gemini File API processing timeout for {file.name}"
f" (Waited {max_wait}s for size {file_size_mb:.1f}MB)"
)
logger.info(f"File {file.name} is {f_state.state.name}, waiting {wait_interval}s...")
await asyncio.sleep(wait_interval)
wait_interval = min(3.0, wait_interval * 1.5)
return file
artemis/agents/video_analyzer/native_coordinator.py
+592
View File
@@ -0,0 +1,592 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Native Gemini coordinator loop for the video analyzer main agent.
Extracted from ``video_analyzer.py`` as a pure structural split. Patchable
collaborators are looked up late through the facade module so
``mock.patch("artemis.agents.video_analyzer.video_analyzer.<name>")`` targets
keep working.
"""
import asyncio
import os
from pathlib import Path
import shutil
import tempfile
from google.genai import types
from artemis.agents.video_analyzer import video_analyzer as _va
from artemis.agents.video_analyzer.reliability import classify_video_failure
from artemis.constants import SAFETY_SETTINGS_BLOCK_NONE
from artemis.data_engine.trace import CURRENT_TRACE_ID, TraceSpan
from artemis.utils.logger import get_logger
logger = get_logger(__name__)
class _DummyResponse:
"""Aggregated view over a consumed stream, shaped like a full response."""
def __init__(self, text, function_calls, parts):
self.text = text
self.function_calls = function_calls
self.candidates = [
types.Candidate(
content=types.Content(
role="model",
parts=parts
if parts
else (
[types.Part.from_text(text=text)] if text else []
),
)
)
]
async def run_native(
analyzer,
ctx,
time_description: str,
purpose: str,
system_prompt: str,
temperature: float,
thinking_level,
) -> tuple[str, str]:
"""Drives the native Gemini coordinator loop to a final answer."""
tools_declaration = analyzer.tools_declaration
contents = [
types.Content(
role="user",
parts=[
types.Part.from_text(
text=(f"Time description: {time_description}\nPurpose: {purpose}")
)
],
)
]
agent_outcome = ""
status = "success"
max_iterations = 10
iterations = 0
last_ledger_index = 0
try:
while iterations < max_iterations:
iterations += 1
logger.info(f"Iteration {iterations}: Invoking Native Gemini SDK...")
analyzer._clean_blackboard()
last_ledger_index = _append_ledger_diff(analyzer, contents, last_ledger_index)
is_final_turn = iterations == max_iterations
_append_turn_guidance(contents, iterations, max_iterations, is_final_turn)
response = await _stream_main_turn(
analyzer,
ctx,
contents,
system_prompt,
temperature,
thinking_level,
tools_declaration,
is_final_turn,
)
function_calls = response.function_calls
if not function_calls:
agent_outcome = response.text
break
if response.candidates and response.candidates[0].content:
contents.append(response.candidates[0].content)
else:
contents.append(
types.Content(
role="model",
parts=[types.Part(function_call=fc) for fc in function_calls],
)
)
tool_response_parts = await _execute_native_tool_calls(analyzer, function_calls)
if tool_response_parts:
contents.append(types.Content(role="user", parts=tool_response_parts))
if iterations >= max_iterations and not agent_outcome:
agent_outcome = (
"Error: Video analyzer agent reached maximum iterations"
" without settling on a final answer."
)
status = "error"
except Exception as e:
logger.error(f"Video analyzer failed: {e}")
failure = classify_video_failure(e)
if failure.should_fallback:
logger.warning(
f"Native coordinator failed with {failure.category.value}; "
"continuing through universal fallback"
)
agent_outcome, status = await analyzer._run_universal(
time_description,
purpose,
system_prompt,
force_fallback=True,
)
else:
agent_outcome = f"Video analysis failed: {e}"
status = "error"
finally:
analyzer._record_reliability_metrics()
await _cleanup_native_run(analyzer)
return agent_outcome, status
def _append_ledger_diff(analyzer, contents: list, last_ledger_index: int) -> int:
"""Appends new blackboard entries (and proofs) since the last coordinator turn."""
new_entries = analyzer.blackboard_entries[last_ledger_index:]
if analyzer.enable_ledger and new_entries:
lines = []
image_parts = []
for e in new_entries:
line = f"[{e['start']}s - {e['end']}s] SUMMARY: {e['summary']}"
screenshot_file = e.get("screenshot") or e.get("screenshot_path")
if screenshot_file and Path(screenshot_file).exists():
line += f" PROOF: {Path(screenshot_file).name}"
image_parts.append(
types.Part.from_text(text=f"PROOF: {Path(screenshot_file).name}")
)
image_parts.append(
types.Part.from_bytes(
data=Path(screenshot_file).read_bytes(),
mime_type="image/jpeg",
)
)
lines.append(line)
diff_text = "NEW BLACKBOARD ENTRIES ADDED: " + " | ".join(lines)
contents.append(
types.Content(
role="user",
parts=[types.Part.from_text(text=diff_text)],
)
)
if image_parts:
contents.append(types.Content(role="user", parts=image_parts))
last_ledger_index = len(analyzer.blackboard_entries)
return last_ledger_index
def _append_turn_guidance(
contents: list, iterations: int, max_iterations: int, is_final_turn: bool
) -> None:
"""Appends per-turn steering messages (final-turn notice or progress nudge)."""
if is_final_turn:
contents.append(
types.Content(
role="user",
parts=[
types.Part.from_text(
text=(
"This is your final iteration; all"
" tools are stripped, and you must"
" provide your final answer directly."
)
)
],
)
)
else:
if iterations > 1:
contents.append(
types.Content(
role="user",
parts=[
types.Part.from_text(
text=(
"You have not completed the video"
" analysis yet (iteration"
f" {iterations} of"
f" {max_iterations})."
)
)
],
)
)
def _record_main_llm_trace(
ctx, trace_id, contents: list, system_prompt: str, tools_declaration, is_final_turn: bool
) -> None:
"""Records the sanitized coordinator LLM call into the trace store."""
serialized_contents = []
for c in contents:
if hasattr(c, "model_dump"):
dumped = c.model_dump()
if (
isinstance(dumped, dict)
and "parts" in dumped
and dumped["parts"]
):
for p in dumped["parts"]:
if (
isinstance(p, dict)
and "inline_data" in p
and p["inline_data"]
and "data" in p["inline_data"]
):
p["inline_data"]["data"] = "<image_bytes_sanitized>"
serialized_contents.append(dumped)
else:
serialized_contents.append(str(c))
ctx.data_engine.record_trace(
type="llm_call",
name="video_analyzer_main",
payload={
"contents": serialized_contents,
"system_instruction": system_prompt,
"tools": (
[t.name for t in tools_declaration]
if (tools_declaration and not is_final_turn)
else []
),
},
parent_trace_id=trace_id,
)
async def _stream_main_turn(
analyzer,
ctx,
contents: list,
system_prompt: str,
temperature: float,
thinking_level,
tools_declaration,
is_final_turn: bool,
):
"""Runs one traced, retry-wrapped streaming turn of the main coordinator."""
with TraceSpan(name="gemini_main_agent_call") as span:
trace_id = CURRENT_TRACE_ID.get()
if ctx.data_engine and trace_id:
_record_main_llm_trace(
ctx, trace_id, contents, system_prompt, tools_declaration, is_final_turn
)
async def run_stream():
return await _consume_main_stream(
analyzer,
ctx,
span,
trace_id,
contents,
system_prompt,
temperature,
thinking_level,
tools_declaration,
is_final_turn,
)
response = await _va._invoke_with_retry(
lambda: asyncio.wait_for(run_stream(), timeout=300),
"Native Gemini video coordinator",
)
span.result = (
f"Function calls: {len(response.function_calls)}"
if response.function_calls
else "Final answer"
)
return response
async def _consume_main_stream(
analyzer,
ctx,
span,
trace_id,
contents: list,
system_prompt: str,
temperature: float,
thinking_level,
tools_declaration,
is_final_turn: bool,
) -> _DummyResponse:
"""Starts the coordinator stream and aggregates it into a response object."""
stream = await asyncio.wait_for(
analyzer.client.aio.models.generate_content_stream(
model=analyzer.model_name,
contents=contents,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
temperature=temperature,
tools=[]
if is_final_turn
else [types.Tool(function_declarations=tools_declaration)],
safety_settings=SAFETY_SETTINGS_BLOCK_NONE,
**(
{
"thinking_config": types.ThinkingConfig(
thinking_level=thinking_level
)
}
if thinking_level
else {}
),
),
),
timeout=60,
)
text = ""
accumulated_parts = []
function_calls = []
async for chunk in stream:
if hasattr(chunk, "usage_metadata") and chunk.usage_metadata:
span.payload["usage_metadata"] = {
"prompt_token_count": getattr(
chunk.usage_metadata,
"prompt_token_count",
0,
),
"candidates_token_count": getattr(
chunk.usage_metadata,
"candidates_token_count",
0,
),
"total_token_count": getattr(
chunk.usage_metadata,
"total_token_count",
0,
),
}
chunk_text = chunk.text or ""
text += chunk_text
if ctx.data_engine and trace_id and chunk_text:
ctx.data_engine.stream_output(
trace_id, chunk_text, is_thinking=False
)
if chunk.function_calls:
function_calls.extend(chunk.function_calls)
candidates = getattr(chunk, "candidates", None)
if candidates and isinstance(candidates, list) and len(candidates) > 0:
content = getattr(candidates[0], "content", None)
parts = getattr(content, "parts", None) if content else None
if parts:
for part in parts:
accumulated_parts.append(part)
if getattr(part, "thought", False) and part.text:
if ctx.data_engine and trace_id:
ctx.data_engine.stream_output(
trace_id,
part.text,
is_thinking=True,
)
else:
if chunk.function_calls:
for fc in chunk.function_calls:
accumulated_parts.append(types.Part(function_call=fc))
return _DummyResponse(text, function_calls, accumulated_parts)
async def _execute_native_tool_calls(analyzer, function_calls: list) -> list:
"""Executes coordinator tool calls (parallel where safe) and collects responses."""
sub_agent_calls = [
fc
for fc in function_calls
if (fc.name.split(":")[-1] if ":" in fc.name else fc.name) == "spawn_sub_agent"
]
audio_only_calls = [
fc
for fc in function_calls
if (fc.name.split(":")[-1] if ":" in fc.name else fc.name)
== "analyze_audio_only"
]
other_calls = [
fc
for fc in function_calls
if (fc.name.split(":")[-1] if ":" in fc.name else fc.name)
not in ["spawn_sub_agent", "analyze_audio_only"]
]
tool_response_parts = []
if sub_agent_calls:
await _run_spawn_calls(analyzer, sub_agent_calls, tool_response_parts)
if audio_only_calls:
await _run_audio_calls(analyzer, audio_only_calls, tool_response_parts)
await _run_other_calls(analyzer, other_calls, tool_response_parts)
return tool_response_parts
async def _run_spawn_calls(
analyzer, sub_agent_calls: list, tool_response_parts: list
) -> None:
"""Runs spawn_sub_agent calls in parallel and appends their responses."""
logger.info(f"Executing {len(sub_agent_calls)} sub-agent calls in parallel...")
async def bounded_spawn(fc):
# exec_spawn_sub_agent limits each actual model leaf.
# Holding the same semaphore around the coordinator
# would deadlock when all slots wait for child slots.
return await analyzer.exec_spawn_sub_agent(
fc.args.get("start_time"),
fc.args.get("end_time"),
fc.args.get("specific_query"),
)
tasks = [bounded_spawn(fc) for fc in sub_agent_calls]
results = await asyncio.gather(*tasks, return_exceptions=True)
for fc, result in zip(sub_agent_calls, results):
if isinstance(result, Exception):
error_msg = f"Sub-agent failed. Error: {result}"
logger.error(error_msg)
tool_response_parts.append(
types.Part.from_function_response(
name=fc.name, response={"error": error_msg}
)
)
else:
tool_response_parts.append(
types.Part.from_function_response(
name=fc.name, response={"result": result}
)
)
async def _run_audio_calls(
analyzer, audio_only_calls: list, tool_response_parts: list
) -> None:
"""Runs analyze_audio_only calls in parallel and appends their responses."""
logger.info(
f"Executing {len(audio_only_calls)} analyze_audio_only calls in parallel..."
)
async def bounded_audio_spawn(fc):
async with _va.API_SEMAPHORE:
return await analyzer.exec_analyze_audio_only(
fc.args.get("start_time"),
fc.args.get("end_time"),
fc.args.get("specific_query"),
)
tasks = [bounded_audio_spawn(fc) for fc in audio_only_calls]
results = await asyncio.gather(*tasks, return_exceptions=True)
for fc, result in zip(audio_only_calls, results):
if isinstance(result, Exception):
error_msg = f"Audio analysis failed. Error: {result}"
logger.error(error_msg)
tool_response_parts.append(
types.Part.from_function_response(
name=fc.name, response={"error": error_msg}
)
)
else:
tool_response_parts.append(
types.Part.from_function_response(
name=fc.name, response={"result": result}
)
)
async def _run_other_calls(
analyzer, other_calls: list, tool_response_parts: list
) -> None:
"""Runs remaining coordinator tool calls sequentially."""
for fc in other_calls:
name = fc.name.split(":")[-1] if ":" in fc.name else fc.name
args = fc.args
logger.info(f"Executing tool '{name}' sequentially...")
if name == "extract_segment_metadata":
try:
res = await analyzer.exec_extract_segment_metadata(
args.get("start_time"), args.get("end_time")
)
tool_response_parts.append(
types.Part.from_function_response(
name=name, response={"result": res}
)
)
except Exception as e:
tool_response_parts.append(
types.Part.from_function_response(
name=name, response={"error": str(e)}
)
)
async def _cleanup_native_run(analyzer) -> None:
"""Releases remaining cloud files and temporary local media after a run."""
if analyzer.cloud_files_to_cleanup:
logger.info(
f"Cleaning up {len(analyzer.cloud_files_to_cleanup)} remaining"
" cloud files in parallel..."
)
tasks = [
asyncio.wait_for(analyzer.client.aio.files.delete(name=name), timeout=30)
for name in list(analyzer.cloud_files_to_cleanup)
]
await asyncio.gather(*tasks, return_exceptions=True)
analyzer.cloud_files_to_cleanup.clear()
if os.environ.get("KEEP_VIDEOS") or os.environ.get("ARTEMIS_DEBUG"):
logger.info(
"Skipping cleanup of local video files and directories due to debug mode."
)
else:
for path in list(analyzer.local_files_to_cleanup):
try:
if path.exists():
logger.info(f"Cleaning up local file {path}...")
path.unlink()
except Exception as e:
logger.error(f"Failed to delete local file {path}: {e}")
for dir_path in list(analyzer.local_dirs_to_cleanup):
try:
resolved = dir_path.resolve()
system_dirs = {
Path("/").resolve(),
Path("/tmp").resolve(),
Path("/var/tmp").resolve(),
Path.home().resolve(),
Path(tempfile.gettempdir()).resolve(),
}
if resolved in system_dirs:
logger.debug(f"Skipping deletion of root/system directory: {dir_path}")
continue
if dir_path.exists():
logger.info(f"Cleaning up local directory {dir_path}...")
shutil.rmtree(dir_path, ignore_errors=True)
except Exception as e:
logger.error(f"Failed to delete local directory {dir_path}: {e}")
artemis/agents/video_analyzer/universal_engine.py
+501
View File
@@ -0,0 +1,501 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Universal (OpenAI-compatible / LangChain) execution paths for the video analyzer.
Extracted from ``video_analyzer.py`` as a pure structural split. Patchable
collaborators (``extract_keyframes_from_video``, ``extract_audio_from_video``,
semaphores, retry helper, ...) are looked up late through the facade module so
``mock.patch("artemis.agents.video_analyzer.video_analyzer.<name>")`` targets
keep working.
"""
import asyncio
import base64
import glob
from pathlib import Path
import re
from typing import Any
from langchain_core.messages import (
BaseMessage,
HumanMessage,
SystemMessage,
ToolMessage,
)
from artemis.agents.video_analyzer import video_analyzer as _va
from artemis.agents.video_analyzer.reliability import classify_video_failure
from artemis.data_engine.trace import CURRENT_TRACE_ID, TraceSpan
from artemis.utils.logger import get_logger
logger = get_logger(__name__)
async def exec_single_chunk_universal(
analyzer,
compressed_path: Path,
raw_path: Path,
start_time: float,
end_time: float | None,
actual_start: float,
prompt_with_context: str,
specific_query: str,
force_fallback: bool = False,
) -> str:
"""Executes sub-agent chunk analysis via keyframe extraction and LangChain ChatModel."""
keyframes = _gather_universal_frames(
analyzer, raw_path, start_time, end_time, actual_start
)
user_blocks, audio_block = await _build_universal_user_blocks(
analyzer, keyframes, actual_start, prompt_with_context, raw_path
)
messages = [
SystemMessage(content=analyzer.sub_system_prompt),
HumanMessage(content=user_blocks),
]
trace_id = CURRENT_TRACE_ID.get()
if analyzer.ctx.data_engine and trace_id:
analyzer.ctx.data_engine.record_trace(
type="llm_call",
name="video_sub_agent_universal",
payload={
"num_keyframes": len(keyframes),
"prompt": prompt_with_context,
"system_instruction": analyzer.sub_system_prompt,
},
parent_trace_id=trace_id,
)
response = await _invoke_universal_with_visual_retry(
analyzer, messages, user_blocks, audio_block, force_fallback
)
final_summary, final_analysis, timeline_events = _extract_submit_payload(response)
await _persist_universal_events(
analyzer,
timeline_events,
start_time,
end_time,
actual_start,
raw_path,
specific_query,
final_summary,
)
end_str = f"{end_time:.1f}s" if end_time is not None else "unknown"
return (
f"[from {start_time:.1f}s to {end_str}] Summary: {final_summary}"
f" Analysis: {final_analysis}"
)
def _gather_universal_frames(
analyzer,
raw_path: Path,
start_time: float,
end_time: float | None,
actual_start: float,
) -> list:
"""Extracts uniform keyframes plus dense action-window frames, merged by timestamp."""
# Universal providers consume stills rather than the uploaded clip. Use
# the raw extracted segment so slowdown/transcode PTS never skew labels.
keyframes = _va.extract_keyframes_from_video(
raw_path,
fps=1.0,
max_frames=30,
max_dimension=1080,
)
dense_offsets = analyzer._dense_action_offsets(
start_time,
end_time if end_time is not None else start_time,
actual_start,
)
targeted_frames = _va.extract_frames_at_timestamps(
raw_path,
dense_offsets,
max_frames=analyzer.max_dense_action_frames,
max_dimension=1080,
)
merged_frames = {round(timestamp, 3): data for timestamp, data in keyframes}
merged_frames.update(
{round(timestamp, 3): data for timestamp, data in targeted_frames}
)
return sorted(merged_frames.items())
async def _build_universal_user_blocks(
analyzer,
keyframes: list,
actual_start: float,
prompt_with_context: str,
raw_path: Path,
) -> tuple[list[dict[str, Any]], dict[str, Any] | None]:
"""Builds the multimodal user content blocks (text, frames, optional audio)."""
user_blocks: list[dict[str, Any]] = [{"type": "text", "text": prompt_with_context}]
for ts_sec, frame_bytes in keyframes:
b64_str = base64.b64encode(frame_bytes).decode("utf-8")
abs_time = actual_start + ts_sec
user_blocks.append(
{
"type": "text",
"text": f"--- Video Keyframe at {abs_time:.1f}s ---",
}
)
user_blocks.append(
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64_str}"},
}
)
audio_block = None
try:
async with _va.TRANSCODE_SEMAPHORE:
audio_path = await _va.extract_audio_from_video(raw_path)
analyzer.local_files_to_cleanup.add(audio_path)
analyzer.local_dirs_to_cleanup.add(audio_path.parent)
audio_block = analyzer._audio_content_block(audio_path)
user_blocks.append(audio_block)
except Exception as audio_error:
logger.info(f"No usable audio track for universal video chunk: {audio_error}")
return user_blocks, audio_block
async def _invoke_universal_with_visual_retry(
analyzer,
messages: list[BaseMessage],
user_blocks: list[dict[str, Any]],
audio_block: dict[str, Any] | None,
force_fallback: bool,
):
"""Invokes the universal model, retrying visually when audio input is rejected."""
with TraceSpan(name="universal_sub_agent_call") as span:
try:
response = await analyzer._invoke_universal_model(
messages,
[_va.UNIVERSAL_SUBMIT_ANSWER_TOOL],
timeout=analyzer.model_call_timeout_seconds,
label="Universal video sub-agent",
force_fallback=force_fallback,
)
except Exception as multimodal_error:
failure = classify_video_failure(multimodal_error)
if audio_block is None or failure.category.value not in {
"bad_request",
"media_processing",
}:
raise
logger.warning(
"Configured video model rejected audio input; retrying this chunk visually"
)
visual_blocks = [block for block in user_blocks if block is not audio_block]
messages = [
SystemMessage(content=analyzer.sub_system_prompt),
HumanMessage(content=visual_blocks),
]
response = await analyzer._invoke_universal_model(
messages,
[_va.UNIVERSAL_SUBMIT_ANSWER_TOOL],
timeout=analyzer.model_call_timeout_seconds,
label="Universal visual-only sub-agent",
force_fallback=force_fallback,
)
span.result = (
f"Received response (tool_calls={len(getattr(response, 'tool_calls', []))})"
)
return response
def _extract_submit_payload(response) -> tuple[str, str, list]:
"""Pulls summary/analysis/timeline events from a universal model response."""
final_summary = "No summary provided."
final_analysis = "No analysis provided."
timeline_events = []
if getattr(response, "tool_calls", None):
for tc in response.tool_calls:
if tc.get("name") == "submit_answer":
args = tc.get("args", {})
final_summary = (
str(args.get("summary", "")).strip().replace("\n", " ") or final_summary
)
final_analysis = (
str(args.get("analysis", "")).strip().replace("\n", " ") or final_analysis
)
timeline_events = args.get("timeline_events", [])
break
elif getattr(response, "content", None):
text_content = str(response.content).strip()
final_summary = text_content[:200].replace("\n", " ")
final_analysis = text_content
return final_summary, final_analysis, timeline_events
async def _persist_universal_events(
analyzer,
timeline_events: list,
start_time: float,
end_time: float | None,
actual_start: float,
raw_path: Path,
specific_query: str,
final_summary: str,
) -> None:
"""Processes timeline events & saves screenshots for the blackboard ledger."""
for event in timeline_events:
if not isinstance(event, dict):
continue
try:
v_ts = float(event.get("verification_timestamp_secs", start_time))
conf = float(event.get("confidence_score", 0.8))
event_start = float(event.get("start_time", start_time))
event_end = float(
event.get(
"end_time",
end_time if end_time is not None else event_start,
)
)
except (TypeError, ValueError):
logger.warning(f"Ignoring malformed universal video event: {event}")
continue
interval_end = end_time if end_time is not None else max(event_end, v_ts)
if (
not 0.0 <= conf <= 1.0
or event_end < event_start
or event_start < start_time
or event_end > interval_end
or v_ts < start_time
or v_ts > interval_end
):
logger.warning(f"Ignoring out-of-range universal video event: {event}")
continue
screenshot_path_str = None
try:
rel_offset = max(0.0, v_ts - actual_start)
out_dir = _va.get_temp_dir("video_analyzer")
out_dir.mkdir(parents=True, exist_ok=True)
existing_imgs = glob.glob(f"{out_dir}/img_*.jpg")
max_idx = -1
for img in existing_imgs:
m = re.search(r"img_(\d+)\.jpg$", img)
if m:
max_idx = max(max_idx, int(m.group(1)))
next_idx = max_idx + 1
shot_path = out_dir / f"img_{next_idx}.jpg"
ffmpeg_cmd = [
"ffmpeg",
"-ss",
str(rel_offset),
"-i",
str(raw_path),
"-vframes",
"1",
"-q:v",
"2",
"-y",
str(shot_path),
]
proc = await asyncio.create_subprocess_exec(
*ffmpeg_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
await proc.communicate()
if shot_path.exists():
analyzer.local_files_to_cleanup.add(shot_path)
screenshot_path_str = str(shot_path)
except Exception as fe:
logger.warning(f"Failed extracting keyframe for event {v_ts}s: {fe}")
entry = {
"start": event_start,
"end": event_end,
"target": specific_query,
"summary": event.get("transcription", final_summary),
"confidence_score": conf,
}
if screenshot_path_str:
entry["screenshot"] = screenshot_path_str
analyzer._record_blackboard_entry(entry)
async def exec_analyze_audio_universal(
analyzer,
audio_path: Path,
start_time: float,
end_time: float | None,
actual_start: float,
prompt_with_context: str,
specific_query: str,
force_fallback: bool = False,
) -> str:
"""Executes audio analysis via Universal LangChain ChatModel."""
messages = [
SystemMessage(content=analyzer.audio_system_prompt),
HumanMessage(
content=[
{"type": "text", "text": prompt_with_context},
analyzer._audio_content_block(audio_path),
]
),
]
response = await analyzer._invoke_universal_model(
messages,
[_va.UNIVERSAL_SUBMIT_ANSWER_TOOL],
timeout=analyzer.model_call_timeout_seconds,
label="Universal audio sub-agent",
force_fallback=force_fallback,
)
final_summary = "No audio summary provided."
final_analysis = "No audio analysis provided."
if getattr(response, "tool_calls", None):
for tc in response.tool_calls:
if tc.get("name") == "submit_answer":
args = tc.get("args", {})
final_summary = (
str(args.get("summary", "")).strip().replace("\n", " ") or final_summary
)
final_analysis = (
str(args.get("analysis", "")).strip().replace("\n", " ") or final_analysis
)
break
elif getattr(response, "content", None):
final_summary = str(response.content)[:200].replace("\n", " ")
final_analysis = str(response.content)
end_val = end_time if end_time is not None else "unknown"
analyzer._record_blackboard_entry(
{
"start": start_time,
"end": end_val,
"target": specific_query,
"summary": final_summary,
"confidence_score": 0.8,
},
modality="audio",
)
end_str = f"{end_time:.1f}s" if end_time is not None else "unknown"
return (
f"[from {start_time:.1f}s to {end_str}] Audio Summary: {final_summary}"
f" Analysis: {final_analysis}"
)
async def run_universal(
analyzer,
time_description: str,
purpose: str,
system_prompt: str,
*,
force_fallback: bool = False,
) -> tuple[str, str]:
"""Runs VideoAnalyzer main reasoning loop using Universal LangChain BaseChatModel."""
messages: list[BaseMessage] = [
SystemMessage(content=system_prompt),
HumanMessage(
content=(
f"Time description: {time_description}\n"
f"Purpose: {purpose}\n"
f"Persistent video blackboard: {analyzer.blackboard_ledger}\n"
"Reuse sufficient prior evidence and do not request an identical "
"interval/query analysis again.\n"
"Please coordinate video metadata extraction and sub-agent analysis."
)
),
]
max_iterations = 6
iterations = 0
agent_outcome = ""
status = "success"
try:
while iterations < max_iterations:
iterations += 1
response = await _va._invoke_with_retry(
lambda: analyzer._invoke_universal_model(
messages,
_va.UNIVERSAL_MAIN_TOOLS,
timeout=300,
label="Universal video coordinator",
force_fallback=force_fallback,
),
"Universal video coordinator",
)
messages.append(response)
if not getattr(response, "tool_calls", None):
agent_outcome = str(response.content)
break
for tc in response.tool_calls:
t_name = tc["name"]
t_args = tc.get("args", {})
t_id = tc.get("id", f"call_{iterations}")
logger.info(f"Executing universal tool '{t_name}' with args {t_args}")
res = await _execute_universal_tool(analyzer, t_name, t_args)
messages.append(ToolMessage(tool_call_id=t_id, name=t_name, content=str(res)))
if not agent_outcome:
agent_outcome = "Video analyzer completed all turns without error."
except Exception as e:
logger.error(f"Universal video analyzer failed: {e}")
agent_outcome = f"Video analysis failed: {e}"
status = "error"
analyzer._record_reliability_metrics()
return agent_outcome, status
async def _execute_universal_tool(analyzer, t_name: str, t_args: dict) -> str:
"""Dispatches one universal coordinator tool call, converting errors to text."""
try:
if t_name == "extract_segment_metadata":
res = await analyzer.exec_extract_segment_metadata(
t_args.get("start_time"), t_args.get("end_time")
)
elif t_name == "spawn_sub_agent":
res = await analyzer.exec_spawn_sub_agent(
t_args.get("start_time"),
t_args.get("end_time"),
t_args.get("specific_query"),
)
elif t_name == "analyze_audio_only":
res = await analyzer.exec_analyze_audio_only(
t_args.get("start_time"),
t_args.get("end_time"),
t_args.get("specific_query"),
)
else:
res = f"Error: Tool '{t_name}' is not recognized."
except Exception as tool_err:
logger.error(f"Error executing tool '{t_name}': {tool_err}")
res = f"Error executing tool '{t_name}': {tool_err}"
return res
artemis/agents/video_analyzer/universal_tools.py
+324
View File
@@ -0,0 +1,324 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tool schema declarations for the video analyzer.
Holds both the OpenAI-style function schemas used by the universal
(LangChain) engine and the builders for the native Gemini
``FunctionDeclaration`` equivalents. Pure data/declarations only.
"""
from typing import Any
from google.genai import types
UNIVERSAL_SUBMIT_ANSWER_TOOL: dict[str, Any] = {
"type": "function",
"function": {
"name": "submit_answer",
"description": "Submit the final findings for the requested segment analysis.",
"parameters": {
"type": "object",
"properties": {
"summary": {
"type": "string",
"description": "A short paragraph summarizing your findings.",
},
"analysis": {
"type": "string",
"description": "A short paragraph of detailed analysis.",
},
"timeline_events": {
"type": "array",
"description": (
"List of detected events matching the query. If no relevant"
" events occurred, return an empty array."
),
"items": {
"type": "object",
"properties": {
"start_time": {
"type": "number",
"description": "Event start time (seconds)",
},
"end_time": {
"type": "number",
"description": "Event end time (seconds)",
},
"transcription": {
"type": "string",
"description": (
"A concise transcription of what happened in the event."
),
},
"confidence_score": {
"type": "number",
"description": "Confidence score between 0.0 and 1.0",
},
"verification_timestamp_secs": {
"type": "number",
"description": (
"Exact timestamp (seconds) for verification screenshot."
),
},
},
"required": [
"start_time",
"end_time",
"transcription",
"confidence_score",
"verification_timestamp_secs",
],
},
},
},
"required": ["summary", "analysis"],
},
},
}
UNIVERSAL_MAIN_TOOLS: list[dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "extract_segment_metadata",
"description": (
"Retrieves metadata (duration_seconds, file_size_mb) for a video segment."
" Does not return video content or perform content analysis."
),
"parameters": {
"type": "object",
"properties": {
"start_time": {
"type": "number",
"description": "Start time in seconds",
},
"end_time": {
"type": "number",
"description": (
"End time in seconds (optional, defaults to latest available time)"
),
},
},
"required": ["start_time"],
},
},
},
{
"type": "function",
"function": {
"name": "spawn_sub_agent",
"description": (
"Spawns a sub-agent to analyze visual and audio content of a video segment."
" Parallelizes chunk analysis across workers. Returns combined summary."
),
"parameters": {
"type": "object",
"properties": {
"start_time": {
"type": "number",
"description": "Start time in seconds",
},
"end_time": {
"type": "number",
"description": "End time in seconds",
},
"specific_query": {
"type": "string",
"description": "Specific query or intent for the sub-agent",
},
},
"required": ["start_time", "specific_query"],
},
},
},
{
"type": "function",
"function": {
"name": "analyze_audio_only",
"description": (
"Extracts and analyzes only the audio track from a video segment."
" More efficient for non-visual tasks."
),
"parameters": {
"type": "object",
"properties": {
"start_time": {
"type": "number",
"description": "Start time in seconds",
},
"end_time": {
"type": "number",
"description": "End time in seconds (optional)",
},
"specific_query": {
"type": "string",
"description": "Specific query or intent for the audio content",
},
},
"required": ["start_time", "specific_query"],
},
},
},
]
def build_native_tools_declaration() -> list[types.FunctionDeclaration]:
"""Builds the native Gemini declarations for the coordinator tools."""
return [
types.FunctionDeclaration(
name="extract_segment_metadata",
description=(
"Retrieves metadata (such as duration_seconds and"
" file_size_mb) for a specific video segment. Does not"
" return the video content or perform content analysis."
),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"start_time": types.Schema(
type=types.Type.NUMBER,
description="Start time in seconds",
),
"end_time": types.Schema(
type=types.Type.NUMBER,
description=(
"End time in seconds (optional, defaults to latest available time)"
),
),
},
required=["start_time"],
),
),
types.FunctionDeclaration(
name="spawn_sub_agent",
description=(
"Spawns a sub-agent to analyze both visual and audio"
" content of a video segment. Parallelizes chunk analysis"
" across workers. Returns combined summary and analysis."
),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"start_time": types.Schema(
type=types.Type.NUMBER,
description="Start time in seconds",
),
"end_time": types.Schema(
type=types.Type.NUMBER,
description=(
"End time in seconds. Use the maximum available"
" time if not specified."
),
),
"specific_query": types.Schema(
type=types.Type.STRING,
description=("Specific query or intent for the sub-agent"),
),
},
required=["start_time", "specific_query"],
),
),
types.FunctionDeclaration(
name="analyze_audio_only",
description=(
"Extracts and analyzes only the audio track from a video"
" segment. More efficient for non-visual tasks."
),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"start_time": types.Schema(
type=types.Type.NUMBER,
description="Start time in seconds",
),
"end_time": types.Schema(
type=types.Type.NUMBER,
description="End time in seconds (optional)",
),
"specific_query": types.Schema(
type=types.Type.STRING,
description=("Specific query or intent for the audio content"),
),
},
required=["start_time", "specific_query"],
),
),
]
def build_submit_answer_declaration() -> types.FunctionDeclaration:
"""Builds the native Gemini declaration for the sub-agent submit_answer tool."""
return types.FunctionDeclaration(
name="submit_answer",
description=("Submit the final findings for the requested segment analysis."),
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"summary": types.Schema(
type=types.Type.STRING,
description=("A short paragraph summarizing of your findings."),
),
"analysis": types.Schema(
type=types.Type.STRING,
description="A short paragraph of detailed analysis.",
),
"timeline_events": types.Schema(
type=types.Type.ARRAY,
description=(
"List of detected events matching the query. If no"
" relevant events occurred in the segment, return"
" an empty array."
),
items=types.Schema(
type=types.Type.OBJECT,
properties={
"start_time": types.Schema(
type=types.Type.NUMBER,
description="Event start time (seconds)",
),
"end_time": types.Schema(
type=types.Type.NUMBER,
description="Event end time (seconds)",
),
"transcription": types.Schema(
type=types.Type.STRING,
description=(
"A concise transcription of what happened in the event. "
),
),
"confidence_score": types.Schema(
type=types.Type.NUMBER,
description=("Confidence score between 0 and 1"),
),
"verification_timestamp_secs": types.Schema(
type=types.Type.NUMBER,
description=(
"Exact timestamp (seconds) for verification screenshot."
),
),
},
required=[
"start_time",
"end_time",
"transcription",
"confidence_score",
"verification_timestamp_secs",
],
),
),
},
required=["summary", "analysis", "timeline_events"],
),
)
artemis/agents/video_analyzer/video_analyzer.py
+109 -2355
View File
File diff suppressed because it is too large. Load diff