Базовый класс и кастомные исключения для API ChatGPT

В эпоху стремительно развивающихся интернет-технологий, особое внимание уделяется выбору оптимальных инструментов для хранения и управления данными. Одним из таких инструментов, заслуживающих особого внимания, является Redis — продвинутое в памяти хранилище структур данных, которое используется как база данных, кэш и брокер сообщений.

Redis обладает несколькими ключевыми характеристиками, делающими его идеальным решением для использования в современных проектах, в том числе и при работе с моделями, связанными с искусственным интеллектом и машинным обучением, такими как GPT (Generative Pre-trained Transformer).

Высокая производительность: Redis хранит данные в оперативной памяти, что обеспечивает чрезвычайно быстрый доступ к ним по сравнению с дисковыми базами данных. Это критически важно для проектов, требующих высокоскоростной обработки больших объемов запросов в реальном времени, например, для обработки запросов к моделям GPT.

Гибкость структур данных: Redis поддерживает широкий спектр структур данных, таких как строки, списки, множества, хэши, упорядоченные множества и пространственные индексы. Это позволяет разработчикам оптимизировать хранение данных с учетом конкретных требований проекта.

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

Масштабируемость и отказоустойчивость: Redis предлагает расширенные возможности для масштабирования, включая репликацию, шардирование и автоматическое переключение на резервный узел (failover), что обеспечивает высокую доступность и надежность хранилища.

В контексте работы с моделями GPT и другими интеллектуальными системами, Redis может выполнять ряд ключевых функций. Например, использование Redis как кэша для хранения промежуточных результатов обработки запросов может значительно ускорить получение ответов от моделей, сократив время, необходимое для их повторного вычисления. Кроме того, Redis может использоваться для управления очередями задач, организации распределенных блокировок и координации работы нескольких экземпляров приложения в распределенной среде.

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

После того как мы углубились в теоретические аспекты и выявили ключевые преимущества использования Redis, настало время рассмотреть, как эти знания применяются на практике. Конкретно, мы сконцентрируемся на реализации конфигурации Redis в файле settings.py проекта.

Файл settings.py играет важную роль в Django-проектах, поскольку содержит все глобальные настройки приложения, включая конфигурацию баз данных, промежуточного программного обеспечения, шаблонов, статических файлов и многого другого. Включение Redis в этот файл обеспечивает централизованное управление параметрами подключения и упрощает доступ к ним из различных частей приложения.

Приведенный код описывает процесс конфигурации клиента Redis в Django-проекте:

# settings.py
...
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = os.getenv('REDIS_PORT', '6379')
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD')
REDIS_URL = f'redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/0'
REDIS_CLIENT_DATA = {
    'host': REDIS_HOST,
    'port': REDIS_PORT,
    'db': 0,
	'password': REDIS_PASSWORD
}
pool = redis.ConnectionPool(**REDIS_CLIENT_DATA)
REDIS_CLIENT = redis.Redis(connection_pool=pool)
...

Давайте разберем эту конфигурацию по частям:

• Получение параметров окружения: Используя os.getenv, мы извлекаем значения для REDIS_HOST,  REDIS_PORT и REDIS_PASSWORD из переменных окружения. Это подход обеспечивает гибкость конфигурации, позволяя легко адаптировать настройки к различным средам (разработка, тестирование, продакшн) без изменения исходного кода. По умолчанию используются стандартные значения для Redis: localhost и 6379.
• Формирование URL Redis: Строка подключения REDIS_URL формируется на основе полученных параметров. Это удобный способ хранить полную информацию о подключении в одной переменной, который может быть использован в других частях приложения, если требуется прямое подключение к Redis без использования созданного клиента.
• redis.ConnectionPool: используется для создания пула соединений с Redis. Пул соединений позволяет эффективно переиспользовать уже установленные соединения, уменьшая накладные расходы на установку новых соединений при каждом запросе. Это особенно важно в высоконагруженных приложениях, где каждая миллисекунда на счету.
• Инициализация клиента Redis: Наконец, создается экземпляр redis.Redis, который используется для взаимодействия с Redis из кода приложения. При создании этого объекта в качестве параметра передается ранее созданный пул соединений.

В проектах, связанных с искусственным интеллектом и машинным обучением, таких как системы, работающие на основе моделей GPT (Generative Pre-trained Transformer), ключевую роль играют данные моделей. Они не только обеспечивают структуру, но и инкапсулируют бизнес-логику, связанную с обработкой и хранением информации о пользователях и взаимодействиях с AI. Рассмотрим модель GptModels, которая является центральной в контексте настройки и использования различных версий моделей GPT в Django-проекте.

class GptModels(models.Model):
    title = models.CharField(_('модель GPT'), max_length=28)
    is_default = models.BooleanField(_('доступна всем по умолчанию'), default=False)
    token = models.CharField(_('токен OpenAI'), max_length=51)
    context_window = models.IntegerField(_('окно количества токенов для передачи истории в запросе'))
    max_request_token = models.IntegerField(_('максимальное количество токенов в запросе'))
    time_window = models.IntegerField(_('окно времени для передачи истории в запросе, мин'), default=30)

    class Meta:
        verbose_name = _('модель GPT OpenAi')
        verbose_name_plural = _('модели GPT OpenAi')

    def __str__(self):
        return self.title

    def clean(self, *args, **kwargs):
        default_model = GptModels.objects.filter(is_default=True).exclude(pk=self.pk)
        if not self.is_default and not default_model:
            raise ValidationError('Необходимо указать хотя бы одну модель по умолчанию для всех.')

    def save(self, *args, **kwargs):
        self.clean()
        with transaction.atomic():
            super().save(*args, **kwargs)
            if self.is_default:
                GptModels.objects.filter(is_default=True).exclude(pk=self.pk).update(is_default=False)

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

• title: Название модели GPT. Это поле позволяет идентифицировать модель по её названию.
• default: Булево значение, указывающее, является ли данная модель доступной по умолчанию для всех пользователей. Это поле гарантирует, что существует хотя бы одна модель, доступная для общего использования.
• token: Токен для аутентификации запросов к API. Обеспечивает защиту доступа и контроль над использованием модели.
• context_window: Определяет количество токенов, которое можно передать в запросе для контекста. Это поле позволяет настроить размер входных данных для модели.
• max_request_token: Максимальное количество токенов, которое может быть использовано в одном запросе пользователя. Ограничивает размер запроса для предотвращения его чрезмерного расширения.
• time_window: Временное окно в минутах, которое используется для передачи истории запросов и ответов модели. Это позволяет создавать более контекстуализированные и осмысленные ответы.

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

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

Связи между моделями в Django позволяют устанавливать отношения "один к одному", "один ко многим" и "многие ко многим" между различными моделями. Эти связи реализуются с помощью полей ForeignKey, OneToOneField и ManyToManyField соответственно.

• ForeignKey: Используется для указания связи "один ко многим". Это позволяет модели ссылаться на другую модель, создавая таким образом иерархическую связь. Например, если у нас есть модель User и модель Article, мы можем использовать ForeignKey для указания того, что каждая статья принадлежит определенному пользователю.
• OneToOneField: Это поле используется для создания связи "один к одному" между двумя моделями. Это может быть полезно, когда один объект модели должен иметь в точности один связанный объект другой модели. Например, если у нас есть модель User и модель UserProfile, которая хранит дополнительную информацию о пользователе.
• ManyToManyField: Используется для создания связи "многие ко многим" между моделями. Этот тип связи позволяет объекту одной модели быть связанным с множеством объектов другой модели и наоборот. Примером может служить модель Student и модель Course, где студенты могут записаться на множество курсов, а курсы могут иметь множество зарегистрированных студентов.

Подробнее рассмотрим реализацию двух ключевых классов моделей в models.py - UserGptModels и HistoryAI, которые иллюстрируют использование связей и наследования в Django.

class UserGptModels(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='approved_models')
    active_model = models.ForeignKey(GptModels, on_delete=models.SET_NULL, null=True, blank=True, related_name='active_for_users', verbose_name=_('Активная модель'))
    approved_models = models.ManyToManyField(to=GptModels, related_name='approved_users')
    time_start = models.DateTimeField(_('время начала окна для передачи истории'), default=now)

    class Meta:
        verbose_name = _('разрешенная GPT модели юзера')
        verbose_name_plural = _('разрешенные GPT модели юзера')

    def __str__(self):
        return f'User: {self.user}, Active model: {self.active_model}'

    def save(self, *args, **kwargs):
        is_new = self._state.adding
        super(UserGptModels, self).save(*args, **kwargs)

        if not self.active_model:
            default_model = GptModels.objects.filter(is_default=True).first()
            if default_model:
                self.active_model = default_model
                self.save(update_fields=['active_model'])

        if is_new and default_model:
            if not self.approved_models.filter(id=default_model.id).exists():
                self.approved_models.add(default_model)

UserGptModels - модель, предназначенная для управления связями между пользователями и моделями GPT, которые они могут использовать. Эта модель выполняет несколько ключевых функций:

• OneToOneField с User гарантирует, что каждый пользователь связан с единственным экземпляром UserGptModels, что обеспечивает простоту управления доступными пользователю моделями GPT.
• ForeignKey на GptModels позволяет определить активную модель GPT для каждого пользователя, с возможностью ее изменения.
• ManyToManyField с GptModels используется для определения набора моделей, которые были одобрены для использования пользователем. Это позволяет реализовать гибкую систему управления доступом к различным моделям GPT.

class HistoryAI(Create):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='history_ai', null=True, blank=True)
    question = models.TextField(_('Вопрос'))
    question_tokens = models.PositiveIntegerField(null=True)
    answer = models.TextField(_('Ответ'))
    answer_tokens = models.PositiveIntegerField(null=True)

    class Meta:
        verbose_name = _('История запросов к ИИ')
        verbose_name_plural = _('История запросов к ИИ')
        ordering = ('created_at',)

    def __str__(self):
        return f'User: {self.user}, Question: {self.question}'

HistoryAI - модель, предназначенная для сохранения истории взаимодействия пользователя с искусственным интеллектом. Эта модель не только хранит вопросы и ответы, но и количество токенов в них, что может быть полезно для анализа использования и оптимизации работы системы. Использование ForeignKey для связи с моделью User позволяет отслеживать, какой пользователь совершал каждый запрос. Create, от которого она наследуется, это абстрактный класс инициирующий дату создания и сортировку.

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

Метод clean

Метод clean предназначен для выполнения дополнительной валидации полей модели, которая не может быть выполнена на уровне базы данных или через стандартные валидаторы Django. Это может включать проверку условий, которые зависят от нескольких полей модели одновременно или от внешних условий. Важно отметить, что метод clean вызывается автоматически при использовании ModelForm, в других случаях его необходимо вызывать вручную в коде перед сохранением объекта.

Например, в модели GptModels, метод clean используется для проверки, что в системе может быть только одна активная модель по умолчанию. Это делается путем проверки существования других экземпляров с флагом default=True при попытке установить текущий экземпляр как по умолчанию.

def clean(self, *args, **kwargs):
    is_there_default_model = GptModels.objects.filter(is_default=True).exclude(pk=self.pk).exists()
    if not self.is_default and not is_there_default_model:
        raise ValidationError('Необходимо указать хотя бы одну модель по умолчанию для всех.')
    if self.is_default and is_there_default_model:
        raise ValidationError('По умолчанию может быть только одна модель.')

Метод save

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

В модели UserGptModels, например, метод save используется для автоматического назначения активной модели GPT, если таковая не была установлена. Это демонстрирует, как можно управлять связанными объектами и их состояниями во время сохранения.

def save(self, *args, **kwargs):
    is_new = self._state.adding
    super(UserGptModels, self).save(*args, **kwargs)

    if not self.active_model:
        default_model = GptModels.objects.filter(is_default=True).first()
        if default_model:
            self.active_model = default_model
            self.save(update_fields=['active_model'])

    if is_new and default_model:
        if not self.approved_models.filter(id=default_model.id).exists():
            self.approved_models.add(default_model)

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

Полиморфизм — это принцип ООП, позволяющий объектам разных классов быть обработанными через один и тот же интерфейс. Этот принцип позволяет функциям или методам работать с разными типами данных, не зная точно, с каким типом они работают в данный момент. Полиморфизм обеспечивает гибкость и масштабируемость кода, позволяя расширять функциональность без изменения существующего кода или только с минимальными изменениями.

Применительно к базовому классу для работы с ChatGPT, полиморфизм позволяет создать унифицированный интерфейс взаимодействия с различными версиями GPT или даже с разными моделями ИИ, не изменяя основную структуру и логику работы с ними. Разработчик может определить общий интерфейс для всех моделей и затем реализовать специфичные для каждой модели детали в наследуемых классах.

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

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

Разработка базового класса GetAnswerGPT для взаимодействия с ChatGPT представляет собой ключевой элемент архитектуры нашего приложения, обеспечивающего интеграцию с искусственным интеллектом. Этот класс демонстрирует, как можно инкапсулировать логику обработки запросов к модели GPT, включая подготовку данных, отправку запросов и обработку ответов.

class GetAnswerGPT():
    """
    Класс для получения ответов от модели GPT.

    ### Args:
    - query_text (`str`): Текст запроса пользователя.
    - assist_prompt (`str`): Промпт для ассистента в head модели.
    - user (`Model`): Пользователь.
    - history_model (`Model`): Модель для хранения истории.
    - chat_id (`int`, optional): ID чата. Defaults to None.

    """
    MAX_TYPING_TIME = 3

    def __init__(self, query_text: str, assist_prompt: str, user: 'Model', history_model: 'Model', chat_id: int = None, temperature: float = 0.5) -> None:
        # Инициализация свойств класса
        self.user = user                    # модель пользователя пославшего запрос
        self.is_user_authenticated = self.user.is_authenticated  # гость или аутентифицированный пользователь
        self.history_model = history_model  # модель для хранения истории
        self.query_text = query_text        # текст запроса пользователя
        self.assist_prompt = assist_prompt  # промпт для ассистента в head модели
        self.query_text_tokens = None       # количество токенов в запросе
        self.chat_id = chat_id              # id чата в котором инициировать typing
        self.temperature = temperature      # уровень энтропии при ответе от GPT модели
        # Дополнительные свойства
        self.assist_prompt_tokens = 0       # количество токенов в промпте ассистента в head модели
        self.all_prompt = []                # общий промпт для запроса
        self.current_time = now()           # текущее время для окна истории
        self.return_text = ''               # текст полученный в ответе от модели
        self.return_text_tokens = None      # количество токенов в ответе
        self.event = asyncio.Event() if chat_id else None  # typing в чат пользователя
        self.user_models = None             # разрешенные GPT модели пользователя

    @property
    def check_long_query(self) -> bool:
        return self.query_text_tokens > self.model.max_request_token

    async def get_answer_chat_gpt(self) -> dict:
        """Основная логика."""
        await self.init_user_model()
        self.query_text_tokens, self.assist_prompt_tokens, _ = await asyncio.gather(
            self.__num_tokens(self.query_text, 4),
            self.__num_tokens(self.assist_prompt, 7),
            self.__check_in_works(),
        )
        if self.check_long_query:
            raise LongQueryError(
                f'{self.user.first_name if self.is_user_authenticated else "Дорогой друг" }, слишком большой текст запроса.\n'
                'Попробуйте сформулировать его короче.'
            )
        try:
            if self.event:
                asyncio.create_task(self.__send_typing_periodically())
            await self.get_prompt()
            await self.httpx_request_to_openai()
            if self.is_user_authenticated:
                asyncio.create_task(self._create_history_ai())
        except Exception as err:
            _, type_err, traceback_str = await handle_exceptions(err, True)
            raise type_err(f'\n\n{str(err)}{traceback_str}')
        finally:
            await self.__del_mess_in_redis()

    async def __send_typing_periodically(self) -> None:
        """"Передаёт TYPING в чат Телеграм откуда пришёл запрос."""
        time_stop = datetime.now() + timedelta(minutes=self.MAX_TYPING_TIME)
        while not self.event.is_set():
            bot.send_chat_action(chat_id=self.chat_id, action=ChatAction.TYPING)
            await asyncio.sleep(2)
            if datetime.now() > time_stop:
                break

    async def httpx_request_to_openai(self) -> None:
        """Делает запрос в OpenAI и выключает typing."""
        transport = AsyncProxyTransport.from_url(settings.SOCKS5)
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.model.token}"
        }
        data = {
            "model": self.model.title,
            "messages": self.all_prompt,
            "temperature": self.temperature
        }
        try:
            async with httpx.AsyncClient(transport=transport) as client:
                response = await client.post(
                    "https://api.openai.com/v1/chat/completions",
                    headers=headers,
                    json=data,
                    timeout=60 * self.MAX_TYPING_TIME,
                )
                response.raise_for_status()
                completion = response.json()
                choices = completion.get('choices')
                if choices and len(choices) > 0:
                    first_choice = choices[0]
                    self.return_text = first_choice['message']['content']
                    self.return_text_tokens = completion.get('usage', {}).get('completion_tokens')
                    self.query_text_tokens = completion.get('usage', {}).get('prompt_tokens')
                else:
                    raise ValueChoicesError(f"`GetAnswerGPT`, ответ не содержит полей 'choices': {json.dumps(completion, ensure_ascii=False, indent=4)}")

        except httpx.HTTPStatusError as http_err:
            raise OpenAIResponseError(f'`GetAnswerGPT`, ответ сервера был получен, но код состояния указывает на ошибку: {http_err}') from http_err
        except httpx.RequestError as req_err:
            raise OpenAIConnectionError(f'`GetAnswerGPT`, проблемы соединения: {req_err}') from req_err
        except json.JSONDecodeError:
            raise OpenAIJSONDecodeError('`GetAnswerGPT`, ошибка при десериализации JSON.')
        except Exception as error:
            raise UnhandledError(f'Необработанная ошибка в `GetAnswerGPT.httpx_request_to_openai()`: {error}') from error
        finally:
            if self.event:
                self.event.set()

    async def _create_history_ai(self):
        """Создаём запись истории в БД для моделей поддерживающих асинхронное сохранение."""
        self.history_instance = await self.history_model.objects.acreate(
            user=self.user,
            question=self.query_text,
            question_tokens=self.query_text_tokens,
            answer=self.return_text,
            answer_tokens=self.return_text_tokens
        )

    async def __num_tokens(self, text: str, corr_token: int = 0) -> int:
        """Считает количество токенов.
        ## Args:
        - text (`str`): текс для которого возвращается количество
        - corr_token (`int`): количество токенов для ролей и разделителей

        """
        try:
            encoding = await tiktoken_async.encoding_for_model(self.model.title)
        except KeyError:
            encoding = await tiktoken_async.get_encoding("cl100k_base")
        return len(encoding.encode(text)) + corr_token

    async def add_to_prompt(self, role: str, content: str) -> None:
        """Добавляет элемент в список all_prompt."""
        self.all_prompt.append({'role': role, 'content': content})

    async def get_prompt(self) -> None:
        """Prompt для запроса в OpenAI и модель user."""
        history = []
        await self.add_to_prompt('system', self.assist_prompt)
        if self.is_user_authenticated:
            history = await sync_to_async(list)(self.history_model.objects.filter(
                user=self.user,
                created_at__range=[self.time_start, self.current_time]
            ).exclude(
                answer__isnull=True
            ).values(
                'question', 'question_tokens', 'answer', 'answer_tokens'
            ))
        token_counter = self.query_text_tokens + self.assist_prompt_tokens
        for item in history:
            question_tokens = item.get('question_tokens', 0)
            answer_tokens = item.get('answer_tokens', 0)
            # +11 - токены для ролей и разделителей: 'system' - 7 'user' - 4
            token_counter += question_tokens + answer_tokens + 11

            if token_counter >= self.model.context_window:
                break

            await self.add_to_prompt('user', item['question'])
            await self.add_to_prompt('assistant', item['answer'])

        await self.add_to_prompt('user', self.query_text)

    async def init_user_model(self):
        """Инициация активной модели юзера и начального времени истории в prompt для запроса."""
        if self.is_user_authenticated:
        	queryset = UserGptModels.objects.select_related('active_model', 'active_prompt')
            self.user_models, created = await queryset.aget_or_create(user=self.user, defaults={'time_start': self.current_time})
            self.model = self.user_models.active_model
            if not created and self.model:
                time_window = timedelta(minutes=self.model.time_window)
                self.time_start = max(self.current_time - time_window, self.user_models.time_start)
            else:
                self.time_start = self.current_time
        else:
            self.model = await GptModels.objects.filter(is_default=True, consumer=self.consumer).afirst()

    @sync_to_async
    def __check_in_works(self) -> bool:
        """Проверяет нет ли уже в работе этого запроса в Redis и добавляет в противном случае."""
        queries = redis_client.lrange(f'gpt_user:{self.user.id}', 0, -1)
        if self.query_text.encode('utf-8') in queries:
            raise InWorkError('Запрос уже находится в работе.')
        redis_client.lpush(f'gpt_user:{self.user.id}', self.query_text)

    @sync_to_async
    def __del_mess_in_redis(self) -> None:
        """Удаляет входящее сообщение из Redis."""
        redis_client.lrem(f'gpt_user:{self.user.id}', 1, self.query_text.encode('utf-8'))

Рассмотрим детали реализации и принципы работы этого класса.

• MAX_TYPING_TIME: максимальное время ожидания ответа.
• Конструктор класса принимает несколько параметров, включая текст запроса пользователя, специальный промпт для ассистента, информацию о пользователе и модели для хранения истории взаимодействий, а также идентификатор чата и параметры настройки запроса к модели GPT.
• check_long_query: проверка на превышение допустимого количества токенов в запросе пользователя. В старых моделях, когда все окно токенов составляло всего 4К, это было актуально. Сейчас опционально.
• get_answer_chat_gpt: основная функция класса, организующая получение ответа от GPT, включая проверку длины запроса, отправку typing в чат, формирование промпта, запрос к OpenAI и обработку ответа.
• __send_typing_periodically: функция, отправляющая статус "печатает" в Телеграмм чат в течение заданного времени ожидания.
• httpx_request_to_openai: выполнение HTTP-запроса к API OpenAI для получения ответа на заданный запрос с использованием асинхронного клиента httpx.
• _create_history_ai: сохранение истории взаимодействия в базу данных.
• __num_tokens: подсчёт количества токенов в тексте запроса или ответа.
• add_to_prompt: добавление элемента в общий промпт запроса.
• get_prompt: формирование окончательного промпта для запроса в OpenAI, включая предыдущие взаимодействия с пользователем.
• init_user_model: инициализация модели пользователя, включая определение активной модели GPT для аутентифицированных пользователей.
• __check_in_works: проверка на наличие текущего запроса в работе с использованием Redis для избежания дублирования.
• __del_mess_in_redis: удаление информации о запросе из Redis после обработки запроса.

Базовый класс GetAnswerGPT является центральным элементом нашего приложения для взаимодействия с искусственным интеллектом ChatGPT. Этот класс включает в себя ряд ключевых функций, использующих асинхронность для эффективной обработки запросов и взаимодействия с внешним API. Подробнее рассмотрим эти методы и принципы их работы.

Инициализация и установка свойств: Конструктор класса инициализирует основные свойства, необходимые для работы с запросами пользователя и моделью ChatGPT. Это включает в себя текст запроса, информацию о пользователе, используемую модель истории (на случай если их несколько), для сохранения запросов и ответов, а также различные параметры запроса к модели GPT, такие как температура и идентификатор чата, который так-же является флагом к запуску функции send_typing_periodically.

Асинхронность в работе с ChatGPT: Одной из ключевых особенностей класса является его асинхронная природа. Методы, такие как get_answer_chat_gpt, send_typing_periodically, и httpx_request_to_openai, используют асинхронные операции для выполнения сетевых запросов и других операций, требующих ожидания, без блокирования главного потока выполнения.
Метод get_answer_chat_gpt демонстрирует асинхронный поток выполнения, включающий в себя инициализацию активной модели пользователя, подсчет токенов запроса и промпта, проверку условий выполнения запроса и, наконец, отправку запроса к модели GPT и обработку ответа.

Асинхронное взаимодействие с API: Метод httpx_request_to_openai представляет собой асинхронный запрос к API OpenAI, используя библиотеку httpx. Этот метод демонстрирует, как можно эффективно обрабатывать внешние HTTP-запросы в асинхронном режиме, что особенно важно при работе с веб-сервисами, где время ответа может быть непредсказуемым. Дополнительно, внедрение поддержки прокси-транспорта гарантирует, что запросы к API будут обрабатываться без риска блокировки, предоставляя надежный канал связи с сервисом OpenAI.

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

Асинхронные утилиты: В классе используются асинхронные утилиты, такие как sync_to_async и database_sync_to_async, для интеграции с синхронным кодом Django, например, при работе с моделями базы данных. Это позволяет эффективно сочетать асинхронную логику обработки запросов к ИИ с синхронным кодом Django, не нарушая общую структуру приложения.

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

Обработка исключений в Python осуществляется с использованием блоков try, except, else, и finally. Ключевая идея заключается в том, чтобы "попытаться" выполнить определенный блок кода (try), который потенциально может вызвать ошибку, и "перехватить" возникшее исключение (except), чтобы обработать его адекватно. Блок else выполняется, если в блоке try ошибок не возникло, а finally выполняется в любом случае, независимо от того, возникло исключение или нет, и часто используется для выполнения необходимых задач по очистке ресурсов.

Важность правильного управления ошибками:

1. Улучшение устойчивости приложения: Адекватная обработка исключений позволяет приложению продолжить работу даже в случае возникновения ошибок, тем самым повышая его надежность и доступность.
2. Безопасность: Неправильная обработка ошибок может привести к утечке чувствительной информации или создать потенциальные уязвимости в безопасности. Корректное управление исключениями помогает предотвратить такие риски.
3. Отладка и диагностика: Предоставление подробной информации об ошибках и исключениях облегчает процесс отладки и диагностики проблем, что важно как на этапе разработки, так и при поддержке продукта.
4. Пользовательский опыт: Вместо "падения" приложения или веб-сайта, корректная обработка ошибок может предоставить пользователю понятное сообщение о возникшей проблеме, что положительно влияет на восприятие продукта.

В контексте интеграции с моделями искусственного интеллекта, такими как ChatGPT, правильное управление ошибками и исключениями становится особенно важным. Внешние запросы к API, обработка ответов, асинхронные операции — все эти аспекты подвержены риску возникновения исключений. Эффективная обработка таких ситуаций не только обеспечивает стабильность работы приложения, но и помогает предоставить конечному пользователю более качественный и предсказуемый сервис.

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

# ai.gpt_exception

import traceback

class LogTracebackExceptionError(Exception):
    """Абстракция для логирования traceback."""
    def __init__(self, message, log_traceback=True):
        super().__init__(message)
        self.log_traceback = log_traceback

class InWorkError(LogTracebackExceptionError):
    """Исключение, вызываемое, если запрос находится в работе."""
    pass

class LongQueryError(LogTracebackExceptionError):
    """Исключение для случаев, когда текст запроса слишком большой."""
    def __init__(self, message="Слишком большой текст запроса. Попробуйте сформулировать его короче."):
        self.message = message
        super().__init__(self.message)

class UnhandledError(LogTracebackExceptionError):
    """Исключение для обработки неожиданных ошибок."""
    pass

class OpenAIRequestError(LogTracebackExceptionError):
    """Ошибки, связанные с запросами к OpenAI."""
    pass

class OpenAIResponseError(OpenAIRequestError):
    """Ошибки, связанные с ответами от OpenAI."""
    pass

class OpenAIConnectionError(OpenAIRequestError):
    """Ошибки соединения при запросах к OpenAI."""
    pass

class OpenAIJSONDecodeError(OpenAIRequestError):
    """Ошибки при десериализации ответов OpenAI."""
    pass

class ValueChoicesError(OpenAIRequestError):
    """Ошибки в содержании ответа."""
    pass

async def handle_exceptions(err: Exception, include_traceback: bool = False) -> tuple[str, type, str]:
    """
    Обработчик исключений для запросов к OpenAI.

    ### Args:
    - err (`Exception`): Объект исключения.
    - include_traceback (`bool`, optional): Флаг для включения трассировки из пространства текущего вызова.

    ### Returns:
    - tuple[`str`, `type`, `str`]: Кортеж с текстом ошибки, типом ошибки и трассировкой.

    """
    user_error_text = 'Что-то пошло не так 🤷🏼\nВозможно большой наплыв запросов, которые я не успеваю обрабатывать 🤯'
    error_messages = {
        InWorkError: 'Я ещё думаю над вашим вопросом.',
        LongQueryError: : 'Слишком большой текст запроса. Попробуйте сформулировать его короче.',
        ValueChoicesError: user_error_text,
        OpenAIResponseError: 'Проблема в получении ответа от искусственного интеллекта. Вероятно, он временно недоступен.',
        OpenAIConnectionError: 'Проблема с подключением... Похоже, искусственный интеллект на мгновение отключился.',
        OpenAIJSONDecodeError: user_error_text,
        UnhandledError: user_error_text,
    }

    error_message = error_messages.get(type(err), user_error_text)
    if isinstance(error_message, str):
        formatted_error_message = error_message
    else:
        formatted_error_message = error_message(err)

    include_traceback = include_traceback or (hasattr(err, 'log_traceback') and err.log_traceback)
    traceback_str = ''
    if include_traceback:
        traceback_str = f'\n\nТрассировка:\n{traceback.format_exc()[-1024:]}'

    return formatted_error_message, type(err), traceback_str

Рассмотрим детальнее классы исключений, определенные в модуле gpt_exception.py, и разберем их предназначение и способы использования.

class LogTracebackExceptionError(Exception):

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

class InWorkError(LogTracebackExceptionError):

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

class LongQueryError(LogTracebackExceptionError):

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

class OpenAIRequestError(LogTracebackExceptionError):

Группа исключений, наследуемых от OpenAIRequestError, охватывает широкий спектр ошибок, связанных с взаимодействием с API OpenAI. Это включает в себя ошибки соединения, ошибки десериализации JSON и ошибки, связанные с содержимым ответа. Разделение этих ошибок на отдельные классы позволяет обработать каждый тип ошибки более точно, предоставляя пользователю более информативные сообщения об ошибках.

async def handle_exceptions(err: Exception, include_traceback: bool = False) -> tuple[str, type, str]:

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

Кастомные исключения в gpt_exception.py не только улучшают управление ошибками в приложении, но и способствуют повышению его надежности и удобства использования. Подход к обработке исключений, основанный на четко определенных классах ошибок, позволяет разработчикам более эффективно реагировать на проблемы, а пользователям — получать понятные и информативные сообщения об ошибках. 

В предыдущих разделах мы обсудили важность и методы создания кастомных исключений в контексте взаимодействия с ChatGPT, включая разработку модуля gpt_exception.py. Однако, для полноценной интеграции этих исключений в архитектуру приложения требуется эффективный механизм их обработки. Функция handle_exceptions играет ключевую роль в этом процессе, предоставляя унифицированный способ реагирования на ошибки и исключения в приложении. Более детально рассмотрим, как работает этот механизм и почему он так важен.

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

async def handle_exceptions(err: Exception, include_traceback: bool = False) -> tuple[str, type, str]:

Входные параметры функции:

• err: объект исключения, который необходимо обработать.
• include_traceback: булевый флаг, указывающий на необходимость включения трассировки стека в ответе, для класса из которого она инициирована.

Функция возвращает кортеж из трех элементов:

1. Строка с сообщением об ошибке для пользователя.
2. Тип исключения, который может быть использован для дальнейшей обработки.
3. Трассировка стека (если запрошено), которая может быть полезна для отладки и логирования.

Ключевые аспекты функции:

• Централизованная обработка ошибок: Этот подход позволяет стандартизировать реакцию на различные типы исключений во всем приложении, упрощая его поддержку и развитие.
• Гибкость: Функция поддерживает обработку широкого спектра исключений, включая как стандартные ошибки Python, так и кастомные исключения, определенные в приложении.
• Пользовательский опыт: Возвращаемые функцией сообщения об ошибках могут быть адаптированы для понимания конечными пользователями, что способствует повышению качества взаимодействия с приложением.
• Отладка: Включение трассировки стека при необходимости помогает разработчикам быстрее находить и устранять причины ошибок.

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

Наследование позволяет новому классу (производному) автоматически наследовать все методы и свойства другого класса (базового). В Python, наследование реализуется путем указания базового класса в круглых скобках при объявлении производного класса:

class BaseClass:
    pass

class DerivedClass(BaseClass):
    pass

Это позволяет DerivedClass наследовать все атрибуты и методы BaseClass, делая их доступными для использования или модификации в производном классе.

Переопределение методов — это возможность производного класса изменять реализацию методов, унаследованных от базового класса. Это особенно полезно, когда требуется специализировать или расширить поведение базового класса в соответствии с уникальными требованиями производного класса. Переопределение методов достигается путем объявления в производном классе метода с тем же именем, что и в базовом классе:

class BaseClass:
    def common_method(self):
        print("Метод базового класса")

class DerivedClass(BaseClass):
    def common_method(self):
        print("Переопределенный метод в производном классе")

В данном примере, вызов метода common_method на экземпляре DerivedClass приведет к выполнению переопределенной версии метода, а не версии из BaseClass.

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

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

В рамках разработки приложений, взаимодействующих с ChatGPT, возникает потребность в создании специализированных решений, адаптированных под конкретные каналы коммуникации или требования бизнеса. Один из таких кейсов — интеграция с веб-сокетами для обеспечения обмена сообщениями в реальном времени между пользователем и ИИ. Рассмотрим класс WSAnswerChatGPT, который расширяет базовый класс GetAnswerGPT, адаптируя его для работы через WebSocket.

# ai.utilities.py

import httpx
import openai
import markdown
from ai.gpt_exception import (OpenAIConnectionError, OpenAIResponseError,
                              UnhandledError, handle_exceptions)
from ai.gpt_query import GetAnswerGPT
from channels.generic.websocket import AsyncWebsocketConsumer
from django.conf import settings
from django.db.models import Model
from httpx_socks import AsyncProxyTransport
from openai import AsyncOpenAI
from telbot.loader import bot
from telbot.models import HistoryAI

ADMIN_ID = settings.TELEGRAM_ADMIN_ID


class WSAnswerChatGPT(GetAnswerGPT):
    MAX_TYPING_TIME = 3

    def __init__(self, channel_layer: AsyncWebsocketConsumer, room_group_name: str, user: Model, query_text: str, message_count: int) -> None:
        assist_prompt = self.init_model_prompt
        history_model = HistoryAI
        super().__init__(query_text, assist_prompt, user, history_model)
        self.channel_layer = channel_layer
        self.room_group_name = room_group_name
        self.message_count = message_count

    async def answer_from_ai(self) -> dict:
        """Основная логика."""
        try:
            await self.get_answer_chat_gpt()
        except Exception as err:
            self.return_text, *_ = await handle_exceptions(err, True)
            await self.handle_error(f'Ошибка в `GetAnswerGPT.answer_from_ai()`: {str(err)}')
        finally:
            if self.user.is_anonymous and self.message_count == 1:
                welcome_text = (
                    'С большой радостью приветствуем тебя! 🌟\n'
                    'Завершив процесс регистрации и авторизации, ты получишь доступ к уникальной возможности: '
                    'вести диалог с ИИ Superstar. Это гораздо больше, чем простые ответы на вопросы — это целый новый мир, '
                    'где ты можешь общаться, учиться и исследовать. Ваш диалог будет тщательно сохранён, '
                    'что позволит легко продолжить общение даже при переходе между страницами сайта. '
                )
                await self.send_chat_message(welcome_text)

    async def send_chat_message(self, message):
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat.message',
                'message': markdown.markdown(message),
                'username': 'Superstar',
            }
        )

    async def httpx_request_to_openai(self) -> None:
        """Делает запрос в OpenAI и выключает typing."""
        try:
            proxy_transport = AsyncProxyTransport.from_url(settings.SOCKS5)
            async with httpx.AsyncClient(transport=proxy_transport) as transport:
                client = AsyncOpenAI(
                    api_key=self.model.token,
                    http_client=transport,
                )
                stream = await client.chat.completions.create(
                    model=self.model.title,
                    messages=self.all_prompt,
                    stream=True,
                )
                first_chunk = True
                async for chunk in stream:
                    self.return_text += chunk.choices[0].delta.content or ""
                    await self.send_chunk_to_websocket(self.return_text, is_start=first_chunk, is_end=False)
                    if first_chunk:
                        first_chunk = False
                await self.send_chunk_to_websocket("", is_end=True)
                self.return_text_tokens = await self.num_tokens(self.return_text, 0)

        except openai.APIStatusError as http_err:
            raise OpenAIResponseError(f'`WSAnswerChatGPT`, ответ сервера был получен, но код состояния указывает на ошибку: {http_err}') from http_err
        except openai.APIConnectionError as req_err:
            raise OpenAIConnectionError(f'`WSAnswerChatGPT`, проблемы соединения: {req_err}') from req_err
        except Exception as error:
            raise UnhandledError(f'Необработанная ошибка в `WSAnswerChatGPT.httpx_request_to_openai()`: {error}') from error

    async def send_chunk_to_websocket(self, chunk, is_start=False, is_end=False):
        """Отправка части текста ответа через веб-сокет с указанием на статус части потока."""
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat.message',
                'message': chunk,
                'username': 'Superstar',
                'is_stream': True,
                'is_start': is_start,
                'is_end': is_end,
            }
        )

    async def handle_error(self, err):
        """Логирование ошибок."""
        error_message = f"Ошибка в блоке Сайт-ChatGPT:\n{err}"
        await bot.send_message(ADMIN_ID, error_message)
        
    @property
    def init_model_prompt(self):
        return (
            """
            Your name is Superstar and are an experienced senior software developer with a strong background in team leadership, mentoring all developers, and delivering high-quality software solutions to clients.
            Your primary language is Russian. When formatting the text, please use only Markdown format.
            """
        )

Особенности и нововведения класса

• WebSocket-специфичные свойства: channel_layer и room_group_name позволяют классу взаимодействовать с группами каналов WebSocket, обеспечивая рассылку сообщений всем подключенным клиентам.
• Переопределение конструктора: в конструкторе добавляются новые параметры, специфичные для работы с WebSocket, и вызывается конструктор базового класса с передачей ему необходимых параметров запроса и пользователя.
• Переопределение httpx_request_to_openai: теперь он отправляет ответа обратно пользователю через веб-сокет посредством стриминга.
• Метод answer_from_ai: является асинхронным и содержит логику получения ответа от ИИ, обработки исключений.
• Обработка ошибок: метод handle_error предназначен для логирования ошибок, возникших в процессе работы с ИИ, обеспечивая возможность отслеживания и устранения проблем. Передав флаг True для include_traceback мы имеем возможность получить трассировку ошибки внутри этого класса, без дублирования её в случае возникновения ошибки на нижнем уровне.
• Кастомный промпт: метод init_model_prompt возвращает промпт, специально адаптированный под контекст использования WebSocket, что позволяет уточнить характеристики диалога с ИИ.

Класс WSAnswerChatGPT демонстрирует, как можно адаптировать и расширять базовую функциональность взаимодействия с ChatGPT для специфических целей, в данном случае — для реализации интерактивного обмена сообщениями в реальном времени. Это подчеркивает гибкость подхода, основанного на принципах ООП. Реализация через наследование базового класса облегчает поддержку кода и внесение изменений, а также обеспечивает централизованное управление общими аспектами работы с ИИ, делая процесс разработки более эффективным и удобным.

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

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

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

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

1. Повышение производительности и эффективности: Асинхронные программы могут обрабатывать больше задач за единицу времени за счет неблокирующего выполнения, что особенно важно для серверных приложений, обрабатывающих большое количество запросов параллельно.
2. Улучшение отзывчивости приложения: В пользовательских интерфейсах асинхронность предотвращает "замерзание" приложения во время выполнения длительных операций, улучшая восприятие приложения пользователями.
3. Более эффективное использование ресурсов: Асинхронные приложения могут эффективнее использовать процессорное время и другие системные ресурсы, поскольку они не тратят время на ожидание завершения операций ввода-вывода.

Асинхронность в интеграции с Telegram: При разработке ботов для Telegram асинхронное программирование позволяет боту одновременно обрабатывать множество входящих сообщений и запросов без задержек в ответах пользователям. Это особенно критично в сценариях, когда бот взаимодействует с внешними API, такими как ChatGPT, где время ответа может быть переменным. Использование асинхронности в таких случаях гарантирует, что бот остается отзывчивым и способен обслуживать большое количество пользователей без потери качества обслуживания.

Интеграция ChatGPT, с платформами обмена сообщениями, открывает новые возможности для взаимодействия с пользователями. Класс TelegramAnswerGPT, наследующий функциональность базового класса GetAnswerGPT, является примером того, как можно адаптировать работу с ИИ под специфику Telegram ботов.

# telbot.chat_gpt.py

from ai.gpt_exception import handle_exceptions
from ai.gpt_query import GetAnswerGPT
from django.conf import settings
from django.db.models import Model
from telbot.models import HistoryAI
from telbot.service_message import send_message_to_chat
from telegram import ParseMode, Update
from telegram.ext import CallbackContext

ADMIN_ID = settings.TELEGRAM_ADMIN_ID


class TelegramAnswerGPT(GetAnswerGPT):

    def __init__(self, update: Update, _: CallbackContext, user: 'Model') -> None:
        query_text = update.effective_message.text
        assist_prompt = self.init_model_prompt
        history_model = HistoryAI
        self.chat_id = update.effective_chat.id
        self.message_id = update.message.message_id
        super().__init__(query_text, assist_prompt, user, history_model, self.chat_id, 0.3)

    async def answer_from_ai(self) -> dict:
        """Основная логика."""
        try:
            await self.get_answer_chat_gpt()
        except Exception as err:
            self.return_text, *_ = await handle_exceptions(err)
            await self.handle_error(f'Ошибка в `GetAnswerGPT.answer_from_ai()`: {str(err)}')
        finally:
            await self.reply_to_user()

    async def handle_error(self, err) -> None:
        """Логирование ошибок."""
        error_message = f"Ошибка в блоке Telegram-ChatGPT:\n{err}"
        send_message_to_chat(ADMIN_ID, error_message)

    async def reply_to_user(self) -> None:
        """Отправляет ответ пользователю."""
        await send_message_to_chat(self.chat_id, self.return_text, self.message_id, ParseMode.MARKDOWN)

    @property
    def init_model_prompt(self) -> str:
        return """
            You are named Superstar, an experienced senior software developer with a strong background in team leadership, mentoring all developers, and delivering high-quality software solutions to clients.
            Your primary language is Russian. When formatting the text, please only use this Markdown format:
            **bold text** _italic text_ [inline URL](http://www.example.com/) `inline fixed-width code` ```preformatted block code with fixed width```
            """

Подробнее рассмотрим ключевые аспекты этой интеграции и добавленную логику.

Конструктор и инициализация: В конструкторе TelegramAnswerGPT происходит инициализация с использованием объекта update от Telegram Bot API, который содержит информацию о входящем сообщении. Это позволяет извлекать текст запроса пользователя, идентификатор чата и другие параметры, необходимые для формирования и отправки ответа. Эта конструкция обеспечивает тесную интеграцию с Telegram API, позволяя классу TelegramAnswerGPT эффективно обрабатывать сообщения от пользователей.

Асинхронная обработка и отправка ответов: Метод answer_from_ai реализует асинхронную логику получения ответа от модели GPT и отправки его пользователю. Это включает в себя обработку возможных исключений и асинхронное взаимодействие с API Telegram для отправки ответных сообщений. Этот подход гарантирует, что бот может обслуживать множество запросов одновременно без существенных задержек, повышая качество пользовательского взаимодействия.

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

Отправка ответов пользователям: Через декоратор @sync_to_async метод reply_to_user, посредством кастомной функции send_message_to_chat, устойчивой к ошибкам в формате полученном от ChatGpt, обеспечивает асинхронную отправку ответов пользователям.

Кастомный промпт для модели: Свойство init_model_prompt демонстрирует, как можно персонализировать запросы в sistem к модели GPT, учитывая специфику контекста использования или предпочтений пользователя.

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

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

1. Специализация поведения: Наследование позволяет производным классам изменять или расширять поведение, унаследованное от базового класса. Это может включать переопределение методов для изменения их поведения, добавление новых методов для расширения функциональности или даже изменение внутренних данных класса для специфических нужд.
2. Модульность и переиспользование: Подклассы могут повторно использовать код базового класса, что способствует сокращению дублирования и упрощает поддержку кода. Это обеспечивает модульность и гибкость архитектуры приложения.
3. Полиморфизм: Наследование позволяет использовать полиморфизм, где объекты производных классов могут быть обработаны как объекты базового класса, что особенно полезно при работе с коллекциями разнотипных объектов, обладающих общим интерфейсом.

Рассмотрим классы WSAnswerChatGPT и TelegramAnswerGPT как примеры специализации базового класса GetAnswerGPT. Каждый из этих классов расширяет базовую функциональность для удовлетворения уникальных требований различных сред взаимодействия:

• WSAnswerChatGPT: Адаптирует базовый класс для интеграции с веб-сокетами, добавляя специфическую логику асинхронной обработки сообщений и управления состоянием сессии в реальном времени.
• TelegramAnswerGPT: Специализируется на взаимодействии с Telegram API, учитывая особенности форматирования сообщений и специфику асинхронной коммуникации через бот-платформу.

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

from ai.gpt_exception import handle_exceptions
from ai.gpt_query import GetAnswerGPT
from django.db.models import Model
from telbot.models import GptModels, ReminderAI


class ReminderGPT(GetAnswerGPT):

    def __init__(self, text: str, user: 'Model', chat_id: int) -> None:
        query_text = text
        assist_prompt = self.init_model_prompt
        history_model = ReminderAI
        self.chat_id = chat_id
        super().__init__(query_text, assist_prompt, user, history_model, self.chat_id, 0.1)

    async def transform(self) -> None:
        try:
            await self.get_answer_chat_gpt()
        except Exception as err:
            _, type_err, traceback_str = await handle_exceptions(err, True)
            raise type_err(f'Ошибка в процессе `ReminderGPT`: {err}{traceback_str}') from err
        return self.return_text

    async def get_prompt(self) -> None:
        self.all_prompt = [
            {'role': 'system', 'content': self.init_model_prompt},
            {'role': 'user', 'content': self.query_text}
        ]

    async def init_user_model(self) -> None:
        self.model = await GptModels.objects.filter(is_default=True).afirst()

    @property
    def init_model_prompt(self) -> str:
        return """
            ЧатGPT, прошу преобразовать следующий текст в формат:
            «дата {числовой формат} время {числовой формат}
            | количество минут за сколько оповестить до наступления дата+время {по умолчанию: 120}
            | повтор напоминания {по умолчанию: N, каждый день: D, каждую неделю: W, каждый месяц: M, каждый год: Y}
            | тело напоминания {исправить ошибки}».
            Ответ для примера: «20.11.2025 17:35|30|N|Запись к врачу»
            """

Ключевые особенности и добавленная логика:

• Конструктор и инициализация: Конструктор ReminderGPT принимает текст напоминания, информацию о пользователе и идентификатор чата как входные данные. Это обеспечивает необходимую информацию для формирования запроса к модели GPT и последующей обработки полученного ответа.
• Специализированный промпт: В init_model_prompt формируется кастомный промпт, предназначенный для инструкции модели GPT о преобразовании текста запроса в форматированное напоминание. Это демонстрирует, как можно настроить взаимодействие с ИИ для выполнения конкретных задач.
• Асинхронное преобразование запроса: Метод transform асинхронно обрабатывает запрос к GPT, пытаясь преобразовать пользовательский ввод в структурированное напоминание. В случае возникновения исключений, метод асинхронно обрабатывает их и передает для дальнейшей обработки.
• Гибкость в определении модели пользователя: Метод init_user_model демонстрирует, как можно динамически переопределять используемую модель GPT в зависимости от контекста, в данном случае выбирая модель по умолчанию.

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

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

Эта статья написана в учебных целях и демонстрирует, как можно использовать объектно-ориентированное программирование (ООП) и асинхронность для реализации взаимодействия приложений на Django с API OpenAI, включая ChatGPT. Показанные примеры методов и классов, таких как GetAnswerGPT и handle_exceptions, служат основой, которую каждый разработчик может адаптировать под свои конкретные задачи и потребности.

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

• top_p (Top P sampling), cтандартное значение 1: Этот параметр используется для управления разнообразием ответов путем выборки из топ-N% самых вероятных слов. Значение 1 означает, что рассматриваются все возможные слова, вне зависимости от их вероятности. Уменьшение значения приводит к тому, что выборка ограничивается узким набором наиболее вероятных слов.
• frequency_penalty (штраф за частотность), значение по умолчанию 0: Этот параметр уменьшает вероятность повторения слов и фраз в ответе. Чем выше значение, тем менее вероятно появление повторений. Значение 0 означает отсутствие штрафа за частотность.
• presence_penalty (штраф за присутствие), значение по умолчанию 0: Подобно frequency_penalty, но вместо уменьшения вероятности повторения уже использованных слов или фраз, он уменьшает вероятность повторного использования тем или идеей. Это может способствовать генерации более разнообразного и творческого контента.
• temperature, значение по умолчанию обычно составляет 1. Этот параметр контролирует степень случайности в ответах. Значение 0 приводит к тому, что модель будет повторять наиболее вероятные ответы, делая вывод детерминированным. При увеличении значения до 1 вывод становится более разнообразным и непредсказуемым.

В классах GetAnswerGPT и WSAnswerChatGPT я намеренно использовал разные подходы взаимодействия с API OpenAl, чтоб наглядно продемонстрировать различные методы работы с API OpenAI. Это подчеркивает, что нет единственного "правильного" пути интеграции, и разработчики могут выбирать подход, наилучшим образом соответствующий их целям и контексту применения. Но удобнее работать конечно же через библиотеку openai-python. 

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

Цель статьи — описать принципы ООП, методы отладки ошибок и демонстрация того, как можно эффективно внедрять взаимодействие с OpenAI API в проекты на Django, подчеркивая важность адаптации и настройки подходов под уникальные требования каждого конкретного проекта.