Fix competency extraction feature

app/domain/__init__.py

@@ -3,6 +3,9 @@
 from .pipeline_execution_settings_dto import PipelineExecutionSettingsDTO
 from .chat.chat_pipeline_execution_dto import ChatPipelineExecutionDTO
 from .chat.chat_pipeline_execution_base_data_dto import ChatPipelineExecutionBaseDataDTO
+from .competency_extraction_pipeline_execution_dto import (
+    CompetencyExtractionPipelineExecutionDTO,
+)
 from app.domain.chat.exercise_chat.exercise_chat_pipeline_execution_dto import (
     ExerciseChatPipelineExecutionDTO,
 )

app/domain/chat/chat_pipeline_execution_dto.py

@@ -2,16 +2,11 @@
 
 from pydantic import Field
 
-from app.domain import PipelineExecutionDTO, PipelineExecutionSettingsDTO
+from app.domain import PipelineExecutionDTO
 from app.domain.pyris_message import PyrisMessage
 from app.domain.data.user_dto import UserDTO
-from app.domain.status.stage_dto import StageDTO
 
 
 class ChatPipelineExecutionDTO(PipelineExecutionDTO):
     chat_history: List[PyrisMessage] = Field(alias="chatHistory", default=[])
     user: Optional[UserDTO]
-    settings: Optional[PipelineExecutionSettingsDTO]
-    initial_stages: Optional[List[StageDTO]] = Field(
-        default=None, alias="initialStages"
-    )

app/domain/competency_extraction_pipeline_execution_dto.py

@@ -0,0 +1,19 @@
+from typing import List, Optional
+
+from pydantic import Field, BaseModel
+
+from . import PipelineExecutionDTO
+from .data.competency_dto import CompetencyTaxonomy
+
+
+class CompetencyExtractionPipelineExecutionDTO(BaseModel):
+    execution: PipelineExecutionDTO
+    course_description: Optional[str] = Field(alias="courseDescription")
+    taxonomy_options: List[CompetencyTaxonomy] = Field(
+        alias="taxonomyOptions", default=[]
+    )
+    max_n: int = Field(
+        alias="maxN",
+        description="Maximum number of competencies to extract from the course description",
+        default=10,
+    )

app/domain/data/competency_dto.py

@@ -3,6 +3,7 @@
 from typing import Optional
 
 from pydantic import BaseModel, Field
+from pydantic.v1 import validator
 
 
 class CompetencyTaxonomy(str, Enum):
@@ -21,3 +22,29 @@ class CompetencyDTO(BaseModel):
     taxonomy: Optional[CompetencyTaxonomy] = None
     soft_due_date: Optional[datetime] = Field(default=None, alias="softDueDate")
     optional: Optional[bool] = None
+
+
+class Competency(BaseModel):
+    title: str = Field(
+        description="Title of the competency that contains no more than 4 words",
+    )
+    description: str = Field(
+        description="Description of the competency as plain string. DO NOT RETURN A LIST OF STRINGS."
+    )
+    taxonomy: CompetencyTaxonomy = Field(
+        description="Selected taxonomy based on bloom's taxonomy"
+    )
+
+    @validator("title")
+    def validate_title(cls, field):
+        """Validate the subject of the competency."""
+        if len(field.split()) > 4:
+            raise ValueError("Title must contain no more than 4 words")
+        return field
+
+    @validator("taxonomy")
+    def validate_selected_taxonomy(cls, field):
+        """Validate the selected taxonomy."""
+        if field not in CompetencyTaxonomy.__members__.keys():
+            raise ValueError(f"Invalid taxonomy: {field}")
+        return field

app/domain/ingestion/ingestion_pipeline_execution_dto.py

@@ -1,17 +1,12 @@
-from typing import List, Optional
+from typing import List
 
 from pydantic import Field
 
-from app.domain import PipelineExecutionDTO, PipelineExecutionSettingsDTO
+from app.domain import PipelineExecutionDTO
 from app.domain.data.lecture_unit_dto import LectureUnitDTO
-from app.domain.status.stage_dto import StageDTO
 
 
 class IngestionPipelineExecutionDto(PipelineExecutionDTO):
     lecture_units: List[LectureUnitDTO] = Field(
         ..., alias="pyrisLectureUnitWebhookDTOS"
     )
-    settings: Optional[PipelineExecutionSettingsDTO]
-    initial_stages: Optional[List[StageDTO]] = Field(
-        default=None, alias="initialStages"
-    )

app/domain/pipeline_execution_dto.py

@@ -1,8 +1,16 @@
-from pydantic import BaseModel
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from app.domain.pipeline_execution_settings_dto import PipelineExecutionSettingsDTO
+from app.domain.status.stage_dto import StageDTO
 
 
 class PipelineExecutionDTO(BaseModel):
-    pass
+    settings: Optional[PipelineExecutionSettingsDTO]
+    initial_stages: Optional[list[StageDTO]] = Field(
+        default=None, alias="initialStages"
+    )
 
     class Config:
         populate_by_name = True

app/domain/status/competency_extraction_status_update_dto.py

@@ -0,0 +1,6 @@
+from app.domain.data.competency_dto import Competency
+from app.domain.status.status_update_dto import StatusUpdateDTO
+
+
+class CompetencyExtractionStatusUpdateDTO(StatusUpdateDTO):
+    result: list[Competency] = []

app/pipeline/__init__.py

@@ -1 +1 @@
-from ..pipeline.pipeline import Pipeline
+from app.pipeline.pipeline import Pipeline

app/pipeline/competency_extraction_pipeline.py

@@ -0,0 +1,94 @@
+import logging
+from typing import Optional
+
+from langchain.output_parsers import PydanticOutputParser
+from langchain_core.prompts import (
+    ChatPromptTemplate,
+)
+
+from app.domain import (
+    CompetencyExtractionPipelineExecutionDTO,
+    PyrisMessage,
+    IrisMessageRole,
+)
+from app.domain.data.text_message_content_dto import TextMessageContentDTO
+from app.domain.data.competency_dto import Competency
+from app.llm import CapabilityRequestHandler, RequirementList, CompletionArguments
+from app.pipeline import Pipeline
+from app.web.status.status_update import CompetencyExtractionCallback
+from app.pipeline.prompts.competency_extraction import system_prompt
+
+logger = logging.getLogger(__name__)
+
+
+class CompetencyExtractionPipeline(Pipeline):
+    callback: CompetencyExtractionCallback
+    request_handler: CapabilityRequestHandler
+    output_parser: PydanticOutputParser
+
+    def __init__(self, callback: Optional[CompetencyExtractionCallback] = None):
+        super().__init__(
+            implementation_id="competency_extraction_pipeline_reference_impl"
+        )
+        self.callback = callback
+        self.request_handler = CapabilityRequestHandler(requirements=RequirementList())
+        self.output_parser = PydanticOutputParser(pydantic_object=Competency)
+
+    def __call__(
+        self,
+        dto: CompetencyExtractionPipelineExecutionDTO,
+        prompt: Optional[ChatPromptTemplate] = None,
+        **kwargs,
+    ):
+        if not dto.course_description:
+            self.callback.error("Course description is required")
+        if not dto.taxonomy_options:
+            self.callback.error("Taxonomy options are required")
+        if not dto.max_n:
+            self.callback.error("Non-zero max_n is required")
+
+        taxonomy_options = ", ".join(dto.taxonomy_options)
+
+        prompt = system_prompt.format(
+            taxonomy_list=taxonomy_options,
+            course_description=dto.course_description,
+            n=dto.max_n,
+        )
+        prompt = PyrisMessage(
+            sender=IrisMessageRole.SYSTEM,
+            contents=[TextMessageContentDTO(text_content=prompt)],
+        )
+
+        self.callback.in_progress("Starting competency extraction")
+
+        response = self.request_handler.chat(
+            [prompt], CompletionArguments(temperature=0.4)
+        )
+        response = response.contents[0].text_content
+
+        print(f"Received response from OpenAI: {response}")
+
+        generated_competencies: list[Competency] = []
+
+        # Find all competencies in the response
+        competencies = response.split("\n\n")
+        for i, competency in enumerate(competencies):
+            print(f"Processing competency {i + 1}: {competency}")
+            if "{" not in competency or "}" not in competency:
+                print("Skipping competency without JSON")
+                continue
+            # Get the competency JSON object
+            start = competency.index("{")
+            end = competency.index("}") + 1
+            competency = competency[start:end]
+            try:
+                competency = self.output_parser.parse(competency)
+                print(f"Generated competency: {competency}")
+                generated_competencies.append(competency)
+                self.callback.done(final_result=generated_competencies)
+            except Exception as e:
+                print(f"Error generating competency: {e}")
+                self.callback.error(f"Error generating competency: {e}")
+        # Mark all remaining competencies as skipped
+        for i in range(len(generated_competencies), len(competencies)):
+            self.callback.skip(f"Skipping competency {i + 1}")

app/pipeline/prompts/competency_extraction.py

@@ -0,0 +1,42 @@
+system_prompt = """
+You are an expert in all topics of computer science and its practical applications.
+Your task consists of three parts:
+1. Read the provided curriculum description a university course.
+2. Extract all learning goals ("competencies") from the course description.
+
+Each competency must contain the following fields:
+
+- title:
+The title of the competency, which is a specific topic or skill. This should be a short phrase of at most 4 words.
+
+- description:
+A detailed description of the competency in 3 to 6 bullet points.
+Each bullet point is a short sentence, at most 15 words.
+Each bullet point illustrates a specific skill or concept of the competency.
+
+- taxonomy:
+The classification of the competency within Bloom's taxonomy.
+You must choose from these options in Bloom's taxonomy: {taxonomy_list}
+
+All competencies must meet the following requirements:
+
+- is mentioned in the course description.
+- corresponds to exactly one subject or skill covered in the course description.
+- is assigned to exactly one level of Bloom's taxonomy.
+- is small and fine-grained. Large topics should be broken down into smaller competencies.
+- does not overlap with other competencies: each competency is unique. Expanding on a previous competency is allowed.
+
+Here is an example competency whose structure you should follow:
+
+{{
+    "title": "Recursion",
+    "description": "- You understand the concept of recursion.
+    - You are able to understand complex recursive implementations.
+    - You are able to implement recursive solutions of medium difficulty independently.",
+    "taxonomy": "ANALYZE"
+}}
+
+Here is the provided course description: {course_description}
+
+Respond with up to {n} competencies extracted from the course description, each in JSON format, split by two newlines.
+"""

app/web/routers/pipelines.py

@@ -9,14 +9,17 @@
 from app.domain import (
     ExerciseChatPipelineExecutionDTO,
     CourseChatPipelineExecutionDTO,
+    CompetencyExtractionPipelineExecutionDTO,
 )
 from app.web.status.status_update import (
     ExerciseChatStatusCallback,
     CourseChatStatusCallback,
+    CompetencyExtractionCallback,
 )
 from app.pipeline.chat.course_chat_pipeline import CourseChatPipeline
 from app.pipeline.chat.exercise_chat_pipeline import ExerciseChatPipeline
 from app.dependencies import TokenValidator
+from app.pipeline.competency_extraction_pipeline import CompetencyExtractionPipeline
 
 router = APIRouter(prefix="/api/v1/pipelines", tags=["pipelines"])
 logger = logging.getLogger(__name__)
@@ -86,6 +89,45 @@ def run_course_chat_pipeline(variant: str, dto: CourseChatPipelineExecutionDTO):
     thread.start()
 
 
+def run_competency_extraction_pipeline_worker(
+    dto: CompetencyExtractionPipelineExecutionDTO, _variant: str
+):
+    try:
+        callback = CompetencyExtractionCallback(
+            run_id=dto.execution.settings.authentication_token,
+            base_url=dto.execution.settings.artemis_base_url,
+            initial_stages=dto.execution.initial_stages,
+            num_iterations=dto.max_n,
+        )
+        pipeline = CompetencyExtractionPipeline(callback=callback)
+    except Exception as e:
+        logger.error(f"Error preparing exercise chat pipeline: {e}")
+        logger.error(traceback.format_exc())
+        capture_exception(e)
+        return
+
+    try:
+        pipeline(dto=dto)
+    except Exception as e:
+        logger.error(f"Error running exercise chat pipeline: {e}")
+        logger.error(traceback.format_exc())
+        callback.error("Fatal error.", exception=e)
+
+
+@router.post(
+    "/competency-extraction/{variant}/run",
+    status_code=status.HTTP_202_ACCEPTED,
+    dependencies=[Depends(TokenValidator())],
+)
+def run_competency_extraction_pipeline(
+    variant: str, dto: CompetencyExtractionPipelineExecutionDTO
+):
+    thread = Thread(
+        target=run_competency_extraction_pipeline_worker, args=(dto, variant)
+    )
+    thread.start()
+
+
 @router.get("/{feature}")
 def get_pipeline(feature: str):
     """

app/web/status/status_update.py

@@ -5,6 +5,9 @@
 import requests
 from abc import ABC
 
+from domain.status.competency_extraction_status_update_dto import (
+    CompetencyExtractionStatusUpdateDTO,
+)
 from ...domain.chat.course_chat.course_chat_status_update_dto import (
     CourseChatStatusUpdateDTO,
 )
@@ -219,3 +222,27 @@ def __init__(
         status = ExerciseChatStatusUpdateDTO(stages=stages)
         stage = stages[current_stage_index]
         super().__init__(url, run_id, status, stage, current_stage_index)
+
+
+class CompetencyExtractionCallback(StatusCallback):
+    def __init__(
+        self,
+        run_id: str,
+        base_url: str,
+        initial_stages: List[StageDTO] = None,
+        num_iterations=10,
+    ):
+        url = f"{base_url}/api/public/pyris/pipelines/competency-extraction/runs/{run_id}/status"
+        current_stage_index = 1 if initial_stages else 0
+        stages = initial_stages or []
+        stages += [
+            StageDTO(
+                weight=10,
+                state=StageStateEnum.NOT_STARTED,
+                name=f"Competency {i + 1}",
+            )
+            for i in range(num_iterations)
+        ]
+        status = CompetencyExtractionStatusUpdateDTO(stages=stages)
+        stage = stages[current_stage_index]
+        super().__init__(url, run_id, status, stage, current_stage_index)