Integrace Gemini API do Telegram-bota: průvodce pro vývojáře

# Asynchronní integrace Gemini API do Telegram bota: praktický návod pro aiogram 3.x

Když váš Telegram bot náhle přestane odpovídat kvůli chybě 429 od Gemini API, není to jen chyba – je to architektuální dluh. V tomto článku krok za krokem rozebereme, jak správně propojit Gemini s asynchronním botem na aiogram 3.x, aby se zabránilo blokování event loopu, obejít limity RPM/TPM a implementovat funkční volání bez ztráty výkonu.

Architektura: od jednoduchého handleru do produkčního řešení

Základní schéma interakce uživatele s botem přes Gemini vypadá triviálně: požadavek → zpracování → odpověď. Realita však vyžaduje další vrstvy. Asynchronní volání synchronního API bez řádné obálky vede k zacyklení event loopu aiogram, zvláště při vysoké zátěži. Proto klíčové komponenty:

• Asyncio Queue — pro buforizaci příchozích požadavků
• Rate Limiter — pro dodržování limitů Gemini (5–60 RPM)
• Response Cache — pro úsporu tokenů u opakovaných požadavků
• Streaming Handler — pro obejití timeoutů Telegramu (10 sekund)

Gemini API ukládá přísné limity i na placených tarifech: maximální zpoždění odpovědi může dosáhnout 15 sekund, zatímco bezplatný limit je jen 5 požadavků za minutu. Bez front a cachování bude uživatelský zážitek zničen.

Nastavení klienta: synchronní SDK v asynchronním prostředí

Google GenAI SDK zůstává synchronní, navzdory všem očekáváním komunity. To znamená, že přímé volání generate_content() uvnitř handleru aiogram zablokuje celý event loop. Řešení – obalit volání do asyncio.to_thread nebo použít 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

Toto je minimální funkční obálka. Pro produkci přidejte zpracování výjimek a metriky výkonu.

Integrace s aiogram: od /start po streamingové odpovědi

Základní handler musí nejen přijímat zprávy, ale i řídit stav: ukazovat „psaní“, rozdělovat dlouhé odpovědi a cachovat výsledky. Telegram omezuje délku zprávy na 4096 znaků – ignorování toho vede k přerušením.

@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("Něco se pokazilo. Zkuste to později.")

Pro dlouhé požadavky použijte streaming – aktualizujte jednu zprávu postupně s generovaným textem. To zabraňuje timeoutům a zlepšuje UX.

Funkční volání: když se Gemini stává agentem

Function Calling umožňuje modelu volat externí API podle kontextu požadavku. Například pokud uživatel napíše „zarezervuj schůzku na zítřek v 15:00“, Gemini může parsovat parametry a volat vaši funkci schedule_meeting.

Hlavní kroky:

• Popište funkci ve formátu JSON Schema – model musí chápat její signaturu.
• Předejte popis do GenerateContentConfig jako tools.
• Zpracujte function_call v odpovědi a proveďte skutečné volání.
• Vraťte výsledek zpět do chatu jako součást dialogu.

Příklad deklarace funkce:

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Vytvoří schůzku s uvedenými účastníky",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

Toto mění statického bota v dynamického asistenta schopného interagovat s externími systémy.

Cachování a rate limiting: ochrana před bankrotem a zákazy

Gemini se účtuje podle tokenů. Bez cachování budete platit za každý opakovaný dotaz. Jednoduchý in-memory cache na bázi MD5 hashu promptu s TTL problém řeší:

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # kontrola TTL a vrácení hodnoty
    
    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 je nutný i na placených tarifech. Implementujte ho jako asynchronní kontextový manažer nebo 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)

Typické chyby a jak jim zabránit

Tři hlavní pasti, na které šlápou všichni:

• Chyba 429 (RESOURCE_EXHAUSTED) — vzniká při překročení RPM, TPM nebo RPD. Řešení: lokální rate limiter + exponenciální backoff.
• Timeouty Telegramu — server čeká na odpověď 10 sekund. Pokud je Gemini pomalý, použijte streaming nebo pošlete mezitím zprávy.
• Zastaralé modely — Google pravidelně ruší podporu starých verzí. V březnu 2026 se Pro modely staly dostupné jen na předplatné. Vždy ověřte aktuálnost modelu před nasazením.

Co je důležité:

• Synchronní SDK Gemini vyžaduje obalení do run_in_executor – jinak zablokujete event loop.
• Rate limiting musí být implementován na straně bota, ne spoléhat se na API.
• Streaming je jediný způsob, jak obejít 10sekundový timeout Telegramu pro dlouhé odpovědi.
• Cachování šetří peníze a snižuje latenci u opakovaných požadavků.
• Function Calling mění bota z textového generátoru na plnohodnotného agenta.

Implementace těchto praktik sníží počet incidentů v produkci a zajistí stabilní chod i při růstu zátěže. Nešetřete na architektuře – šetřete na tokenech.

— Editorial Team