| import os, json, re |
| import gradio as gr |
| import numpy as np |
| import matplotlib |
| matplotlib.use('Agg') |
| import matplotlib.pyplot as plt |
|
|
| from evaluator import KANEvaluator |
| from teacher import GeminiTeacher, EXERCISES, CATEGORY_LABELS |
| from profiler import LearnerProfiler |
| from reporter import ReporterAgent |
|
|
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) |
| MODEL_PATH = os.path.join(BASE_DIR, 'tinykan_cosine.pth') |
| GEMINI_KEY = os.environ.get('GEMINI_API_KEY', '') |
| PROFILE_PATH = os.path.join(BASE_DIR, 'profiles.json') |
|
|
| print("Initializing agents...") |
| evaluator = KANEvaluator(MODEL_PATH) |
| teacher = GeminiTeacher(GEMINI_KEY) if GEMINI_KEY else None |
| profiler = LearnerProfiler(PROFILE_PATH) |
| reporter = ReporterAgent() |
| print("All agents initialized") |
|
|
| |
| |
| |
|
|
| |
| |
| |
| INK = "#1A2332" |
| SUBINK = "#64748B" |
| RULE = "#E2E5EA" |
| PAPER = "#F7F8FA" |
| CARD = "#FFFFFF" |
| ACCENT = "#4F46E5" |
| DATA = "#0F766E" |
| LIGHT = "#C7D2FE" |
|
|
| theme = gr.themes.Base( |
| primary_hue=gr.themes.colors.indigo, |
| neutral_hue=gr.themes.colors.slate, |
| font=[gr.themes.GoogleFont("Inter"), "system-ui", "sans-serif"], |
| font_mono=[gr.themes.GoogleFont("JetBrains Mono"), "monospace"], |
| ) |
|
|
| CSS = f""" |
| @import url('https://fonts.googleapis.com/css2?family=Newsreader:ital,wght@0,400;0,500;0,600;1,400&family=Inter:wght@400;500;600&family=JetBrains+Mono:wght@400;500&display=swap'); |
| |
| .gradio-container {{ background: {PAPER} !important; max-width: 1180px !important; font-family: 'Inter', sans-serif; }} |
| |
| #masthead {{ |
| border-bottom: 1px solid {RULE}; padding-bottom: 18px; margin-bottom: 20px; |
| }} |
| #masthead .kicker {{ font-size: 11px; letter-spacing: 1.5px; text-transform: uppercase; color: {SUBINK}; font-weight: 500; }} |
| #masthead h1 {{ font-family: 'Newsreader', serif; font-size: 26px; font-weight: 600; color: {INK}; margin: 4px 0 0 0; }} |
| |
| /* --- Signature element: agent pipeline status strip --- */ |
| #agent-strip {{ |
| display: flex; align-items: center; gap: 0; margin: 14px 0 0 0; |
| background: {CARD}; border: 1px solid {RULE}; border-radius: 10px; padding: 10px 16px; |
| }} |
| .agent-node {{ |
| display: flex; align-items: center; gap: 8px; font-size: 12px; font-weight: 500; color: {SUBINK}; |
| padding: 4px 10px; border-radius: 20px; transition: all 0.2s ease; |
| }} |
| .agent-node .dot {{ |
| width: 7px; height: 7px; border-radius: 50%; background: {RULE}; flex-shrink: 0; |
| }} |
| .agent-node.active {{ background: {LIGHT}; color: {ACCENT}; }} |
| .agent-node.active .dot {{ background: {ACCENT}; }} |
| .agent-arrow {{ color: {RULE}; font-size: 13px; margin: 0 2px; }} |
| |
| .section-label {{ |
| font-size: 11px; letter-spacing: 1.2px; text-transform: uppercase; font-weight: 600; |
| color: {ACCENT}; border-bottom: 1px solid {RULE}; padding-bottom: 6px; margin-bottom: 12px; |
| }} |
| .sheet {{ |
| background: {CARD}; border: 1px solid {RULE}; border-radius: 12px; padding: 20px 22px; |
| box-shadow: 0 1px 2px rgba(26,35,50,0.04), 0 4px 12px rgba(26,35,50,0.03); |
| }} |
| .note {{ font-size: 12px; color: {SUBINK}; font-style: italic; }} |
| |
| /* Scores/data in monospace for a "measured" feel */ |
| .score-value, .data-figure {{ font-family: 'JetBrains Mono', monospace; font-weight: 500; }} |
| |
| [role="option"], [role="option"] *, |
| ul[role="listbox"] li, ul[role="listbox"] li *, |
| .wrap-inner, .wrap-inner *, .icon-wrap, .icon-wrap *, |
| .dropdown-arrow, .dropdown-arrow *, svg.dropdown-arrow, |
| [data-testid="dropdown"], [data-testid="dropdown"] svg, [data-testid="dropdown"] .icon-wrap, |
| .options, .options *, .options li, .options .item, |
| [data-testid="dropdown"] ul li, [data-testid="dropdown"] ul li *, |
| ul.options li, ul.options li *, |
| div.options li, div.options div, |
| .gr-dropdown ul li, .gr-dropdown ul li *, |
| select, select option {{ cursor: pointer !important; }} |
| """ |
|
|
| TYPE_CHOICES = [(label, key) for key, label in CATEGORY_LABELS.items()] + \ |
| [('Automatic (Planner-selected)', 'auto')] |
|
|
|
|
| def agent_strip_html(active=None): |
| """Renders the 4-agent pipeline status strip, highlighting whichever |
| agent(s) just acted, so the multi-agent flow is visible at a glance.""" |
| active = active or [] |
| nodes = [ |
| ('evaluator', 'Evaluator'), |
| ('teacher', 'Teacher'), |
| ('planner', 'Planner'), |
| ('reporter', 'Reporter'), |
| ] |
| parts = [] |
| for i, (key, label) in enumerate(nodes): |
| cls = "agent-node active" if key in active else "agent-node" |
| parts.append(f'<div class="{cls}"><span class="dot"></span>{label}</div>') |
| if i < len(nodes) - 1: |
| parts.append('<span class="agent-arrow">→</span>') |
| return f'<div id="agent-strip">{"".join(parts)}</div>' |
|
|
|
|
| def plot_xai(result): |
| plt.rcParams['font.family'] = 'serif' |
| fig, axes = plt.subplots(1, 3, figsize=(15, 4.6)) |
| fig.patch.set_facecolor('#ffffff') |
|
|
| dark, light, mid = '#0F766E', '#C7D2FE', '#4F46E5' |
| feat_names = ['cos_MiniLM','cos_MPNet','cos_RoBERTa', |
| '|M-MP|','|M-R|','|MP-R|','mean','max','min'] |
|
|
| ax1 = axes[0] |
| feat_colors = [dark if i < 3 else mid if i < 6 else light for i in range(9)] |
| bars = ax1.barh(feat_names, result.importance*100, color=feat_colors, |
| edgecolor='white', linewidth=0.6, height=0.6) |
| ax1.set_xlabel('Importance (%)', fontsize=9.5, color=dark) |
| ax1.set_title('Feature attribution', fontsize=11, fontweight='bold', color=dark, pad=8) |
| ax1.tick_params(labelsize=8, colors=dark) |
| for s in ['top','right']: ax1.spines[s].set_visible(False) |
| for s in ['left','bottom']: ax1.spines[s].set_color('#E2E5EA') |
| for bar, val in zip(bars, result.importance): |
| ax1.text(bar.get_width()+0.3, bar.get_y()+bar.get_height()/2, |
| f'{val*100:.1f}%', va='center', fontsize=7.2, color=dark) |
| ax1.set_facecolor('#ffffff') |
|
|
| ax2 = axes[1] |
| wedges, _, autotexts = ax2.pie( |
| [result.gate_kan*100, result.gate_linear*100], |
| labels=['KAN branch', 'Linear branch'], |
| colors=[dark, light], |
| autopct='%1.0f%%', startangle=90, |
| textprops={'fontsize':9,'color':dark}, |
| wedgeprops={'edgecolor':'white','linewidth':2}) |
| for at in autotexts: at.set_fontsize(9.5); at.set_color('white'); at.set_fontweight('bold') |
| ax2.set_title('Gate distribution', fontsize=11, fontweight='bold', color=dark, pad=8) |
|
|
| ax3 = axes[2] |
| levels = ['A1','A2','B1','B2','C1','C2'] |
| thresholds = [0, 0.40, 0.55, 0.70, 0.80, 0.90, 1.0] |
| shades = ['#EEF2FF','#E0E7FF','#C7D2FE','#A5B4FC','#4F46E5','#0F766E'] |
| for i, level in enumerate(levels): |
| ax3.barh(0, thresholds[i+1]-thresholds[i], left=thresholds[i], |
| height=0.32, color=shades[i], edgecolor='white') |
| tc = dark if i < 3 else 'white' |
| ax3.text((thresholds[i]+thresholds[i+1])/2, 0, level, ha='center', va='center', |
| fontsize=9, fontweight='bold', color=tc) |
| ax3.axvline(x=result.score, color='#B91C1C', linewidth=2.2, zorder=5) |
| ax3.plot(result.score, 0, 'v', color='#B91C1C', markersize=10, zorder=6) |
| ax3.text(result.score, 0.24, f'{result.score_pct:.1f}%', |
| ha='center', fontsize=9.5, fontweight='bold', color='#B91C1C', |
| bbox=dict(boxstyle='round,pad=0.25', facecolor='white', edgecolor='#B91C1C', linewidth=1.2)) |
| ax3.set_xlim(0,1); ax3.set_ylim(-0.5,0.55) |
| ax3.set_title('Response-quality position', fontsize=11, fontweight='bold', color=dark, pad=8) |
| ax3.set_yticks([]); ax3.set_facecolor('#ffffff') |
| for s in ['left','top','right']: ax3.spines[s].set_visible(False) |
| ax3.spines['bottom'].set_color('#E2E5EA') |
| ax3.tick_params(colors=dark, labelsize=8) |
|
|
| plt.tight_layout() |
| path = '/tmp/xai_plot.png' |
| plt.savefig(path, dpi=170, bbox_inches='tight', facecolor='white') |
| plt.close() |
| return path |
|
|
| def plot_history(score_history): |
| if len(score_history) < 2: |
| return None |
| fig, ax = plt.subplots(figsize=(9, 3.2)) |
| fig.patch.set_facecolor('#ffffff') |
| x = list(range(1, len(score_history)+1)) |
| y = [s*100 for s in score_history] |
| ax.plot(x, y, color='#4F46E5', linewidth=2, marker='o', markersize=4, |
| markerfacecolor='#C7D2FE', markeredgecolor='#4F46E5') |
| ax.fill_between(x, y, color='#C7D2FE', alpha=0.2) |
| ax.set_ylim(0, 100) |
| ax.set_xlabel('Session', fontsize=9.5, color='#1A2332') |
| ax.set_ylabel('Score (%)', fontsize=9.5, color='#1A2332') |
| ax.set_title('Score progression', fontsize=11, fontweight='bold', color='#1A2332', pad=8) |
| ax.tick_params(labelsize=8, colors='#1A2332') |
| for s in ['top','right']: ax.spines[s].set_visible(False) |
| for s in ['left','bottom']: ax.spines[s].set_color('#E2E5EA') |
| ax.set_facecolor('#ffffff') |
| plt.tight_layout() |
| path = '/tmp/history_plot.png' |
| plt.savefig(path, dpi=160, bbox_inches='tight', facecolor='white') |
| plt.close() |
| return path |
|
|
| def planner_panel(name): |
| p = profiler.get(name) |
| status = profiler.planner_status(p) |
| return f"**Planner — {p.current_cefr}** \n{status['message']}" |
|
|
| def start_session(name, cefr, ex_type, session_state): |
| session_state = dict(session_state) if session_state else {} |
| last_given = session_state.get('last_given', {}) |
|
|
| if not name.strip(): |
| return "Enter your name to begin. / Entrez votre nom pour commencer.", "", None, "", agent_strip_html([]), session_state |
|
|
| name = name.strip() |
| profile = profiler.get(name, cefr) |
| effective_cefr = cefr if profile.total_exercises == 0 else ( |
| profile.current_cefr if profile.total_exercises >= profiler.MIN_SESSIONS else cefr) |
|
|
| if ex_type == 'auto': |
| available = [e['type'] for e in EXERCISES.get(effective_cefr, EXERCISES['B1'])] |
| chosen_type = profiler.weak_exercise_type(name, list(set(available))) |
| else: |
| chosen_type = ex_type |
|
|
| key = (name, effective_cefr, chosen_type) |
| exercise = teacher.get_exercise(effective_cefr, chosen_type, exclude_instruction=last_given.get(key)) if teacher else None |
| if not exercise: |
| from teacher import Exercise |
| exercise = Exercise('translation', |
| 'Traduisez: "Il est important de protéger l\'environnement."', |
| 'Translate into English: "Il est important de protéger l\'environnement."', |
| 'It is important to protect the environment.', |
| effective_cefr, 'General', ['important','protect','environment']) |
| last_given[key] = exercise.instruction |
| session_state['last_given'] = last_given |
|
|
| session_state['current_exercise'] = {'exercise': exercise, 'name': name, 'cefr': effective_cefr} |
|
|
| adapted_note = " (level adapted by Planner / niveau adapté)" if effective_cefr != cefr else "" |
| profile_md = (f"**{name}**{adapted_note} · Exercise level / Niveau **{effective_cefr}** · " |
| f"{profile.total_exercises} sessions recorded / enregistrées · " |
| f"{profile.avg_score*100:.0f}% average / moyenne") |
|
|
| label = CATEGORY_LABELS.get(exercise.type, exercise.type.title()) |
| lines = [f"**{label} — {effective_cefr}**", "", |
| f"FR — {exercise.instruction}", "", |
| f"EN — {exercise.instruction_en}"] |
| if exercise.hints: |
| lines.append("") |
| lines.append(f"*Hints / Indices: {', '.join(exercise.hints)}*") |
| exercise_md = "\n".join(lines) |
|
|
| return profile_md, exercise_md, None, planner_panel(name), agent_strip_html(['planner']), session_state |
|
|
| def evaluate_answer(learner_answer, session_state): |
| session_state = dict(session_state) if session_state else {} |
| current_exercise = session_state.get('current_exercise') |
|
|
| if not current_exercise: |
| return "Start a session first.", "", None, "", "", agent_strip_html([]), session_state |
| if not learner_answer.strip(): |
| return "Write an answer first.", "", None, "", "", agent_strip_html([]), session_state |
|
|
| exercise = current_exercise['exercise'] |
| name = current_exercise['name'] |
| cefr = current_exercise['cefr'] |
|
|
| source_match = re.search(r'"([^"]+)"', exercise.instruction) |
| source_sentence = source_match.group(1) if source_match else None |
| result = evaluator.evaluate(exercise.reference, learner_answer, category=exercise.type, source_sentence=source_sentence) |
|
|
| if teacher: |
| feedback = teacher.get_feedback(exercise, learner_answer, result.score, result.xai_explanation) |
| feedback_md = ( |
| f"**Correction** \n{feedback.correction}\n\n" |
| f"**Explanation** \n{feedback.explanation}\n\n" |
| f"**Suggestion** \n{feedback.improvement}\n\n" |
| f"**Encouragement** \n{feedback.encouragement}\n\n" |
| f"**Next focus** \n{feedback.next_focus}" |
| ) |
| else: |
| feedback_md = result.feedback |
|
|
| profile = profiler.update(name, result.score, exercise.type, cefr, learner_answer) |
|
|
| score_md = ( |
| f"### Response-quality equivalent / Qualité de réponse : {result.cefr} ({result.score_pct:.1f}%)\n" |
| f"<span class='note'>Reflects the semantic quality of this specific answer, not the exercise's " |
| f"target level ({cefr}). / Reflète la qualité sémantique de cette réponse précise, " |
| f"pas le niveau de l'exercice ({cefr}).</span>\n\n" |
| f"| Indicator / Indicateur | Value / Valeur |\n|---|---|\n" |
| f"| Confidence / Confiance | {result.confidence*100:.0f}% |\n" |
| f"| KAN branch | {result.gate_kan*100:.0f}% |\n" |
| f"| Linear branch | {result.gate_linear*100:.0f}% |\n\n" |
| f"*{result.xai_explanation}*" |
| ) |
|
|
| profile_md = (f"**{name}** · Exercise level / Niveau **{profile.current_cefr}** · " |
| f"{profile.total_exercises} sessions recorded / enregistrées · " |
| f"{profile.avg_score*100:.0f}% average / moyenne") |
|
|
| active_agents = ['evaluator'] |
| if teacher: |
| active_agents.append('teacher') |
| active_agents.append('planner') |
|
|
| return score_md, feedback_md, plot_xai(result), profile_md, planner_panel(name), agent_strip_html(active_agents), session_state |
|
|
| def show_report(name): |
| if not name.strip(): |
| return "Enter a name above first. / Entrez un nom d'abord.", "", None, agent_strip_html([]) |
| p = profiler.get(name.strip()) |
| r = profiler.report(name.strip()) |
| if 'message' in r: |
| return r['message'], "", None, agent_strip_html([]) |
| md = ( |
| f"**{r['name']}** · currently / actuellement **{r['cefr']}**\n\n" |
| f"| Metric / Indicateur | Value / Valeur |\n|---|---|\n" |
| f"| Sessions | {r['total']} |\n" |
| f"| Average / Moyenne | {r['avg']} |\n" |
| f"| Best / Meilleur | {r['best']} |\n" |
| f"| Trend / Tendance | {r['progression']} |\n\n" |
| f"Recent / Récents : {' · '.join(r['recent'])}" |
| ) |
| rep = reporter.generate_report(p) |
| report_narrative = ( |
| f"**{rep.trend_summary}**\n\n" |
| f"{rep.strength} \n{rep.weakness}\n\n" |
| f"{rep.stability_note}\n\n" |
| f"**Recommandation du Reporter :** {rep.recommendation}" |
| ) |
| return md, report_narrative, plot_history(p.score_history), agent_strip_html(['reporter']) |
|
|
| with gr.Blocks(theme=theme, css=CSS, title="Adaptive English Practice") as app: |
|
|
| gr.HTML(""" |
| <div id="masthead"> |
| <div class="kicker">Multi-Agent Language Learning System</div> |
| <h1>Adaptive English Practice</h1> |
| </div> |
| """) |
| agent_strip = gr.HTML(agent_strip_html([])) |
| session_state = gr.State({}) |
|
|
| with gr.Row(): |
| with gr.Column(scale=1, elem_classes="sheet"): |
| gr.HTML('<div class="section-label">Learner / Apprenant</div>') |
| name_input = gr.Textbox(label="Name / Nom", placeholder="e.g. Mouna Khelifi") |
| cefr_input = gr.Dropdown(['A1','A2','B1','B2','C1','C2'], value='B1', |
| label="Exercise level / Niveau de l'exercice") |
| type_input = gr.Dropdown(TYPE_CHOICES, value='translation', |
| label="Exercise category / Catégorie d'exercice") |
| btn_start = gr.Button("Begin session / Commencer", variant="primary") |
| profile_md = gr.Markdown("") |
| planner_md = gr.Markdown("", elem_classes="note") |
|
|
| with gr.Column(scale=2, elem_classes="sheet"): |
| gr.HTML('<div class="section-label">Exercise / Exercice</div>') |
| exercise_md = gr.Markdown("*Select a level and click Begin session. / " |
| "Choisissez un niveau et cliquez sur Commencer.*") |
| answer_box = gr.Textbox(label="Your answer / Votre réponse", lines=3) |
| btn_eval = gr.Button("Submit answer / Soumettre", variant="primary") |
|
|
| gr.HTML('<div style="height:22px"></div>') |
|
|
| with gr.Row(): |
| with gr.Column(scale=1, elem_classes="sheet"): |
| gr.HTML('<div class="section-label">Evaluation — TinyKAN-Distilled</div>') |
| score_md = gr.Markdown("*No submission yet. / Aucune réponse soumise.*") |
| with gr.Column(scale=1, elem_classes="sheet"): |
| gr.HTML('<div class="section-label">Feedback — Teacher Agent</div>') |
| feedback_md = gr.Markdown("*No submission yet. / Aucune réponse soumise.*") |
|
|
| with gr.Accordion("Explainability (XAI detail) / Détail explicatif", open=False): |
| xai_plot = gr.Image(show_label=False, type="filepath") |
|
|
| gr.HTML('<div style="height:10px"></div>') |
|
|
| with gr.Accordion("Learner history / Historique de l'apprenant", open=False): |
| hist_name = gr.Textbox(label="Name / Nom") |
| btn_report = gr.Button("Load history / Charger l'historique", size="sm") |
| report_md = gr.Markdown("") |
| history_plot = gr.Image(show_label=False, type="filepath") |
| gr.HTML('<div class="section-label" style="margin-top:14px;">Reporter agent analysis / Analyse du Reporter</div>') |
| reporter_md = gr.Markdown("*Load a history above to generate an analysis. / Chargez un historique ci-dessus pour générer une analyse.*") |
|
|
| btn_start.click(fn=start_session, |
| inputs=[name_input, cefr_input, type_input, session_state], |
| outputs=[profile_md, exercise_md, xai_plot, planner_md, agent_strip, session_state]) |
|
|
| btn_eval.click(fn=evaluate_answer, |
| inputs=[answer_box, session_state], |
| outputs=[score_md, feedback_md, xai_plot, profile_md, planner_md, agent_strip, session_state]) |
|
|
| btn_report.click(fn=show_report, inputs=[hist_name], outputs=[report_md, reporter_md, history_plot, agent_strip]) |
|
|
| if __name__ == '__main__': |
| app.launch( |
| server_name='0.0.0.0', |
| server_port=int(os.environ.get('PORT', 7860)), |
| pwa=True |
| ) |
