wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
296 lines
12 KiB
Python
296 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from typing import TYPE_CHECKING
|
|
import uuid
|
|
|
|
from deeptutor.learning.grading import classify_error, grade_answer
|
|
from deeptutor.learning.mastery import compute_mastery
|
|
from deeptutor.learning.models import (
|
|
ErrorRecord,
|
|
LearningModule,
|
|
LearningProgress,
|
|
LearningStage,
|
|
PendingQuestion,
|
|
QuizAttempt,
|
|
RetryAttempt,
|
|
)
|
|
from deeptutor.learning.storage import LearningStore
|
|
|
|
if TYPE_CHECKING:
|
|
from deeptutor.learning.scheduler import SpacedRepetitionScheduler
|
|
|
|
|
|
class LearningService:
|
|
def __init__(self, store: LearningStore | None = None) -> None:
|
|
self._store = store or LearningStore()
|
|
|
|
def get_or_create(self, book_id: str) -> LearningProgress:
|
|
existing = self._store.load(book_id)
|
|
if existing is not None:
|
|
return existing
|
|
progress = LearningProgress(book_id=book_id)
|
|
self._store.save(progress) # persist immediately to prevent race
|
|
return progress
|
|
|
|
def init_modules(self, progress: LearningProgress, modules: list[LearningModule]) -> None:
|
|
"""Initialize the runnable module set (replace semantics)."""
|
|
self.replace_modules(progress, modules)
|
|
|
|
def replace_modules(self, progress: LearningProgress, modules: list[LearningModule]) -> None:
|
|
"""Replace all modules and clean stale KP state."""
|
|
new_kp_ids = {kp.id for m in modules for kp in m.knowledge_points}
|
|
|
|
# Clean stale KP state
|
|
for key in list(progress.mastery_levels.keys()):
|
|
if key not in new_kp_ids:
|
|
del progress.mastery_levels[key]
|
|
for key in list(progress.knowledge_types.keys()):
|
|
if key not in new_kp_ids:
|
|
del progress.knowledge_types[key]
|
|
for key in list(progress.repetition_states.keys()):
|
|
if key not in new_kp_ids:
|
|
del progress.repetition_states[key]
|
|
progress.error_records = [
|
|
r for r in progress.error_records if r.knowledge_point_id in new_kp_ids
|
|
]
|
|
progress.feynman_retries = {
|
|
k: v for k, v in progress.feynman_retries.items() if k in new_kp_ids
|
|
}
|
|
progress.feynman_explanations = {
|
|
k: v for k, v in progress.feynman_explanations.items() if k in new_kp_ids
|
|
}
|
|
progress.review_queue = [
|
|
t for t in progress.review_queue if t.knowledge_point_id in new_kp_ids
|
|
]
|
|
# Clear global stage failure records — different modules should not share failure counts
|
|
progress.stage_failure_counts = {}
|
|
progress.stage_failure_notes = {}
|
|
|
|
# Set new modules
|
|
progress.modules = list(modules)
|
|
for mod in modules:
|
|
for kp in mod.knowledge_points:
|
|
progress.knowledge_types[kp.id] = kp.type
|
|
|
|
def advance_stage(self, progress: LearningProgress, next_stage: LearningStage) -> None:
|
|
progress.current_stage = next_stage
|
|
progress.updated_at = time.time()
|
|
|
|
def switch_module(self, progress: LearningProgress, module_id: str) -> bool:
|
|
"""Point the session at ``module_id`` and reset it to that module's
|
|
first teaching stage (EXPLAIN). Mutates ``progress`` in place and returns
|
|
whether the module exists. The caller is responsible for persisting
|
|
(``save``) — typically *after* cancelling any in-flight turn so the
|
|
turn's teardown cannot overwrite the switch with stale progress.
|
|
"""
|
|
found = any(m.id == module_id for m in progress.modules)
|
|
if found:
|
|
progress.current_module_id = module_id
|
|
progress.current_kp_index = 0
|
|
progress.current_stage = LearningStage.EXPLAIN
|
|
progress.updated_at = time.time()
|
|
return found
|
|
|
|
def record_quiz_attempt(self, progress: LearningProgress, attempt: QuizAttempt) -> None:
|
|
if not attempt.is_correct and attempt.error_type is not None:
|
|
# Find existing error record for this question + knowledge point.
|
|
existing = None
|
|
for rec in progress.error_records:
|
|
if (
|
|
rec.question_id == attempt.question_id
|
|
and rec.knowledge_point_id == attempt.knowledge_point_id
|
|
):
|
|
existing = rec
|
|
break
|
|
|
|
if existing is not None:
|
|
existing.retry_history.append(
|
|
RetryAttempt(
|
|
timestamp=time.time(),
|
|
is_correct=False,
|
|
attempt_number=len(existing.retry_history) + 1,
|
|
)
|
|
)
|
|
existing.status = "retrying"
|
|
else:
|
|
record = ErrorRecord(
|
|
id=uuid.uuid4().hex,
|
|
question_id=attempt.question_id,
|
|
knowledge_point_id=attempt.knowledge_point_id,
|
|
module_id=attempt.module_id,
|
|
error_type=attempt.error_type,
|
|
self_attribution=attempt.self_attribution,
|
|
status="active",
|
|
)
|
|
progress.error_records.append(record)
|
|
|
|
elif attempt.is_correct:
|
|
# Graduate any active error record for this question + knowledge point.
|
|
for rec in progress.error_records:
|
|
if (
|
|
rec.question_id == attempt.question_id
|
|
and rec.knowledge_point_id == attempt.knowledge_point_id
|
|
and rec.status in ("active", "retrying")
|
|
):
|
|
rec.retry_history.append(
|
|
RetryAttempt(
|
|
timestamp=time.time(),
|
|
is_correct=True,
|
|
attempt_number=len(rec.retry_history) + 1,
|
|
)
|
|
)
|
|
rec.status = "graduated"
|
|
break
|
|
|
|
progress.quiz_attempts.append(attempt)
|
|
progress.updated_at = time.time()
|
|
|
|
def calculate_mastery(self, progress: LearningProgress, kp_id: str) -> float:
|
|
"""Mastery 0..1 for *kp_id* from its attempt history (policy in mastery.py)."""
|
|
correctness = [
|
|
a.is_correct for a in progress.quiz_attempts if a.knowledge_point_id == kp_id
|
|
]
|
|
return compute_mastery(correctness)
|
|
|
|
def update_mastery(self, progress: LearningProgress, kp_id: str, level: float) -> None:
|
|
progress.mastery_levels[kp_id] = level
|
|
progress.updated_at = time.time()
|
|
|
|
def grade_and_record(
|
|
self,
|
|
progress: LearningProgress,
|
|
*,
|
|
question_id: str,
|
|
knowledge_point_id: str,
|
|
module_id: str,
|
|
user_answer: str,
|
|
expected_answer: str,
|
|
question_type: str = "short",
|
|
self_attribution: str = "",
|
|
scheduler: SpacedRepetitionScheduler | None = None,
|
|
) -> bool:
|
|
"""Grade one answer and fold it through the full post-answer pipeline.
|
|
|
|
record attempt -> recompute mastery -> advance the spaced-repetition
|
|
state -> rebuild the review queue -> persist. This is the single source
|
|
of truth for what happens when a student answers, shared by every
|
|
interactive stage. Grading is fail-closed: with no stored expected
|
|
answer the attempt is recorded wrong, never right.
|
|
"""
|
|
is_correct = bool(expected_answer) and grade_answer(
|
|
user_answer, expected_answer, question_type
|
|
)
|
|
self.record_quiz_attempt(
|
|
progress,
|
|
QuizAttempt(
|
|
question_id=question_id,
|
|
knowledge_point_id=knowledge_point_id,
|
|
module_id=module_id,
|
|
is_correct=is_correct,
|
|
user_answer=user_answer,
|
|
self_attribution=self_attribution,
|
|
error_type=None if is_correct else classify_error(user_answer),
|
|
),
|
|
)
|
|
if knowledge_point_id:
|
|
self.update_mastery(
|
|
progress, knowledge_point_id, self.calculate_mastery(progress, knowledge_point_id)
|
|
)
|
|
kp_type = progress.knowledge_types.get(knowledge_point_id)
|
|
if kp_type is not None and scheduler is not None:
|
|
state = progress.repetition_states.get(
|
|
knowledge_point_id
|
|
) or scheduler.get_initial_state(kp_type)
|
|
progress.repetition_states[knowledge_point_id] = state
|
|
scheduler.schedule_next(state, kp_type, is_correct)
|
|
progress.review_queue = scheduler.build_review_queue(progress)
|
|
self.save(progress)
|
|
return is_correct
|
|
|
|
# ── Loop-driven tutoring helpers ─────────────────────────────────────
|
|
|
|
def set_pending_question(self, progress: LearningProgress, pending: PendingQuestion) -> None:
|
|
"""Store the question the tutor just posed so its expected answer can
|
|
be graded deterministically on a later turn (never via the model)."""
|
|
progress.pending_question = pending
|
|
progress.updated_at = time.time()
|
|
self.save(progress)
|
|
|
|
def clear_pending_question(self, progress: LearningProgress) -> None:
|
|
progress.pending_question = None
|
|
progress.updated_at = time.time()
|
|
self.save(progress)
|
|
|
|
def record_qualitative(
|
|
self,
|
|
progress: LearningProgress,
|
|
kp_id: str,
|
|
*,
|
|
passed: bool,
|
|
evidence: str = "",
|
|
) -> None:
|
|
"""Record the qualitative (CONCEPT / DESIGN) gate outcome.
|
|
|
|
The boolean is the gate of record; ``mastery_levels`` is nudged only so
|
|
the map's colour matches the gate (full on pass, capped on fail).
|
|
"""
|
|
progress.qualitative_mastery[kp_id] = bool(passed)
|
|
current = progress.mastery_levels.get(kp_id, 0.0)
|
|
progress.mastery_levels[kp_id] = max(current, 1.0) if passed else min(current, 0.4)
|
|
if evidence:
|
|
progress.feynman_explanations[kp_id] = evidence
|
|
progress.updated_at = time.time()
|
|
self.save(progress)
|
|
|
|
def list_progress(self) -> dict:
|
|
"""Return summary of all book progress with per-book error info."""
|
|
logger = logging.getLogger(__name__)
|
|
|
|
book_ids = self._store.list_all()
|
|
summaries = []
|
|
errors = []
|
|
for bid in book_ids:
|
|
try:
|
|
progress = self._store.load(bid)
|
|
if progress is None:
|
|
continue
|
|
# Only count KPs from current modules (exclude stale IDs)
|
|
current_kp_ids = {kp.id for m in progress.modules for kp in m.knowledge_points}
|
|
total_kps = len(current_kp_ids)
|
|
total_mastery = sum(
|
|
progress.mastery_levels.get(kp_id, 0) for kp_id in current_kp_ids
|
|
)
|
|
# Derive display name from first module, fall back to book_id
|
|
display_name = ""
|
|
if progress.modules:
|
|
display_name = progress.modules[0].name or ""
|
|
summaries.append(
|
|
{
|
|
"book_id": progress.book_id,
|
|
"name": display_name or progress.book_id,
|
|
"modules_count": len(progress.modules),
|
|
"kp_count": total_kps,
|
|
"current_stage": progress.current_stage.value
|
|
if progress.current_stage
|
|
else "",
|
|
# Average mastery across current KPs (not the % of KPs mastered).
|
|
"avg_mastery_pct": round(total_mastery / total_kps * 100)
|
|
if total_kps
|
|
else 0,
|
|
"updated_at": progress.updated_at,
|
|
}
|
|
)
|
|
except Exception:
|
|
logger.warning("Failed to load progress for book %s, skipping", bid, exc_info=True)
|
|
errors.append({"book_id": bid, "error": "Failed to load"})
|
|
continue
|
|
return {"summaries": summaries, "errors": errors}
|
|
|
|
def save(self, progress: LearningProgress) -> None:
|
|
self._store.save(progress)
|
|
|
|
|
|
__all__ = ["LearningService"]
|