Source code for council.contexts._message_collection

import abc
from typing import Callable, Iterable, Optional

from more_itertools import first
from typing_extensions import TypeGuard

from ._chat_message import ChatMessage, ChatMessageKind
from ..utils import Option


[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): 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)