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

А зачем?
Вопрос неочевидный. Ведь LLM не является путём к созданию AGI:
LLM — это, по сути, высокоточная имитация человеческой речи без подлинного понимания смысла;
Модели блестяще справляются с генерацией текстов, решением задач и написанием кода, но делают это не через логику и абстрактное мышление, а через статистический анализ и математические закономерности;
Их интеллект экстраполяция паттернов из обучающих данных, а не осмысленное познание.
Изучая архитектурные подходы к реализации AGI, я пришел к выводу, что нужен прототип. Поэтому в этой статье обсудим разработку MVP на базе LLM-модели, а выбор архитектуры отложим для следующей статьи.
Стоп, что?
Наша цель создать чат-бота с конкретной личностью и с каким-никаким запоминанием контекста для телеграмм.
Сначала мой выбор упал на реализацию LLM с нуля, но от этой идеи пришлось быстро отказаться. Создание собственной LLM слишком дорогое, долгое и не имеет никакого смысла.
Далее выбор пал на дообучение. Так называемый fine-tuning.
Fine-tuning — это когда взяли часть уже существующей модели и обучили часть под свои задачи. Такая часть называется адаптер.
У меня было несколько попыток дообучения, и обе не увенчались каким-либо большим успехом. Бот генерировал мусор.
Основная проблема заключалась в датасете. Мусор на входе, мусор на выходе. Даже создание собственного датасета на 3 тысячи диалогов не особо помогло. Этого оказалось мало. Можно стоило, конечно, потратить еще полгода и добить 10 тысяч диалогов, но это потеряло смысл.
Оказывается, технология дообучения — это долго, дорого и вообще непонятно, как это всё работает. Поэтому от нее после пары неудачных попыток пришлось отказаться.
Тогда я узнал про RAG, и тут уже намного интереснее.
RAG (Retrieval-Augmented Generation) – это система подключения к LLM, которая по своей сути напоминает поисковик. Она ищет в базе данных или датасете информацию, максимально похожую на промпт, и отправляет его в генерирующую часть. Там LLM смотрит и генерирует уже ответ.
Очень хорошие статьи про RAG есть от этого автора в двух частях. Оттуда же картинка.

Данная часть статьи не является гайдом как таковым, это про то, что у меня получилось. Я буду очень рад послушать критику и возможные улучшения. Всем спасибо <3.
Перед тем как начать, обращу внимание на то, что у меня 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 - это датасет.
Тут руки полностью развязаны:
Можно использовать уже готовые, такие как RuTurboAlpaca;
Можно сделать свой, как поступил я.
Самое важное, должна быть такая структура:
{
"messages":
[
{
"role": "user", "content": "Привет! Как тебя зовут?"},
{
"role": "assistant", "content": "Привет! Я Гриша, чат-бот. А как тебя зовут?"
}
]
}Это для того, чтобы наш чат-бот понимал, где пользователи, а где его текст для обучения.
Дальше создадим файл 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 и импортируем библиотеки. Все библиотеки будут подчеркиваться из-за того, что мы их еще не заполнили и файлы пустые. Так и должно быть. Когда мы закончим проект, они исчезнут.
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:
Ищет похожие диалоги в датасете
Берет ответ ассистента из найденного диалога
Добавляет, как пример ответа
Новая функция для очистки ответа 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(), которое координирует все модули бота.
prompt - сформированный текст промпта
return_tensors="pt" - возвращать как PyTorch тензоры
truncation=True - обрезать если длиннее max_length
max_length=2048 - максимальная длина в токенах
Переносит данные на то же устройство, где модель:
Если модель на GPU, то и тензоры на GPU
Если модель на CPU, то и тензоры на CPU
**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()Этот файл отвечает за память. Импорт библиотек:
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)
)Сделаем основные методы для работы с памятью диалогов.
get_last_messages() - получить последние N сообщений;
add_message() - добавить сообщение в историю;
get_history() - получить всю историю чата;
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}")В этом файле мы реализуем систему самообучения на паттернах. Бот запоминает, бот учиться. Библиотеки:
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 системы. Импорт:
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 Falseadd_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)
}Итак, почти дошли до финала, этот файл отвечает за работу с телеграмом. Библиотеки:
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("Обработчики настроены: приватные чаты + группы")Мы добрались до конца. Этот файл запустит нашего бота. У него есть только токен подключения, который вам нужно получить у отца ботов. Также он проверит наличие 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 уже можно ознакомиться у меня в тк канале. А также оставлю список используемой литературы как дополнении. Там очень много полезного и интересного.