"""Tool base classes & registry for agentic video understanding.
Design notes
------------
* Tools are pure-Python objects with a JSON-Schema-flavored ``parameters`` dict
so we can render them into the system prompt as an OpenAI-style tool spec.
* ``run(args, ctx)`` returns a :class:`ToolResult` with ``ok`` flag and a
JSON-serializable ``data`` payload. Some tools (e.g. ``zoom_crop``) may
additionally append items to ``ctx.next_round_images`` to inject extra
visual tokens into the next round of the streaming dialog.
* ``ToolContext`` is a thin handle the *agent main loop* (agent.py)
passes in. Tools must NOT import video_processor.py to avoid coupling.
"""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
log = logging.getLogger("agent.tools")
# ---------------------------------------------------------------------------
# Result type
# ---------------------------------------------------------------------------
@dataclass
class ToolResult:
name: str
ok: bool
data: Any = None
error: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {"name": self.name, "ok": self.ok}
if self.data is not None:
d["data"] = self.data
if self.error:
d["error"] = self.error
return d
def to_prompt_str(self) -> str:
return f"{json.dumps(self.to_dict(), ensure_ascii=False)}"
# ---------------------------------------------------------------------------
# Per-sample context handed to tools
# ---------------------------------------------------------------------------
@dataclass
class ToolContext:
"""Mutable per-sample state shared between agent loop and tool impls."""
sample_id: str
task: str
# Streaming clock state
current_round: int = 0
total_rounds: int = 0
question_time: int = 0
# Frames available on disk (1-second granularity)
frame_paths: List[str] = field(default_factory=list)
# Next round of inference may pull frame_idx from here (replay/skip)
next_frame_idx: Optional[int] = None
# Images tools want to inject into the *next* round's prompt as
# additional tokens (e.g. zoom_crop output). Cleared after consume.
next_round_images: List[str] = field(default_factory=list)
# Free-form extra text to append to next user turn
next_round_extra_text: List[str] = field(default_factory=list)
# Set by abstain() to short-circuit the loop with a fixed answer
abstain_reason: Optional[str] = None
# Counters / log
tool_call_log: List[Dict[str, Any]] = field(default_factory=list)
# Per-round budget (reset by agent loop each round)
calls_this_round: int = 0
# ------------------------------------------------------------------
# Optional extensions (Phase-A scaffolding). All default to None so
# legacy tools / call sites that don't know about them keep working.
# See report_doc/Agentic流式视频理解_Memory与工具体系设计.md for design.
# ------------------------------------------------------------------
memory: Optional[Any] = None # tools.memory.MemoryStore
tracks: Optional[Dict[str, Any]] = None # active SAM3 sessions
audio_cache: Optional[Dict[str, Any]] = None
budget: Optional[Any] = None # tools.efficiency.Budget
# Sampling rate hint (frames-per-second) honored by efficiency tool
sample_rate_fps: Optional[float] = None
# Pending defer target (seconds); agent loop fast-forwards until reached
defer_until_t: Optional[int] = None
# Per-sample zoom_crop history: { t_frame: [ [x1,y1,x2,y2], ... ] }.
# Used by ZoomCropTool to dedupe near-identical crops (IoU >= 0.5)
# and enforce per-frame / per-sample call caps so the model cannot
# spam zoom_crop and starve other tools.
zoom_history: Optional[Dict[int, List[List[float]]]] = None
# Per-physical-round tool counters used by promptfix v2 anti-loop
# heuristics. The agent main loop is responsible for resetting these
# whenever a new physical round starts (cf. _last_phys_round handling
# in agent.py). Tools read & write them to:
# * memsearch_history: detect repeated memory_search results and
# redirect the model to replay (改动 1).
# * zoom_count_this_phys_round: enforce a soft 5-call cap on
# zoom_crop per physical round (改动 7).
memsearch_history: List[Dict[str, Any]] = field(default_factory=list)
zoom_count_this_phys_round: int = 0
# ---------------------------------------------------------------------------
# Tool base class
# ---------------------------------------------------------------------------
class Tool:
name: str = ""
description: str = ""
# JSON-Schema-like dict; only top-level "type": "object" with "properties"
parameters: Dict[str, Any] = {"type": "object", "properties": {}}
def run(self, args: Dict[str, Any], ctx: ToolContext) -> ToolResult: # noqa: D401
raise NotImplementedError
# -- prompt rendering -------------------------------------------------
def to_schema(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
if not tool.name:
raise ValueError(f"Tool {tool!r} has empty name")
if tool.name in self._tools:
log.warning("Overriding existing tool %s", tool.name)
self._tools[tool.name] = tool
def get(self, name: str) -> Optional[Tool]:
return self._tools.get(name)
def names(self) -> List[str]:
return sorted(self._tools)
def schemas(self) -> List[Dict[str, Any]]:
return [t.to_schema() for t in self._tools.values()]
def dispatch(self, name: str, args: Dict[str, Any], ctx: ToolContext) -> ToolResult:
tool = self.get(name)
if tool is None:
return ToolResult(name=name, ok=False, error=f"unknown tool: {name}")
try:
res = tool.run(args or {}, ctx)
if not isinstance(res, ToolResult):
res = ToolResult(name=name, ok=True, data=res)
return res
except Exception as e: # noqa: BLE001
log.exception("Tool %s failed", name)
return ToolResult(name=name, ok=False, error=f"{type(e).__name__}: {e}")
# ---------------------------------------------------------------------------
# Default registry factory
# ---------------------------------------------------------------------------
def build_default_registry() -> ToolRegistry:
"""Build the MVP toolset (no external models required).
Honors the env var ``STREAMO_DISABLE_ZOOM_CROP`` (any truthy value:
"1", "true", "yes") to drop ``ZoomCropTool`` from the registry. This
is used for ablations that probe whether zoom_crop is genuinely
necessary or whether the 27B agent over-relies on it.
"""
import os
from .temporal import ReplayTool
from .meta import AbstainTool
from .spatial import ZoomCropTool
disable_zoom = os.environ.get(
"STREAMO_DISABLE_ZOOM_CROP", ""
).strip().lower() in ("1", "true", "yes")
reg = ToolRegistry()
tools = [ReplayTool(), AbstainTool()]
if not disable_zoom:
tools.append(ZoomCropTool())
for t in tools:
reg.register(t)
return reg