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

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

А зачем?

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

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

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

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

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

Стоп, что?

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

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

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

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

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

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

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

Очень хорошие статьи про 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|>system\n{SYSTEM_PROMPT}\n<|im_end|>\n"] # Контекст времени и даты current_time = datetime.now() prompt_parts.append(f"<|im_start|>context\nСейчас: {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|>context\nТекущий пользователь: {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|>history\nИстория диалога:\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|>examples\nПример ответа:\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|>user\n{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|>instruction\nОтвечая, обязательно используй имя пользователя: {user_name}\n<|im_end|>\n") else: prompt_parts.append( "<|im_start|>instruction\nЕсли не знаешь имя пользователя, спроси его или признайся, что не помнишь.\n<|im_end|>\n") # Маркер для ответа prompt_parts.append("<|im_start|>assistant\n") 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'(\b\w+\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 уже можно ознакомиться у меня в тк канале. А также оставлю список используемой литературы как дополнении. Там очень много полезного и интересного.