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")
# NOTE: session-specific data (current_exercise, last_given) is now stored in
# a per-browser-session gr.State component instead of module-level globals,
# to prevent concurrent users from overwriting each other's session data.
# ---------------------------------------------------------------------
# Visual identity -- refined, professional SaaS-style palette
# ---------------------------------------------------------------------
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'
{label}
')
if i < len(nodes) - 1:
parts.append('→')
return f'{"".join(parts)}
'
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"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}).\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("""
""")
agent_strip = gr.HTML(agent_strip_html([]))
session_state = gr.State({})
with gr.Row():
with gr.Column(scale=1, elem_classes="sheet"):
gr.HTML('Learner / Apprenant
')
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('Exercise / Exercice
')
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('')
with gr.Row():
with gr.Column(scale=1, elem_classes="sheet"):
gr.HTML('Evaluation — TinyKAN-Distilled
')
score_md = gr.Markdown("*No submission yet. / Aucune réponse soumise.*")
with gr.Column(scale=1, elem_classes="sheet"):
gr.HTML('Feedback — Teacher Agent
')
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('')
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('Reporter agent analysis / Analyse du Reporter
')
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
)