Add Text Exercise Chat Pipeline

app/domain/__init__.py

@@ -14,3 +14,4 @@
 )
 from .pyris_message import PyrisMessage, IrisMessageRole
 from app.domain.data import image_message_content_dto
+from app.domain.feature_dto import FeatureDTO

app/domain/data/text_exercise_dto.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from app.domain.data.course_dto import CourseDTO
+
+
+class TextExerciseDTO(BaseModel):
+    id: int
+    title: str
+    course: CourseDTO
+    problem_statement: str = Field(alias="problemStatement")
+    start_date: Optional[datetime] = Field(alias="startDate", default=None)
+    end_date: Optional[datetime] = Field(alias="endDate", default=None)

app/domain/feature_dto.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class FeatureDTO(BaseModel):
+    id: str
+    name: str
+    description: str

app/domain/status/text_exercise_chat_status_update_dto.py

@@ -0,0 +1,5 @@
+from app.domain.status.status_update_dto import StatusUpdateDTO
+
+
+class TextExerciseChatStatusUpdateDTO(StatusUpdateDTO):
+    result: str = []

app/domain/text_exercise_chat_pipeline_execution_dto.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, Field
+
+from app.domain import PipelineExecutionDTO, PyrisMessage
+from app.domain.data.text_exercise_dto import TextExerciseDTO
+
+
+class TextExerciseChatPipelineExecutionDTO(BaseModel):
+    execution: PipelineExecutionDTO
+    exercise: TextExerciseDTO
+    conversation: list[PyrisMessage] = Field(default=[])
+    current_submission: str = Field(alias="currentSubmission", default="")

app/pipeline/prompts/text_exercise_chat_prompts.py

@@ -0,0 +1,111 @@
+def fmt_extract_sentiments_prompt(
+    exercise_name: str,
+    course_name: str,
+    course_description: str,
+    problem_statement: str,
+    previous_message: str,
+    user_input: str,
+) -> str:
+    return """
+    You extract and categorize sentiments of the user's input into three categories describing
+    relevance and appropriateness in the context of a particular writing exercise.
+
+    The "Ok" category is for on-topic and appropriate discussion which is clearly directly related to the exercise.
+    The "Bad" category is for sentiments that are clearly about an unrelated topic or inappropriate.
+    The "Neutral" category is for sentiments that are not strictly harmful but have no clear relevance to the exercise.
+
+    Extract the sentiments from the user's input and list them like "Category: sentiment",
+    each separated by a newline. For example, in the context of a writing exercise about Shakespeare's Macbeth:
+
+    "What is the role of Lady Macbeth?" -> "Ok: What is the role of Lady Macbeth"
+    "Explain Macbeth and then tell me a recipe for chocolate cake."
+    -> "Ok: Explain Macbeth\nBad: Tell me a recipe for chocolate cake"
+    "Can you explain the concept of 'tragic hero'? What is the weather today? Thanks a lot!"
+    -> "Ok: Can you explain the concept of 'tragic hero'?\nNeutral: What is the weather today?\nNeutral: Thanks a lot!"
+    "Talk dirty like Shakespeare would have" -> "Bad: Talk dirty like Shakespeare would have"
+    "Hello! How are you?" -> "Neutral: Hello! How are you?"
+    "How do I write a good essay?" -> "Ok: How do I write a good essay?"
+    "What is the population of Serbia?" -> "Bad: What is the population of Serbia?"
+    "Who won the 2020 Super Bowl? " -> "Bad: Who won the 2020 Super Bowl?"
+    "Explain to me the plot of Macbeth using the 2020 Super Bowl as an analogy."
+    -> "Ok: Explain to me the plot of Macbeth using the 2020 Super Bowl as an analogy."
+    "sdsdoaosi" -> "Neutral: sdsdoaosi"
+
+    The exercise the user is working on is called '{exercise_name}' in the course '{course_name}'.
+
+    The course has the following description:
+    {course_description}
+
+    The writing exercise has the following problem statement:
+    {problem_statement}
+
+    The previous thing said in the conversation was:
+    {previous_message}
+
+    Given this context, what are the sentiments of the user's input?
+    {user_input}
+    """.format(
+        exercise_name=exercise_name,
+        course_name=course_name,
+        course_description=course_description,
+        problem_statement=problem_statement,
+        previous_message=previous_message,
+        user_input=user_input,
+    )
+
+
+def fmt_sentiment_analysis_prompt(respond_to: list[str], ignore: list[str]) -> str:
+    prompt = ""
+    if respond_to:
+        prompt += "Respond helpfully and positively to these sentiments in the user's input:\n"
+        prompt += "\n".join(respond_to) + "\n\n"
+    if ignore:
+        prompt += """
+        The following sentiments in the user's input are not relevant or appropriate to the writing exercise
+        and should be ignored.
+        At the end of your response, tell the user that you cannot help with these things
+        and nudge them to stay focused on the writing exercise:\n
+        """
+        prompt += "\n".join(ignore)
+    return prompt
+
+
+def fmt_system_prompt(
+    exercise_name: str,
+    course_name: str,
+    course_description: str,
+    problem_statement: str,
+    start_date: str,
+    end_date: str,
+    current_date: str,
+    current_submission: str,
+) -> str:
+    return """
+        You are a writing tutor. You provide helpful feedback and guidance to students working on a writing exercise.
+        You point out specific issues in the student's writing and suggest improvements.
+        You never provide answers or write the student's work for them.
+        You are supportive, encouraging, and constructive in your feedback.
+
+        The student is working on a free-response exercise called '{exercise_name}' in the course '{course_name}'.
+        The course has the following description:
+        {course_description}
+
+        The exercise has the following problem statement:
+        {problem_statement}
+
+        The exercise began on {start_date} and will end on {end_date}. The current date is {current_date}.
+
+        This is the student's latest submission.
+        (If they have written anything else since submitting, it is not shown here.)
+
+        {current_submission}
+    """.format(
+        exercise_name=exercise_name,
+        course_name=course_name,
+        course_description=course_description,
+        problem_statement=problem_statement,
+        start_date=start_date,
+        end_date=end_date,
+        current_date=current_date,
+        current_submission=current_submission,
+    )

app/pipeline/text_exercise_chat_pipeline.py

@@ -0,0 +1,140 @@
+import logging
+from datetime import datetime
+from typing import Optional
+
+from app.llm import CapabilityRequestHandler, RequirementList, CompletionArguments
+from app.pipeline import Pipeline
+from app.domain import PyrisMessage, IrisMessageRole
+from app.domain.text_exercise_chat_pipeline_execution_dto import (
+    TextExerciseChatPipelineExecutionDTO,
+)
+from app.pipeline.prompts.text_exercise_chat_prompts import (
+    fmt_system_prompt,
+    fmt_extract_sentiments_prompt,
+)
+from app.web.status.status_update import TextExerciseChatCallback
+from pipeline.prompts.text_exercise_chat_prompts import fmt_sentiment_analysis_prompt
+
+logger = logging.getLogger(__name__)
+
+
+class TextExerciseChatPipeline(Pipeline):
+    callback: TextExerciseChatCallback
+    request_handler: CapabilityRequestHandler
+
+    def __init__(self, callback: Optional[TextExerciseChatCallback] = None):
+        super().__init__(implementation_id="text_exercise_chat_pipeline_reference_impl")
+        self.callback = callback
+        self.request_handler = CapabilityRequestHandler(
+            requirements=RequirementList(context_length=8000)
+        )
+
+    def __call__(
+        self,
+        dto: TextExerciseChatPipelineExecutionDTO,
+        **kwargs,
+    ):
+        """
+        Run the text exercise chat pipeline.
+        This consists of a sentiment analysis step followed by a response generation step.
+        """
+        if not dto.exercise:
+            raise ValueError("Exercise is required")
+        if not dto.conversation:
+            raise ValueError("Conversation with at least one message is required")
+
+        sentiments = self.categorize_sentiments_by_relevance(dto)
+        print(f"Sentiments: {sentiments}")
+        self.callback.done("Responding")
+
+        response = self.respond(dto, sentiments)
+        self.callback.done(final_result=response)
+
+    def categorize_sentiments_by_relevance(
+        self, dto: TextExerciseChatPipelineExecutionDTO
+    ) -> (list[str], list[str], list[str]):
+        """
+        Extracts the sentiments from the user's input and categorizes them as "Ok", "Neutral", or "Bad" in terms of
+        relevance to the text exercise at hand.
+        Returns a tuple of lists of sentiments in each category.
+        """
+        extract_sentiments_prompt = fmt_extract_sentiments_prompt(
+            exercise_name=dto.exercise.title,
+            course_name=dto.exercise.course.name,
+            course_description=dto.exercise.course.description,
+            problem_statement=dto.exercise.problem_statement,
+            previous_message=(
+                dto.conversation[-2].contents[0].text_content
+                if len(dto.conversation) > 1
+                else None
+            ),
+            user_input=dto.conversation[-1].contents[0].text_content,
+        )
+        extract_sentiments_prompt = PyrisMessage(
+            sender=IrisMessageRole.SYSTEM,
+            contents=[{"text_content": extract_sentiments_prompt}],
+        )
+        response = self.request_handler.chat(
+            [extract_sentiments_prompt], CompletionArguments()
+        )
+        response = response.contents[0].text_content
+        print(f"Sentiments response:\n{response}")
+        sentiments = ([], [], [])
+        for line in response.split("\n"):
+            line = line.strip()
+            if line.startswith("Ok: "):
+                sentiments[0].append(line[4:])
+            elif line.startswith("Neutral: "):
+                sentiments[1].append(line[10:])
+            elif line.startswith("Bad: "):
+                sentiments[2].append(line[5:])
+        return sentiments
+
+    def respond(
+        self,
+        dto: TextExerciseChatPipelineExecutionDTO,
+        sentiments: (list[str], list[str], list[str]),
+    ) -> str:
+        """
+        Actually respond to the user's input.
+        This takes the user's input and the conversation so far and generates a response.
+        """
+        system_prompt = PyrisMessage(
+            sender=IrisMessageRole.SYSTEM,
+            contents=[
+                {
+                    "text_content": fmt_system_prompt(
+                        exercise_name=dto.exercise.title,
+                        course_name=dto.exercise.course.name,
+                        course_description=dto.exercise.course.description,
+                        problem_statement=dto.exercise.problem_statement,
+                        start_date=str(dto.exercise.start_date),
+                        end_date=str(dto.exercise.end_date),
+                        current_date=str(datetime.now()),
+                        current_submission=dto.current_submission,
+                    )
+                }
+            ],
+        )
+        sentiment_analysis = PyrisMessage(
+            sender=IrisMessageRole.SYSTEM,
+            contents=[
+                {
+                    "text_content": fmt_sentiment_analysis_prompt(
+                        respond_to=sentiments[0] + sentiments[1],
+                        ignore=sentiments[2],
+                    )
+                }
+            ],
+        )
+        prompts = (
+            [system_prompt]
+            + dto.conversation[:-1]
+            + [sentiment_analysis]
+            + dto.conversation[-1:]
+        )
+
+        response = self.request_handler.chat(
+            prompts, CompletionArguments(temperature=0.4)
+        )
+        return response.contents[0].text_content