Telegram-бот с GPT на Python (aiogram) через AI Gateway
Развёрнутый пример Telegram-бота с AI на Python: aiogram и OpenAI SDK с base_url AI Gateway, память диалога, системный промпт и защита от перерасхода через лимиты ключа.
Соберём Telegram-бота на aiogram, который помнит контекст диалога и держит расход под контролем. Это развёрнутая версия базового рецепта со страницы инструмента Telegram-бот: добавим память по каждому пользователю, системный промпт и лимиты. Модель зовём через OpenAI SDK, направленный на AI Gateway.
Подготовка
Нужны два секрета: токен бота от @BotFather (/newbot) и ключ AI Gateway sk-aigate-… из кабинета. Для бота заведите отдельный ключ со своим потолком расхода и RPM-лимитом. Базовый URL — https://ai-gatewey.ru/v1. Оба секрета читаем из окружения.
pip install aiogram openaiРеализация
- Клиент и хранилище диалогов
Настраиваем
AsyncOpenAIна AI Gateway и заводим на каждого пользователя короткую историю сообщений (dequeс ограничением длины — так контекст не растёт бесконечно):pythonpythonimport osfrom collections import defaultdict, dequefrom openai import AsyncOpenAIai = AsyncOpenAI(base_url="https://ai-gatewey.ru/v1",api_key=os.environ["AIGATE_API_KEY"],)SYSTEM = {"role": "system", "content": "Ты дружелюбный ассистент. Отвечай кратко и по делу."}# по 10 последних реплик на пользователяhistory: dict[int, deque] = defaultdict(lambda: deque(maxlen=10)) - Обработчик с памятью диалога
Перед запросом добавляем реплику пользователя в его историю и передаём модели системный промпт плюс контекст; ответ тоже кладём в историю:
pythonpythonfrom aiogram.types import Messageasync def on_message(msg: Message):hist = history[msg.from_user.id]hist.append({"role": "user", "content": msg.text})resp = await ai.chat.completions.create(model="deepseek/deepseek-v4-flash",messages=[SYSTEM, *hist],)answer = resp.choices[0].message.contenthist.append({"role": "assistant", "content": answer})await msg.answer(answer) - Команда сброса контекста
Дадим пользователю способ начать диалог заново — команду
/reset, которая чистит его историю:pythonpythonfrom aiogram.filters import Commandasync def on_reset(msg: Message):history.pop(msg.from_user.id, None)await msg.answer("Контекст очищен.") - Сборка и запуск
Регистрируем хендлеры и запускаем polling. Расход по токенам виден в кабинете AI Gateway в реальном времени:
pythonpythonimport asynciofrom aiogram import Bot, Dispatcher, Fbot = Bot(os.environ["TELEGRAM_TOKEN"])dp = Dispatcher()dp.message.register(on_reset, Command("reset"))dp.message.register(on_message, F.text) # без фильтра сюда придут фото и стикеры, а msg.text будет Noneasyncio.run(dp.start_polling(bot))
Какие модели подойдут
В ботах много коротких запросов, поэтому важнее скорость и цена за обращение, чем максимальный интеллект:
Частые вопросы
Почему история хранится в deque, а не в базе?
deque(maxlen=…) в памяти — простейший вариант для старта: он сам вытесняет старые реплики. Для продакшена контекст лучше держать в Redis или БД, иначе после перезапуска бота диалоги обнулятся.
Как ограничить число запросов на пользователя?
Добавьте middleware aiogram с проверкой частоты (например, не чаще N сообщений в минуту на from_user.id). Это защищает и от спама, и от неожиданного расхода на публичном боте.
Можно ли включить стриминг ответа?
Да, но в Telegram нет «печатающегося» текста: обычно копят чанки и раз в несколько токенов редактируют сообщение через edit_text. Для большинства ботов проще отдавать готовый ответ целиком.