Integration der Gemini API in einen Telegram-Bot: Leitfaden für Entwickler

Asynchrone Integration der Gemini-API in einen Telegram-Bot: Ein praktischer Leitfaden mit aiogram 3.x

Wenn Ihr Telegram-Bot aufgrund eines 429-Fehlers der Gemini-API plötzlich nicht mehr reagiert, ist das nicht nur ein Bug – es ist technische Schuld. In diesem Artikel gehen wir Schritt für Schritt vor: Wie Sie Gemini richtig in einen asynchronen Bot mit aiogram 3.x integrieren, um die Ereignisschleife nicht zu blockieren, RPM/TPM-Limits zu umgehen und Funktionsaufrufe ohne Leistungseinbußen zu implementieren.

Architektur: Vom einfachen Handler zur Produktionslösung

Der grundlegende Benutzer-Bot-Interaktionsfluss über Gemini sieht einfach aus: Anfrage → Verarbeitung → Antwort. Aber reale Szenarien erfordern zusätzliche Schichten. Das synchrone Aufrufen einer API ohne ordnungsgemäße Wrapper friert die aiogram-Ereignisschleife ein, besonders unter hoher Last. Deshalb sind die Schlüsselkomponenten:

• Asyncio Queue – zum Puffern eingehender Anfragen
• Rate Limiter – zur Einhaltung der Gemini-Limits (5–60 RPM)
• Response Cache – zum Sparen von Tokens bei wiederholten Anfragen
• Streaming Handler – zur Umgehung von Telegram-Zeitüberschreitungen (10 Sekunden)

Die Gemini-API erzwingt selbst bei kostenpflichtigen Tarifen strenge Limits: Die maximale Antwortlatenz kann 15 Sekunden betragen, und der kostenlose Tarif ist auf nur 5 Anfragen pro Minute begrenzt. Ohne Warteschlangen und Caching wird Ihre Benutzererfahrung darunter leiden.

Client-Einrichtung: Synchrones SDK in einer asynchronen Umgebung

Das Google GenAI SDK bleibt trotz der Hoffnungen der Community synchron. Das bedeutet, dass ein direkter Aufruf von generate_content() in einem aiogram-Handler die gesamte Ereignisschleife blockiert. Die Lösung: Wickeln Sie die Aufrufe in asyncio.to_thread oder verwenden Sie 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

Dies ist ein minimaler funktionsfähiger Wrapper. Fügen Sie für die Produktion Ausnahmebehandlung und Ausführungsmetriken hinzu.

Integration mit aiogram: Vom /start zum Streaming von Antworten

Ein grundlegender Handler muss mehr tun, als nur Nachrichten zu akzeptieren: Zustand verwalten, indem er „tippt" anzeigt, lange Antworten aufteilen und Ergebnisse zwischenspeichern. Telegram begrenzt die Nachrichtenlänge auf 4096 Zeichen – ignorieren Sie dies, und Sie erhalten abgeschnittene Antworten.

@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-Fehler: {e}")
        await message.answer("Irgendwas ist schiefgelaufen. Versuchen Sie es später noch einmal.")

Verwenden Sie für langlaufende Anfragen Streaming – aktualisieren Sie eine einzelne Nachricht, während der Text generiert wird. Dies umgeht Zeitüberschreitungen und verbessert die Benutzererfahrung.

Funktionsaufrufe: Wenn Gemini zum Agenten wird

Function Calling ermöglicht es dem Modell, basierend auf dem Anforderungskontext externe APIs aufzurufen. Wenn ein Benutzer beispielsweise sagt „Buche ein Meeting für morgen um 15 Uhr", kann Gemini die Parameter parsen und Ihre schedule_meeting-Funktion aufrufen.

Wichtige Schritte:

• Beschreiben Sie die Funktion im JSON-Schema-Format – damit das Modell ihre Signatur versteht.
• Übergeben Sie die Beschreibung in GenerateContentConfig als tools.
• Behandeln Sie function_call in der Antwort und führen Sie den eigentlichen Aufruf aus.
• Geben Sie das Ergebnis als Teil der Konversation zurück in den Chat.

Beispiel für eine Funktionsdeklaration:

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Erstellt ein Meeting mit den angegebenen Teilnehmern",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

Dies verwandelt einen statischen Bot in einen dynamischen Assistenten, der mit externen Systemen interagieren kann.

Caching und Ratenbegrenzung: Schutz vor Bankrott und Sperren

Gemini berechnet nach Tokens. Ohne Caching zahlen Sie für jede wiederholte Frage. Ein einfacher In-Memory-Cache mit MD5-Hash des Prompts und TTL behebt das:

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # TTL prüfen und Werte zurückgeben
    
    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()

Ratenbegrenzung ist auch bei kostenpflichtigen Tarifen unerlässlich. Implementieren Sie sie als asynchronen Kontextmanager oder 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)

Häufige Fehler und wie man sie vermeidet

Drei klassische Fallstricke, auf die jeder stößt:

• Fehler 429 (RESOURCE_EXHAUSTED) – ausgelöst durch Überschreitung von RPM, TPM oder RPD. Lösung: Lokaler Ratenbegrenzer + exponentielles Backoff.
• Telegram-Zeitüberschreitungen – Der Server wartet 10 Sekunden auf eine Antwort. Wenn Gemini langsam ist, verwenden Sie Streaming oder senden Sie Zwischennachrichten.
• Veraltete Modelle – Google stellt regelmäßig alte Versionen ein. Im März 2026 wurden Pro-Modelle auf Abonnement umgestellt. Überprüfen Sie immer die Modellverfügbarkeit vor dem Einsatz.

Wichtige Erkenntnisse:

• Das synchrone Gemini-SDK benötigt run_in_executor-Wrapper – sonst blockieren Sie die Ereignisschleife.
• Implementieren Sie die Ratenbegrenzung auf der Bot-Seite, verlassen Sie sich nicht auf die API.
• Streaming ist der einzige Weg, um Telegrams 10-Sekunden-Timeout bei langen Antworten zu umgehen.
• Caching spart Geld und reduziert die Latenz bei Wiederholungen.
• Function Calling entwickelt Ihren Bot vom Textgenerator zum vollwertigen Agenten.

Die Übernahme dieser Praktiken reduziert Produktionsausfälle und sorgt für Stabilität bei steigender Last. Sparen Sie nicht an der Architektur – sparen Sie stattdessen an Tokens.

— Editorial Team