diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 96078dc5fb..5233930417 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -954,10 +954,6 @@ def _collect_protected_skill_names(messages: List[Dict[str, Any]], prune_boundar _CHARS_PER_TOKEN = 4 -# Flat per-image token estimate (realistic ceiling; matches Claude Code's constant). -_IMAGE_TOKEN_ESTIMATE = 1600 -# Same figure in char-budget currency. -_IMAGE_CHAR_EQUIVALENT = _IMAGE_TOKEN_ESTIMATE * _CHARS_PER_TOKEN _SUMMARY_FAILURE_COOLDOWN_SECONDS = 600 # Fallback handoff preserves continuity anchors only, not a transcript copy. @@ -1072,14 +1068,18 @@ def _bullets(items: list[str], limit: int = 8) -> str: def _content_length_for_budget(raw_content: Any) -> int: - """Effective char-length of message content for budgeting: text by length plus ``_IMAGE_CHAR_EQUIVALENT`` per image.""" + """Effective char-length of message content for budgeting: text by length plus the learned + per-image price (``agent.image_token_cost``, same figure the trigger estimator uses) per image.""" if isinstance(raw_content, str): return len(raw_content) if not isinstance(raw_content, list): return len(str(raw_content or "")) + from agent.image_token_cost import current_image_token_cost + + image_chars = current_image_token_cost() * _CHARS_PER_TOKEN # Any text-bearing part counts its text; image_url payload size is irrelevant. return sum( - (_IMAGE_CHAR_EQUIVALENT if _is_image_part(p) else len(p.get("text", "") or "")) if isinstance(p, dict) else len(str(p)) + (image_chars if _is_image_part(p) else len(p.get("text", "") or "")) if isinstance(p, dict) else len(str(p)) for p in raw_content ) diff --git a/agent/image_token_cost.py b/agent/image_token_cost.py new file mode 100644 index 0000000000..1fc5e570ee --- /dev/null +++ b/agent/image_token_cost.py @@ -0,0 +1,135 @@ +"""Per-image token cost learned from the provider's own usage, never from a vendor formula. + +A flat per-image constant is wrong in both directions: a 1920x1080 screenshot costs ~1,100 tokens +on one provider and 4,000+ on a local mmproj model. The provider prices every image exactly on the +request that carries it, so the cost is observable: with a fresh usage anchor (real prompt count of +the previous response), the residual between the next real ``prompt_tokens`` and +``anchor + text-only delta`` is the price of the N images that delta introduced (#70328). + +The learned value is kept per ``model@host`` in ``~/.hermes/cache/image_token_costs.json`` so a new +session starts calibrated, and bound per turn through a ContextVar so every estimator +(preflight trigger, tail-budget walk, gateway hygiene) prices images the same way. +""" + +from __future__ import annotations + +import contextlib +import logging +from contextvars import ContextVar +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +DEFAULT_IMAGE_TOKEN_COST = 1500 +# Observations outside this band are text-estimate noise, not an image price. +_MIN_PLAUSIBLE, _MAX_PLAUSIBLE = 64, 32_768 +_EMA_ALPHA = 0.5 + +_image_cost_var: ContextVar[Optional[int]] = ContextVar("hermes_image_token_cost", default=None) +_LEARNED: Dict[str, int] = {} +_LOADED = False + + +def _cache_path(): + from agent.model_metadata import _cache_file + + return _cache_file("image_token_costs.json") + + +def _key(model: Any, base_url: Any) -> str: + from utils import base_url_hostname + + return f"{model or ''}@{base_url_hostname(base_url or '') or ''}" + + +def _load() -> None: + global _LOADED + if _LOADED: + return + _LOADED = True + from agent.model_metadata import _load_json_dict + + for k, v in _load_json_dict(_cache_path()).items(): + if isinstance(v, int) and _MIN_PLAUSIBLE <= v <= _MAX_PLAUSIBLE: + _LEARNED[k] = v + + +def learned_image_token_cost(model: Any, base_url: Any) -> int: + """Learned per-image cost for ``model@host``, else the flat default.""" + _load() + return _LEARNED.get(_key(model, base_url), DEFAULT_IMAGE_TOKEN_COST) + + +def current_image_token_cost() -> int: + """Per-image cost bound for the running turn (see :func:`image_cost_context`), else the default.""" + bound = _image_cost_var.get() + return bound if bound is not None else DEFAULT_IMAGE_TOKEN_COST + + +@contextlib.contextmanager +def image_cost_context(cost: Optional[int]): + token = _image_cost_var.set(cost) + try: + yield + finally: + _image_cost_var.reset(token) + + +def bind_image_token_cost(agent: Any) -> None: + """Bind the agent's learned per-image cost to the current context for the rest of the turn.""" + _image_cost_var.set(learned_image_token_cost(getattr(agent, "model", None), getattr(agent, "base_url", None))) + + +def count_images(messages: List[Dict[str, Any]]) -> int: + from agent.model_metadata import _count_image_tokens + + return sum(_count_image_tokens(m, 1) for m in messages if isinstance(m, dict)) + + +def calibrate_from_usage(agent: Any, messages: List[Dict[str, Any]], prompt_tokens: Any) -> Optional[int]: + """Learn the per-image cost from the response that just priced ``messages``. + + Requires the PREVIOUS anchor (real count of the prior request) to still match: the residual + ``prompt_tokens - (anchor + text-only delta)`` is then the provider's price for the images the + delta introduced. Returns the new learned cost, or None when this response teaches nothing + (no anchor, no new images, implausible residual).""" + from agent.usage_anchor import anchored_context_tokens + + anchor = getattr(agent, "_usage_anchor", None) + try: + real = int(prompt_tokens or 0) + except (TypeError, ValueError): + return None + if real <= 0 or not isinstance(anchor, dict) or not isinstance(messages, list): + return None + base_count = int(anchor.get("base_count") or 0) + delta = messages[base_count:] + if delta and isinstance(delta[0], dict) and delta[0].get("role") == "assistant": + delta = delta[1:] + n_images = count_images(delta) + if n_images <= 0: + return None + with image_cost_context(0): + text_only = anchored_context_tokens(messages, anchor) + if text_only is None: + return None + per_image = (real - text_only) // n_images + if not _MIN_PLAUSIBLE <= per_image <= _MAX_PLAUSIBLE: + return None + key = _key(getattr(agent, "model", None), getattr(agent, "base_url", None)) + _load() + prior = _LEARNED.get(key) + learned = per_image if prior is None else int(prior + _EMA_ALPHA * (per_image - prior)) + _LEARNED[key] = learned + _image_cost_var.set(learned) + try: + from utils import atomic_json_write + + atomic_json_write(_cache_path(), dict(_LEARNED), indent=0, separators=(",", ":")) + except Exception: + logger.debug("image token cost persist failed", exc_info=True) + logger.info( + "Image token cost calibrated from provider usage: %s images priced %s tokens each (learned %s for %s)", + n_images, f"{per_image:,}", f"{learned:,}", key, + ) + return learned diff --git a/agent/model_metadata.py b/agent/model_metadata.py index db66bf1a33..d24df995f1 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -1977,15 +1977,18 @@ def estimate_tokens_rough(text: str) -> int: def estimate_messages_tokens_rough(messages: List[Dict[str, Any]], *, charge_stale_thinking: bool = True) -> int: - """Rough token estimate for a message list (pre-flight only). Images cost a flat ~1500 tokens - each rather than their base64 length. ``charge_stale_thinking=False`` mirrors the tail-budget + """Rough token estimate for a message list (pre-flight only). Images cost the per-image price + learned from provider usage (``agent.image_token_cost``; flat default before calibration) + rather than their base64 length. ``charge_stale_thinking=False`` mirrors the tail-budget walk (``context_compressor._estimate_msg_budget_tokens``): on non-echo routes stale reasoning rides the wire only for the NEWEST assistant turn, so excluding it keeps the compaction TRIGGER in the same size class as the walk — otherwise reasoning-heavy sessions fire preflight forever.""" - _IMAGE_TOKEN_COST = 1500 + from agent.image_token_cost import current_image_token_cost + + image_cost = current_image_token_cost() if not charge_stale_thinking: messages = _strip_stale_thinking_for_estimate(messages) - return sum(_estimate_message_tokens_cached(msg, _IMAGE_TOKEN_COST) for msg in messages) + return sum(_estimate_message_tokens_cached(msg, image_cost) for msg in messages) # Thinking-text keys replayed for at most the newest assistant turn on non-echo routes — must stay @@ -2020,7 +2023,7 @@ def _strip_stale_thinking_for_estimate(messages: List[Dict[str, Any]]) -> List[D # estimate. Because the api_messages build shallow-copies history dicts each iteration, the copies share the # same content strings — so unchanged history messages hit the memo even though the outer dicts are fresh # objects every turn. -_MSG_TOKENS_CACHE: Dict[Any, Tuple[list, int]] = {} +_MSG_TOKENS_CACHE: Dict[Any, Tuple[list, int, int]] = {} # pins, text tokens, image count _MSG_TOKENS_CACHE_MAX = 4096 @@ -2041,19 +2044,23 @@ def _msg_fingerprint(value: Any, pins: list) -> Any: def _estimate_message_tokens_cached(msg: Any, image_cost: int) -> int: - def _compute() -> int: - return _estimate_message_tokens_without_images(msg) + _count_image_tokens(msg, image_cost) + """Text tokens + images x ``image_cost``; the memo holds text and image COUNT so a recalibrated + per-image price re-prices cached rows without invalidating them.""" + def _compute() -> Tuple[int, int]: + return _estimate_message_tokens_without_images(msg), _count_image_tokens(msg, 1) try: pins: list = [] key = _msg_fingerprint(msg, pins) hash(key) except Exception: - return _compute() + text, images = _compute() + return text + images * image_cost cached = _MSG_TOKENS_CACHE.get(key) if cached is not None: - return cached[1] - tokens = _compute() - _MSG_TOKENS_CACHE[key] = (pins, tokens) + return cached[1] + cached[2] * image_cost + text, images = _compute() + tokens = text + images * image_cost + _MSG_TOKENS_CACHE[key] = (pins, text, images) while len(_MSG_TOKENS_CACHE) > _MSG_TOKENS_CACHE_MAX: try: _MSG_TOKENS_CACHE.pop(next(iter(_MSG_TOKENS_CACHE))) diff --git a/agent/turn_context.py b/agent/turn_context.py index 06b6cb7e38..6dfad44e56 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -23,6 +23,7 @@ from agent.memory_manager import build_memory_context_block from agent.memory_provider import is_trivial_prompt from agent.message_metadata import append_message, stamp_message_timestamp from agent.model_metadata import estimate_messages_tokens_rough, estimate_request_tokens_rough +from agent.image_token_cost import bind_image_token_cost from agent.usage_anchor import anchored_context_tokens, restore_usage_anchor logger = logging.getLogger(__name__) @@ -834,6 +835,8 @@ def build_turn_context( persist_user_platform_id, persist_user_display_kind, persist_user_display_metadata, ) _hydrate_from_history(agent, conversation_history) + # Every estimator this turn prices images at the cost learned from this model's real usage. + bind_image_token_cost(agent) # Append the user message now that close persistence is safe. append_message(messages, user_msg) current_turn_user_idx = len(messages) - 1 diff --git a/agent/turn_usage.py b/agent/turn_usage.py index 4dfac5ca37..f3c43a690c 100644 --- a/agent/turn_usage.py +++ b/agent/turn_usage.py @@ -15,6 +15,7 @@ from contextlib import suppress from dataclasses import dataclass from typing import Any, Dict, List +from agent.image_token_cost import calibrate_from_usage from agent.usage_anchor import capture_usage_anchor, set_usage_anchor from agent.usage_pricing import estimate_usage_cost, normalize_usage @@ -119,6 +120,9 @@ def record_response_usage( # transcript (main-loop ONLY; MoA uses pre-fold aggregator usage). The display meter # anchors on the turn's FIRST response: later same-turn responses inflate # prompt_tokens with replayed thinking. Display-only; compression math uses real usage. + # The provider just priced this request exactly: if the delta since the previous anchor + # introduced images, the residual is their real per-image cost (learned before re-anchoring). + calibrate_from_usage(agent, messages, aggregator_usage.prompt_tokens) _new_anchor = capture_usage_anchor( aggregator_usage.prompt_tokens, aggregator_usage.output_tokens, messages ) diff --git a/evals/token_accounting/ab_image_cost_calibration.py b/evals/token_accounting/ab_image_cost_calibration.py new file mode 100644 index 0000000000..080c32f4ae --- /dev/null +++ b/evals/token_accounting/ab_image_cost_calibration.py @@ -0,0 +1,189 @@ +"""A/B probe: does the compaction trigger learn the real per-image cost from provider usage? (#70328) + +Real ``AIAgent.run_conversation`` against a local fake chat-completions server whose +``usage.prompt_tokens`` prices every image at ``IMAGE_REAL`` tokens (a multimodal local model: +several thousand per screenshot) while the flat default is 1,500. A GUI loop appends one +screenshot per turn on a small window; the question is whether compaction fires BEFORE the real +prompt crosses the provider window (the fake returns a context-overflow 400 past it, like +llama.cpp), and what the estimator believes when it does. + + python evals/token_accounting/ab_image_cost_calibration.py --out /tmp/result.json +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import tempfile +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +CONTEXT_LENGTH = 65_536 +THRESHOLD = 40_000 +IMAGE_REAL = 4_000 +TEXT_PER_TURN = 400 # real tokens of text the provider sees per turn (system + user + reply) +TURNS = 14 + + +def _count_images(messages) -> int: + n = 0 + for m in messages: + c = m.get("content") + if isinstance(c, list): + n += sum(1 for p in c if isinstance(p, dict) and p.get("type") in ("image_url", "image", "input_image")) + return n + + +class _FakeVisionChat: + """Prices a request as text_turns*TEXT_PER_TURN + images*IMAGE_REAL; 400s past the window.""" + + def __init__(self) -> None: + self.requests: list[dict] = [] + self.overflows = 0 + server = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_a): # noqa: D401 + pass + + def do_POST(self): + n = int(self.headers.get("content-length", 0)) + body = json.loads(self.rfile.read(n) or b"{}") + msgs = body.get("messages") or [] + server.requests.append(body) + images = _count_images(msgs) + prompt = len(msgs) * TEXT_PER_TURN // 2 + images * IMAGE_REAL + if os.environ.get("AB_DEBUG"): + sys.stderr.write(f"SERVER msgs={len(msgs)} images={images} prompt={prompt} roles={[m.get('role') for m in msgs]} types={[type(m.get('content')).__name__ for m in msgs]}\n") + if prompt > CONTEXT_LENGTH: + server.overflows += 1 + data = json.dumps({"error": {"message": f"request ({prompt} tokens) exceeds the available context size ({CONTEXT_LENGTH} tokens)", "type": "exceed_context_size"}}).encode() + self.send_response(400) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(data))) + self.end_headers() + self.wfile.write(data) + return + usage = {"prompt_tokens": prompt, "completion_tokens": 5, "total_tokens": prompt + 5} + if body.get("stream") is True: + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + for chunk in ( + {"id": "m", "object": "chat.completion.chunk", "model": "m", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}, "finish_reason": None}]}, + {"id": "m", "object": "chat.completion.chunk", "model": "m", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], "usage": usage}, + ): + self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode()) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + return + data = json.dumps({ + "id": "x", "object": "chat.completion", "created": 0, "model": body.get("model", "m"), + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": usage, + }).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.base_url = f"http://127.0.0.1:{self.server.server_address[1]}/v1" + + def close(self) -> None: + self.server.shutdown() + self.server.server_close() + + +def _screenshot_turn(i: int) -> list: + return [{"type": "text", "text": f"screenshot {i}, click next"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64," + "A" * 2_000}}] + + +def run(out_path: str) -> dict: + from run_agent import AIAgent + + wire = _FakeVisionChat() + agent = AIAgent( + api_key="k", base_url=wire.base_url, provider="custom", model="vision-local-ab", + quiet_mode=True, skip_context_files=True, skip_memory=True, enabled_toolsets=[], max_iterations=3, + ) + agent.compression_enabled = True + cc = agent.context_compressor + cc.context_length = CONTEXT_LENGTH + cc.threshold_tokens = THRESHOLD + compress_calls: list[dict] = [] + original = agent._compress_context + + def counting(messages, system_message, **kw): + # Real compaction would need a summarizer; drop everything but the last 2 rows like one. + compress_calls.append({"turn": len(wire.requests), "approx_tokens": int(kw.get("approx_tokens") or 0), + "images_in_history": _count_images(messages), "real_prompt_if_sent": len(messages) * TEXT_PER_TURN // 2 + _count_images(messages) * IMAGE_REAL}) + kept = [{"role": "user", "content": "[compressed summary]"}, {"role": "assistant", "content": "ok"}] + messages[-2:] + sys_prompt = system_message.get("content") if isinstance(system_message, dict) else system_message + return kept, kw.get("active_system_prompt") or sys_prompt + + agent._compress_context = counting # type: ignore[method-assign] + from agent.image_token_cost import current_image_token_cost, learned_image_token_cost + history: list = [] + per_turn = [] + try: + for i in range(TURNS): + r = agent.run_conversation(_screenshot_turn(i), conversation_history=history) + history = r["messages"] + per_turn.append({"turn": i + 1, "completed": bool(r.get("completed")), "images": _count_images(history), + "provider_overflows_so_far": wire.overflows, "compress_calls_so_far": len(compress_calls)}) + finally: + wire.close() + # Compaction sizing: how many screenshots does the protected tail keep under the tail budget? + # With images priced under their real cost the walk protects too many rows and the + # "compacted" request is still over the window (the #70328 "cannot compress further" loop). + from agent.image_token_cost import image_cost_context + walk_history = [m for i in range(40) for m in ({"role": "user", "content": _screenshot_turn(i)}, {"role": "assistant", "content": "ok"})] + tail_budget = int(THRESHOLD * cc.summary_target_ratio) + with image_cost_context(learned_image_token_cost("vision-local-ab", wire.base_url)): + cut = cc._find_tail_cut_by_tokens(walk_history, 0, token_budget=tail_budget) + tail = walk_history[cut:] + tail_real = _count_images(tail) * IMAGE_REAL + len(tail) * TEXT_PER_TURN // 2 + result = { + "head": subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True).stdout.strip(), + "image_real_cost": IMAGE_REAL, "context_length": CONTEXT_LENGTH, "threshold": THRESHOLD, + "learned_image_cost_after": learned_image_token_cost("vision-local-ab", wire.base_url), + "provider_overflows": wire.overflows, "compress_calls": compress_calls, "per_turn": per_turn, + "tail_budget": tail_budget, "tail_images_kept": _count_images(tail), "tail_real_tokens": tail_real, + # The tail walk has an 8-row hard floor, so a screenshot-per-row tail can exceed the budget by + # design; the invariant is that the walk's own accounting of that tail tracks the provider's. + "tail_walk_estimate_error_pct": round(100 * (_count_images(tail) * learned_image_token_cost("vision-local-ab", wire.base_url) + len(tail) * TEXT_PER_TURN // 2 - tail_real) / tail_real, 1), + "verdict": "PASS" if wire.overflows == 0 and compress_calls and abs(learned_image_token_cost("vision-local-ab", wire.base_url) - IMAGE_REAL) / IMAGE_REAL < 0.15 else "FAIL", + } + Path(out_path).write_text(json.dumps(result, indent=2), encoding="utf-8") + return result + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--out", required=True) + args = ap.parse_args() + tmp = Path(tempfile.mkdtemp(prefix="ab-image-cost-")) + os.environ["HERMES_HOME"] = str(tmp / "home") + (tmp / "home").mkdir(parents=True) + # An unknown custom model is treated as non-vision (images replaced by text); declare it. + (tmp / "home" / "config.yaml").write_text("model:\n supports_vision: true\n", encoding="utf-8") + result = run(args.out) + print(json.dumps({k: v for k, v in result.items() if k != "per_turn"}, indent=2)) + return 0 if result["verdict"] == "PASS" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/gateway/run_turn.py b/gateway/run_turn.py index ba8cf52793..f6ed37af2a 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -621,19 +621,22 @@ class GatewayTurnMixin: # row (real count + delta of what was appended since, survives gateway restarts), else the # rough estimate (runs 30-50% high, which only fires hygiene early — safe). Do NOT compensate # with a threshold multiplier. + from agent.image_token_cost import image_cost_context, learned_image_token_cost _anchored = None - if session_entry.last_prompt_tokens <= 0: - from agent.usage_anchor import persisted_anchor_tokens - _session_db = getattr(self, "_session_db", None) - _anchored = persisted_anchor_tokens( - getattr(_session_db, "_db", _session_db), session_entry.session_id, history, - ) - if session_entry.last_prompt_tokens > 0: - _approx_tokens, _token_source = session_entry.last_prompt_tokens, "actual" - elif _anchored is not None: - _approx_tokens, _token_source = _anchored, "anchored" - else: - _approx_tokens, _token_source = estimate_messages_tokens_rough(history), "estimated" + # Images in any local delta/estimate are priced at the cost learned from this model's usage. + with image_cost_context(learned_image_token_cost(hs.model, hs.base_url)): + if session_entry.last_prompt_tokens <= 0: + from agent.usage_anchor import persisted_anchor_tokens + _session_db = getattr(self, "_session_db", None) + _anchored = persisted_anchor_tokens( + getattr(_session_db, "_db", _session_db), session_entry.session_id, history, + ) + if session_entry.last_prompt_tokens > 0: + _approx_tokens, _token_source = session_entry.last_prompt_tokens, "actual" + elif _anchored is not None: + _approx_tokens, _token_source = _anchored, "anchored" + else: + _approx_tokens, _token_source = estimate_messages_tokens_rough(history), "estimated" # Hard safety valve: force compression at an extreme message count regardless of tokens, # breaking the disconnect → no token data → no compression spiral. 5000 clears 1M+ sessions. diff --git a/tests/agent/test_compressor_image_tokens.py b/tests/agent/test_compressor_image_tokens.py index 323ca3c134..924af658c7 100644 --- a/tests/agent/test_compressor_image_tokens.py +++ b/tests/agent/test_compressor_image_tokens.py @@ -9,12 +9,8 @@ creative workflows that iterate on images across many turns. from __future__ import annotations -from agent.context_compressor import ( - _CHARS_PER_TOKEN, - _IMAGE_CHAR_EQUIVALENT, - _IMAGE_TOKEN_ESTIMATE, - _content_length_for_budget, -) +from agent.context_compressor import _CHARS_PER_TOKEN, _content_length_for_budget +from agent.image_token_cost import DEFAULT_IMAGE_TOKEN_COST, image_cost_context class TestContentLengthForBudget: @@ -36,17 +32,13 @@ class TestContentLengthForBudget: - def test_image_estimate_constant_is_reasonable(self): - """Sanity-check the estimate aligns with real provider billing. - - Anthropic ≈ width*height/750 → ~1600 for 1000×1200. - OpenAI GPT-4o high-detail 2048×2048 ≈ 1445. - Gemini 258/tile × 6 tiles for a 2048×2048 ≈ 1548. - Anything in the 800-2000 range is defensible. Enforce bounds so an - accidental edit doesn't drop it to e.g. 16. - """ - assert 800 <= _IMAGE_TOKEN_ESTIMATE <= 2500 - assert _IMAGE_CHAR_EQUIVALENT == _IMAGE_TOKEN_ESTIMATE * _CHARS_PER_TOKEN + def test_image_priced_at_the_learned_cost(self): + """The budget walk charges each image at the per-image price learned from provider usage + (the same figure the trigger estimator uses), falling back to the flat default.""" + content = [{"type": "text", "text": "look"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}] + assert _content_length_for_budget(content) == 4 + DEFAULT_IMAGE_TOKEN_COST * _CHARS_PER_TOKEN + with image_cost_context(4_000): + assert _content_length_for_budget(content) == 4 + 4_000 * _CHARS_PER_TOKEN class TestTokenBudgetWithImages: diff --git a/tests/agent/test_image_token_cost.py b/tests/agent/test_image_token_cost.py new file mode 100644 index 0000000000..4a20342361 --- /dev/null +++ b/tests/agent/test_image_token_cost.py @@ -0,0 +1,65 @@ +"""Per-image token cost learned from provider usage (agent/image_token_cost.py, #70328). + +A flat per-image constant undercounts multimodal local models 2-4x (a GUI loop then hits +provider 400s before compaction can fire) and overcounts providers that downscale. The +provider prices every image exactly on the request that carries it, so the residual between +the real prompt count and ``anchor + text-only delta`` teaches the per-image cost. +""" + +from types import SimpleNamespace + +from agent import image_token_cost as itc +from agent.model_metadata import estimate_messages_tokens_rough +from agent.usage_anchor import anchored_context_tokens, capture_usage_anchor + + +def _img(): + return {"role": "user", "content": [{"type": "text", "text": "look"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64," + "A" * 4000}}]} + + +def _agent(anchor): + return SimpleNamespace(_usage_anchor=anchor, model="vision-local", base_url="http://127.0.0.1:8080/v1") + + +def _isolate(monkeypatch, tmp_path): + monkeypatch.setattr(itc, "_LEARNED", {}) + monkeypatch.setattr(itc, "_LOADED", True) + monkeypatch.setattr(itc, "_cache_path", lambda: tmp_path / "image_token_costs.json") + + +def test_residual_on_an_image_delta_teaches_the_per_image_cost(monkeypatch, tmp_path): + """Two screenshots appended since the last real reading; the provider reports 8,000 tokens + more than the text-only projection -> 4,000 per image, and every estimator prices images at + that from now on (trigger and budget walk read the same bound value).""" + _isolate(monkeypatch, tmp_path) + history = [{"role": "user", "content": "start"}, {"role": "assistant", "content": "ok"}] + anchor = capture_usage_anchor(10_000, 5, history) + history += [_img(), {"role": "assistant", "content": "looking"}, _img()] + agent = _agent(anchor) + with itc.image_cost_context(0): + text_only = anchored_context_tokens(history, anchor) + with itc.image_cost_context(None): + learned = itc.calibrate_from_usage(agent, history, text_only + 2 * 4_000) + assert learned == 4_000 + assert itc.current_image_token_cost() == 4_000 + assert estimate_messages_tokens_rough([_img()]) >= 4_000 + # Persisted per model@host: a fresh process starts calibrated. + monkeypatch.setattr(itc, "_LEARNED", {}) + monkeypatch.setattr(itc, "_LOADED", False) + assert itc.learned_image_token_cost("vision-local", "http://127.0.0.1:8080/v1") == 4_000 + assert itc.learned_image_token_cost("other-model", "http://127.0.0.1:8080/v1") == itc.DEFAULT_IMAGE_TOKEN_COST + + +def test_nothing_learned_without_images_or_anchor(monkeypatch, tmp_path): + """Text-only deltas, missing anchors and implausible residuals teach nothing: a text-estimate + error must never be mistaken for an image price.""" + _isolate(monkeypatch, tmp_path) + history = [{"role": "user", "content": "start"}, {"role": "assistant", "content": "ok"}] + anchor = capture_usage_anchor(10_000, 5, history) + history += [{"role": "user", "content": "no image here"}] + assert itc.calibrate_from_usage(_agent(anchor), history, 50_000) is None + history += [_img()] + assert itc.calibrate_from_usage(_agent(None), history, 50_000) is None + assert itc.calibrate_from_usage(_agent(anchor), history, 10_001) is None # residual < plausible floor + assert itc._LEARNED == {} diff --git a/website/docs/developer-guide/context-compression-and-caching.md b/website/docs/developer-guide/context-compression-and-caching.md index 309c552ddb..63aa3fff2e 100644 --- a/website/docs/developer-guide/context-compression-and-caching.md +++ b/website/docs/developer-guide/context-compression-and-caching.md @@ -96,6 +96,15 @@ provider-proven overflow all compress immediately. Opaque provider blobs (`encrypted_content` on Codex reasoning / compaction items) contribute 0 to every local estimate; only real usage ever prices them. +Images are priced at the per-image cost **learned from the provider's usage** +(`agent/image_token_cost.py`), not a vendor formula: on a response whose delta +since the previous anchor introduced N images, the residual between the real +`prompt_tokens` and the text-only projection is N × the provider's price. The +value is kept per `model@host` in `~/.hermes/cache/image_token_costs.json` and +bound per turn so the trigger estimator, the tail-budget walk and gateway +hygiene all use the same figure. Before the first vision turn a flat 1,500 +default applies. + #### Failure cooldown and provider-proven overflow A failed or stalled summary attempt arms a per-session **failure cooldown**