Intégration de l'API Gemini dans un bot Telegram : guide pour développeurs

Intégration asynchrone de l’API Gemini dans un bot Telegram : Guide pratique avec aiogram 3.x

Lorsque votre bot Telegram cesse soudainement de répondre à cause d’une erreur 429 de l’API Gemini, ce n’est pas qu’un bug — c’est de la dette technique. Dans cet article, nous décomposons le problème étape par étape : comment intégrer correctement Gemini dans un bot asynchrone sur aiogram 3.x pour éviter de bloquer la boucle d’événements, contourner les limites RPM/TPM, et implémenter les appels de fonctions sans sacrifier les performances.

Architecture : Du gestionnaire simple à la solution de production

Le flux d’interaction utilisateur-bot de base via Gemini semble simple : requête → traitement → réponse. Mais les scénarios réels exigent des couches supplémentaires. Appeler une API synchrone de manière asynchrone sans encapsulage approprié fige la boucle d’événements d’aiogram, surtout sous forte charge. C’est pourquoi les composants clés sont :

• File d’attente asyncio — pour tamponner les requêtes entrantes
• Limiteur de débit — pour respecter les limites de Gemini (5–60 RPM)
• Cache de réponses — pour économiser des jetons sur les requêtes répétées
• Gestionnaire de streaming — pour contourner les timeouts Telegram (10 secondes)

L’API Gemini impose des limites strictes même sur les plans payants : la latence maximale de réponse peut atteindre 15 secondes, et le niveau gratuit est plafonné à seulement 5 requêtes par minute. Sans files d’attente et cache, votre expérience utilisateur sera fichue.

Configuration du client : SDK synchrone dans un environnement asynchrone

Le SDK Google GenAI reste synchrone malgré les espoirs de la communauté. Appeler generate_content() directement dans un gestionnaire aiogram bloque toute la boucle d’événements. La solution : encapsuler les appels dans asyncio.to_thread ou utiliser run_in_executor.

import asyncio
from google import genai

class AsyncGeminiClient:
    def __init__(self, model: str = "gemini-3-flash-preview"):
        self.client = genai.Client()
        self.model = model
    
    async def generate(self, prompt: str) -> str:
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(
            None,
            self._sync_generate,
            prompt
        )
        return response
    
    def _sync_generate(self, prompt: str) -> str:
        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt
        )
        return response.text

Ceci est un wrapper minimal fonctionnel. Pour la production, ajoutez la gestion des exceptions et les métriques d’exécution.

Intégration avec aiogram : De /start aux réponses en streaming

Un gestionnaire de base doit faire plus que simplement accepter les messages : gérer l’état en affichant « typing », diviser les réponses longues, et mettre en cache les résultats. Telegram limite la longueur des messages à 4096 caractères — ignorez cela et vous aurez des coupures.

@router.message()
async def handle_message(message: types.Message):
    await message.bot.send_chat_action(chat_id=message.chat.id, action="typing")
    
    try:
        response = await gemini.generate(message.text)
        if len(response) > 4000:
            for i in range(0, len(response), 4000):
                await message.answer(response[i:i+4000])
        else:
            await message.answer(response)
    except Exception as e:
        logging.error(f"Gemini error: {e}")
        await message.answer("Chto-then poshlo not so. Poprobuyte pozzhe.")

Pour les requêtes longues, utilisez le streaming — mettez à jour un seul message au fur et à mesure que le texte est généré. Cela évite les timeouts et améliore l’UX.

Appels de fonctions : Quand Gemini devient un agent

Function Calling permet au modèle d’invoquer des API externes en fonction du contexte de la requête. Par exemple, si un utilisateur dit « réservez une réunion pour demain à 15 h », Gemini peut analyser les paramètres et appeler votre fonction schedule_meeting.

Étapes clés :

• Décrire la fonction au format JSON Schema — pour que le modèle comprenne sa signature.
• Passer la description dans GenerateContentConfig sous tools.
• Gérer function_call dans la réponse et exécuter l’appel réel.
• Renvoyer le résultat dans le chat comme partie de la conversation.

Exemple de déclaration de fonction :

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Withzdayot meetingsu with ukazannymi uchastnikami",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

Cela transforme un bot statique en assistant dynamique capable d’interagir avec des systèmes externes.

Cache et limitation de débit : Protection contre la faillite et les bannissements

Gemini facture par jetons. Sans cache, vous payez pour chaque question répétée. Un cache en mémoire simple utilisant le hachage MD5 du prompt avec TTL résout cela :

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # check TTL and vozvrat values
    
    def set(self, key: str, value: str):
        self._cache[key] = (value, datetime.now())
    
    @staticmethod
    def hash_prompt(prompt: str) -> str:
        return hashlib.md5(prompt.lower().strip().encode()).hexdigest()

La limitation de débit est essentielle même sur les niveaux payants. Implémentez-la comme gestionnaire de contexte asynchrone ou middleware :

class AsyncRateLimiter:
    def __init__(self, max_requests: int, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = []
        self._lock = asyncio.Lock()
    
    async def acquire(self):
        async with self._lock:
            now = time.time()
            self.requests = [t for t in self.requests if now - t < self.time_window]
            if len(self.requests) >= self.max_requests:
                sleep_time = self.time_window - (now - self.requests[0])
                await asyncio.sleep(sleep_time + 0.1)
                return await self.acquire()
            self.requests.append(now)

Erreurs courantes et comment les éviter

Trois pièges classiques que tout le monde rencontre :

• Erreur 429 (RESOURCE_EXHAUSTED) — déclenchée par dépassement de RPM, TPM ou RPD. Solution : limiteur de débit local + backoff exponentiel.
• Timeouts Telegram — le serveur attend 10 secondes pour une réponse. Si Gemini est lent, utilisez le streaming ou envoyez des messages intermédiaires.
• Modèles obsolètes — Google retire régulièrement les anciennes versions. En mars 2026, les modèles Pro sont passés en abonnement uniquement. Vérifiez toujours la disponibilité du modèle avant déploiement.

Points clés :

• Le SDK Gemini synchrone nécessite un encapsulage dans run_in_executor — sinon vous bloquez la boucle d’événements.
• Implémentez la limitation de débit côté bot, ne comptez pas sur l’API.
• Le streaming est la seule façon de contourner le timeout de 10 secondes de Telegram pour les réponses longues.
• Le cache économise de l’argent et réduit la latence sur les répétitions.
• Function Calling fait évoluer votre bot d’un générateur de texte vers un agent complet.

Adopter ces pratiques réduit les incidents en production et maintient la stabilité à mesure que la charge augmente. Ne lésinez pas sur l’architecture — économisez sur les jetons à la place.

— Editorial Team