import abc
from typing import Callable, Iterable, Optional
from more_itertools import first
from typing_extensions import TypeGuard
from ..utils import Option
from ._chat_message import ChatMessage, ChatMessageKind
[docs]
class MessageCollection(abc.ABC):
"""
Base class to manage collection of :class:`ChatMessage`
"""
@property
@abc.abstractmethod
def messages(self) -> Iterable[ChatMessage]:
"""
Iterates over the collection of messages, in the order they appear in the context
Returns:
Iterable[ChatMessage]
"""
pass
@property
@abc.abstractmethod
def reversed(self) -> Iterable[ChatMessage]:
"""
Iterates over the collection of messages, in reversed order
Returns:
Iterable[ChatMessage]
"""
pass
@property
def last_message(self) -> Optional[ChatMessage]:
"""
Returns the last message, if any, otherwise `None`.
Returns:
Optional[ChatMessage]:
"""
return first(self.reversed, None)
@property
def try_last_message(self) -> Option[ChatMessage]:
"""
Returns the last message, wrapped into an :class:`Option`
Returns:
Option[ChatMessage]:
"""
return Option(self.last_message)
@property
def last_user_message(self) -> Optional[ChatMessage]:
"""
Returns the last message of kind :attr:`ChatMessageKind.User`, if any, otherwise `None`
Returns:
Optional[ChatMessage]:
"""
return self._last_message_filter(self.message_kind_predicate(ChatMessageKind.User))
@property
def try_last_user_message(self) -> Option[ChatMessage]:
"""
Returns the last message of kind :attr:`ChatMessageKind.User, wrapped into an :class:`Option`
Returns:
Option[ChatMessage]:
"""
return Option(self.last_user_message)
@property
def last_agent_message(self) -> Optional[ChatMessage]:
"""
Returns the last message of kind :attr:`ChatMessageKind.Agent`, if any, otherwise `None`
Returns:
Optional[ChatMessage]:
"""
return self._last_message_filter(self.message_kind_predicate(ChatMessageKind.Agent))
@property
def try_last_agent_message(self) -> Option[ChatMessage]:
"""
Returns the last message of kind :attr:`ChatMessageKind.Agent` wrapped into an :class:`Option`
Returns:
Option[ChatMessage]
"""
return Option(self.last_agent_message)
[docs]
def last_message_from_skill(self, skill_name: str) -> Optional[ChatMessage]:
"""
Returns the last message generated by a given skill, if any, otherwise `None`
Parameters:
skill_name(str): Name of the skill
Returns:
Optional[ChatMessage]:
"""
def predicate(message: ChatMessage) -> bool:
return message.is_of_kind(ChatMessageKind.Skill) and message.is_from_source(skill_name)
return self._last_message_filter(predicate)
[docs]
def try_last_message_from_skill(self, skill_name: str) -> Option[ChatMessage]:
"""
Returns the last message generated by a given skill, wrapped into an :class:`Option`
Parameters:
skill_name(str): Name of the skill
Returns:
Option[ChatMessage]:
"""
return Option(self.last_message_from_skill(skill_name))
def _last_message_filter(self, predicate: Callable[[ChatMessage], bool]) -> Optional[ChatMessage]:
def typeguard_predicate(message: ChatMessage) -> TypeGuard[Optional[ChatMessage]]:
return isinstance(message, ChatMessage) and predicate(message)
return first(filter(typeguard_predicate, self.reversed), None)
@staticmethod
def message_kind_predicate(kind: ChatMessageKind) -> Callable[[ChatMessage], bool]:
return lambda m: m.is_of_kind(kind)