diff --git a/docs/docs/configuration/genai/config.md b/docs/docs/configuration/genai/config.md
index feb29956b7..bfcb9d1373 100644
--- a/docs/docs/configuration/genai/config.md
+++ b/docs/docs/configuration/genai/config.md
@@ -344,7 +344,7 @@ To start using Gemini, you must first get an API key from [Google AI Studio](htt
1. Navigate to .
- Set **Provider** to `gemini`
- Set **API key** to your Gemini API key (or use an environment variable such as `{FRIGATE_GEMINI_API_KEY}`)
- - Set **Model** to the desired model (e.g., `gemini-2.5-flash`)
+ - Set **Model** to the desired model (e.g., `gemini-3.5-flash-lite`)
@@ -354,12 +354,18 @@ genai:
my_provider:
provider: gemini
api_key: "{FRIGATE_GEMINI_API_KEY}"
- model: gemini-2.5-flash
+ model: gemini-3.5-flash-lite
```
+:::tip
+
+Gemini 3.x models charge a (roughly) flat ~1,100 tokens per image regardless of resolution, so a 180p preview frame costs the same as a 480p recording frame. Set the review [image source](/configuration/genai/genai_review#image-source) to `recordings` to send higher quality frames at no extra cost, as long as recordings are enabled for the camera.
+
+:::
+
:::note
To use a different Gemini-compatible API endpoint, set the `provider_options` with the `base_url` key to your provider's API URL. For example:
diff --git a/frigate/genai/plugins/gemini.py b/frigate/genai/plugins/gemini.py
index 9956f0157d..d55fbeb402 100644
--- a/frigate/genai/plugins/gemini.py
+++ b/frigate/genai/plugins/gemini.py
@@ -2,14 +2,17 @@
import base64
import binascii
+import io
import json
import logging
+import time
from collections.abc import AsyncGenerator
from typing import Any
from google import genai
from google.genai import errors, types
from google.genai.types import FunctionCallingConfigMode
+from PIL import Image
from frigate.config import GenAIProviderEnum
from frigate.genai import GenAIClient, register_genai_provider
@@ -21,6 +24,11 @@ logger = logging.getLogger(__name__)
# under it so the request fails as a log line rather than a 400.
GEMINI_MAX_INLINE_BYTES = 15 * 1024 * 1024
+# Image-token probes run on the enrichment thread, so they get a short timeout,
+# no retries, and a cooldown after failure instead of the client's defaults.
+COUNT_TOKENS_TIMEOUT_MS = 5000
+COUNT_TOKENS_RETRY_INTERVAL = 300
+
def _decode_thought_signature(value: Any) -> bytes | None:
"""Decode a base64-encoded thought_signature carried across conversation turns."""
@@ -93,9 +101,14 @@ class GeminiClient(GenAIClient):
"""Generative AI client for Frigate using Gemini."""
provider: genai.Client
+ _image_token_cache: dict[tuple[int, int], int]
+ _image_token_retry_at: float
def _init_provider(self) -> genai.Client:
"""Initialize the client."""
+ self._image_token_cache = {}
+ self._image_token_retry_at = 0.0
+
# Merge provider_options into HttpOptions
http_options_dict: dict[str, Any] = {
"timeout": int(self.timeout * 1000), # requires milliseconds
@@ -232,6 +245,71 @@ class GeminiClient(GenAIClient):
# Gemini Pro Vision has a 1M token context window
return 1000000
+ def estimate_image_tokens(self, width: int, height: int) -> float:
+ """Ask Gemini how many prompt tokens an image of these dimensions costs.
+
+ Gemini bills images far above the base pixel heuristic (Gemini 3 models
+ charge ~1100 tokens for a 320x180 frame, not ~46), and the cost differs
+ between model generations, so it is measured with the free count_tokens
+ endpoint and cached per (width, height). Falls back to the base heuristic
+ if the request fails, and does not probe again for
+ COUNT_TOKENS_RETRY_INTERVAL seconds.
+ """
+ cached = self._image_token_cache.get((width, height))
+
+ if cached is not None:
+ return cached
+
+ if time.monotonic() < self._image_token_retry_at:
+ return super().estimate_image_tokens(width, height)
+
+ img = Image.new("RGB", (width, height), (128, 128, 128))
+ buf = io.BytesIO()
+ img.save(buf, format="JPEG", quality=60)
+ image = types.Part.from_bytes(data=buf.getvalue(), mime_type="image/jpeg")
+
+ try:
+ baseline = self._count_tokens(["."])
+ with_image = self._count_tokens([".", image])
+ tokens = max(1, with_image - baseline)
+ except Exception as e:
+ self._image_token_retry_at = time.monotonic() + COUNT_TOKENS_RETRY_INTERVAL
+ logger.debug(
+ "Gemini image-token count 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(
+ "Gemini model '%s' uses %d tokens for %dx%d images",
+ self.genai_config.model,
+ tokens,
+ width,
+ height,
+ )
+ return tokens
+
+ def _count_tokens(self, contents: list[Any]) -> int:
+ """Return Gemini's prompt-token count for contents, without generating."""
+ total = self.provider.models.count_tokens(
+ model=self.genai_config.model,
+ contents=contents,
+ config=types.CountTokensConfig(
+ http_options=types.HttpOptions(
+ timeout=COUNT_TOKENS_TIMEOUT_MS,
+ retry_options=types.HttpRetryOptions(attempts=1),
+ )
+ ),
+ ).total_tokens
+
+ if total is None:
+ raise ValueError("count_tokens returned no total")
+
+ return total
+
def chat_with_tools(
self,
messages: list[dict[str, Any]],
diff --git a/frigate/test/test_genai_providers.py b/frigate/test/test_genai_providers.py
index 8ae8fb34f1..f4cdccaac3 100644
--- a/frigate/test/test_genai_providers.py
+++ b/frigate/test/test_genai_providers.py
@@ -431,6 +431,56 @@ class TestGeminiProvider(unittest.TestCase):
self.assertEqual(final["content"], "ok")
+class TestGeminiImageTokens(unittest.TestCase):
+ def _client(self, count_tokens):
+ client = _make_client("gemini", model="gemini-3-flash", api_key="k")
+ client.provider = SimpleNamespace(
+ models=SimpleNamespace(count_tokens=count_tokens)
+ )
+ return client
+
+ def test_measures_image_cost_and_caches_per_size(self):
+ count_tokens = MagicMock(
+ side_effect=lambda model, contents, config: SimpleNamespace(
+ total_tokens=2 if len(contents) == 1 else 1102
+ )
+ )
+ client = self._client(count_tokens)
+
+ self.assertEqual(client.estimate_image_tokens(320, 180), 1100)
+ self.assertEqual(client.estimate_image_tokens(320, 180), 1100)
+ self.assertEqual(count_tokens.call_count, 2)
+
+ image = count_tokens.call_args.kwargs["contents"][1]
+ self.assertEqual(image.inline_data.mime_type, "image/jpeg")
+ self.assertEqual(count_tokens.call_args.kwargs["model"], "gemini-3-flash")
+ http_options = count_tokens.call_args.kwargs["config"].http_options
+ self.assertEqual(http_options.timeout, 5000)
+ self.assertEqual(http_options.retry_options.attempts, 1)
+
+ client.estimate_image_tokens(640, 360)
+ self.assertEqual(count_tokens.call_count, 4)
+
+ def test_failure_falls_back_to_heuristic_and_waits_before_retrying(self):
+ count_tokens = MagicMock(side_effect=RuntimeError("offline"))
+ client = self._client(count_tokens)
+
+ with patch("frigate.genai.plugins.gemini.time.monotonic", return_value=1000):
+ self.assertEqual(client.estimate_image_tokens(320, 180), 320 * 180 / 1250)
+ self.assertEqual(client.estimate_image_tokens(640, 360), 640 * 360 / 1250)
+ self.assertEqual(count_tokens.call_count, 1)
+
+ with patch("frigate.genai.plugins.gemini.time.monotonic", return_value=1300):
+ client.estimate_image_tokens(320, 180)
+ self.assertEqual(count_tokens.call_count, 2)
+
+ def test_missing_total_falls_back_to_heuristic(self):
+ client = self._client(
+ MagicMock(return_value=SimpleNamespace(total_tokens=None))
+ )
+ self.assertEqual(client.estimate_image_tokens(320, 180), 320 * 180 / 1250)
+
+
# ---------------------------------------------------------------------------
# Ollama
# ---------------------------------------------------------------------------