Integracja Gemini API w bot Telegram: przewodnik dla deweloperów

# Asynchroniczna integracja API Gemini z botem Telegram: praktyczny przewodnik po aiogram 3.x

Kiedy Twój bot Telegram nagle przestaje odpowiadać z powodu błędu 429 od API Gemini, to nie jest po prostu błąd — to dług architektoniczny. W tym artykule krok po kroku omawiamy, jak poprawnie podłączyć Gemini do asynchronicznego bota na aiogram 3.x, aby uniknąć blokady event loopa, obejść limity RPM/TPM i zaimplementować wywołania funkcyjne bez utraty wydajności.

Architektura: od prostego handlera do rozwiązania produkcyjnego

Podstawowy schemat interakcji użytkownika z botem przez Gemini wygląda banalnie: zapytanie → przetwarzanie → odpowiedź. Ale rzeczywistość wymaga dodatkowych warstw. Asynchroniczne wywołanie synchronicznego API bez odpowiedniej otoczki doprowadzi do zawieszenia event loopa aiogram, szczególnie przy wysokiej obciążeniu. Dlatego kluczowe komponenty:

• Asyncio Queue — do buforowania przychodzących zapytań
• Rate Limiter — do przestrzegania ograniczeń Gemini (5–60 RPM)
• Response Cache — do oszczędzania tokenów na powtarzających się zapytaniach
• Streaming Handler — do obejścia limitów czasu Telegram (10 sekund)

API Gemini nakłada surowe ograniczenia nawet na płatnych planach: maksymalne opóźnienie odpowiedzi może sięgać 15 sekund, a darmowy limit to zaledwie 5 zapytań na minutę. Bez kolejek i cachowania doświadczenie użytkownika zostanie zniszczone.

Konfiguracja klienta: synchroniczny SDK w środowisku asynchronicznym

SDK Google GenAI pozostaje synchroniczny, mimo wszystkich oczekiwań społeczności. Oznacza to, że bezpośrednie wywołanie generate_content() wewnątrz handlera aiogram zablokuje cały cykl zdarzeń. Rozwiązanie — owinąć wywołania w asyncio.to_thread lub użyć 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

To minimalna działająca otoczka. Do produkcji dodaj obsługę wyjątków i metryki wykonania.

Integracja z aiogram: od /start do odpowiedzi streamingowych

Podstawowy handler musi nie tylko przyjmować wiadomości, ale i zarządzać stanem: pokazywać „pisze”, dzielić długie odpowiedzi i cachować wyniki. Telegram ogranicza długość wiadomości do 4096 znaków — ignorowanie tego doprowadzi do przerw.

@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("Coś poszło nie tak. Spróbuj później.")

Dla długich zapytań użyj streamingu — aktualizuj jedną wiadomość w miarę generowania tekstu. To zapobiega timeoutom i poprawia UX.

Wywołania funkcyjne: kiedy Gemini staje się agentem

Function Calling pozwala modelowi wywoływać zewnętrzne API w kontekście zapytania. Na przykład, jeśli użytkownik napisze „zarezerwuj spotkanie na jutro o 15:00", Gemini może sparsować parametry i wywołać twoją funkcję schedule_meeting.

Główne kroki:

• Opisz funkcję w formacie JSON Schema — model musi rozumieć jej sygnaturę.
• Przekaż opis do GenerateContentConfig jako tools.
• Obsłuż function_call w odpowiedzi i wykonaj rzeczywiste wywołanie.
• Zwróć wynik z powrotem do czatu jako część dialogu.

Przykład deklaracji funkcji:

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Tworzy spotkanie z podanymi uczestnikami",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

To przekształca statycznego bota w dynamicznego asystenta zdolnego do interakcji z zewnętrznymi systemami.

Cachowanie i rate limiting: ochrona przed bankructwem i banami

Gemini jest rozliczane według tokenów. Bez cache'u będziesz płacić za każde powtarzające się pytanie. Prosty cache w pamięci oparty na hashu MD5 promptu z TTL rozwiązuje problem:

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # sprawdzenie TTL i zwrot wartości
    
    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()

Rate limiter jest obowiązkowy nawet na płatnych planach. Zaimplementuj go jako asynchroniczny menedżer kontekstu lub 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)

Typowe błędy i jak ich uniknąć

Trzy główne pułapki, na które wpadają wszyscy:

• Błąd 429 (RESOURCE_EXHAUSTED) — pojawia się przy przekroczeniu RPM, TPM lub RPD. Rozwiązanie: lokalny rate limiter + wykładniczy backoff.
• Timeouty Telegram — serwer czeka na odpowiedź 10 sekund. Jeśli Gemini jest wolny, użyj streamingu lub wysyłaj wiadomości pośrednie.
• Nieaktualne modele — Google regularnie wycofuje stare wersje. W marcu 2026 modele Pro stały się dostępne tylko na subskrypcji. Zawsze sprawdzaj aktualność modelu przed wdrożeniem.

Co ważne:

• Synchroniczny SDK Gemini wymaga owinięcia w run_in_executor — inaczej zablokujesz event loop.
• Rate limiting musi być zaimplementowany po stronie bota, a nie polegać na API.
• Streaming to jedyny sposób na obejście 10-sekundowego timeoutu Telegram dla długich odpowiedzi.
• Cachowanie oszczędza pieniądze i zmniejsza latencję dla powtarzających się zapytań.
• Function Calling przekształca bota z generatora tekstu w pełnoprawnego agenta.

Wdrożenie tych praktyk zmniejszy liczbę incydentów w produkcji i zapewni stabilną pracę nawet przy wzroście obciążenia. Nie oszczędzaj na architekturze — oszczędzaj na tokenach.

— Editorial Team