File size: 6,177 Bytes
0bdbf4d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """Efficiency-control tools: set_sample_rate / sleep / defer / resume.
Plus a :class:`Budget` dataclass that the agent main loop checks each
round to decide whether to short-circuit a sample.
These are pure clock / counter manipulations and require no external
services. They mutate :class:`ToolContext`.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Optional
from .base import Tool, ToolContext, ToolResult
# ---------------------------------------------------------------------------
# Budget
# ---------------------------------------------------------------------------
@dataclass
class Budget:
max_tool_calls: int = 24
max_frames_seen: int = 600
max_tokens: int = 200_000
tool_calls: int = 0
frames_seen: int = 0
tokens_used: int = 0
def add_call(self) -> None:
self.tool_calls += 1
def add_frame(self) -> None:
self.frames_seen += 1
def add_tokens(self, n: int) -> None:
self.tokens_used += int(n)
def exhausted(self) -> bool:
# Any limit set to <=0 means "unlimited" for that dimension.
if self.max_tool_calls > 0 and self.tool_calls >= self.max_tool_calls:
return True
if self.max_frames_seen > 0 and self.frames_seen >= self.max_frames_seen:
return True
if self.max_tokens > 0 and self.tokens_used >= self.max_tokens:
return True
return False
def to_dict(self) -> Dict[str, Any]:
def _fmt(used: int, cap: int) -> str:
return f"{used}/{cap if cap > 0 else 'inf'}"
return {
"tool_calls": _fmt(self.tool_calls, self.max_tool_calls),
"frames_seen": _fmt(self.frames_seen, self.max_frames_seen),
"tokens_used": _fmt(self.tokens_used, self.max_tokens),
}
# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------
_VALID_FPS = (0.25, 0.5, 1.0, 2.0)
# Lock window around question_time: within ``LOCK_BEFORE_Q`` rounds before
# question_time, and any time at/after question_time, ``set_sample_rate``
# is rejected and fps is forced back to 1.0. This prevents the agent from
# striding past question_time with a slow sample rate, which would skip
# the only frame at which </Response> is allowed and produce a null result.
LOCK_BEFORE_Q = 5
class SetSampleRateTool(Tool):
name = "set_sample_rate"
description = (
"Change how often the streaming clock advances per actual second. "
"Use 0.25 fps for static / repetitive scenes to save budget; use "
"2.0 fps only when fast motion is critical. Default is 1.0."
)
parameters = {
"type": "object",
"properties": {
"fps": {"type": "number",
"description": "One of 0.25, 0.5, 1.0, 2.0."},
},
"required": ["fps"],
}
def run(self, args: Dict[str, Any], ctx: ToolContext) -> ToolResult:
try:
fps = float(args["fps"])
except (KeyError, TypeError, ValueError) as e:
return ToolResult(self.name, False, error=f"bad fps: {e}")
# Reject any rate change within the lock window around question_time:
# the streaming clock must hit question_time at fps=1.0 so the model
# gets a chance to emit </Response>. Also reset fps back to 1.0.
q = int(getattr(ctx, "question_time", 0) or 0)
if q > 0 and ctx.current_round >= q - LOCK_BEFORE_Q:
ctx.sample_rate_fps = 1.0
return ToolResult(
self.name, False,
error=(
f"set_sample_rate is locked from t={q - LOCK_BEFORE_Q} "
f"(question_time={q}) onward; fps forced back to 1.0."
),
)
# Clamp to the allowed set
fps = min(_VALID_FPS, key=lambda x: abs(x - fps))
ctx.sample_rate_fps = fps
return ToolResult(self.name, True, data={"fps": fps})
class SleepTool(Tool):
name = "sleep"
description = (
"Skip ahead by `seconds` (alias of `wait`, but also marks the span "
"as 'low-info' so the summarizer will not pollute memory)."
)
parameters = {
"type": "object",
"properties": {
"seconds": {"type": "number"},
},
"required": ["seconds"],
}
def run(self, args: Dict[str, Any], ctx: ToolContext) -> ToolResult:
# Disabled: sleep can stride past question_time and cause null
# answers. Use set_sample_rate (locked near question_time) instead.
return ToolResult(
self.name, False,
error="sleep is disabled in OVO-Bench evaluation (it can "
"skip past question_time). Use set_sample_rate.",
)
class DeferTool(Tool):
name = "defer"
description = (
"Postpone answering until streaming reaches time `t` (seconds). "
"Until then the agent will not invoke the model on intermediate "
"frames (they are silently fast-forwarded)."
)
parameters = {
"type": "object",
"properties": {
"t": {"type": "integer"},
},
"required": ["t"],
}
def run(self, args: Dict[str, Any], ctx: ToolContext) -> ToolResult:
# Disabled: defer can stride past question_time and cause null
# answers. Use set_sample_rate (locked near question_time) instead.
return ToolResult(
self.name, False,
error="defer is disabled in OVO-Bench evaluation (it can "
"skip past question_time). Use set_sample_rate.",
)
class ResumeTool(Tool):
name = "resume"
description = "Cancel an active `defer` and resume normal streaming."
parameters = {"type": "object", "properties": {}, "required": []}
def run(self, args: Dict[str, Any], ctx: ToolContext) -> ToolResult:
# Disabled: defer is disabled, so resume has nothing to cancel.
return ToolResult(
self.name, False,
error="resume is disabled (defer is disabled too).",
)
|