feat: add adaptive assessment session

Author: dixonfrederick

src/nokaman/api/app.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from nokaman import __version__
+from nokaman.eval.adaptive import adaptive_session
 from nokaman.eval.metrics import placement_test
 from nokaman.eval.pipeline import evaluate_demo, evaluate_text
 from nokaman.rubrics.registry import SUPPORTED_LANGUAGES
@@ -25,6 +26,12 @@ class PlacementReq(BaseModel):
     answers: list[str] = Field(..., min_length=1)
 
 
+class AdaptiveReq(BaseModel):
+    language: str = "en"
+    answers: list[str] = Field(default_factory=list)
+    administered_ids: list[str] = Field(default_factory=list)
+
+
 @app.get("/health")
 def health() -> dict:
     return {
@@ -54,3 +61,10 @@ def assess_placement(req: PlacementReq) -> dict:
     if req.language not in SUPPORTED_LANGUAGES:
         raise HTTPException(400, f"unsupported language {req.language}")
     return placement_test(req.language, req.answers)
+
+
+@app.post("/assess/adaptive")
+def assess_adaptive(req: AdaptiveReq) -> dict:
+    if req.language not in SUPPORTED_LANGUAGES:
+        raise HTTPException(400, f"unsupported language {req.language}")
+    return adaptive_session(req.language, req.answers, administered_ids=req.administered_ids)

src/nokaman/eval/adaptive.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from nokaman.models.cefr import score_to_cefr
+from nokaman.models.toy import ToyAbilityModel
+from nokaman.rubrics.registry import get_language_meta
+
+CEFR_TARGETS = {
+    "A1": 22.0,
+    "A2": 42.0,
+    "B1": 57.0,
+    "B2": 72.0,
+    "C1": 84.0,
+    "C2": 94.0,
+}
+
+
+def adaptive_session(
+    language: str,
+    answers: list[str] | None = None,
+    administered_ids: list[str] | None = None,
+) -> dict:
+    code = language.strip().lower()
+    meta = get_language_meta(code)
+    answer_list = [answer for answer in (answers or []) if answer.strip()]
+    answer_estimates = estimate_answer_scores(code, answer_list)
+    ability_score = _running_ability(answer_estimates)
+    administered = set(administered_ids or [])
+    next_prompt = select_next_prompt(
+        ability_score=ability_score,
+        prompt_bank=build_prompt_bank(code),
+        administered_ids=administered,
+    )
+    return {
+        "language": code,
+        "language_name": meta["name"],
+        "n_answers": len(answer_list),
+        "ability_score": round(ability_score, 2),
+        "cefr": score_to_cefr(ability_score),
+        "answer_estimates": answer_estimates,
+        "next_prompt": next_prompt,
+        "complete": next_prompt is None,
+        "model": "AdaptiveHeuristicSession",
+    }
+
+
+def estimate_answer_scores(language: str, answers: list[str]) -> list[dict]:
+    model = ToyAbilityModel(language=language)
+    estimates = []
+    for index, answer in enumerate(answers, start=1):
+        scored = model.score_text(answer, skill="writing")
+        estimates.append(
+            {
+                "answer_index": index,
+                "score": scored["score"],
+                "cefr": scored["cefr"],
+                "tokens": scored["features"]["tokens"],
+            }
+        )
+    return estimates
+
+
+def select_next_prompt(
+    ability_score: float,
+    prompt_bank: Iterable[dict],
+    administered_ids: set[str] | list[str] | None = None,
+) -> dict | None:
+    administered = set(administered_ids or [])
+    candidates = [prompt for prompt in prompt_bank if str(prompt["id"]) not in administered]
+    if not candidates:
+        return None
+    target = max(0.0, min(100.0, float(ability_score)))
+    return min(
+        candidates,
+        key=lambda prompt: (
+            abs(float(prompt["target_score"]) - target),
+            float(prompt["target_score"]),
+            str(prompt["id"]),
+        ),
+    )
+
+
+def build_prompt_bank(language: str) -> list[dict]:
+    code = language.strip().lower()
+    meta = get_language_meta(code)
+    language_name = meta["name"]
+    templates = {
+        "A1": f"Introduce yourself in simple {language_name} sentences.",
+        "A2": f"Describe your daily routine in {language_name} with times and places.",
+        "B1": f"Explain a recent problem you solved while learning {language_name}.",
+        "B2": f"Compare two study strategies and defend your preference in {language_name}.",
+        "C1": f"Analyze how culture affects communication style in {language_name}.",
+        "C2": f"Write a nuanced argument about language policy and education in {language_name}.",
+    }
+    return [
+        {
+            "id": f"{code}_{band.lower()}_adaptive",
+            "language": code,
+            "difficulty_cefr": band,
+            "target_score": target,
+            "prompt": templates[band],
+        }
+        for band, target in CEFR_TARGETS.items()
+    ]
+
+
+def _running_ability(answer_estimates: list[dict]) -> float:
+    if not answer_estimates:
+        return CEFR_TARGETS["A2"]
+    return sum(float(item["score"]) for item in answer_estimates) / len(answer_estimates)

tests/test_adaptive.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from nokaman.eval.adaptive import adaptive_session, build_prompt_bank, select_next_prompt
+
+
+def test_select_next_prompt_uses_nearest_unadministered_prompt() -> None:
+    prompt_bank = [
+        {"id": "a2", "target_score": 42.0},
+        {"id": "b1", "target_score": 57.0},
+        {"id": "b2", "target_score": 72.0},
+    ]
+    selected = select_next_prompt(ability_score=60.0, prompt_bank=prompt_bank, administered_ids={"b1"})
+    assert selected is not None
+    assert selected["id"] == "b2"
+
+
+def test_adaptive_session_starts_at_a2_prompt() -> None:
+    result = adaptive_session("en", answers=[])
+    assert result["ability_score"] == 42.0
+    assert result["cefr"] == "A2"
+    assert result["next_prompt"]["difficulty_cefr"] == "A2"
+    assert not result["complete"]
+
+
+def test_adaptive_session_tracks_answer_estimates_and_skips_ids() -> None:
+    first_prompt = build_prompt_bank("en")[1]
+    result = adaptive_session(
+        "en",
+        answers=["I study English every day because it helps me talk with more people."],
+        administered_ids=[first_prompt["id"]],
+    )
+    assert result["n_answers"] == 1
+    assert result["answer_estimates"][0]["tokens"] > 0
+    assert result["next_prompt"]["id"] != first_prompt["id"]

tests/test_api.py

@@ -24,3 +24,17 @@ def test_assess_text() -> None:
     )
     assert r.status_code == 200
     assert "cefr" in r.json()
+
+
+def test_assess_adaptive() -> None:
+    r = client.post(
+        "/assess/adaptive",
+        json={
+            "language": "en",
+            "answers": ["I study English every day because it helps me travel."],
+        },
+    )
+    assert r.status_code == 200
+    payload = r.json()
+    assert payload["model"] == "AdaptiveHeuristicSession"
+    assert payload["next_prompt"]["language"] == "en"