Gemini gratuit avec bascule automatique vers Gemini payant pour le STT et l’agent de conversation
Mise à jour 20.09.26 : j’ai ajouté Secure Assist, une petite intégration qui gère automatiquement un moteur principal et un moteur de secours pour la Conversation et le STT. Projet encore en test, si quelqu'un veut tester.
GitHub :
Petite évolution de mon projet Ikabot (petit nom de mon assistant HA).
J’utilise Gemini à deux endroits dans mon pipeline Home Assistant :
- pour le STT, donc transformer ce que je dis en texte ;
- pour l’agent de conversation, qui traite ensuite la demande.
J’utilise en priorité mes configurations Gemini gratuites.
Le problème est qu’il arrive que le service gratuit renvoie une erreur ou soit momentanément indisponible. Dans un assistant vocal, si le STT ou l’agent de conversation tombe, toute la chaîne s’arrête.
Je voulais donc quelque chose d’assez simple :
Gemini gratuit
↓
OK
↓
on continue normalement
Et si ça ne fonctionne pas :
Gemini gratuit
↓
erreur
↓
Gemini payant
↓
on continue
Le but est donc de rester sur le gratuit en fonctionnement normal et de n’utiliser le payant qu’en secours.
Je précise dès le départ que tout ça est encore en test chez moi.
Je ne suis pas développeur Home Assistant et je ne prétends pas maîtriser toutes les subtilités de l’API interne de HA. J’ai fait ça pour répondre à mon besoin, ça fonctionne actuellement chez moi, mais il y a certainement des choses qui peuvent être améliorées ou faites plus proprement.
Donc si certains connaissent mieux cette partie de Home Assistant, voient une erreur ou ont une meilleure méthode, je suis clairement preneur des remarques et améliorations.
Principe
Chez moi j’ai deux configurations Google AI.
La première est utilisée normalement :
Google AI Conversation gratuit
conversation.google_ai_conversation_2
Google AI STT gratuit
stt.google_ai_stt_2
La deuxième utilise mon accès avec facturation activée :
Google AI Conversation payant
conversation.google_ai_conversation
Google AI STT payant
stt.google_ai_stt
J’ai ensuite créé une petite intégration personnalisée :
Ikabot Gemini Fallback
Elle crée deux nouveaux éléments utilisables dans le pipeline Assist :
Ikabot Gemini secours
pour l’agent de conversation,
et :
Ikabot STT secours
pour la reconnaissance vocale.
Le pipeline utilise donc uniquement ces deux éléments.
C’est ensuite l’intégration qui essaye le Gemini gratuit et qui bascule automatiquement vers le payant en cas de problème.
Pour le STT
Micro
↓
Ikabot STT secours
↓
Google AI STT gratuit
↓
si erreur
↓
Google AI STT payant
↓
texte
Pour l’agent de conversation
Texte
↓
Ikabot Gemini secours
↓
Google AI Conversation gratuit
↓
si erreur
↓
Google AI Conversation payant
↓
réponse
Le changement est donc transparent pour le pipeline.
Création de l’intégration
Dans :
/config/custom_components/
j’ai créé le dossier :
ikabot_gemini_fallback
Ce qui donne :
/config/custom_components/ikabot_gemini_fallback/
Avec quatre fichiers :
ikabot_gemini_fallback/
├── __init__.py
├── config_flow.py
├── manifest.json
└── stt.py
Le __init__.py gère principalement le fallback de l’agent de conversation.
Le stt.py s’occupe du fallback STT.
manifest.json
Fichier :
/config/custom_components/ikabot_gemini_fallback/manifest.json
Contenu :
{
"domain": "ikabot_gemini_fallback",
"name": "Ikabot Gemini Fallback",
"version": "1.1.0",
"documentation": "https://www.home-assistant.io/",
"dependencies": [
"conversation",
"stt"
],
"after_dependencies": [
"google_generative_ai_conversation"
],
"codeowners": [],
"config_flow": true,
"single_config_entry": true,
"integration_type": "service",
"iot_class": "local_push"
}
L’intégration dépend donc des composants conversation et stt.
J’ai également ajouté :
"after_dependencies": [
"google_generative_ai_conversation"
]
afin que Google AI soit chargé avant notre intégration de fallback.
config_flow.py
Fichier :
/config/custom_components/ikabot_gemini_fallback/config_flow.py
Contenu :
from homeassistant import config_entries
DOMAIN = "ikabot_gemini_fallback"
class IkabotGeminiFallbackConfigFlow(
config_entries.ConfigFlow,
domain=DOMAIN,
):
"""Configuration de Ikabot Gemini Fallback."""
VERSION = 1
async def async_step_user(self, user_input=None):
"""Créer l'intégration de fallback."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
return self.async_create_entry(
title="Ikabot Gemini secours",
data={},
)
Il n’y a volontairement presque rien dedans.
Il sert simplement à pouvoir ajouter l’intégration depuis l’interface Home Assistant.
Pour l’instant, les IDs des moteurs principal et de secours sont directement renseignés dans les fichiers Python.
Une amélioration future pourrait être de pouvoir les sélectionner directement depuis l’interface.
Fallback de l’agent de conversation
init.py
Fichier :
/config/custom_components/ikabot_gemini_fallback/__init__.py
Voici mon fichier actuel :
import logging
from typing import Literal
from homeassistant.components import conversation
from homeassistant.components.conversation import (
AbstractConversationAgent,
ConversationInput,
ConversationResult,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import MATCH_ALL
from homeassistant.core import HomeAssistant
DOMAIN = "ikabot_gemini_fallback"
# Gemini gratuit
PRIMARY_AGENT = "conversation.google_ai_conversation_2"
# Gemini payant
FALLBACK_AGENT = "conversation.google_ai_conversation"
PLATFORMS = ["stt"]
_LOGGER = logging.getLogger(__name__)
def _result_is_error(result: ConversationResult) -> bool:
"""Détecter une réponse d'erreur renvoyée par Home Assistant."""
try:
response_type = getattr(result.response, "response_type", None)
response_type_value = getattr(
response_type,
"value",
response_type,
)
return response_type_value == "error"
except Exception:
return False
class IkabotGeminiFallbackAgent(AbstractConversationAgent):
"""Gemini gratuit avec bascule automatique vers Gemini payant."""
def __init__(self, hass: HomeAssistant) -> None:
self.hass = hass
@property
def supported_languages(self) -> list[str] | Literal["*"]:
return MATCH_ALL
async def async_process(
self,
user_input: ConversationInput,
) -> ConversationResult:
"""Traiter avec Gemini gratuit puis Gemini payant si nécessaire."""
try:
_LOGGER.debug(
"Ikabot Gemini : tentative avec Gemini gratuit (%s)",
PRIMARY_AGENT,
)
result = await conversation.async_converse(
hass=self.hass,
text=user_input.text,
conversation_id=user_input.conversation_id,
context=user_input.context,
language=user_input.language,
agent_id=PRIMARY_AGENT,
device_id=user_input.device_id,
satellite_id=user_input.satellite_id,
extra_system_prompt=user_input.extra_system_prompt,
)
if not _result_is_error(result):
_LOGGER.debug(
"Ikabot Gemini : réponse Gemini gratuit OK"
)
return result
_LOGGER.warning(
"Ikabot Gemini : Gemini gratuit a retourné "
"response_type=error -> bascule vers Gemini payant"
)
except Exception as err:
_LOGGER.warning(
"Ikabot Gemini : exception Gemini gratuit (%s) "
"-> bascule vers Gemini payant",
err,
)
try:
_LOGGER.info(
"Ikabot Gemini : utilisation du secours payant (%s)",
FALLBACK_AGENT,
)
return await conversation.async_converse(
hass=self.hass,
text=user_input.text,
conversation_id=user_input.conversation_id,
context=user_input.context,
language=user_input.language,
agent_id=FALLBACK_AGENT,
device_id=user_input.device_id,
satellite_id=user_input.satellite_id,
extra_system_prompt=user_input.extra_system_prompt,
)
except Exception as err:
_LOGGER.exception(
"Ikabot Gemini : échec également du Gemini payant : %s",
err,
)
raise
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
) -> bool:
"""Charger Ikabot Gemini Fallback."""
agent = IkabotGeminiFallbackAgent(hass)
conversation.async_set_agent(
hass,
entry,
agent,
)
await hass.config_entries.async_forward_entry_setups(
entry,
PLATFORMS,
)
_LOGGER.info(
"Ikabot Gemini Fallback chargé : "
"Conversation gratuit -> payant + STT fallback"
)
return True
async def async_unload_entry(
hass: HomeAssistant,
entry: ConfigEntry,
) -> bool:
"""Décharger Ikabot Gemini Fallback."""
conversation.async_unset_agent(
hass,
entry,
)
unload_ok = await hass.config_entries.async_unload_platforms(
entry,
PLATFORMS,
)
return unload_ok
Comment fonctionne le fallback Conversation
Les deux agents sont définis ici :
PRIMARY_AGENT = "conversation.google_ai_conversation_2"
FALLBACK_AGENT = "conversation.google_ai_conversation"
Il faut évidemment remplacer ces IDs par ceux de votre installation.
Le premier appel est fait vers PRIMARY_AGENT, donc normalement tout passe par le Gemini gratuit.
Certaines erreurs provenant de Gemini ne remontent pas forcément sous forme d’exception Python. Home Assistant peut déjà avoir intercepté l’erreur et renvoyer un ConversationResult avec :
response_type = error
Dans ce cas, le except n’est jamais exécuté.
J’ai donc ajouté une vérification du résultat.
La bascule se fait actuellement dans deux cas :
Exception Python
ou :
ConversationResult avec response_type = error
Si le gratuit échoue, la même demande est renvoyée à l’agent payant avec le même texte, le même contexte, la même langue, le même conversation_id, etc.
Seul l’agent change.
Fallback STT
stt.py
Fichier :
/config/custom_components/ikabot_gemini_fallback/stt.py
Contenu :
import asyncio
import logging
from collections.abc import AsyncIterable
from typing import Any, override
from homeassistant.components import stt
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
_LOGGER = logging.getLogger(__name__)
PRIMARY_STT = "stt.google_ai_stt_2"
FALLBACK_STT = "stt.google_ai_stt"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Créer le moteur STT de secours."""
async_add_entities([IkabotFallbackSTT(hass)])
class IkabotFallbackSTT(stt.SpeechToTextEntity):
"""STT Gemini gratuit avec bascule automatique vers le payant."""
_attr_name = "Ikabot STT secours"
_attr_unique_id = "ikabot_stt_secours"
def __init__(self, hass: HomeAssistant) -> None:
self.hass = hass
self._attr_extra_state_attributes = {
"principal": PRIMARY_STT,
"secours": FALLBACK_STT,
"dernier_moteur": "aucun",
}
def _engine(self, entity_id: str) -> stt.SpeechToTextEntity | None:
"""Récupérer une entité STT existante."""
entity = stt.async_get_speech_to_text_entity(self.hass, entity_id)
if entity is self:
return None
return entity
def _capability(
self,
attribute: str,
fallback: list[Any],
) -> list[Any]:
"""Reprendre les capacités du STT Google déjà chargé."""
for entity_id in (PRIMARY_STT, FALLBACK_STT):
engine = self._engine(entity_id)
if engine is not None:
value = getattr(engine, attribute, None)
if value:
return list(value)
return fallback
@property
@override
def supported_languages(self) -> list[str]:
return self._capability(
"supported_languages",
["fr-FR", "fr-BE", "fr-CH", "fr-CA"],
)
@property
@override
def supported_formats(self) -> list[stt.AudioFormats]:
return self._capability(
"supported_formats",
[stt.AudioFormats.WAV, stt.AudioFormats.OGG],
)
@property
@override
def supported_codecs(self) -> list[stt.AudioCodecs]:
return self._capability(
"supported_codecs",
[stt.AudioCodecs.PCM, stt.AudioCodecs.OPUS],
)
@property
@override
def supported_bit_rates(self) -> list[stt.AudioBitRates]:
return self._capability(
"supported_bit_rates",
[stt.AudioBitRates.BITRATE_16],
)
@property
@override
def supported_sample_rates(self) -> list[stt.AudioSampleRates]:
return self._capability(
"supported_sample_rates",
[stt.AudioSampleRates.SAMPLERATE_16000],
)
@property
@override
def supported_channels(self) -> list[stt.AudioChannels]:
return self._capability(
"supported_channels",
[stt.AudioChannels.CHANNEL_MONO],
)
@property
@override
def audio_processing(self) -> stt.SpeechAudioProcessing:
"""Utiliser les mêmes préférences de traitement que Gemini."""
for entity_id in (PRIMARY_STT, FALLBACK_STT):
engine = self._engine(entity_id)
if engine is not None:
return engine.audio_processing
return stt.DEFAULT_AUDIO_PROCESSING
async def _call_engine(
self,
entity_id: str,
metadata: stt.SpeechMetadata,
audio_data: bytes,
) -> stt.SpeechResult | None:
"""Envoyer une copie du même audio à un moteur STT."""
engine = self._engine(entity_id)
if engine is None:
_LOGGER.warning("STT introuvable : %s", entity_id)
return None
if not engine.available:
_LOGGER.warning("STT indisponible : %s", entity_id)
return None
if not engine.check_metadata(metadata):
_LOGGER.error(
"Métadonnées audio non prises en charge par %s : %s",
entity_id,
metadata,
)
return None
async def replay_audio() -> AsyncIterable[bytes]:
yield audio_data
try:
return await engine.internal_async_process_audio_stream(
metadata=metadata,
stream=replay_audio(),
)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("Exception lors de l'appel STT %s", entity_id)
return None
def _set_last_engine(self, engine_name: str) -> None:
self._attr_extra_state_attributes = {
"principal": PRIMARY_STT,
"secours": FALLBACK_STT,
"dernier_moteur": engine_name,
}
self.async_write_ha_state()
@override
async def async_process_audio_stream(
self,
metadata: stt.SpeechMetadata,
stream: AsyncIterable[bytes],
) -> stt.SpeechResult:
"""Essayer le STT gratuit, puis rejouer l'audio au STT payant."""
audio = bytearray()
async for chunk in stream:
audio.extend(chunk)
if not audio:
_LOGGER.warning("Aucun audio reçu par Ikabot STT secours.")
self._set_last_engine("erreur")
return stt.SpeechResult(None, stt.SpeechResultState.ERROR)
audio_data = bytes(audio)
primary_result = await self._call_engine(
PRIMARY_STT,
metadata,
audio_data,
)
if (
primary_result is not None
and primary_result.result == stt.SpeechResultState.SUCCESS
and primary_result.text
):
self._set_last_engine("gratuit")
_LOGGER.debug("Ikabot STT : transcription réussie avec le gratuit.")
return primary_result
_LOGGER.warning(
"Ikabot STT : moteur gratuit en erreur. "
"Bascule vers le moteur payant."
)
fallback_result = await self._call_engine(
FALLBACK_STT,
metadata,
audio_data,
)
if (
fallback_result is not None
and fallback_result.result == stt.SpeechResultState.SUCCESS
and fallback_result.text
):
self._set_last_engine("payant")
_LOGGER.warning(
"Ikabot STT : transcription réussie avec le moteur payant."
)
return fallback_result
self._set_last_engine("erreur")
_LOGGER.error("Ikabot STT : échec du gratuit ET du payant.")
if fallback_result is not None:
return fallback_result
if primary_result is not None:
return primary_result
return stt.SpeechResult(None, stt.SpeechResultState.ERROR)
Comment fonctionne le fallback STT
Comme pour Conversation, les deux moteurs sont définis au début :
PRIMARY_STT = "stt.google_ai_stt_2"
FALLBACK_STT = "stt.google_ai_stt"
La différence avec l’agent de conversation est qu’on travaille ici avec un flux audio.
Une fois que le premier moteur STT a lu ce flux, on ne peut pas simplement le redonner tel quel au deuxième moteur.
J’ai donc choisi de mettre temporairement l’audio de la phrase en mémoire :
audio = bytearray()
async for chunk in stream:
audio.extend(chunk)
Puis :
audio_data = bytes(audio)
Je peux ensuite rejouer exactement le même audio au moteur gratuit puis, si nécessaire, au moteur payant.
Le moteur est considéré comme ayant correctement fonctionné uniquement si j’obtiens un résultat SUCCESS avec du texte.
Sinon, le même audio est envoyé au moteur de secours.
Suivre quel moteur a réellement répondu
J’ai ajouté trois attributs sur l’entité Ikabot STT secours :
principal
secours
dernier_moteur
Le dernier moteur peut donc prendre par exemple :
gratuit
payant
erreur
C’est surtout pratique actuellement pour les essais et pour vérifier que le fallback a réellement été utilisé.
Installation
Une fois les quatre fichiers créés :
__init__.py
config_flow.py
manifest.json
stt.py
je redémarre complètement Home Assistant.
Ensuite :
Paramètres
→ Appareils et services
→ Ajouter une intégration
Je cherche :
Ikabot Gemini Fallback
et je l’ajoute.
J’obtiens alors mon nouvel agent de conversation et mon nouveau moteur STT de secours.
Configuration du pipeline Assist
Dans le pipeline Assist, je sélectionne :
Reconnaissance vocale
Ikabot STT secours
Agent de conversation
Ikabot Gemini secours
Au final, mon pipeline ressemble donc à ça :
Wake word
↓
Ikabot STT secours
↓
STT gratuit
↓
STT payant si nécessaire
↓
Ikabot Gemini secours
↓
Conversation gratuite
↓
Conversation payante si nécessaire
↓
TTS
Le pipeline lui-même ne change donc jamais de moteur.
C’est le composant intermédiaire qui gère le secours.
Les valeurs à adapter
Dans __init__.py :
PRIMARY_AGENT = "conversation.google_ai_conversation_2"
FALLBACK_AGENT = "conversation.google_ai_conversation"
Dans stt.py :
PRIMARY_STT = "stt.google_ai_stt_2"
FALLBACK_STT = "stt.google_ai_stt"
Ce sont mes propres IDs.
Il faut donc bien récupérer ceux présents dans votre Home Assistant avant de reprendre le code.
Quelques limites actuelles
Le système fonctionne actuellement chez moi mais il reste volontairement assez simple.
Les moteurs principal et secondaire sont encore écrits directement dans le code Python.
Une évolution intéressante serait d’avoir une page de configuration permettant de sélectionner directement :
Conversation principale
Conversation de secours
STT principal
STT de secours
sans modifier les fichiers.
Le STT conserve aussi temporairement en mémoire l’intégralité de la phrase afin de pouvoir la rejouer au moteur payant.
Pour quelques secondes de voix ça ne me pose pas de problème, mais c’est quand même un point à connaître.
Enfin, on utilise ici des classes et méthodes internes de Home Assistant. Il est donc tout à fait possible qu’une future mise à jour nécessite une adaptation du composant.
Ce qu’il me reste à tester
Je veux encore observer le comportement sur plusieurs types d’erreurs :
503
quota atteint
timeout
perte de connexion
moteur indisponible
Je veux également vérifier sur la durée que le payant n’est appelé que lorsqu’il est réellement nécessaire.
Une autre évolution qui pourrait être intéressante serait d’ajouter quelques compteurs :
requêtes passées par le gratuit
nombre de bascules vers le payant
nombre d’échecs complets
Ça permettrait de voir si ce fallback sert réellement souvent.
Pour finir
Comme pour le reste de mon projet Ikabot, je partage surtout ici ce que j’ai testé et ce qui fonctionne chez moi.
Ce n’est pas une intégration officielle et je ne considère pas encore ça comme terminé.
Pour l’instant j’obtiens bien le fonctionnement que je cherchais :
STT Gemini gratuit
→ payant si erreur
Conversation Gemini gratuite
→ payante si erreur
et surtout, tout cela reste transparent pour l’assistant vocal.
Je ne maîtrise clairement pas toutes les subtilités internes de Home Assistant, donc si vous voyez une mauvaise pratique, quelque chose qui risque de poser problème ou une façon plus propre de faire, je suis preneur.
Si ça peut servir de base à quelqu’un, ou être amélioré collectivement, tant mieux.
Et au moins maintenant, quand Gemini gratuit décide de faire une pause café, Ikabot a une roue de secours. 
