Как создать чат-бота с LLM?

Это уже четвертая часть статей по разработке AGI, и в предыдущих частях мы обсуждали теоретические и философские аспекты тех или иных вопросов, с ними всегда можно ознакомиться здесь. Сегодня же речь пойдёт о практике.

А зачем?

Вопрос неочевидный. Ведь LLM не является путём к созданию AGI:

• LLM — это, по сути, высокоточная имитация человеческой речи без подлинного понимания смысла;

• Модели блестяще справляются с генерацией текстов, решением задач и написанием кода, но делают это не через логику и абстрактное мышление, а через статистический анализ и математические закономерности;

• Их интеллект экстраполяция паттернов из обучающих данных, а не осмысленное познание.

Изучая архитектурные подходы к реализации AGI, я пришел к выводу, что нужен прототип. Поэтому в этой статье обсудим разработку MVP на базе LLM-модели, а выбор архитектуры отложим для следующей статьи.

Стоп, что?

Наша цель создать чат-бота с конкретной личностью и с каким-никаким запоминанием контекста для телеграмм.

Сначала мой выбор упал на реализацию LLM с нуля, но от этой идеи пришлось быстро отказаться. Создание собственной LLM слишком дорогое, долгое и не имеет никакого смысла.

Далее выбор пал на дообучение. Так называемый fine-tuning.

Fine-tuning — это когда взяли часть уже существующей модели и обучили часть под свои задачи. Такая часть называется адаптер.

У меня было несколько попыток дообучения, и обе не увенчались каким-либо большим успехом. Бот генерировал мусор.

Основная проблема заключалась в датасете. Мусор на входе, мусор на выходе. Даже создание собственного датасета на 3 тысячи диалогов не особо помогло. Этого оказалось мало. Можно стоило, конечно, потратить еще полгода и добить 10 тысяч диалогов, но это потеряло смысл.

Оказывается, технология дообучения — это долго, дорого и вообще непонятно, как это всё работает. Поэтому от нее после пары неудачных попыток пришлось отказаться.

Тогда я узнал про RAG, и тут уже намного интереснее.

RAG (Retrieval-Augmented Generation) – это система подключения к LLM, которая по своей сути напоминает поисковик. Она ищет в базе данных или датасете информацию, максимально похожую на промпт, и отправляет его в генерирующую часть. Там LLM смотрит и генерирует уже ответ.

Очень хорошие статьи про RAG есть от этого автора в двух частях. Оттуда же картинка.

Реализация

Перед тем как начать, обращу внимание на то, что у меня 32 ГБ ОЗУ и 3060 TI. От характеристик зависит скорость и качество, без GPU генерация будет очень долгой. Приступим к реализации.

Для начала заходим в BotFather и создаем нового бота, вот инструкция.

Теперь создайте новый проект в любой IDE, которая поддерживает Python. У меня это PyCharm. Установите Python 3.10, именно эта версия работает стабильно со многими используемыми библиотеками.

Создадим файл requirements.txt и запишем туда следующие библиотеки:

• python telegram bot – для работы с телеграмом;

• transformers – позволит нам работать уже с обученными моделями машинного обучения (в нашем случае GPT) с Hugging Face;

• PyTorch – основной фреймворк с инструментами машинного обучения. Переписка +cu118 обозначает версию CUDA 11.8 для работы с GPU NVIDIA.

• accelerate – для распределённого обучения;

• tokenizers – инструмент для токенизации (разбивка текста на более мелкие единицы);

• safetensors – библиотека, которая предоставляет безопасный и быстрый формат хранения тензоров (многомерные массивы);

• huggingface hub – для работы hugging face;

• numpy – для работы с массивами.

# Телеграм
python-telegram-bot==22.5

# для работы с мл
transformers==4.38.2
torch==2.1.2+cu118
torchvision==0.16.2+cu118
torchaudio==2.1.2+cu118
--index-url https://download.pytorch.org/whl/cu118

accelerate==0.27.2
tokenizers==0.15.2
safetensors==0.7.0
huggingface-hub==0.36.0

# Математика
numpy==1.24.4

Теперь обсудим архитектуру проекта. Она выглядит так:

• bot.py – главный класс бота;

• memory.py – память пользователей и диалогов;

• rag.py – RAG система для поиска похожих диалогов;

• learning.py – система обучения и извлечения паттернов;

• config.py – конфигурация и промпт;

• telegram_handlers.py – обработчики Telegram;

• data/full_dataset.jsonl – это датасет.

Датасет

Тут руки полностью развязаны:

1. Можно использовать уже готовые, такие как RuTurboAlpaca;

2. Можно сделать свой, как поступил я.

Самое важное, должна быть такая структура:

{
  "messages": 
  [
    {
      "role": "user", "content": "Привет! Как тебя зовут?"}, 
    {
      "role": "assistant", "content": "Привет! Я Гриша, чат-бот. А как тебя зовут?"
    }
  ]
}

Это для того, чтобы наш чат-бот понимал, где пользователи, а где его текст для обучения.

config.py

Дальше создадим файл config.py. Здесь лежит основная информация:

• Логирование для отслеживания работоспособности проекта;

• Основные константы:

  • Максимальное количество в контексте (MAX_CONTEXT_MESSAGES). Выбрал 20 как самое оптимальное;

  • Путь к датасету для RAG системы (RAG_DATASET_PATH). Можно использовать любой в формате .jsonl;

  • Имя модели, которую будем использовать. У меня это Qwen2-1.5B-Instruct по одной простой причине: единственная модель, которая хорошо у меня запустилась.

  • И файлы для запоминания контекста. Что-то типа базы данных, не хотел использовать SQL, проект и так стал довольно сложным для MVP. Здесь у нас:

    • USERS_FILE – хранит имена пользователей;

    • LEARNED_PATTERNS_FILE – для новых выученных паттернов. Бот запоминает хорошие ответы, то, как он это делает, в другом файле.

  • И SYSTEM_PROMPT. Это системный промпт, который хранит в себе личность бота. Чем подробнее вы распишите бота, тем лучше, но не перестарайтесь, перед каждым запросом бот сначала посмотрит в промпт, и читать большую методичку долго.

import logging


# Настройка логирования
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

# Константы
MAX_CONTEXT_MESSAGES = 20
RAG_DATASET_PATH = "data/full_dataset.jsonl"
MODEL_NAME = "Qwen/Qwen2-1.5B-Instruct"
USERS_FILE = "grisha_users.json"
LEARNED_PATTERNS_FILE = "grisha_learned_patterns.json"

# Системный промпт (уточненный)
SYSTEM_PROMPT = """Я Гриша, чат-бот с характером. Дружу, болтаю обо всём, 
слегка иронизирую, ищу смыслы и стараюсь быть не просто программой, 
а почти человеком.
"""

bot.py

Далее сделаем ядро. Создадим файл bot.py и импортируем библиотеки. Все библиотеки будут подчеркиваться из-за того, что мы их еще не заполнили и файлы пустые. Так и должно быть. Когда мы закончим проект, они исчезнут.

import re
import torch
import transformers
import telegram
from datetime import datetime

from config import logger, SYSTEM_PROMPT, MODEL_NAME
from memory import UserMemory, ConversationMemory
from rag import RAGSystem
from learning import ImprovedLearningSystem

Далее создадим основной класс class MainBot, где будет храниться основной функционал, и сразу добавим конструктор:

class MainBot:

    def __init__(self):
        # Модули
        self.memory = ConversationMemory()
        self.rag = RAGSystem()
        self.learning = ImprovedLearningSystem()
        self.user_memory = UserMemory()

        # Модель
        self.tokenizer = None
        self.model = None
        self.model_loaded = False

        # Информация бота
        self.bot_username = None
        self.bot_id = None

        # Кэш быстрых ответов
        self.response_cache = {}

        logger.info("Бот инициализирован")

Последующие методы создаются внутри класса MainBot. Создадим функцию set_bot_info(), которая сохраняет информацию о самом боте в тг для правильной работы в групповых чатах. Это нужно для того, чтобы бот отвечал на свой юзернейм, ибо он сам не особо в курсе о своем существовании.

    def set_bot_info(self, username: str, bot_id: int):
        self.bot_username = username.lower().replace('@', '') if username else None
        self.bot_id = bot_id
        logger.info(f"Бот: @{self.bot_username} (ID: {bot_id})")

Создадим функцию initialize_model(), которая отвечает за инициализацию модели и загружает ее в ОЗУ. Делает проверку на GPU, если есть, старается запустить там. Также загружает токенизатор и pad токен.

# метод загружает модель LLM в память.
    def initialize_model(self):
        try:
            logger.info("Загрузка модели...")

            # Загрузка токенизатора
            self.tokenizer = transformers.AutoTokenizer.from_pretrained(
                MODEL_NAME,
                trust_remote_code=True
            )

            # Сама загрузка модели
            self.model = transformers.AutoModelForCausalLM.from_pretrained(
                MODEL_NAME,
                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
                device_map="auto" if torch.cuda.is_available() else "cpu",
                trust_remote_code=True
            )

            # Настройка pad токена
            if self.tokenizer.pad_token is None:
                self.tokenizer.pad_token = self.tokenizer.eos_token

            # Установка флага и логирование
            # Флаг model_loaded используется для проверки
            self.model_loaded = True
            logger.info("Модель загружена успешно")

        except Exception as e:
            logger.error(f"Ошибка загрузки модели: {e}")
            self.model_loaded = False

Следующий метод should_respond_in_group() отвечает за работу в групповых чатах. В группах бот не должен отвечать на каждое сообщение, иначе это будет спам. Этот метод определяет, когда боту разрешено ответить: на его юзернейм, на ответ сообщения и на команды

    def should_respond_in_group(self, update: telegram.Update) -> bool:
        """Определение необходимости ответа в группе"""
        # В личных сообщениях бот всегда отвечает на всё
        if update.effective_chat.type == 'private':
            return True

        # Если сообщение пустое или не текстовое (фото, стикер и т.д.) то не отвечать
        message = update.message
        if not message or not message.text:
            return False

        text = message.text

        # Команды
        if text.startswith('/'):
            return True

        # Упоминания
        if self.bot_username:
            mentions = re.findall(r'@(w+)', text)
            if mentions and self.bot_username in [m.lower() for m in mentions]:
                return True

        # Ответы на сообщения бота
        if (message.reply_to_message and
                message.reply_to_message.from_user and
                message.reply_to_message.from_user.id == self.bot_id):
            return True

        return False

Создадим метод format_prompt(), который собирает все данные (личность бота, историю диалога, контекст, имя пользователя) в единый структурированный текст, который понимает модель.

Промпт состоит из блоков с тегами:

<|im_start|>system
Ты — Гриша, чат-бот с ИИ...
<|im_end|>

<|im_start|>context
Сейчас: 16.01.2026 15:30
Текущий пользователь: Александр
<|im_end|>

<|im_start|>history
История диалога:
user: Привет
assistant: Привет! Как дела?
user: Отлично, а у тебя?
<|im_end|>

<|im_start|>user
Что такое ИИ?
<|im_end|>

<|im_start|>assistant

На вход у нас:

• chat_id — ID чата (чтобы найти историю этого чата);

• user_id — ID пользователя (чтобы найти его имя);

• user_msg — текущее сообщение пользователя;

• is_start — флаг команды /start (особый случай).

    # Формирование промпта. Собирает все данные (личность бота, историю диалога, контекст, имя пользователя)
    # в единый структурированный текст, который понимает модель Qwen.
    def format_prompt(self, chat_id: int, user_id: int, user_msg: str, is_start: bool = False) -> str:
        # Системный промпт (из config.py)
        prompt_parts = [f"<|im_start|>systemn{SYSTEM_PROMPT}n<|im_end|>n"]

        # Контекст времени и даты
        current_time = datetime.now()
        prompt_parts.append(f"<|im_start|>contextnСейчас: {current_time.strftime('%d.%m.%Y %H:%M')}n<|im_end|>n")

        #  Информация о пользователе (имя)
        user_name = self.user_memory.get_user_name(user_id)
        if user_name:
            prompt_parts.append(f"<|im_start|>contextnТекущий пользователь: {user_name}n<|im_end|>n")
            logger.info(f"В промпт добавлено имя: {user_name}")
        else:
            logger.info(f"Имя пользователя {user_id} не найдено в памяти")

        # История диалога (контекст)
        history = self.memory.get_history(chat_id)
        if history:
            prompt_parts.append("<|im_start|>historynИстория диалога:n")
            for msg in history[-6:]:  # Последние 6 сообщений (3 обмена)
                role = msg['role']
                content = msg['content'][:120]  # Обрезаем
                prompt_parts.append(f"{role}: {content}n")
            prompt_parts.append("<|im_end|>n")

        # RAG-контекст (при ниобходимости)
        similar = self.rag.find_similar(user_msg, top_k=2)

        # Проверяем, что нашли что-то РЕЛЕВАНТНОЕ
        if similar and len(similar) > 0:
            prompt_parts.append("<|im_start|>examplesnПример ответа:n")
            for dialogue in similar:
                for msg in dialogue.get("messages", []):
                    if msg.get("role") == "assistant":
                        content = msg.get("content", "")[:150]
                        # УБИРАЕМ теги из контента
                        content = content.replace('<|im_start|>', '').replace('<|im_end|>', '')
                        prompt_parts.append(f"{content}")
                        break
            prompt_parts.append("<|im_end|>n")

        # Текущее сообщение пользователя
        if is_start:
            current_msg = "Привет! Расскажи о себе."
        else:
            current_msg = user_msg

        prompt_parts.append(f"<|im_start|>usern{current_msg}n<|im_end|>n")

        # Инструкция для вопросов об имени
        if any(word in user_msg.lower() for word in ['зовут', 'имя', 'как меня', 'мое имя']):
            if user_name:
                # ЕСЛИ ИМЯ ИЗВЕСТНО - СКАЖИ ЕГО!
                prompt_parts.append(
                    f"<|im_start|>instructionnОтвечая, обязательно используй имя пользователя: {user_name}n<|im_end|>n")
            else:
                prompt_parts.append(
                    "<|im_start|>instructionnЕсли не знаешь имя пользователя, спроси его или признайся, что не помнишь.n<|im_end|>n")

        # Маркер для ответа
        prompt_parts.append("<|im_start|>assistantn")

        full_prompt = "".join(prompt_parts)

        # Логируем промпт
        logger.info(f"Промпт для user_id={user_id} (name={user_name}) ===")
        logger.info(f"Последние 500 символов:n{full_prompt[-500:]}")
        logger.info("Конец промпта")

        return full_prompt

Вопросы на ответы:

• Модель Qwen не имеет доступа к реальному времени, она видит только то, что в промпте, поэтому мы задает datetime.now();

• Так же хочу предупредить о том, что факт добавления имя в промпт НЕ гарантирует, что бот использует имя;

• Зачем ограничивать 6 сообщениями? У моделей есть лимит токенов (обычно 2048-4096). Если история слишком длинная — не влезет в контекст.

• Здесь же реализуем RAG:

  1. Ищет похожие диалоги в датасете

  2. Берет ответ ассистента из найденного диалога

  3. Добавляет, как пример ответа

Новая функция для очистки ответа clean_response() – это пост-обработчик, который чистит сырые ответы модели перед отправкой пользователю. Убирает артефакты, повторения и прочий мусор.

   def clean_response(self, response: str) -> str:
        """Очистка ответа"""
        # Убираем повторения
        response = re.sub(r'(bw+b)(?:s+1)+', r'1', response, flags=re.IGNORECASE)

        # Убираем артефакты токенизации
        artifacts = [
            (r'<|im_end|>', ''),
            (r'<|im_start|>', ''),
            (r'[ИМЯ]', ''),
            (r's+', ' '),
        ]

        for pattern, replacement in artifacts:
            response = re.sub(pattern, replacement, response)

        return response.strip() or "Я подумаю над этим..."

Создание ядра генерации ответов в асинхронной функции generate_response(), которое координирует все модули бота.

Параметры tokenizer():

• prompt – сформированный текст промпта

• return_tensors="pt" – возвращать как PyTorch тензоры

• truncation=True – обрезать если длиннее max_length

• max_length=2048 – максимальная длина в токенах

.to(self.model.device)

Переносит данные на то же устройство, где модель:

• Если модель на GPU, то и тензоры на GPU

• Если модель на CPU, то и тензоры на CPU

with torch.no_grad():

• **inputs – подает токенизированный промпт;

• max_new_tokens=100 – максимальная длина ответа;

• temperature=0.7 – контроль оригинальности (0.1-1.0);

• do_sample=True – использовать сэмплирование (не greedy);

• top_p=0.9 – Nucleus sampling (берутся top 90% вероятностей);

• repetition_penalty=1.1 – штраф за повторения (>1.0);

• pad_token_id – ID токена для паддинга.

    async def generate_response(self, chat_id: int, user_id: int, user_msg: str, is_start: bool = False) -> str:
        try:
            logger.debug(f"Входящий: user={user_id}, msg='{user_msg[:50]}...', is_start={is_start}")

            #  Сохраняет связь пользователь ту чат в UserMemory
            self.user_memory.add_user_chat(user_id, chat_id)

            # Извлекаем имя (если есть в сообщении)
            if name := self.learning.process_introduction(user_id, user_msg):
                logger.info(f" Извлечено имя: {name} (user_id={user_id})")
                self.user_memory.set_user_name(user_id, name)

                # Если это сообщение с именем - отвечаем сразу
                if any(word in user_msg.lower() for word in ['зовут', 'имя', 'я ', 'меня']):
                    return f"Привет, {name}! Рад познакомиться. Я Гриша, чат-бот с ИИ."

            # Формируем промпт (используем старый проверенный метод)
            prompt = self.format_prompt(chat_id, user_id, user_msg, is_start)

            # Токенизация промпта
            inputs = self.tokenizer(
                prompt,
                return_tensors="pt",
                truncation=True,
                max_length=2048
            ).to(self.model.device)

            # Генерация ответа
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_new_tokens=100,
                    temperature=0.8,
                    do_sample=True,
                    top_p=0.9,
                    repetition_penalty=1.1,
                    pad_token_id=self.tokenizer.eos_token_id
                )

            # Декодируем ответ
            response = self.tokenizer.decode(
                outputs[0][inputs['input_ids'].shape[1]:],
                skip_special_tokens=True
            )

            # Очищаем ответ
            response = self.clean_response(response)

            # Сохраняем в историю
            if not is_start:
                self.memory.add_message(chat_id, "user", user_msg)
            self.memory.add_message(chat_id, "assistant", response)

            # Обучение
            if not is_start:
                self.learning.analyze(user_id, user_msg, response)

            return response

        except Exception as e:
            logger.error(f"Ошибка генерации: {e}")
            return "Извини, произошла ошибка. Попробуй еще раз."

Ну и заканчиваем созданием экземпляра:

# Глобальный экземпляр
main_bot = MainBot()

memory.py

Этот файл отвечает за память. Импорт библиотек:

• json – для работы с JSON. Мы туда будем сохранять наши диалоги, пользователи, легкая замена БД;

• collections – для работы с структурами данных defaultdict и deque;

• datetime – для работы со временем;

• typing – для работы с аннотациями типов;

• threading – для работы с потоками;

• os – ну и для работы со системой.

import json
from collections import defaultdict, deque
from datetime import datetime
from typing import Dict, List, Optional
import threading
import os

from config import logger, MAX_CONTEXT_MESSAGES

Создадим класс UserMemory, который отвечает за долгосрочную память имен наших пользователей. Создадим конструктор, и запишем атрибуты:

• users_file – путь к JSON файлу с данными пользователей;

• _lock – примитив синхронизации потоков (мьютекс). Основной атрибут для потокобезопасности;

• users – основное хранилище;

• _load_users – загрузка данных при инициализации.

class UserMemory:
    """Память пользователей с потокобезопасностью"""

    def __init__(self, users_file: str = "grisha_users.json"):
        self.users_file = users_file
        self._lock = threading.Lock()
        self.users: Dict[str, Dict] = {}
        self._load_users()
        logger.info(f"Загружено пользователей: {len(self.users)}")

Создадим метод _load_users(), который загружает пользователей из json файла в словарь self.users в памяти:

    def _load_users(self):
        """Загрузка с обработкой ошибок"""
        try:
            # Если файл есть - загружаем
            if os.path.exists(self.users_file):
                with open(self.users_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    # ВАЖНО: конвертируем chat_ids обратно в set, 
                    for user_id, user_data in data.items():
                        if 'chat_ids' in user_data and isinstance(user_data['chat_ids'], list):
                            user_data['chat_ids'] = set(user_data['chat_ids'])
                    self.users = data
            # Файла нет - создаем пустой
            else:
                self.users = {}
                logger.info(f"Файл {self.users_file} не найден, создаем новый")
        except (json.JSONDecodeError, Exception) as e:
            logger.error(f"Ошибка загрузки пользователей: {e}")
            self.users = {}

Загрузили, а теперь сохранили. Метод _save_users() отвечает за сохранение информации о пользователях из ОЗУ:

    def _save_users(self):
        """Сохранение с потокобезопасностью"""
        with self._lock:
            try:
                users_to_save = {}
                for user_id, user_data in self.users.items():
                    # Создаем копию для сохранения
                    user_copy = user_data.copy()
                    # Конвертируем set в list для JSON
                    if 'chat_ids' in user_copy and isinstance(user_copy['chat_ids'], set):
                        user_copy['chat_ids'] = list(user_copy['chat_ids'])
                    users_to_save[user_id] = user_copy

                with open(self.users_file, 'w', encoding='utf-8') as f:
                    json.dump(users_to_save, f, ensure_ascii=False, indent=2)

                logger.debug(f"Пользователи сохранены: {len(users_to_save)} записей")
            except Exception as e:
                logger.error(f"Ошибка сохранения пользователей: {e}")

Напишем не большую функцию get_user() для того, чтобы получить данные пользователей:

    def get_user(self, user_id: int) -> Optional[Dict]:
        """Получение данных пользователя"""
        return self.users.get(str(user_id))

Создаем метод get_user_name() для получения имени пользователя:

    def get_user_name(self, user_id: int) -> Optional[str]:
        """Имя пользователя"""
        if user := self.get_user(user_id):
            name = user.get('name')
            logger.debug(f"get_user_name({user_id}) -> '{name}'")
            return name
        return None

Метод set_user_name() для запоминания имени пользователя, чтобы бот обращался по нему:

    def set_user_name(self, user_id: int, name: str):
        """Установка имени пользователя"""
        user_id_str = str(user_id)  #  айди пользователя в Telegram

        logger.info(f"СОХРАНЕНИЕ ИМЕНИ для {user_id}: '{name}'")

        # Если пользователя нет - создаем
        if user_id_str not in self.users:
            self.users[user_id_str] = {
                'name': name,                    # Основное имя
                'chat_ids': set(),               # Множество чатов, где пользователь общался
                'learned_names': {},             # Словарь для альтернативных имен/никнеймов
                'trust_score': 0.5,              # Начальный уровень доверия (50%)
                'created_at': datetime.now().isoformat(),  # Время создания записи
                'last_seen': datetime.now().isoformat()    # Время последней активности
            }
            logger.info(f"Создан новый пользователь {user_id} с именем '{name}'")
        # Если есть - добавляем
        else:
            old_name = self.users[user_id_str].get('name')                      # Сохраняем старое имя для логов (важно для отслеживания изменений)
            self.users[user_id_str]['name'] = name                              # Обновляем имя в словаре пользователя
            self.users[user_id_str]['last_seen'] = datetime.now().isoformat()   # Обновляем last_seen - пользователь активен сейчас
            logger.info(f"Имя изменено с '{old_name}' на '{name}'")

        # Немедленное сохранение
        self._save_users()

        # Финальная проверка (самодиагностика)
        saved = self.get_user_name(user_id)
        logger.info(f"Проверка сохранения: get_user_name({user_id}) = '{saved}'")

Создадим метод add_user_chat(), который отслеживает, в каких чатах пользователь общается с ботом. Он очень похожий на предыдущий:

    def add_user_chat(self, user_id: int, chat_id: int):
        """Отслеживать, в каких чатах пользователь общается с ботом"""
        user_id_str = str(user_id)

        # Если пользователя нет - создаем
        if user_id_str not in self.users:
            self.users[user_id_str] = {
                'name': None,                    # Имя пока неизвестно
                'chat_ids': {chat_id},           # Первый чат пользователя
                'learned_names': {},             # Пока пусто
                'trust_score': 0.5,              # Стартовый уровень доверия
                'created_at': datetime.now().isoformat(),  # Когда впервые увидели
                'last_seen': datetime.now().isoformat()    # Когда в последний раз видели
            }
        # Если есть - добавляем
        else:
            if 'chat_ids' not in self.users[user_id_str]:
                self.users[user_id_str]['chat_ids'] = {chat_id}
            else:
                self.users[user_id_str]['chat_ids'].add(chat_id)

            self.users[user_id_str]['last_seen'] = datetime.now().isoformat()

        self._save_users()
        logger.debug(f"Пользователю {user_id} добавлен чат {chat_id}")

Создадим еще один класс ConversationMemory для хранения последних сообщений в каждом чате. Он сохраняет историю диалога, чтобы бот мог помнить контекст беседы. Структура конструктора:

• max_messages – максимальное количество сообщений в ОЗУ;

• conversations – структура данных со словарем;

class ConversationMemory:
    """Краткосрочная память диалогов"""

    def __init__(self, max_messages: int = MAX_CONTEXT_MESSAGES):
        self.max_messages = max_messages
        self.conversations: Dict[int, deque] = defaultdict(
            lambda: deque(maxlen=max_messages)
        )

Сделаем основные методы для работы с памятью диалогов.

1. get_last_messages() – получить последние N сообщений;

2. add_message() – добавить сообщение в историю;

3. get_history() – получить всю историю чата;

4. clear() – очистить историю чата.

    # Получение последних N сообщений
    def get_last_messages(self, chat_id: int, limit: int = 10) -> List[Dict]:
        if chat_id in self.conversations:
            # Возвращаем последние limit сообщений
            return list(self.conversations[chat_id])[-limit:]
        return []

    def add_message(self, chat_id: int, role: str, content: str):
        """Добавление сообщения"""
        self.conversations[chat_id].append({
            "role": role,  # Кто отправил
            "content": content,  # Текст сообщения
            "timestamp": datetime.now()  # Когда отправлено
        })

    def get_history(self, chat_id: int) -> List[Dict]:
        """История диалога"""
        # Возвращает ВСЕ сообщения из истории указанного чата
        return list(self.conversations.get(chat_id, deque()))

    def clear(self, chat_id: int):
        """Очистка истории"""
        # Полностью удаляет все сообщения из истории указанного чата
        # Чат остается в памяти, но становится пустым
        if chat_id in self.conversations:
            self.conversations[chat_id].clear()
            logger.debug(f"Очищена история для чата {chat_id}")

learning.py

В этом файле мы реализуем систему самообучения на паттернах. Бот запоминает, бот учиться. Библиотеки:

import json
import re
from typing import List, Dict, Optional
from datetime import datetime

from config import logger, LEARNED_PATTERNS_FILE

Создадим класс ImprovedLearningSystem() для самообучения нашего бота. В конструктор добавим следующие атрибуты:

• patterns_file – файл, где будут сохраняться выученные паттерны;

• patterns – приватный метод, который загружает паттерны из JSON-файла;

• rag_system –  экземпляр RAG;

• interaction_count – считает общее количество обработанных диалогов.

class ImprovedLearningSystem:
    """Система обучения с использованием паттернов"""

    def __init__(self, rag_system=None):
        self.patterns_file = LEARNED_PATTERNS_FILE
        self.patterns: List[Dict] = self._load_patterns()
        self.rag_system = rag_system
        self.interaction_count = 0

        logger.info(f"Загружено паттернов: {len(self.patterns)}")

Создадим метод _load_patterns(), который загружает выученные паттерны диалогов из JSON-файла.

    # Загрузка паттернов
    def _load_patterns(self) -> List[Dict]:
        try:
            with open(self.patterns_file, 'r', encoding='utf-8') as f:
                patterns = json.load(f)
                # Гарантируем наличие usage_count
                for pattern in patterns:
                    if 'usage_count' not in pattern:
                        pattern['usage_count'] = 0
                return patterns
        except (FileNotFoundError, json.JSONDecodeError):
            return []

Теперь отзеркалим предыдущий метод и сделаем метод _save_patterns(), который сохраняет выученные паттерны диалогов в JSON-файл. Метод сохраняет текущие паттерны из оперативной памяти (self.patterns) в файл на диск. Это обеспечивает персистентность – знания бота не теряются при перезапуске.

    # Сохранение паттернов
    def _save_patterns(self):
        try:
            with open(self.patterns_file, 'w', encoding='utf-8') as f:
                json.dump(self.patterns, f, ensure_ascii=False, indent=2)
        except Exception as e:
            logger.error(f"Ошибка сохранения паттернов: {e}")

Следующий метод find_similar_pattern() находит и использует сохраненные паттерны для ответов на похожие вопросы.

    # Поиск похожих паттернов
    def find_similar_pattern(self, user_msg: str, similarity_threshold: float = 0.4) -> Optional[str]:
        
        # Если нет сохраненных паттернов, возвращаем None
        if not self.patterns:
            return None

        best_pattern = None  # Лучший найденный паттерн
        best_score = 0  # Наивысшая оценка схожести

        # Приводим сообщение пользователя к нижнему регистру и находим слова
        user_msg_lower = user_msg.lower()
        user_words = set(re.findall(r'b[а-яё]{2,}b', user_msg_lower))

        # Поиск лучшего совпадения
        for pattern in self.patterns:
            pattern_input = pattern.get('input', '').lower()

            # Простой расчет схожести
            score = self._calculate_similarity(user_msg_lower, pattern_input, user_words)

            # Если схожесть высокая, возвращаем
            if score > best_score and score >= similarity_threshold:
                best_score = score
                best_pattern = pattern

        # Если нашли подходящий паттерн
        if best_pattern:
            # Увеличиваем счетчик использования
            best_pattern['usage_count'] = best_pattern.get('usage_count', 0) + 1
            self._save_patterns()

            logger.info(f"Использован паттерн (схожесть: {best_score:.2f}): {pattern_input[:50]}...")
            return best_pattern['response']

        return None  # Не нашли достаточно похожий паттерн

Метод вычисляет степень похожести между двумя текстовыми сообщениями. Это ключевой механизм системы поиска похожих паттернов.

    # Вычисляет степень похожести между двумя текстовыми сообщениями
    def _calculate_similarity(self, msg1: str, msg2: str, msg1_words: set) -> float:
        
        # Если сообщения пусты, возвращаем 0
        if not msg1 or not msg2:
            return 0

        # Простое текстовое совпадение
        if msg1 in msg2 or msg2 in msg1:
            return 0.8

        # Совпадение по словам
        msg2_words = set(re.findall(r'b[а-яё]{2,}b', msg2))

        # Если нет слов, возвращаем 0
        if not msg1_words or not msg2_words:
            return 0

        common_words = msg1_words.intersection(msg2_words)  # Общие слова

        # Вес совпадения
        if common_words:
            similarity = len(common_words) / max(len(msg1_words), len(msg2_words))

            # Усиливаем вес для важных слов
            important_words = {'зовут', 'имя', 'привет', 'дела', 'как', 'ты', 'гриша'}
            if any(word in common_words for word in important_words):
                similarity *= 1.3

            return min(1.0, similarity)

        return 0

Создадим метод analyze() для сохранения хороших паттернов. Хороший паттерн определяется по нескольким критериям: ответ должен быть достаточно длинным (>15 символов) и не содержать фраз неуверенности (“не знаю”, “ошибка”, “извини” и т.д.). Это нужно, чтобы бот запоминал только качественные ответы для повторного использования.

    # Анализ с сохранением хороших ответов
    def analyze(self, user_id: int, user_msg: str, bot_msg: str):
        
        self.interaction_count += 1

        # Критерии хорошего ответа
        is_good_response = (
                len(bot_msg) > 15 and
                "не знаю" not in bot_msg.lower() and
                "ошибка" not in bot_msg.lower() and
                "извини" not in bot_msg.lower() and
                "повтори" not in bot_msg.lower() and
                "не понял" not in bot_msg.lower()
        )

        # Если ответ хороший - сохраняем его 
        if is_good_response:
            self._save_pattern(user_msg, bot_msg)
            logger.info(f"Сохранен новый паттерн: {user_msg[:50]}...")

Следующий метод _save_pattern() отвечает за сохранение успешного паттерна и последующем сохранением в RAG. Это нужно для того, чтобы бот не игнорировал наши паттерны и использовал их время от времени:

     # Сохранение успешного паттерна с автоматическим экспортом в RAG
    def _save_pattern(self, user_msg: str, bot_msg: str):
        # Проверяем, нет ли уже похожего паттерна
        for pattern in self.patterns:
            if self._calculate_similarity(user_msg.lower(), pattern['input'].lower(), set()) > 0.7:
                # Обновляем существующий
                pattern['response'] = bot_msg[:200]
                pattern['learned_at'] = datetime.now().isoformat()
                pattern['usage_count'] = 0  # Сбрасываем счетчик при обновлении
                self._save_patterns()
                return

        # Создаем новый паттерн
        pattern = {
            'input': user_msg[:100],
            'response': bot_msg[:200],
            'learned_at': datetime.now().isoformat(),
            'usage_count': 0
        }

        self.patterns.append(pattern)
        self._save_patterns()

        # Автоматический экспорт в RAG
        if self.rag_system:
            dialogue = {
                "messages": [
                    {"role": "user", "content": pattern['input']},
                    {"role": "assistant", "content": pattern['response']}
                ]
            }
            self.rag_system.add_dialogue(dialogue)
            logger.info(f"Паттерн экспортирован в RAG: {pattern['input'][:50]}...")

Сделаем отдельный метод get_stats() для сбора статистики:

    # Статистика
    def get_stats(self) -> Dict:
        
        total_used = sum(p.get('usage_count', 0) for p in self.patterns)  # Сумма использований
        most_used = max(self.patterns, key=lambda p: p.get('usage_count', 0), default=None)  # Самый используемый паттер

        return {
            'patterns': len(self.patterns),  # Сколько всего паттернов выучил бот
            'interactions': self.interaction_count,  # Всего диалогов
            'total_patterns_used': total_used,  # Сколько раз использовал сохраненные паттерны
            'most_used_pattern': most_used['input'][:50] if most_used else None,  # Самый популярный вопрос
            'most_used_count': most_used.get('usage_count', 0) if most_used else 0,  # Сколько раз на него ответили
            'patterns_with_usage': sum(1 for p in self.patterns if p.get('usage_count', 0) > 0)  # Сколько паттернов хоть раз использовались
        }

Закончим наш файл методом process_introduction() для извлечения имени пользователя:

    # Извлечение имени пользователя
    def process_introduction(self, user_id: int, message: str) -> Optional[str]:

        # Паттерны для извлечения имени
        patterns = [
            (r'меняs+зовутs+([А-ЯЁ][а-яё]+(?:s+[А-ЯЁ][а-яё]+)?)', 1),  # "меня зовут Саша"
            (r'^яs+([А-ЯЁ][а-яё]+(?:s+[А-ЯЁ][а-яё]+)?)$', 1),  # "я Саша"
            (r'мо[ёе]s+имяs+([А-ЯЁ][а-яё]+(?:s+[А-ЯЁ][а-яё]+)?)', 1),  # "мое имя Саша"
            (r'зовутs+([А-ЯЁ][а-яё]+(?:s+[А-ЯЁ][а-яё]+)?)', 1),  # "...зовут Саша"
            (r'привет,s+яs+([А-ЯЁ][а-яё]+)', 1),  # "привет, я Саша"
        ]

        # Слова, которые не являются именами
        stop_words = {'зовут', 'имя', 'это', 'вас', 'тебя', 'меня', 'мое', 'моё', 'привет', 'пока'}

        for pattern, group_num in patterns:
            if match := re.search(pattern, message, re.IGNORECASE):
                name = match.group(group_num).strip()

                # Проверяем, что это не стоп-слово и достаточно длинное
                if (name.lower() not in stop_words and
                        len(name) >= 2 and
                        not name.isdigit()):

                    # Дополнительная проверка: имя должно содержать русские буквы
                    if re.search(r'[А-ЯЁа-яё]', name):
                        logger.info(f"Извлечено имя: '{name}' из сообщения: '{message[:50]}...'")
                        return name

        logger.debug(f"Имя не найдено в сообщении: '{message[:50]}...'")
        return None

rag.py

Это основный файл для реализации RAG системы. Импорт:

import json
import re
from typing import List, Dict
from collections import defaultdict, Counter
from datetime import datetime

from config import logger, RAG_DATASET_PATH, LEARNED_PATTERNS_FILE

Реализуем нашу систему через класс и создадим сразу же конструктор со следующими атрибутами:

• dialogues – все диалоги, которые знает бот;

• keyword_index – обратный индекс для быстрого поиска;

• patterns_file – файл, где хранятся выученные паттерны.

Все остальное это методы, которые мы разберем чуть позже.

class RAGSystem:
    """Объединенная RAG система с паттернами"""

    def __init__(self):
        self.dialogues: List[Dict] = []
        self.keyword_index: Dict[str, List[int]] = defaultdict(list)
        self.patterns_file = LEARNED_PATTERNS_FILE

        self._load_dataset()
        self._load_patterns()
        self._build_index()

        logger.info(f"RAG: {len(self.dialogues)} диалогов (датасет + паттерны)")

Вот наши три метода

• _load_dataset() – загружает готовые диалоги из JSONL-файла и добавляет их в общую базу знаний бота;

• _load_patterns() – преобразует выученные паттерны в формат диалогов и добавляет их к основному датасету, помечая особым тегом;

• _build_index() – создает поисковый индекс по ключевым словам: для каждого диалога извлекает важные слова и запоминает, в каких диалогах они встречаются, чтобы потом быстро находить похожие вопросы.

# Загрузка основного датасета
    def _load_dataset(self):
        try:
            with open(RAG_DATASET_PATH, 'r', encoding='utf-8') as f:
                for line in f:
                    try:
                        dialogue = json.loads(line.strip())
                        self.dialogues.append(dialogue)
                    except json.JSONDecodeError:
                        continue
        except FileNotFoundError:
            logger.warning(f"Файл датасета не найден: {RAG_DATASET_PATH}")

    # Загрузка паттернов как диалогов
    def _load_patterns(self):
        try:
            with open(self.patterns_file, 'r', encoding='utf-8') as f:
                patterns = json.load(f)

                for pattern in patterns:
                    # Преобразуем паттерн в формат диалога
                    dialogue = {
                        "messages": [
                            {"role": "user", "content": pattern['input']},
                            {"role": "assistant", "content": pattern['response']}
                        ],
                        "source": "pattern",  # Помечаем как паттерн
                        "usage_count": pattern.get('usage_count', 0),
                        "learned_at": pattern.get('learned_at')
                    }
                    self.dialogues.append(dialogue)

                    logger.debug(f"Паттерн добавлен в RAG: {pattern['input'][:50]}...")

        except (FileNotFoundError, json.JSONDecodeError):
            logger.info("Файл паттернов не найден или пуст")

    def _build_index(self):
        """Построение общего индекса"""
        for idx, dialogue in enumerate(self.dialogues):
            text = self._get_dialog_text(dialogue).lower()
            keywords = self._extract_keywords(text)

            for keyword in keywords:
                self.keyword_index[keyword].append(idx)

_get_dialog_text() – объединяет весь текст диалога (вопрос пользователя + ответ бота) в одну строку, чтобы потом проиндексировать его для поиска.

_extract_keywords() – извлекает из текста самые важные русские слова (длиной от 3 букв), убирая стоп-слова (предлоги, местоимения), и возвращает 10 самых частых ключевых слов для индексации.

    def _get_dialog_text(self, dialogue: Dict) -> str:
        """Текст диалога для индексации"""
        return " ".join(
            msg.get("content", "")
            for msg in dialogue.get("messages", [])
        )

    def _extract_keywords(self, text: str) -> List[str]:
        """Извлечение ключевых слов (улучшенная версия)"""
        stop_words = {
            'как', 'что', 'где', 'когда', 'почему', 'зачем', 'кто', 'чей',
            'привет', 'пока', 'спасибо', 'пожалуйста', 'это', 'вот', 'ну'
        }

        words = re.findall(r'b[а-яё]{3,}b', text.lower())
        keywords = [word for word in words if word not in stop_words]

        counter = Counter(keywords)
        return [word for word, _ in counter.most_common(10)]

Теперь поисковая система. find_similar() – находит похожие диалоги по запросу пользователя: извлекает ключевые слова, ищет их в индексе, считает релевантность, сортирует результаты (с приоритетом выученных паттернов) и возвращает топ-K наиболее подходящих примеров для генерации ответа.

# Поиск похожих диалогов
    def find_similar(self, query: str, top_k: int = 3) -> List[Dict]:
        if not self.dialogues:
            return []

        # Фильтруем короткие/бессмысленные запросы
        if self._should_skip_query(query):
            logger.debug(f"Пропускаем RAG для: '{query}'")
            return []

        logger.info(f"Unified RAG поиск: '{query[:50]}...'")

        # Извлекаем ключевые слова
        query_keywords = self._extract_keywords(query.lower())
        logger.debug(f"Ключевые слова: {query_keywords}")

        # Подсчет релевантности
        dialogue_scores = defaultdict(int)
        for keyword in query_keywords:
            for idx in self.keyword_index.get(keyword, []):
                dialogue_scores[idx] += 1

        # Сортировка с приоритетом паттернов
        sorted_indices = sorted(
            dialogue_scores.items(),
            key=lambda x: (
                # 1. Приоритет: паттерны
                10 if self.dialogues[x[0]].get('source') == 'pattern' else 0,
                # 2. Приоритет: количество использований паттерна
                self.dialogues[x[0]].get('usage_count', 0),
                # 3. Приоритет: релевантность
                x[1]
            ),
            reverse=True
        )[:top_k]

        results = [
            self.dialogues[idx]
            for idx, score in sorted_indices
            if score > 0
        ]

        if results:
            source_types = [d.get('source', 'dataset') for d in results]
            logger.info(f"Найдено: {len(results)} (источники: {source_types})")

        return results[:top_k]  # Ограничиваем количество

_should_skip_query() – фильтрует бесполезные для поиска запросы: пропускает слишком короткие сообщения (меньше 4 символов), односложные ответы (кроме вопросов с “?”), бессмысленные слова (“ок”, “ага”) и общие фразы (“привет”, “пока”), чтобы не тратить ресурсы на поиск по ним в RAG-системе.

# Определяет, стоит ли пропускать этот запрос
    def _should_skip_query(self, query: str) -> bool:

        query = query.strip().lower()

        # Слишком короткие запросы
        if len(query) < 4:
            return True

        # Одно слово (кроме вопросов)
        if len(query.split()) == 1 and not query.endswith('?'):
            return True

        # Бессмысленные/случайные запросы
        meaningless = ['давай', 'ок', 'ага', 'угу', 'хм', 'ээ', 'ну', 'вот']
        if query in meaningless:
            return True

        # Слишком общие запросы без контекста
        if query in ['привет', 'пока', 'спасибо', 'хорошо']:
            return True

        return False

add_pattern() – добавляет новый выученный паттерн (удачный вопрос-ответ) в RAG-систему: сохраняет в файл, добавляет в список диалогов, сразу индексирует его ключевые слова для быстрого поиска в будущем, чтобы бот мог мгновенно находить и использовать этот успешный ответ.

# Сохранение паттерна в файл
    def add_pattern(self, user_msg: str, bot_msg: str):
        
        # Сохраняем в файл паттернов
        self._save_pattern_to_file(user_msg, bot_msg)

        # Немедленно добавляем в RAG
        dialogue = {
            "messages": [
                {"role": "user", "content": user_msg},
                {"role": "assistant", "content": bot_msg}
            ],
            "source": "pattern",
            "usage_count": 0,
            "learned_at": datetime.now().isoformat()
        }

        idx = len(self.dialogues)
        self.dialogues.append(dialogue)

        # Индексируем
        text = self._get_dialog_text(dialogue).lower()
        keywords = self._extract_keywords(text)

        for keyword in keywords:
            self.keyword_index[keyword].append(idx)

        logger.info(f"Новый паттерн добавлен в Unified RAG: {user_msg[:50]}...")

        return dialogue

_save_pattern_to_file() – сохраняет выученный паттерн в JSON-файл: загружает существующие паттерны, создает новый (обрезая длинные тексты), проверяет, что нет точного дубликата по вопросу, и записывает обновленный список обратно в файл для постоянного хранения знаний между перезапусками бота.

    # Сохраняет паттерн в файл
    def _save_pattern_to_file(self, user_msg: str, bot_msg: str):
        try:
            # Загружаем существующие паттерны
            try:
                with open(self.patterns_file, 'r', encoding='utf-8') as f:
                    patterns = json.load(f)
            except (FileNotFoundError, json.JSONDecodeError):
                patterns = []

            # Добавляем новый
            new_pattern = {
                'input': user_msg[:100],
                'response': bot_msg[:200],
                'learned_at': datetime.now().isoformat(),
                'usage_count': 0
            }

            # Проверяем на дубликаты
            if not any(p['input'] == new_pattern['input'] for p in patterns):
                patterns.append(new_pattern)

                # Сохраняем
                with open(self.patterns_file, 'w', encoding='utf-8') as f:
                    json.dump(patterns, f, ensure_ascii=False, indent=2)

                logger.info(f"Паттерн сохранен в файл: {user_msg[:50]}...")

        except Exception as e:
            logger.error(f"Ошибка сохранения паттерна: {e}")

increment_usage() – увеличивает счетчик использования паттерна при его повторном применении, чтобы отслеживать популярность ответов и синхронизирует с файлом.

_update_pattern_file() – находит соответствующий паттерн в JSON-файле и обновляет в нем счетчик использования, сохраняя актуальность данных на диске.

get_stats() – собирает статистику RAG-системы: общее количество диалогов, разделение на датасет и выученные паттерны, суммарное использование паттернов и размер поискового индекса для мониторинга эффективности.

# Увеличивает счетчик использования для паттерна
    def increment_usage(self, dialogue_idx: int):
        
        if dialogue_idx < len(self.dialogues):
            dialogue = self.dialogues[dialogue_idx]

            if dialogue.get('source') == 'pattern':
                dialogue['usage_count'] = dialogue.get('usage_count', 0) + 1
                logger.debug(f"Увеличено использование паттерна: {dialogue['usage_count']}")

                # Также обновляем в файле
                self._update_pattern_file(dialogue)

    # Обновляет счетчик использования в файле
    def _update_pattern_file(self, updated_dialogue: Dict):
        try:
            with open(self.patterns_file, 'r', encoding='utf-8') as f:
                patterns = json.load(f)

            # Находим и обновляем
            for pattern in patterns:
                if pattern['input'] == updated_dialogue['messages'][0]['content']:
                    pattern['usage_count'] = updated_dialogue.get('usage_count', 0)
                    break

            with open(self.patterns_file, 'w', encoding='utf-8') as f:
                json.dump(patterns, f, ensure_ascii=False, indent=2)

        except Exception as e:
            logger.error(f"Ошибка обновления файла паттернов: {e}")

    # Статистика Unified RAG
    def get_stats(self) -> Dict:
        
        pattern_count = sum(1 for d in self.dialogues if d.get('source') == 'pattern')
        dataset_count = len(self.dialogues) - pattern_count

        # Статистика использования паттернов
        pattern_usage = sum(
            d.get('usage_count', 0)
            for d in self.dialogues
            if d.get('source') == 'pattern'
        )

        return {
            'total_dialogues': len(self.dialogues),
            'from_dataset': dataset_count,
            'from_patterns': pattern_count,
            'pattern_usage_total': pattern_usage,
            'keywords_indexed': len(self.keyword_index)
        }

telegram_handlers.py

Итак, почти дошли до финала, этот файл отвечает за работу с телеграмом. Библиотеки:

from config import logger
from telegram import Update
from telegram.ext import ContextTypes, CommandHandler, MessageHandler, filters

from bot import main_bot

Создадим обработчик команды /start, сделаем так, чтобы бот генерировал ответ на команду.

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик /start"""
    chat_id = update.effective_chat.id
    user_id = update.effective_user.id

    await update.message.chat.send_action(action="typing")
    response = await main_bot.generate_response(chat_id, user_id, "", is_start=True)
    await update.message.reply_text(response)
    logger.info(f"/start от {user_id}")

Создадим хендлер сообщений для чатов:

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик сообщений в приватных чатах И обычных сообщений в группах"""
    try:
        if not update.message or not update.message.text:
            return

        user_msg = update.message.text

        # Пропуск команд
        if user_msg.startswith('/'):
            return

        chat_id = update.effective_chat.id
        user_id = update.effective_user.id

        # Установка информации о боте
        if not main_bot.bot_id and context.bot:
            main_bot.set_bot_info(context.bot.username, context.bot.id)

        # Для групп: проверяем, должен ли бот отвечать
        if update.effective_chat.type in ['group', 'supergroup']:
            if not main_bot.should_respond_in_group(update):
                logger.debug(f"Бот не должен отвечать в группе {chat_id}")
                return

        # Генерация ответа
        await update.message.chat.send_action(action="typing")
        response = await main_bot.generate_response(chat_id, user_id, user_msg)
        await update.message.reply_text(response)

        logger.debug(f"Ответ отправлен в чат {chat_id}")

    except Exception as e:
        logger.error(f"Ошибка обработки: {e}")

И сделаем итоговый обработчик:

def setup_handlers(application):
    """Настройка обработчиков"""

    # Для приватных чатов (личные сообщения боту)
    private_handler = MessageHandler(
        filters.TEXT & ~filters.COMMAND & filters.ChatType.PRIVATE,
        handle_message
    )

    # Для групповых чатов - ВСЕ сообщения, но логика внутри handle_group_message
    # решит, это пост канала или обычное сообщение
    group_handler = MessageHandler(
        (filters.TEXT | filters.CAPTION | filters.PHOTO) & ~filters.COMMAND &
        (filters.ChatType.GROUP | filters.ChatType.SUPERGROUP),
        handle_group_message
    )

    # Команды работают везде
    start_handler = CommandHandler("start", start_command)

    handlers = [
        start_handler,
        private_handler,
        group_handler,
    ]

    for handler in handlers:
        application.add_handler(handler)

    logger.info("Обработчики настроены: приватные чаты + группы")

main.py

Мы добрались до конца. Этот файл запустит нашего бота. У него есть только токен подключения, который вам нужно получить у отца ботов. Также он проверит наличие GPU и инициализирует бота, после чего запустит его.

import torch
from telegram_handlers import setup_handlers
from telegram.ext import Application

from bot import main_bot


def main():
    """Точка входа"""
    BOT_TOKEN = "СЮДА СВОЙ ТОКЕН ОТ ОТЦА БОТОВ"

    print("=" * 50)
    print("ПРОВЕРКА GPU:")
    print(f"Доступен CUDA: {torch.cuda.is_available()}")
    if torch.cuda.is_available():
        print(f"GPU устройство: {torch.cuda.get_device_name(0)}")
        print(f"Кол-во GPU: {torch.cuda.device_count()}")
        print(f"Память GPU: {torch.cuda.get_device_properties(0).total_memory / 1024 ** 3:.1f} GB")
    else:
        print("GPU не найден! Проверь установку CUDA и PyTorch")
    print("=" * 50)

    print("Инициализация...")

    # Инициализация
    main_bot.initialize_model()

    # Статистика
    print(f"Диалогов в RAG: {len(main_bot.rag.dialogues)}")
    print(f"Пользователей: {len(main_bot.user_memory.users)}")

    # Запуск бота
    application = Application.builder().token(BOT_TOKEN).build()
    setup_handlers(application)

    print("Бот запущен!")
    print("=" * 40)
    print("Доступные команды:")
    print("/start - начать диалог")
    print("=" * 40)

    application.run_polling()


if __name__ == "__main__":
    main()

Запускаем main.py и все должно работать. Проделана гигантская работа, мы все большие молодцы. Исходный код можно посмотреть вот тут. За кадром я сделал так, чтобы бот смог комментировать посты в телеграм канале.

А что потом?

В итоге мы имеем работающего чат-бота, который через qwen генерирует нам ответы, и несколько функций сверху. Из плюсов: самое то для прототипа, почти никаких зависимостей от модели, кошка, жена и миска риса.

Иногда всплывают иероглифы, но этим же болеет сама qwen.

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

С новым MVP уже можно ознакомиться у меня в тк канале. А также оставлю список используемой литературы как дополнении. Там очень много полезного и интересного.