Разработка AI-бота для Telegram на базе LLM

Разработка AI-бота для Telegram на базе LLM

Telegram Bot с LLM — самый быстрый способ дать пользователям доступ к AI-ассистенту. Telegram Bot API поддерживает: текст, голосовые сообщения (через Whisper), изображения (через vision API), встроенные кнопки, группы и каналы.

Базовый AI-бот с aiogram 3.x

import asyncio
from aiogram import Bot, Dispatcher, Router, types, F
from aiogram.filters import CommandStart, Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.storage.redis import RedisStorage
from anthropic import AsyncAnthropic

BOT_TOKEN = "BOT_TOKEN"
ANTHROPIC_API_KEY = "ANTHROPIC_API_KEY"

bot = Bot(token=BOT_TOKEN)
storage = RedisStorage.from_url("redis://localhost:6379")
dp = Dispatcher(storage=storage)
router = Router()
dp.include_router(router)
anthropic_client = AsyncAnthropic(api_key=ANTHROPIC_API_KEY)

# История диалога — хранится в FSM context
@router.message(CommandStart())
async def start(message: types.Message, state: FSMContext):
    await state.update_data(history=[])
    await message.answer(
        "Привет! Я AI-ассистент на базе Claude. Задай любой вопрос.",
        reply_markup=get_main_keyboard()
    )

def get_main_keyboard():
    from aiogram.utils.keyboard import ReplyKeyboardBuilder
    builder = ReplyKeyboardBuilder()
    builder.button(text="🗑 Очистить историю")
    builder.button(text="ℹ️ О боте")
    builder.adjust(2)
    return builder.as_markup(resize_keyboard=True)

@router.message(F.text == "🗑 Очистить историю")
async def clear_history(message: types.Message, state: FSMContext):
    await state.update_data(history=[])
    await message.answer("История диалога очищена.")

@router.message(F.text & ~F.text.startswith("/"))
async def handle_text(message: types.Message, state: FSMContext):
    data = await state.get_data()
    history = data.get("history", [])

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

    # Показываем "печатает..."
    await bot.send_chat_action(message.chat.id, "typing")

    # Streaming ответ
    full_response = ""
    sent_message = await message.answer("...")

    async with anthropic_client.messages.stream(
        model="claude-haiku-4-5",
        max_tokens=2048,
        system="Ты — полезный ассистент. Отвечай кратко и по существу.",
        messages=history,
    ) as stream:
        async for text in stream.text_stream:
            full_response += text
            # Обновляем сообщение каждые 50 символов
            if len(full_response) % 50 == 0:
                await sent_message.edit_text(full_response)

    await sent_message.edit_text(full_response)

    # Сохраняем ответ в историю
    history.append({"role": "assistant", "content": full_response})
    # Ограничиваем историю — 20 сообщений
    await state.update_data(history=history[-20:])

Обработка голосовых сообщений

import io
from groq import AsyncGroq

groq_client = AsyncGroq(api_key="GROQ_API_KEY")

@router.message(F.voice)
async def handle_voice(message: types.Message, state: FSMContext):
    await bot.send_chat_action(message.chat.id, "typing")

    # Скачиваем голосовое сообщение
    voice = message.voice
    file = await bot.get_file(voice.file_id)
    file_bytes = await bot.download_file(file.file_path)

    # Транскрибируем через Whisper на Groq (быстро)
    transcription = await groq_client.audio.transcriptions.create(
        file=("voice.ogg", file_bytes.read()),
        model="whisper-large-v3",
        language="ru",
    )

    text = transcription.text
    await message.answer(f"📝 Распознано: {text}")

    # Обрабатываем как текстовое сообщение
    await handle_text_with_content(message, state, text)

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

import base64

@router.message(F.photo)
async def handle_photo(message: types.Message, state: FSMContext):
    await bot.send_chat_action(message.chat.id, "typing")

    # Берём фото наилучшего качества
    photo = message.photo[-1]
    file = await bot.get_file(photo.file_id)
    file_bytes = await bot.download_file(file.file_path)

    image_data = base64.standard_b64encode(file_bytes.read()).decode()
    caption = message.caption or "Что на этом изображении?"

    response = await anthropic_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_data}},
                {"type": "text", "text": caption},
            ]
        }]
    )

    await message.answer(response.content[0].text)

Rate limiting и billing защита

from aiogram.filters import BaseFilter
from collections import defaultdict
import time

class RateLimitFilter(BaseFilter):
    def __init__(self, rate_limit: int = 10, period: int = 60):
        self.rate_limit = rate_limit
        self.period = period
        self.user_requests = defaultdict(list)

    async def __call__(self, message: types.Message) -> bool:
        user_id = message.from_user.id
        now = time.time()

        # Убираем старые запросы
        self.user_requests[user_id] = [
            t for t in self.user_requests[user_id]
            if now - t < self.period
        ]

        if len(self.user_requests[user_id]) >= self.rate_limit:
            await message.answer("⏳ Слишком много запросов. Подождите минуту.")
            return False

        self.user_requests[user_id].append(now)
        return True

# Применяем фильтр
@router.message(RateLimitFilter(rate_limit=10), F.text)
async def handle_with_limit(message: types.Message, state: FSMContext):
    await handle_text(message, state)

Практический кейс: клиентская поддержка

Результат: Telegram-бот для SaaS-продукта. 73% обращений обрабатывались без участия поддержки. Время ответа: 24 часа → мгновенно.

Сроки

• Базовый бот с диалогом: 2–3 дня
• Голос + изображения: 2–3 дня
• Rate limiting + admin panel: 1 неделя
• Публикация + мониторинг: +2–3 дня