Как создать Telegram-бота с AI: Пошаговое руководство

Telegram-боты с AI — один из самых популярных запросов, которые мы получаем от клиентов. Бизнес хочет автоматизировать поддержку, создать персонального ассистента или просто дать пользователям удобный интерфейс к языковой модели. В этом руководстве мы пошагово создадим полноценного Telegram-бота с AI, который умеет вести диалог с памятью, обрабатывать изображения и экономно расходовать бюджет на API.

Технологический стек

Для проекта мы выбрали следующие инструменты:

Python 3.11+ — основной язык разработки
aiogram 3 — современный асинхронный фреймворк для Telegram Bot API
OpenAI API / Anthropic API — языковые модели для генерации ответов
Redis — хранение истории диалогов и кэширование
Docker — контейнеризация для деплоя
systemd или Docker Compose — запуск в production

Почему aiogram 3, а не python-telegram-bot

Aiogram 3 — это полностью асинхронный фреймворк, написанный с нуля на asyncio. Преимущества:

Нативная поддержка async/await
Встроенная система middleware
Удобная система фильтров и роутеров
Активное сообщество и хорошая документация на русском языке
Высокая производительность при большом количестве запросов

Шаг 1: Регистрация бота в Telegram

Откройте Telegram и найдите бота @BotFather. Отправьте команду /newbot и следуйте инструкциям:

Введите имя бота (отображаемое имя в чатах)
Введите username бота (должен заканчиваться на bot)
Сохраните полученный API-токен — он потребуется далее

Дополнительные настройки через BotFather:

/setdescription — описание бота
/setabouttext — текст "О боте"
/setuserpic — аватарка бота
/setcommands — список команд

Шаг 2: Настройка проекта

Структура проекта

telegram-ai-bot/
├── bot/
│   ├── __init__.py
│   ├── main.py           # Точка входа
│   ├── config.py          # Конфигурация
│   ├── handlers/
│   │   ├── __init__.py
│   │   ├── commands.py    # Обработчики команд
│   │   └── messages.py    # Обработчики сообщений
│   ├── services/
│   │   ├── __init__.py
│   │   ├── ai_service.py  # Работа с AI API
│   │   └── memory.py      # Управление памятью диалогов
│   └── middlewares/
│       ├── __init__.py
│       └── throttling.py  # Ограничение частоты запросов
├── .env                   # Переменные окружения
├── .env.example
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

Установка зависимостей

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

aiogram==3.13.1
openai==1.56.0
anthropic==0.39.0
redis==5.2.1
python-dotenv==1.0.1
aiohttp==3.11.9

Установите зависимости:

python -m venv venv
source venv/bin/activate  # Linux/macOS
pip install -r requirements.txt

Конфигурация

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

# Telegram
BOT_TOKEN=your_telegram_bot_token

# AI Provider (openai или anthropic)
AI_PROVIDER=anthropic

# OpenAI
OPENAI_API_KEY=your_openai_api_key
OPENAI_MODEL=gpt-4o

# Anthropic
ANTHROPIC_API_KEY=your_anthropic_api_key
ANTHROPIC_MODEL=claude-3-5-sonnet-latest

# Redis
REDIS_URL=redis://localhost:6379/0

# Settings
MAX_HISTORY_MESSAGES=20
MAX_TOKENS=2048
SYSTEM_PROMPT=Ты — полезный AI-ассистент. Отвечай кратко и по существу на русском языке.

Файл bot/config.py:

from dataclasses import dataclass
from os import getenv
from dotenv import load_dotenv

load_dotenv()


@dataclass
class Config:
    bot_token: str = getenv("BOT_TOKEN", "")
    ai_provider: str = getenv("AI_PROVIDER", "anthropic")

    openai_api_key: str = getenv("OPENAI_API_KEY", "")
    openai_model: str = getenv("OPENAI_MODEL", "gpt-4o")

    anthropic_api_key: str = getenv("ANTHROPIC_API_KEY", "")
    anthropic_model: str = getenv("ANTHROPIC_MODEL", "claude-3-5-sonnet-latest")

    redis_url: str = getenv("REDIS_URL", "redis://localhost:6379/0")

    max_history: int = int(getenv("MAX_HISTORY_MESSAGES", "20"))
    max_tokens: int = int(getenv("MAX_TOKENS", "2048"))
    system_prompt: str = getenv(
        "SYSTEM_PROMPT",
        "Ты — полезный AI-ассистент. Отвечай кратко и по существу."
    )


config = Config()

Шаг 3: Сервис памяти диалогов

Для полноценного диалога боту нужно помнить контекст разговора. Используем Redis для хранения истории:

# bot/services/memory.py
import json
from redis.asyncio import Redis
from bot.config import config


class ConversationMemory:
    def __init__(self):
        self.redis = Redis.from_url(config.redis_url, decode_responses=True)
        self.max_history = config.max_history

    def _key(self, user_id: int) -> str:
        return f"chat_history:{user_id}"

    async def add_message(self, user_id: int, role: str, content: str):
        """Добавить сообщение в историю."""
        key = self._key(user_id)
        message = json.dumps({"role": role, "content": content})
        await self.redis.rpush(key, message)

        # Ограничиваем историю
        length = await self.redis.llen(key)
        if length > self.max_history * 2:  # *2 для пар user/assistant
            await self.redis.ltrim(key, -self.max_history * 2, -1)

        # TTL 24 часа — автоочистка неактивных чатов
        await self.redis.expire(key, 86400)

    async def get_history(self, user_id: int) -> list[dict]:
        """Получить историю диалога."""
        key = self._key(user_id)
        messages = await self.redis.lrange(key, 0, -1)
        return [json.loads(msg) for msg in messages]

    async def clear_history(self, user_id: int):
        """Очистить историю диалога."""
        await self.redis.delete(self._key(user_id))

    async def close(self):
        """Закрыть соединение с Redis."""
        await self.redis.close()


memory = ConversationMemory()

Шаг 4: AI-сервис

Создаём универсальный сервис, который работает с OpenAI и Anthropic:

# bot/services/ai_service.py
import logging
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
from bot.config import config
from bot.services.memory import memory

logger = logging.getLogger(__name__)


class AIService:
    def __init__(self):
        self.provider = config.ai_provider

        if self.provider == "openai":
            self.openai_client = AsyncOpenAI(api_key=config.openai_api_key)
        elif self.provider == "anthropic":
            self.anthropic_client = AsyncAnthropic(
                api_key=config.anthropic_api_key
            )

    async def get_response(self, user_id: int, message: str) -> str:
        """Получить ответ от AI-модели."""
        # Сохраняем сообщение пользователя
        await memory.add_message(user_id, "user", message)

        # Получаем историю
        history = await memory.get_history(user_id)

        try:
            if self.provider == "openai":
                response = await self._openai_request(history)
            elif self.provider == "anthropic":
                response = await self._anthropic_request(history)
            else:
                response = "Ошибка: неизвестный AI-провайдер"

            # Сохраняем ответ ассистента
            await memory.add_message(user_id, "assistant", response)
            return response

        except Exception as e:
            logger.error(f"AI API error: {e}")
            return (
                "Произошла ошибка при обращении к AI. "
                "Попробуйте позже или напишите /reset для сброса диалога."
            )

    async def _openai_request(self, history: list[dict]) -> str:
        """Запрос к OpenAI API."""
        messages = [
            {"role": "system", "content": config.system_prompt}
        ] + history

        response = await self.openai_client.chat.completions.create(
            model=config.openai_model,
            messages=messages,
            max_tokens=config.max_tokens,
            temperature=0.7,
        )
        return response.choices[0].message.content

    async def _anthropic_request(self, history: list[dict]) -> str:
        """Запрос к Anthropic API."""
        response = await self.anthropic_client.messages.create(
            model=config.anthropic_model,
            system=config.system_prompt,
            messages=history,
            max_tokens=config.max_tokens,
            temperature=0.7,
        )
        return response.content[0].text


ai_service = AIService()

Шаг 5: Обработчики команд и сообщений

Обработчики команд

# bot/handlers/commands.py
from aiogram import Router
from aiogram.filters import Command
from aiogram.types import Message
from bot.services.memory import memory

router = Router()


@router.message(Command("start"))
async def cmd_start(message: Message):
    """Приветственное сообщение."""
    await message.answer(
        "Привет! Я — AI-ассистент от QZX Studio.\n\n"
        "Просто напишите мне сообщение, и я постараюсь помочь.\n\n"
        "Команды:\n"
        "/reset — сбросить историю диалога\n"
        "/help — показать справку\n"
        "/model — текущая модель"
    )


@router.message(Command("reset"))
async def cmd_reset(message: Message):
    """Сброс истории диалога."""
    await memory.clear_history(message.from_user.id)
    await message.answer(
        "История диалога очищена. Начнём сначала!"
    )


@router.message(Command("help"))
async def cmd_help(message: Message):
    """Справка по командам."""
    await message.answer(
        "Я могу помочь с:\n"
        "• Ответами на вопросы\n"
        "• Написанием и анализом текстов\n"
        "• Помощью с кодом\n"
        "• Переводами\n"
        "• Анализом изображений (отправьте фото)\n\n"
        "Просто напишите сообщение или отправьте фото с подписью."
    )

Обработчики сообщений

# bot/handlers/messages.py
from aiogram import Router, F
from aiogram.types import Message
from bot.services.ai_service import ai_service

router = Router()


@router.message(F.text)
async def handle_text(message: Message):
    """Обработка текстовых сообщений."""
    # Показываем статус "печатает"
    await message.bot.send_chat_action(
        chat_id=message.chat.id,
        action="typing"
    )

    response = await ai_service.get_response(
        user_id=message.from_user.id,
        message=message.text
    )

    # Telegram ограничивает сообщения 4096 символами
    if len(response) <= 4096:
        await message.answer(response, parse_mode="Markdown")
    else:
        # Разбиваем длинные ответы на части
        for i in range(0, len(response), 4096):
            chunk = response[i:i + 4096]
            await message.answer(chunk, parse_mode="Markdown")


@router.message(F.photo)
async def handle_photo(message: Message):
    """Обработка фотографий (Vision API)."""
    await message.bot.send_chat_action(
        chat_id=message.chat.id,
        action="typing"
    )

    # Получаем файл фото в максимальном разрешении
    photo = message.photo[-1]
    file = await message.bot.get_file(photo.file_id)
    file_url = f"https://api.telegram.org/file/bot{message.bot.token}/{file.file_path}"

    caption = message.caption or "Опиши, что на этом изображении."

    response = await ai_service.get_response(
        user_id=message.from_user.id,
        message=f"[Пользователь отправил изображение: {file_url}]\n{caption}"
    )

    await message.answer(response, parse_mode="Markdown")

Шаг 6: Middleware для rate limiting

Защищаем бота от спама и экономим бюджет на API:

# bot/middlewares/throttling.py
import time
from typing import Any, Awaitable, Callable, Dict
from aiogram import BaseMiddleware
from aiogram.types import Message
from redis.asyncio import Redis
from bot.config import config


class ThrottlingMiddleware(BaseMiddleware):
    def __init__(self, rate_limit: float = 2.0):
        """
        rate_limit — минимальный интервал между сообщениями
        в секундах.
        """
        self.rate_limit = rate_limit
        self.redis = Redis.from_url(
            config.redis_url, decode_responses=True
        )

    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}"

        last_time = await self.redis.get(key)
        current_time = time.time()

        if last_time and current_time - float(last_time) < self.rate_limit:
            await event.answer(
                "Пожалуйста, подождите немного перед следующим сообщением."
            )
            return

        await self.redis.set(key, str(current_time), ex=int(self.rate_limit) + 1)
        return await handler(event, data)

Шаг 7: Точка входа

# bot/main.py
import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode

from bot.config import config
from bot.handlers import commands, messages
from bot.middlewares.throttling import ThrottlingMiddleware
from bot.services.memory import memory

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


async def main():
    # Инициализация бота
    bot = Bot(
        token=config.bot_token,
        default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN)
    )

    # Инициализация диспетчера
    dp = Dispatcher()

    # Подключаем middleware
    dp.message.middleware(ThrottlingMiddleware(rate_limit=2.0))

    # Подключаем роутеры
    dp.include_router(commands.router)
    dp.include_router(messages.router)

    logger.info("Бот запущен")

    try:
        # Удаляем вебхук и запускаем polling
        await bot.delete_webhook(drop_pending_updates=True)
        await dp.start_polling(bot)
    finally:
        await memory.close()
        await bot.session.close()


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

Шаг 8: Docker и деплой

Dockerfile

FROM python:3.11-slim

WORKDIR /app

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

COPY . .

CMD ["python", "-m", "bot.main"]

Docker Compose

version: "3.8"

services:
  bot:
    build: .
    env_file: .env
    depends_on:
      - redis
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

volumes:
  redis_data:

Запуск

# Локальная разработка
python -m bot.main

# Production с Docker
docker compose up -d

# Просмотр логов
docker compose logs -f bot

Оптимизация стоимости

AI API может быстро «съесть» бюджет, если не оптимизировать использование. Вот проверенные стратегии:

1. Ограничение длины истории

Не храните весь диалог — последних 10-20 сообщений обычно достаточно. Мы уже реализовали это в ConversationMemory.

2. Сжатие контекста

Для длинных диалогов используйте суммаризацию:

async def summarize_history(self, user_id: int) -> str:
    """Сжатие длинной истории в краткое резюме."""
    history = await self.get_history(user_id)

    if len(history) < 10:
        return None

    # Используем дешёвую модель для суммаризации
    summary_prompt = (
        "Сделай краткое резюме этого диалога в 2-3 предложениях, "
        "сохранив ключевые факты и контекст:"
    )
    # ... вызов API с дешёвой моделью

3. Выбор модели по задаче

Не все запросы требуют самой мощной модели:

def select_model(self, message: str) -> str:
    """Выбор модели в зависимости от сложности задачи."""
    # Простые запросы — дешёвая модель
    if len(message) < 100 and "код" not in message.lower():
        return "gpt-4o-mini"  # или claude-3-haiku

    # Сложные запросы — мощная модель
    return config.openai_model

4. Кэширование частых запросов

import hashlib

async def get_cached_response(self, message: str) -> str | None:
    """Проверяем кэш для частых запросов."""
    key = f"cache:{hashlib.md5(message.encode()).hexdigest()}"
    cached = await self.redis.get(key)
    return cached

async def cache_response(self, message: str, response: str):
    """Кэшируем ответ на 1 час."""
    key = f"cache:{hashlib.md5(message.encode()).hexdigest()}"
    await self.redis.set(key, response, ex=3600)

5. Мониторинг расходов

Добавьте логирование использования токенов:

async def _openai_request(self, history: list[dict]) -> str:
    response = await self.openai_client.chat.completions.create(
        model=config.openai_model,
        messages=messages,
        max_tokens=config.max_tokens,
    )

    # Логируем использование токенов
    usage = response.usage
    logger.info(
        f"Tokens used: "
        f"input={usage.prompt_tokens}, "
        f"output={usage.completion_tokens}, "
        f"total={usage.total_tokens}"
    )

    return response.choices[0].message.content

Расширенные возможности

Поддержка голосовых сообщений

@router.message(F.voice)
async def handle_voice(message: Message):
    """Обработка голосовых сообщений через Whisper."""
    file = await message.bot.get_file(message.voice.file_id)
    file_path = f"/tmp/{message.voice.file_id}.ogg"

    await message.bot.download_file(file.file_path, file_path)

    # Транскрипция через OpenAI Whisper
    with open(file_path, "rb") as audio:
        transcript = await openai_client.audio.transcriptions.create(
            model="whisper-1",
            file=audio,
            language="ru"
        )

    # Обрабатываем как текстовое сообщение
    response = await ai_service.get_response(
        user_id=message.from_user.id,
        message=transcript.text
    )
    await message.answer(response)

Inline-кнопки для обратной связи

from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton

def feedback_keyboard() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(inline_keyboard=[
        [
            InlineKeyboardButton(text="👍 Полезно", callback_data="feedback:good"),
            InlineKeyboardButton(text="👎 Не помогло", callback_data="feedback:bad"),
        ]
    ])

Заключение

Мы создали полноценного Telegram-бота с AI, который умеет:

Вести диалог с сохранением контекста
Работать с двумя AI-провайдерами (OpenAI и Anthropic)
Ограничивать частоту запросов
Экономить бюджет на API
Легко деплоиться через Docker

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

Если вам нужен Telegram-бот с AI для бизнеса — с кастомной логикой, интеграциями и поддержкой — команда QZX Studio готова разработать решение под ваши задачи.

Содержание (29)

Содержание