Интеграция Google Gemini API в асинхронный Telegram-бот на aiogram 3.x и Python

В прошлую пятницу, ровно в 18:47, когда я уже мысленно открывал великолепный, наполненный витаминами, напиток, мне прилетело сообщение от тимлида: «Бот лежит, пользователи жалуются, Gemini API возвращает 429». Наш корпоративный Telegram-бот, который должен был помогать саппорту отвечать на тикеты, просто встал колом. Причина оказалась до банальности простой: мы не учли rate limiting и думали, что 50 RPM (запросов в минуту) на бесплатном тарифе — это «бесконечно много». С тех пор мы переписали архитектуру, добавили очереди, кэширование и middleware для retry. В этой статье разберу, как с нуля подружить Gemini API с Telegram-ботом на aiogram 3.x, не наступая на те же грабли.

Архитектура: что и зачем

Классическая схема выглядит так:
[Telegram User] → [aiogram Bot] → [Gemini API]
↓
[Response Queue / Cache]

Но в реальном продакшене появляется дополнительная обвязка:

[Telegram] → [aiogram] → [asyncio Queue] → [Rate Limiter] → [Gemini API]
↑ ↓
[Response Cache] ←───────────────── [Streaming Handler]

Почему это важно? Gemini API имеет жёсткие лимиты:
1.Бесплатный тариф: 5–15 RPM в зависимости от модели
2.Ответы могут идти до 10–15 секунд на длинных промптах.
3.Платный: до 60 RPM для Gemini 2.0 Flash
Если просто вызывать await client.models.generate_content() внутри хендлера — вы положите ивент-луп aiogram и получите таймауты от Telegram. Асинхронность aiogram здесь не спасает — блокирующий вызов остаётся блокирующим.

Шаг 1. Установка и настройка Gemini API

Ставим библиотеку Google GenAI SDK (она же google-genai):

pip install google-genai aiogram python-dotenv

Скрытый текст

Важно: библиотека google-generativeai устарела и с мая 2025 года не поддерживается. Используйте именно google-genai

Создаём .env файл:

GEMINI_API_KEY=your-api-key-hereTELEGRAM_BOT_TOKEN=your-bot-token

Получить API-ключ можно в Google AI Studio → API Keys → Create API Key.

Базовый клиент:

# gemini_client.pyimport osfrom google import genaifrom google.genai import typesclass GeminiClient:    def __init__(self, model: str = "gemini-3-flash-preview"):        self.client = genai.Client()  # ключ берётся из GEMINI_API_KEY        self.model = model        async def generate(self, prompt: str) -> str:        # Обратите внимание: это синхронный вызов!        # await здесь не поможет, нужен asyncio.to_thread        response = self.client.models.generate_content(            model=self.model,            contents=prompt        )        return response.text

Скрытый текст

Важное замечание: клиент google-genai синхронный! В асинхронном aiogram его вызовы будут блокировать ивент-луп.

Шаг 2. Асинхронная обёртка через asyncio.to_thread

Чтобы не вешать весь бот на каждый запрос к Gemini, используем asyncio.to_thread:

# async_gemini.pyimport asynciofrom google import genaiclass 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,  # используем дефолтный ThreadPoolExecutor            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

Скрытый текст

Это минимально жизнеспособный вариант. Для продакшена понадобится ещё очередь запросов.

Шаг 3. Интеграция с aiogram

Базовый хендлер для aiogram 3.x:

# bot.pyimport asyncioimport loggingfrom aiogram import Bot, Dispatcher, Router, typesfrom aiogram.filters import Commandfrom aiogram.enums import ParseModefrom dotenv import load_dotenvfrom async_gemini import AsyncGeminiClientload_dotenv()logging.basicConfig(level=logging.INFO)router = Router()gemini = AsyncGeminiClient()@router.message(Command("start"))async def cmd_start(message: types.Message):    await message.answer(        "Привет! Я бот с Gemini API. Просто напиши мне вопрос, и я отвечу."    )@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)        # Telegram имеет лимит 4096 символов на сообщение        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(            "Что-то пошло не так. Попробуйте позже или сформулируйте запрос иначе."        )async def main():    bot = Bot(token=os.getenv("TELEGRAM_BOT_TOKEN"))    dp = Dispatcher()    dp.include_router(router)        await dp.start_polling(bot)if __name__ == "__main__":    asyncio.run(main())

Шаг 4. Функциональный вызов (Function Calling) для расширения возможностей

Gemini умеет не только генерировать текст, но и вызывать внешние функции. Это полезно, если бот должен работать с реальными данными: бронировать встречи, проверять статус заказа, искать информацию в базе. Пример для планирования встреч:

# function_calling.pyfrom google import genaifrom google.genai import types# Описываем функцию, которую Gemini может вызватьschedule_meeting_function = {    "name": "schedule_meeting",    "description": "Создаёт встречу с указанными участниками",    "parameters": {        "type": "object",        "properties": {            "attendees": {                "type": "array",                "items": {"type": "string"},                "description": "Список email участников",            },            "date": {                "type": "string",                "description": "Дата встречи (ГГГГ-ММ-ДД)",            },            "time": {                "type": "string",                "description": "Время встречи (ЧЧ:ММ)",            },            "topic": {                "type": "string",                "description": "Тема встречи",            },        },        "required": ["attendees", "date", "time", "topic"],    },}# Реальная функция, которую мы будем вызыватьdef schedule_meeting(attendees: list, date: str, time: str, topic: str):    # Здесь может быть вызов Google Calendar API, БД и т.д.    return f"Встреча '{topic}' запланирована на {date} в {time} с {', '.join(attendees)}"# Интеграция с Geminiclient = genai.Client()tools = types.Tool(function_declarations=[schedule_meeting_function])config = types.GenerateContentConfig(tools=[tools])response = client.models.generate_content(    model="gemini-3-flash-preview",    contents="Запланируй встречу с bob@company.com и alice@company.com на 15.04.2026 в 14:00 по поводу запуска продукта",    config=config,)if response.candidates[0].content.parts[0].function_call:    fc = response.candidates[0].content.parts[0].function_call    print(f"Gemini хочет вызвать: {fc.name}")    print(f"С аргументами: {fc.args}")    # Вызываем нашу функцию с аргументами от Gemini    result = schedule_meeting(**fc.args)    print(result)

Это мощный паттерн, который превращает простого чат-бота в настоящего агента.

Шаг 5. Кэширование ответов (чтобы не платить дважды)

Gemini API тарифицируется по токенам. Если пользователи часто задают одни и те же вопросы (например, «как сбросить пароль»), вы будете платить за каждый запрос. Решение — простой in-memory кэш:

# cache.pyimport hashlibfrom datetime import datetime, timedeltafrom typing import Optionalclass SimpleCache:    def __init__(self, ttl_seconds: int = 3600):        self._cache = {}        self._ttl = ttl_seconds        def get(self, key: str) -> Optional[str]:        if key in self._cache:            value, timestamp = self._cache[key]            if datetime.now() - timestamp < timedelta(seconds=self._ttl):                return value            else:                del self._cache[key]        return None        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()

В хендлере добавляем:

cache = SimpleCache(ttl_seconds=7200)  # 2 часа@router.message()async def handle_message(message: types.Message):    prompt_hash = cache.hash_prompt(message.text)    cached = cache.get(prompt_hash)        if cached:        await message.answer(cached)        return        response = await gemini.generate(message.text)    cache.set(prompt_hash, response)    await message.answer(response)

Грабли (то, о чём не пишут в документации)

Грабли №1: 429 ошибка в пятницу вечером

Самая частая проблема — RESOURCE_EXHAUSTED (429). Причины:

1. RPM-лимит. Бесплатный тариф даёт 5 RPM, платный — до 60 RPM

2. TPM-лимит. Ограничение на количество токенов в минуту (1M для бесплатного тарифа).

3. RPD-лимит. Ограничение на количество запросов в день (25–1500 в зависимости от модели).

Решение: используйте rate limiter на стороне бота:

# rate_limiter.pyimport asyncioimport timeclass 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)

Грабли №2: Таймауты от Telegram

Telegram ждёт ответ от бота 10 секунд. Если Gemini думает дольше, вы получите таймаут и пользователь увидит ошибку. Решение — показывать промежуточные сообщения или использовать streaming:

# streaming_example.pyasync def generate_stream(prompt: str):    for chunk in client.models.generate_content_stream(        model="gemini-3-flash-preview",        contents=prompt    ):        yield chunk.text

В aiogram можно обновлять одно сообщение:

sent = await message.answer("Думаю...")full_response = ""async for chunk in gemini.generate_stream(message.text):    full_response += chunk    if len(full_response) % 100 == 0:  # обновляем каждые 100 символов        try:            await sent.edit_text(full_response)        except:            passawait sent.edit_text(full_response)

Грабли №3: Модели устаревают быстрее, чем вы читаете документацию

Gemini обновляется каждые несколько месяцев. На момент написания статьи актуальны:
gemini-3-flash-preview — быстрая и дешёвая модель
gemini-3.1-pro-preview — мощная, но дорогая ($2.00 за 1M входных токенов, $12.00 за 1M выходных)
С марта 2026 года Pro-модели недоступны на бесплатном тарифе — только платная подписка

Заключение

Интеграция Gemini API в Telegram-бота — задача на пару часов, если знать все подводные камни. Ключевые выводы:

1. Используйте asyncio.to_thread или очереди, чтобы не блокировать ивент-луп aiogram.

2. Внедряйте rate limiter и retry-логику до того, как получите 429 в продакшене.

3. Кэшируйте частые запросы — экономия на токенах может быть существенной.

4. Function Calling — ваш друг, если бот должен взаимодействовать с реальными сервисами.

Что бы вы добавили? Сталкивались ли вы с проблемами при интеграции LLM в ботов? Может, у кого-то есть опыт использования Gemini API в высоконагруженных проектах? Давайте обсудим в комментариях — особенно интересно услышать про ваши кейсы с 429 и таймаутами.