Refactor genai (#22752)
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions

* Switch to a feature-based roles so it is easier to choose models for different tasks

* Fallback and try llama-swap format

* List models supported by provider

* Cleanup

* Add frontend

* Improve model loading

* Make it possible to update genai without restarting

* Cleanup

* Cleanup

* Mypy
This commit is contained in:
Nicolas Mowen
2026-04-03 17:13:52 -06:00
committed by GitHub
parent bb77a01779
commit 9cb76d0bd9
20 changed files with 363 additions and 140 deletions
+68 -38
View File
@@ -1,15 +1,15 @@
"""GenAI client manager for Frigate.
Manages GenAI provider clients from Frigate config. Configuration is read only
in _update_config(); no other code should read config.genai. Exposes clients
by role: tool_client, vision_client, embeddings_client.
Manages GenAI provider clients from Frigate config. Clients are created lazily
on first access so that providers whose roles are never used (e.g. chat when
no chat feature is active) are never initialized.
"""
import logging
from typing import TYPE_CHECKING, Optional
from frigate.config import FrigateConfig
from frigate.config.camera.genai import GenAIRoleEnum
from frigate.config.camera.genai import GenAIConfig, GenAIRoleEnum
if TYPE_CHECKING:
from frigate.genai import GenAIClient
@@ -21,68 +21,98 @@ class GenAIClientManager:
"""Manages GenAI provider clients from Frigate config."""
def __init__(self, config: FrigateConfig) -> None:
self._tool_client: Optional[GenAIClient] = None
self._vision_client: Optional[GenAIClient] = None
self._embeddings_client: Optional[GenAIClient] = None
self._configs: dict[str, GenAIConfig] = {}
self._role_map: dict[GenAIRoleEnum, str] = {}
self._clients: dict[str, "GenAIClient"] = {}
self.update_config(config)
def update_config(self, config: FrigateConfig) -> None:
"""Build role clients from current Frigate config.genai.
"""Store provider configs and build the role→name mapping.
Called from __init__ and can be called again when config is reloaded.
Each role (tools, vision, embeddings) gets the client for the provider
that has that role in its roles list.
Clients are not created here; they are instantiated lazily on first
access via a role property or list_models().
"""
from frigate.genai import PROVIDERS, load_providers
self._tool_client = None
self._vision_client = None
self._embeddings_client = None
self._configs = {}
self._role_map = {}
self._clients = {}
if not config.genai:
return
load_providers()
for _name, genai_cfg in config.genai.items():
for name, genai_cfg in config.genai.items():
if not genai_cfg.provider:
continue
provider_cls = PROVIDERS.get(genai_cfg.provider)
if not provider_cls:
if genai_cfg.provider not in PROVIDERS:
logger.warning(
"Unknown GenAI provider %s in config, skipping.",
genai_cfg.provider,
)
continue
try:
client = provider_cls(genai_cfg)
except Exception as e:
logger.exception(
"Failed to create GenAI client for provider %s: %s",
genai_cfg.provider,
e,
)
continue
self._configs[name] = genai_cfg
for role in genai_cfg.roles:
if role == GenAIRoleEnum.tools:
self._tool_client = client
elif role == GenAIRoleEnum.vision:
self._vision_client = client
elif role == GenAIRoleEnum.embeddings:
self._embeddings_client = client
self._role_map[role] = name
def _get_client(self, name: str) -> "Optional[GenAIClient]":
"""Return the client for *name*, creating it on first access."""
if name in self._clients:
return self._clients[name]
from frigate.genai import PROVIDERS
genai_cfg = self._configs.get(name)
if not genai_cfg:
return None
if not genai_cfg.provider:
return None
provider_cls = PROVIDERS.get(genai_cfg.provider)
if not provider_cls:
return None
try:
client: "GenAIClient" = provider_cls(genai_cfg)
except Exception as e:
logger.exception(
"Failed to create GenAI client for provider %s: %s",
genai_cfg.provider,
e,
)
return None
self._clients[name] = client
return client
@property
def tool_client(self) -> "Optional[GenAIClient]":
"""Client configured for the tools role (e.g. chat with function calling)."""
return self._tool_client
def chat_client(self) -> "Optional[GenAIClient]":
"""Client configured for the chat role (e.g. chat with function calling)."""
name = self._role_map.get(GenAIRoleEnum.chat)
return self._get_client(name) if name else None
@property
def vision_client(self) -> "Optional[GenAIClient]":
"""Client configured for the vision role (e.g. review descriptions, object descriptions)."""
return self._vision_client
def description_client(self) -> "Optional[GenAIClient]":
"""Client configured for the descriptions role (e.g. review descriptions, object descriptions)."""
name = self._role_map.get(GenAIRoleEnum.descriptions)
return self._get_client(name) if name else None
@property
def embeddings_client(self) -> "Optional[GenAIClient]":
"""Client configured for the embeddings role."""
return self._embeddings_client
name = self._role_map.get(GenAIRoleEnum.embeddings)
return self._get_client(name) if name else None
def list_models(self) -> dict[str, list[str]]:
"""Return available models keyed by config entry name."""
result: dict[str, list[str]] = {}
for name in self._configs:
client = self._get_client(name)
if client:
result[name] = client.list_models()
return result