Telegram бот с AI на Python: пошаговое руководство с потоковой генерацией (2026)

Кратко

• Создаём полноценного AI-бота для Telegram на Python — от первого сообщения до продакшена
• Используем Bot API 9.5 со встроенным стримингом ответов (релиз 1 марта 2026)
• Подключаем GPT, Claude, Gemini и 50+ моделей через единый OpenAI-совместимый API
• Полный рабочий код: обработка сообщений, потоковая генерация, история диалога, обработка ошибок
• Два режима работы: polling для разработки, webhook для деплоя

Содержание

• Что нового: Bot API 9.5 и нативный streaming
• Архитектура бота
• Подготовка: создаём бота и получаем ключи
• Базовый бот: первый AI-ответ за 50 строк
• Потоковая генерация: streaming в реальном времени
• Продвинутые возможности
• Мультимодельный роутинг
• Обработка ошибок и fallback
• Деплой: от polling к webhook
• Часто задаваемые вопросы (FAQ)
• Итоги

Что нового: Bot API 9.5 и нативный streaming

1 марта 2026 года Telegram выпустил Bot API 9.5, и главное нововведение — метод sendMessageDraft стал доступен всем ботам. Это значит, что теперь любой бот может отправлять ответы потоково, как ChatGPT — текст появляется посимвольно в реальном времени.

До 9.5 реализовать streaming можно было только через постоянное редактирование сообщения (editMessageText), что давало рваный UX и нагружало API rate limit. Теперь streaming — нативная функция платформы.

Что ещё появилось в 9.5:

• Форматирование даты и времени — бот может отправлять локализованные даты, которые автоматически отображаются в часовом поясе пользователя
• Кастомные теги участников — управление тегами в группах через API
• Стилизация кнопок (из 9.4) — кастомные цвета и эмодзи на инлайн-кнопках

Для AI-ботов streaming — критически важная функция. Пользователи привыкли видеть генерацию текста в реальном времени, и бот, который молчит 10 секунд, а потом выдаёт стену текста, воспринимается как сломанный.

Архитектура бота

Наш бот работает по простой, но надёжной схеме:

Пользователь → Telegram → Bot API → Python-обработчик
                                          ↓
                                    AI API (OpenAI-совместимый)
                                          ↓
                                    Streaming ответ → Telegram

Стек:

Подготовка: создаём бота и получаем ключи

Шаг 1. Создание бота в Telegram

1. Откройте @BotFather в Telegram
2. Отправьте /newbot
3. Введите отображаемое имя (например, My AI Assistant)
4. Введите username (должен заканчиваться на bot, например my_ai_assistant_bot)
5. Скопируйте токен — строку вида 7123456789:AAH...

Шаг 2. Получение AI API ключа

Для подключения AI моделей нужен API-ключ, совместимый с OpenAI протоколом. Варианты:

Если вам нужен доступ к нескольким моделям без отдельных аккаунтов — агрегатор с OpenAI-совместимым интерфейсом упрощает жизнь: один base_url, один ключ, все модели.

Шаг 3. Установка зависимостей

pip install python-telegram-bot openai

Шаг 4. Переменные окружения

Создайте файл .env:

TELEGRAM_BOT_TOKEN=7123456789:AAHxxxxx
OPENAI_API_KEY=sk-xxxxx
OPENAI_BASE_URL=https://api.openai.com/v1  # или URL агрегатора
AI_MODEL=gpt-4.1-mini

Базовый бот: первый AI-ответ за 50 строк

Начнём с минимального рабочего бота — без streaming, без истории, чистая механика:

import os
from telegram import Update
from telegram.ext import (
    ApplicationBuilder, CommandHandler,
    MessageHandler, filters, ContextTypes
)
from openai import AsyncOpenAI

# Инициализация AI клиента
ai = AsyncOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)
MODEL = os.getenv("AI_MODEL", "gpt-4.1-mini")

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "Привет! Я AI-бот. Напиши мне что-нибудь, и я отвечу."
    )

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_text = update.message.text

    # Показываем индикатор «печатает...»
    await update.message.chat.send_action("typing")

    response = await ai.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "Ты полезный AI-ассистент. Отвечай кратко и по делу."},
            {"role": "user", "content": user_text},
        ],
    )

    reply = response.choices[0].message.content
    await update.message.reply_text(reply)

def main():
    app = ApplicationBuilder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    app.run_polling()

if __name__ == "__main__":
    main()

Запуск:

python bot.py

Бот заработал. Но у него нет ни памяти, ни стриминга. Исправим это.

Потоковая генерация: streaming в реальном времени

Bot API 9.5 добавил нативный streaming через sendMessageDraft, но python-telegram-bot пока не обернул этот метод в удобный API. Поэтому используем проверенный подход — streaming через OpenAI SDK с периодическим редактированием сообщения. Это работает надёжно и выглядит как «живая печать»:

import os
import asyncio
from telegram import Update
from telegram.ext import (
    ApplicationBuilder, CommandHandler,
    MessageHandler, filters, ContextTypes
)
from openai import AsyncOpenAI

ai = AsyncOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)
MODEL = os.getenv("AI_MODEL", "gpt-4.1-mini")

# История диалогов: {chat_id: [messages]}
conversation_history: dict[int, list[dict]] = {}
MAX_HISTORY = 20  # Максимум пар сообщений в памяти

SYSTEM_PROMPT = """Ты полезный AI-ассистент в Telegram.
Отвечай кратко и структурированно. Используй Markdown для форматирования."""


def get_history(chat_id: int) -> list[dict]:
    if chat_id not in conversation_history:
        conversation_history[chat_id] = []
    return conversation_history[chat_id]


def trim_history(chat_id: int):
    history = conversation_history.get(chat_id, [])
    if len(history) > MAX_HISTORY * 2:
        conversation_history[chat_id] = history[-(MAX_HISTORY * 2):]


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    conversation_history[chat_id] = []
    await update.message.reply_text(
        "Привет! Я AI-бот. Задай любой вопрос.\n\n"
        "Команды:\n"
        "/clear — очистить историю диалога\n"
        "/model — показать текущую модель"
    )


async def clear(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    conversation_history[chat_id] = []
    await update.message.reply_text("История очищена.")


async def model_info(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(f"Текущая модель: `{MODEL}`", parse_mode="Markdown")


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    user_text = update.message.text

    # Добавляем сообщение пользователя в историю
    history = get_history(chat_id)
    history.append({"role": "user", "content": user_text})

    messages = [{"role": "system", "content": SYSTEM_PROMPT}] + history

    # Отправляем placeholder-сообщение
    bot_msg = await update.message.reply_text("⏳")

    try:
        # Запускаем streaming
        stream = await ai.chat.completions.create(
            model=MODEL,
            messages=messages,
            stream=True,
        )

        full_text = ""
        last_update = ""
        chunk_count = 0

        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                full_text += delta
                chunk_count += 1

                # Обновляем сообщение каждые 15 чанков или при завершении
                if chunk_count % 15 == 0 and full_text != last_update:
                    try:
                        await bot_msg.edit_text(full_text + " ▌")
                        last_update = full_text
                    except Exception:
                        pass  # Rate limit — пропускаем обновление

        # Финальное обновление без курсора
        if full_text:
            try:
                await bot_msg.edit_text(full_text, parse_mode="Markdown")
            except Exception:
                await bot_msg.edit_text(full_text)  # Fallback без Markdown

            # Сохраняем ответ в историю
            history.append({"role": "assistant", "content": full_text})
            trim_history(chat_id)
        else:
            await bot_msg.edit_text("Не удалось получить ответ. Попробуйте ещё раз.")

    except Exception as e:
        await bot_msg.edit_text(f"Ошибка: {type(e).__name__}")


def main():
    app = ApplicationBuilder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear))
    app.add_handler(CommandHandler("model", model_info))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    print(f"Бот запущен, модель: {MODEL}")
    app.run_polling()


if __name__ == "__main__":
    main()

Разберём ключевые моменты:

Streaming через OpenAI SDK. Параметр stream=True возвращает асинхронный итератор. Каждый чанк содержит delta.content — фрагмент текста длиной 1–5 токенов.

Периодическое обновление. Обновляем сообщение каждые 15 чанков (~0.5–1 секунда). Чаще — упрёмся в rate limit Telegram API. Реже — пользователь не видит «живую» генерацию.

Курсор ▌. Мигающий курсор в конце текста создаёт ощущение печати. Убираем его в финальном обновлении.

Fallback без Markdown. Если AI сгенерировал невалидный Markdown, финальное обновление отправляется без parse_mode.

Продвинутые возможности

Обработка изображений

Современные модели (GPT-4.1, Claude 4.6, Gemini 3.1) поддерживают мультимодальный ввод. Добавим обработку фотографий:

async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    photo = update.message.photo[-1]  # Берём максимальное разрешение
    file = await photo.get_file()

    # Получаем URL файла
    file_url = file.file_path
    caption = update.message.caption or "Что на этом изображении?"

    history = get_history(chat_id)
    history.append({
        "role": "user",
        "content": [
            {"type": "text", "text": caption},
            {"type": "image_url", "image_url": {"url": file_url}},
        ],
    })

    bot_msg = await update.message.reply_text("🔍 Анализирую изображение...")

    response = await ai.chat.completions.create(
        model="gpt-4.1",  # Мультимодальная модель
        messages=[{"role": "system", "content": SYSTEM_PROMPT}] + history,
    )

    reply = response.choices[0].message.content
    await bot_msg.edit_text(reply)
    history.append({"role": "assistant", "content": reply})

Подробнее о работе с Vision, TTS и Whisper API — в руководстве по мультимодальным AI API.

Системный промпт под задачу

Системный промпт определяет характер бота. Несколько готовых вариантов:

# Программист-помощник
SYSTEM_PROMPT = """Ты опытный программист. Отвечай с примерами кода.
Используй Markdown: ```язык для блоков кода. Объясняй кратко."""

# Переводчик
SYSTEM_PROMPT = """Ты профессиональный переводчик.
Определи язык ввода и переведи: с русского на английский, с английского на русский.
Отвечай только переводом, без пояснений."""

# Бизнес-ассистент
SYSTEM_PROMPT = """Ты бизнес-ассистент. Помогаешь с анализом, стратегией,
написанием текстов. Ответы структурируй списками и заголовками."""

Подробнее об оптимизации промптов — в нашей статье про снижение расходов на AI API.

Мультимодельный роутинг

Зачем платить за GPT-5.4 на простых вопросах? Реализуем переключение моделей:

MODELS = {
    "fast": "gpt-4.1-mini",      # Быстрые простые ответы: ~$0.40/1M токенов
    "smart": "claude-sonnet-4-6", # Код и анализ: ~$3/1M токенов
    "pro": "gpt-5.4",            # Сложные задачи: ~$10/1M токенов
    "vision": "gpt-4.1",         # Работа с изображениями
}

# Текущая модель для каждого чата
user_model: dict[int, str] = {}

async def set_model(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    args = context.args

    if not args or args[0] not in MODELS:
        models_list = "\n".join(
            f"• `{key}` — {val}" for key, val in MODELS.items()
        )
        await update.message.reply_text(
            f"Использование: /setmodel <режим>\n\n{models_list}",
            parse_mode="Markdown",
        )
        return

    user_model[chat_id] = args[0]
    model_name = MODELS[args[0]]
    await update.message.reply_text(
        f"Модель переключена на `{model_name}` ({args[0]})",
        parse_mode="Markdown",
    )

Такой подход позволяет пользователю выбирать между скоростью и качеством. Подробнее о сравнении моделей — в бенчмарке 8 моделей.

Обработка ошибок и fallback

В продакшене AI API может быть временно недоступен. Реализуем автоматический fallback:

FALLBACK_CHAIN = ["gpt-4.1-mini", "gemini-3-flash", "deepseek-v3"]

async def call_ai_with_fallback(messages: list[dict]) -> str:
    """Вызывает AI API с автоматическим переключением при ошибке."""
    last_error = None

    for model in FALLBACK_CHAIN:
        try:
            response = await ai.chat.completions.create(
                model=model,
                messages=messages,
                timeout=30,
            )
            return response.choices[0].message.content
        except Exception as e:
            last_error = e
            print(f"Модель {model} недоступна: {e}")
            continue

    raise last_error

Ключевые принципы:

• Таймаут 30 секунд — не заставляйте пользователя ждать бесконечно
• Fallback-цепочка — от дешёвой модели к ещё более дешёвой, не от дорогой к дешёвой
• Логирование — каждый fallback записывайте в лог для мониторинга

Подробнее о стратегиях обработки ошибок — в справочнике по ошибкам AI API.

Деплой: от polling к webhook

Polling отлично работает на этапе разработки, но для продакшена webhook надёжнее и эффективнее.

Вариант 1: fly.io (бесплатный уровень)

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "bot.py"]

# fly.toml
app = "my-ai-telegram-bot"
primary_region = "ams"

[http_service]
  internal_port = 8080
  force_https = true

[env]
  AI_MODEL = "gpt-4.1-mini"

Для webhook-режима замените app.run_polling() на:

from aiohttp import web

async def webhook_handler(request):
    data = await request.json()
    update = Update.de_json(data, app.bot)
    await app.process_update(update)
    return web.Response(status=200)

# В main():
await app.bot.set_webhook(url="https://my-ai-telegram-bot.fly.dev/webhook")
web_app = web.Application()
web_app.router.add_post("/webhook", webhook_handler)
web.run_app(web_app, port=8080)

Вариант 2: VPS с systemd

# /etc/systemd/system/telegram-bot.service
[Unit]
Description=AI Telegram Bot
After=network.target

[Service]
User=botuser
WorkingDirectory=/home/botuser/bot
EnvironmentFile=/home/botuser/bot/.env
ExecStart=/home/botuser/bot/venv/bin/python bot.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot

Чеклист перед деплоем

• Переменные окружения вынесены в .env, не захардкожены
• .env добавлен в .gitignore
• Установлен таймаут на AI API вызовы
• Реализован fallback между моделями
• Добавлена команда /start с описанием бота
• Протестированы длинные ответы (Telegram лимит — 4096 символов на сообщение)
• Логирование ошибок включено

Итоги

Что мы создали:

1. Базовый AI-бот — 50 строк, отвечает на любой вопрос
2. Streaming-бот — живая генерация текста с историей диалога
3. Мультимодельный роутинг — переключение между GPT, Claude, Gemini
4. Продакшен-версия — fallback, обработка ошибок, webhook-деплой

Telegram — идеальная платформа для AI-ботов: 900 миллионов активных пользователей, мощный Bot API с нативным streaming (9.5), бесплатная инфраструктура доставки сообщений. А с OpenAI-совместимым API подключение любой модели — это замена одной строки.

Полезные ссылки

• Telegram Bot API документация
• python-telegram-bot на GitHub
• Руководство по AI Agent на Python — если хотите добавить боту инструменты (поиск, калькулятор, база данных)
• Function Calling: полное руководство — как научить бота вызывать внешние API
• Настройка API в Cursor, Claude Code, Cline — если используете AI не только в Telegram