# pylint: disable=W0613,C0115
"""All models for the OpenAI Function Calling API app."""
import logging
from functools import cached_property
from typing import Any, Optional, Union
from django.db import models
from django.db.utils import IntegrityError
from django.http import HttpRequest
from rest_framework import serializers
from smarter.apps.account.models import MetaDataWithOwnershipModel
from smarter.apps.chatbot.models import ChatBot, get_cached_chatbot_by_request
from smarter.apps.plugin.models import PluginMeta
from smarter.common.conf import smarter_settings
from smarter.common.const import SMARTER_CHAT_SESSION_KEY_NAME
from smarter.common.exceptions import SmarterConfigurationError, SmarterValueError
from smarter.lib import json
from smarter.lib.cache import lazy_cache as cache
from smarter.lib.django import waffle
from smarter.lib.django.models import TimestampedModel
from smarter.lib.django.request import SmarterRequestMixin
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
[docs]
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return waffle.switch_is_active(SmarterWaffleSwitches.PROMPT_LOGGING)
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
[docs]
class Chat(MetaDataWithOwnershipModel):
"""Chat model."""
class Meta:
verbose_name_plural = "Chats"
unique_together = (SMARTER_CHAT_SESSION_KEY_NAME, "url")
session_key = models.CharField(max_length=255, blank=False, null=False, unique=True)
chatbot = models.ForeignKey(ChatBot, on_delete=models.CASCADE, blank=False, null=False)
ip_address = models.GenericIPAddressField(blank=False, null=False)
user_agent = models.CharField(max_length=255, blank=False, null=False)
url = models.URLField(blank=False, null=False)
def __str__(self):
# pylint: disable=E1136
return f"{self.id} - {self.ip_address} - {self.url}" # type: ignore[return]
[docs]
def delete(self, *args, **kwargs):
if self.session_key:
cache.delete(self.session_key)
super().delete(*args, **kwargs)
[docs]
class ChatHistory(TimestampedModel):
"""Chat history model."""
class Meta:
verbose_name_plural = "Chat History"
chat = models.ForeignKey(Chat, on_delete=models.CASCADE)
request = models.JSONField(
blank=True,
null=True,
encoder=json.SmarterJSONEncoder,
)
response = models.JSONField(
blank=True,
null=True,
encoder=json.SmarterJSONEncoder,
)
messages = models.JSONField(
blank=True,
null=True,
encoder=json.SmarterJSONEncoder,
)
def __str__(self):
return f"{self.chat.id}" # type: ignore[return]
@property
def chat_history(self) -> list[dict]:
"""
Used by the Reactapp (via ChatConfigView) to display the chat history.
"""
history = self.messages if self.messages else self.request.get("messages", []) if self.request else []
# response = self.response.get("choices", []) if self.response else []
# response = response[0] if response else {}
# response = response.get("message", {})
# history.append(response)
return history
[docs]
class ChatPluginUsage(TimestampedModel):
"""Plugin selection history model."""
class Meta:
verbose_name_plural = "Plugin Usage"
chat = models.ForeignKey(Chat, on_delete=models.CASCADE)
plugin = models.ForeignKey(PluginMeta, on_delete=models.CASCADE)
input_text = models.TextField(blank=True, null=True)
def __str__(self):
return f"{self.chat.id} - {self.plugin.name}" # type: ignore[return]
# --------------------------------------------------------------------------------
# Serializers
# --------------------------------------------------------------------------------
[docs]
class ChatSerializer(serializers.ModelSerializer):
class Meta:
model = Chat
fields = "__all__"
[docs]
class ChatPluginUsageSerializer(serializers.ModelSerializer):
"""Serializer for the ChatPluginUsage model."""
chat = ChatSerializer(read_only=True)
class Meta:
model = ChatPluginUsage
fields = "__all__"
[docs]
class ChatHelper(SmarterRequestMixin):
"""
Helper class for working with :class:`Chat` objects.
This class provides methods for creating and retrieving :class:`Chat` objects,
as well as managing the cache for chat sessions. It is designed to simplify
the process of interacting with chat-related data and to ensure consistent
handling of chat sessions, chatbots, and associated metadata.
**Features**
- Abstracts the logic for creating and retrieving chat sessions.
- Manages caching of chat objects to improve performance and reduce database queries.
- Provides access to related chat history, tool calls, and plugin usage.
- Integrates with Django's request and session handling.
- Ensures that chat sessions are always associated with a valid :class:`ChatBot` and :class:`Account`.
**Usage**
Typically, this class is instantiated with a Django :class:`HttpRequest` object and a session key.
Optionally, a :class:`ChatBot` instance can be provided to associate the chat session with a specific chatbot.
Example
-------
.. code-block:: python
helper = ChatHelper(request, session_key)
if helper.ready:
chat = helper.chat
chatbot = helper.chatbot
history = helper.history
:param request: The Django HttpRequest object for the current session.
:type request: django.http.HttpRequest
:param session_key: The session key identifying the chat session.
:type session_key: Optional[str]
:param chatbot: An optional ChatBot instance to associate with the chat session.
:type chatbot: Optional[ChatBot]
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:raises SmarterValueError: If neither a session key nor a ChatBot instance is provided.
:raises SmarterConfigurationError: If there is an error creating a new Chat object.
.. note::
This class is intended for internal use within the Smarter platform and
should not be used directly in user-facing code without proper validation.
.. todo::
- Remove the session_key parameter and rely solely on the ChatBot instance for chat session management.
.. seealso::
- :class:`smarter.apps.chatbot.models.ChatBot`
- :class:`smarter.apps.account.models.Account`
- :class:`smarter.apps.chat.models.Chat`
- :class:`smarter.lib.django.request.SmarterRequestMixin`
"""
_chat: Optional[Chat] = None
_chatbot: Optional[ChatBot] = None
_chat_tool_call: Optional[Union[models.QuerySet, list]] = None
_chat_plugin_usage: Optional[Union[models.QuerySet, list]] = None
_history: Optional[dict] = None
[docs]
def __init__(
self, request: HttpRequest, session_key: Optional[str], *args, chatbot: Optional[ChatBot] = None, **kwargs
) -> None:
"""
Initialize the ChatHelper instance.
:param request: The Django HttpRequest object for the current session.
:type request: django.http.HttpRequest
:param session_key: The session key identifying the chat session.
:type session_key: Optional[str]
:param chatbot: An optional ChatBot instance to associate with the chat session.
:type chatbot: Optional[ChatBot]
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:raises SmarterValueError: If neither a session key nor a ChatBot instance is provided.
:raises SmarterConfigurationError: If there is an error creating a new Chat object.
"""
logger.debug(
"%s.__init__() - received request: %s session_key: %s, chatbot: %s",
self.formatted_class_name,
self.smarter_build_absolute_uri(request),
session_key,
chatbot,
)
if not request:
raise SmarterValueError(f"{self.formatted_class_name} request object is required.")
super().__init__(request, session_key=session_key, **kwargs)
self._chat = None
self._chatbot = chatbot
self._chat_tool_call = None
self._chat_plugin_usage = None
self._history = None
if not session_key and not chatbot:
raise SmarterValueError(
f"{self.formatted_class_name} either a session_key or a ChatBot instance is required"
)
if chatbot:
logger.debug("%s.__init__() received ChatBot instance: %s", self.formatted_class_name, chatbot)
logger.debug(
"%s.__init__() - reinitializing account from chatbot.account: %s",
self.formatted_class_name,
self.account,
)
self.user_profile = chatbot.user_profile
if session_key:
self._session_key = session_key
logger.debug(
"%s.__init__() - setting session_key to %s from session_key parameter",
self.formatted_class_name,
self._session_key,
)
if self.session_key:
logger.debug("%s.__init__() received session_key: %s", self.formatted_class_name, session_key)
self._chat = self.get_cached_chat()
logger.debug(
"%s.__init__() - %s with session_key: %s, chat: %s",
self.formatted_class_name,
"is ready" if self.ready else "is not ready",
self.session_key,
self._chat,
)
def __str__(self):
return self.session_key
@property
def ready(self) -> bool:
"""
Check if the ChatHelper is ready to use.
This property returns ``True`` if the chat instance is available and all required
attributes are set, otherwise returns ``False``. It is useful for determining
whether the ChatHelper is fully initialized and ready for chat operations.
:returns: ``True`` if the ChatHelper is ready to use, otherwise ``False``.
:rtype: bool
"""
return bool(super().ready) and bool(self._session_key) and bool(self._chat) and bool(self._chatbot)
[docs]
def to_json(self) -> dict[str, Any]:
"""
Convert the ChatHelper instance to a JSON serializable dictionary.
This method returns a dictionary representation of the ChatHelper instance,
including key metadata and related objects such as the chat, chatbot, chat history,
and a unique client string.
:returns: A dictionary containing the serialized state of the ChatHelper.
:rtype: dict[str, Any]
"""
return self.sorted_dict(
{
**super().to_json(),
"ready": self.ready,
"session_key": self.session_key,
"chat": self.chat.id if self.chat else None, # type: ignore[return]
"chatbot": self.chatbot.id if self.chatbot else None, # type: ignore[return]
"history": self.history,
"unique_client_string": self.unique_client_string,
}
)
@property
def chat(self):
"""
Get the chat instance for the current request.
:returns: The Chat instance associated with the current session.
:rtype: Chat
"""
return self._chat
@property
def chatbot(self):
"""
Returns a lazy instance of the ChatBot.
Examples
--------
- ``https://hr.3141-5926-5359.alpha.api.example.com/chatbot/``
returns ``ChatBot(name='hr', account=Account(...))``
:returns: The ChatBot instance.
:rtype: ChatBot
"""
if self._chatbot:
return self._chatbot
self._chatbot = get_cached_chatbot_by_request(request=self.smarter_request)
@property
def chat_history(self) -> Union[models.QuerySet, list]:
"""
Get the most recent chat history for the current chat session.
:returns: The most recent ChatHistory instance's chat_history field, or an empty list if none found.
:rtype: Union[models.QuerySet, list]
"""
rec = ChatHistory.objects.filter(chat=self.chat).order_by("-created_at").first()
return rec.chat_history if rec else []
@property
def chat_tool_call(self) -> Union[models.QuerySet, list]:
"""
Get the most recent chat tool call history for the current chat session.
:returns: A queryset of ChatToolCall instances for the current chat session, ordered by creation date.
:rtype: Union[models.QuerySet, list]
"""
if self._chat_tool_call:
return self._chat_tool_call
self._chat_tool_call = ChatToolCall.objects.filter(chat=self.chat).order_by("-created_at") or []
return self._chat_tool_call
@property
def chat_plugin_usage(self) -> Union[models.QuerySet, list]:
"""
Get the most recent chat plugin usage history for the current chat session.
:returns: A queryset of ChatPluginUsage instances for the current chat session, ordered by creation date.
:rtype: Union[models.QuerySet, list]
"""
if self._chat_plugin_usage:
return self._chat_plugin_usage
self._chat_plugin_usage = ChatPluginUsage.objects.filter(chat=self.chat).order_by("-created_at") or []
return self._chat_plugin_usage
@property
def history(self) -> dict:
"""
Serialize the most recent logged history output for the chat session.
:returns: A dictionary containing serialized chat, chat history, tool calls, and plugin usage.
:rtype: dict
"""
if self._history:
return self._history
chat_serializer = ChatSerializer(self.chat)
chat_tool_call_serializer = ChatToolCallSerializer(self.chat_tool_call, many=True)
chat_plugin_usage_serializer = ChatPluginUsageSerializer(self.chat_plugin_usage, many=True)
self._history = {
"chat": chat_serializer.data,
"chat_history": self.chat_history,
"chat_tool_call_history": chat_tool_call_serializer.data,
"chat_plugin_usage_history": chat_plugin_usage_serializer.data,
# these two will be added upstream.
"chatbot_request_history": None, # ChatBotRequests
}
return self._history
[docs]
def get_cached_chat(self) -> Optional[Chat]:
"""
Get the chat instance for the current request.
This method retrieves the Chat instance associated with the current session key
from the cache. If the Chat instance is not found in the cache, it attempts to
retrieve it from the database. If it still cannot be found, a new Chat instance
is created using the provided ChatBot and request metadata.
:returns: The Chat instance associated with the current session, or ``None`` if not found.
:rtype: Optional[Chat]
"""
if not self.smarter_request:
logger.error("%s - request object is required for ChatHelper.", self.formatted_class_name)
return None
chat: Chat = cache.get(self.session_key) # type: ignore[assignment]
if chat:
logger.debug(
"%s - retrieved cached Chat: %s session_key: %s", self.formatted_class_name, chat, chat.session_key
)
return chat
if self.session_key:
try:
chat = Chat.objects.get(session_key=self.session_key)
logger.debug(
"%s - retrieved Chat instance: %s session_key: %s",
self.formatted_class_name,
chat,
chat.session_key,
)
except Chat.DoesNotExist:
pass
if not chat:
if not self.chatbot:
raise SmarterValueError(
f"{self.formatted_class_name} ChatBot instance is required for creating a Chat object."
)
try:
# modify the unit test server URL
# to a more Django friendly URL.
django_friendly_url = self.url or ""
django_friendly_url = django_friendly_url.replace("http://testserver/", "http://testserver.local/")
chat = Chat.objects.create(
session_key=self.session_key,
user_profile=self.user_profile,
chatbot=self.chatbot,
ip_address=self.ip_address,
user_agent=self.user_agent,
url=django_friendly_url,
)
except IntegrityError as e:
raise SmarterConfigurationError(f"{self.formatted_class_name} - IntegrityError: {str(e)}") from e
cache.set(key=self.session_key, value=chat, timeout=smarter_settings.chat_cache_expiration or 300)
if waffle.switch_is_active(SmarterWaffleSwitches.CACHE_LOGGING):
logger.debug(
"%s - cached chat instance: %s session_key: %s", self.formatted_class_name, chat, chat.session_key
)
if not chat.chatbot:
raise ValueError(f"{self.formatted_class_name} ChatBot instance is required for Chat object.")
return chat