Programming LLM: Automatic Feedback for Programming Exercises (#234)

athena/athena/endpoints.py

@@ -1,6 +1,6 @@
 # type: ignore # too much weird behavior of mypy with decorators
 import inspect
-from fastapi import Depends, BackgroundTasks
+from fastapi import Depends, BackgroundTasks, Body
 from pydantic import BaseModel, ValidationError
 from typing import TypeVar, Callable, List, Union, Any, Coroutine, Type
 
@@ -14,10 +14,10 @@
 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
 
-
 E = TypeVar('E', bound=Exercise)
 S = TypeVar('S', bound=Submission)
 F = TypeVar('F', bound=Feedback)
+G = TypeVar('G', bound=bool)
 
 # Config type
 C = TypeVar("C", bound=BaseModel)
@@ -75,7 +75,7 @@ async def wrapper(
             exercise: exercise_type,
             submissions: List[submission_type],
             module_config: module_config_type = Depends(get_dynamic_module_config_factory(module_config_type))):
-        
+
         # Retrieve existing metadata for the exercise and submissions
         exercise_meta = get_stored_exercise_meta(exercise) or {}
         exercise_meta.update(exercise.meta)
@@ -152,7 +152,7 @@ class SubmissionSelectorRequest(BaseModel):
         exercise: exercise_type
         submission_ids: List[int]
         module_config: module_config_type = Depends(get_dynamic_module_config_factory(module_config_type))
-    
+
         class Config:
             # Allow camelCase field names in the API (converted to snake_case)
             alias_generator = to_camel
@@ -270,7 +270,8 @@ def feedback_provider(func: Union[
     Callable[[E, S], List[F]],
     Callable[[E, S], Coroutine[Any, Any, List[F]]],
     Callable[[E, S, C], List[F]],
-    Callable[[E, S, C], Coroutine[Any, Any, List[F]]]
+    Callable[[E, S, C], Coroutine[Any, Any, List[F]]],
+    Callable[[E, S, G, C], List[F]],
 ]):
     """
     Provide feedback to the Assessment Module Manager.
@@ -302,15 +303,17 @@ def feedback_provider(func: Union[
     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
 
     @app.post("/feedback_suggestions", responses=module_responses)
     @authenticated
     @with_meta
     async def wrapper(
             exercise: exercise_type,
             submission: submission_type,
+            is_graded: is_graded_type = Body(False),
             module_config: module_config_type = Depends(get_dynamic_module_config_factory(module_config_type))):
-        
+
         # Retrieve existing metadata for the exercise, submission and feedback
         exercise.meta.update(get_stored_exercise_meta(exercise) or {})
         submission.meta.update(get_stored_submission_meta(submission) or {})
@@ -322,6 +325,9 @@ async def wrapper(
         if "module_config" in inspect.signature(func).parameters:
             kwargs["module_config"] = module_config
 
+        if "is_graded" in inspect.signature(func).parameters:
+            kwargs["is_graded"] = is_graded
+
         # Call the actual provider
         if inspect.iscoroutinefunction(func):
             feedbacks = await func(exercise, submission, **kwargs)
@@ -399,11 +405,11 @@ def evaluation_provider(func: Union[
     @authenticated
     @with_meta
     async def wrapper(
-            exercise: exercise_type, 
-            submission: submission_type, 
-            true_feedbacks: List[feedback_type], 
+            exercise: exercise_type,
+            submission: submission_type,
+            true_feedbacks: List[feedback_type],
             predicted_feedbacks: List[feedback_type],
-        ):
+    ):
         # Retrieve existing metadata for the exercise, submission and feedback
         exercise.meta.update(get_stored_exercise_meta(exercise) or {})
         submission.meta.update(get_stored_submission_meta(submission) or {})
@@ -417,4 +423,4 @@ async def wrapper(
             evaluation = func(exercise, submission, true_feedbacks, predicted_feedbacks)
 
         return evaluation
-    return wrapper
+    return wrapper

athena/athena/models/db_feedback.py

@@ -13,6 +13,7 @@ class DBFeedback(Model):
     description = Column(String)
     credits = Column(Float, nullable=False)
     structured_grading_instruction_id = Column(BigInteger)
+    is_graded = Column(Boolean, nullable=True)
     meta = Column(JSON, nullable=False)
 
     # not in the schema, but used in the database to distinguish between feedbacks and feedback suggestions

athena/athena/models/db_programming_exercise.py

@@ -9,9 +9,9 @@ class DBProgrammingExercise(DBExercise, Base):
     __tablename__ = "programming_exercises"
 
     programming_language: str = Column(String, nullable=False)  # type: ignore
-    solution_repository_url: str = Column(String, nullable=False) # type: ignore
-    template_repository_url: str = Column(String, nullable=False) # type: ignore
-    tests_repository_url: 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")

athena/athena/models/db_programming_submission.py

@@ -8,7 +8,7 @@
 
 class DBProgrammingSubmission(DBSubmission, Base):
     __tablename__ = "programming_submissions"
-    repository_url: str = Column(String, nullable=False)  # type: ignore
+    repository_uri: str = Column(String, nullable=False)  # type: ignore
 
     exercise_id = Column(BigIntegerWithAutoincrement, ForeignKey("programming_exercises.id", ondelete="CASCADE"), index=True)
 

athena/athena/schemas/feedback.py

@@ -20,6 +20,9 @@ class Feedback(Schema, ABC):
     structured_grading_instruction_id: Optional[int] = Field(None,
                                                              description="The id of the structured grading instruction that this feedback belongs to.",
                                                              example=1)
+    is_graded: Optional[bool] = Field(None,
+                                                             description="Graded or non graded.",
+                                                             example=False)
 
     meta: dict = Field({}, example={})
 

athena/athena/schemas/programming_exercise.py

@@ -13,42 +13,42 @@ class ProgrammingExercise(Exercise):
     type: ExerciseType = Field(ExerciseType.programming, const=True)
 
     programming_language: str = Field(description="The programming language that is used for this exercise.", example="java")
-    solution_repository_url: AnyUrl = Field(description="URL to the solution git repository, which contains the "
+    solution_repository_uri: AnyUrl = Field(description="URL to the solution git repository, which contains the "
                                                         "reference solution.",
                                             example="http://localhost:3000/api/example-solutions/1")
-    template_repository_url: AnyUrl = Field(description="URL to the template git repository, which is the starting "
+    template_repository_uri: AnyUrl = Field(description="URL to the template git repository, which is the starting "
                                                         "point for students.",
                                             example="http://localhost:3000/api/example-template/1")
-    tests_repository_url: AnyUrl = Field(description="URL to the tests git repository, which contains the tests that "
+    tests_repository_uri: AnyUrl = Field(description="URL to the tests git repository, which contains the tests that "
                                                      "are used to automatically grade the exercise.",
                                          example="http://localhost:3000/api/example-tests/1")
 
 
     def get_solution_zip(self) -> ZipFile:
         """Return the solution repository as a ZipFile object."""
-        return get_repository_zip(self.solution_repository_url)
+        return get_repository_zip(self.solution_repository_uri)
 
 
     def get_solution_repository(self) -> Repo:
         """Return the solution repository as a Repo object."""
-        return get_repository(self.solution_repository_url)
+        return get_repository(self.solution_repository_uri)
 
 
     def get_template_zip(self) -> ZipFile:
         """Return the template repository as a ZipFile object."""
-        return get_repository_zip(self.template_repository_url)
+        return get_repository_zip(self.template_repository_uri)
 
 
     def get_template_repository(self) -> Repo:
         """Return the template repository as a Repo object."""
-        return get_repository(self.template_repository_url)
+        return get_repository(self.template_repository_uri)
 
 
     def get_tests_zip(self) -> ZipFile:
         """Return the tests repository as a ZipFile object."""
-        return get_repository_zip(self.tests_repository_url)
+        return get_repository_zip(self.tests_repository_uri)
 
 
     def get_tests_repository(self) -> Repo:
         """Return the tests repository as a Repo object."""
-        return get_repository(self.tests_repository_url)
+        return get_repository(self.tests_repository_uri)

athena/athena/schemas/programming_submission.py

@@ -8,17 +8,17 @@
 
 class ProgrammingSubmission(Submission):
     """Submission on a programming exercise."""
-    repository_url: str = Field(example="https://lms.example.com/assignments/1/submissions/1/download")
+    repository_uri: str = Field(example="https://lms.example.com/assignments/1/submissions/1/download")
 
 
     def get_zip(self) -> ZipFile:
         """Return the submission repository as a ZipFile object."""
-        return get_repository_zip(self.repository_url)
+        return get_repository_zip(self.repository_uri)
 
 
     def get_repository(self) -> Repo:
         """Return the submission repository as a Repo object."""
-        return get_repository(self.repository_url)
+        return get_repository(self.repository_uri)
 
     def get_code(self, file_path: str) -> str:
         """

module_programming_llm/module_programming_llm/__main__.py

@@ -2,12 +2,23 @@
 
 import tiktoken
 
-from athena import app, submission_selector, submissions_consumer, feedback_consumer, feedback_provider
+from athena import (
+    app,
+    submission_selector,
+    submissions_consumer,
+    feedback_consumer,
+    feedback_provider,
+)
 from athena.programming import Exercise, Submission, Feedback
 from athena.logger import logger
 from module_programming_llm.config import Configuration
 
-from module_programming_llm.generate_suggestions_by_file import generate_suggestions_by_file
+from module_programming_llm.generate_graded_suggestions_by_file import (
+    generate_suggestions_by_file as generate_graded_suggestions_by_file,
+)
+from module_programming_llm.generate_non_graded_suggestions_by_file import (
+    generate_suggestions_by_file as generate_non_graded_suggestions_by_file,
+)
 
 
 @submissions_consumer
@@ -27,12 +38,18 @@ def process_incoming_feedback(exercise: Exercise, submission: Submission, feedba
 
 
 @feedback_provider
-async def suggest_feedback(exercise: Exercise, submission: Submission, module_config: Configuration) -> List[Feedback]:
-    logger.info("suggest_feedback: Suggestions for submission %d of exercise %d were requested", submission.id, exercise.id)
-    return await generate_suggestions_by_file(exercise, submission, module_config.approach, module_config.debug)
+async def suggest_feedback(exercise: Exercise, submission: Submission, is_graded: bool, module_config: Configuration) -> List[Feedback]:
+    logger.info("suggest_feedback: %s suggestions for submission %d of exercise %d were requested",
+                "Graded" if is_graded else "Non-graded", submission.id, exercise.id)
+    if is_graded:
+        return await generate_graded_suggestions_by_file(exercise, submission, module_config.graded_approach,
+                                                         module_config.debug)
+    return await generate_non_graded_suggestions_by_file(exercise, submission, module_config.non_graded_approach,
+                                                             module_config.debug)
+
 
 
 if __name__ == "__main__":
     # Preload for token estimation later
     tiktoken.get_encoding("cl100k_base")
-    app.start()
+    app.start()