326 lines
13 KiB
Python
326 lines
13 KiB
Python
import json
|
||
import os
|
||
import re
|
||
from typing import List
|
||
from openai import OpenAI
|
||
from sqlalchemy.orm import Session
|
||
from . import models
|
||
from .curriculum import QUESTIONS, SKILLS
|
||
from .program_content import get_lesson
|
||
|
||
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
|
||
|
||
|
||
SYSTEM_PROMPT = """
|
||
Tu es Professeur TOP, un professeur virtuel français pour enfants de 8 à 12 ans.
|
||
Règles :
|
||
- Tu parles toujours en français simple et chaleureux.
|
||
- Tu donnes des explications très courtes, puis un mini exemple.
|
||
- Tu tiens compte du niveau de l'élève et de ses faiblesses indiquées dans le contexte.
|
||
- Tu n'inventes pas la progression : elle est fournie dans le contexte.
|
||
- Tu encourages sans infantiliser.
|
||
- Quand l'élève se trompe, tu expliques calmement puis proposes une question très simple.
|
||
- Tu enseignes principalement le programme national français niveau primaire/cycle 3.
|
||
""".strip()
|
||
|
||
TURN_TAKING_PROMPT = """
|
||
Consignes de rythme oral:
|
||
- Fais des tours de parole courts: 1 a 3 phrases maximum, sauf demande explicite.
|
||
- Termine souvent par une question ou une consigne courte pour laisser l'enfant parler.
|
||
- Ne fais pas de longs monologues: une idee a la fois.
|
||
- Si l’élève t’interrompt, arrête-toi, écoute son message, puis reprends seulement si c’est utile.
|
||
- Si l'interruption semble accidentelle ou hors sujet, reponds tres brievement puis continue.
|
||
""".strip()
|
||
|
||
TTS_PROFILES = [
|
||
{
|
||
"id": "rigolote",
|
||
"label": "Rigolote",
|
||
"description": "La plus joueuse et cartoon.",
|
||
"voice": "coral",
|
||
"speed": 1.08,
|
||
"instructions": (
|
||
"Parle en français avec une énergie joyeuse, malicieuse et très expressive. "
|
||
"Tu es Professeur TOP, un professeur amusant, chaleureux et un peu théâtral. "
|
||
"Le ton doit rester clair pour un enfant, avec des intonations souriantes, "
|
||
"des fins de phrases dynamiques et une diction très vivante."
|
||
),
|
||
},
|
||
{
|
||
"id": "petillante",
|
||
"label": "Pétillante",
|
||
"description": "Enjouée, dynamique et encourageante.",
|
||
"voice": "shimmer",
|
||
"speed": 1.03,
|
||
"instructions": (
|
||
"Parle en français avec une voix lumineuse, motivante et positive. "
|
||
"Tu es Professeur TOP: enthousiaste, rassurant et très engageant, "
|
||
"sans caricature excessive. Garde une diction nette et chaleureuse."
|
||
),
|
||
},
|
||
{
|
||
"id": "douce",
|
||
"label": "Douce",
|
||
"description": "Calme, rassurante et patiente.",
|
||
"voice": "sage",
|
||
"speed": 0.98,
|
||
"instructions": (
|
||
"Parle en français avec une voix douce, calme et très rassurante. "
|
||
"Tu es Professeur TOP dans une version posée, bienveillante et patiente. "
|
||
"Le rythme est fluide, jamais pressé, avec une intonation apaisante."
|
||
),
|
||
},
|
||
{
|
||
"id": "sobre",
|
||
"label": "Sobre",
|
||
"description": "La plus neutre et sérieuse.",
|
||
"voice": "alloy",
|
||
"speed": 1.0,
|
||
"instructions": (
|
||
"Parle en français avec une voix claire, naturelle et sobre. "
|
||
"Tu es Professeur TOP dans une version très lisible, professionnelle et mesurée. "
|
||
"Reste chaleureux, mais sans effet théâtral."
|
||
),
|
||
},
|
||
]
|
||
|
||
|
||
def seed_skills(db: Session) -> None:
|
||
for skill in SKILLS:
|
||
existing = db.query(models.Skill).filter(models.Skill.code == skill["code"]).first()
|
||
if not existing:
|
||
db.add(models.Skill(**skill))
|
||
db.commit()
|
||
|
||
|
||
def ensure_student_mastery(db: Session, student: models.Student) -> None:
|
||
all_skills = db.query(models.Skill).all()
|
||
for skill in all_skills:
|
||
found = (
|
||
db.query(models.StudentSkillMastery)
|
||
.filter_by(student_id=student.id, skill_id=skill.id)
|
||
.first()
|
||
)
|
||
if not found:
|
||
db.add(models.StudentSkillMastery(student_id=student.id, skill_id=skill.id))
|
||
db.commit()
|
||
|
||
|
||
def get_student_context(db: Session, student_id: int, include_history: bool = True) -> str:
|
||
student = db.query(models.Student).filter_by(id=student_id).first()
|
||
mastery = (
|
||
db.query(models.StudentSkillMastery, models.Skill)
|
||
.join(models.Skill, models.Skill.id == models.StudentSkillMastery.skill_id)
|
||
.filter(models.StudentSkillMastery.student_id == student_id)
|
||
.all()
|
||
)
|
||
recent_messages = []
|
||
if include_history:
|
||
recent_messages = (
|
||
db.query(models.Message)
|
||
.filter_by(student_id=student_id)
|
||
.order_by(models.Message.created_at.desc())
|
||
.limit(6)
|
||
.all()
|
||
)
|
||
|
||
assignment = db.query(models.StudentProgramAssignment).filter_by(student_id=student_id).first()
|
||
lesson = get_lesson(assignment.lesson_id) if assignment else None
|
||
|
||
lines: List[str] = [
|
||
f"Élève: {student.first_name}, {student.age} ans, classe {student.grade}.",
|
||
"Progression par compétence:",
|
||
]
|
||
if lesson:
|
||
lines.append(
|
||
f"Leçon programme sélectionnée par le professeur: {lesson['title']} ({lesson['subject']}, {lesson['grade']})."
|
||
)
|
||
first_asset = (lesson.get("assets") or [None])[0]
|
||
if first_asset:
|
||
section_titles = [section["title"] for section in first_asset.get("sections", [])]
|
||
if section_titles:
|
||
lines.append("Étapes de la première fiche: " + ", ".join(section_titles) + ".")
|
||
for mastery_row, skill in mastery:
|
||
lines.append(
|
||
f"- {skill.label}: score={mastery_row.mastery_score:.1f}, confiance={mastery_row.confidence:.2f}, preuves={mastery_row.evidence_count}"
|
||
)
|
||
if include_history:
|
||
lines.append("Historique récent:")
|
||
for message in reversed(recent_messages):
|
||
lines.append(f"- {message.role}: {message.content}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def build_llm_reply(db: Session, student_id: int, user_message: str) -> str:
|
||
context = get_student_context(db, student_id)
|
||
response = client.responses.create(
|
||
model="gpt-4.1-mini",
|
||
input=[
|
||
{"role": "system", "content": SYSTEM_PROMPT},
|
||
{"role": "system", "content": TURN_TAKING_PROMPT},
|
||
{
|
||
"role": "user",
|
||
"content": f"Contexte pédagogique:\n{context}\n\nMessage de l'élève:\n{user_message}",
|
||
},
|
||
],
|
||
temperature=0.7,
|
||
)
|
||
return response.output_text.strip()
|
||
|
||
|
||
def build_lesson_reply(db: Session, student_id: int, user_message: str, lesson_context: str) -> dict:
|
||
context = get_student_context(db, student_id, include_history=False)
|
||
response = client.responses.create(
|
||
model="gpt-4.1-mini",
|
||
input=[
|
||
{"role": "system", "content": SYSTEM_PROMPT},
|
||
{"role": "system", "content": TURN_TAKING_PROMPT},
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"Tu reçois un contexte courant de leçon lié à la card affichée. "
|
||
"Tu dois parler uniquement de cette card, sauf si l'élève pose une question de clarification immédiate. "
|
||
"N'utilise que les nombres, mots de vocabulaire et actions explicitement autorisés dans le contexte de la card. "
|
||
"Si un nombre n'est pas affiché ou autorisé dans ce contexte, tu n'en parles pas. "
|
||
"Ne propose jamais 'un autre nombre' si le contexte ne donne pas cet autre nombre. "
|
||
"Si l'élève veut continuer mais que la card courante est terminée, mets `advance_lesson_step` à true. "
|
||
"N'utilise pas l'historique des séances précédentes pour choisir le thème de ta réponse. "
|
||
"Ta réponse visible doit être très courte: maximum 24 mots, 1 phrase d'explication "
|
||
"puis éventuellement 1 question courte. Évite les phrases longues avec plusieurs propositions. "
|
||
"Si la réponse de l'élève montre clairement qu'il a compris la card courante, mets "
|
||
"`advance_lesson_step` à true. Sinon, mets-le à false. "
|
||
"Réponds strictement en JSON valide avec les clés `reply` et `advance_lesson_step`. "
|
||
"`reply` contient uniquement ce que Professeur TOP dit à l'élève, sans JSON visible."
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": (
|
||
f"Contexte pédagogique général:\n{context}\n\n"
|
||
f"Contexte courant de la card affichée:\n{lesson_context}\n\n"
|
||
f"Message de l'élève:\n{user_message}"
|
||
),
|
||
},
|
||
],
|
||
temperature=0.4,
|
||
)
|
||
raw_text = response.output_text.strip()
|
||
json_text = raw_text
|
||
if json_text.startswith("```"):
|
||
json_text = re.sub(r"^```(?:json)?\s*", "", json_text)
|
||
json_text = re.sub(r"\s*```$", "", json_text)
|
||
if "{" in json_text and "}" in json_text:
|
||
json_text = json_text[json_text.find("{") : json_text.rfind("}") + 1]
|
||
try:
|
||
parsed = json.loads(json_text)
|
||
except json.JSONDecodeError:
|
||
return {"reply": raw_text, "advance_lesson_step": False}
|
||
|
||
reply = str(parsed.get("reply") or "").strip()
|
||
advance_value = parsed.get("advance_lesson_step")
|
||
advance_lesson_step = (
|
||
advance_value
|
||
if isinstance(advance_value, bool)
|
||
else str(advance_value).strip().lower() in {"true", "oui", "yes", "1"}
|
||
)
|
||
return {
|
||
"reply": reply or raw_text,
|
||
"advance_lesson_step": advance_lesson_step,
|
||
}
|
||
|
||
|
||
def list_tts_profiles() -> list[dict]:
|
||
return [
|
||
{
|
||
"id": profile["id"],
|
||
"label": profile["label"],
|
||
"description": profile["description"],
|
||
}
|
||
for profile in TTS_PROFILES
|
||
]
|
||
|
||
|
||
def synthesize_speech(text: str, profile_id: str) -> bytes:
|
||
clean_text = text.strip()
|
||
if not clean_text:
|
||
raise ValueError("Texte à synthétiser manquant")
|
||
|
||
profile = next((item for item in TTS_PROFILES if item["id"] == profile_id), None)
|
||
if not profile:
|
||
raise ValueError("Profil de voix inconnu")
|
||
|
||
response = client.audio.speech.create(
|
||
model="gpt-4o-mini-tts",
|
||
voice=profile["voice"],
|
||
input=clean_text[:4096],
|
||
instructions=profile["instructions"],
|
||
response_format="mp3",
|
||
speed=profile["speed"],
|
||
)
|
||
return response.content
|
||
|
||
|
||
def transcribe_audio(filename: str, audio_bytes: bytes, content_type: str | None = None) -> str:
|
||
file_payload = (filename, audio_bytes, content_type or "application/octet-stream")
|
||
transcript = client.audio.transcriptions.create(
|
||
model="gpt-4o-mini-transcribe",
|
||
file=file_payload,
|
||
language="fr",
|
||
prompt=(
|
||
"Transcris uniquement en français. "
|
||
"Le locuteur est un enfant qui repond a un exercice scolaire. "
|
||
"Les phrases peuvent etre tres courtes, par exemple: nous chantons, je lis, 8 plus 5."
|
||
),
|
||
)
|
||
return (getattr(transcript, "text", "") or "").strip()
|
||
|
||
|
||
def pick_next_skill(db: Session, student_id: int) -> models.Skill:
|
||
weakest = (
|
||
db.query(models.StudentSkillMastery)
|
||
.filter_by(student_id=student_id)
|
||
.order_by(models.StudentSkillMastery.mastery_score.asc())
|
||
.first()
|
||
)
|
||
return db.query(models.Skill).filter_by(id=weakest.skill_id).first()
|
||
|
||
|
||
def evaluate_answer(db: Session, student_id: int, skill_code: str, answer: str):
|
||
q = QUESTIONS[skill_code]
|
||
normalized_student = answer.strip().lower()
|
||
normalized_expected = q["expected_answer"].strip().lower()
|
||
correct = normalized_student == normalized_expected
|
||
|
||
skill = db.query(models.Skill).filter_by(code=skill_code).first()
|
||
mastery = (
|
||
db.query(models.StudentSkillMastery)
|
||
.filter_by(student_id=student_id, skill_id=skill.id)
|
||
.first()
|
||
)
|
||
|
||
if correct:
|
||
mastery.mastery_score = min(100.0, mastery.mastery_score + 8)
|
||
mastery.confidence = min(1.0, mastery.confidence + 0.15)
|
||
feedback = q["feedback_ok"]
|
||
else:
|
||
mastery.mastery_score = max(0.0, mastery.mastery_score - 6)
|
||
mastery.confidence = min(1.0, mastery.confidence + 0.1)
|
||
feedback = q["feedback_ko"]
|
||
|
||
mastery.evidence_count += 1
|
||
|
||
db.add(
|
||
models.AssessmentAttempt(
|
||
student_id=student_id,
|
||
skill_code=skill_code,
|
||
question=q["question"],
|
||
expected_answer=q["expected_answer"],
|
||
student_answer=answer,
|
||
is_correct=1 if correct else 0,
|
||
feedback=feedback,
|
||
)
|
||
)
|
||
db.commit()
|
||
db.refresh(mastery)
|
||
return correct, feedback, mastery.mastery_score
|