Persist Structured Grading Instructions (Modeling)

athena/athena/app.py

@@ -69,4 +69,4 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
     return JSONResponse(
         status_code=422,
         content={"detail": exc.errors()},
-    )
+    )

athena/athena/models/__init__.py

@@ -14,4 +14,5 @@
 from .db_modeling_submission import DBModelingSubmission
 from .db_programming_feedback import DBProgrammingFeedback
 from .db_text_feedback import DBTextFeedback
-from .db_modeling_feedback import DBModelingFeedback
+from .db_modeling_feedback import DBModelingFeedback
+from .db_structured_grading_criterion import DBStructuredGradingCriterion

athena/athena/models/db_exercise.py

@@ -1,18 +1,29 @@
 from sqlalchemy import Column, String, Float, JSON, Enum as SqlEnum
-
+from sqlalchemy.orm import relationship
+from athena.database import Base
 from athena.schemas import ExerciseType
 from .model import Model
 from .big_integer_with_autoincrement import BigIntegerWithAutoincrement
 
 
-class DBExercise(Model):
+class DBExercise(Model, Base):
+    __tablename__ = "exercise"
     id = Column(BigIntegerWithAutoincrement, primary_key=True, index=True, nullable=False)
     lms_url = Column(String, index=True, nullable=False)
     title = Column(String, index=True, nullable=False)
-    type = Column(SqlEnum(ExerciseType), index=True, nullable=False)
     max_points = Column(Float, index=True, nullable=False)
     bonus_points = Column(Float, index=True, nullable=False)
     grading_instructions = Column(String)
     problem_statement = Column(String)
     grading_criteria = Column(JSON, nullable=True)
     meta = Column(JSON, nullable=False)
+
+    structured_grading_criterion = relationship("DBStructuredGradingCriterion", back_populates="exercise", uselist=False)  # Use uselist=False for one-to-one
+
+    # Polymorphism, discriminator attribute
+    type = Column(SqlEnum(ExerciseType), index=True, nullable=False)
+
+    __mapper_args__ = {
+        'polymorphic_identity': 'exercise',
+        'polymorphic_on': 'type'
+    }

athena/athena/models/db_modeling_exercise.py

@@ -1,14 +1,17 @@
-from sqlalchemy import Column, String
+from athena.schemas.exercise_type import ExerciseType
+from sqlalchemy import Column, String, ForeignKey
 from sqlalchemy.orm import relationship
-
-from athena.database import Base
 from .db_exercise import DBExercise
-
-
-class DBModelingExercise(DBExercise, Base):
+from .big_integer_with_autoincrement import BigIntegerWithAutoincrement
+class DBModelingExercise(DBExercise):
     __tablename__ = "modeling_exercises"
 
-    example_solution: str = Column(String)  # type: ignore
-
+    example_solution = Column(String)  # type: ignore
+
+    id = Column(BigIntegerWithAutoincrement, ForeignKey('exercise.id'), primary_key=True)
     submissions = relationship("DBModelingSubmission", back_populates="exercise")
     feedbacks = relationship("DBModelingFeedback", back_populates="exercise")
+
+    __mapper_args__ = {
+        'polymorphic_identity': ExerciseType.modeling.value
+    }

athena/athena/models/db_modeling_feedback.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from sqlalchemy import Column, ForeignKey, JSON
+from sqlalchemy import Column, ForeignKey, JSON, String
 from sqlalchemy.orm import relationship
 
 from athena.database import Base
@@ -11,7 +11,8 @@
 class DBModelingFeedback(DBFeedback, Base):
     __tablename__ = "modeling_feedbacks"
 
-    element_ids: Optional[list[str]] = Column(JSON)  # type: ignore
+    element_ids: Optional[list[str]] = Column(JSON) # type: ignore
+    reference: Optional[str] = Column(String, nullable=True) # type: ignore
 
     exercise_id = Column(BigIntegerWithAutoincrement, ForeignKey("modeling_exercises.id", ondelete="CASCADE"), index=True)
     submission_id = Column(BigIntegerWithAutoincrement, ForeignKey("modeling_submissions.id", ondelete="CASCADE"), index=True)

athena/athena/models/db_programming_exercise.py

@@ -1,17 +1,23 @@
-from sqlalchemy import Column, String
+from athena.schemas.exercise_type import ExerciseType
+from sqlalchemy import Column, String, ForeignKey
 from sqlalchemy.orm import relationship
 
-from athena.database import Base
 from .db_exercise import DBExercise
+from .big_integer_with_autoincrement import BigIntegerWithAutoincrement
 
 
-class DBProgrammingExercise(DBExercise, Base):
+class DBProgrammingExercise(DBExercise):
     __tablename__ = "programming_exercises"
 
+    id = Column(BigIntegerWithAutoincrement, ForeignKey('exercise.id'), primary_key=True)
     programming_language: str = Column(String, nullable=False)  # type: ignore
     solution_repository_uri: str = Column(String, nullable=False) # type: ignore
     template_repository_uri: str = Column(String, nullable=False) # type: ignore
     tests_repository_uri: str = Column(String, nullable=False)  # type: ignore
 
     submissions = relationship("DBProgrammingSubmission", back_populates="exercise")
     feedbacks = relationship("DBProgrammingFeedback", back_populates="exercise")
+
+    __mapper_args__ = {
+        'polymorphic_identity': ExerciseType.programming.value
+    }

athena/athena/models/db_structured_grading_criterion.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column, JSON, String, ForeignKey
+from sqlalchemy.orm import relationship
+
+from athena.database import Base
+from .big_integer_with_autoincrement import BigIntegerWithAutoincrement
+
+
+class DBStructuredGradingCriterion(Base):
+    __tablename__ = "structured_grading_criterion"
+    id = Column(BigIntegerWithAutoincrement, primary_key=True, index=True,
+                autoincrement=True)
+    exercise_id = Column(BigIntegerWithAutoincrement, ForeignKey("exercise.id", ondelete="CASCADE"), index=True, unique=True) # Only one cached instruction per exercise
+    instructions_hash = Column(String, nullable=False)
+    structured_grading_criterion = Column(JSON, nullable=False)
+    lms_url = Column(String, nullable=False)
+
+    exercise = relationship("DBExercise", back_populates="structured_grading_criterion")

athena/athena/models/db_text_exercise.py

@@ -1,14 +1,20 @@
-from sqlalchemy import Column, String
+from athena.schemas.exercise_type import ExerciseType
+from sqlalchemy import Column, String, ForeignKey
 from sqlalchemy.orm import relationship
 
-from athena.database import Base
 from .db_exercise import DBExercise
+from .big_integer_with_autoincrement import BigIntegerWithAutoincrement
 
 
-class DBTextExercise(DBExercise, Base):
+class DBTextExercise(DBExercise):
     __tablename__ = "text_exercises"
+    id = Column(BigIntegerWithAutoincrement, ForeignKey('exercise.id'), primary_key=True)
 
     example_solution: str = Column(String)  # type: ignore
 
     submissions = relationship("DBTextSubmission", back_populates="exercise")
     feedbacks = relationship("DBTextFeedback", back_populates="exercise")
+
+    __mapper_args__ = {
+        'polymorphic_identity': ExerciseType.text.value
+    }

athena/athena/schemas/__init__.py

@@ -13,4 +13,5 @@
 from .modeling_feedback import ModelingFeedback
 from .modeling_exercise import ModelingExercise
 from .modeling_submission import ModelingSubmission
-from .grading_criterion import GradingCriterion, StructuredGradingInstruction, StructuredGradingCriterion
+from .grading_criterion import GradingCriterion, StructuredGradingInstruction
+from .structured_grading_criterion import StructuredGradingCriterion

athena/athena/schemas/grading_criterion.py

@@ -23,7 +23,4 @@ class GradingCriterion(Schema, ABC):
     title: Optional[str] = Field(None, example="Some instructions")
     structured_grading_instructions: List[StructuredGradingInstruction] = Field(
         [], example=[{"credits": 1.0, "gradingScale": "Good", "instructionDescription": "Some instructions", "feedback": "Nicely done!", "usageCount": 1},
-                     {"credits": 0.0, "gradingScale": "Bad", "instructionDescription": "Some instructions", "feedback": "Try again!", "usageCount": 0}])
-
-class StructuredGradingCriterion(BaseModel):
-    criteria: List[GradingCriterion]
+                     {"credits": 0.0, "gradingScale": "Bad", "instructionDescription": "Some instructions", "feedback": "Try again!", "usageCount": 0}])

athena/athena/schemas/modeling_feedback.py

@@ -1,11 +1,10 @@
-from typing import Optional, List
-
+from typing import List, Optional
 from pydantic import Field
-
 from .feedback import Feedback
 
 
 class ModelingFeedback(Feedback):
     """Feedback on a modeling exercise."""
 
     element_ids: Optional[List[str]] = Field([], description="referenced diagram element IDs", example=["id_1"])
+    reference: Optional[str] = Field(None, description="reference to the diagram element", example="ClassAttribute:5a337bdf-da00-4bd0-a6f0-78ba5b84330e")

athena/athena/schemas/structured_grading_criterion.py

@@ -0,0 +1,7 @@
+from typing import List
+from athena.schemas.grading_criterion import GradingCriterion
+from pydantic import BaseModel
+
+
+class StructuredGradingCriterion(BaseModel):
+    criteria: List[GradingCriterion]

athena/athena/storage/structured_grading_criterion_storage.py

@@ -0,0 +1,31 @@
+from typing import Optional
+from athena.contextvars import get_lms_url
+from athena.database import get_db
+
+from athena.models import DBStructuredGradingCriterion
+from athena.schemas import StructuredGradingCriterion
+
+def get_structured_grading_criterion(exercise_id: int, current_hash: Optional[str] = None) -> Optional[StructuredGradingCriterion]:
+    lms_url = get_lms_url()
+    with get_db() as db:
+        cache_entry = db.query(DBStructuredGradingCriterion).filter_by(
+            exercise_id=exercise_id, lms_url=lms_url
+        ).first()
+        if cache_entry is not None and (current_hash is None or cache_entry.instructions_hash == current_hash):  # type: ignore
+            return StructuredGradingCriterion.parse_obj(cache_entry.structured_grading_criterion)
+    return None
+
+def store_structured_grading_criterion(
+    exercise_id: int, hash: str, structured_instructions: StructuredGradingCriterion
+):
+    lms_url = get_lms_url()
+    with get_db() as db:
+        db.merge(
+            DBStructuredGradingCriterion(
+                exercise_id=exercise_id,
+                lms_url=lms_url,
+                instructions_hash=hash,
+                structured_grading_criterion=structured_instructions.dict(),
+            )
+        )
+        db.commit()

llm_core/llm_core/models/openai.py

@@ -67,7 +67,7 @@ class OpenAIModelConfig(ModelConfig):
 
         model_name: OpenAIModel = Field(default=default_openai_model, # type: ignore
                                         description="The name of the model to use.")
-        max_tokens: PositiveInt = Field(1000, description="""\
+        max_tokens: PositiveInt = Field(4000, description="""\
 The maximum number of [tokens](https://platform.openai.com/tokenizer) to generate in the chat completion.
 
 The total length of input tokens and generated tokens is limited by the model's context length. \

modules/modeling/module_modeling_llm/module_modeling_llm/apollon_transformer/apollon_json_transformer.py

@@ -6,7 +6,7 @@
 class ApollonJSONTransformer:
 
     @staticmethod
-    def transform_json(model: str) -> tuple[str, dict[str, str], str]:
+    def transform_json(model: str) -> tuple[str, dict[str, str], str, dict[str, str]]:
         """
         Serialize a given Apollon diagram model to a string representation.
         This method converts the UML diagram model into a format similar to mermaid syntax, called "apollon".
@@ -25,11 +25,11 @@ def transform_json(model: str) -> tuple[str, dict[str, str], str]:
         # Convert the UML diagram to the apollon representation
         apollon_representation = parser.to_apollon()
 
-        # Extract elements and relations with their corresponding IDs and names
-        names = {
-            **{element['name']: element['id'] for element in parser.get_elements()},
-            **{relation['name']: relation['id'] for relation in parser.get_relations()}
-        }
-    
-        return apollon_representation, names, diagram_type
+        # Get the mapping of element, method, and attribute names to their corresponding IDs
+        # This is used to resolve references to as the apollon representation only contains names and not IDs
+        names = parser.get_element_id_mapping()
+
+        id_type_mapping = parser.get_id_to_type_mapping()
+
+        return apollon_representation, names, diagram_type, id_type_mapping
 

modules/modeling/module_modeling_llm/module_modeling_llm/apollon_transformer/parser/element.py

@@ -1,4 +1,7 @@
-from typing import Dict, Any, List, Optional
+# apollon_transformer/parser/element.py
+
+from typing import Dict, Any, List, Optional, Tuple
+from string import ascii_uppercase
 
 class Element:
     """
@@ -17,15 +20,61 @@ def __init__(self, data: Dict[str, Any], element_dict: Optional[Dict[str, Any]]
         self.method_refs: List[str] = data.get('methods', [])
         self.attributes: List[str] = []
         self.methods: List[str] = []
+        self.attribute_id_mapping: Dict[str, str] = {}
+        self.method_id_mapping: Dict[str, str] = {}
         if element_dict is not None:
             self.resolve_references(element_dict)
 
     def resolve_references(self, element_dict: Dict[str, Any]):
         """
-        Resolve attribute and method references using the provided element dictionary. The json data contains only references to other elements that represent attributes and methods. This method resolves these references to the actual names of the attributes and methods by looking up the corresponding elements via their IDs in the provided element dictionary.
+        Resolve attribute and method references using the provided element dictionary.
+        Ensures uniqueness among attribute and method names within the class.
         """
-        self.attributes = [element_dict[ref].get("name", "") for ref in self.attribute_refs if ref in element_dict]
-        self.methods = [element_dict[ref].get('name', '') for ref in self.method_refs if ref in element_dict]
+        # Resolve attributes
+        self.attributes, self.attribute_id_mapping = self._resolve_uniqueness(
+            self.attribute_refs, element_dict)
+
+        # Resolve methods
+        self.methods, self.method_id_mapping = self._resolve_uniqueness(
+            self.method_refs, element_dict)
+
+    def _resolve_uniqueness(
+        self, refs: List[str], element_dict: Dict[str, Any]
+    ) -> Tuple[List[str], Dict[str, str]]:
+        name_counts: Dict[str, int] = {}
+        unique_full_names: List[str] = []
+        id_mapping: Dict[str, str] = {}
+        for ref in refs:
+            if ref in element_dict:
+                full_name = element_dict[ref].get("name", "")
+                simplified_name = self.extract_name(full_name)
+                count = name_counts.get(simplified_name, 0)
+                if count > 0:
+                    suffix = f"#{ascii_uppercase[count - 1]}"
+                    simplified_name_with_suffix = f"{simplified_name}{suffix}"
+                else:
+                    simplified_name_with_suffix = simplified_name
+                name_counts[simplified_name] = count + 1
+                unique_full_names.append(full_name)
+                id_mapping[simplified_name_with_suffix] = ref
+        return unique_full_names, id_mapping
+
+    @staticmethod
+    def extract_name(full_name: str) -> str:
+        """
+        Extracts the simplified name from the full attribute or method name.
+        Removes visibility symbols, type annotations, and parameters.
+        """
+        # Remove visibility symbols and leading/trailing spaces
+        name = full_name.lstrip('+-#~ ').strip()
+        # For attributes, split on ':'
+        if ':' in name:
+            name = name.split(':')[0].strip()
+        # For methods, split on '('
+        elif '(' in name:
+            name = name.split('(')[0].strip()
+        return name
+
 
     def to_apollon(self) -> str:
         parts = [f"[{self.type}] {self.name}"]
@@ -41,6 +90,6 @@ def to_apollon(self) -> str:
             parts.append("{\n" + "\n".join(details) + "\n}")
 
         return " ".join(parts)
-
+        
     def __getitem__(self, key):
-        return self.__dict__[key]
+        return self.__dict__[key]

modules/modeling/module_modeling_llm/module_modeling_llm/apollon_transformer/parser/uml_parser.py

@@ -93,4 +93,30 @@ def get_elements(self) -> List[Element]:
         return self.elements
 
     def get_relations(self) -> List[Relation]:
-        return self.relations
+        return self.relations
+
+    def get_element_id_mapping(self) -> Dict[str, str]:
+        """
+        Creates a mapping from element names to their IDs, including attributes and methods.
+        """
+        mapping = {}
+        for element in self.elements:
+            mapping[element.name] = element.id
+            for simplified_name_with_suffix, attr_id in element.attribute_id_mapping.items():
+                mapping[f"{element.name}.{simplified_name_with_suffix}"] = attr_id
+            for simplified_name_with_suffix, method_id in element.method_id_mapping.items():
+                mapping[f"{element.name}.{simplified_name_with_suffix}"] = method_id
+        for relation in self.relations:
+            mapping[relation.name] = relation.id
+        return mapping
+
+    def get_id_to_type_mapping(self) -> Dict[str, str]:
+        """
+        Creates a mapping from IDs to their types, including attributes and methods.
+        """
+        mapping = {}
+        for element in self.data['elements'].values():
+            mapping[element['id']] = element['type']
+        for relation in self.data['relationships'].values():
+            mapping[relation['id']] = relation['type']
+        return mapping