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_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/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()

modules/modeling/module_modeling_llm/module_modeling_llm/core/generate_suggestions.py

@@ -1,4 +1,4 @@
-from athena.schemas.grading_criterion import StructuredGradingCriterion
+from athena.schemas.structured_grading_criterion import StructuredGradingCriterion
 from langchain_core.output_parsers import PydanticOutputParser
 from langchain_core.prompts import ChatPromptTemplate
 

modules/modeling/module_modeling_llm/module_modeling_llm/core/get_structured_grading_instructions.py

@@ -1,9 +1,13 @@
-from typing import List, Optional
+import hashlib
+import json
+from typing import Any, Dict, List, Optional
+from athena import logger
 from athena.metadata import emit_meta
+from athena.storage.structured_grading_criterion_storage import get_structured_grading_criterion, store_structured_grading_criterion
 from langchain_core.output_parsers import PydanticOutputParser
 from langchain_core.prompts import ChatPromptTemplate
 
-from athena.schemas.grading_criterion import GradingCriterion, StructuredGradingCriterion
+from athena.schemas import GradingCriterion, StructuredGradingCriterion
 from llm_core.utils.predict_and_parse import predict_and_parse
 from module_modeling_llm.config import BasicApproachConfig
 from module_modeling_llm.models.exercise_model import ExerciseModel
@@ -19,6 +23,13 @@ async def get_structured_grading_instructions(
 
     if grading_criteria:
         return StructuredGradingCriterion(criteria=grading_criteria)
+
+    # Check if we have cached instructions for this exercise
+    current_hash = get_grading_instructions_hash(exercise_model)
+    cached_instructions = get_structured_grading_criterion(exercise_model.exercise_id, current_hash)
+    if cached_instructions:
+        print("Using cached instructions")
+        return cached_instructions
 
     chat_prompt = ChatPromptTemplate.from_messages([
         ("system", config.generate_suggestions_prompt.structured_grading_instructions_system_message),
@@ -53,5 +64,22 @@ async def get_structured_grading_instructions(
 
     if not grading_instruction_result:
         raise ValueError("No structured grading instructions were returned by the model.")
+
+    # Cache the grading instructions
+    hash = get_grading_instructions_hash(exercise_model)
+    store_structured_grading_criterion(exercise_model.exercise_id, hash, grading_instruction_result)
+
+    return grading_instruction_result
+
+def get_grading_instructions_hash(exercise: ExerciseModel) -> str:
+
+    hashable_data = {
+        "problem_statement": exercise.problem_statement,
+        "grading_instructions": exercise.grading_instructions,
+        "sample_solution": exercise.transformed_example_solution,
+        "max_points": exercise.max_points,
+        "bonus_points": exercise.bonus_points,
+    }
 
-    return grading_instruction_result
+    json_string = json.dumps(hashable_data, sort_keys=True, default=str)
+    return hashlib.sha256(json_string.encode()).hexdigest()