administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-08-02 17:12:16 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Increase mypy coverage and fixes (#22632)

Browse Source
This commit is contained in:
Nicolas Mowen
2026-03-25 09:28:48 -06:00
committed by GitHub
parent 04a2f42d11
commit 80c4ce2b5d
11 changed files with 140 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/genai/gemini.py
+45 -22
View File
@@ -2,10 +2,11 @@
import json
import logging
from typing import Any, Optional
from typing import Any, AsyncGenerator, Optional
from google import genai
from google.genai import errors, types
from google.genai.types import FunctionCallingConfigMode
from frigate.config import GenAIProviderEnum
from frigate.genai import GenAIClient, register_genai_provider
@@ -19,10 +20,10 @@ class GeminiClient(GenAIClient):
provider: genai.Client
def _init_provider(self):
def _init_provider(self) -> genai.Client:
"""Initialize the client."""
# Merge provider_options into HttpOptions
http_options_dict = {
http_options_dict: dict[str, Any] = {
"timeout": int(self.timeout * 1000), # requires milliseconds
"retry_options": types.HttpRetryOptions(
attempts=3,
@@ -54,7 +55,7 @@ class GeminiClient(GenAIClient):
] + [prompt]
try:
# Merge runtime_options into generation_config if provided
generation_config_dict = {"candidate_count": 1}
generation_config_dict: dict[str, Any] = {"candidate_count": 1}
generation_config_dict.update(self.genai_config.runtime_options)
if response_format and response_format.get("type") == "json_schema":
@@ -65,7 +66,7 @@ class GeminiClient(GenAIClient):
response = self.provider.models.generate_content(
model=self.genai_config.model,
contents=contents,
contents=contents, # type: ignore[arg-type]
config=types.GenerateContentConfig(
**generation_config_dict,
),
@@ -78,6 +79,8 @@ class GeminiClient(GenAIClient):
return None
try:
if response.text is None:
return None
description = response.text.strip()
except (ValueError, AttributeError):
# No description was generated
@@ -102,7 +105,7 @@ class GeminiClient(GenAIClient):
"""
try:
# Convert messages to Gemini format
gemini_messages = []
gemini_messages: list[types.Content] = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
@@ -110,7 +113,11 @@ class GeminiClient(GenAIClient):
# Map roles to Gemini format
if role == "system":
# Gemini doesn't have system role, prepend to first user message
if gemini_messages and gemini_messages[0].role == "user":
if (
gemini_messages
and gemini_messages[0].role == "user"
and gemini_messages[0].parts
):
gemini_messages[0].parts[
0
].text = f"{content}\n\n{gemini_messages[0].parts[0].text}"
@@ -136,7 +143,7 @@ class GeminiClient(GenAIClient):
types.Content(
role="function",
parts=[
types.Part.from_function_response(function_response)
types.Part.from_function_response(function_response) # type: ignore[misc,call-arg,arg-type]
],
)
)
@@ -171,19 +178,25 @@ class GeminiClient(GenAIClient):
if tool_choice:
if tool_choice == "none":
tool_config = types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(mode="NONE")
function_calling_config=types.FunctionCallingConfig(
mode=FunctionCallingConfigMode.NONE
)
)
elif tool_choice == "auto":
tool_config = types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(mode="AUTO")
function_calling_config=types.FunctionCallingConfig(
mode=FunctionCallingConfigMode.AUTO
)
)
elif tool_choice == "required":
tool_config = types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(mode="ANY")
function_calling_config=types.FunctionCallingConfig(
mode=FunctionCallingConfigMode.ANY
)
)
# Build request config
config_params = {"candidate_count": 1}
config_params: dict[str, Any] = {"candidate_count": 1}
if gemini_tools:
config_params["tools"] = gemini_tools
@@ -197,7 +210,7 @@ class GeminiClient(GenAIClient):
response = self.provider.models.generate_content(
model=self.genai_config.model,
contents=gemini_messages,
contents=gemini_messages, # type: ignore[arg-type]
config=types.GenerateContentConfig(**config_params),
)
@@ -291,7 +304,7 @@ class GeminiClient(GenAIClient):
messages: list[dict[str, Any]],
tools: Optional[list[dict[str, Any]]] = None,
tool_choice: Optional[str] = "auto",
):
) -> AsyncGenerator[tuple[str, Any], None]:
"""
Stream chat with tools; yields content deltas then final message.
@@ -299,7 +312,7 @@ class GeminiClient(GenAIClient):
"""
try:
# Convert messages to Gemini format
gemini_messages = []
gemini_messages: list[types.Content] = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
@@ -307,7 +320,11 @@ class GeminiClient(GenAIClient):
# Map roles to Gemini format
if role == "system":
# Gemini doesn't have system role, prepend to first user message
if gemini_messages and gemini_messages[0].role == "user":
if (
gemini_messages
and gemini_messages[0].role == "user"
and gemini_messages[0].parts
):
gemini_messages[0].parts[
0
].text = f"{content}\n\n{gemini_messages[0].parts[0].text}"
@@ -333,7 +350,7 @@ class GeminiClient(GenAIClient):
types.Content(
role="function",
parts=[
types.Part.from_function_response(function_response)
types.Part.from_function_response(function_response) # type: ignore[misc,call-arg,arg-type]
],
)
)
@@ -368,19 +385,25 @@ class GeminiClient(GenAIClient):
if tool_choice:
if tool_choice == "none":
tool_config = types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(mode="NONE")
function_calling_config=types.FunctionCallingConfig(
mode=FunctionCallingConfigMode.NONE
)
)
elif tool_choice == "auto":
tool_config = types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(mode="AUTO")
function_calling_config=types.FunctionCallingConfig(
mode=FunctionCallingConfigMode.AUTO
)
)
elif tool_choice == "required":
tool_config = types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(mode="ANY")
function_calling_config=types.FunctionCallingConfig(
mode=FunctionCallingConfigMode.ANY
)
)
# Build request config
config_params = {"candidate_count": 1}
config_params: dict[str, Any] = {"candidate_count": 1}
if gemini_tools:
config_params["tools"] = gemini_tools
@@ -399,7 +422,7 @@ class GeminiClient(GenAIClient):
stream = await self.provider.aio.models.generate_content_stream(
model=self.genai_config.model,
contents=gemini_messages,
contents=gemini_messages, # type: ignore[arg-type]
config=types.GenerateContentConfig(**config_params),
)