Feat/llm responses

chatsky/conditions/__init__.py

@@ -11,3 +11,4 @@
 )
 from chatsky.conditions.slots import SlotsExtracted
 from chatsky.conditions.service import ServiceFinished
+from chatsky.conditions.llm import LLMCondition

chatsky/conditions/llm.py

@@ -0,0 +1,73 @@
+"""
+LLM Conditions
+--------------
+This module provides LLM-based conditions.
+"""
+
+from pydantic import Field
+
+from chatsky.core import BaseCondition, Context
+from chatsky.core.script_function import AnyResponse
+from chatsky.llm.methods import BaseMethod
+from chatsky.llm.utils import context_to_history, message_to_langchain
+from chatsky.llm.filters import BaseHistoryFilter, DefaultFilter
+
+
+class LLMCondition(BaseCondition):
+    """
+    LLM-based condition.
+    Uses prompt to produce result from model and evaluates the result using given method.
+    """
+
+    model_name: str
+    """
+    Key of the model in the :py:attr:`~chatsky.core.pipeline.Pipeline.models` dictionary.
+    """
+    prompt: AnyResponse = Field(default="", validate_default=True)
+    """
+    Condition prompt.
+    """
+    history: int = 1
+    """
+    Number of dialogue turns to keep in history. `-1` for full history.
+    """
+    filter_func: BaseHistoryFilter = Field(default_factory=DefaultFilter)
+    """
+    Filter function to filter messages that will go the models context.
+    """
+    max_size: int = 1000
+    """
+    Maximum size of any message in chat in symbols. If exceed the limit will raise ValueError.
+    """
+    method: BaseMethod
+    """
+    Method that takes model's output and returns boolean.
+    """
+
+    async def call(self, ctx: Context) -> bool:
+        model = ctx.pipeline.models[self.model_name]
+
+        if model.system_prompt == "":
+            history_messages = []
+        else:
+            history_messages = [await message_to_langchain(model.system_prompt, ctx=ctx, source="system")]
+
+        if not (self.history == 0 or len(ctx.responses) == 0 or len(ctx.requests) == 0):
+            history_messages.extend(
+                await context_to_history(
+                    ctx=ctx,
+                    length=self.history,
+                    filter_func=self.filter_func,
+                    model_name=self.model_name,
+                    max_size=self.max_size,
+                )
+            )
+
+        history_messages.append(
+            await message_to_langchain(self.prompt, ctx=ctx, source="system", max_size=self.max_size)
+        )
+        history_messages.append(
+            await message_to_langchain(ctx.last_request, ctx=ctx, source="human", max_size=self.max_size)
+        )
+
+        return await model.condition(history_messages, self.method)

chatsky/core/pipeline.py

@@ -29,6 +29,7 @@
 from chatsky.core.service.actor import Actor
 from chatsky.core.node_label import AbsoluteNodeLabel, AbsoluteNodeLabelInitTypes
 from chatsky.core.script_parsing import JSONImporter, Path
+from chatsky.llm.llm_api import LLM_API
 
 logger = logging.getLogger(__name__)
 
@@ -78,6 +79,10 @@ class Pipeline(BaseModel, extra="forbid", arbitrary_types_allowed=True):
     """
     Slots configuration.
     """
+    models: Dict[str, LLM_API] = Field(default_factory=dict)
+    """
+    LLM models.
+    """
     messenger_interface: MessengerInterface = Field(default_factory=CLIMessengerInterface)
     """
     A `MessengerInterface` instance for this pipeline.
@@ -116,6 +121,7 @@ def __init__(
         *,
         default_priority: float = None,
         slots: GroupSlot = None,
+        models: dict = None,
         messenger_interface: MessengerInterface = None,
         context_storage: Union[DBContextStorage, dict] = None,
         pre_services: ServiceGroupInitTypes = None,
@@ -133,6 +139,7 @@ def __init__(
             "fallback_label": fallback_label,
             "default_priority": default_priority,
             "slots": slots,
+            "models": models,
             "messenger_interface": messenger_interface,
             "context_storage": context_storage,
             "pre_services": pre_services,

chatsky/llm/__init__.py

@@ -0,0 +1,3 @@
+from chatsky.llm.filters import BaseHistoryFilter, FromModel, IsImportant
+from chatsky.llm.methods import BaseMethod, LogProb, Contains
+from chatsky.llm.llm_api import LLM_API

chatsky/llm/_langchain_imports.py

@@ -0,0 +1,25 @@
+from typing import Any
+
+try:
+    from langchain_core.output_parsers import StrOutputParser
+    from langchain_core.language_models.chat_models import BaseChatModel
+    from langchain_core.messages.base import BaseMessage
+    from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
+    from langchain_core.outputs.llm_result import LLMResult
+
+    langchain_available = True
+except ImportError:  # pragma: no cover
+    StrOutputParser = Any
+    BaseChatModel = Any
+    BaseMessage = Any
+    HumanMessage = Any
+    SystemMessage = Any
+    AIMessage = Any
+    LLMResult = Any
+
+    langchain_available = False
+
+
+def check_langchain_available():  # pragma: no cover
+    if not langchain_available:
+        raise ImportError("Langchain is not available. Please install it with `pip install chatsky[llm]`.")

chatsky/llm/filters.py

@@ -0,0 +1,107 @@
+"""
+Filters
+---------
+This module contains a collection of basic functions for history filtering to avoid cluttering LLMs context window.
+"""
+
+import abc
+from enum import Enum
+from logging import Logger
+from typing import Union
+
+from pydantic import BaseModel
+
+from chatsky.core.message import Message
+from chatsky.core.context import Context
+
+logger = Logger(name=__name__)
+
+
+class Return(Enum):
+    Request = 1
+    Response = 2
+    Turn = 3
+
+
+class BaseHistoryFilter(BaseModel, abc.ABC):
+    """
+    Base class for all message history filters.
+    """
+
+    @abc.abstractmethod
+    def call(self, ctx: Context, request: Message, response: Message, model_name: str) -> Union[Return, int]:
+        """
+        :param ctx: Context object.
+        :param request: Request message.
+        :param response: Response message.
+        :param model_name: Name of the model in the Pipeline.models.
+
+        :return: Instance of Return enum or a corresponding int value.
+        """
+        raise NotImplementedError
+
+    def __call__(self, ctx: Context, request: Message, response: Message, model_name: str):
+        """
+        :param ctx: Context object.
+        :param request: Request message.
+        :param response: Response message.
+        :param model_name: Name of the model in the Pipeline.models.
+        """
+        try:
+            result = self.call(ctx, request, response, model_name)
+
+            if isinstance(result, int):
+                result = Return(result)
+        except Exception as exc:
+            logger.warning(exc)
+            return []
+        if result == Return.Turn:
+            return [request, response]
+        if result == Return.Response:
+            return [response]
+        if result == Return.Request:
+            return [request]
+        return []
+
+
+class MessageFilter(BaseHistoryFilter):
+    @abc.abstractmethod
+    def call(self, ctx, message, model_name) -> bool:
+        raise NotImplementedError
+
+    def __call__(self, ctx, request, response, model_name):
+        return (
+            int(self.call(ctx, request, model_name)) * Return.Request.value
+            + int(self.call(ctx, response, model_name)) * Return.Response.value
+        )
+
+
+class DefaultFilter(BaseHistoryFilter):
+    def call(self, ctx: Context, request: Message, response: Message, model_name: str) -> Union[Return, int]:
+        return Return.Turn
+
+
+class IsImportant(MessageFilter):
+    """
+    Filter that checks if the "important" field in a Message.misc is True.
+    """
+
+    def call(self, ctx: Context, message: Message, model_name: str) -> bool:
+        if message is not None and message.misc is not None and message.misc.get("important", None):
+            return True
+        return False
+
+
+class FromModel(MessageFilter):
+    """
+    Filter that checks if the message was sent by the model.
+    """
+
+    def call(self, ctx: Context, message: Message, model_name: str) -> bool:
+        if (
+            message is not None
+            and message.annotations is not None
+            and message.annotations.get("__generated_by_model__") == model_name
+        ):
+            return True
+        return False

chatsky/llm/llm_api.py

@@ -0,0 +1,77 @@
+"""
+LLM responses.
+--------------
+Wrapper around langchain.
+"""
+
+from typing import Union, Type, Optional
+from pydantic import BaseModel
+import logging
+from chatsky.core.message import Message
+from chatsky.llm.methods import BaseMethod
+from chatsky.llm._langchain_imports import StrOutputParser, BaseChatModel, BaseMessage, check_langchain_available
+
+
+class LLM_API:
+    """
+    This class acts as a wrapper for all LLMs from langchain
+    and handles message exchange between remote model and chatsky classes.
+    """
+
+    def __init__(
+        self,
+        model: BaseChatModel,
+        system_prompt: Optional[str] = "",
+    ) -> None:
+        """
+        :param model: Model object
+        :param system_prompt: System prompt for the model
+        """
+        check_langchain_available()
+        self.model: BaseChatModel = model
+        self.parser = StrOutputParser()
+        self.system_prompt = Message(text=system_prompt)
+
+    async def respond(
+        self,
+        history: list[BaseMessage],
+        message_schema: Union[None, Type[Message], Type[BaseModel]] = None,
+    ) -> Message:
+        """
+        Process and structure the model's response based on the provided schema.
+
+        :param history: List of previous messages in the conversation
+        :param message_schema: Schema for structuring the output, defaults to None
+        :return: Processed model response
+
+        :raises ValueError: If message_schema is not None, Message, or BaseModel
+        """
+
+        if message_schema is None:
+            result = await self.parser.ainvoke(await self.model.ainvoke(history))
+            return Message(text=result)
+        elif issubclass(message_schema, Message):
+            # Case if the message_schema describes Message structure
+            structured_model = self.model.with_structured_output(message_schema, method="json_mode")
+            model_result = await structured_model.ainvoke(history)
+            logging.debug(f"Generated response: {model_result}")
+            return Message.model_validate(model_result)
+        elif issubclass(message_schema, BaseModel):
+            # Case if the message_schema describes Message.text structure
+            structured_model = self.model.with_structured_output(message_schema)
+            model_result = await structured_model.ainvoke(history)
+            return Message(text=message_schema.model_validate(model_result).model_dump_json())
+        else:
+            raise ValueError
+
+    async def condition(self, history: list[BaseMessage], method: BaseMethod) -> bool:
+        """
+        Execute a conditional method on the conversation history.
+
+        :param history: List of previous messages in the conversation
+        :param method: Method to evaluate the condition
+
+        :return: Boolean result of the condition evaluation
+        """
+        result = await method(history, await self.model.agenerate([history], logprobs=True, top_logprobs=10))
+        return result