Создаем ChatGPT ассистента на основе КПТ методики Альберта Эллиса

В рамках статьи мы создадим приложение в виде интеллектуального ассистента, а также познакомимся и освоим на базовом уровне следующие инструменты:

• Haystack - разработка приложений на базе искусственного интеллекта.
• Chainlit - интеграция чата в веб-интерфейсе.

Требования

• Python >= 3.9
• OpenAI api ключ (https://platform.openai.com/api-keys

Знакомство с Haystack

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

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

Используя Haystack, весь процесс работы с данными может быть поделен на три этапа: Сырые данные (Raw Data), Компоненты (Components) и Пайплайн (Pipeline).

1. Любой пользовательский запрос - это сырые данные, будь то текст, аудио, файл или изображение. Их мы воспринимаем как "Raw Data".
2. При получении сырых данных нам потребуется их подготовка. Например, определение формата, классификация, конвертация из одного формата в другой, форматирование, очистка от нежелательных символов, объединение в один результирующий файл и т.д. За этот этап отвечают компоненты ("Components").
3. От первого пользовательского запроса до получения конечного результата данные "протекают" этап за этапом из одного компонента в другой (Pipeline).

Резюмируем:

На самом верхнем уровне у нас есть конвейер (Pipeline), в который добавляются компоненты (Components).

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

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

Детальнее о всех доступных компонентах и примерах можно ознакомиться в документации: https://docs.haystack.deepset.ai/docs/get_started

Схема взаимодействия

Рассмотрим обобщенную схему взаимодействия приложения, построенного по принципу коммуникации с внешними LLM-моделями.

1. Получаем запрос от пользователя.
2. Модифицируем запрос, добавляя дополнительные инструкции, условия и ограничения, если требуется. Эта модификация позволяет настроить взаимодействие с моделью на определённую форму коммуникации.
3. Транслируем запрос в модель.
4. Получаем ответ.
5. При необходимости модифицируем ответ перед отправкой пользователю, например: можем извлечь определённую часть ответа.
6. Транслируем пользователю готовый результат.

Задача

Мы хотим создать ассистента, который будет эмулировать терапевтическую сессию по методике рационально-эмоционально-поведенческой терапии "АВС"

Есть три варианта реализации задачи (от сложного к простому):

1. Fine-tuning/дообучение модели под конкретную специфику:
   • Настройка LLM моделей под конкретную задачу путем дообучения на собственном массиве данных.
2. Retrieval-Augmented Generation (RAG)/Поисково-улучшенная генерация:
   • Предварительный поиск наиболее релевантного документа из собственной базы данных и передача в LLM модели в виде контекста с указанием строгих инструкций для работы и генерации ответа исключительно на основе переданного контекста.
3. Assigning Roles and Prompt Engineering/Назначение ролей и инструкционный "промптинг":
   • Искусственное ограничение модели с помощью ролей и инструкционных запросов или подсказок. Поскольку ChatGPT и подобные большие модели обучены на большом массиве текстов (книги, Википедия и другие источники), такой подход возможен и является наиболее простым способом.

В данном материале мы рассмотрим третий вариант. Сперва узнаем, знаком ли ChatGPT с нужной нам методикой.

Подготовка

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

# Устанавливаем зависимости
pip install haystack-ai \
            python-dotenv \
            chainlit \
            langfuse-haystack

# Создаём и настраиваем .env файл
OPENAI_API_KEY= #Получить и указать openAi api ключ (https://platform.openai.com/api-keys)

# Создаём рабочий файл main.ру и подготавливаем базовый скелет
from dotenv import dotenv_values
import logging

# Подгружаем файл переменных окружения
config = dotenv_values(".env")
OPENAI_API_KEY = config["OPENAI_API_KEY"]

# Добавим логирование для отслежевания ошибок
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


async def main() -> None:
    # Здесь будет дальнейший код...


if __name__ == "__main__":
    main()

Опишем этапы, через которые нам потребуется пройти

# main.py
from dotenv import dotenv_values
import logging

# Подгружаем файл переменных окружения
config = dotenv_values(".env")
OPENAI_API_KEY = config["OPENAI_API_KEY"]

# Добавим логирования приложения для отслежевание ошибок
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


async def main() -> None:
    # 1. Получаем пользовательский запрос
    # 2. Дополняем пользовательский запрос, нашей собственной инстр
    # 2.1. Назначаем роль
    # 2.2. Задаём строгую структуру взаимодействия, чтобы исклю
    # 2.3. Добавляем память/контекст в виде предидущего диалога

    # 3. Транслируем в LLM модель

    # 4. Получаем результат, проводим анализ ответа и извлекаем пред

    # 5. Транслируем готовый результат пользователю.

if __name__ == "__main__":
    main()

Создание интеллектуального агента

Веб интерфейс чата

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

# Предидуший код...

# Импортируем chainlit
import chainlit as cl

# Применяем функцию обёртку on_message, для нашей основной функции
@cl.on_message
async def main(message: cl.Message) -> None:
    # 1. Получаем пользовательский запрос
    user_query = message.content

    # В целях тестирования работоспособности транслируем пользовате
    await cl.Message(
        content=user_query,
    ).send()


if __name__ == "__main__":
    main()

После чего запускаем приложение командой chainlit run main.py -w открываем в браузере по адресу: http://localhost:8000 и тестируем

Настройка haystack

Выше мы уже рассмотрели теоретические аспекты, теперь приступим к настройке компонентов.

# Предидуший код...

# Импортируем компонент OpenAIChatGenerator для взаимодействия с Ch
from haystack.components.generators.chat import OpenAIChatGenerator
# Импортируем компонент DynamicChatPromptBuilder для построения шаб
from haystack.components.builders import DynamicChatPromptBuilder
# Импортируем вспомогательный класс ChatMessage для удобного постро
from haystack.dataclasses import ChatMessage

@cl.on_message
async def main(message: cl.Message) -> None:
    # 1. Получаем пользовательский запрос
    user_query = message.content

    # Производим настройку
    gpt = OpenAIChatGenerator(api_key=Secret.from_token(OPENAI_API_KEY))
    prompt_builder = DynamicChatPromptBuilder()
    # Создаём конвейер и добавляем в него компоненты
    pipe = Pipeline()
    pipe.add_component("chat_prompt_builder", prompt_builder)
    pipe.add_component("gpt", gpt)
    # Связываем компоненты между собой
    pipe.connect("chat_prompt_builder.prompt", "llm.messages")

    # Добавляем пользовательское сообщение полученное из чата
    messages = [
        ChatMessage.from_user(user_query)
    ]

    # Транслируем в ChatGpt
    response = pipe.run(
        data={
            "prompt_builder": {"prompt_source": messages}
        }
    )

    # Извлекаем готовый результат:
    response_message = response["gpt"]["replies"][0].content

    # Отправляем обратно пользователю
    await cl.Message(
        content=response_message,
    ).send()


if __name__ == "__main__":
    main()

На данном этапе мы уже реализовали приложение, которое транслирует пользовательское сообщение в ChatGPT, дожидается ответа и возвращает результат. Однако пока оно является обычным транслятором без передачи истории сообщений, дополнительных инструкций и указаний следовать определенным правилам. Давайте доработаем наш код, добавив системные сообщения и память:

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

ChatGPT категоризирует сообщения на 4 типа:

• assistant: сообщение, содержащее текст модели
• user: сообщение, содержащее пользовательский текст
• system: сообщение, содержащее системную инструкцию
• function: сообщение, содержащее информацию о доступных инструментах

from haystack import Pipeline
from haystack.dataclasses import ChatMessage, ChatRole
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.builders import DynamicChatPromptBuilder
from haystack.utils import Secret
from dotenv import dotenv_values
import logging
import chainlit as cl
import re

config = dotenv_values(".env")
OPENAI_API_KEY = config["OPENAI_API_KEY"]
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Подготавливаем список для хранения истории
message_hist = []


# Подготавливаем функцию для извлечения истории сообщений чата.
def get_history(messages: list) -> list:
    # Фильтруем, исключая системные сообщения и возвращаем 5 послед
    messages = list(filter(lambda message: message.role == ChatRole.USER, messages))
    return [
        *messages[-5:]
    ]


# Подготавливаем функцию для построения системной инструкции
def build_template(query, history):
    # Строим шаблон запроса к ChatGpt. Данный шаблон строит динамическо

    # 1. Системное сообщение - статическое/неизменяемое для каждого зап
    # 2. history - динамическое, 5 последних сообщений взаимодействия м
    # 3. query - динамическое, текущий запрос пользователя

    return [
        ChatMessage.from_system(f"""
Твоя роль, выступать в роли AI ассистента. AI ассистент явл
который проводит терапевтическую сессию по КПТ методике Аль
Задача ассистента вести логически последовательный диалог п
Проанализируй пользовательский запрос и сформулируй логичес

Ожидаемый ответ:
Reflection: [Процесс саморефлексии. Анализ пользовательског
Throught: [Логическая цепочка рассуждения AI ассистента. На
Answer: [Ответ Аі ассистента на основе Throught и Reflectio

Ниже приведена предидущая история разговора:
"""),
        *history,
        ChatMessage.from_user(f"Query: {query}")
    ]

# Подготавливаем вспомагательную функцию для извлечения готового ре
async def extract_answer(response) -> str:
    pattern = r"Answer:\s*(.*)"
    match = re.search(pattern, response)
    if match:
        result = match.group(1).strip()
        return result
    else:
        return ""


# Подготавливаем вспомагательную функцию для извлечения процесса ра
@cl.step
async def extract_throught(response) -> str:
    pattern = r"Throught:\s*(.*)"
    match = re.search(pattern, response)
    if match:
        result = match.group(1).strip()
        return result
    else:
        return ""


# Обновляем функцию main с учётом новых функций
@cl.on_message
async def main(message: cl.Message) -> None:
    # 1. Получаем пользовательский запрос
    user_query = message.content

    # Инициализируем компонент взаимодействия с ChatGpt
    gpt = OpenAIChatGenerator(api_key=Secret.from_token(OPENAI_API_KEY))
    # Инициализируем компонент построения запросов для ChatGpt
    prompt_builder = DynamicChatPromptBuilder()
    # Создаём конвейер и добавляем в него компоненты
    pipe = Pipeline()
    pipe.add_component("prompt_builder", prompt_builder)
    pipe.add_component("gpt", gpt)
    # Связываем внутри рipline компоненты между собой, чтобы pipile
    pipe.connect("prompt_builder.prompt", "gpt.messages")

    # Строим шаблон сообщения
    messages = build_template(user_query, get_history(message_hist))

    # Транслируем в ChatGpt
    response = pipe.run(
        data={
            "prompt_builder": {"prompt_source": messages}
        }
    )

    # Извлекаем готовый результат:
    response_message = response["gpt"]["replies"][0].content
    # Извлекаем процесс рассуждения
    throught = await extract_throught(response_message)
    # Извлекаем ответ от ChatGpt
    answer = await extract_answer(response_message)

    # Добавляем новый ответ от ChatGpt в историю
    messages.append(ChatMessage.from_assistant(response_message))
    message_hist.extend(messages)

    # Отправляем извлечённый ответ пользователю
    await cl.Message(
        content=answer,
    ).send()


if __name__ == "__main__":
    main()

Проверка результата

Как можно видеть, процесс довольно прост: получив пользовательский запрос, мы дополняем его инструкцией, в которой просим следовать определенной структуре:

1. Анализ пользовательского запроса.
2. Процесс рассуждения/саморефлексии для определения дальнейшего этапа терапевтической сессии.
3. Формирование логически последовательного ответа.
4. Подготовка строгой структуры ответа, чтобы мы могли извлечь данные по паттерну и транслировать пользователю готовый ответ.

Дополнительные ссылки

• Haystack-документация
• OpenApi-ключ
• Chainlit-интерфейс
• Гайд по промпт-инжинирингу