Test for image token usage in llama.cpp so we can more appropriately decide how many frames to include

frigate/data_processing/post/review_descriptions.py

@@ -39,6 +39,7 @@ logger = logging.getLogger(__name__)
 
 RECORDING_BUFFER_EXTENSION_PERCENT = 0.10
 MIN_RECORDING_DURATION = 10
+MAX_IMAGE_TOKENS = 24000
 
 
 class ReviewDescriptionProcessor(PostProcessorApi):
@@ -65,9 +66,12 @@ class ReviewDescriptionProcessor(PostProcessorApi):
     ) -> int:
         """Calculate optimal number of frames based on context size, image source, and resolution.
 
-        Token usage varies by resolution: larger images (ultra-wide aspect ratios) use more tokens.
-        Estimates ~1 token per 1250 pixels. Targets 98% context utilization with safety margin.
-        Capped at 20 frames.
+        Per-image token cost is asked of the GenAI provider so providers that know
+        their model's true cost (e.g. llama.cpp can probe the loaded mmproj) can
+        diverge from the default ~1-token-per-1250-pixels heuristic. The frame
+        budget is bounded by both the remaining context window and a fixed
+        MAX_IMAGE_TOKENS ceiling so cheap-per-image models get more frames while
+        expensive-per-image models stay reined in.
         """
         client = self.genai_manager.description_client
 
@@ -105,14 +109,13 @@ class ReviewDescriptionProcessor(PostProcessorApi):
                 width = target_width
                 height = int(target_width / aspect_ratio)
 
-        pixels_per_image = width * height
-        tokens_per_image = pixels_per_image / 1250
+        tokens_per_image = client.estimate_image_tokens(width, height)
         prompt_tokens = 3800
         response_tokens = 300
-        available_tokens = context_size - prompt_tokens - response_tokens
-        max_frames = int(available_tokens / tokens_per_image)
-
-        return min(max(max_frames, 3), 20)
+        context_budget = context_size - prompt_tokens - response_tokens
+        image_token_budget = min(context_budget, MAX_IMAGE_TOKENS)
+        max_frames = int(image_token_budget / tokens_per_image)
+        return max(max_frames, 3)
 
     def process_data(
         self, data: dict[str, Any], data_type: PostProcessDataEnum

frigate/genai/__init__.py

@@ -356,6 +356,14 @@ Guidelines:
         """Get the context window size for this provider in tokens."""
         return 4096
 
+    def estimate_image_tokens(self, width: int, height: int) -> float:
+        """Estimate prompt tokens consumed by a single image of the given dimensions.
+
+        Default heuristic: ~1 token per 1250 pixels. Providers that can measure or
+        know their model's exact image-token cost should override.
+        """
+        return (width * height) / 1250
+
     def embed(
         self,
         texts: list[str] | None = None,

frigate/genai/llama_cpp.py

@@ -42,6 +42,8 @@ class LlamaCppClient(GenAIClient):
     _supports_vision: bool
     _supports_audio: bool
     _supports_tools: bool
+    _image_token_cache: dict[tuple[int, int], int]
+    _text_baseline_tokens: int | None
 
     def _init_provider(self) -> str | None:
         """Initialize the client and query model metadata from the server."""
@@ -52,6 +54,8 @@ class LlamaCppClient(GenAIClient):
         self._supports_vision = False
         self._supports_audio = False
         self._supports_tools = False
+        self._image_token_cache = {}
+        self._text_baseline_tokens = None
 
         base_url = (
             self.genai_config.base_url.rstrip("/")
@@ -272,6 +276,91 @@ class LlamaCppClient(GenAIClient):
             return self._context_size
         return 4096
 
+    def estimate_image_tokens(self, width: int, height: int) -> float:
+        """Probe the llama.cpp server to learn the model's image-token cost at the
+        requested dimensions.
+
+        llama.cpp's image tokenization is a deterministic function of dimensions and
+        the loaded mmproj, so the result is cached per (width, height) for the
+        lifetime of the process. Falls back to the base pixel heuristic if the
+        server is unreachable or the response is malformed.
+        """
+        if self.provider is None:
+            return super().estimate_image_tokens(width, height)
+
+        cached = self._image_token_cache.get((width, height))
+
+        if cached is not None:
+            return cached
+
+        try:
+            baseline = self._probe_baseline_tokens()
+            with_image = self._probe_image_prompt_tokens(width, height)
+            tokens = max(1, with_image - baseline)
+        except Exception as e:
+            logger.debug(
+                "llama.cpp image-token probe failed for %dx%d (%s); using heuristic",
+                width,
+                height,
+                e,
+            )
+            return super().estimate_image_tokens(width, height)
+
+        self._image_token_cache[(width, height)] = tokens
+        logger.debug(
+            "llama.cpp model '%s' uses ~%d tokens for %dx%d images",
+            self.genai_config.model,
+            tokens,
+            width,
+            height,
+        )
+        return tokens
+
+    def _probe_baseline_tokens(self) -> int:
+        """Return prompt_tokens for a minimal text-only request. Cached after first call."""
+        if self._text_baseline_tokens is not None:
+            return self._text_baseline_tokens
+
+        self._text_baseline_tokens = self._probe_prompt_tokens(
+            [{"type": "text", "text": "."}]
+        )
+        return self._text_baseline_tokens
+
+    def _probe_image_prompt_tokens(self, width: int, height: int) -> int:
+        """Return prompt_tokens for a single synthetic image plus minimal text."""
+        img = Image.new("RGB", (width, height), (128, 128, 128))
+        buf = io.BytesIO()
+        img.save(buf, format="JPEG", quality=60)
+        encoded = base64.b64encode(buf.getvalue()).decode("utf-8")
+        return self._probe_prompt_tokens(
+            [
+                {"type": "text", "text": "."},
+                {
+                    "type": "image_url",
+                    "image_url": {"url": f"data:image/jpeg;base64,{encoded}"},
+                },
+            ]
+        )
+
+    def _probe_prompt_tokens(self, content: list[dict[str, Any]]) -> int:
+        """POST a 1-token chat completion and return reported prompt_tokens.
+
+        Uses a generous timeout to absorb a cold model load on the first probe
+        when the server lazily loads models on demand (e.g. llama-swap).
+        """
+        payload = {
+            "model": self.genai_config.model,
+            "messages": [{"role": "user", "content": content}],
+            "max_tokens": 1,
+        }
+        response = requests.post(
+            f"{self.provider}/v1/chat/completions",
+            json=payload,
+            timeout=60,
+        )
+        response.raise_for_status()
+        return int(response.json()["usage"]["prompt_tokens"])
+
     def _build_payload(
         self,
         messages: list[dict[str, Any]],