diff --git a/backend/app/models/llm/constants.py b/backend/app/models/llm/constants.py
index a731b00a7..2b5209cf4 100644
--- a/backend/app/models/llm/constants.py
+++ b/backend/app/models/llm/constants.py
@@ -176,7 +176,7 @@ def normalize_bcp47_language(value: str) -> str:
DEFAULT_STT_MODEL = "gemini-2.5-pro"
-DEFAULT_TTS_MODEL = "gemini-3.1-flash-tts-preview"
+DEFAULT_TTS_MODEL = "gemini-3.8-flash-tts"
DEFAULT_TTS_VOICE = "Kore"
DEFAULT_RAG_MODEL = "gpt-4o"
@@ -199,21 +199,24 @@ def normalize_bcp47_language(value: str) -> str:
DEFAULT_ELEVENLABS_STT_MODEL = "scribe_v2"
DEFAULT_ELEVENLABS_TTS_MODEL = "eleven_v3"
-# BCP-47 to language tag -> Gemini ISO 639-1 code (Indic + English)
+# Kaapi BCP-47 tag -> Gemini TTS language. Full locales, not bare ISO codes:
+# en-IN keeps the Indian accent where "en" does not. Unlisted tags fall back
+# to auto-detect.
BCP47_LOCALE_TO_GEMINI_LANG: dict[str, str] = {
- "en-IN": "en",
- "hi-IN": "hi",
- "bn-IN": "bn",
- "ta-IN": "ta",
- "te-IN": "te",
- "mr-IN": "mr",
- "gu-IN": "gu",
- "kn-IN": "kn",
- "ml-IN": "ml",
- "pa-IN": "pa",
- "od-IN": "or",
- # "as-IN": "as", //not supported by Gemini
- "sd-IN": "sd",
+ "en-IN": "en-IN",
+ "hi-IN": "hi-IN",
+ "bn-IN": "bn-IN",
+ "ta-IN": "ta-IN",
+ "te-IN": "te-IN",
+ "mr-IN": "mr-IN",
+ "gu-IN": "gu-IN",
+ "kn-IN": "kn-IN",
+ "ml-IN": "ml-IN",
+ "pa-IN": "pa-IN",
+ # Kaapi uses the non-standard "od-IN" for Odia; the real tag is "or-IN".
+ "od-IN": "or-IN",
+ # "as-IN": not supported by Gemini
+ "sd-IN": "sd-IN",
}
# BCP-47 language tag → ElevenLabs ISO 639-1 code (Indic + English)
diff --git a/backend/app/models/llm/request.py b/backend/app/models/llm/request.py
index 870d1ef1c..6210833fd 100644
--- a/backend/app/models/llm/request.py
+++ b/backend/app/models/llm/request.py
@@ -89,7 +89,7 @@ class TextLLMParams(ParamSerialization, SQLModel):
"(e.g. {'type': 'enabled', 'budget_tokens': 4096})"
),
)
- thinking_level: Literal["low", "medium", "high"] | None = Field(
+ thinking_level: Literal["minimal", "low", "medium", "high"] | None = Field(
default=None,
description="Google thinking level for thinking-capable Gemini models",
)
diff --git a/backend/app/services/llm/mappers.py b/backend/app/services/llm/mappers.py
index 53fec7741..3543f792a 100644
--- a/backend/app/services/llm/mappers.py
+++ b/backend/app/services/llm/mappers.py
@@ -267,8 +267,9 @@ def map_kaapi_to_google_params(
Supported Mapping:
- model → model
- instructions → instructions (for STT prompts, if available)
- - temperature -> temperature parameter (0-2)
- - top_p / max_output_tokens → same name (text only)
+ - temperature / top_p → not sent (deprecated by Gemini)
+ - max_output_tokens → same name (text only)
+ - language → full Gemini locale via BCP47_LOCALE_TO_GEMINI_LANG; unsupported → auto-detect (TTS only)
- thinking_level → thinking_config.thinking_level (text only)
- output_schema → output_schema, converted to Gemini's shape (text only)
- knowledge_base_ids → FileSearch tool store names (text only)
@@ -296,19 +297,11 @@ def map_kaapi_to_google_params(
google_params["model"] = model
if completion_type == CompletionType.TEXT:
- # Text completion - instructions, temperature, reasoning, knowledge_base_ids
+ # Text completion - instructions, reasoning, knowledge_base_ids
instructions = params.get("instructions")
if instructions:
google_params["instructions"] = instructions
- temperature = params.get("temperature")
- if temperature is not None:
- google_params["temperature"] = temperature
-
- top_p = params.get("top_p")
- if top_p is not None:
- google_params["top_p"] = top_p
-
max_output_tokens = params.get("max_output_tokens")
if max_output_tokens is not None:
google_params["max_output_tokens"] = max_output_tokens
@@ -353,7 +346,7 @@ def map_kaapi_to_google_params(
# Language: Only set if explicitly provided (None/missing = auto-detect)
language = params.get("language")
if language:
- google_lang = BCP47_LOCALE_TO_GEMINI_LANG.get(language) or None
+ google_lang = BCP47_LOCALE_TO_GEMINI_LANG.get(language)
if not google_lang:
warnings.append(
f"Unsupported language '{language}' for Gemini TTS, using auto-detect"
@@ -362,7 +355,7 @@ def map_kaapi_to_google_params(
google_params["language"] = google_lang
elif completion_type == CompletionType.STT:
- # STT mode - instructions, temperature, input_language, output_language, response_format
+ # STT mode - instructions, input_language, output_language, response_format
# Apply smart default for input_language
input_language = params.get("input_language") or "auto"
google_params["input_language"] = input_language
@@ -372,10 +365,6 @@ def map_kaapi_to_google_params(
if instructions:
google_params["instructions"] = instructions
- temperature = params.get("temperature")
- if temperature is not None:
- google_params["temperature"] = temperature
-
output_language = params.get("output_language")
if output_language:
google_params["output_language"] = output_language
diff --git a/backend/app/services/llm/providers/google_aistudio.py b/backend/app/services/llm/providers/google_aistudio.py
index 4ccb9eb76..05b56f08c 100644
--- a/backend/app/services/llm/providers/google_aistudio.py
+++ b/backend/app/services/llm/providers/google_aistudio.py
@@ -1,29 +1,33 @@
-import logging
import base64
-from typing import Any
+import io
+import logging
+import wave
+from typing import Any, cast
+
from google import genai
from google.genai import errors as genai_errors
-from google.genai.types import (
- FileSearch,
- GenerateContentResponse,
- GenerateContentConfig,
- ThinkingConfig,
- SpeechConfig,
- Tool,
- VoiceConfig,
- PrebuiltVoiceConfig,
-)
+# Interactions raises its own error hierarchy, not google.genai.errors; in
+# google-genai 2.29 this private module is the only path that exports it.
+from google.genai._gaos.lib import compat_errors as interactions_errors
+from google.genai.interactions import Interaction
+
+from app.core.audio_utils import (
+ AudioRef,
+ convert_pcm_to_mp3,
+ convert_pcm_to_ogg,
+ pcm_to_wav,
+)
from app.models.llm import (
- NativeCompletionConfig,
+ ImageContent,
LLMCallResponse,
- QueryParams,
LLMResponse,
- Usage,
- TextOutput,
- TextContent,
- ImageContent,
+ NativeCompletionConfig,
PDFContent,
+ QueryParams,
+ TextContent,
+ TextOutput,
+ Usage,
)
from app.models.llm.constants import (
DEFAULT_STT_MODEL,
@@ -32,15 +36,8 @@
DEFAULT_TTS_VOICE,
CompletionType,
)
-from app.models.llm.response import AudioOutput, AudioContent
+from app.models.llm.response import AudioContent, AudioOutput
from app.services.llm.providers.base import BaseProvider, ContentPart, MultiModalInput
-from app.services.llm.mappers import BCP47_LOCALE_TO_GEMINI_LANG
-from app.core.audio_utils import (
- AudioRef,
- convert_pcm_to_mp3,
- convert_pcm_to_ogg,
- pcm_to_wav,
-)
logger = logging.getLogger(__name__)
@@ -64,54 +61,49 @@ def create_client(credentials: dict[str, Any]) -> Any:
@staticmethod
def format_parts(
parts: list[ContentPart],
- ) -> list[dict]:
- items = []
+ ) -> list[dict[str, Any]]:
+ """Render Kaapi content parts as Interactions content items."""
+ items: list[dict[str, Any]] = []
for part in parts:
if isinstance(part, TextContent):
- items.append({"text": part.value})
-
- elif isinstance(part, ImageContent):
- if part.format == "base64":
- items.append(
- {
- "inline_data": {
- "data": part.value,
- "mime_type": part.mime_type,
- }
- }
- )
- else:
- items.append(
- {
- "file_data": {
- "file_uri": part.value,
- "mime_type": part.mime_type,
- "display_name": None,
- }
- }
- )
+ items.append({"type": "text", "text": part.value})
+ continue
+
+ if isinstance(part, ImageContent):
+ item: dict[str, Any] = {"type": "image"}
elif isinstance(part, PDFContent):
- if part.format == "base64":
- items.append(
- {
- "inline_data": {
- "data": part.value,
- "mime_type": part.mime_type,
- }
- }
- )
- else:
- items.append(
- {
- "file_data": {
- "file_uri": part.value,
- "mime_type": part.mime_type,
- "display_name": None,
- }
- }
- )
+ item = {"type": "document"}
+ else:
+ continue
+
+ if part.format == "base64":
+ item["data"] = part.value
+ else:
+ item["uri"] = part.value
+ if part.mime_type:
+ item["mime_type"] = part.mime_type
+ items.append(item)
return items
+ @staticmethod
+ def _extract_usage(interaction: Interaction, provider: str) -> Usage:
+ usage = interaction.usage
+ if usage is None:
+ logger.warning(
+ f"[GoogleAIProvider._extract_usage] Interaction missing usage, using zeros | "
+ f"provider={provider}, interaction_id={interaction.id}"
+ )
+ return Usage(
+ input_tokens=0, output_tokens=0, total_tokens=0, reasoning_tokens=0
+ )
+
+ return Usage(
+ input_tokens=usage.total_input_tokens or 0,
+ output_tokens=usage.total_output_tokens or 0,
+ total_tokens=usage.total_tokens or 0,
+ reasoning_tokens=usage.total_thought_tokens or 0,
+ )
+
def _execute_stt(
self,
completion_config: NativeCompletionConfig,
@@ -147,9 +139,8 @@ def _execute_stt(
instructions = generation_params.get("instructions", "")
input_language = generation_params.get("input_language") or "auto"
output_language = generation_params.get("output_language", "")
- temperature = generation_params.get("temperature") or 0.0
+ thinking_level = generation_params.get("thinking_level") or "low"
- # Build transcription/translation instruction
if input_language == "auto":
lang_instruction = (
"Detect the spoken language automatically and transcribe the audio"
@@ -161,7 +152,6 @@ def _execute_stt(
lang_instruction += f" and translate to {output_language} in the native script of {output_language} and only return transcribed script in {output_language}."
forced_transcription_text = "Only return transcribed text and no other text."
- # Merge user instructions with language instructions
if instructions:
merged_instruction = (
f"{instructions}. {lang_instruction}. {forced_transcription_text}"
@@ -170,43 +160,49 @@ def _execute_stt(
merged_instruction = f"{lang_instruction}. {forced_transcription_text}"
logger.info(
- f"The merged instructions is {merged_instruction} and output language is {output_language} and input language is {input_language}"
+ f"[GoogleAIProvider._execute_stt] Built transcription prompt | "
+ f"provider={provider}, model={model}, input_language={input_language}, "
+ f"output_language={output_language}"
)
- # Materialize the AudioRef to a temp file so the genai SDK can upload it.
with resolved_input.to_path() as audio_path:
gemini_file = self.client.files.upload(file=audio_path)
- contents = []
- if merged_instruction:
- contents.append(merged_instruction)
- contents.append(gemini_file)
-
- response: GenerateContentResponse = self.client.models.generate_content(
- model=model,
- contents=contents,
- # TODO switch back default thinking configs for reasoning supported models in future
- config=GenerateContentConfig(
- thinking_config=ThinkingConfig(
- include_thoughts=True, thinking_budget=1000
- ),
- temperature=temperature,
+ interaction = cast(
+ Interaction,
+ self.client.interactions.create(
+ model=model,
+ input=[
+ {
+ "type": "user_input",
+ "content": [
+ {"type": "text", "text": merged_instruction},
+ {
+ "type": "audio",
+ "uri": gemini_file.uri,
+ "mime_type": gemini_file.mime_type,
+ },
+ ],
+ }
+ ],
+ generation_config={"thinking_level": thinking_level},
),
)
- # Validate response has required fields
- if not response.response_id:
+ response_id = interaction.id
+ if not response_id or interaction.status != "completed":
error_message = (
- "[GEMINI] STT response is missing a response_id. This indicates "
- "an unexpected upstream payload from Gemini. Retry the request; "
- "if the issue persists, contact Kaapi."
+ f"[GEMINI] STT interaction did not complete (status="
+ f"{interaction.status}, errors={interaction.errors}). Retry the "
+ f"request; if the issue persists, contact Kaapi."
)
logger.warning(
- f"[GoogleAIProvider._execute_stt] {error_message} | provider={provider}, model={model}"
+ f"[GoogleAIProvider._execute_stt] {error_message} | "
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- if not response.text:
+ if not interaction.output_text:
error_message = (
"[GEMINI] STT response is missing transcribed text. Gemini "
"returned an empty result — verify the audio is audible and in "
@@ -215,51 +211,30 @@ def _execute_stt(
)
logger.warning(
f"[GoogleAIProvider._execute_stt] {error_message} | "
- f"provider={provider}, model={model}, response_id={response.response_id}"
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- # Extract usage metadata with null checks
- if response.usage_metadata:
- input_tokens = response.usage_metadata.prompt_token_count or 0
- output_tokens = response.usage_metadata.candidates_token_count or 0
- total_tokens = response.usage_metadata.total_token_count or 0
- reasoning_tokens = response.usage_metadata.thoughts_token_count or 0
- else:
- logger.warning(
- f"[GoogleAIProvider._execute_stt] Response missing usage_metadata, using zeros | provider={provider}"
- )
- input_tokens = 0
- output_tokens = 0
- total_tokens = 0
- reasoning_tokens = 0
-
- # Build response
llm_response = LLMCallResponse(
response=LLMResponse(
- provider_response_id=response.response_id,
- model=response.model_version or model,
+ provider_response_id=response_id,
+ model=interaction.model or model,
provider=provider,
output=TextOutput(
content=TextContent(
- value=response.text, language_code=output_language
+ value=interaction.output_text, language_code=output_language
)
),
),
- usage=Usage(
- input_tokens=input_tokens,
- output_tokens=output_tokens,
- total_tokens=total_tokens,
- reasoning_tokens=reasoning_tokens,
- ),
+ usage=self._extract_usage(interaction, provider),
)
if include_provider_raw_response:
- llm_response.provider_raw_response = response.model_dump()
+ llm_response.provider_raw_response = interaction.model_dump(mode="json")
logger.info(
f"[GoogleAIProvider._execute_stt] Successfully generated STT response | "
- f"request_id={response.response_id}, provider={provider}, model={model}"
+ f"request_id={response_id}, provider={provider}, model={model}"
)
return llm_response, None
@@ -283,7 +258,6 @@ def _execute_tts(
provider = completion_config.provider
generation_params = completion_config.params
- # Validate input is a text string
if not isinstance(resolved_input, str):
error_message = (
f"[KAAPI] TTS validation failed: {provider} TTS requires a text "
@@ -305,54 +279,53 @@ def _execute_tts(
)
return None, error_message
- # Extract params with defaults (language is optional — Gemini auto-detects from script)
model = generation_params.get("model") or DEFAULT_TTS_MODEL
voice = generation_params.get("voice") or DEFAULT_TTS_VOICE
-
+ # Optional: Gemini auto-detects the language from the script when unset.
language = generation_params.get("language")
-
- # Extract optional params
response_format = generation_params.get("response_format", "wav")
- # Extract Gemini-specific params from provider_specific.gemini
provider_specific = generation_params.get("provider_specific", {})
gemini_params = provider_specific.get("gemini", {})
-
director_notes = gemini_params.get("director_notes", "")
- # Build Gemini TTS config
- config_kwargs = {
- "response_modalities": ["AUDIO"],
- "speech_config": SpeechConfig(
- voice_config=VoiceConfig(
- prebuilt_voice_config=PrebuiltVoiceConfig(voice_name=voice)
- ),
- language_code=language,
- ),
- }
+ transcript: dict[str, Any] = {
+ "type": "text",
+ "text": f"{resolved_input}",
+ }
if director_notes:
- config_kwargs["system_instruction"] = director_notes
+ transcript["annotations"] = [
+ {"type": "speech_metadata", "style": director_notes}
+ ]
- config = GenerateContentConfig(**config_kwargs)
- decorated_resolved_input = f"{resolved_input}"
- # Execute TTS
- response: GenerateContentResponse = self.client.models.generate_content(
- model=model, contents=decorated_resolved_input, config=config
+ speech_config: dict[str, Any] = {"voice": voice}
+ if language:
+ speech_config["language"] = language
+
+ interaction = cast(
+ Interaction,
+ self.client.interactions.create(
+ model=model,
+ input=[{"type": "user_input", "content": [transcript]}],
+ generation_config={"speech_config": [speech_config]},
+ response_format={"type": "audio"},
+ ),
)
- if not response.response_id:
+
+ response_id = interaction.id
+ if not response_id or interaction.status != "completed":
error_message = (
- "[GEMINI] TTS response is missing a response_id. This indicates "
- "an unexpected upstream payload from Gemini. Retry the request; "
- "if the issue persists, contact Kaapi."
+ f"[GEMINI] TTS interaction did not complete (status="
+ f"{interaction.status}, errors={interaction.errors}). Retry the "
+ f"request; if the issue persists, contact Kaapi."
)
logger.warning(
- f"[GoogleAIProvider._execute_tts] {error_message} | provider={provider}, model={model}"
+ f"[GoogleAIProvider._execute_tts] {error_message} | "
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- try:
- raw_audio_bytes = response.candidates[0].content.parts[0].inline_data.data
- except (IndexError, AttributeError) as e:
+ if interaction.output_audio is None:
error_message = (
"[GEMINI] Failed to extract audio bytes from TTS response: "
"Gemini was unable to generate audio from the provided input. "
@@ -362,12 +335,11 @@ def _execute_tts(
)
logger.warning(
f"[GoogleAIProvider._execute_tts] {error_message} | "
- f"provider={provider}, model={model}, response_id={response.response_id}, cause={type(e).__name__}",
- exc_info=True,
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- if not raw_audio_bytes:
+ if not interaction.output_audio.data:
error_message = (
"[GEMINI] TTS response is missing generated audio data. This is "
"typically a Gemini server-side error. Wait a minute and retry; "
@@ -375,111 +347,82 @@ def _execute_tts(
)
logger.warning(
f"[GoogleAIProvider._execute_tts] {error_message} | "
- f"provider={provider}, model={model}, response_id={response.response_id}"
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- # Post-process audio format conversion if needed
- # Gemini TTS natively outputs 24kHz 16-bit raw PCM — wrap in WAV container
+ raw_audio_bytes = base64.b64decode(interaction.output_audio.data)
+ # Preview TTS models return headerless 24 kHz PCM (and reject audio/l16),
+ # while 3.8+ return a full WAV; normalise to PCM for pcm_to_wav / convert_pcm_to_*.
+ if raw_audio_bytes.startswith(b"RIFF"):
+ with wave.open(io.BytesIO(raw_audio_bytes), "rb") as wav_file:
+ raw_audio_bytes = wav_file.readframes(wav_file.getnframes())
+
actual_format = "wav"
- wav_bytes = pcm_to_wav(raw_audio_bytes)
- encoded_content = base64.b64encode(wav_bytes).decode("ascii")
-
- if response_format and response_format != "wav":
- # Need to convert from WAV to requested format
- logger.info(
- f"[GoogleAIProvider._execute_tts] Converting audio from WAV to {response_format} | provider={provider}"
- )
-
- if response_format == "mp3":
- converted_bytes, convert_error = convert_pcm_to_mp3(raw_audio_bytes)
- if convert_error:
- error_message = (
- f"[KAAPI] Post-processing failure: unable to convert "
- f"Gemini PCM audio to MP3 ({convert_error}). Falling "
- f"back to WAV is possible by setting response_format='wav'."
- )
- logger.error(
- f"[GoogleAIProvider._execute_tts] {error_message} | "
- f"provider={provider}, model={model}, pcm_bytes={len(raw_audio_bytes)}"
- )
- return None, error_message
- encoded_content = base64.b64encode(converted_bytes or b"").decode(
- "ascii"
+ if response_format == "mp3":
+ converted_bytes, convert_error = convert_pcm_to_mp3(raw_audio_bytes)
+ if convert_error:
+ error_message = (
+ f"[KAAPI] Post-processing failure: unable to convert "
+ f"Gemini PCM audio to MP3 ({convert_error}). Falling "
+ f"back to WAV is possible by setting response_format='wav'."
)
- actual_format = "mp3"
-
- elif response_format == "ogg":
- converted_bytes, convert_error = convert_pcm_to_ogg(raw_audio_bytes)
- if convert_error:
- error_message = (
- f"[KAAPI] Post-processing failure: unable to convert "
- f"Gemini PCM audio to OGG ({convert_error}). Falling "
- f"back to WAV is possible by setting response_format='wav'."
- )
- logger.error(
- f"[GoogleAIProvider._execute_tts] {error_message} | "
- f"provider={provider}, model={model}, pcm_bytes={len(raw_audio_bytes)}"
- )
- return None, error_message
- encoded_content = base64.b64encode(converted_bytes or b"").decode(
- "ascii"
+ logger.error(
+ f"[GoogleAIProvider._execute_tts] {error_message} | "
+ f"provider={provider}, model={model}, pcm_bytes={len(raw_audio_bytes)}"
)
- actual_format = "ogg"
- else:
- logger.warning(
- f"[GoogleAIProvider._execute_tts] Unsupported response_format '{response_format}', returning native WAV | provider={provider}"
+ return None, error_message
+ encoded_content = base64.b64encode(converted_bytes or b"").decode("ascii")
+ actual_format = "mp3"
+
+ elif response_format == "ogg":
+ converted_bytes, convert_error = convert_pcm_to_ogg(raw_audio_bytes)
+ if convert_error:
+ error_message = (
+ f"[KAAPI] Post-processing failure: unable to convert "
+ f"Gemini PCM audio to OGG ({convert_error}). Falling "
+ f"back to WAV is possible by setting response_format='wav'."
+ )
+ logger.error(
+ f"[GoogleAIProvider._execute_tts] {error_message} | "
+ f"provider={provider}, model={model}, pcm_bytes={len(raw_audio_bytes)}"
)
- response_format = "wav"
- logger.info(
- f"[GoogleAIProvider._execute_tts] Audio conversion successful: {actual_format.upper()} ({len(raw_audio_bytes)} bytes) | provider={provider}"
- )
- response_mime_type = f"audio/{response_format}"
-
- # Extract usage metadata
- if response.usage_metadata:
- input_tokens = response.usage_metadata.prompt_token_count or 0
- output_tokens = response.usage_metadata.candidates_token_count or 0
- total_tokens = response.usage_metadata.total_token_count or 0
- reasoning_tokens = response.usage_metadata.thoughts_token_count or 0
+ return None, error_message
+ encoded_content = base64.b64encode(converted_bytes or b"").decode("ascii")
+ actual_format = "ogg"
+
else:
- logger.warning(
- f"[GoogleAIProvider._execute_tts] Response missing usage_metadata, using zeros | provider={provider}"
+ if response_format and response_format != "wav":
+ logger.warning(
+ f"[GoogleAIProvider._execute_tts] Unsupported response_format "
+ f"'{response_format}', returning native WAV | provider={provider}"
+ )
+ encoded_content = base64.b64encode(pcm_to_wav(raw_audio_bytes)).decode(
+ "ascii"
)
- input_tokens = 0
- output_tokens = 0
- total_tokens = 0
- reasoning_tokens = 0
- # Build response
llm_response = LLMCallResponse(
response=LLMResponse(
- provider_response_id=response.response_id,
- model=response.model_version or model,
+ provider_response_id=response_id,
+ model=interaction.model or model,
provider=provider,
- # output=LLMOutput(audio_bytes=audio_bytes, audio_format=actual_format),
output=AudioOutput(
content=AudioContent(
format="base64",
value=encoded_content,
- mime_type=response_mime_type,
+ mime_type=f"audio/{actual_format}",
)
),
),
- usage=Usage(
- input_tokens=input_tokens,
- output_tokens=output_tokens,
- total_tokens=total_tokens,
- reasoning_tokens=reasoning_tokens,
- ),
+ usage=self._extract_usage(interaction, provider),
)
if include_provider_raw_response:
- llm_response.provider_raw_response = response.model_dump()
+ llm_response.provider_raw_response = interaction.model_dump(mode="json")
logger.info(
f"[GoogleAIProvider._execute_tts] Successfully generated TTS response | "
- f"request_id={response.response_id}, provider={provider}, model={model}, audio_size={len(raw_audio_bytes)} bytes"
+ f"request_id={response_id}, provider={provider}, model={model}, audio_size={len(raw_audio_bytes)} bytes"
)
return llm_response, None
@@ -490,122 +433,91 @@ def _execute_text(
resolved_input: str | list[ContentPart] | MultiModalInput,
include_provider_raw_response: bool = False,
) -> tuple[LLMCallResponse | None, str | None]:
- model = completion_config.params.get("model") or DEFAULT_TEXT_MODELS["google"]
+ provider = completion_config.provider
+ params = completion_config.params
+ model = params.get("model") or DEFAULT_TEXT_MODELS["google"]
if isinstance(resolved_input, MultiModalInput):
- gemini_parts = self.format_parts(resolved_input.parts)
- contents = [{"role": "user", "parts": gemini_parts}]
+ content = self.format_parts(resolved_input.parts)
elif isinstance(resolved_input, list):
- gemini_parts = self.format_parts(resolved_input)
- contents = [{"role": "user", "parts": gemini_parts}]
+ content = self.format_parts(resolved_input)
else:
- contents = [{"role": "user", "parts": [{"text": resolved_input}]}]
-
- instructions = completion_config.params.get("instructions", "")
- temperature = completion_config.params.get("temperature", None)
- thinking_level = completion_config.params.get("reasoning", None)
- knowledge_base_ids = completion_config.params.get("knowledge_base_ids", None)
-
- generation_kwargs = {}
+ content = [{"type": "text", "text": resolved_input}]
+
+ instructions = params.get("instructions")
+ # Kaapi configs carry thinking_config.thinking_level; native configs use reasoning.
+ thinking_level = (params.get("thinking_config") or {}).get(
+ "thinking_level"
+ ) or params.get("reasoning")
+ max_output_tokens = params.get("max_output_tokens")
+ knowledge_base_ids = params.get("knowledge_base_ids")
+
+ request: dict[str, Any] = {
+ "model": model,
+ "input": [{"type": "user_input", "content": content}],
+ }
if instructions:
- generation_kwargs["system_instruction"] = instructions
+ request["system_instruction"] = instructions
- if temperature is not None:
- generation_kwargs["temperature"] = temperature
-
- if thinking_level is not None:
- generation_kwargs["thinking_config"] = ThinkingConfig(
- include_thoughts=False, thinking_level=thinking_level
- )
+ generation_config: dict[str, Any] = {}
+ if thinking_level:
+ generation_config["thinking_level"] = thinking_level
+ if max_output_tokens is not None:
+ generation_config["max_output_tokens"] = max_output_tokens
+ if generation_config:
+ request["generation_config"] = generation_config
if knowledge_base_ids:
- generation_kwargs["tools"] = [
- Tool(file_search=FileSearch(file_search_store_names=knowledge_base_ids))
+ request["tools"] = [
+ {"type": "file_search", "file_search_store_names": knowledge_base_ids}
]
- response = self.client.models.generate_content(
- model=model,
- contents=contents,
- config=GenerateContentConfig(**generation_kwargs),
- )
-
- provider = completion_config.provider
+ # create() is typed Interaction | Stream; we never pass stream=True.
+ interaction = cast(Interaction, self.client.interactions.create(**request))
- if not response.response_id:
+ response_id = interaction.id
+ if not response_id or interaction.status != "completed":
error_message = (
- "[GEMINI] Text response is missing a response_id. This "
- "indicates an unexpected upstream payload from Gemini. Retry "
- "the request; if the issue persists, contact Kaapi."
+ f"[GEMINI] Text interaction did not complete (status="
+ f"{interaction.status}, errors={interaction.errors}). This "
+ f"typically means the response was blocked by Gemini's safety "
+ f"filters or failed upstream. Review the prompt, then retry."
)
logger.warning(
- f"[GoogleAIProvider._execute_text] {error_message} | provider={provider}, model={model}"
+ f"[GoogleAIProvider._execute_text] {error_message} | "
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- if not response.text:
- # Gemini commonly returns no text when the response is blocked by
- # safety filters or when the candidate finishes with a non-STOP
- # reason. Surface the finish_reason / block_reason if available so
- # the caller can act on it.
- finish_reason = None
- block_reason = None
- try:
- finish_reason = response.candidates[0].finish_reason
- except (IndexError, AttributeError):
- pass
- try:
- block_reason = response.prompt_feedback.block_reason
- except AttributeError:
- pass
-
+ if not interaction.output_text:
error_message = (
f"[GEMINI] Text response is missing generated content "
- f"(finish_reason={finish_reason}, block_reason={block_reason}). "
- f"This typically means the response was blocked by Gemini's "
- f"safety filters, truncated by token limits, or the model "
- f"returned no candidates. Review the prompt and safety "
- f"settings, then retry."
+ f"(status={interaction.status}). This typically means the "
+ f"model returned no text output — e.g. the response was blocked "
+ f"by Gemini's safety filters or consumed by thinking tokens. "
+ f"Review the prompt and token limits, then retry."
)
logger.warning(
f"[GoogleAIProvider._execute_text] {error_message} | "
- f"provider={provider}, model={model}, response_id={response.response_id}"
+ f"provider={provider}, model={model}, response_id={response_id}"
)
return None, error_message
- if response.usage_metadata:
- input_tokens = response.usage_metadata.prompt_token_count or 0
- output_tokens = response.usage_metadata.candidates_token_count or 0
- total_tokens = response.usage_metadata.total_token_count or 0
- reasoning_tokens = response.usage_metadata.thoughts_token_count or 0
- else:
- logger.warning(
- f"[GoogleAIProvider._execute_text] Response missing usage_metadata, using zeros | provider={completion_config.provider}"
- )
- input_tokens = 0
- output_tokens = 0
- total_tokens = 0
- reasoning_tokens = 0
-
llm_response = LLMCallResponse(
response=LLMResponse(
- provider_response_id=response.response_id,
- model=response.model_version or model,
- provider=completion_config.provider,
- output=TextOutput(content=TextContent(value=response.text)),
- ),
- usage=Usage(
- input_tokens=input_tokens,
- output_tokens=output_tokens,
- total_tokens=total_tokens,
- reasoning_tokens=reasoning_tokens,
+ provider_response_id=response_id,
+ model=interaction.model or model,
+ provider=provider,
+ output=TextOutput(content=TextContent(value=interaction.output_text)),
),
+ usage=self._extract_usage(interaction, provider),
)
if include_provider_raw_response:
- llm_response.provider_raw_response = response.model_dump(mode="json")
+ llm_response.provider_raw_response = interaction.model_dump(mode="json")
logger.info(
f"[GoogleAIProvider._execute_text] Successfully generated text response | "
- f"request_id={response.response_id}, provider={completion_config.provider}, model={model}"
+ f"request_id={response_id}, provider={provider}, model={model}"
)
return llm_response, None
@@ -616,8 +528,9 @@ def execute(
resolved_input: str | list[ContentPart] | MultiModalInput,
include_provider_raw_response: bool = False,
) -> tuple[LLMCallResponse | None, str | None]:
+ provider = completion_config.provider
+ completion_type = completion_config.type
try:
- completion_type = completion_config.type
if completion_type == CompletionType.STT:
return self._execute_stt(
completion_config=completion_config,
@@ -647,7 +560,69 @@ def execute(
)
logger.warning(
f"[GoogleAIProvider.execute] {error_message} | "
- f"provider={completion_config.provider}, type={completion_type}",
+ f"provider={provider}, type={completion_type}",
+ exc_info=True,
+ )
+ return None, error_message
+
+ except interactions_errors.APIStatusError as e:
+ code = e.status_code
+ if code == 429:
+ error_message = (
+ f"[GEMINI] Rate limit / quota exceeded (code: 429): "
+ f"{e.message}. You have hit Gemini's per-minute or per-day "
+ f"quota for this model. Wait at least 1 minute and retry; "
+ f"if the issue persists, request a quota increase from "
+ f"Google or contact Kaapi."
+ )
+ elif code in (401, 403):
+ error_message = (
+ f"[GEMINI] Authentication / permission denied (code: "
+ f"{code}): {e.message}. Verify the Gemini API key is valid, "
+ f"not expired, and has access to the requested model and "
+ f"project."
+ )
+ elif code == 404:
+ error_message = (
+ f"[GEMINI] Resource not found (code: 404): {e.message}. "
+ f"Check that the model name and any referenced IDs in "
+ f"your config are correct and available in your region."
+ )
+ elif code == 400:
+ error_message = (
+ f"[GEMINI] Bad request (code: 400): {e.message}. Review "
+ f"your config parameters and input payload — the request "
+ f"shape, model, or content may be invalid for this Gemini "
+ f"endpoint."
+ )
+ elif code is not None and code >= 500:
+ error_message = (
+ f"[GEMINI] Server error (code: {code}): {e.message}. This "
+ f"is typically transient (Gemini overloaded, internal "
+ f"error, or deadline exceeded) — retry in a few seconds. "
+ f"If the issue persists, contact Kaapi."
+ )
+ else:
+ error_message = (
+ f"[GEMINI] Client error (code: {code}): {e.message}. "
+ f"Review the request configuration; if the issue persists, "
+ f"contact Kaapi."
+ )
+ logger.warning(
+ f"[GoogleAIProvider.execute] {error_message} | "
+ f"provider={provider}, type={completion_type}",
+ exc_info=True,
+ )
+ return None, error_message
+
+ except interactions_errors.APIConnectionError as e:
+ error_message = (
+ f"[KAAPI] Could not reach Gemini ({type(e).__name__}): {e}. "
+ f"Retry the request; if the issue persists, contact Kaapi."
+ )
+ logger.error(
+ f"[GoogleAIProvider.execute] {error_message} | "
+ f"provider={provider}, type={completion_type}",
exc_info=True,
)
return None, error_message
@@ -747,7 +722,16 @@ def execute(
f"failure. Contact Kaapi if the issue persists."
)
logger.error(
- f"[GoogleAIProvider.execute] {error_message} | provider={completion_config.provider}",
+ f"[GoogleAIProvider.execute] {error_message} | provider={provider}",
exc_info=True,
)
return None, error_message
+
+ error_message = (
+ f"[KAAPI] Unsupported completion type '{completion_type}' for "
+ f"Gemini. Use one of: text, stt, tts."
+ )
+ logger.warning(
+ f"[GoogleAIProvider.execute] {error_message} | provider={provider}"
+ )
+ return None, error_message
diff --git a/backend/app/services/llm/providers/google_gcp.py b/backend/app/services/llm/providers/google_gcp.py
index 172d24953..d0155136f 100644
--- a/backend/app/services/llm/providers/google_gcp.py
+++ b/backend/app/services/llm/providers/google_gcp.py
@@ -32,7 +32,6 @@
DEFAULT_TTS_VOICE,
CompletionType,
)
-from app.models.llm.request import ImageContent, PDFContent
from app.models.llm.response import AudioContent, AudioOutput
from app.services.llm.providers.base import BaseProvider, ContentPart, MultiModalInput
@@ -313,106 +312,6 @@ def _extract_usage(data: dict) -> Usage:
reasoning_tokens=reasoning_tokens,
)
- @staticmethod
- def _format_content_parts(parts: list[ContentPart]) -> list[dict]:
- """Render Kaapi content parts as GCP REST `parts` entries (camelCase
- keys, matching the generateContent wire format used elsewhere in
- this file — not the genai SDK's snake_case Python bindings)."""
- items = []
- for part in parts:
- if isinstance(part, TextContent):
- items.append({"text": part.value})
- elif isinstance(part, (ImageContent, PDFContent)):
- if part.format == "base64":
- items.append(
- {"inlineData": {"mimeType": part.mime_type, "data": part.value}}
- )
- else:
- items.append(
- {
- "fileData": {
- "mimeType": part.mime_type,
- "fileUri": part.value,
- }
- }
- )
- return items
-
- def _execute_text(
- self,
- completion_config: NativeCompletionConfig,
- resolved_input: str | list[ContentPart] | MultiModalInput,
- include_provider_raw_response: bool = False,
- ) -> tuple[LLMCallResponse | None, str | None]:
- """Execute a text completion via Google GCP generateContent."""
- provider = completion_config.provider
- params = completion_config.params
-
- if isinstance(resolved_input, MultiModalInput):
- gemini_parts = self._format_content_parts(resolved_input.parts)
- elif isinstance(resolved_input, list):
- gemini_parts = self._format_content_parts(resolved_input)
- else:
- gemini_parts = [{"text": resolved_input}]
-
- model = params.get("model") or DEFAULT_TEXT_MODELS["google"]
- instructions = params.get("instructions")
- temperature = params.get("temperature")
-
- payload: dict[str, Any] = {
- "contents": [{"role": "user", "parts": gemini_parts}],
- }
- if instructions:
- payload["systemInstruction"] = {"parts": [{"text": instructions}]}
-
- generation_config: dict[str, Any] = {}
- if temperature is not None:
- generation_config["temperature"] = temperature
- if generation_config:
- payload["generationConfig"] = generation_config
-
- data, err = self._post(
- model, payload, log_context=f"provider={provider}, type=text"
- )
- if err:
- return None, err
-
- try:
- text = data["candidates"][0]["content"]["parts"][0]["text"]
- except (KeyError, IndexError, TypeError):
- error_message = (
- "[GOOGLE_GCP] Text response is missing generated content. "
- "Google GCP returned a 200 response but the expected "
- "candidates[0].content.parts[0].text path is absent — this "
- "typically means the response was blocked by safety filters "
- "or truncated. Review the prompt and safety settings, then "
- "retry."
- )
- logger.warning(
- f"[GoogleGCPProvider._execute_text] {error_message} | "
- f"provider={provider}, model={model}, response_id={data.get('responseId')}"
- )
- return None, error_message
-
- llm_response = LLMCallResponse(
- response=LLMResponse(
- provider_response_id=data.get("responseId")
- or f"google-gcp-{uuid.uuid4().hex}",
- model=data.get("modelVersion") or model,
- provider=provider,
- output=TextOutput(content=TextContent(value=text)),
- ),
- usage=self._extract_usage(data),
- )
-
- if include_provider_raw_response:
- llm_response.provider_raw_response = data
-
- logger.info(
- f"[GoogleGCPProvider._execute_text] Generated text response | provider={provider}, model={model}"
- )
- return llm_response, None
-
def _execute_stt(
self,
completion_config: NativeCompletionConfig,
@@ -492,8 +391,8 @@ def _execute_stt(
instructions = params.get("instructions")
input_language = params.get("input_language") or "auto"
output_language = params.get("output_language")
- temperature = params.get("temperature")
max_output_tokens = params.get("max_output_tokens") or 2048
+ thinking_level = params.get("thinking_level") or "low"
# Build transcription/translation instruction
if input_language == "auto":
@@ -515,9 +414,10 @@ def _execute_stt(
else:
prompt = f"{lang_instruction}. {forced}"
- generation_config: dict[str, Any] = {"maxOutputTokens": max_output_tokens}
- if temperature is not None:
- generation_config["temperature"] = temperature
+ generation_config: dict[str, Any] = {
+ "maxOutputTokens": max_output_tokens,
+ "thinkingConfig": {"thinkingLevel": thinking_level},
+ }
payload = {
"contents": [
@@ -808,12 +708,14 @@ def _execute_text(
parts = [{"text": resolved_input}]
instructions = params.get("instructions")
- temperature = params.get("temperature")
max_output_tokens = params.get("max_output_tokens")
+ thinking_level = (params.get("thinking_config") or {}).get(
+ "thinking_level"
+ ) or params.get("reasoning")
generation_config: dict[str, Any] = {}
- if temperature is not None:
- generation_config["temperature"] = temperature
+ if thinking_level:
+ generation_config["thinkingConfig"] = {"thinkingLevel": thinking_level}
if max_output_tokens is not None:
generation_config["maxOutputTokens"] = max_output_tokens
diff --git a/backend/app/tests/models/llm/test_request.py b/backend/app/tests/models/llm/test_request.py
index fdc1e5cd5..ab6509a00 100644
--- a/backend/app/tests/models/llm/test_request.py
+++ b/backend/app/tests/models/llm/test_request.py
@@ -1,4 +1,7 @@
-from app.models.llm.request import build_kaapi_completion_config
+import pytest
+from pydantic import ValidationError
+
+from app.models.llm.request import TextLLMParams, build_kaapi_completion_config
from app.services.llm.mappers import kaapi_params_as_dict
@@ -67,3 +70,13 @@ def test_unset_temperature_survives_json_round_trip(self) -> None:
assert "temperature" not in kaapi_params_as_dict(
round_tripped.completion.params
)
+
+
+class TestTextLLMParamsThinkingLevel:
+ @pytest.mark.parametrize("level", ["minimal", "low", "medium", "high"])
+ def test_accepts_gemini_thinking_levels(self, level: str) -> None:
+ assert TextLLMParams(thinking_level=level).thinking_level == level
+
+ def test_rejects_unknown_thinking_level(self) -> None:
+ with pytest.raises(ValidationError):
+ TextLLMParams(thinking_level="extreme")
diff --git a/backend/app/tests/services/llm/providers/test_google_aistudio.py b/backend/app/tests/services/llm/providers/test_google_aistudio.py
index 0b44fb428..3848a4cfb 100644
--- a/backend/app/tests/services/llm/providers/test_google_aistudio.py
+++ b/backend/app/tests/services/llm/providers/test_google_aistudio.py
@@ -1,355 +1,688 @@
-"""
-Tests for the Google AI provider (STT and TTS).
-"""
-
import base64
+from typing import Any
+from unittest.mock import MagicMock, patch
+import httpx
import pytest
-from unittest.mock import MagicMock, patch
-from types import SimpleNamespace
+from google.genai._gaos.lib import compat_errors
+from google.genai.interactions import Interaction
-from app.models.llm import (
- NativeCompletionConfig,
- QueryParams,
-)
+from app.core.audio_utils import AudioRef, pcm_to_wav
+from app.models.llm import NativeCompletionConfig, QueryParams, TextContent
from app.models.llm.constants import CompletionType
from app.services.llm.providers.google_aistudio import GoogleAIProvider
+SAMPLE_PCM_BYTES = b"\x00\x01" * 1000
+GEMINI_REQUEST = httpx.Request("POST", "https://generativelanguage.googleapis.com")
+
+
+def make_interaction(
+ *,
+ text: str | None = "Hello world",
+ audio: bytes | None = None,
+ interaction_id: str | None = "int_123",
+ model: str | None = "gemini-2.5-pro",
+ status: str = "completed",
+ errors: list[dict[str, str]] | None = None,
+ usage: dict[str, int] | None = None,
+) -> Interaction:
+ """Build a real SDK Interaction from a wire-shaped payload.
+
+ The SDK derives output_text / output_audio from the model_output step, so
+ going through model_validate keeps the tests honest about those accessors.
+ """
+ content: list[dict[str, Any]] = []
+ if text is not None:
+ content.append({"type": "text", "text": text})
+ if audio is not None:
+ content.append(
+ {
+ "type": "audio",
+ "data": base64.b64encode(audio).decode("ascii"),
+ "mime_type": "audio/l16",
+ }
+ )
+ payload: dict[str, Any] = {
+ "status": status,
+ "model": model,
+ "steps": [{"type": "model_output", "content": content}],
+ "usage": usage
+ if usage is not None
+ else {
+ "total_input_tokens": 50,
+ "total_output_tokens": 100,
+ "total_thought_tokens": 7,
+ "total_tokens": 157,
+ },
+ }
+ if interaction_id is not None:
+ payload["id"] = interaction_id
+ if errors is not None:
+ payload["errors"] = errors
+ return Interaction.model_validate(payload)
-def mock_google_response(
- text: str = "Transcribed text",
- model: str = "gemini-2.5-pro",
- response_id: str = "resp_123",
-) -> SimpleNamespace:
- """Create a mock Google AI response object."""
- usage = SimpleNamespace(
- prompt_token_count=50,
- candidates_token_count=100,
- total_token_count=150,
- thoughts_token_count=0,
- )
- response = SimpleNamespace(
- response_id=response_id,
- model_version=model,
- text=text,
- usage_metadata=usage,
- model_dump=lambda: {
- "response_id": response_id,
- "model_version": model,
- "text": text,
- "usage_metadata": {
- "prompt_token_count": 50,
- "candidates_token_count": 100,
- "total_token_count": 150,
- "thoughts_token_count": 0,
- },
- },
- )
- return response
+def create_kwargs(mock_client: MagicMock) -> dict[str, Any]:
+ mock_client.interactions.create.assert_called_once()
+ return mock_client.interactions.create.call_args.kwargs
-class TestGoogleAIProviderSTT:
- """Test cases for GoogleAIProvider STT functionality."""
+@pytest.fixture
+def mock_client() -> MagicMock:
+ return MagicMock()
+
+
+@pytest.fixture
+def provider(mock_client: MagicMock) -> GoogleAIProvider:
+ return GoogleAIProvider(client=mock_client)
+
+@pytest.fixture
+def query_params() -> QueryParams:
+ return QueryParams(input="Hello there")
+
+
+class TestGoogleAIProviderText:
@pytest.fixture
- def mock_client(self):
- """Create a mock Google AI client."""
- client = MagicMock()
- # Mock file upload
- mock_file = MagicMock()
- mock_file.name = "test_audio.wav"
- client.files.upload.return_value = mock_file
- return client
+ def text_config(self) -> NativeCompletionConfig:
+ return NativeCompletionConfig(
+ provider="google-native",
+ type=CompletionType.TEXT,
+ params={"model": "gemini-2.5-pro"},
+ )
+
+ def test_text_happy_path(self, provider, mock_client, text_config, query_params):
+ mock_client.interactions.create.return_value = make_interaction(text="Hi!")
+
+ result, error = provider.execute(text_config, query_params, "Hello there")
+
+ assert error is None
+ assert result.response.output.content.value == "Hi!"
+ assert result.response.provider_response_id == "int_123"
+ assert result.response.model == "gemini-2.5-pro"
+ assert result.usage.input_tokens == 50
+ assert result.usage.output_tokens == 100
+ assert result.usage.reasoning_tokens == 7
+ assert result.usage.total_tokens == 157
+ kwargs = create_kwargs(mock_client)
+ assert kwargs["model"] == "gemini-2.5-pro"
+ assert kwargs["input"] == [
+ {"type": "user_input", "content": [{"type": "text", "text": "Hello there"}]}
+ ]
+
+ def test_text_instructions_become_system_instruction(
+ self, provider, mock_client, text_config, query_params
+ ):
+ text_config.params["instructions"] = "Be terse."
+ mock_client.interactions.create.return_value = make_interaction()
+
+ provider.execute(text_config, query_params, "Ping")
+
+ assert create_kwargs(mock_client)["system_instruction"] == "Be terse."
+
+ def test_text_minimal_request_has_no_optional_fields(
+ self, provider, mock_client, text_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction()
+
+ provider.execute(text_config, query_params, "Ping")
+
+ assert set(create_kwargs(mock_client)) == {"model", "input"}
+
+ def test_text_thinking_config_takes_precedence_over_reasoning(
+ self, provider, mock_client, text_config, query_params
+ ):
+ text_config.params["thinking_config"] = {"thinking_level": "high"}
+ text_config.params["reasoning"] = "low"
+ mock_client.interactions.create.return_value = make_interaction()
+
+ provider.execute(text_config, query_params, "Ping")
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "thinking_level": "high"
+ }
+ def test_text_reasoning_used_as_thinking_level_fallback(
+ self, provider, mock_client, text_config, query_params
+ ):
+ text_config.params["reasoning"] = "low"
+ mock_client.interactions.create.return_value = make_interaction()
+
+ provider.execute(text_config, query_params, "Ping")
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "thinking_level": "low"
+ }
+
+ def test_text_max_output_tokens_in_generation_config(
+ self, provider, mock_client, text_config, query_params
+ ):
+ text_config.params["max_output_tokens"] = 256
+ mock_client.interactions.create.return_value = make_interaction()
+
+ provider.execute(text_config, query_params, "Ping")
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "max_output_tokens": 256
+ }
+
+ def test_text_sampling_params_not_sent(
+ self, provider, mock_client, text_config, query_params
+ ):
+ text_config.params["temperature"] = 0.2
+ text_config.params["top_p"] = 0.9
+ mock_client.interactions.create.return_value = make_interaction()
+
+ result, error = provider.execute(text_config, query_params, "Ping")
+
+ assert error is None
+ assert set(create_kwargs(mock_client)) == {"model", "input"}
+
+ def test_text_list_of_content_parts(
+ self, provider, mock_client, text_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction()
+
+ result, error = provider.execute(
+ text_config, query_params, [TextContent(value="part one")]
+ )
+
+ assert error is None
+ assert create_kwargs(mock_client)["input"] == [
+ {"type": "user_input", "content": [{"type": "text", "text": "part one"}]}
+ ]
+
+ def test_text_knowledge_base_ids_set_file_search_tool(
+ self, provider, mock_client, text_config, query_params
+ ):
+ store_names = ["fileSearchStores/store-1", "fileSearchStores/store-2"]
+ text_config.params["knowledge_base_ids"] = store_names
+ mock_client.interactions.create.return_value = make_interaction()
+
+ result, error = provider.execute(text_config, query_params, "Ask the KB")
+
+ assert error is None
+ assert create_kwargs(mock_client)["tools"] == [
+ {"type": "file_search", "file_search_store_names": store_names}
+ ]
+
+ def test_text_missing_model_uses_default(self, provider, mock_client, query_params):
+ config = NativeCompletionConfig(
+ provider="google-native", type=CompletionType.TEXT, params={}
+ )
+ mock_client.interactions.create.return_value = make_interaction(model=None)
+
+ result, error = provider.execute(config, query_params, "Hi")
+
+ assert error is None
+ assert create_kwargs(mock_client)["model"] == "gemini-2.5-pro"
+ assert result.response.model == "gemini-2.5-pro"
+
+ def test_text_include_provider_raw_response(
+ self, provider, mock_client, text_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction(text="raw")
+
+ result, error = provider.execute(
+ text_config, query_params, "Hi", include_provider_raw_response=True
+ )
+
+ assert error is None
+ assert result.provider_raw_response["id"] == "int_123"
+ assert result.provider_raw_response["output_text"] == "raw"
+
+ def test_text_missing_usage_defaults_to_zeros(
+ self, provider, mock_client, text_config, query_params
+ ):
+ interaction = make_interaction()
+ interaction.usage = None
+ mock_client.interactions.create.return_value = interaction
+
+ result, error = provider.execute(text_config, query_params, "Hi")
+
+ assert error is None
+ assert result.usage.input_tokens == 0
+ assert result.usage.output_tokens == 0
+ assert result.usage.total_tokens == 0
+ assert result.usage.reasoning_tokens == 0
+
+ def test_text_missing_interaction_id_returns_error(
+ self, provider, mock_client, text_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction(
+ interaction_id=None
+ )
+
+ result, error = provider.execute(text_config, query_params, "Hi")
+
+ assert result is None
+ assert error.startswith("[GEMINI] Text interaction did not complete")
+
+ @pytest.mark.parametrize("status", ["failed", "incomplete"])
+ def test_text_non_completed_status_returns_error(
+ self, provider, mock_client, text_config, query_params, status
+ ):
+ mock_client.interactions.create.return_value = make_interaction(
+ status=status,
+ errors=[{"code": "SAFETY", "message": "blocked by policy"}],
+ )
+
+ result, error = provider.execute(text_config, query_params, "Hi")
+
+ assert result is None
+ assert error.startswith("[GEMINI]")
+ assert f"status={status}" in error
+ assert "SAFETY" in error
+ assert "blocked by policy" in error
+
+ def test_text_empty_output_text_returns_error(
+ self, provider, mock_client, text_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction(text=None)
+
+ result, error = provider.execute(text_config, query_params, "Hi")
+
+ assert result is None
+ assert error.startswith("[GEMINI]")
+ assert "missing generated content" in error
+
+
+class TestGoogleAIProviderSTT:
@pytest.fixture
- def provider(self, mock_client):
- """Create a GoogleAIProvider instance with mock client."""
- return GoogleAIProvider(client=mock_client)
+ def mock_client(self) -> MagicMock:
+ client = MagicMock()
+ uploaded = MagicMock()
+ uploaded.name = "files/abc"
+ uploaded.uri = "https://generativelanguage.googleapis.com/v1beta/files/abc"
+ uploaded.mime_type = "audio/wav"
+ client.files.upload.return_value = uploaded
+ return client
@pytest.fixture
- def stt_config(self):
- """Create a basic STT completion config."""
+ def stt_config(self) -> NativeCompletionConfig:
return NativeCompletionConfig(
provider="google-aistudio-native",
type=CompletionType.STT,
- params={
- "model": "gemini-2.5-pro",
- },
+ params={"model": "gemini-2.5-pro"},
)
@pytest.fixture
- def audio_ref(self):
- from app.core.audio_utils import AudioRef
-
+ def audio_ref(self) -> AudioRef:
return AudioRef(bytes_=b"fake audio data", mime_type="audio/wav")
- @pytest.fixture
- def query_params(self):
- """Create basic query parameters."""
- return QueryParams(input="Test audio input")
+ @staticmethod
+ def _prompt(mock_client: MagicMock) -> str:
+ content = create_kwargs(mock_client)["input"][0]["content"]
+ return content[0]["text"]
def test_stt_success_with_auto_language(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test successful STT execution with auto language detection."""
- mock_response = mock_google_response(text="Hello world")
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = make_interaction(
+ text="Hello world"
+ )
result, error = provider.execute(stt_config, query_params, audio_ref)
assert error is None
- assert result is not None
assert result.response.output.content.value == "Hello world"
+ assert result.response.provider_response_id == "int_123"
assert result.response.model == "gemini-2.5-pro"
assert result.response.provider == "google-aistudio-native"
assert result.usage.input_tokens == 50
assert result.usage.output_tokens == 100
- assert result.usage.total_tokens == 150
-
- # Verify file upload was called with a materialized temp path matching the AudioRef mime.
- mock_client.files.upload.assert_called_once()
+ assert result.usage.total_tokens == 157
uploaded_path = mock_client.files.upload.call_args.kwargs["file"]
assert uploaded_path.endswith(".wav")
- mock_client.models.generate_content.assert_called_once()
+ assert "Detect the spoken language automatically" in self._prompt(mock_client)
- # Verify instruction contains auto-detect
- call_args = mock_client.models.generate_content.call_args
- assert "Detect the spoken language automatically" in call_args[1]["contents"][0]
+ def test_stt_references_uploaded_file_uri(
+ self, provider, mock_client, stt_config, query_params, audio_ref
+ ):
+ mock_client.interactions.create.return_value = make_interaction()
- def test_stt_with_specific_input_language(
+ provider.execute(stt_config, query_params, audio_ref)
+
+ kwargs = create_kwargs(mock_client)
+ assert kwargs["model"] == "gemini-2.5-pro"
+ assert kwargs["input"][0]["type"] == "user_input"
+ assert kwargs["input"][0]["content"][1] == {
+ "type": "audio",
+ "uri": "https://generativelanguage.googleapis.com/v1beta/files/abc",
+ "mime_type": "audio/wav",
+ }
+
+ def test_stt_defaults_thinking_level_to_low(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test STT with specific input language."""
- stt_config.params["input_language"] = "English"
+ mock_client.interactions.create.return_value = make_interaction()
+
+ provider.execute(stt_config, query_params, audio_ref)
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "thinking_level": "low"
+ }
+
+ def test_stt_explicit_thinking_level_overrides_default(
+ self, provider, mock_client, stt_config, query_params, audio_ref
+ ):
+ stt_config.params["thinking_level"] = "high"
+ mock_client.interactions.create.return_value = make_interaction()
- mock_response = mock_google_response(text="Transcribed English text")
- mock_client.models.generate_content.return_value = mock_response
+ provider.execute(stt_config, query_params, audio_ref)
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "thinking_level": "high"
+ }
+
+ def test_stt_sampling_params_not_sent(
+ self, provider, mock_client, stt_config, query_params, audio_ref
+ ):
+ stt_config.params["temperature"] = 0.4
+ stt_config.params["top_p"] = 0.9
+ mock_client.interactions.create.return_value = make_interaction()
result, error = provider.execute(stt_config, query_params, audio_ref)
assert error is None
- assert result is not None
+ kwargs = create_kwargs(mock_client)
+ assert set(kwargs) == {"model", "input", "generation_config"}
+ assert kwargs["generation_config"] == {"thinking_level": "low"}
+
+ def test_stt_with_specific_input_language(
+ self, provider, mock_client, stt_config, query_params, audio_ref
+ ):
+ stt_config.params["input_language"] = "English"
+ mock_client.interactions.create.return_value = make_interaction()
+
+ result, error = provider.execute(stt_config, query_params, audio_ref)
- # Verify instruction contains specific language
- call_args = mock_client.models.generate_content.call_args
- assert "Transcribe the audio from English" in call_args[1]["contents"][0]
+ assert error is None
+ assert "Transcribe the audio from English" in self._prompt(mock_client)
def test_stt_with_translation(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test STT with translation to different output language."""
stt_config.params["input_language"] = "Spanish"
stt_config.params["output_language"] = "English"
-
- mock_response = mock_google_response(text="Translated text")
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = make_interaction()
result, error = provider.execute(stt_config, query_params, audio_ref)
assert error is None
- assert result is not None
-
- # Verify instruction contains translation
- call_args = mock_client.models.generate_content.call_args
- instruction = call_args[1]["contents"][0]
- assert "Transcribe the audio from Spanish" in instruction
- assert "translate to English" in instruction
+ prompt = self._prompt(mock_client)
+ assert "Transcribe the audio from Spanish" in prompt
+ assert "translate to English" in prompt
+ assert result.response.output.content.language_code == "English"
def test_stt_with_custom_instructions(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test STT with custom instructions."""
stt_config.params["instructions"] = "Include timestamps"
-
- mock_response = mock_google_response(text="Transcribed with timestamps")
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = make_interaction()
result, error = provider.execute(stt_config, query_params, audio_ref)
assert error is None
- assert result is not None
-
- # Verify custom instructions are included
- call_args = mock_client.models.generate_content.call_args
- instruction = call_args[1]["contents"][0]
- assert "Include timestamps" in instruction
+ assert self._prompt(mock_client).startswith("Include timestamps. ")
- def test_stt_with_type_error(
+ def test_stt_empty_output_text_returns_error(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test handling of TypeError (invalid parameters)."""
- mock_client.models.generate_content.side_effect = TypeError(
- "unexpected keyword argument 'invalid_param'"
- )
+ mock_client.interactions.create.return_value = make_interaction(text=None)
result, error = provider.execute(stt_config, query_params, audio_ref)
assert result is None
- assert error is not None
- assert "Invalid or unexpected parameter in Config" in error
+ assert error.startswith("[GEMINI]")
+ assert "missing transcribed text" in error
- def test_stt_with_generic_exception(
+ def test_stt_failed_status_returns_error(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test handling of unexpected exceptions."""
- mock_client.files.upload.side_effect = Exception("File upload failed")
+ mock_client.interactions.create.return_value = make_interaction(status="failed")
result, error = provider.execute(stt_config, query_params, audio_ref)
assert result is None
- assert error is not None
- assert "[KAAPI] Unexpected error" in error
+ assert "STT interaction did not complete (status=failed" in error
- def test_stt_with_invalid_input_type(
- self, provider, mock_client, stt_config, query_params
+ def test_stt_missing_interaction_id_returns_error(
+ self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test STT execution with invalid input type (non-string)."""
- # Pass a dict instead of a string path
- invalid_input = {"invalid": "data"}
+ mock_client.interactions.create.return_value = make_interaction(
+ interaction_id=None
+ )
- result, error = provider.execute(stt_config, query_params, invalid_input)
+ result, error = provider.execute(stt_config, query_params, audio_ref)
assert result is None
- assert error is not None
- assert "STT requires an AudioRef" in error
+ assert error.startswith("[GEMINI] STT interaction did not complete")
- def test_stt_with_valid_audio_ref(
+ def test_stt_type_error(
self, provider, mock_client, stt_config, query_params, audio_ref
):
- """Test STT execution with valid file path string."""
- mock_response = mock_google_response(text="Valid transcription")
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.side_effect = TypeError(
+ "unexpected keyword argument 'invalid_param'"
+ )
result, error = provider.execute(stt_config, query_params, audio_ref)
- assert error is None
- assert result is not None
- assert result.response.output.content.value == "Valid transcription"
-
-
-def mock_tts_google_response(
- audio_bytes: bytes = b"\x00\x01\x02\x03",
- model: str = "gemini-2.5-pro-preview-tts",
- response_id: str = "resp_tts_123",
-) -> SimpleNamespace:
- """Create a mock Google AI TTS response object with audio data."""
- usage = SimpleNamespace(
- prompt_token_count=10,
- candidates_token_count=0,
- total_token_count=10,
- thoughts_token_count=0,
- )
-
- inline_data = SimpleNamespace(data=audio_bytes)
- part = SimpleNamespace(inline_data=inline_data)
- content = SimpleNamespace(parts=[part])
- candidate = SimpleNamespace(content=content)
-
- response = SimpleNamespace(
- response_id=response_id,
- model_version=model,
- candidates=[candidate],
- usage_metadata=usage,
- model_dump=lambda: {
- "response_id": response_id,
- "model_version": model,
- "usage_metadata": {
- "prompt_token_count": 10,
- "candidates_token_count": 0,
- "total_token_count": 10,
- "thoughts_token_count": 0,
- },
- },
- )
- return response
+ assert result is None
+ assert "Invalid or unexpected parameter in Config" in error
+ def test_stt_generic_exception(
+ self, provider, mock_client, stt_config, query_params, audio_ref
+ ):
+ mock_client.files.upload.side_effect = Exception("File upload failed")
-SAMPLE_PCM_BYTES = b"\x00\x01" * 1000
+ result, error = provider.execute(stt_config, query_params, audio_ref)
+ assert result is None
+ assert "[KAAPI] Unexpected error" in error
-class TestGoogleAIProviderTTS:
- """Test cases for GoogleAIProvider TTS functionality."""
+ def test_stt_invalid_input_type(
+ self, provider, mock_client, stt_config, query_params
+ ):
+ result, error = provider.execute(stt_config, query_params, {"invalid": "data"})
- @pytest.fixture
- def mock_client(self):
- """Create a mock Google AI client."""
- return MagicMock()
+ assert result is None
+ assert "STT requires an AudioRef" in error
+ mock_client.interactions.create.assert_not_called()
- @pytest.fixture
- def provider(self, mock_client):
- """Create a GoogleAIProvider instance with mock client."""
- return GoogleAIProvider(client=mock_client)
+class TestGoogleAIProviderTTS:
@pytest.fixture
- def tts_config(self):
- """Create a basic TTS completion config."""
+ def tts_config(self) -> NativeCompletionConfig:
return NativeCompletionConfig(
provider="google-aistudio-native",
type=CompletionType.TTS,
params={
"model": "gemini-2.5-pro-preview-tts",
"voice": "Kore",
- "language": "en-US",
+ "language": "en-IN",
},
)
@pytest.fixture
- def query_params(self):
- """Create basic query parameters."""
- return QueryParams(input="Hello world")
+ def pcm_interaction(self) -> Interaction:
+ return make_interaction(
+ text=None, audio=SAMPLE_PCM_BYTES, model="gemini-2.5-pro-preview-tts"
+ )
+
+ def test_tts_request_shape(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ result, error = provider.execute(tts_config, query_params, "Say this text")
+
+ assert error is None
+ kwargs = create_kwargs(mock_client)
+ assert kwargs["model"] == "gemini-2.5-pro-preview-tts"
+ assert kwargs["input"] == [
+ {
+ "type": "user_input",
+ "content": [
+ {"type": "text", "text": "Say this text"}
+ ],
+ }
+ ]
+ assert kwargs["response_format"] == {"type": "audio"}
+
+ def test_tts_passes_full_bcp47_language(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ provider.execute(tts_config, query_params, "Hello")
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "speech_config": [{"voice": "Kore", "language": "en-IN"}]
+ }
+
+ def test_tts_omits_language_when_unset(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ del tts_config.params["language"]
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ provider.execute(tts_config, query_params, "Hello")
+
+ assert create_kwargs(mock_client)["generation_config"] == {
+ "speech_config": [{"voice": "Kore"}]
+ }
+
+ def test_tts_director_notes_become_style_annotation(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ tts_config.params["provider_specific"] = {
+ "gemini": {"director_notes": "Speak in a cheerful tone"}
+ }
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert error is None
+ transcript = create_kwargs(mock_client)["input"][0]["content"][0]
+ assert transcript["annotations"] == [
+ {"type": "speech_metadata", "style": "Speak in a cheerful tone"}
+ ]
+ assert "system_instruction" not in create_kwargs(mock_client)
+
+ def test_tts_without_director_notes_has_no_annotations(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ provider.execute(tts_config, query_params, "Hello")
+
+ transcript = create_kwargs(mock_client)["input"][0]["content"][0]
+ assert "annotations" not in transcript
+
+ def test_tts_default_wav_wraps_pcm_once(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert error is None
+ assert result.response.output.content.mime_type == "audio/wav"
+ decoded = base64.b64decode(result.response.output.content.value)
+ assert decoded == pcm_to_wav(SAMPLE_PCM_BYTES, sample_rate=24000)
- def test_tts_success_mp3_format(
+ def test_tts_wav_returned_by_server_is_not_double_wrapped(
+ self, provider, mock_client, tts_config, query_params
+ ):
+ # 3.8 TTS models reply with a full WAV rather than raw PCM.
+ server_wav = pcm_to_wav(SAMPLE_PCM_BYTES, sample_rate=24000)
+ mock_client.interactions.create.return_value = make_interaction(
+ text=None, audio=server_wav
+ )
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert error is None
+ decoded = base64.b64decode(result.response.output.content.value)
+ assert decoded == pcm_to_wav(SAMPLE_PCM_BYTES, sample_rate=24000)
+ assert decoded.count(b"RIFF") == 1
+
+ def test_tts_wav_returned_by_server_is_unwrapped_before_mp3(
self, provider, mock_client, tts_config, query_params
):
- """Test TTS with MP3 response format conversion."""
tts_config.params["response_format"] = "mp3"
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ server_wav = pcm_to_wav(SAMPLE_PCM_BYTES, sample_rate=24000)
+ mock_client.interactions.create.return_value = make_interaction(
+ text=None, audio=server_wav
+ )
- fake_mp3_bytes = b"fake-mp3-content"
with patch(
"app.services.llm.providers.google_aistudio.convert_pcm_to_mp3",
- return_value=(fake_mp3_bytes, None),
+ return_value=(b"fake-mp3", None),
+ ) as mock_convert:
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert error is None
+ # ffmpeg conversion is a local external process; mocked to check its input.
+ assert mock_convert.call_args.args[0] == SAMPLE_PCM_BYTES
+
+ def test_tts_mp3_format(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ tts_config.params["response_format"] = "mp3"
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ with patch(
+ "app.services.llm.providers.google_aistudio.convert_pcm_to_mp3",
+ return_value=(b"fake-mp3-content", None),
) as mock_convert:
result, error = provider.execute(tts_config, query_params, "Hello world")
assert error is None
- assert result is not None
assert result.response.output.content.mime_type == "audio/mp3"
- decoded = base64.b64decode(result.response.output.content.value)
- assert decoded == fake_mp3_bytes
+ assert base64.b64decode(result.response.output.content.value) == (
+ b"fake-mp3-content"
+ )
mock_convert.assert_called_once_with(SAMPLE_PCM_BYTES)
- def test_tts_success_ogg_format(
- self, provider, mock_client, tts_config, query_params
+ def test_tts_ogg_format(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
):
- """Test TTS with OGG response format conversion."""
tts_config.params["response_format"] = "ogg"
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = pcm_interaction
- fake_ogg_bytes = b"fake-ogg-content"
with patch(
"app.services.llm.providers.google_aistudio.convert_pcm_to_ogg",
- return_value=(fake_ogg_bytes, None),
+ return_value=(b"fake-ogg-content", None),
) as mock_convert:
result, error = provider.execute(tts_config, query_params, "Hello world")
assert error is None
- assert result is not None
assert result.response.output.content.mime_type == "audio/ogg"
- decoded = base64.b64decode(result.response.output.content.value)
- assert decoded == fake_ogg_bytes
+ assert base64.b64decode(result.response.output.content.value) == (
+ b"fake-ogg-content"
+ )
mock_convert.assert_called_once_with(SAMPLE_PCM_BYTES)
+ def test_tts_unsupported_format_falls_back_to_wav(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
+ ):
+ tts_config.params["response_format"] = "flac"
+ mock_client.interactions.create.return_value = pcm_interaction
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert error is None
+ assert result.response.output.content.mime_type == "audio/wav"
+ decoded = base64.b64decode(result.response.output.content.value)
+ assert decoded.startswith(b"RIFF")
+
def test_tts_mp3_conversion_failure(
- self, provider, mock_client, tts_config, query_params
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
):
- """Test error when MP3 conversion fails."""
tts_config.params["response_format"] = "mp3"
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = pcm_interaction
with patch(
"app.services.llm.providers.google_aistudio.convert_pcm_to_mp3",
@@ -362,12 +695,10 @@ def test_tts_mp3_conversion_failure(
assert "ffmpeg not found" in error
def test_tts_ogg_conversion_failure(
- self, provider, mock_client, tts_config, query_params
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
):
- """Test error when OGG conversion fails."""
tts_config.params["response_format"] = "ogg"
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = pcm_interaction
with patch(
"app.services.llm.providers.google_aistudio.convert_pcm_to_ogg",
@@ -378,280 +709,231 @@ def test_tts_ogg_conversion_failure(
assert result is None
assert "unable to convert Gemini PCM audio to OGG" in error
+ def test_tts_missing_output_audio_returns_error(
+ self, provider, mock_client, tts_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction(text="oops")
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert result is None
+ assert error.startswith("[GEMINI] Failed to extract audio bytes")
+
+ def test_tts_incomplete_status_returns_error(
+ self, provider, mock_client, tts_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction(
+ text=None, audio=SAMPLE_PCM_BYTES, status="incomplete"
+ )
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert result is None
+ assert "TTS interaction did not complete (status=incomplete" in error
+
+ def test_tts_missing_interaction_id_returns_error(
+ self, provider, mock_client, tts_config, query_params
+ ):
+ mock_client.interactions.create.return_value = make_interaction(
+ text=None, audio=SAMPLE_PCM_BYTES, interaction_id=None
+ )
+
+ result, error = provider.execute(tts_config, query_params, "Hello")
+
+ assert result is None
+ assert error.startswith("[GEMINI] TTS interaction did not complete")
+
def test_tts_empty_input(self, provider, mock_client, tts_config, query_params):
- """Test error when text input is empty."""
result, error = provider.execute(tts_config, query_params, " ")
assert result is None
assert "text input is empty" in error
- mock_client.models.generate_content.assert_not_called()
+ mock_client.interactions.create.assert_not_called()
def test_tts_non_string_input(
self, provider, mock_client, tts_config, query_params
):
- """Test error when input is not a string."""
result, error = provider.execute(tts_config, query_params, {"invalid": "data"})
assert result is None
assert "TTS requires a text string" in error
- def test_tts_no_usage_metadata(
- self, provider, mock_client, tts_config, query_params
+ def test_tts_missing_usage_defaults_to_zeros(
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
):
- """Test TTS response when usage_metadata is None (defaults to zeros)."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_response.usage_metadata = None
- mock_client.models.generate_content.return_value = mock_response
+ pcm_interaction.usage = None
+ mock_client.interactions.create.return_value = pcm_interaction
result, error = provider.execute(tts_config, query_params, "Hello")
assert error is None
- assert result is not None
assert result.usage.input_tokens == 0
assert result.usage.output_tokens == 0
assert result.usage.total_tokens == 0
assert result.usage.reasoning_tokens == 0
def test_tts_include_provider_raw_response(
- self, provider, mock_client, tts_config, query_params
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
):
- """Test TTS with include_provider_raw_response=True."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = pcm_interaction
result, error = provider.execute(
tts_config, query_params, "Hello", include_provider_raw_response=True
)
assert error is None
- assert result is not None
- assert result.provider_raw_response is not None
- assert isinstance(result.provider_raw_response, dict)
+ assert result.provider_raw_response["id"] == "int_123"
+ assert result.provider_raw_response["output_audio"]["mime_type"] == "audio/l16"
def test_tts_without_provider_raw_response(
- self, provider, mock_client, tts_config, query_params
+ self, provider, mock_client, tts_config, query_params, pcm_interaction
):
- """Test TTS without raw response (default)."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = pcm_interaction
result, error = provider.execute(tts_config, query_params, "Hello")
assert error is None
assert result.provider_raw_response is None
- def test_tts_with_director_notes(
+ def test_tts_model_falls_back_to_config_model(
self, provider, mock_client, tts_config, query_params
):
- """Test TTS with Gemini-specific director_notes parameter."""
- tts_config.params["provider_specific"] = {
- "gemini": {"director_notes": "Speak in a cheerful tone"}
- }
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
+ mock_client.interactions.create.return_value = make_interaction(
+ text=None, audio=SAMPLE_PCM_BYTES, model=None
+ )
result, error = provider.execute(tts_config, query_params, "Hello")
assert error is None
- assert result is not None
- # Verify config was passed with system_instruction
- call_args = mock_client.models.generate_content.call_args
- config_arg = call_args[1]["config"]
- assert config_arg.system_instruction == "Speak in a cheerful tone"
-
- def test_tts_without_director_notes(
- self, provider, mock_client, tts_config, query_params
- ):
- """Test TTS without director_notes (no system_instruction in config)."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
-
- result, error = provider.execute(tts_config, query_params, "Hello")
+ assert result.response.model == "gemini-2.5-pro-preview-tts"
- assert error is None
- call_args = mock_client.models.generate_content.call_args
- config_arg = call_args[1]["config"]
- assert (
- not hasattr(config_arg, "system_instruction")
- or config_arg.system_instruction is None
+ def test_tts_default_model(self, provider, mock_client, query_params):
+ config = NativeCompletionConfig(
+ provider="google-aistudio-native", type=CompletionType.TTS, params={}
+ )
+ mock_client.interactions.create.return_value = make_interaction(
+ text=None, audio=SAMPLE_PCM_BYTES, model=None
)
- def test_tts_model_version_fallback(
- self, provider, mock_client, tts_config, query_params
- ):
- """Test that model falls back to config model when model_version is None."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_response.model_version = None
- mock_client.models.generate_content.return_value = mock_response
-
- result, error = provider.execute(tts_config, query_params, "Hello")
+ result, error = provider.execute(config, query_params, "Hello")
assert error is None
- assert result.response.model == "gemini-2.5-pro-preview-tts"
+ assert create_kwargs(mock_client)["model"] == "gemini-3.8-flash-tts"
def test_tts_generic_exception(
self, provider, mock_client, tts_config, query_params
):
- """Test handling of unexpected exceptions."""
- mock_client.models.generate_content.side_effect = Exception("API unavailable")
+ mock_client.interactions.create.side_effect = Exception("API unavailable")
result, error = provider.execute(tts_config, query_params, "Hello")
assert result is None
assert "[KAAPI] Unexpected error" in error
- def test_tts_type_error(self, provider, mock_client, tts_config, query_params):
- """Test handling of TypeError (invalid parameters)."""
- mock_client.models.generate_content.side_effect = TypeError(
- "unexpected keyword argument"
- )
-
- result, error = provider.execute(tts_config, query_params, "Hello")
-
- assert result is None
- assert "Invalid or unexpected parameter in Config" in error
-
- def test_tts_passes_correct_model_and_content(
- self, provider, mock_client, tts_config, query_params
- ):
- """Test that the correct model and text content are passed to the API."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
-
- result, error = provider.execute(tts_config, query_params, "Say this text")
-
- assert error is None
- call_args = mock_client.models.generate_content.call_args
- assert call_args[1]["model"] == "gemini-2.5-pro-preview-tts"
- assert call_args[1]["contents"] == "Say this text"
-
- def test_tts_passes_correct_voice_config(
- self, provider, mock_client, tts_config, query_params
- ):
- """Test that voice and language are correctly configured."""
- mock_response = mock_tts_google_response(audio_bytes=SAMPLE_PCM_BYTES)
- mock_client.models.generate_content.return_value = mock_response
-
- result, error = provider.execute(tts_config, query_params, "Hello")
- assert error is None
- call_args = mock_client.models.generate_content.call_args
- config_arg = call_args[1]["config"]
- assert config_arg.response_modalities == ["AUDIO"]
- voice_name = (
- config_arg.speech_config.voice_config.prebuilt_voice_config.voice_name
- )
- assert voice_name == "Kore"
- assert config_arg.speech_config.language_code == "en-US"
+def status_error(status_code: int, status: str, message: str) -> Exception:
+ body = {"error": {"code": status_code, "status": status, "message": message}}
+ response = httpx.Response(status_code, json=body, request=GEMINI_REQUEST)
+ return compat_errors.APIError.generate(status_code, body, message, response)
-class TestGoogleAIProviderText:
- """Coverage for the text completion path (`_execute_text`)."""
-
- @pytest.fixture
- def mock_client(self):
- return MagicMock()
-
- @pytest.fixture
- def provider(self, mock_client):
- return GoogleAIProvider(client=mock_client)
-
+class TestGoogleAIProviderErrors:
@pytest.fixture
- def text_config(self):
+ def text_config(self) -> NativeCompletionConfig:
return NativeCompletionConfig(
provider="google-native",
type=CompletionType.TEXT,
- params={"model": "gemini-2.5-pro", "temperature": 0.2},
+ params={"model": "gemini-2.5-pro"},
)
- @pytest.fixture
- def query_params(self):
- return QueryParams(input="Hello there")
-
- def test_text_happy_path(self, provider, mock_client, text_config, query_params):
- mock_client.models.generate_content.return_value = mock_google_response(
- text="Hi!"
- )
- result, error = provider.execute(text_config, query_params, "Hello there")
- assert error is None
- assert result.response.output.content.value == "Hi!"
- assert result.usage.input_tokens == 50
- assert result.usage.output_tokens == 100
-
- def test_text_with_instructions_and_reasoning(
- self, provider, mock_client, text_config, query_params
+ @pytest.mark.parametrize(
+ ("status_code", "status", "expected_prefix"),
+ [
+ (400, "INVALID_ARGUMENT", "[GEMINI] Bad request (code: 400)"),
+ (
+ 401,
+ "UNAUTHENTICATED",
+ "[GEMINI] Authentication / permission denied (code: 401)",
+ ),
+ (
+ 403,
+ "PERMISSION_DENIED",
+ "[GEMINI] Authentication / permission denied (code: 403)",
+ ),
+ (404, "NOT_FOUND", "[GEMINI] Resource not found (code: 404)"),
+ (
+ 429,
+ "RESOURCE_EXHAUSTED",
+ "[GEMINI] Rate limit / quota exceeded (code: 429)",
+ ),
+ (500, "INTERNAL", "[GEMINI] Server error (code: 500)"),
+ (503, "UNAVAILABLE", "[GEMINI] Server error (code: 503)"),
+ (409, "ABORTED", "[GEMINI] Client error (code: 409)"),
+ ],
+ )
+ def test_api_status_error_maps_to_gemini_message(
+ self,
+ provider,
+ mock_client,
+ text_config,
+ query_params,
+ status_code,
+ status,
+ expected_prefix,
):
- text_config.params["instructions"] = "Be terse."
- text_config.params["reasoning"] = "low"
- mock_client.models.generate_content.return_value = mock_google_response(
- text="ok"
+ mock_client.interactions.create.side_effect = status_error(
+ status_code, status, "upstream detail"
)
- result, error = provider.execute(text_config, query_params, "Ping")
- assert error is None
- config_arg = mock_client.models.generate_content.call_args[1]["config"]
- assert config_arg.system_instruction == "Be terse."
- assert config_arg.thinking_config is not None
- def test_text_missing_response_id_returns_error(
- self, provider, mock_client, text_config, query_params
- ):
- resp = mock_google_response(text="x")
- resp.response_id = None
- mock_client.models.generate_content.return_value = resp
result, error = provider.execute(text_config, query_params, "Hi")
+
assert result is None
- assert "response_id" in error
+ assert error.startswith(f"{expected_prefix}: upstream detail")
- def test_text_missing_text_returns_error(
+ def test_api_status_error_without_envelope_uses_exception_message(
self, provider, mock_client, text_config, query_params
):
- resp = mock_google_response(text="")
- resp.candidates = [SimpleNamespace(finish_reason="SAFETY")]
- resp.prompt_feedback = SimpleNamespace(block_reason="HATE")
- mock_client.models.generate_content.return_value = resp
+ response = httpx.Response(502, text="Bad Gateway", request=GEMINI_REQUEST)
+ mock_client.interactions.create.side_effect = compat_errors.APIStatusError(
+ "gateway exploded", response=response, body="Bad Gateway"
+ )
+
result, error = provider.execute(text_config, query_params, "Hi")
+
assert result is None
- assert "missing generated content" in error
+ assert error.startswith("[GEMINI] Server error (code: 502): gateway exploded")
- def test_text_accepts_list_of_content_parts(
+ def test_timeout_maps_to_could_not_reach_gemini(
self, provider, mock_client, text_config, query_params
):
- from app.models.llm import TextContent
-
- mock_client.models.generate_content.return_value = mock_google_response(
- text="seen"
+ mock_client.interactions.create.side_effect = compat_errors.APITimeoutError(
+ request=GEMINI_REQUEST
)
- result, error = provider.execute(
- text_config, query_params, [TextContent(value="part one")]
- )
- assert error is None
- contents = mock_client.models.generate_content.call_args[1]["contents"]
- assert contents[0]["parts"][0]["text"] == "part one"
- def test_text_with_knowledge_base_ids_sets_file_search_tool(
+ result, error = provider.execute(text_config, query_params, "Hi")
+
+ assert result is None
+ assert error.startswith("[KAAPI] Could not reach Gemini (APITimeoutError)")
+
+ def test_connection_error_maps_to_could_not_reach_gemini(
self, provider, mock_client, text_config, query_params
):
- store_names = ["fileSearchStores/store-1", "fileSearchStores/store-2"]
- text_config.params["knowledge_base_ids"] = store_names
- mock_client.models.generate_content.return_value = mock_google_response(
- text="grounded"
+ mock_client.interactions.create.side_effect = compat_errors.APIConnectionError(
+ request=GEMINI_REQUEST
)
- result, error = provider.execute(text_config, query_params, "Ask the KB")
+ result, error = provider.execute(text_config, query_params, "Hi")
- assert error is None
- config_arg = mock_client.models.generate_content.call_args[1]["config"]
- assert config_arg.tools[0].file_search.file_search_store_names == store_names
+ assert result is None
+ assert error.startswith("[KAAPI] Could not reach Gemini (APIConnectionError)")
- def test_text_without_knowledge_base_ids_sets_no_tools(
+ def test_unsupported_completion_type_returns_kaapi_error(
self, provider, mock_client, text_config, query_params
):
- mock_client.models.generate_content.return_value = mock_google_response(
- text="plain"
- )
+ text_config.type = "video"
- result, error = provider.execute(text_config, query_params, "No KB")
+ result, error = provider.execute(text_config, query_params, "Hi")
- assert error is None
- config_arg = mock_client.models.generate_content.call_args[1]["config"]
- assert not config_arg.tools
+ assert result is None
+ assert error.startswith("[KAAPI] Unsupported completion type 'video'")
+ mock_client.interactions.create.assert_not_called()
diff --git a/backend/app/tests/services/llm/providers/test_google_gcp.py b/backend/app/tests/services/llm/providers/test_google_gcp.py
index 5263a588d..f73d046d5 100644
--- a/backend/app/tests/services/llm/providers/test_google_gcp.py
+++ b/backend/app/tests/services/llm/providers/test_google_gcp.py
@@ -318,7 +318,7 @@ def test_text_completion_happy_path(self, provider, query):
assert resp.response.output.content.value == "generated answer"
payload = mock_post.call_args.kwargs["json"]
assert payload["contents"][0]["parts"] == [{"text": "hello"}]
- assert payload["generationConfig"]["temperature"] == 0.2
+ assert "generationConfig" not in payload
def test_raw_response_included_when_requested(
self, provider, stt_config, query, audio_ref
@@ -334,7 +334,7 @@ def test_raw_response_included_when_requested(
assert resp.provider_raw_response == raw
# ── text ─────────────────────────────────────────────────────────────────
- def test_text_forwards_instructions_and_temperature(self, provider, query):
+ def test_text_forwards_instructions_and_drops_temperature(self, provider, query):
config = NativeCompletionConfig(
provider="google-native",
type=CompletionType.TEXT,
@@ -351,7 +351,72 @@ def test_text_forwards_instructions_and_temperature(self, provider, query):
provider.execute(config, query, "hello")
payload = mock_post.call_args.kwargs["json"]
assert payload["systemInstruction"] == {"parts": [{"text": "be concise"}]}
- assert payload["generationConfig"]["temperature"] == 0.4
+ assert "generationConfig" not in payload
+
+ def test_text_thinking_config_takes_precedence_over_reasoning(
+ self, provider, query
+ ):
+ config = NativeCompletionConfig(
+ provider="google-native",
+ type=CompletionType.TEXT,
+ params={
+ "model": "gemini-2.5-flash",
+ "thinking_config": {"thinking_level": "high"},
+ "reasoning": "low",
+ },
+ )
+ with patch(
+ "app.services.llm.providers.google_gcp.requests.post",
+ return_value=_mock_http_ok(_stt_response("hi there")),
+ ) as mock_post:
+ provider.execute(config, query, "hello")
+ payload = mock_post.call_args.kwargs["json"]
+ assert payload["generationConfig"] == {
+ "thinkingConfig": {"thinkingLevel": "high"}
+ }
+
+ def test_text_reasoning_used_as_thinking_level_fallback(self, provider, query):
+ config = NativeCompletionConfig(
+ provider="google-native",
+ type=CompletionType.TEXT,
+ params={"model": "gemini-2.5-flash", "reasoning": "minimal"},
+ )
+ with patch(
+ "app.services.llm.providers.google_gcp.requests.post",
+ return_value=_mock_http_ok(_stt_response("hi there")),
+ ) as mock_post:
+ provider.execute(config, query, "hello")
+ payload = mock_post.call_args.kwargs["json"]
+ assert payload["generationConfig"] == {
+ "thinkingConfig": {"thinkingLevel": "minimal"}
+ }
+
+ def test_stt_defaults_thinking_level_to_low_and_drops_temperature(
+ self, provider, stt_config, query, audio_ref
+ ):
+ stt_config.params["temperature"] = 0.3
+ with patch(
+ "app.services.llm.providers.google_gcp.requests.post",
+ return_value=_mock_http_ok(_stt_response("hi there")),
+ ) as mock_post:
+ resp, err = provider.execute(stt_config, query, audio_ref)
+ assert err is None
+ assert mock_post.call_args.kwargs["json"]["generationConfig"] == {
+ "maxOutputTokens": 2048,
+ "thinkingConfig": {"thinkingLevel": "low"},
+ }
+
+ def test_stt_explicit_thinking_level_overrides_default(
+ self, provider, stt_config, query, audio_ref
+ ):
+ stt_config.params["thinking_level"] = "high"
+ with patch(
+ "app.services.llm.providers.google_gcp.requests.post",
+ return_value=_mock_http_ok(_stt_response("hi there")),
+ ) as mock_post:
+ provider.execute(stt_config, query, audio_ref)
+ generation_config = mock_post.call_args.kwargs["json"]["generationConfig"]
+ assert generation_config["thinkingConfig"] == {"thinkingLevel": "high"}
def test_text_accepts_multimodal_list_input(self, provider, query):
config = NativeCompletionConfig(
diff --git a/backend/app/tests/services/llm/test_mappers.py b/backend/app/tests/services/llm/test_mappers.py
index f0540813f..ab888d2a0 100644
--- a/backend/app/tests/services/llm/test_mappers.py
+++ b/backend/app/tests/services/llm/test_mappers.py
@@ -199,7 +199,6 @@ def test_text_completion_basic(self):
assert result == {
"model": "gemini-2.5-pro",
- "temperature": 0.7,
"video_part_config": {"videoMetadata": {"fps": 1.0}},
"media_resolution": "MEDIA_RESOLUTION_LOW",
}
@@ -217,7 +216,7 @@ def test_text_completion_with_reasoning(self):
assert result["model"] == "gemini-2.5-pro"
assert result["reasoning"] == "high"
- assert result["temperature"] == 0.5
+ assert "temperature" not in result
assert warnings == []
def test_text_completion_knowledge_base_passthrough(self):
@@ -260,10 +259,31 @@ def test_text_completion_top_p_and_max_output_tokens(self):
kaapi_params.model_dump(exclude_none=True), completion_type="text"
)
- assert result["top_p"] == 0.8
+ assert "top_p" not in result
assert result["max_output_tokens"] == 512
assert warnings == []
+ def test_text_completion_temperature_and_top_p_dropped_silently(self):
+ kaapi_params = TextLLMParams(model="gemini-2.5-pro", temperature=0.0, top_p=0.5)
+
+ result, warnings = map_kaapi_to_google_params(
+ kaapi_params.model_dump(exclude_none=True), completion_type="text"
+ )
+
+ assert "temperature" not in result
+ assert "top_p" not in result
+ assert warnings == []
+
+ def test_stt_completion_drops_temperature_silently(self):
+ kaapi_params = STTLLMParams(model="gemini-2.5-pro", temperature=0.3)
+
+ result, warnings = map_kaapi_to_google_params(
+ kaapi_params.model_dump(exclude_none=True), completion_type="stt"
+ )
+
+ assert "temperature" not in result
+ assert warnings == []
+
def test_text_completion_thinking_level_maps_to_thinking_config(self):
kaapi_params = TextLLMParams(model="gemini-2.5-pro", thinking_level="high")
@@ -325,9 +345,8 @@ def test_tts_completion_with_voice(self):
assert result["model"] == "gemini-2.5-pro"
assert result["voice"] == "Orus"
- assert (
- result["language"] == "en"
- ) # Mapped from en-IN to en via BCP47_LOCALE_TO_GEMINI_LANG
+ # Full locale is kept: collapsing en-IN to "en" loses the Indian accent.
+ assert result["language"] == "en-IN"
assert result["response_format"] == "wav" # Default
assert warnings == []
@@ -389,12 +408,12 @@ def test_tts_language_missing_uses_auto_detect(self):
assert "language" not in result # Not set = auto-detect
assert warnings == []
- def test_tts_unsupported_language_warns_and_auto_detects(self):
- """Test TTS with unsupported language generates warning and uses auto-detect."""
+ @pytest.mark.parametrize("language", ["fr-FR", "as-IN"])
+ def test_tts_unsupported_language_falls_back_to_auto_detect(self, language):
kaapi_params = TTSLLMParams(
model="gemini-2.5-pro",
voice="Kore",
- language="fr-FR", # French not in BCP47_LOCALE_TO_GEMINI_LANG
+ language=language,
)
result, warnings = map_kaapi_to_google_params(
@@ -403,14 +422,36 @@ def test_tts_unsupported_language_warns_and_auto_detects(self):
assert result["model"] == "gemini-2.5-pro"
assert result["voice"] == "Kore"
- assert result["response_format"] == "wav" # Default
- assert "language" not in result # Not set, falls back to auto-detect
- assert len(warnings) == 1
- assert "Unsupported language 'fr-FR'" in warnings[0]
- assert "auto-detect" in warnings[0]
+ assert result["response_format"] == "wav"
+ assert "language" not in result
+ assert warnings == [
+ f"Unsupported language '{language}' for Gemini TTS, using auto-detect"
+ ]
+
+ def test_tts_odia_maps_to_standard_locale(self):
+ # Kaapi's canonical Odia tag is the non-standard od-IN.
+ kaapi_params = TTSLLMParams(
+ model="gemini-2.5-pro", voice="Kore", language="odia"
+ )
- def test_tts_supported_language_maps_correctly(self):
- """Test TTS with supported BCP-47 language maps to Gemini language code."""
+ result, warnings = map_kaapi_to_google_params(
+ kaapi_params.model_dump(exclude_none=True), completion_type="tts"
+ )
+
+ assert result["language"] == "or-IN"
+ assert warnings == []
+
+ def test_tts_without_language_leaves_it_unset(self):
+ kaapi_params = TTSLLMParams(model="gemini-2.5-pro", voice="Kore")
+
+ result, warnings = map_kaapi_to_google_params(
+ kaapi_params.model_dump(exclude_none=True), completion_type="tts"
+ )
+
+ assert "language" not in result
+ assert warnings == []
+
+ def test_tts_indic_language_keeps_full_locale(self):
kaapi_params = TTSLLMParams(
model="gemini-2.5-pro", voice="Kore", language="hi-IN"
)
@@ -421,9 +462,7 @@ def test_tts_supported_language_maps_correctly(self):
assert result["model"] == "gemini-2.5-pro"
assert result["voice"] == "Kore"
- assert (
- result["language"] == "hi"
- ) # BCP47_LOCALE_TO_GEMINI_LANG maps hi-IN -> hi
+ assert result["language"] == "hi-IN"
assert result["response_format"] == "wav" # Default
assert warnings == []
@@ -1175,9 +1214,7 @@ def test_text_completion_maps_via_google_mapper(self, db: Session):
assert native_config.provider == "google-native"
assert native_config.params["model"] == "gemini-2.5-pro"
- def test_unsupported_language_emits_warning(self, db: Session):
- """Languages not in BCP47_LOCALE_TO_GEMINI_LANG fall back to auto-detect
- and surface a warning, rather than silently being dropped."""
+ def test_unlisted_language_falls_back_to_auto_detect(self, db: Session):
kaapi_config = build_kaapi_completion_config(
provider="google",
type="tts",
@@ -1192,9 +1229,10 @@ def test_unsupported_language_emits_warning(self, db: Session):
)
assert native_config.provider == "google-native"
- assert "language" not in native_config.params # dropped
- assert len(warnings) == 1
- assert "xx-YY" in warnings[0]
+ assert "language" not in native_config.params
+ assert warnings == [
+ "Unsupported language 'xx-YY' for Gemini TTS, using auto-detect"
+ ]
class TestTransformGoogleGCPRouting:
@@ -1413,7 +1451,7 @@ def test_transform_google_text_completion(self, db: Session):
assert result.provider == "google-aistudio-native"
assert result.type == "text"
assert result.params["model"] == "gemini-2.5-pro"
- assert result.params["temperature"] == 0.7
+ assert "temperature" not in result.params
assert result.params["reasoning"] == "high"
assert warnings == []
@@ -1457,6 +1495,6 @@ def test_transform_google_tts_completion(self, db: Session):
assert result.type == "tts"
assert result.params["model"] == "gemini-2.5-flash-preview-tts"
assert result.params["voice"] == "Kore"
- assert result.params["language"] == "hi" # Mapped from hi-IN
+ assert result.params["language"] == "hi-IN"
assert result.params["response_format"] == "wav" # Default
assert warnings == []
diff --git a/backend/app/tests/services/llm/test_multimodal.py b/backend/app/tests/services/llm/test_multimodal.py
index b7e8acce2..b34574192 100644
--- a/backend/app/tests/services/llm/test_multimodal.py
+++ b/backend/app/tests/services/llm/test_multimodal.py
@@ -1,6 +1,8 @@
import pytest
from unittest.mock import MagicMock
+from google.genai.interactions import Interaction
+
from app.models.llm.request import (
TextInput,
AudioInput,
@@ -224,15 +226,12 @@ class TestGoogleAIFormatParts:
def test_text_part(self):
parts = [TextContent(value="hello")]
result = GoogleAIProvider.format_parts(parts)
- assert result == [{"text": "hello"}]
+ assert result == [{"type": "text", "text": "hello"}]
def test_image_base64_part(self):
parts = [ImageContent(format="base64", value="abc123", mime_type="image/png")]
result = GoogleAIProvider.format_parts(parts)
- assert len(result) == 1
- assert result[0] == {
- "inline_data": {"data": "abc123", "mime_type": "image/png"}
- }
+ assert result == [{"type": "image", "data": "abc123", "mime_type": "image/png"}]
def test_image_url_part(self):
parts = [
@@ -243,22 +242,27 @@ def test_image_url_part(self):
)
]
result = GoogleAIProvider.format_parts(parts)
- assert result[0] == {
- "file_data": {
- "file_uri": "https://example.com/img.jpg",
+ assert result == [
+ {
+ "type": "image",
+ "uri": "https://example.com/img.jpg",
"mime_type": "image/jpeg",
- "display_name": None,
}
- }
+ ]
+
+ def test_image_without_mime_type_omits_key(self):
+ parts = [ImageContent(format="url", value="https://example.com/img.jpg")]
+ result = GoogleAIProvider.format_parts(parts)
+ assert result == [{"type": "image", "uri": "https://example.com/img.jpg"}]
def test_pdf_base64_part(self):
parts = [
PDFContent(format="base64", value="pdf123", mime_type="application/pdf")
]
result = GoogleAIProvider.format_parts(parts)
- assert result[0] == {
- "inline_data": {"data": "pdf123", "mime_type": "application/pdf"}
- }
+ assert result == [
+ {"type": "document", "data": "pdf123", "mime_type": "application/pdf"}
+ ]
def test_pdf_url_part(self):
parts = [
@@ -269,13 +273,13 @@ def test_pdf_url_part(self):
)
]
result = GoogleAIProvider.format_parts(parts)
- assert result[0] == {
- "file_data": {
- "file_uri": "https://example.com/doc.pdf",
+ assert result == [
+ {
+ "type": "document",
+ "uri": "https://example.com/doc.pdf",
"mime_type": "application/pdf",
- "display_name": None,
}
- }
+ ]
def test_mixed_parts(self):
parts = [
@@ -286,10 +290,9 @@ def test_mixed_parts(self):
PDFContent(format="base64", value="pdf", mime_type="application/pdf"),
]
result = GoogleAIProvider.format_parts(parts)
- assert len(result) == 3
- assert "text" in result[0]
- assert "file_data" in result[1]
- assert "inline_data" in result[2]
+ assert [item["type"] for item in result] == ["text", "image", "document"]
+ assert result[1]["uri"] == "https://img.com/a.jpg"
+ assert result[2]["data"] == "pdf"
class TestResolveImageContent:
@@ -433,15 +436,24 @@ def test_string_input(self):
class TestGoogleAIExecuteTextRouting:
def _make_provider(self):
mock_client = MagicMock()
- mock_resp = MagicMock()
- mock_resp.response_id = "resp_gai_123"
- mock_resp.model_version = "gemini-2.0-flash"
- mock_resp.text = "response text"
- mock_resp.usage_metadata.prompt_token_count = 10
- mock_resp.usage_metadata.candidates_token_count = 5
- mock_resp.usage_metadata.total_token_count = 15
- mock_resp.usage_metadata.thoughts_token_count = 0
- mock_client.models.generate_content.return_value = mock_resp
+ mock_client.interactions.create.return_value = Interaction.model_validate(
+ {
+ "id": "int_gai_123",
+ "status": "completed",
+ "model": "gemini-2.0-flash",
+ "steps": [
+ {
+ "type": "model_output",
+ "content": [{"type": "text", "text": "response text"}],
+ }
+ ],
+ "usage": {
+ "total_input_tokens": 10,
+ "total_output_tokens": 5,
+ "total_tokens": 15,
+ },
+ }
+ )
return GoogleAIProvider(client=mock_client), mock_client
def _make_config(self, **extra_params):
@@ -454,6 +466,13 @@ def _make_config(self, **extra_params):
def _make_query(self):
return QueryParams(input="test")
+ @staticmethod
+ def _user_content(mock_client):
+ user_input = mock_client.interactions.create.call_args.kwargs["input"]
+ assert len(user_input) == 1
+ assert user_input[0]["type"] == "user_input"
+ return user_input[0]["content"]
+
def test_multimodal_input(self):
provider, mock_client = self._make_provider()
mm = MultiModalInput(
@@ -468,9 +487,10 @@ def test_multimodal_input(self):
resolved_input=mm,
)
assert error is None
- call_kwargs = mock_client.models.generate_content.call_args[1]
- assert call_kwargs["contents"][0]["role"] == "user"
- assert len(call_kwargs["contents"][0]["parts"]) == 2
+ assert self._user_content(mock_client) == [
+ {"type": "text", "text": "describe"},
+ {"type": "image", "data": "img", "mime_type": "image/png"},
+ ]
def test_list_input(self):
provider, mock_client = self._make_provider()
@@ -481,8 +501,9 @@ def test_list_input(self):
resolved_input=parts,
)
assert error is None
- call_kwargs = mock_client.models.generate_content.call_args[1]
- assert call_kwargs["contents"][0]["role"] == "user"
+ assert self._user_content(mock_client) == [
+ {"type": "image", "data": "img", "mime_type": "image/png"}
+ ]
def test_string_input(self):
provider, mock_client = self._make_provider()
@@ -492,11 +513,10 @@ def test_string_input(self):
resolved_input="hello",
)
assert error is None
- call_kwargs = mock_client.models.generate_content.call_args[1]
- assert call_kwargs["contents"][0]["parts"] == [{"text": "hello"}]
+ assert self._user_content(mock_client) == [{"type": "text", "text": "hello"}]
+ assert response.response.output.content.value == "response text"
def test_missing_model_falls_back_to_default(self):
- """No model in params → provider uses DEFAULT_TEXT_MODELS['google']."""
provider, mock_client = self._make_provider()
config = NativeCompletionConfig(
provider="google-aistudio-native", type="text", params={}
@@ -508,10 +528,11 @@ def test_missing_model_falls_back_to_default(self):
)
assert error is None
assert response is not None
- call_kwargs = mock_client.models.generate_content.call_args[1]
- assert call_kwargs["model"] == "gemini-2.5-pro"
+ assert mock_client.interactions.create.call_args.kwargs["model"] == (
+ "gemini-2.5-pro"
+ )
- def test_instructions_passed_to_config(self):
+ def test_instructions_passed_as_system_instruction(self):
provider, mock_client = self._make_provider()
response, error = provider.execute(
completion_config=self._make_config(instructions="be helpful"),
@@ -519,14 +540,12 @@ def test_instructions_passed_to_config(self):
resolved_input="hello",
)
assert error is None
- call_kwargs = mock_client.models.generate_content.call_args[1]
- config = call_kwargs["config"]
- assert config.system_instruction == "be helpful"
+ call_kwargs = mock_client.interactions.create.call_args.kwargs
+ assert call_kwargs["system_instruction"] == "be helpful"
- def test_no_usage_metadata(self):
+ def test_no_usage(self):
provider, mock_client = self._make_provider()
- mock_resp = mock_client.models.generate_content.return_value
- mock_resp.usage_metadata = None
+ mock_client.interactions.create.return_value.usage = None
response, error = provider.execute(
completion_config=self._make_config(),
query=self._make_query(),
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index caf23c32b..095f43b83 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -44,7 +44,7 @@ dependencies = [
"celery>=5.3.0,<6.0.0",
"redis>=5.0.0,<6.0.0",
"flower>=2.0.1",
- "google-genai>=1.59.0",
+ "google-genai>=2.25.0",
"sarvamai>=0.1.25",
"pydub>=0.25.1",
"jiwer>=3.1.0",
diff --git a/backend/uv.lock b/backend/uv.lock
index 675933ab5..87072d6f1 100644
--- a/backend/uv.lock
+++ b/backend/uv.lock
@@ -338,7 +338,7 @@ requires-dist = [
{ name = "gevent", specifier = ">=25.9.1" },
{ name = "google-auth", specifier = ">=2.49.1" },
{ name = "google-cloud-storage", specifier = ">=3.10.1" },
- { name = "google-genai", specifier = ">=1.59.0" },
+ { name = "google-genai", specifier = ">=2.25.0" },
{ name = "httpx", specifier = ">=0.25.1,<1.0.0" },
{ name = "indic-nlp-library", specifier = ">=0.92" },
{ name = "jinja2", specifier = ">=3.1.4,<4.0.0" },
@@ -1373,15 +1373,15 @@ wheels = [
[[package]]
name = "google-auth"
-version = "2.49.1"
+version = "2.61.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "cryptography" },
{ name = "pyasn1-modules" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/ea/80/6a696a07d3d3b0a92488933532f03dbefa4a24ab80fb231395b9a2a1be77/google_auth-2.49.1.tar.gz", hash = "sha256:16d40da1c3c5a0533f57d268fe72e0ebb0ae1cc3b567024122651c045d879b64", size = 333825, upload-time = "2026-03-12T19:30:58.135Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/0b/9788e913f2202da49068c27ce821eebcf96319240a89d7bb11f206d6471f/google_auth-2.61.0.tar.gz", hash = "sha256:37f0815967322e8c32b12bf422531e8b637cafdaae0acbb9141117cfe6a96f23", size = 398379, upload-time = "2026-10-07T20:13:14.827Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/e9/eb/c6c2478d8a8d633460be40e2a8a6f8f429171997a35a96f81d3b680dec83/google_auth-2.49.1-py3-none-any.whl", hash = "sha256:195ebe3dca18eddd1b3db5edc5189b76c13e96f29e73043b923ebcf3f1a860f7", size = 240737, upload-time = "2026-03-12T19:30:53.159Z" },
+ { url = "https://files.pythonhosted.org/packages/1e/68/c2ff6043d93fda69e5c12a834d0be3b3fa6c762e89732160532566be621b/google_auth-2.61.0-py3-none-any.whl", hash = "sha256:ca60266a37475ae68b63bac007272f46094b3d571abe92aded329b6cfb568025", size = 268021, upload-time = "2026-10-07T20:13:13.2Z" },
]
[package.optional-dependencies]
@@ -1444,7 +1444,7 @@ wheels = [
[[package]]
name = "google-genai"
-version = "1.67.0"
+version = "2.29.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@@ -1458,9 +1458,9 @@ dependencies = [
{ name = "typing-extensions" },
{ name = "websockets" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/a2/07/59a498f81f2c7b0649eacda2ea470b7fd8bd7149f20caba22962081bdd51/google_genai-1.67.0.tar.gz", hash = "sha256:897195a6a9742deb6de240b99227189ada8b2d901d61bdfba836c3092021eab6", size = 506972, upload-time = "2026-03-12T20:39:16.241Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/21/2b/977616ef350e51d3d263eb43be23a91be793544cd937b51c0fac313be233/google_genai-2.29.0.tar.gz", hash = "sha256:124d3eef9ff6b9e4ea548f95ff5b7721683e7990c0a936b006d9e2ab61ea897d", size = 706902, upload-time = "2026-10-07T22:22:57.153Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/6e/c2/562aa1f086e53529ffbeb5b43d5d8bc42c1b968102b5e2163fad005ce298/google_genai-1.67.0-py3-none-any.whl", hash = "sha256:58b0484ff2d4335fa53c724b489e9f807fcca8115d9cdbd8fdf341121fbd6d2d", size = 733542, upload-time = "2026-03-12T20:39:14.615Z" },
+ { url = "https://files.pythonhosted.org/packages/b8/27/a3dede4a1deb21f7fde6a171daa039bf0ea149a5e7826a785e8f109a6e13/google_genai-2.29.0-py3-none-any.whl", hash = "sha256:7adda55cb6746fb46d38ad3202c3372840b96eede964b5fb3df1469cd4998570", size = 1167603, upload-time = "2026-10-07T22:22:55.215Z" },
]
[[package]]
diff --git a/docs/wiki/modules/llm-call.md b/docs/wiki/modules/llm-call.md
index de5749b90..e74c761c5 100644
--- a/docs/wiki/modules/llm-call.md
+++ b/docs/wiki/modules/llm-call.md
@@ -28,13 +28,13 @@ All paths relative to `backend/app/`.
- `LLMCallConfig` — one-of: saved reference (`id` + `version`) XOR ad-hoc `blob` (validator-enforced)
- `ConfigBlob` — `completion` + optional `prompt_template` (`PromptTemplate.template`, plain string; `{{input}}` interpolation is llm-chain-only) + `input_guardrails`/`output_guardrails`
- `CompletionConfig` — discriminated union on `provider`: `KaapiCompletionConfig` (standardized params: `TextLLMParams`/`STTLLMParams`/`TTSLLMParams`), `NativeCompletionConfig` (pass-through), `ProxyCompletionConfig` (client's own endpoint)
-- `TextLLMParams` reasoning knobs: `reasoning` + `effort` (OpenAI-style), `thinking` (Anthropic adaptive-thinking container, forwarded as-is) and `thinking_level` (Gemini). A knob must be **declared here** to survive config save — pydantic's default extra policy is ignore, so an undeclared key is dropped silently at validation with no error.
+- `TextLLMParams` reasoning knobs: `reasoning` + `effort` (OpenAI-style), `thinking` (Anthropic adaptive-thinking container, forwarded as-is) and `thinking_level` (Gemini: `minimal`/`low`/`medium`/`high`; `minimal` is flash-only). A knob must be **declared here** to survive config save — pydantic's default extra policy is ignore, so an undeclared key is dropped silently at validation with no error.
- `QueryParams` — per-call input + `ConversationConfig`
- `models/guardrails/` — validator config shapes
## Services / CRUD
- `services/llm/` — `mappers.py` (Kaapi params → provider API), `providers/`, `chain/`, `guardrails.py`, `jobs.py`
-- `mappers.py` is the **only** param mapper in the codebase; the assessment fork was merged back into it. Callers: `crud/evaluations/{batch,fast,judge}.py`, `crud/assessment/batch.py`, `services/assessment/api/batch.py`, `services/assessment/prefilter/request_builder.py`. Structured output is keyed `output_schema` for every provider (OpenAI `text.format`, Anthropic `output_config.format`, Gemini `output_schema`), and Anthropic reads `effort` into `output_config.effort`. Every param that only assessment set (`top_p`, `max_output_tokens`, `thinking_level`, `thinking`, `output_schema`) defaults to `None` on `TextLLMParams` and is dropped by `ParamSerialization._dump_compact`, so a config that does not set it maps exactly as before. Anthropic never receives `temperature`/`top_p`: the Messages API returns 400 for a non-default sampling value on every Claude model Kaapi serves, so the mapper drops both, warning only when the value was not Anthropic's own default of 1.0. `normalize_llm_text` no longer lives here; it moved to `services/assessment/validators.py`, the only place that calls it.
+- `mappers.py` is the **only** param mapper in the codebase; the assessment fork was merged back into it. Callers: `crud/evaluations/{batch,fast,judge}.py`, `crud/assessment/batch.py`, `services/assessment/api/batch.py`, `services/assessment/prefilter/request_builder.py`. Structured output is keyed `output_schema` for every provider (OpenAI `text.format`, Anthropic `output_config.format`, Gemini `output_schema`), and Anthropic reads `effort` into `output_config.effort`. Every param that only assessment set (`top_p`, `max_output_tokens`, `thinking_level`, `thinking`, `output_schema`) defaults to `None` on `TextLLMParams` and is dropped by `ParamSerialization._dump_compact`, so a config that does not set it maps exactly as before. Anthropic never receives `temperature`/`top_p`: the Messages API returns 400 for a non-default sampling value on every Claude model Kaapi serves, so the mapper drops both, warning only when the value was not Anthropic's own default of 1.0. Gemini likewise never receives `temperature`/`top_p` for text or STT: newer models return 400 for them and they have been no-ops since 3.6 Flash, so the mapper does not send them. Gemini TTS `language` maps through `BCP47_LOCALE_TO_GEMINI_LANG` to a full locale (`en-IN`, not `en`, so the accent survives; Kaapi's `od-IN` → `or-IN`); an unlisted tag (e.g. `as-IN`) is dropped with a warning and Gemini auto-detects. `normalize_llm_text` no longer lives here; it moved to `services/assessment/validators.py`, the only place that calls it.
- `services/guardrails/` — validator execution
- `crud/llm.py`, `crud/llm_chain.py`, `crud/config/` — persistence
@@ -43,6 +43,7 @@ All paths relative to `backend/app/`.
## External
- OpenAI / Gemini / Anthropic (via `services/llm/providers/`), proxy endpoints, Langfuse tracing.
+- Gemini has two transports: AI Studio (`providers/google_aistudio.py`) calls the **Interactions API** (`client.interactions.create`; TTS strips the WAV header when present — 3.8 models return WAV, preview models return raw 24 kHz PCM and reject `audio/l16` — then Kaapi re-wraps to wav/mp3/ogg), while Vertex (`providers/google_gcp.py`) stays on raw REST **generateContent**. Thinking level resolves the same way in both: `thinking_config.thinking_level` → `reasoning`, omitted if neither; STT defaults to `low`. Interactions errors come from `google.genai._gaos.lib.compat_errors`, not `google.genai.errors` (which `files.upload` still raises).
## Gotchas
- Saved config resolution = `config_id` + `version` pinned; ad-hoc blob never persisted.