Merge branch 'feature/pipeline-subsystem-v1' of github.com:ls1intum/P…

…yris into feature/pipeline-subsystem-v1

app/pipeline/__init__.py

@@ -1 +1,2 @@
-from pipeline.pipeline import AbstractPipeline
+from pipeline.pipeline import Pipeline
+from pipeline.chat.simple_chat_pipeline import SimpleChatPipeline

app/pipeline/chat/chat_pipeline.py

@@ -0,0 +1,21 @@
+from abc import ABC, abstractmethod
+
+from domain import IrisMessage
+from pipeline import Pipeline
+
+
+class ProgrammingExerciseTutorChatPipeline(Pipeline, ABC):
+    """
+    Abstract class for the programming exercise tutor chat pipeline implementations.
+    This class defines the signature of all implementations of this Iris feature.
+    """
+
+    def __call__(self, query: IrisMessage, **kwargs) -> IrisMessage:
+        return self._run(query)
+
+    @abstractmethod
+    def _run(self, query: IrisMessage) -> IrisMessage:
+        """
+        Runs the pipeline and returns the response message.
+        """
+        raise NotImplementedError

app/pipeline/chat/simple_chat_pipeline.py

@@ -5,26 +5,23 @@
 
 from domain import IrisMessage, IrisMessageRole
 from llm.langchain import IrisLangchainChatModel
-from pipeline import AbstractPipeline
+from pipeline.chat.chat_pipeline import ProgrammingExerciseTutorChatPipeline
 
 
-class SimpleChatPipeline(AbstractPipeline):
+class SimpleChatPipeline(ProgrammingExerciseTutorChatPipeline):
     """A simple chat pipeline that uses our custom langchain chat model for our own request handler"""
 
     llm: IrisLangchainChatModel
     pipeline: Runnable
 
-    def __init__(self, llm: IrisLangchainChatModel, name=None):
+    def __init__(self, llm: IrisLangchainChatModel):
         self.llm = llm
         self.pipeline = {"query": itemgetter("query")} | llm | StrOutputParser()
-        super().__init__(name=name)
+        super().__init__(implementation_id="simple_chat_pipeline")
 
-    def __call__(self, query: IrisMessage, **kwargs) -> IrisMessage:
+    def _run(self, query: IrisMessage) -> IrisMessage:
         """
-        Runs the pipeline and is intended to be called by __call__
-        :param query: The query
-        :param kwargs: keyword arguments
-        :return: IrisMessage
+        Gets a response from the langchain chat model
         """
         if query is None:
             raise ValueError("IrisMessage must not be None")

app/pipeline/chat/tutor_chat_pipeline.py

@@ -7,22 +7,19 @@
 
 from domain import IrisMessage, IrisMessageRole
 from llm.langchain import IrisLangchainChatModel
-from pipeline import AbstractPipeline
-
+from pipeline.chat.chat_pipeline import ProgrammingExerciseTutorChatPipeline
 
 logger = logging.getLogger(__name__)
 
 
-class TutorChatPipeline(AbstractPipeline):
+class TutorChatPipelineReferenceImpl(ProgrammingExerciseTutorChatPipeline):
     """Tutor chat pipeline that answers exercises related questions from students."""
 
     llm: IrisLangchainChatModel
     pipeline: Runnable
-    prompt_str: str
-    prompt: ChatPromptTemplate
 
-    def __init__(self, llm: IrisLangchainChatModel, name=None):
-        super().__init__(name=name)
+    def __init__(self, llm: IrisLangchainChatModel):
+        super().__init__(implementation_id="tutor_chat_pipeline_reference_impl")
         # Set the langchain chat model
         self.llm = llm
         # Load the prompt from a file
@@ -31,21 +28,20 @@ def __init__(self, llm: IrisLangchainChatModel, name=None):
             os.path.join(dirname, "../prompts/iris_tutor_chat_prompt.txt", "r")
         ) as file:
             logger.debug("Loading tutor chat prompt...")
-            self.prompt_str = file.read()
+            prompt_str = file.read()
         # Create the prompt
-        self.prompt = ChatPromptTemplate.from_messages(
+        prompt = ChatPromptTemplate.from_messages(
             [
-                SystemMessagePromptTemplate.from_template(self.prompt_str),
+                SystemMessagePromptTemplate.from_template(prompt_str),
             ]
         )
         # Create the pipeline
-        self.pipeline = self.prompt | llm | StrOutputParser()
+        self.pipeline = prompt | llm | StrOutputParser()
 
-    def __call__(self, query: IrisMessage, **kwargs) -> IrisMessage:
+    def _run(self, query: IrisMessage) -> IrisMessage:
         """
         Runs the pipeline
             :param query: The query
-            :param kwargs: keyword arguments
             :return: IrisMessage
         """
         if query is None:

app/pipeline/pipeline.py

@@ -1,21 +1,17 @@
-from abc import ABCMeta, abstractmethod
+from abc import ABCMeta
 
-from domain import IrisMessage
 
-
-class AbstractPipeline(metaclass=ABCMeta):
+class Pipeline(metaclass=ABCMeta):
     """Abstract class for all pipelines"""
 
-    name: str
+    implementation_id: str
 
-    def __init__(self, name=None):
-        self.name = name
+    def __init__(self, implementation_id=None):
+        self.implementation_id = implementation_id
 
-    @abstractmethod
-    def __call__(self, **kwargs) -> IrisMessage:
+    def __call__(self, **kwargs):
         """
-        Runs the pipeline and is intended to be called by __call__
-        :param kwargs: keyword arguments
-        :return: IrisMessage
+        Extracts the required parameters from the kwargs runs the pipeline.
         """
         raise NotImplementedError
+