blizzarman/polyglot-tutor
0
1"""Gradio entrypoint.2 3One tab per skill (wired milestone by milestone) plus a Diagnostics tab that4shows the resolved configuration, pings the configured LLM and runs the CEFR5classifier on demand. M1 ships the Reading tab: level-verified generated6texts (or the learner's own text) with judge-gated comprehension questions.7 8UI note: the question block uses `@gr.render`, Gradio's mechanism for dynamic9UI — components are created fresh from state on every change instead of being10patched through update payloads (which proved fragile in the installed Gradio11major version).12"""13 14import gradio as gr15 16from tutor.config import Settings, get_settings17from tutor.ml.cefr.inference import create_cefr_classifier18from tutor.services.asr.base import ASRError19from tutor.services.asr.factory import create_asr_client20from tutor.services.cache import FileCache21from tutor.services.dictation import DictationResult22from tutor.services.llm.base import ChatMessage, LLMError23from tutor.services.llm.factory import create_llm_client24from tutor.services.reading import (25 CEFR_LEVELS,26 ReadingError,27 ReadingExercise,28 build_reading_exercise,29 exercise_from_user_text,30)31 32 33def format_feedback(questions: list[dict], picked: list[str | None]) -> str:34 """Score the learner's answers against the exercise questions (pure, tested)."""35 if any(answer is None for answer in picked):36 return "Answer all the questions first."37 lines, score = [], 038 for index, (question, answer) in enumerate(zip(questions, picked, strict=True)):39 correct = question["options"][question["answer_index"]]40 good = answer == correct41 score += int(good)42 mark = "✅" if good else "❌"43 explanation = question.get("explanation", "")44 lines.append(f"{mark} **Q{index + 1}** — correct answer: **{correct}**. {explanation}")45 header = f"## Score: {score}/{len(questions)}"46 if score == len(questions):47 header += " 🎉"48 return header + "\n\n" + "\n\n".join(lines)49 50 51def _normalize(answer: str) -> str:52 """Casefold + strip accents, so 'Café ' matches 'cafe' for cloze scoring."""53 import unicodedata54 55 stripped = unicodedata.normalize("NFD", answer.strip().casefold())56 return "".join(ch for ch in stripped if unicodedata.category(ch) != "Mn")57 58 59def format_cloze_feedback(blanks: list[dict], typed: list[str | None]) -> str:60 """Score cloze answers; tolerant to case, accents and surrounding spaces."""61 filled = [(answer or "").strip() for answer in typed]62 if any(not answer for answer in filled):63 return "Fill in every blank first."64 lines, score = [], 065 for index, (blank, answer) in enumerate(zip(blanks, filled, strict=True)):66 good = _normalize(answer) == _normalize(blank["answer"])67 score += int(good)68 mark = "✅" if good else "❌"69 hint = f" — {blank['hint']}" if blank.get("hint") else ""70 if good:71 lines.append(f"{mark} **{index + 1}.** **{blank['answer']}**{hint}")72 else:73 lines.append(74 f"{mark} **{index + 1}.** you wrote *{answer}* — "75 f"correct: **{blank['answer']}**{hint}"76 )77 header = f"## Score: {score}/{len(blanks)}"78 if score == len(blanks):79 header += " 🎉"80 return header + "\n\n" + "\n\n".join(lines)81 82 83def render_cloze_text(text: str, blanks: list[dict]) -> str:84 """Replace each blank with a numbered placeholder, right to left to keep offsets valid."""85 rendered = text86 for index in range(len(blanks) - 1, -1, -1):87 blank = blanks[index]88 start, end = blank["start"], blank["start"] + len(blank["answer"])89 rendered = f"{rendered[:start]}**\\[{index + 1}: ____\\]**{rendered[end:]}"90 return rendered91 92 93def render_dictation_feedback(result: DictationResult) -> str:94 """Reference sentence with errors marked, plus WER and an error breakdown."""95 pieces: list[str] = []96 for op in result.ops:97 if op.op == "equal":98 pieces.append(op.ref_word or "")99 elif op.op == "substitute":100 pieces.append(f"~~{op.hyp_word}~~ **{op.ref_word}**")101 elif op.op == "delete":102 pieces.append(f"**[missing: {op.ref_word}]**")103 elif op.op == "insert":104 pieces.append(f"~~{op.hyp_word}~~")105 marked = " ".join(piece for piece in pieces if piece)106 107 accuracy = round((1.0 - result.wer) * 100)108 header = f"## {accuracy}% correct" + (" 🎉" if result.is_perfect else "")109 breakdown = (110 f"WER {result.wer:.0%} · {result.hits} correct, "111 f"{result.substitutions} wrong, {result.deletions} missed, "112 f"{result.insertions} extra"113 )114 if result.is_perfect:115 return f"{header}\n\n{breakdown}"116 legend = "_Marked below: **correct word**, ~~your version~~, **[missing: word]**._"117 return f"{header}\n\n{breakdown}\n\n{legend}\n\n> {marked}"118 119 120def build_app(settings: Settings | None = None) -> gr.Blocks:121 settings = settings or get_settings()122 llm = create_llm_client(settings)123 cache = FileCache(settings.cache_dir)124 125 classifier_cache: dict = {}126 127 def get_classifier():128 """Lazy, memoized; may raise (callers decide how to surface it)."""129 if "classifier" not in classifier_cache:130 classifier_cache["classifier"] = create_cefr_classifier(settings)131 return classifier_cache["classifier"]132 133 asr_cache: dict = {}134 135 def get_asr():136 """Lazy, memoized ASR client (model load is deferred to first use)."""137 if "asr" not in asr_cache:138 asr_cache["asr"] = create_asr_client(settings)139 return asr_cache["asr"]140 141 # ------------------------------------------------------------- Diagnostics142 def estimate_level(text: str) -> str:143 if not text.strip():144 return "Paste a text first."145 try:146 classifier = get_classifier()147 if classifier is None:148 return "CEFR model not configured (set CEFR_MODEL_ID or CEFR_MODEL_PATH)."149 prediction = classifier.classify_text(text)150 except Exception as exc: # surfaced to the UI, never crashes the app151 return f"❌ {type(exc).__name__}: {exc}"152 top = sorted(prediction.per_level.items(), key=lambda kv: -kv[1])[:2]153 detail = ", ".join(f"{lvl} {p:.0%}" for lvl, p in top)154 caveat = ""155 if len(text.split()) < 30:156 caveat = (157 "\n\n⚠️ Very short input — the model is trained on sentences and "158 "passages; results on single words or fragments are anecdotal."159 )160 return (161 f"**{prediction.level}** — score {prediction.score:.2f}/5 "162 f"({detail}; {prediction.n_chunks} chunk(s))" + caveat163 )164 165 async def ping_llm() -> str:166 try:167 response = await llm.complete(168 [ChatMessage(role="user", content="Reply with exactly one word: pong")],169 temperature=0.0,170 max_tokens=16,171 )172 except LLMError as exc:173 return f"❌ {exc}"174 return f"✅ [{response.model}] {response.text.strip()}"175 176 async def transcribe_audio(audio_path: str | None) -> str:177 if not audio_path:178 return "Record or upload an audio clip first."179 from pathlib import Path180 181 try:182 transcription = await get_asr().transcribe(Path(audio_path), language="en")183 except ASRError as exc:184 return f"❌ {exc}"185 except Exception as exc:186 return f"❌ {type(exc).__name__}: {exc}"187 lang = f" ({transcription.language})" if transcription.language else ""188 return f"**Transcription{lang}:** {transcription.text}" if transcription.text else "(empty)"189 190 # ----------------------------------------------------------------- Reading191 def _classifier_or_none():192 try:193 return get_classifier()194 except Exception:195 return None # Reading degrades to unverified texts instead of failing196 197 def _exercise_info(exercise: ReadingExercise) -> str:198 if exercise.source == "generated":199 verified = exercise.classified_level or "unverified — classifier not configured"200 info = f"Requested **{exercise.requested_level}** · classifier says **{verified}**"201 if exercise.classifier_score is not None:202 info += f" (score {exercise.classifier_score:.2f}/5)"203 if exercise.attempts > 1:204 info += f" · {exercise.attempts} generation attempts"205 if exercise.classified_level and exercise.classified_level != exercise.requested_level:206 info += (207 "\n\n⚠️ The rewrite still classifies off-target — served with its honest level."208 )209 return info210 level = exercise.classified_level or "unverified — classifier not configured"211 info = f"Your text · classifier says **{level}**"212 if exercise.classifier_score is not None:213 info += f" (score {exercise.classifier_score:.2f}/5)"214 return info215 216 async def fetch_exercise(level: str, topic: str, activity: str):217 try:218 exercise = await build_reading_exercise(219 llm,220 _classifier_or_none(),221 cache,222 level=level,223 topic=topic,224 activity=activity,225 model_name=settings.llm_model,226 )227 except (ReadingError, LLMError) as exc:228 return f"❌ {exc}", "", None229 except Exception as exc:230 return f"❌ {type(exc).__name__}: {exc}", "", None231 # In cloze mode the intact text would reveal every answer — the gapped232 # version is rendered inside @gr.render instead.233 shown_text = "" if exercise.activity == "cloze" else exercise.text234 return shown_text, _exercise_info(exercise), exercise.model_dump()235 236 async def use_own_text(text: str, activity: str):237 try:238 exercise = await exercise_from_user_text(239 llm,240 _classifier_or_none(),241 cache,242 text=text,243 activity=activity,244 model_name=settings.llm_model,245 )246 except (ReadingError, LLMError) as exc:247 return f"❌ {exc}", "", None248 except Exception as exc:249 return f"❌ {type(exc).__name__}: {exc}", "", None250 shown_text = "" if exercise.activity == "cloze" else exercise.text251 return shown_text, _exercise_info(exercise), exercise.model_dump()252 253 # ---------------------------------------------------------------------- UI254 with gr.Blocks(title="Polyglot Tutor") as app:255 gr.Markdown(256 "# 🌍 Polyglot Tutor\n"257 f"Adaptive language tutor — learning **{settings.default_target_lang}** "258 f"from **{settings.default_source_lang}**. *M1: Reading is live.*"259 )260 261 with gr.Tab("📖 Reading"):262 gr.Markdown(263 "Pick your level and an exercise type, then get a text written **and "264 "verified** at that level by the CEFR classifier — or paste your own "265 "English text."266 )267 with gr.Row():268 level_dd = gr.Dropdown(choices=CEFR_LEVELS, value="B1", label="Your CEFR level")269 topic_tb = gr.Textbox(label="Topic (optional)", placeholder="e.g. the ocean")270 activity_radio = gr.Radio(271 choices=[272 ("Comprehension questions", "questions"),273 ("Fill in the blanks", "cloze"),274 ],275 value="questions",276 label="Exercise type",277 )278 fetch_btn = gr.Button("📖 Get a text", variant="primary")279 with gr.Accordion("…or use your own English text", open=False):280 own_tb = gr.Textbox(label="Your text", lines=5)281 own_btn = gr.Button("Use my text")282 gr.Markdown("---")283 text_md = gr.Markdown()284 info_md = gr.Markdown()285 exercise_state = gr.State(None)286 287 @gr.render(inputs=exercise_state)288 def render_activity(exercise: dict | None):289 if not exercise:290 return291 292 if exercise.get("activity") == "cloze" and exercise.get("cloze"):293 blanks = exercise["cloze"]["blanks"]294 gr.Markdown(render_cloze_text(exercise["text"], blanks))295 inputs = []296 for index, blank in enumerate(blanks):297 inputs.append(298 gr.Textbox(299 label=f"Blank {index + 1}", placeholder="type the missing word"300 )301 )302 options = ", ".join(303 sorted([*blank.get("distractors", []), blank["answer"]])304 )305 hint_label = f"💡 Hint for blank {index + 1}"306 if blank.get("hint"):307 hint_label += f" — {blank['hint']}"308 with gr.Accordion(hint_label, open=False):309 gr.Markdown(f"Options: {options}")310 submit_btn = gr.Button("Check my answers", variant="primary")311 feedback_md = gr.Markdown()312 313 def grade_cloze(*typed: str | None) -> str:314 return format_cloze_feedback(blanks, list(typed))315 316 submit_btn.click(grade_cloze, inputs=inputs, outputs=feedback_md)317 return318 319 questions = exercise["questions"]320 radios = [321 gr.Radio(322 choices=question["options"],323 label=f"{index + 1}. {question['question']}",324 )325 for index, question in enumerate(questions)326 ]327 submit_btn = gr.Button("Check my answers", variant="primary")328 feedback_md = gr.Markdown()329 330 def grade(*picked: str | None) -> str:331 return format_feedback(questions, list(picked))332 333 submit_btn.click(grade, inputs=radios, outputs=feedback_md)334 335 outputs = [text_md, info_md, exercise_state]336 fetch_btn.click(337 fetch_exercise,338 inputs=[level_dd, topic_tb, activity_radio],339 outputs=outputs,340 api_name="reading_fetch",341 )342 own_btn.click(343 use_own_text,344 inputs=[own_tb, activity_radio],345 outputs=outputs,346 api_name="reading_own",347 )348 349 with gr.Tab("🎧 Listening"):350 gr.Markdown("TTS audio + dictation with ASR scoring — **M2**.")351 with gr.Tab("✍️ Writing"):352 gr.Markdown("LLM correction with typed errors, per-learner error profile — **M3**.")353 with gr.Tab("🗣️ Speaking"):354 gr.Markdown("Read-aloud exercises with pronunciation scoring — **M5**.")355 356 with gr.Tab("⚙️ Diagnostics"):357 gr.Markdown(358 f"- env: `{settings.app_env}`\n"359 f"- LLM: `{settings.llm_provider}` / `{settings.llm_model}`\n"360 f"- ASR: `{settings.asr_provider}` · TTS: `{settings.tts_provider}` · "361 f"storage: `{settings.storage_backend}`"362 )363 ping_button = gr.Button("Ping LLM", variant="primary")364 ping_output = gr.Textbox(label="LLM response", interactive=False)365 ping_button.click(ping_llm, outputs=ping_output)366 367 gr.Markdown("### CEFR quick check (M1 model)")368 cefr_source = settings.cefr_model_path or settings.cefr_model_id or "not configured"369 gr.Markdown(f"Model source: `{cefr_source}`")370 cefr_input = gr.Textbox(label="English text", lines=4, placeholder="Paste a text...")371 cefr_button = gr.Button("Estimate CEFR level")372 cefr_output = gr.Markdown()373 cefr_button.click(estimate_level, inputs=cefr_input, outputs=cefr_output)374 375 gr.Markdown("### ASR quick check (M2 model)")376 gr.Markdown(377 f"Provider: `{settings.asr_provider}`"378 + (f" / `{settings.asr_model}`" if settings.asr_provider != "fake" else "")379 )380 asr_input = gr.Audio(sources=["upload", "microphone"], type="filepath", label="Audio")381 asr_button = gr.Button("Transcribe")382 asr_output = gr.Markdown()383 asr_button.click(transcribe_audio, inputs=asr_input, outputs=asr_output)384 385 return app386 387 388def main() -> None:389 settings = get_settings()390 auth: tuple[str, str] | None = None391 if settings.gradio_auth_username and settings.gradio_auth_password:392 auth = (393 settings.gradio_auth_username,394 settings.gradio_auth_password.get_secret_value(),395 )396 build_app(settings).launch(397 server_name=settings.host,398 server_port=settings.port,399 auth=auth,400 )401 402 403if __name__ == "__main__":404 main()405