From b5706104990fc5017db6981d1a411eb60b38f9f5 Mon Sep 17 00:00:00 2001 From: vieneny <161407215+vieneny@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:29:08 +0800 Subject: [PATCH] fix(video): bound universal analyzer frame inputs --- .../agents/video_analyzer/universal_engine.py | 67 ++++++++++- .../agents/video_analyzer/video_analyzer.py | 1 + artemis/config/agent.py | 11 ++ .../agents/test_video_analyzer_universal.py | 113 ++++++++++++++++++ .../unit/agents/test_video_frame_sampling.py | 60 ++++++++++ 5 files changed, 251 insertions(+), 1 deletion(-) create mode 100644 tests/unit/agents/test_video_frame_sampling.py diff --git a/artemis/agents/video_analyzer/universal_engine.py b/artemis/agents/video_analyzer/universal_engine.py index 96ade3fe7..b5d791e0e 100644 --- a/artemis/agents/video_analyzer/universal_engine.py +++ b/artemis/agents/video_analyzer/universal_engine.py @@ -23,6 +23,7 @@ import asyncio import base64 +from bisect import bisect_left, bisect_right import glob from pathlib import Path import re @@ -59,9 +60,23 @@ async def exec_single_chunk_universal( ) -> 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) + source_frame_count = len(keyframes) + frame_limit = analyzer.max_total_frames + if frame_limit is not None: + actions = analyzer._action_timestamps( + start_time, end_time if end_time is not None else start_time + ) + keyframes = _select_video_frames( + keyframes, frame_limit, [timestamp - actual_start for timestamp in actions] + ) user_blocks, audio_block = await _build_universal_user_blocks( - analyzer, keyframes, actual_start, prompt_with_context, raw_path + analyzer, + keyframes, + actual_start, + prompt_with_context, + raw_path, + sampling_source_count=source_frame_count if frame_limit is not None else None, ) messages = [ @@ -76,6 +91,7 @@ async def exec_single_chunk_universal( name="video_sub_agent_universal", payload={ "num_keyframes": len(keyframes), + "source_num_keyframes": source_frame_count, "prompt": prompt_with_context, "system_instruction": analyzer.sub_system_prompt, }, @@ -137,15 +153,64 @@ def _gather_universal_frames( return sorted(merged_frames.items()) +def _select_video_frames( + frames: list[tuple[float, bytes]], limit: int, action_offsets: list[float] +) -> list[tuple[float, bytes]]: + """Bound sorted extracted frames without changing their timestamps or bytes.""" + if type(limit) is not int or limit < 2: + raise ValueError("Video frame limit must be an integer of at least two") + if len(frames) <= limit: + return frames + times = [timestamp for timestamp, _ in frames] + selected = {0, len(frames) - 1} + # Favor recent actions, adding a before/after pair only when both fit. + # An action outside the extracted range has no evidenced pair. + for action in sorted(set(action_offsets), reverse=True): + before = bisect_left(times, action) - 1 + after = bisect_right(times, action) + if before < 0 or after >= len(frames): + continue + pair = {before, after} + if len(selected | pair) <= limit: + selected.update(pair) + # Fill the remaining slots across time, with an earlier-frame tie break. + while len(selected) < limit: + selected.add( + max( + (index for index in range(len(frames)) if index not in selected), + key=lambda index: ( + min(abs(times[index] - times[chosen]) for chosen in selected), + -times[index], + ), + ) + ) + return [frames[index] for index in sorted(selected)] + + async def _build_universal_user_blocks( analyzer, keyframes: list, actual_start: float, prompt_with_context: str, raw_path: Path, + *, + sampling_source_count: int | None = None, ) -> 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}] + if sampling_source_count is not None: + user_blocks.append( + { + "type": "text", + "text": ( + f"Video frame sampling: selected {len(keyframes)} of " + f"{sampling_source_count} extracted frames in timestamp order. " + "Extracted endpoints and fitting action-adjacent pairs are prioritized. " + "This does not show every video instant or every action; " + "do not infer what happened between the supplied frames." + ), + } + ) for ts_sec, frame_bytes in keyframes: b64_str = base64.b64encode(frame_bytes).decode("utf-8") diff --git a/artemis/agents/video_analyzer/video_analyzer.py b/artemis/agents/video_analyzer/video_analyzer.py index 9704a5796..5ce909a95 100644 --- a/artemis/agents/video_analyzer/video_analyzer.py +++ b/artemis/agents/video_analyzer/video_analyzer.py @@ -178,6 +178,7 @@ def __init__(self, ctx: ArtemisContext): self.max_dense_action_frames = int( getattr(self.video_config, "max_dense_action_frames", 24) ) + self.max_total_frames = getattr(self.video_config, "max_total_frames", None) self.native_max_retries = int(getattr(self.video_config, "native_max_retries", 1)) self.model_call_timeout_seconds = float( getattr(self.video_config, "model_call_timeout_seconds", 120.0) diff --git a/artemis/config/agent.py b/artemis/config/agent.py index 6fb00d398..01fe6f7e8 100644 --- a/artemis/config/agent.py +++ b/artemis/config/agent.py @@ -95,6 +95,17 @@ class VideoAnalyzerConfig(BaseModel): le=120, description="Maximum additional action-proximal keyframes per chunk.", ) + max_total_frames: int | None = Field( + default=None, + ge=2, + le=150, + strict=True, + description=( + "Optional cap on merged frames in one universal video request. " + "Keeps extracted endpoints and prioritizes action-adjacent frame pairs; " + "None preserves the existing frame selection." + ), + ) native_max_retries: int = Field( default=1, ge=0, diff --git a/tests/unit/agents/test_video_analyzer_universal.py b/tests/unit/agents/test_video_analyzer_universal.py index a718aec49..03e24e1d3 100644 --- a/tests/unit/agents/test_video_analyzer_universal.py +++ b/tests/unit/agents/test_video_analyzer_universal.py @@ -14,6 +14,8 @@ """Unit tests for Universal VideoAnalyzer dual-path execution engine.""" +import asyncio +import base64 from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest @@ -67,6 +69,117 @@ def test_universal_tool_schemas(): assert "analyze_audio_only" in tool_names +@pytest.mark.asyncio +@pytest.mark.parametrize("frame_limit", [None, 8]) +async def test_video_frame_cap_applies_before_model_and_discloses_sampling( + mock_context, tmp_path, frame_limit +): + agent = VideoAnalyzer(mock_context) + agent.max_total_frames = frame_limit + frames = [(float(index), bytes([index])) for index in range(53)] + source_bytes = b"unchanged-original-recording" + recording = tmp_path / "recording.mp4" + recording.write_bytes(source_bytes) + prompt = "Check only the supplied test interval. Preserve this full query." + response = AIMessage(content="Observed only supplied frames.") + with ( + patch( + "artemis.agents.video_analyzer.universal_engine._gather_universal_frames", + return_value=frames, + ), + patch.object(agent, "_action_timestamps", return_value=[125.1, 150.1]), + patch( + "artemis.agents.video_analyzer.video_analyzer.extract_audio_from_video", + new=AsyncMock(side_effect=ValueError("no audio")), + ), + patch.object( + agent, "_invoke_universal_model", new=AsyncMock(return_value=response) + ) as invoke, + patch( + "artemis.agents.video_analyzer.universal_engine._persist_universal_events", + new=AsyncMock(), + ), + ): + await agent._exec_single_chunk_universal( + compressed_path=recording, + raw_path=recording, + start_time=100.0, + end_time=153.0, + actual_start=100.0, + prompt_with_context=prompt, + specific_query="test interval", + ) + messages, tools = invoke.await_args.args[:2] + assert messages[0].content == agent.sub_system_prompt + assert messages[1].content[0] == {"type": "text", "text": prompt} + assert tools == [UNIVERSAL_SUBMIT_ANSWER_TOOL] + parts = messages[1].content + images = [part for part in parts if part["type"] == "image_url"] + assert len(images) == (8 if frame_limit else 53) + decoded = [base64.b64decode(part["image_url"]["url"].split(",", 1)[1]) for part in images] + assert decoded == sorted(decoded) + assert decoded[0] == frames[0][1] and decoded[-1] == frames[-1][1] + if frame_limit: + assert {frames[index][1] for index in [25, 26, 50, 51]} <= set(decoded) + assert "selected 8 of 53 extracted frames" in parts[1]["text"] + assert "does not show every video instant or every action" in parts[1]["text"] + else: + assert len(parts) == 1 + 2 * len(frames) + assert recording.read_bytes() == source_bytes + assert len(frames) == 53 + + +@pytest.mark.asyncio +async def test_concurrent_video_chunks_keep_their_own_sampling_counts(mock_context, tmp_path): + agent = VideoAnalyzer(mock_context) + agent.max_total_frames = 8 + sent = [] + + async def invoke(messages, *_args, **_kwargs): + await asyncio.sleep(0) + sent.append(messages[1].content) + return AIMessage(content="Observed only supplied frames.") + + def gather(_agent, path, *_args): + return [(float(index), bytes([index])) for index in range(int(path.stem))] + + with ( + patch( + "artemis.agents.video_analyzer.universal_engine._gather_universal_frames", + side_effect=gather, + ), + patch.object(agent, "_action_timestamps", return_value=[]), + patch( + "artemis.agents.video_analyzer.video_analyzer.extract_audio_from_video", + new=AsyncMock(side_effect=ValueError("no audio")), + ), + patch.object(agent, "_invoke_universal_model", side_effect=invoke), + patch( + "artemis.agents.video_analyzer.universal_engine._persist_universal_events", + new=AsyncMock(), + ), + ): + await asyncio.gather( + *( + agent._exec_single_chunk_universal( + compressed_path=tmp_path / f"{count}.mp4", + raw_path=tmp_path / f"{count}.mp4", + start_time=0.0, + end_time=60.0, + actual_start=0.0, + prompt_with_context=f"Source chunk {count}", + specific_query="test interval", + ) + for count in [53, 51] + ) + ) + assert len(sent) == 2 + for parts in sent: + count = int(parts[0]["text"].split()[-1]) + assert f"selected 8 of {count} extracted frames" in parts[1]["text"] + assert len([part for part in parts if part["type"] == "image_url"]) == 8 + + @pytest.mark.asyncio async def test_exec_single_chunk_universal(mock_context, tmp_path): """Test keyframe extraction and LangChain tool invocation in universal sub-agent.""" diff --git a/tests/unit/agents/test_video_frame_sampling.py b/tests/unit/agents/test_video_frame_sampling.py new file mode 100644 index 000000000..01af35cda --- /dev/null +++ b/tests/unit/agents/test_video_frame_sampling.py @@ -0,0 +1,60 @@ +"""Video request bounds without a device, model call, or loss of source media.""" + +import pytest +from pydantic import ValidationError + +from artemis.agents.video_analyzer.universal_engine import _select_video_frames +from artemis.config.agent import VideoAnalyzerConfig + + +@pytest.mark.parametrize("count", [0, 1, 2, 7, 8, 9, 51, 53, 54, 150]) +def test_selection_is_bounded_and_preserves_original_order_timestamps_and_bytes(count): + frames = [(index / 4, bytes([index])) for index in range(count)] + before = frames.copy() + selected = _select_video_frames(frames, 8, [1.1, 7.1, 11.1]) + assert frames == before + assert len(selected) == min(count, 8) + assert selected == sorted(selected) + assert all(frame in frames for frame in selected) + assert len({timestamp for timestamp, _ in selected}) == len(selected) + if count: + assert selected[0] is frames[0] and selected[-1] is frames[-1] + if count <= 8: + assert selected is frames + + +def test_recent_action_pairs_are_preserved_and_unrelated_ranges_do_not_consume_slots(): + frames = [(float(index), bytes([index])) for index in range(53)] + actions = [-4.0, 25.1, 50.1, 90.0] + selected = _select_video_frames(frames, 8, actions) + assert {frames[index] for index in [0, 25, 26, 50, 51, 52]} <= set(selected) + assert selected == _select_video_frames(frames, 8, list(reversed(actions))) + assert selected == _select_video_frames(frames, 8, [25.1, 50.1]) + + +def test_tight_limit_preserves_endpoints_and_prioritizes_most_recent_action_pair(): + frames = [(float(index), bytes([index])) for index in range(20)] + assert _select_video_frames(frames, 2, [4.5, 15.5]) == [frames[0], frames[-1]] + assert _select_video_frames(frames, 4, [4.5, 15.5]) == [ + frames[0], + frames[15], + frames[16], + frames[-1], + ] + + +def test_exact_action_timestamp_keeps_frames_on_both_sides(): + frames = [(float(index), bytes([index])) for index in range(12)] + selected = _select_video_frames(frames, 4, [6.0]) + assert selected == [frames[0], frames[5], frames[7], frames[-1]] + + +@pytest.mark.parametrize("limit", [0, 1, 151, True, 8.5]) +def test_total_frame_config_rejects_invalid_limits(limit): + with pytest.raises(ValidationError): + VideoAnalyzerConfig(max_total_frames=limit) + + +@pytest.mark.parametrize("limit", [None, 2, 8, 150]) +def test_total_frame_config_is_optional_and_bounded(limit): + assert VideoAnalyzerConfig(max_total_frames=limit).max_total_frames == limit