Telegram-бот с GPT: от идеи до продакшена за неделю

TL;DR: Рабочий Telegram-бот с GPT — это примерно 300 строк кода на aiogram 3.x плюс клиент OpenAI, Redis для контекста и Docker-образ с systemd на старте. Неделя — реалистичный срок, если не закапываться в архитектурную эстетику. Основные проблемы в проде не в коде, а в трёх местах: лимиты Telegram и OpenAI, раздувающийся контекст диалогов и флуд от пользователей. Ниже — план по дням и код, который можно запустить как есть.

Что собираем: стек и архитектура

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

Python 3.12+ и aiogram 3.x — асинхронная, стабильная, активно развивающаяся библиотека для Bot API.
AsyncOpenAI из пакета openai>=1.x (или anthropic, если нужен Claude — API схожий по духу). Старый openai.ChatCompletion из версий 0.x больше не существует, код на нём даже не запустится.
Redis — для контекста диалога и антифлуда. Для MVP отдельная PostgreSQL нужна реже, чем кажется: история диалога умещается в Redis с TTL.
Docker — один образ, один контейнер, перезапуск через systemd.

Общая схема простая: Telegram шлёт обновления в бота (polling или webhook), хендлер aiogram собирает контекст из Redis, запрос уходит в LLM, ответ сохраняется и отправляется пользователю. Asyncio здесь не опция, а необходимость: пока GPT думает десять секунд, бот должен обслуживать остальных.

Если думаете о более сложных сценариях — агенты с инструментами, память, RAG по документам — лучше сразу посмотреть, как устроены AI-агенты на практике и как организовать память AI-систем. Для недельного плана это оверхед, для второй итерации — обязательное чтение.

Polling или webhook

Моя позиция: для MVP — long polling. Одна команда dp.start_polling(bot), никаких сертификатов, nginx и доменных имён. Работает за NAT, подходит для VPS за любой фаервол. Webhook нужен, когда у вас уже есть веб-сервер (например, FastAPI), жёсткие требования к задержке или необходимость объединить несколько ботов на одном хосте. Разница в latency между polling и webhook у aiogram минимальна — реальная задержка у вас будет в ответе LLM, а не в доставке апдейта.

День 1–2: MVP на aiogram 3.x

Первые два дня — каркас, который отвечает на сообщения. Никаких абстракций сверх необходимого.

Установка:

pip install aiogram openai redis

Минимальный бот:

import asyncio
import logging
import os

from aiogram import Bot, Dispatcher, F
from aiogram.filters import CommandStart
from aiogram.types import Message
from openai import AsyncOpenAI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

bot = Bot(token=os.environ["TELEGRAM_TOKEN"])
dp = Dispatcher()
openai_client = AsyncOpenAI()  # ключ берётся из OPENAI_API_KEY

SYSTEM_PROMPT = "Ты полезный помощник. Отвечай кратко и по делу."


@dp.message(CommandStart())
async def cmd_start(message: Message) -> None:
    await message.answer("Пришлите любой вопрос — отвечу через GPT.")


@dp.message(F.text)
async def handle_message(message: Message) -> None:
    await bot.send_chat_action(message.chat.id, "typing")
    try:
        response = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": message.text},
            ],
            max_tokens=1000,
        )
        await message.answer(response.choices[0].message.content)
    except Exception:
        logger.exception("LLM request failed")
        await message.answer("Ошибка, попробуйте ещё раз через минуту.")


async def main() -> None:
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Три детали, которые здесь важны. Во-первых, фильтры CommandStart() и F.text — это способ aiogram 3.x разводить апдейты по хендлерам; фабрики dp.message_handler из 2.x удалены. Во-вторых, send_chat_action вешает индикатор «печатает…» — без него пользователь при ответе в 5–15 секунд решит, что бот завис. В-третьих, ловим общий Exception и логируем через logger.exception — со стектрейсом, иначе в проде вы будете дебажить вслепую.

Выбор модели: для чат-сценариев gpt-4o-mini — мой дефолт по соотношению цена/качество; дорогие флагманы имеет смысл брать, только если обычный ответ заметно хуже по метрике, а не «вроде бы хуже». Актуальные модели и цены смотрите в официальном прайс-листе OpenAI — он меняется быстрее, чем я успеваю обновлять статьи.

Команды для бота удобнее регистрировать через BotFather (/setcommands), а не прописывать их интерпретацию вручную.

День 3: контекст диалога в Redis

Без истории бот общается как человек с антероградной амнезией: каждый вопрос обрабатывается в вакууме. Контекст храним в Redis как JSON-список сообщений с TTL.

import json

import redis.asyncio as redis

redis_client = redis.from_url(os.environ["REDIS_URL"])
CONTEXT_TTL = 3600
MAX_MESSAGES = 20  # пар «вопрос-ответ» хватит за глаза


async def get_context(user_id: int) -> list[dict]:
    raw = await redis_client.get(f"ctx:{user_id}")
    return json.loads(raw) if raw else []


async def append_to_context(user_id: int, role: str, content: str) -> list[dict]:
    context = await get_context(user_id)
    context.append({"role": role, "content": content})
    context = context[-MAX_MESSAGES:]
    await redis_client.set(
        f"ctx:{user_id}", json.dumps(context, ensure_ascii=False), ex=CONTEXT_TTL
    )
    return context

В хендлере перед запросом берём контекст, после ответа — дописываем оба сообщения:

context = await append_to_context(message.from_user.id, "user", message.text)
response = await openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "system", "content": SYSTEM_PROMPT}, *context],
    max_tokens=1000,
)
await append_to_context(message.from_user.id, "assistant", response.choices[0].message.content)

Типовая грабля: обрезать контекст по количеству сообщений — ненадёжно. Один длинный вопрос превысит лимит модели по токенам. Правильнее считать токены через tiktoken и выкидывать самые старые сообщения, пока суммарный объём не станет меньше порога:

import tiktoken

encoder = tiktoken.encoding_for_model("gpt-4o-mini")


def trim_to_tokens(context: list[dict], max_tokens: int = 6000) -> list[dict]:
    while context and sum(len(encoder.encode(m["content"])) for m in context) > max_tokens:
        context.pop(0)
    return context

Полная история диалогов «навсегда» — отдельная тема, там нужны PostgreSQL и продуманная схема; идеи по организации долгосрочной памяти разбирал в статье про память AI-систем. Для MVP TTL в час и двадцать сообщений покрывают подавляющее большинство реальных диалогов.

День 4: защита от флуда и очередь запросов

Пользователь с автокликером или просто нетерпеливый человек может сгенерировать десятки запросов в минуту. Каждый — деньги. Антифлуд в aiogram 3.x делается middleware:

from collections.abc import Awaitable, Callable
from typing import Any

from aiogram import BaseMiddleware
from aiogram.types import Message


class ThrottlingMiddleware(BaseMiddleware):
    def __init__(self, rate: float = 3.0):
        self.rate = rate

    async def __call__(
        self,
        handler: Callable[[Message, dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: dict[str, Any],
    ) -> Any:
        user_id = event.from_user.id
        key = f"throttle:{user_id}"
        if await redis_client.set(key, "1", px=int(self.rate * 1000), nx=True):
            return await handler(event, data)
        await event.answer("Подождите несколько секунд — ещё обрабатываю предыдущий запрос.")


dp.message.middleware(ThrottlingMiddleware(rate=3.0))

Логика через SET ... NX PX атомарна: ключ живёт rate миллисекунд, и пока он существует — запрос отклоняется. Пользователю это нужно объяснить, иначе он решит, что бот сломался, и начнёт слать сообщения ещё агрессивнее.

Второй момент — таймауты. Ответ LLM может занять 30+ секунд, а Telegram-вебхуки требуют быстрого ответа (поэтому с webhook сразу делайте ack и обрабатывайте запрос фоном). Для polling достаточно таймаута на стороне клиента:

from openai import AsyncOpenAI

openai_client = AsyncOpenAI(timeout=45.0, max_retries=2)

Встроенные ретраи max_retries покрывают временные сбои и RateLimitError с экспоненциальной паузой. Не пишите собственный цикл ретраев поверх — получите двойные запросы и двойные деньги.

Очередь из asyncio.Queue на этом этапе не нужна: aiogram сам обрабатывает апдейты конкурентно. Очередь и воркеры становятся оправданными, когда вы хотите жёстко ограничить параллелизм к OpenAI (например, держаться под RPM-лимитом) — тогда ставьте asyncio.Semaphore(5) вокруг вызова, это три строки вместо класса воркеров.

День 5: ошибки, логирование, длинные ответы

Что реально ломается в проде, в порядке частоты:

1. RateLimitError от OpenAI. На молодых аккаунтах лимиты низкие. Решение — ретраи (уже есть), резервная модель или честное сообщение пользователю. Ловим отдельно:

from openai import RateLimitError, APIError

try:
    response = await openai_client.chat.completions.create(...)
except RateLimitError:
    await message.answer("Сервис перегружен, попробуйте через минуту.")
except APIError:
    logger.exception("OpenAI API error")
    await message.answer("Временный сбой. Перешлите вопрос позже.")

2. Длинные ответы. Telegram ограничивает сообщение 4096 символами — при превышении send_message бросит TelegramBadRequest. Простое решение — резать ответ на части:

async def reply_long(message: Message, text: str) -> None:
    for i in range(0, len(text), 4000):
        await message.answer(text[i : i + 4000])

Резать лучше по абзацам, а не по ровно 4000 символов — иначе регулярно разъедаете markdown-разметку посередине и получаете can't parse entities. Парсинг разметки в LLM-ответах — вообще отдельная боль: надёжнее отправлять как plain text, чем лечить битый MarkdownV2.

3. Файлы и голосовые. Как только бот «умеет отвечать», пользователи начинают слать документы и аудио. Заранее решите, что с этим делать: хотя бы вежливо отклоняйте (@dp.message(F.voice) → «Пока понимаю только текст»).

4. Зависший polling. При рвущемся соединении start_polling может умереть молча. Поэтому — автоперезапуск через systemd (следующий раздел), а не nohup в tmux.

Логирование: минимум — logging в файл с ротацией; в идеале — JSON-логи и сбор через любой агент. Что логировать обязательно: user_id, длину запроса/ответа, response.usage.total_tokens (для контроля расходов), время ответа, exception со стектрейсом. Что логировать не надо: полные тексты переписки — это и этически сомнительно, и нарушение требований к хранению персональных данных при росте аудитории.

День 6: контроль расходов

Токены — главная переменная статья бюджета. Три рабочих рычага, без магии:

Обрезка контекста из дня 3 — уже сделано. Системный промпт тоже считается в каждом запросе, поэтому простыня из тысячи токенов инструкций — это постоянный налог; держите её короткой.
Лимиты на пользователя в день через Redis-счётчик. Например, 50 сообщений в сутки — типичному пользователю хватит, а ботоводы упрётся:

DAILY_LIMIT = 50


async def check_daily_limit(user_id: int) -> bool:
    key = f"daily:{user_id}"
    count = await redis_client.incr(key)
    if count == 1:
        await redis_client.expire(key, 86400)
    return count <= DAILY_LIMIT

Мониторинг потребления. Суммируйте usage.total_tokens из каждого ответа по дням и выводите в простой дашборд или хотя бы в лог — «внезапная» тысяча долларов на счётчике всегда следствие отсутствия этой статистики, а не хитрой атаки.

Альтернативные провайдеры: Anthropic SDK (AsyncAnthropic, метод client.messages.create) совместим по структуре сценария «системный промпт + история», миграция — дело пары часов. Локальные модели через Ollama имеют смысл при строжайшей конфиденциальности или очень больших объёмах, но на MVP-неделе качество и время внедрения проигрывают облаку.

День 7: деплой в Docker + systemd

Dockerfile:

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY bot.py .

CMD ["python", "bot.py"]

requirements.txt:

aiogram>=3.10
openai>=1.50
redis>=5.0
tiktoken

Сборка и unit-файл systemd — перезапуск при падении, автостарт после ребута:

# /etc/systemd/system/gptbot.service
[Unit]
Description=GPT Telegram bot
After=docker.service

[Service]
Restart=always
RestartSec=5
ExecStart=/usr/bin/docker run --rm --name gptbot \
  --env-file /opt/gptbot/.env \
  -e REDIS_URL=redis://host.docker.internal:6379 \
  gptbot:latest
ExecStop=/usr/bin/docker stop gptbot

[Install]
WantedBy=multi-user.target

.env с токенами (права 600, владелец root):

TELEGRAM_TOKEN=123:abc
OPENAI_API_KEY=sk-...

systemctl enable --now gptbot
journalctl -u gptbot -f   # живые логи

Redis на том же VPS ставьте отдельным контейнером или из пакетов ОС — на этом масштабе разницы нет. Когда понадобятся миграции схемы, несколько инстансов и балансировка — значит, продукт выстрелил, и пора смотреть в сторону полноценного бэкенда; выбор между FastAPI и Django для такого сценария разбирал в отдельной статье.

Структура плана по дням (шпаргалка)

День 1–2: aiogram-каркас, интеграция OpenAI, минимальная обработка ошибок. Критерий готовности: бот отвечает на вопросы с VPS, а не с ноутбука.
День 3: контекст в Redis, обрезка по токенам через tiktoken.
День 4: throttling-middleware, таймауты и ретраи, семафор на параллелизм.
День 5: типизация ошибок OpenAI, разбивка длинных ответов, обработка голосовых, JSON-логи.
День 6: дневные лимиты, учёт usage, тест «что будет при тысяче запросов» на дружественной нагрузке.
День 7: Docker, systemd, env-файлы, чек-лист аварий «бот молчит / бот дублит ответы / OpenAI вернул 429».

Пропустить можно день 4–5 (если аудитория — три друга), нельзя — день 6 и 7. Бот без контроля расходов и автоперезапуска — это кредит: рано или поздно он предъявит счёт.

FAQ

Обязательно ли брать OpenAI? Нет. Код из статьи адаптируется под Anthropic за пару десятков строк (свой AsyncClient и формат ответа); локальные модели через Ollama — полноценный вариант при наличии GPU, но качество диалога на малых моделях заметно ниже.

Polling или webhook для продакшена? Polling до нескольких тысяч пользователей. Webhook — когда появляется веб-часть, требования к задержке или несколько ботов. Мигрировать позже — пара часов работы.

Как хранить персональные данные пользователей? Минимум: TTL на контекст, никаких полных переписок в логах, шифрование на уровне диска VPS. Если бот растёт — добавляется PostgreSQL с политикой ретенции и, в зависимости от юрисдикции, требования к согласию на обработку.

А если пользователь пришлёт prompt injection — «игнорируй инструкции»? В сценарии «бот-помощник» это некритично: злоумышленник испортит только свой диалог. Критично становится, когда у бота есть доступ к инструментам и данным — тогда выходите на территорию AI-агентов с валидацией действий, что уже отдельный разговор.

Сколько это стоит в месяц? VPS начального уровня плюс расходы на токены; итог зависит целиком от вашей аудитории и модели, поэтому честную оценку считайте только после недели замеров usage. Методику оценки бюджетов на AI-продукты я разбирал в статье «Сколько стоит MVP», а про критерии готовности MVP вообще — в «MVP в 2026».

Нужна помощь с Telegram-ботом или интеграцией LLM в продукт? Напишите мне — обсудим задачу и подберём архитектуру под ваши сроки.

Что собираем: стек и архитектура

Polling или webhook

День 1–2: MVP на aiogram 3.x

День 3: контекст диалога в Redis

День 4: защита от флуда и очередь запросов

День 5: ошибки, логирование, длинные ответы

День 6: контроль расходов

День 7: деплой в Docker + systemd

Структура плана по дням (шпаргалка)

FAQ

Похожие статьи

LangChain vs LlamaIndex: Выбираем инструмент для работы с LLM

FastAPI vs Django: Выбираем фреймворк для AI-продуктов

Как построить RAG-пайплайн для корпоративных документов: от идеи до продакшна

Обсудить проект