GenAI Optimizations (#23006)

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

* Limit based on frames per second

* handle zone case sensitivity

* Improve formatting

* Add observations field so model can build CoT before outputting used fields

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]],