f5b627256f
Full backend implementation with custom Odoo modules: - encoach_api: Core API, user management, JWT auth - encoach_exam: Exam generation (reading, writing, listening, speaking) - encoach_evaluate: AI-powered evaluation (writing, speaking) - encoach_training: Training tips and walkthrough - encoach_storage: File storage management - encoach_payment: Stripe, PayPal, Paymob integration - encoach_mail: Email notifications Made-with: Cursor
341 lines
12 KiB
Python
341 lines
12 KiB
Python
import json
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from odoo.addons.encoach_ai.models.constants import GPT_MODELS, TEMPERATURE
|
|
from odoo.addons.encoach_ai.services.openai_service import EncoachOpenAIService
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
WRITING_RESPONSE_TEMPLATE = json.dumps({
|
|
"comment": "comment about student's response quality",
|
|
"overall": 0.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": "..."},
|
|
},
|
|
})
|
|
|
|
SPEAKING_RESPONSE_TEMPLATE = json.dumps({
|
|
"comment": "extensive comment about answer quality",
|
|
"overall": 0.0,
|
|
"task_response": {
|
|
"Fluency and Coherence": {"grade": 0.0, "comment": "extensive comment..."},
|
|
"Lexical Resource": {"grade": 0.0, "comment": "..."},
|
|
"Grammatical Range and Accuracy": {"grade": 0.0, "comment": "..."},
|
|
"Pronunciation": {"grade": 0.0, "comment": "..."},
|
|
},
|
|
})
|
|
|
|
SHORT_ANSWER_TEMPLATE = json.dumps({
|
|
"exercises": [
|
|
{"id": "exercise-id", "correct": True, "correct_answer": "the correct answer"},
|
|
],
|
|
})
|
|
|
|
SUMMARY_TEMPLATE = json.dumps({
|
|
"sections": [
|
|
{
|
|
"code": "section-code",
|
|
"name": "Section Name",
|
|
"grade": 0.0,
|
|
"evaluation": "Detailed evaluation text...",
|
|
"suggestions": "Improvement suggestions...",
|
|
"bullet_points": ["point 1", "point 2"],
|
|
},
|
|
],
|
|
})
|
|
|
|
SPEAKING_TASK_INSTRUCTIONS = {
|
|
1: (
|
|
'Address the student as "you". '
|
|
"If the answers are not 2 or 3 sentences long, "
|
|
"warn the student that they should be."
|
|
),
|
|
2: 'Address the student as "you".',
|
|
3: (
|
|
'Address the student as "you" and pay special attention '
|
|
"to coherence between the answers."
|
|
),
|
|
}
|
|
|
|
|
|
class EncoachGradingService:
|
|
|
|
def __init__(self, env):
|
|
self.env = env
|
|
self.ai = EncoachOpenAIService(env)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Writing grading
|
|
# ------------------------------------------------------------------
|
|
|
|
def grade_writing(self, question, answer, task, attachment=None):
|
|
"""Grade a writing answer using GPT-4o with IELTS writing rubric.
|
|
|
|
Returns dict with comment, overall, task_response, fixed_text,
|
|
perfect_answer, and ai_detection results.
|
|
"""
|
|
system_msg = (
|
|
"You are a helpful assistant designed to output JSON "
|
|
f"on this format: {WRITING_RESPONSE_TEMPLATE}"
|
|
)
|
|
|
|
if task == 1:
|
|
user_prompt = (
|
|
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.\n"
|
|
'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".\n'
|
|
f'Question: "{question}"\nAnswer: "{answer}"'
|
|
)
|
|
else:
|
|
user_prompt = (
|
|
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.\n"
|
|
f'Question: "{question}"\nAnswer: "{answer}"'
|
|
)
|
|
|
|
messages = [
|
|
{"role": "system", "content": system_msg},
|
|
]
|
|
|
|
if attachment and task == 1:
|
|
messages.append({
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": user_prompt},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": attachment},
|
|
},
|
|
],
|
|
})
|
|
else:
|
|
messages.append({"role": "user", "content": user_prompt})
|
|
|
|
result = self.ai.prediction(
|
|
model=GPT_MODELS["grading"],
|
|
messages=messages,
|
|
temperature=TEMPERATURE["grading"],
|
|
fields_to_check=["comment"],
|
|
)
|
|
if not result:
|
|
return None
|
|
|
|
result["fixed_text"] = self._get_fixed_text(answer)
|
|
result["perfect_answer"] = self._get_perfect_answer(question)
|
|
result["ai_detection"] = self._detect_ai(answer)
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# Speaking grading
|
|
# ------------------------------------------------------------------
|
|
|
|
def grade_speaking(self, task, qa_pairs):
|
|
"""Grade speaking using GPT-4o with IELTS speaking rubric.
|
|
|
|
qa_pairs: list of dicts with 'question' and 'answer' keys.
|
|
"""
|
|
system_msg = (
|
|
"You are a helpful assistant designed to output JSON "
|
|
f"on this format: {SPEAKING_RESPONSE_TEMPLATE}"
|
|
)
|
|
|
|
qa_text = "\n".join(
|
|
f'Question: "{p["question"]}"\nAnswer: "{p["answer"]}"'
|
|
for p in qa_pairs
|
|
)
|
|
task_instruction = SPEAKING_TASK_INSTRUCTIONS.get(task, "")
|
|
|
|
user_prompt = (
|
|
f"Evaluate the given Speaking Part {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 detailed commentary highlighting both "
|
|
f"strengths and weaknesses in the response. {task_instruction}\n"
|
|
f"{qa_text}"
|
|
)
|
|
|
|
messages = [
|
|
{"role": "system", "content": system_msg},
|
|
{"role": "user", "content": user_prompt},
|
|
]
|
|
|
|
result = self.ai.prediction(
|
|
model=GPT_MODELS["grading"],
|
|
messages=messages,
|
|
temperature=TEMPERATURE["grading"],
|
|
fields_to_check=["comment"],
|
|
)
|
|
if not result:
|
|
return None
|
|
|
|
if task == 2 and qa_pairs:
|
|
combined_answer = " ".join(p["answer"] for p in qa_pairs)
|
|
result["fixed_text"] = self._get_fixed_text(combined_answer)
|
|
result["perfect_answer"] = self._get_perfect_answer(
|
|
qa_pairs[0]["question"]
|
|
)
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# Short-answer grading
|
|
# ------------------------------------------------------------------
|
|
|
|
def grade_short_answers(self, text, questions, answers):
|
|
"""Evaluate short answers against a reading/listening passage."""
|
|
system_msg = (
|
|
"You are a helpful assistant designed to output JSON "
|
|
f"on this format: {SHORT_ANSWER_TEMPLATE}"
|
|
)
|
|
|
|
qa_text = "\n".join(
|
|
f'Q{i + 1}: "{q}" — Student answer: "{a}"'
|
|
for i, (q, a) in enumerate(zip(questions, answers))
|
|
)
|
|
|
|
user_prompt = (
|
|
"Evaluate each student answer against the passage. For each answer, "
|
|
"determine if it is correct and provide the correct answer.\n\n"
|
|
f'Passage: "{text}"\n\n{qa_text}'
|
|
)
|
|
|
|
messages = [
|
|
{"role": "system", "content": system_msg},
|
|
{"role": "user", "content": user_prompt},
|
|
]
|
|
|
|
return self.ai.prediction(
|
|
model=GPT_MODELS["grading"],
|
|
messages=messages,
|
|
temperature=TEMPERATURE["grading"],
|
|
check_blacklisted=False,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Grading summary
|
|
# ------------------------------------------------------------------
|
|
|
|
def generate_grading_summary(self, sections):
|
|
"""Generate a summary for an entire exam session using GPT-3.5-turbo."""
|
|
system_msg = (
|
|
"You are a helpful assistant designed to output JSON "
|
|
f"on this format: {SUMMARY_TEMPLATE}"
|
|
)
|
|
|
|
user_prompt = (
|
|
"Generate a detailed evaluation summary for the following IELTS exam "
|
|
"sections. For each section, provide an evaluation, suggestions for "
|
|
"improvement, and key bullet points.\n\n"
|
|
+ json.dumps(sections)
|
|
)
|
|
|
|
messages = [
|
|
{"role": "system", "content": system_msg},
|
|
{"role": "user", "content": user_prompt},
|
|
]
|
|
|
|
return self.ai.prediction(
|
|
model=GPT_MODELS["secondary"],
|
|
messages=messages,
|
|
temperature=TEMPERATURE["grading"],
|
|
check_blacklisted=False,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# AI detection (GPTZero)
|
|
# ------------------------------------------------------------------
|
|
|
|
def _detect_ai(self, text):
|
|
"""Call GPTZero API to detect AI-generated text."""
|
|
api_key = (
|
|
self.env["ir.config_parameter"]
|
|
.sudo()
|
|
.get_param("encoach.gptzero_api_key", "")
|
|
)
|
|
if not api_key:
|
|
_logger.warning("GPTZero API key not configured")
|
|
return None
|
|
|
|
try:
|
|
resp = httpx.post(
|
|
"https://api.gptzero.me/v2/predict/text",
|
|
headers={
|
|
"x-api-key": api_key,
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={
|
|
"document": text,
|
|
"version": "",
|
|
"multilingual": False,
|
|
},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception:
|
|
_logger.exception("GPTZero API call failed")
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helper prompts
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get_perfect_answer(self, question):
|
|
messages = [
|
|
{"role": "system", "content": "You are an IELTS writing expert."},
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"Write a perfect answer for this IELTS writing task: "
|
|
f'"{question}"'
|
|
),
|
|
},
|
|
]
|
|
result = self.ai.prediction(
|
|
model=GPT_MODELS["secondary"],
|
|
messages=messages,
|
|
temperature=TEMPERATURE["grading"],
|
|
response_format={"type": "json_object"},
|
|
check_blacklisted=False,
|
|
)
|
|
if result and "answer" in result:
|
|
return result["answer"]
|
|
return result
|
|
|
|
def _get_fixed_text(self, text):
|
|
messages = [
|
|
{"role": "system", "content": "You are a grammar correction assistant. Output JSON."},
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"Fix the grammatical and spelling errors in this text, "
|
|
f'keeping the original meaning: "{text}"'
|
|
),
|
|
},
|
|
]
|
|
result = self.ai.prediction(
|
|
model=GPT_MODELS["secondary"],
|
|
messages=messages,
|
|
temperature=TEMPERATURE["grading"],
|
|
response_format={"type": "json_object"},
|
|
check_blacklisted=False,
|
|
)
|
|
if result and "text" in result:
|
|
return result["text"]
|
|
return result
|