Athena: Add Artemis competencies (#276)

Author: ahmetsenturk

athena/athena/athena/endpoints.py

@@ -3,14 +3,14 @@
 from fastapi import Depends, BackgroundTasks, Body
 from pydantic import ConfigDict, BaseModel, ValidationError
 from pydantic.alias_generators import to_camel
-from typing import TypeVar, Callable, List, Union, Any, Coroutine, Type, Optional
+from typing import TypeVar, Callable, List, Union, Any, Coroutine, Type
 
 from athena.app import app
 from athena.authenticate import authenticated
 from athena.metadata import with_meta
 from athena.module_config import get_dynamic_module_config_factory
 from athena.logger import logger
-from athena.schemas import Exercise, Submission, Feedback, LearnerProfile
+from athena.schemas import Exercise, Submission, Feedback, LearnerProfile, Competency
 from athena.storage import get_stored_submission_meta, get_stored_exercise_meta, get_stored_feedback_meta, \
     store_exercise, store_feedback, store_feedback_suggestions, store_submissions, get_stored_submissions
 
@@ -273,6 +273,8 @@ def feedback_provider(func: Union[
     Callable[[E, S, G, C, LearnerProfile], Coroutine[Any, Any, List[F]]],
     Callable[[E, S, G, C, LearnerProfile, S], List[F]],
     Callable[[E, S, G, C, LearnerProfile, S], Coroutine[Any, Any, List[F]]],
+    Callable[[E, S, G, C, LearnerProfile, S, List[Competency]], List[F]],
+    Callable[[E, S, G, C, LearnerProfile, S, List[Competency]], Coroutine[Any, Any, List[F]]],
 ]):
     """
     Provide feedback to the Assessment Module Manager.
@@ -318,13 +320,22 @@ def feedback_provider(func: Union[
         >>> @feedback_provider
         ... async def async_suggest_feedback_with_profile(exercise: Exercise, submission: Submission, module_config: Optional[dict], learner_profile: Optional[LearnerProfile], latest_submission: Optional[Submission]):
         ...     # suggest feedback here using module_config and learner_profile and return it as a list
+
+        With competencies (both synchronous and asynchronous forms):
+        >>> @feedback_provider
+        ... def sync_suggest_feedback_with_profile(exercise: Exercise, submission: Submission, module_config: Optional[dict], learner_profile: Optional[LearnerProfile], latest_submission: Optional[Submission], competencies: Optional[List[Competency]]):
+        ...     # suggest feedback here using module_config and learner_profile and return it as a list
+        >>> @feedback_provider
+        ... async def async_suggest_feedback_with_profile(exercise: Exercise, submission: Submission, module_config: Optional[dict], learner_profile: Optional[LearnerProfile], latest_submission: Optional[Submission], competencies: Optional[List[Competency]]):
+        ...     # suggest feedback here using module_config and learner_profile and return it as a list
     """
     exercise_type = inspect.signature(func).parameters["exercise"].annotation
     submission_type = inspect.signature(func).parameters["submission"].annotation
     module_config_type = inspect.signature(func).parameters["module_config"].annotation if "module_config" in inspect.signature(func).parameters else None
     is_graded_type = inspect.signature(func).parameters["is_graded"].annotation if "is_graded" in inspect.signature(func).parameters else None
     learner_profile_type = inspect.signature(func).parameters["learner_profile"].annotation if "learner_profile" in inspect.signature(func).parameters else None
     latest_submission_type = inspect.signature(func).parameters["latest_submission"].annotation if "latest_submission" in inspect.signature(func).parameters else None
+    competencies_type = inspect.signature(func).parameters["competencies"].annotation if "competencies" in inspect.signature(func).parameters else None
 
     @app.post("/feedback_suggestions", responses=module_responses)
     @authenticated
@@ -335,6 +346,7 @@ async def wrapper(
             isGraded: is_graded_type = Body(True, alias="isGraded"),
             learner_profile: learner_profile_type = Body(None, alias="learnerProfile"),
             latest_submission: latest_submission_type = Body(None, alias="latestSubmission"),
+            competencies: competencies_type = Body(None, alias="competencies"),
             module_config: module_config_type = Depends(get_dynamic_module_config_factory(module_config_type))):
 
         # Retrieve existing metadata for the exercise, submission and feedback
@@ -358,13 +370,17 @@ async def wrapper(
         if "latest_submission" in inspect.signature(func).parameters:
             kwargs["latest_submission"] = latest_submission
 
+        if "competencies" in inspect.signature(func).parameters:
+            kwargs["competencies"] = competencies
+
         # Filter out unexpected kwargs and log warnings
         accepted_params = set(inspect.signature(func).parameters.keys())
         all_possible_kwargs = {
             "module_config": module_config,
             "is_graded": isGraded,
             "learner_profile": learner_profile,
             "latest_submission": latest_submission,
+            "competencies": competencies,
         }
 
         # Only pass accepted kwargs

athena/athena/athena/schemas/__init__.py

@@ -15,4 +15,6 @@
 from .modeling_submission import ModelingSubmission
 from .grading_criterion import GradingCriterion, StructuredGradingInstruction
 from .structured_grading_criterion import StructuredGradingCriterion
-from .learner_profile import LearnerProfile
+from .learner_profile import LearnerProfile
+from .competency import Competency
+from .competency_taxonomy import CompetencyTaxonomy

athena/athena/athena/schemas/competency.py

@@ -0,0 +1,26 @@
+from abc import ABC
+
+from pydantic import Field, validator
+
+from .schema import Schema
+from .competency_taxonomy import CompetencyTaxonomy
+
+
+class Competency(Schema, ABC):
+    """A competency that is required by the lecturer to be mastered by the student, enhanced with module-specific metadata."""
+    id: int = Field(example=1)
+    title: str = Field("", description="The title of the competency.", example="Competency 1")
+    description: str = Field("", description="The description of the competency.", example="Competency 1 description")
+    taxonomy: CompetencyTaxonomy = Field(None, description="The taxonomy of the competency.")
+
+    meta: dict = Field(default_factory=dict, example={"internal_id": "5"})
+
+    @validator('taxonomy', pre=True)
+    def validate_taxonomy(cls, v):
+        """Validate and convert taxonomy to the correct format."""
+        if isinstance(v, str):
+            return CompetencyTaxonomy.from_any_case(v)
+        return v
+
+    class Config:
+        orm_mode = True

athena/athena/athena/schemas/competency_taxonomy.py

@@ -0,0 +1,21 @@
+from enum import Enum
+
+
+class CompetencyTaxonomy(Enum):
+    """The taxonomy of competencies that is used to structure the competencies."""
+    REMEMBER = "remember"
+    UNDERSTAND = "understand"
+    APPLY = "apply"
+    ANALYZE = "analyze"
+    EVALUATE = "evaluate"
+    CREATE = "create"
+
+    @classmethod
+    def from_any_case(cls, value):
+        """Convert any case to the correct enum value."""
+        if isinstance(value, str):
+            value_lower = value.lower()
+            for member in cls:
+                if member.value == value_lower:
+                    return member
+        raise ValueError(f"Invalid taxonomy value: {value}")

athena/modules/text/module_text_llm/module_text_llm/__main__.py

@@ -13,7 +13,7 @@
 )
 from athena.text import Exercise, Submission, Feedback
 from athena.logger import logger
-from athena.schemas import LearnerProfile
+from athena.schemas import LearnerProfile, Competency
 
 from module_text_llm.config import Configuration
 from module_text_llm.evaluation import get_feedback_statistics, get_llm_statistics
@@ -61,6 +61,7 @@ async def suggest_feedback(
     is_graded: bool,
     learner_profile: Optional[LearnerProfile] = None,
     latest_submission: Optional[Submission] = None,
+    competencies: Optional[List[Competency]] = None,
 ) -> List[Feedback]:
     logger.info(
         "suggest_feedback: %s suggestions for submission %d of exercise %d were requested, with approach: %s",
@@ -77,6 +78,7 @@ async def suggest_feedback(
         is_graded=is_graded,
         learner_profile=learner_profile,
         latest_submission=latest_submission,
+        competencies=competencies,
     )
 
 

athena/modules/text/module_text_llm/module_text_llm/approach_controller.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict, List, Optional
 from athena.text import Exercise, Submission, Feedback
-from athena.schemas.learner_profile import LearnerProfile
+from athena.schemas import LearnerProfile, Competency
 import inspect
 
 from module_text_llm.registry import APPROACH_IMPLEMENTATIONS
@@ -15,6 +15,7 @@ async def generate_suggestions(
     is_graded: bool,
     learner_profile: Optional[LearnerProfile] = None,
     latest_submission: Optional[Submission] = None,
+    competencies: Optional[List[Competency]] = None,
 ) -> List[Feedback]:
     implementation_func = APPROACH_IMPLEMENTATIONS.get(config.type)
     if implementation_func is None:
@@ -36,6 +37,9 @@ async def generate_suggestions(
     if "latest_submission" in sig.parameters:
         kwargs["latest_submission"] = latest_submission
 
+    if "competencies" in sig.parameters:
+        kwargs["competencies"] = competencies
+
     return await implementation_func(
         exercise,
         submission,

athena/modules/text/module_text_llm/module_text_llm/default_approach/generate_suggestions.py

@@ -1,7 +1,7 @@
 from typing import List, Optional
 
 from athena import emit_meta
-from athena.schemas import LearnerProfile
+from athena.schemas import LearnerProfile, Competency
 from athena.text import Exercise, Submission, Feedback
 from athena.logger import logger
 
@@ -44,6 +44,7 @@ def _prepare_analysis_prompt_input(
     exercise: Exercise,
     submission: Submission,
     latest_submission: Optional[Submission] = None,
+    competencies: Optional[List[Competency]] = None,
 ) -> dict:
     """Prepare input data for the submission analysis prompt."""
     return {
@@ -57,6 +58,7 @@ def _prepare_analysis_prompt_input(
         "submission": add_sentence_numbers(submission.text),
         "previous_submission": add_sentence_numbers(
             latest_submission.text) if latest_submission is not None else "Previous submission is not available.",
+        "competencies": competencies or [],
     }
 
 
@@ -243,7 +245,7 @@ async def _generate_graded_feedback(
 
     result = await predict_and_parse(
         model=config.model,
-        chat_prompt=chat_prompt, 
+        chat_prompt=chat_prompt,
         prompt_input=prompt_input,
         pydantic_object=AssessmentModel,
         tags=[
@@ -321,6 +323,7 @@ async def generate_suggestions(
     is_graded: bool,
     learner_profile: Optional[LearnerProfile],
     latest_submission: Optional[Submission] = None,
+    competencies: Optional[List[Competency]] = None,
 ) -> List[Feedback]:
     """Generate feedback suggestions for a student submission using a two-step LLM approach."""
     if latest_submission is None:
@@ -337,11 +340,16 @@ async def generate_suggestions(
             return []
         return _convert_to_feedback_objects(result, exercise, submission, is_graded)
 
+    if competencies is None:
+        logger.info("Competencies are not provided.")
+    else:
+        logger.info("Competencies are provided: %s", competencies)
+
     # Setup learner profile with fallbacks
     learner_profile = _setup_learner_profile(learner_profile, config)
 
     # Prepare input for submission analysis
-    prompt_input = _prepare_analysis_prompt_input(exercise, submission, latest_submission)
+    prompt_input = _prepare_analysis_prompt_input(exercise, submission, latest_submission, competencies)
 
     # Analyze the submission
     submission_analysis = await _analyze_submission(

athena/modules/text/module_text_llm/module_text_llm/default_approach/prompts/submission_analysis_prompt.py

@@ -5,7 +5,12 @@
 You are an educational evaluator reviewing a student's progress on a text-based exercise.
 
 You will:
-1. Analyze the problem statement, sample solution, and grading instructions to extract core competencies required to solve the task.
+0. Observe if the required competencies are set by the instructor. If they are, move to step 2 and skip the step 1:
+
+Required Competencies:
+{competencies}
+
+1. Only if the required competencies are not provided, then analyze the problem statement, sample solution, and grading instructions to extract core competencies required to solve the task.
     - For each competency, define the expected cognitive level (e.g., Recall, Understand, Apply, Analyze, Evaluate, Create).
     - Reference the grading_instruction_id if relevant.
 

athena/modules/text/module_text_llm/module_text_llm/default_approach/schemas/__init__.py

@@ -1,8 +1,6 @@
 from .feedback_models import FeedbackType, FeedbackModel, AssessmentModel
 from .analysis_models import (
     CompetencyStatus,
-    CognitiveLevel,
-    RequiredCompetency,
     CompetencyChange,
     EnhancedCompetencyEvaluation,
     SubmissionAnalysis,
@@ -13,8 +11,6 @@
     "FeedbackModel", 
     "AssessmentModel",
     "CompetencyStatus",
-    "CognitiveLevel",
-    "RequiredCompetency",
     "CompetencyChange",
     "EnhancedCompetencyEvaluation",
     "SubmissionAnalysis",

athena/modules/text/module_text_llm/module_text_llm/default_approach/schemas/analysis_models.py

@@ -1,7 +1,9 @@
-from pydantic import BaseModel, Field
 from typing import List, Optional, Literal
 from enum import Enum
 
+from athena.schemas import Competency
+from pydantic import BaseModel, Field
+
 
 class CompetencyStatus(str, Enum):
     NOT_ATTEMPTED = "Not Attempted"
@@ -10,21 +12,6 @@ class CompetencyStatus(str, Enum):
     CORRECT = "Correct"
 
 
-class CognitiveLevel(str, Enum):
-    RECALL = "Recall"
-    UNDERSTAND = "Understand"
-    APPLY = "Apply"
-    ANALYZE = "Analyze"
-    EVALUATE = "Evaluate"
-    CREATE = "Create"
-
-
-class RequiredCompetency(BaseModel):
-    description: str = Field(..., description="The skill or knowledge the student is expected to demonstrate.")
-    cognitive_level: Optional[CognitiveLevel] = Field(None, description="Bloom's Taxonomy level")
-    grading_instruction_id: Optional[int] = Field(None, description="Reference to grading instruction if applicable")
-
-
 class CompetencyChange(BaseModel):
     type: Literal["added", "removed", "modified", "unchanged"]
     is_positive: Optional[bool] = Field(None, description="Whether the change is positive (improvement)")
@@ -35,8 +22,8 @@ class CompetencyChange(BaseModel):
 
 
 class EnhancedCompetencyEvaluation(BaseModel):
-    competency: RequiredCompetency
-    status: CompetencyStatus
+    competency: Competency = Field(..., description="The competency that was evaluated")
+    status: CompetencyStatus = Field(..., description="The student's status for the competency")
     evidence: Optional[str] = Field(None, description="Quote or paraphrase from student's current submission")
     line_start: Optional[int] = Field(None, description="Start line number for evidence")
     line_end: Optional[int] = Field(None, description="End line number for evidence")