Архитектура Telegram-бота 2026 — Middleware, Redis и подключение нейросетей

Большинство уроков для начинающих предлагают писать весь код бота в одном файле main.py. Это отличный подход для первого запуска “Hello World”, но в реальных проектах, где нужно поддерживать сложную логику, такой код быстро становится проблемой.

Сегодня мы напишем асинхронного Telegram-бота на Python, используя подходы коммерческой разработки (Clean Architecture). Мы будем использовать aiogram 3.x в качестве фреймворка, Pydantic для строгой типизации настроек, Redis для хранения состояния (FSM) и DeepSeek API для интеграции с нейросетью.

Что нам понадобится

Для начала работы убедитесь, что у вас установлены:

• Python 3.10+ (рекомендую 3.12 для максимальной скорости).
• Redis (локально или в Docker).
• Токен бота от @BotFather.
• API ключ от DeepSeek (или OpenAI/Anthropic — код легко адаптируется).

Настройка окружения

Для управления зависимостями и версиями Python я настоятельно рекомендую использовать uv — это ультра-быстрый инструмент, который заменяет pip, pip-tools и venv.

1. Установка uv (если еще нет):

   curl -LsSf https://astral.sh/uv/install.sh | sh

2. Инициализация проекта и виртуального окружения:

   mkdir my-bot && cd my-bot
   uv init
   uv venv
   source .venv/bin/activate  # Для Linux/macOS
   # .venv\Scripts\activate   # Для Windows

3. Установка библиотек:

   uv pip install aiogram redis pydantic-settings aiohttp

   (Если вы используете классический pip, просто введите pip install ...)

Правильная архитектура проекта

Главная сложность при работе с LLM (языковыми моделями) в ботах — необходимость помнить контекст диалога. Если пользователь написал “Привет”, а следующим сообщением “Как дела?”, нейросеть должна понимать, что это продолжение беседы.

Обычные переменные Python здесь не помогут: при перезапуске бота или обновлении кода вся память в оперативной памяти (RAM) сотрется. Поэтому для хранения истории мы используем быстрое Key-Value хранилище Redis.

Почему именно Redis?

1. MemoryStorage (RAM) — самый простой вариант, но при любом перезапуске бота (деплой, ошибка) все диалоги пользователей исчезнут. Это плохой UX.
2. База данных (PostgreSQL/SQLite) — можно хранить стейты там, но это медленнее и создает лишнюю нагрузку на диск.
3. Redis — золотая середина. Он хранит данные в оперативной памяти (молниеносно быстро), но умеет сохранять их на диск. Плюс, у него есть встроенный механизм TTL (Time To Live) — старые диалоги удалятся сами, не засоряя память.

Когда Redis не подойдет? Если вы строите “тяжелую” CRM или ERP систему, где потеря даже одного состояния (например, наполовину заполненной анкеты) критична для бизнеса, или если вам нужно делать сложные выборки по истории диалогов (SQL-запросы). В таких случаях лучше использовать PostgreSQL.

Структура нашего проекта будет модульной:

bot/
├── config.py          # Валидация и загрузка переменных окружения
├── main.py            # Точка входа (Entrypoint)
├── middlewares/       # Прослойки (DI - внедрение зависимостей)
└── handlers/
    └── chat.py        # Логика обработки сообщений

1. Безопасные настройки — config.py

Использовать os.getenv напрямую в коде — плохая практика. Если вы забудете добавить токен в .env, бот может запуститься, но упасть с ошибкой в самый неподходящий момент. Лучше использовать Pydantic Settings: он проверит наличие всех переменных еще на старте.

from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import SecretStr

class Settings(BaseSettings):
    BOT_TOKEN: SecretStr
    DEEPSEEK_API_KEY: SecretStr
    REDIS_URL: str = "redis://localhost:6379/0"

    model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8')

config = Settings()

Теперь, если переменная отсутствует, приложение сразу сообщит об ошибке.

2. Middleware — Внедрение зависимостей (Dependency Injection)

В aiogram 3 отлично реализован механизм Middleware. Это позволяет нам создать один экземпляр сессии aiohttp (HTTP-клиент) при старте бота и “пробрасывать” его в каждый хендлер. Это значительно экономит ресурсы сервера, так как не нужно открывать новое TCP-соединение на каждое сообщение пользователя.

Создадим файл middlewares/api.py:

from typing import Callable, Dict, Any, Awaitable
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject
import aiohttp

class HTTPMiddleware(BaseMiddleware):
    def __init__(self, session: aiohttp.ClientSession):
        self.session = session

    async def __call__(
        self,
        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
        event: TelegramObject,
        data: Dict[str, Any]
    ) -> Any:
        # Кладём сессию в data, чтобы она была доступна в хендлере
        data["session"] = self.session
        return await handler(event, data)

3. Умный хендлер с памятью — handlers/chat.py

А вот и “мозги”. Мы будем использовать Redis (через aioredis), чтобы хранить историю переписки.

from aiogram import Router, F
from aiogram.types import Message
from aiogram.fsm.context import FSMContext
import aiohttp
import json
from config import config

router = Router()

@router.message()
async def chat_with_ai(message: Message, session: aiohttp.ClientSession, state: FSMContext):
    # Показываем статус "печатает..." для обратной связи
    await message.bot.send_chat_action(chat_id=message.chat.id, action="typing")
    
    # 1. Загружаем контекст диалога из Redis
    data = await state.get_data()
    history = data.get("history", [])
    
    # 2. Добавляем вопрос пользователя
    history.append({"role": "user", "content": message.text})
    
    # Ограничиваем историю последними 10 сообщениями (экономия токенов)
    history = history[-10:]

    # 3. Формируем запрос к DeepSeek API
    headers = {
        "Authorization": f"Bearer {config.DEEPSEEK_API_KEY.get_secret_value()}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "deepseek-chat",
        "messages": history,
        "temperature": 0.7
    }

    try:
        async with session.post("https://api.deepseek.com/chat/completions", json=payload, headers=headers) as resp:
            if resp.status != 200:
                await message.answer(f"Ошибка API: {resp.status}")
                return

            result = await resp.json()
            ai_answer = result['choices'][0]['message']['content']
            
            # 4. Сохраняем ответ ассистента в историю
            history.append({"role": "assistant", "content": ai_answer})
            await state.update_data(history=history)
            
            # Отправляем ответ пользователю (с поддержкой Markdown)
            await message.answer(ai_answer, parse_mode="Markdown")
            
    except Exception as e:
        await message.answer(f"Произошла ошибка при запросе: {e}")

4. Сборка приложения — main.py

В точке входа мы инициализируем хранилище RedisStorage и подключаем нашу Middleware. Обратите внимание на использование lifespan логики через контекстный менеджер — мы открываем сессию при старте и обязательно закрываем её при выходе.

import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.fsm.storage.redis import RedisStorage, DefaultKeyBuilder
from redis.asyncio import from_url
import aiohttp

from config import config
from handlers import chat
from middlewares.api import HTTPMiddleware

async def main():
    logging.basicConfig(level=logging.INFO)

    # Инициализация Redis (FSM)
    redis = from_url(config.REDIS_URL)
    storage = RedisStorage(redis=redis, key_builder=DefaultKeyBuilder(with_bot_id=True))
    
    bot = Bot(token=config.BOT_TOKEN.get_secret_value())
    dp = Dispatcher(storage=storage)

    # Создаем одну сессию на всё время жизни бота
    async with aiohttp.ClientSession() as session:
        # Подключаем Middleware
        dp.message.middleware(HTTPMiddleware(session))
        
        # Регистрируем роутеры
        dp.include_router(chat.router)

        await bot.delete_webhook(drop_pending_updates=True)
        try:
            await dp.start_polling(bot)
        finally:
            await bot.session.close()
            await redis.close()

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

Преимущества такой архитектуры

1. Масштабируемость: Добавить новый функционал (например, генерацию картинок) можно просто создав новый роутер в handlers/image.py. Код остается чистым и понятным.
2. Производительность: aiohttp сессия переиспользуется (keep-alive), а Redis хранит данные в памяти. Бот отвечает пользователю с минимальной задержкой.
3. Контекст: Благодаря FSM в Redis, бот помнит историю диалога даже при перезагрузке сервера.

Поздравляю! Вы создали полноценного AI-ассистента с архитектурой уровня Middle Python Developer. Теперь его можно смело развивать и показывать в портфолио.