import asyncio from typing import List, Dict from app.services.abc import IWritingService, ILLMService, IAIDetectorService from app.configs.constants import GPTModels, TemperatureSettings, FieldsAndExercises from app.helpers import TextHelper, ExercisesHelper class WritingService(IWritingService): def __init__(self, llm: ILLMService, ai_detector: IAIDetectorService): self._llm = llm self._ai_detector = ai_detector async def get_writing_task_general_question(self, task: int, topic: str, difficulty: str): messages = [ { "role": "system", "content": ( 'You are a helpful assistant designed to output JSON on this format: {"prompt": "prompt content"}' ) }, *self._get_writing_args(task, topic, difficulty) ] llm_model = GPTModels.GPT_3_5_TURBO if task == 1 else GPTModels.GPT_4_O response = await self._llm.prediction( llm_model, messages, ["prompt"], TemperatureSettings.GEN_QUESTION_TEMPERATURE ) question = response["prompt"].strip() return { "question": self._add_newline_before_hyphen(question) if task == 1 else question, "difficulty": difficulty, "topic": topic } @staticmethod def _get_writing_args(task: int, topic: str, difficulty: str) -> List[Dict]: writing_args = { "1": { "prompt": ( 'Craft a prompt for an IELTS Writing Task 1 General Training exercise that instructs the ' 'student to compose a letter. The prompt should present a specific scenario or situation, ' f'based on the topic of "{topic}", requiring the student to provide information, ' 'advice, or instructions within the letter. Make sure that the generated prompt is ' f'of {difficulty} difficulty and does not contain forbidden subjects in muslim countries.' ), "instructions": ( 'The prompt should end with "In the letter you should" followed by 3 bullet points of what ' 'the answer should include.' ) }, "2": { # TODO: Should the muslim disclaimer be here as well? "prompt": ( f'Craft a comprehensive question of {difficulty} difficulty like the ones for IELTS ' 'Writing Task 2 General Training that directs the candidate to delve into an in-depth ' f'analysis of contrasting perspectives on the topic of "{topic}".' ), "instructions": ( 'The question should lead to an answer with either "theories", "complicated information" or ' 'be "very descriptive" on the topic.' ) } } messages = [ { "role": "user", "content": writing_args[str(task)]["prompt"] }, { "role": "user", "content": writing_args[str(task)]["instructions"] } ] return messages async def grade_writing_task(self, task: int, question: str, answer: str): bare_minimum = 100 if task == 1 else 180 if not TextHelper.has_words(answer): return self._zero_rating("The answer does not contain enough english words.") elif not TextHelper.has_x_words(answer, bare_minimum): return self._zero_rating("The answer is insufficient and too small to be graded.") else: template = self._get_writing_template() messages = [ { "role": "system", "content": ( f'You are a helpful assistant designed to output JSON on this format: {template}' ) }, { "role": "user", "content": ( f'Evaluate the given Writing Task {task} response based on the IELTS grading system, ' 'ensuring a strict assessment that penalizes errors. Deduct points for deviations ' 'from the task, and assign a score of 0 if the response fails to address the question. ' 'Additionally, provide a detailed commentary highlighting both strengths and ' 'weaknesses in the response. ' f'\n Question: "{question}" \n Answer: "{answer}"') } ] if task == 1: messages.append({ "role": "user", "content": ( 'Refer to the parts of the letter as: "Greeting Opener", "bullet 1", "bullet 2", ' '"bullet 3", "closer (restate the purpose of the letter)", "closing greeting"' ) }) llm_model = GPTModels.GPT_3_5_TURBO if task == 1 else GPTModels.GPT_4_O temperature = ( TemperatureSettings.GRADING_TEMPERATURE if task == 1 else TemperatureSettings.GEN_QUESTION_TEMPERATURE ) evaluation_promise = self._llm.prediction( llm_model, messages, ["comment"], temperature ) perfect_answer_minimum = 150 if task == 1 else 250 perfect_answer_promise = self._get_perfect_answer(question, perfect_answer_minimum) fixed_text_promise = self._get_fixed_text(answer) ai_detection_promise = self._ai_detector.run_detection(answer) prediction_result, perfect_answer_result, fixed_text_result, ai_detection_result = await asyncio.gather( evaluation_promise, perfect_answer_promise, fixed_text_promise, ai_detection_promise ) response = prediction_result response["perfect_answer"] = perfect_answer_result["perfect_answer"] response["overall"] = ExercisesHelper.fix_writing_overall( response["overall"], response["task_response"] ) response['fixed_text'] = fixed_text_result if ai_detection_result is not None: response['ai_detection'] = ai_detection_result return response async def _get_fixed_text(self, text): messages = [ { "role": "system", "content": ( 'You are a helpful assistant designed to output JSON on this format: ' '{"fixed_text": "fixed test with no misspelling errors"}' ) }, { "role": "user", "content": ( 'Fix the errors in the given text and put it in a JSON. ' f'Do not complete the answer, only replace what is wrong. \n The text: "{text}"' ) } ] response = await self._llm.prediction( GPTModels.GPT_3_5_TURBO, messages, ["fixed_text"], 0.2, False ) return response["fixed_text"] async def _get_perfect_answer(self, question: str, size: int) -> Dict: messages = [ { "role": "system", "content": ( 'You are a helpful assistant designed to output JSON on this format: ' '{"perfect_answer": "perfect answer for the question"}' ) }, { "role": "user", "content": f'Write a perfect answer for this writing exercise of a IELTS exam. Question: {question}' }, { "role": "user", "content": f'The answer must have at least {size} words' } ] return await self._llm.prediction( GPTModels.GPT_4_O, messages, ["perfect_answer"], TemperatureSettings.GEN_QUESTION_TEMPERATURE ) @staticmethod def _zero_rating(comment: str): return { 'comment': comment, 'overall': 0, 'task_response': { 'Task Achievement': { "grade": 0.0, "comment": "" }, 'Coherence and Cohesion': { "grade": 0.0, "comment": "" }, 'Lexical Resource': { "grade": 0.0, "comment": "" }, 'Grammatical Range and Accuracy': { "grade": 0.0, "comment": "" } } } @staticmethod def _get_writing_template(): return { "comment": "comment about student's response quality", "overall": 0.0, "task_response": { "Task Achievement": { "grade": 0.0, "comment": "comment about Task Achievement of the student's response" }, "Coherence and Cohesion": { "grade": 0.0, "comment": "comment about Coherence and Cohesion of the student's response" }, "Lexical Resource": { "grade": 0.0, "comment": "comment about Lexical Resource of the student's response" }, "Grammatical Range and Accuracy": { "grade": 0.0, "comment": "comment about Grammatical Range and Accuracy of the student's response" } } } @staticmethod def _add_newline_before_hyphen(s): return s.replace(" -", "\n-")