Друзья, приветствую вас в очередной статье. Сегодня я расскажу, как использовать LLAMA3 ИИ в своих проектах. После небольшой подготовки мы приступим к созданию полноценного Telegram бота.

Сегодня мы:

• Научимся устанавливать LLama3 на локальную машину.

• Научимся бесплатно запускать LLama3 через платформу GROQ.

• Разберемся с преимуществами и недостатками первого и второго способа развертывания LLama3.

• Напишем полноценного Telegram бота с использованием aiogram3, который сможет работать как с локальной версией LLAMA3, так и через сервис GROQ (технически он сможет работать с любой подключенной нейросетью).

• Запустим Telegram бота на VPS сервере (опционально).

Подготовка

Для того чтобы следовать этому руководству, вам потребуется:

• VPS сервер (важно, чтобы с европейским или США API).

• База данных PostgreSQL (о том, как запустить её за пару минут, я писал [ТУТ]).

• Базовые знания по написанию ботов через aiogram3 (в моём профиле на Хабре вы найдёте множество публикаций, в которых подробно рассмотрена тема создания Telegram ботов).

• Установленный Docker на локальной машине и на VPS сервере (если вы новичок, установите Docker Desktop).

Надеюсь, что подготовку вы уже выполнили.

Развертывание локальной версии нейросети LLAMA с использованием Docker

Откройте консоль и выполните следующую команду:

docker run -d -v ollama:/root/.ollama -p 11434:11434 --name ollama ollama/ollama

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

Для запуска самой модели выполните команду:

docker exec -it ollama ollama run llama3:8b

Эта команда загрузит и запустит языковую модель LLAMA3:8b (4.7GB). Доступна также более крупная версия LLama3, 70b (40GB). Вы можете запускать и другие модели, список которых доступен [здесь].

Чтобы запустить другую модель, используйте команду:

docker exec -it ollama ollama run model_name:tag

Для выхода из диалога с LLM отправьте команду /bye, затем последовательно выполните CTRL+P и CTRL+Q, чтобы выйти из интерактивного режима.

Обратите внимание. Так вы просто свернете контейнер, но он все равно будет запущен. Если вы хотите удалить контейнер, то воспользуйтесь командами:

docker stop ollama
docker rm ollama
docker rmi ollama/ollama

Обратите внимание. После остановки (удаления) контейнера ollama локальное взаимодействие с ним будет невозможным!

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

Использование GROQ для взаимодействия с LLAMA3

Альтернативой локальному запуску LLAMA3 является использование сервиса GROQ. Для этого:

1. Зайдите в консоль GROQ.

2. Зарегистрируйтесь и выполните авторизацию.

3. Перейдите в раздел KEYS.

4. Нажмите на «Create API Key».

5. Скопируйте созданный ключ и сохраните его в надёжном месте.

Этих простых действий будет достаточно для дальнейшего использования GROQ в качестве посредника между вашим Python проектом и LLama3.

Преимущества и недостатки использования LLama3 в локальном виде или через GROQ

Локальный запуск LLAMA3

Преимущества:

1. Контроль над данными и конфиденциальность:

   • Полный контроль над данными, так как они не передаются через сторонние серверы.

2. Настраиваемость:

   • Возможность полной настройки системы и оптимизации под специфические задачи.

3. Единовременные затраты:

   • Высокие начальные затраты на оборудование, но отсутствие постоянных расходов на аренду облачных ресурсов.

4. Отсутствие зависимости от интернета:

   • Модель может работать автономно.

Недостатки:

1. Высокие начальные затраты:

   • Значительные затраты на покупку и установку оборудования.

2. Техническое обслуживание:

   • Необходимость регулярного обслуживания и обновления оборудования.

3. Ограниченная масштабируемость:

   • Масштабирование требует физического обновления или добавления новых машин.

Использование платформы GROQ

Преимущества:

1. Масштабируемость:

   • Легкость масштабирования вычислительных ресурсов без необходимости физического обновления оборудования.

2. Обновления и поддержка:

   • Доступ к последним обновлениям и поддержке от команды GROQ.

3. Экономия времени:

   • Быстрое развертывание и настройка.

4. Гибкость в оплате:

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

Недостатки:

1. Возможные расходы:

   • В будущем платформа станет платной.

2. Зависимость от интернета:

   • Требуется постоянное интернет-соединение.

3. Меньший контроль над данными:

   • Данные передаются через сторонние серверы.

4. Ограниченния платформы (текущие и будущие):

   • Сама платформа уже сейчас устанавливает лимиты и ограничения. К примеру есть лимиты по запросам к модели (скрин ниже).

   • Ещё пример - на данный момент нельзя пользоваться данной платформой с РФ IP адресов. При чем, вы не только не сможете без VPN зайти на сайт, но ещё и VPN должен стоять на вашем компьютере (сервере, если IP РФ), если вы в РФ и если вы будете использовать GROQ в своих проектах.

Выбор между локальным запуском LLAMA и использованием платформы GROQ зависит от ваших конкретных нужд и приоритетов. Для полного контроля и автономности лучше подходит локальный запуск, несмотря на высокие начальные затраты и необходимость обслуживания. Для гибкости, быстрого масштабирования и меньших начальных затрат оптимальным будет использование платформы GROQ, несмотря на постоянные расходы и зависимость от интернет-соединения.

Теперь давайте рассмотрим программное использование LLama3 с локальным запуском и запуском через платформу GROQ.

Я подготовил две демо-версии, изучив которые, вы поймёте, как происходит программное взаимодействие с LLAMA3 через Python.

Демо локального использования

from openai import OpenAI

client = OpenAI(
    base_url='http://localhost:11434/v1',
    api_key='ollama',
)

dialog_history = []

while True:
    user_input = input("Введите ваше сообщение ('stop' для завершения): ")

    if user_input.lower() == "stop":
        break

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

    response = client.chat.completions.create(
        model="llama3:8b",
        messages=dialog_history,
    )

    # Извлекаем содержимое ответа
    response_content = response.choices[0].message.content
    print("Ответ модели:", response_content)

    # Добавляем ответ модели в историю диалога
    dialog_history.append({
        "role": "assistant",
        "content": response_content,
    })

Давайте разберемся с этим кодом.

Импорт и настройка клиента

from openai import OpenAI

client = OpenAI(
    base_url='http://localhost:11434/v1',
    api_key='ollama',
)

Здесь мы импортируем библиотеку OpenAI и создаем экземпляр клиента OpenAI, указывая базовый URL для API (локальный сервер на порту 11434) и API-ключ (в данном случае 'ollama').

Такую возможность мы получаем после того как локально развернули Docker контейнер Ollama. Далее, выполняя запросы о которых поговорим далее, мы имитируем диалог с ботом через консоль (описывал выше).

Инициализация истории диалога

dialog_history = []

Создаем пустой список dialog_history, который будет хранить историю сообщений диалога между пользователем и моделью. Такой-же, но более продвинутый формат хранения истории мы будем использовать в нашем боте.

Основной цикл для взаимодействия с пользователем

while True:
    user_input = input("Введите ваше сообщение ('stop' для завершения): ")

    if user_input.lower() == "stop":
        break

Запускаем бесконечный цикл, который будет принимать ввод от пользователя. Если пользователь введет 'stop', цикл прерывается.

Добавление сообщения пользователя в историю диалога

dialog_history.append({
        "role": "user",
        "content": user_input,
    })

Сообщение пользователя добавляется в dialog_history в виде словаря с ключами role (роль 'user') и content (содержимое сообщения).

Отправка запроса модели и получение ответа

response = client.chat.completions.create(
        model="llama3:8b",
        messages=dialog_history,
    )

Отправляется запрос к модели llama3:8b (если разворачивали контейнер с Llama3, то на вашем компьютере уже установлена модель llama3:8b, иначе загрузите модель отдельно) с текущей историей диалога. Метод client.chat.completions.create создает завершение чата на основе переданных сообщений.

Обработка ответа модели

response_content = response.choices[0].message.content
print("Ответ модели:", response_content)

Ответ модели извлекается из объекта response и выводится на экран.

Добавление ответа модели в историю диалога

dialog_history.append({
        "role": "assistant",
        "content": response_content,
    })

Ответ модели добавляется в dialog_history как сообщение от ассистента.

Обратите внимание. В данном примере я использовал модуль openai, а это значит, что данный клиент будет работать в любом боте, который использовал нейронки ChatGPT, Bing и так далее (сейчас GitHub завален проектами, использующими библиотеку openai).

Демо клиента через GROQ

from groq import Groq
from decouple import config

client = Groq(
    api_key=config("GROQ_API_KEY"),
)

dialog_history = []

while True:
    user_input = input("Введите ваше сообщение ('stop' для завершения): ")

    if user_input.lower() == "stop":
        break

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

    models = ["gemma-7b-it", "llama3-70b-8192", "llama3-8b-8192", "mixtral-8x7b-32768"]
    chat_completion = client.chat.completions.create(
        messages=dialog_history,
        model=models[1],D
    )

    response = chat_completion.choices[0].message.content
    print("Ответ модели:", response)

    # Добавляем ответ модели в историю диалога
    dialog_history.append({
        "role": "assistant",
        "content": response,
    })

Особо внимательные могут заметить, что синтаксис модуля GROQ не особо отличается от синтаксиса OPENAI и это совсем не случайно. Ведь, если мы посмотрим под капот модуля groq, то увидим что основывается он на openai.

А это, как вы поняли, нам на руку.

Импорт и настрйка клиента

from groq import Groq
from decouple import config

client = Groq(
    api_key=config("GROQ_API_KEY"),
)

Я использовал модуль python-decouple, чтоб импортировать из файла .env GROQ_API_KEY.

Доступные модели:

models = ["gemma-7b-it", "llama3-70b-8192", "llama3-8b-8192", "mixtral-8x7b-32768"]

На данный момент GROQ поддерживает представленные выше модели. В коде я взял за основу "llama3-70b-8192". Это там самая модель, которая весит более 40GB и самое приятное тут то, что скачивать модель нам не нужно, а достаточно получить у GROQ api key и установить модуль groq.

В остальном отличий от работы со своей локальной моделью через openai нет.

Надеюсь, что данные примеры вам понятны. Переходим к боту

Создаем бота

Сегодня мы создадим бота-ассистента, который будет иметь следующие основные функции:

1. Добавление пользователя в базу данных по клику на "Старт" и присваивание ему статуса "Не в диалоге" / "В диалоге"

2. Запуск диалога по клику на кнопку «Начать диалог».

3. Полное удаление истории общения по клику на кнопку «Очистить историю».

Каждый пользователь может находиться в одном из двух состояний: «В диалоге» или «Не в диалоге». Это позволит в будущем расширить функционал бота, добавив интерактивное меню с информацией о нас, выбором моделей и другими опциями.

Логика сохранения истории

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

• id: уникальный идентификатор диалога (автоинкрементируемый номер).

• user_id: Telegram ID пользователя.

• message: JSON с сообщением бота или пользователя.

Такой подход позволит нам сохранять историю диалога для каждого пользователя по его Telegram ID, а также обеспечить возможность очистки истории.

Хранение информации о пользователе

Для хранения информации о пользователе создадим таблицу users. В этой таблице будут следующие поля:

• user_id: уникальный идентификатор пользователя в Telegram.

• user_login: имя пользователя в Telegram.

• full_name: полное имя пользователя.

• in_dialog: текущее состояние пользователя («В диалоге» (True) или «Не в диалоге» (False)).

• date_reg: дата и время регистрации пользователя.

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

Подготовка перед написанием кода

Начну с того, что весь код бота можно найти в моем публичном репозитории Easy_LLama3_Bot. Там вы найдете бота, демки для работы с локальной и GROQ версией Llama3 и пример настройки Dockerfile для быстрого запуска на VPS сервере.

Зависимости (requirementsl.txt)

asyncpg-lite~=0.3.1.3
aiogram~=3.7.0
python-decouple
groq
pytz
openai

• asyncpg-lite - библиотека для асинхронной работы с PostgreSQL (делал подробное описание Asynpg-lite: лёгкость асинхронных операций на PostgreSQL с SQLAlchemy)

• aiogram3 - фреймворк для создания телеграмм ботов через Python (в моём профиле на Хабре вы найдёте множество публикаций, в которых подробно рассмотрена тема создания Telegram ботов на aiogram3)

• python-decouple - модуль для удобной работы с .env

• groq и openai описывал выше. Выберите для своего бота один из вариантов.

• pytz - простой модуль для работы с часовыми поясами

ENV-файл (.env)

GROQ_API_KEY=your_groq_token
BOT_API_KEY=your_bot_token
ADMINS=admin1,admin2
ROOT_PASS=your_root_password
PG_LINK=postgresql://username:password@host:port/dbname

Замените данные на свои. Предварительно не забудьте узнать свой телеграмм ID, развернуть базу данных и создать токен бота и токен GROQ_API. О том как создавать токен телеграмм бота через BotFather вы можете узнать тут или, в целом, на просторах интернета.

Структура проекта

- db_handler
    - __init__.py: Инициализация модуля.
    - db_funk.py: Функции для взаимодействия с PostgreSQL.

- handlers
    - __init__.py: Инициализация модуля.
    - user_router.py: Основной и единственный роутер в котором весь код

- keyboards
    - __init__.py: Инициализация модуля.
    - kbs.py: Файл со всеми клавиатурами.

- utils
    - __init__.py: Инициализация модуля.
    - utils.py: Файл с утилитами.

- .env
- .dockerignorefile
- Dockerfile
- Makefile: файл для удобного запуска и управления контейнерами
- .gitignorefile
- aiogram_run.py: файл для запуска бота
- create_bot.py: файл с настройками бота
- llama_groq_demo.py: демка LLama3с работой через GROQ
- llama_localhost_demo.py: демка LLama3с с локальным запуском
- README.md: Короткое описание проекта с GitHub

Файл с настройками бота (create_bot.py):

import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from asyncpg_lite import DatabaseManager
from decouple import config
from groq import Groq
from openai import OpenAI

client_groq = Groq(api_key=config("GROQ_API_KEY"))
local_client = OpenAI(base_url='http://localhost:11434/v1', api_key='ollama')

# получаем список администраторов из .env
admins = [int(admin_id) for admin_id in config('ADMINS').split(',')]

# инициализируем логирование и выводим в переменную для отдельного использования в нужных местах
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# инициализируем объект, который будет отвечать за взаимодействие с базой данных
db_manager = DatabaseManager(db_url=config('PG_LINK'), deletion_password=config('ROOT_PASS'))

# инициализируем объект бота, передавая ему parse_mode=ParseMode.HTML по умолчанию
bot = Bot(token=config('BOT_API_KEY'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))

# инициализируем объект бота
dp = Dispatcher()

Если вы читали мои статьи по Aiogram3 то вам должно быть все понятно. Единственное на что хочу обратить внимание - это настройка клиентов для работы с нейронками.

Вам необходимо выбрать один вариант или GROQ или локальную версию. В коде вывел два для демонстрации.

Файл для запуска бота (aiogram_run.py):

import asyncio
from create_bot import bot, dp, admins
from db_handler.db_funk import get_all_users
from handlers.user_router import user_router
from aiogram.types import BotCommand, BotCommandScopeDefault


# Функция, которая настроит командное меню (дефолтное для всех пользователей)
async def set_commands():
    commands = [BotCommand(command='start', description='Старт'),
                BotCommand(command='restart', description='Очистить диалог')]
    await bot.set_my_commands(commands, BotCommandScopeDefault())


# Функция, которая выполнится когда бот запустится
async def start_bot():
    await set_commands()
    count_users = await get_all_users(count=True)
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, f'Я запущен?. Сейчас в базе данных <b>{count_users}</b> пользователей.')
    except:
        pass


# Функция, которая выполнится когда бот завершит свою работу
async def stop_bot():
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, 'Бот остановлен. За что??')
    except:
        pass


async def main():
    # регистрация роутеров
    dp.include_router(user_router)

    # регистрация функций
    dp.startup.register(start_bot)
    dp.shutdown.register(stop_bot)

    # запуск бота в режиме long polling при запуске бот очищает все обновления, которые были за его моменты бездействия
    try:
        await bot.delete_webhook(drop_pending_updates=True)
        await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
    finally:
        await bot.session.close()


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

Тут вы видите базовую структуру файла для запуска бота. Такую я использую в каждом своем проекте.

Если коротко, то тут мы описали функции, которые выполняются при запуске бота и при завершении работы.

Зарегистрировали командное меню и подключили единственный роутер, который будет в данном проекте. Для более глубокого понимания темы - читайте мои прошлые статьи.

Напишем хендлер для работы с базой данных (db_handler/db_funk.py):

import json
from sqlalchemy import Integer, String, BigInteger, TIMESTAMP, JSON, Boolean
from create_bot import db_manager
import asyncio


# функция, которая создаст таблицу с пользователями
async def create_table_users(table_name='users'):
    async with db_manager as client:
        columns = [
            {"name": "user_id", "type": BigInteger, "options": {"primary_key": True, "autoincrement": False}},
            {"name": "full_name", "type": String},
            {"name": "user_login", "type": String},
            {"name": "in_dialog", "type": Boolean},
            {"name": "date_reg", "type": TIMESTAMP},
        ]
        await client.create_table(table_name=table_name, columns=columns)


async def create_table_dialog_history(table_name='dialog_history'):
    async with db_manager as client:
        columns = [
            {"name": "id", "type": Integer, "options": {"primary_key": True, "autoincrement": True}},
            {"name": "user_id", "type": BigInteger},
            {"name": "message", "type": JSON}
        ]
        await client.create_table(table_name=table_name, columns=columns)


# функция, для получения информации по конкретному пользователю
async def get_user_data(user_id: int, table_name='users'):
    async with db_manager as client:
        user_data = await client.select_data(table_name=table_name, where_dict={'user_id': user_id}, one_dict=True)
    return user_data


# функция, для получения всех пользователей (для админки)
async def get_all_users(table_name='users', count=False):
    async with db_manager as client:
        all_users = await client.select_data(table_name=table_name)
    if count:
        return len(all_users)
    else:
        return all_users


# функция, для добавления пользователя в базу данных
async def insert_user(user_data: dict, table_name='users', conflict_column='user_id'):
    async with db_manager as client:
        await client.insert_data_with_update(table_name=table_name,
                                             records_data=user_data,
                                             conflict_column=conflict_column,
                                             update_on_conflict=False)


async def get_dialog_history(db_client, user_id: int, table_name='dialog_history'):
    dialog_history_msg = []
    dialog_history = await db_client.select_data(table_name=table_name, where_dict={'user_id': user_id})
    for msg in dialog_history:
        message = json.loads(msg.get('message'))
        dialog_history_msg.append(message)
    return dialog_history_msg


async def add_message_to_dialog_history(user_id: int, message: dict, table_name='dialog_history', return_history=False):
    async with db_manager as client:
        await client.insert_data_with_update(table_name=table_name,
                                             records_data={'user_id': user_id,
                                                           'message': json.dumps(message)},
                                             conflict_column='id')
        if return_history:
            dialog_history = await get_dialog_history(client, user_id)
            return dialog_history


async def update_dialog_status(client, user_id: int, status: bool, table_name='users'):
    await client.update_data(table_name=table_name,
                             where_dict={'user_id': user_id},
                             update_dict={'in_dialog': status})


async def clear_dialog(user_id: int, dialog_status: bool, table_name='dialog_history'):
    async with db_manager as client:
        await client.delete_data(table_name=table_name, where_dict={'user_id': user_id})
        await update_dialog_status(client, user_id, dialog_status)


async def get_dialog_status(user_id: int, table_name='users'):
    async with db_manager as client:
        user_data = await client.select_data(table_name=table_name, where_dict={'user_id': user_id}, one_dict=True)
    return user_data.get('in_dialog')

Да, тут нужно остановиться подробнее.

Для понимания этого кода вам нужно ознакомиться с синтаксисом Asyncpg-lite. Надеюсь, что вы это сделали.

Как я описывал выше, для работы с базой данных PostgreSQL нам необходимы 2 таблицы: users и dialog_history.

# функция, которая создаст таблицу с пользователями
async def create_table_users(table_name='users'):
    async with db_manager as client:
        columns = [
            {"name": "user_id", "type": BigInteger, "options": {"primary_key": True, "autoincrement": False}},
            {"name": "full_name", "type": String},
            {"name": "user_login", "type": String},
            {"name": "in_dialog", "type": Boolean},
            {"name": "date_reg", "type": TIMESTAMP},
        ]
        await client.create_table(table_name=table_name, columns=columns)


async def create_table_dialog_history(table_name='dialog_history'):
    async with db_manager as client:
        columns = [
            {"name": "id", "type": Integer, "options": {"primary_key": True, "autoincrement": True}},
            {"name": "user_id", "type": BigInteger},
            {"name": "message", "type": JSON}
        ]
        await client.create_table(table_name=table_name, columns=columns)

Благодаря этим 2 простым функциям мы эти таблицы и создадим. Выполнять код можно прямо в файле db_funk.py (для этого оставил импорт asyncio) и в конце файла пример вызова.

Функция для получения информации о пользователе:

# функция, для получения информации по конкретному пользователю
async def get_user_data(user_id: int, table_name='users'):
    async with db_manager as client:
        user_data = await client.select_data(table_name=table_name, where_dict={'user_id': user_id}, one_dict=True)
    return user_data

Данную функцию можно будет использовать под следующие задачи:

• Проверка на наличие пользователя в базе данных (если пользователя не будет в БД, то функция вернет пустой список)

• Отображение информации о пользователе в личном профиле (в данном проекте не применяется, как сделать личный профиль при помощи данной функции я писал в этой статье «Telegram Боты на Aiogram 3.x: Профиль, админ-панель и реферальная система»)

Получаем информацию о всех пользователях

async def get_all_users(table_name='users', count=False):
    async with db_manager as client:
        all_users = await client.select_data(table_name=table_name)
    if count:
        return len(all_users)
    else:
        return all_users

Отображение информации о всех пользователях в админ-панеле. В данном проекте функция используется для вывода количества пользователей в базе данных при запуске бота.

Пример с просмотром пользователей в админке при помощи этой функции и ее полное описание найдете в этой статье «Telegram Боты на Aiogram 3.x: Профиль, админ-панель и реферальная система».

Функция для добавления пользователя:

async def insert_user(user_data: dict, table_name='users', conflict_column='user_id'):
    async with db_manager as client:
        await client.insert_data_with_update(table_name=table_name,
                                             records_data=user_data,
                                             conflict_column=conflict_column,
                                             update_on_conflict=False)

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

Функция для получения истории диалога пользователя:

async def get_dialog_history(db_client, user_id: int, table_name='dialog_history'):
    dialog_history_msg = []
    dialog_history = await db_client.select_data(table_name=table_name, where_dict={'user_id': user_id})
    for msg in dialog_history:
        message = json.loads(msg.get('message'))
        dialog_history_msg.append(message)
    return dialog_history_msg

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

Функция для добавления записи в таблицу dialog_history

async def add_message_to_dialog_history(user_id: int, message: dict, table_name='dialog_history', return_history=False):
    async with db_manager as client:
        await client.insert_data_with_update(table_name=table_name,
                                             records_data={'user_id': user_id,
                                                           'message': json.dumps(message)},
                                             conflict_column='id')
        if return_history:
            dialog_history = await get_dialog_history(client, user_id)
            return dialog_history

Обратите внимание. Данная функция, при передаче параметра return_history=True будет возвращать весь диалог пользователя в виде списка питоновских словарей. Такое решение принял для оптимизации обращений к базе данных. Далее, когда мы начнем рассматривать пример кода бота, вам станет этот момент более понятным.

Обновление статуса диалога (в диалоге или нет) для каждого пользователя:

async def update_dialog_status(client, user_id: int, status: bool, table_name='users'):
    await client.update_data(table_name=table_name,
                             where_dict={'user_id': user_id},
                             update_dict={'in_dialog': status})

Для работы функции достаточно передать user_id и новый статус (True или False) и в таблице произойдет обновление в колонке in_dialog. Вокруг этого статуса мы, в дальнейшем, будем строить логику проверок и вывод функционала.

Функция для полной очистки истории диалогов пользователя:

async def clear_dialog(user_id: int, dialog_status: bool, table_name='dialog_history'):
    async with db_manager as client:
        await client.delete_data(table_name=table_name, where_dict={'user_id': user_id})
        await update_dialog_status(client, user_id, dialog_status)

Функция для получения статуса диалога:

async def get_dialog_status(user_id: int, table_name='users'):
    async with db_manager as client:
        user_data = await client.select_data(table_name=table_name, where_dict={'user_id': user_id}, one_dict=True)
    return user_data.get('in_dialog')

В целом, обратите внимание на то, как перекликаются функции в файлу db_funk.py. Желательно, чтоб вы разобрались с данной логикой. Так как понимание этих моментов позволят вам, в целом, понять как работает бот.

На данный момент:

• Мы разобрались как подключить Llama3 к своему проекту (локально и через GROQ)

• Посмотрели как работает програмное взаимодействие с Llama3 (через две демки)

• Настроили структуру бота

• Подготовили базу данных и функции для взаимодействия с ней

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

Файл handlers/user_router.py:

from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message
from create_bot import bot, client_groq, local_client
from db_handler.db_funk import (get_user_data, insert_user, clear_dialog,
                                add_message_to_dialog_history, get_dialog_status)
from keyboards.kbs import start_kb, stop_speak
from utils.utils import get_now_time
from aiogram.utils.chat_action import ChatActionSender

user_router = Router()


# хендлер команды старт
@user_router.message(Command(commands=['start', 'restart']))
async def cmd_start(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        user_info = await get_user_data(user_id=message.from_user.id)

    if len(user_info) == 0:
        await insert_user(user_data={
            'user_id': message.from_user.id,
            'full_name': message.from_user.full_name,
            'user_login': message.from_user.username,
            'in_dialog': False,
            'date_reg': get_now_time()
        })
        await message.answer(text='Привет! Давай начнем общаться. Для этого просто нажми на кнопку "Начать диалог"',
                             reply_markup=start_kb())
    else:
        await clear_dialog(user_id=message.from_user.id, dialog_status=False)
        await message.answer(text='Диалог очищен. Начнем общаться?', reply_markup=start_kb())


# Хендлер для начала диалога
@user_router.message(F.text.lower().contains('начать диалог'))
async def start_speak(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        await clear_dialog(user_id=message.from_user.id, dialog_status=True)
        await message.answer(text='Диалог начат. Введите ваше сообщение:', reply_markup=stop_speak())


@user_router.message(F.text.lower().contains('завершить диалог'))
async def start_speak(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        await clear_dialog(user_id=message.from_user.id, dialog_status=False)
        await message.answer(text='Диалог очищен! Начнем общаться?', reply_markup=start_kb())


# Хендлер для обработки текстовых сообщений
@user_router.message(F.text)
async def handle_message(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        check_open = await get_dialog_status(message.from_user.id)
        if check_open is False:
            await message.answer(text='Для того чтоб начать общение со мной, пожалуйста, нажмите на кнопку '
                                      '"Начать диалог".', reply_markup=start_kb())
            return
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        # формируем словарь с сообщением пользователя
        user_msg_dict = {"role": "user", "content": message.text}

        # сохраняем сообщение в базу данных и получаем историю диалога
        dialog_history = await add_message_to_dialog_history(user_id=message.from_user.id,
                                                             message=user_msg_dict,
                                                             return_history=True)

        #Пример работы с GROQ
        chat_completion = client_groq.chat.completions.create(model="llama3-70b-8192", messages=dialog_history)
        message_llama = await message.answer(text=chat_completion.choices[0].message.content, reply_markup=stop_speak())

        '''
        # Пример работы с локальной моделью
        
        chat_completion = local_client.chat.completions.create(model="llama3:8b", messages=dialog_history)
        message_llama = await message.answer(text=chat_completion.choices[0].message.content, reply_markup=stop_speak())
        '''

    # формируем словарь с сообщением ассистента
    assistant_msg = {"role": "assistant", "content": message_llama.text}

    # сохраняем сообщение ассистента в базу данных
    await add_message_to_dialog_history(user_id=message.from_user.id, message=assistant_msg,
                                        return_history=False)

Импорты:

from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message
from create_bot import bot, client_groq, local_client
from db_handler.db_funk import (get_user_data, insert_user, clear_dialog,
                                add_message_to_dialog_history, get_dialog_status)
from keyboards.kbs import start_kb, stop_speak
from utils.utils import get_now_time
from aiogram.utils.chat_action import ChatActionSender

Из того на что стоит обратить внимание - это импорты клавиатур (далее покажу вам код) и импорты клиентов Llama3.

• Если вы хотите использовать локальную версию Llama3, то вам достаточно импортировать local_client.

• Для использования клиента через платформу GROQ достаточно импортировать client_groq.

Клавиатуры:

from aiogram.types import KeyboardButton, ReplyKeyboardMarkup


def start_kb():
    kb_list = [[KeyboardButton(text="▶️ Начать диалог")]]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Чтоб начать диалог с ботом жмите ?:"
    )


def stop_speak():
    kb_list = [[KeyboardButton(text="❌ Завершить диалог")]]
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Чтоб завершить диалог с ботом жмите ?:"
    )

Как вы видите в боте будет всего 2 текстовые клавиатуры (можно было объеденить и в одну функцию, но я решил что лучше передать их явно. Подробно тему текстовых клавиатур я рассматривал в статье Telegram Боты на Aiogram 3.x: Текстовая клавиатура и Командное меню.

Создаем роутер. Он у нас в проекте будет единственным:

user_router = Router()

Теперь напишем обработчик функции "start" и "restart":

@user_router.message(Command(commands=['start', 'restart']))
async def cmd_start(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        user_info = await get_user_data(user_id=message.from_user.id)

    if len(user_info) == 0:
        await insert_user(user_data={
            'user_id': message.from_user.id,
            'full_name': message.from_user.full_name,
            'user_login': message.from_user.username,
            'in_dialog': False,
            'date_reg': get_now_time()
        })
        await message.answer(text='Привет! Давай начнем общаться. Для этого просто нажми на кнопку "Начать диалог"',
                             reply_markup=start_kb())
    else:
        await clear_dialog(user_id=message.from_user.id, dialog_status=False)
        await message.answer(text='Диалог очищен. Начнем общаться?', reply_markup=start_kb())

Я решил объеденить под две команды один обработчик.

Смысл данной функции в следующем:

• Если пользователя не было в базе данных, то мы его добавляем и после отправляем сообщение "Привет! Давай начнем общаться. Для этого просто нажми на кнопку "Начать диалог"" с клавиатурой "Начать диалог". Кроме того, статус пользователя станет "Не в диалоге".

• Если пользователь уже был в базе данных, то мы очистим его историю общения и установим статус "Не в диалоге" (поэтому подвязал команду restart, так как она более явно демонстирует тот процесс, который происходит)

Функция начала диалога:

@user_router.message(F.text.lower().contains('начать диалог'))
async def start_speak(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        await clear_dialog(user_id=message.from_user.id, dialog_status=True)
        await message.answer(text='Диалог начат. Введите ваше сообщение:', reply_markup=stop_speak())

Данная функция реагирует на словосочетание "начать диалог" выполняя 3 действия:

1. Меняет статус диалога для пользователя на "В диалоге" (in_dialog=True)

2. Отправляет сообщение пользователю "Диалог начат. Введите ваше сообщение" с клавиатурой с возможностью "Завершить диалог"

3. Очищает историю диалога

Функция завершения диалога:

@user_router.message(F.text.lower().contains('завершить диалог'))
async def start_speak(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        await clear_dialog(user_id=message.from_user.id, dialog_status=False)
        await message.answer(text='Диалог очищен! Начнем общаться?', reply_markup=start_kb())

Данная функция реагирует на словосочетание "начать диалог" выполняя 3 действия:

1. Меняет статус диалога для пользователя на "Не в диалоге" (in_dialog=False)

2. Отправляет сообщение пользователю "Диалог очищен! Начнем общаться?" с клавиатурой с возможностью "Завершить диалог".

3. Очищает историю диалога

Главная функция для диалога (на ней остановимся подробнее):

@user_router.message(F.text)
async def handle_message(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        check_open = await get_dialog_status(message.from_user.id)
        if check_open is False:
            await message.answer(text='Для того чтоб начать общение со мной, пожалуйста, нажмите на кнопку '
                                      '"Начать диалог".', reply_markup=start_kb())
            return
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        # формируем словарь с сообщением пользователя
        user_msg_dict = {"role": "user", "content": message.text}

        # сохраняем сообщение в базу данных и получаем историю диалога
        dialog_history = await add_message_to_dialog_history(user_id=message.from_user.id,
                                                             message=user_msg_dict,
                                                             return_history=True)

        #Пример работы с GROQ
        chat_completion = client_groq.chat.completions.create(model="llama3-70b-8192", messages=dialog_history)
        message_llama = await message.answer(text=chat_completion.choices[0].message.content, reply_markup=stop_speak())

        '''
        # Пример работы с локальной моделью
        
        chat_completion = local_client.chat.completions.create(model="llama3:8b", messages=dialog_history)
        message_llama = await message.answer(text=chat_completion.choices[0].message.content, reply_markup=stop_speak())
        '''

    # формируем словарь с сообщением ассистента
    assistant_msg = {"role": "assistant", "content": message_llama.text}

    # сохраняем сообщение ассистента в базу данных
    await add_message_to_dialog_history(user_id=message.from_user.id, message=assistant_msg,
                                        return_history=False)

Данная функция будет реагировать на любое текстовое сообщение от пользователя.

На старте идет проверка находится ли пользователь в диалоге:

• Не находится - тогда бот отправляет сообщение "Для того чтоб начать общение со мной, пожалуйста, нажмите на кнопку '"Начать диалог"" с клавиатурой "Начать диалог". После отправки этого сообщения функция завершается. Таким образом мы не запускаем диалог с Llama3 пока пользователь не окажется в статусе "В диалоге" (не нажмет на кнопку "Начать диалог")

• Находится.

Если пользователь находится в диалоге, то запустится логика, в рамках которой:

• Будет сформировано сообщение от пользователя в формате: {"role": "user", "content": message.text}

• Сообщение будет сохранено в базе данных

• Мы вернем всю историю диалога пользователя и клиента Llama3 в виде списка питоновских словарей

• История общений (весь список словарей) будет отправлена клиенту Llama3 (GROQ или локальному)

• Мы перехватим ответ от клиента Llama3

• Сохраним сообщение от клиета Llama3 в базу данных, закрепив его за пользователем

Если в этом моменте есть трудности в понимании, то вернитесь выше в часть статьи, где я рассматривал демки. Там логика работы такая-же, просто сообщения мы сохраняли не в базе данных, а в обычном списке.

В примере что вы видите я оставил модель Llama3 через GROQ, но, если ваша машина позволяет тянуть большие нейронки - можно воспользоваться локальной версией. Для этого просто импортируйте клиент с openai и раскоментируйте кусок в коде, который отвечает за работу с локальным клиентом Llama3:

chat_completion = local_client.chat.completions.create(model="llama3:8b", messages=dialog_history)
message_llama = await message.answer(text=chat_completion.choices[0].message.content, reply_markup=stop_speak())

Обратите внимание! Для пользования сервисом GROQ вам необходимо запустить VPN, как для доступа к сайту GROQ, так и для работы с моделью Llama3.

Запустим бота на VPS сервере (если он у вас, конечно, есть).

В данном моменте я поделюсь самым простым способом запуска Python проекта на VPS сервере через Docker (даже если вы абсолютно не знакомы с этой технологией, просто повторяйте за мной).

1. В корне проекта бота создаем файл Dockerfile и заполняем его

FROM python

WORKDIR /usr/src/app
# Копируем и устанавливаем зависимости Python
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

# Копируем все файлы из текущей директории в рабочую директорию контейнера
COPY . .

# Команда запуска контейнера
CMD ["/bin/bash", "-c", "python aiogram_run.py"]

2. Создаем в корне проекта файл .env с переменными окружения (это можно сделать и на сервере через утилиту nano)

3. Закидываем все файлы проекта, вместе с .env (если с ним не забудьте репозиторий на гите сделать приватным!) и Dockerfile

4. Заходим на VPS сервер

5. Устанавливаем Docker, если он ещё не был установлен

6. Создаем папку и закидываем с GitHub в нее файлы бота (git clone или git pull)

7. Создаем свой именной образ:

docker build -t my_image_name .

8. Запускаем контейнер

docker run -it -d --env-file .env --restart=unless-stopped --name container_name my_image_name

Тут вы видите, как привязать env-file к рабочему контейнеру. Убедитесь, что файл .env в проекте и в нем указаны все необходимые для бота переменные:

GROQ_API_KEY=your_groq_token
BOT_API_KEY=your_bot_token
ADMINS=admin1,admin2
ROOT_PASS=your_root_password
PG_LINK=postgresql://username:password@host:port/dbname

Просмотр логов

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

docker attach container_name

Для выхода из интерактивного режима воспользуйтесь комбинацией клавиш CTRL+P, CTRL+Q.

Если все настроено корректно, то после запуска контейнера произойдет и запуск бота. Проверим.

А теперь давайте проверим насколько бот запоминает контекст беседы.

А теперь посмотрим что в базе данных у нас происходит:

Теперь я нажму на кнопку "Очистить диалог" и обновлю таблицу:

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

Теперь давайте спросим бота о том что он обо мне знает.

Напоминаю, что полный код проекта с демками тут - EasyLlamaBot

Поклацать бота можно тут: Llama3Bot

Заключение

Дорогие друзья, вот и подошла к концу эта статья. Я понимаю, что материал может показаться сложным для тех, у кого нет достаточного опыта работы с языком Python и фреймворком Aiogram3. Тем не менее, я старался сделать изложение и код максимально доступными и понятным для каждого читателя.

Теперь вы знакомы с нейросетью Llama3 и знаете, как работать с ней локально и через платформу GROQ. Саму нейросеть, теперь, вы можете использовать не только в телеграмм ботах, но и в любой другом своем проекте.

На создание этой статьи, включая написание кода, я потратил все выходные. Очень надеюсь на ваш позитивный отклик в виде лайков, подписок и донатов (кнопка находится под статьей). Без вашей поддержки я просто физически не смогу продолжать подготовку такого обширного и детализированного контента, ведь у меня есть основная работа, семья и необходимость отдыхать.

Надеюсь на вашу поддержку.

Если у вас возникнут вопросы, пишите в комментариях, личных сообщениях или мессенджерах (контактные данные указаны в моем профиле).

Не забудьте также подписаться на мой Telegram-канал. В ближайшее время я планирую начать публикацию видео контента и эксклюзивных материалов, которые не будут опубликованы на Хабре (вход в канал бесплатный).

Благодарю вас за внимание и до скорого!