Source code for knowledgespaces.assessment.adaptive

"""
Adaptive assessment: item selection policies and termination criteria.

Provides three item-selection policies for adaptive assessment:

- Expected Information Gain (EIG): pick the item whose response is
  expected to reduce the entropy of the state posterior the most.
- Half-split: the classic KST questioning rule — pick the item whose
  current mastery probability is closest to 1/2, i.e., the item that
  best bisects the plausible states (Falmagne & Doignon 1988; Falmagne
  & Doignon 2011, ch. 13).
- Informative: Equation (13.14), with mastery masses weighting the
  entropies after each hypothetical update. This differs from Bayesian EIG.

References:
    Cover, T. M., & Thomas, J. A. (2006).
    Elements of Information Theory, 2nd ed. Wiley.

    Falmagne, J.-C., & Doignon, J.-P. (1988). A class of stochastic
    procedures for the assessment of knowledge. British Journal of
    Mathematical and Statistical Psychology, 41(1), 1-23.

    Falmagne, J.-C., & Doignon, J.-P. (2011).
    Learning Spaces, Chapter 13. Springer-Verlag.
"""

from __future__ import annotations

import random
from dataclasses import dataclass

import numpy as np

from knowledgespaces.assessment.blim import StatePosterior, shannon_entropy
from knowledgespaces.assessment.multiplicative import MultiplicativePosterior


[docs] @dataclass(frozen=True) class ItemScore: """Score of an item under a selection policy.""" item: str score: float
[docs] def select_item_eig( posterior: StatePosterior, candidates: list[str] | None = None, exclude: set[str] | None = None, *, rng: random.Random | None = None, ) -> ItemScore: """Select the item maximizing Expected Information Gain. EIG(q) = H(current) - E[H(posterior after observing q)] where the expectation is over both possible responses (correct/incorrect), weighted by their marginal probability. Parameters ---------- posterior : StatePosterior Current state distribution. candidates : list[str] or None Items to consider. If None, uses all items in the domain. exclude : set[str] or None Items to exclude (e.g. already asked). Applied after candidates. rng : random.Random or None Choose uniformly among exact computed score ties when supplied. Otherwise choose the first candidate (sorted domain by default). Returns ------- ItemScore The best item and its EIG score. Raises ------ ValueError If no candidates are available, or if candidates contains items not in the domain. Notes ----- The EIG for an item :math:`q` is derived from the definition of mutual information between the latent state :math:`K` and the response :math:`R_q`: .. math:: \\mathrm{EIG}(q) &= I(K; R_q) \\\\ &= H(K) - E_{R_q}\\bigl[H(K \\mid R_q)\\bigr] \\\\ &= H(K) - \\bigl[\\,P(R_q{=}1)\\,H(K \\mid R_q{=}1) + P(R_q{=}0)\\,H(K \\mid R_q{=}0)\\bigr]. The marginal :math:`P(R_q{=}1) = \\sum_K P(R_q{=}1\\mid K)\\,P(K)` comes from the law of total probability; the conditional posteriors :math:`P(K \\mid R_q)` are obtained by Bayes' rule and renormalized to sum to 1. References ---------- Cover & Thomas (2006), *Elements of Information Theory*, §2. """ blim = posterior.blim if candidates is None: candidates = sorted(blim.structure.domain) if exclude: candidates = [c for c in candidates if c not in exclude] if not candidates: raise ValueError("No candidate items available.") extra = set(candidates) - blim.structure.domain if extra: raise ValueError(f"Candidates contain items not in the domain: {extra}") current_entropy = posterior.entropy scores = [] for item in dict.fromkeys(candidates): # Marginal probability of a correct response on q: # P(R_q=1) = sum_K P(R_q=1 | K) * P(K). lh_correct = blim.likelihood_vector(item, True) prob_correct = float(np.sum(lh_correct * posterior.probabilities)) # Posterior P(K | R_q=1), renormalized by Bayes' rule. post_c = lh_correct * posterior.probabilities sum_c = post_c.sum() if sum_c > 0: post_c = post_c / sum_c entropy_correct = shannon_entropy(post_c) # Posterior P(K | R_q=0), renormalized by Bayes' rule. lh_incorrect = blim.likelihood_vector(item, False) post_i = lh_incorrect * posterior.probabilities sum_i = post_i.sum() if sum_i > 0: post_i = post_i / sum_i entropy_incorrect = shannon_entropy(post_i) # Expected entropy after observing this item expected_entropy = prob_correct * entropy_correct + (1 - prob_correct) * entropy_incorrect eig = current_entropy - expected_entropy scores.append(ItemScore(item, float(eig))) return _best_score(scores, rng)
[docs] def select_item_half_split( posterior: StatePosterior | MultiplicativePosterior, candidates: list[str] | None = None, exclude: set[str] | None = None, *, rng: random.Random | None = None, ) -> ItemScore: """Select an item by the classic half-split questioning rule. Picks the item whose current mastery probability :math:`p_q = P(q \\in K) = \\sum_{K \\ni q} P(K)` is closest to 1/2 — the item that best bisects the probability mass over plausible states. This is the canonical questioning rule of knowledge space theory (Falmagne & Doignon 1988; Falmagne & Doignon 2011, ch. 13), provided here as the standard baseline against which entropy-based selection (:func:`select_item_eig`) is compared. Parameters ---------- posterior : StatePosterior or MultiplicativePosterior Current state distribution. candidates : list[str] or None Items to consider. If None, uses all items in the domain. exclude : set[str] or None Items to exclude (e.g. already asked). Applied after candidates. rng : random.Random or None Uniform choice among exact computed ties, as in Definition 13.4.7. If omitted, ties use candidate order for backward compatibility. Returns ------- ItemScore The selected item; ``score = min(p_q, 1 - p_q)`` in ``[0, 0.5]``, higher meaning closer to a perfect bisection. Ties are broken by candidate order (sorted domain order when ``candidates`` is None), matching the deterministic tie-breaking of :func:`select_item_eig`. Raises ------ ValueError If no candidate items are available, or if candidates contains items not in the domain. Notes ----- Unlike EIG, the half-split criterion does not use the error parameters β and η at selection time: it depends only on the current state distribution. Under a noise-free local independence model the two rules coincide; with noise they can diverge, which is precisely what makes half-split the informative baseline. """ structure = posterior.structure if candidates is None: candidates = sorted(structure.domain) if exclude: candidates = [c for c in candidates if c not in exclude] if not candidates: raise ValueError("No candidate items available.") extra = set(candidates) - structure.domain if extra: raise ValueError(f"Candidates contain items not in the domain: {extra}") mastery = posterior.marginal_mastery() scores = [] for item in dict.fromkeys(candidates): p = mastery[item] score = min(p, 1.0 - p) scores.append(ItemScore(item, float(score))) return _best_score(scores, rng)
[docs] def select_item_informative( posterior: StatePosterior | MultiplicativePosterior, candidates: list[str] | None = None, exclude: set[str] | None = None, *, rng: random.Random | None = None, ) -> ItemScore: """Select by Learning Spaces (2011), Definition 13.4.8 / Eq. (13.14). Minimize ``p_q H(update(q, 1)) + (1-p_q) H(update(q, 0))``, where ``p_q`` is the current mastery mass. The returned score is current entropy minus this criterion, in bits; it can be negative. It is not Bayesian expected information gain, which weights by predictive response probabilities under the BLIM instead of mastery masses. Supports Bayesian and multiplicative updates. Zero-weight outcomes contribute zero without trying to condition on impossible evidence. Candidates and exclusions follow :func:`select_item_half_split`. With ``rng``, draw uniformly among exact computed ties, as in Eq. (13.15). Otherwise choose the first candidate (sorted domain by default). No almost-sure convergence theorem for arbitrary combinations is implied. """ candidates = sorted(posterior.structure.domain) if candidates is None else candidates candidates = list(dict.fromkeys(q for q in candidates if not exclude or q not in exclude)) if not candidates: raise ValueError("No candidate items available.") extra = set(candidates) - posterior.structure.domain if extra: raise ValueError(f"Candidates contain items not in the domain: {extra}") mastery = posterior.marginal_mastery() scores = [] for item in candidates: p = mastery[item] entropy = 0.0 for response, weight in ((True, p), (False, 1 - p)): if weight > 0: entropy += weight * posterior.update(item, response).entropy scores.append(ItemScore(item, posterior.entropy - entropy)) return _best_score(scores, rng)
def _best_score(scores: list[ItemScore], rng: random.Random | None) -> ItemScore: best = max(scores, key=lambda entry: entry.score) if rng is None: return best return rng.choice([entry for entry in scores if entry.score == best.score])
[docs] def is_converged( posterior: StatePosterior | MultiplicativePosterior, threshold: float = 0.85, ) -> bool: """Check if the assessment has converged. Convergence occurs when the most likely state has probability above the threshold. Parameters ---------- posterior : StatePosterior or MultiplicativePosterior Current state distribution. threshold : float Probability threshold for convergence. Must be in (0, 1]. Default 0.85. Raises ------ ValueError If threshold is not in (0, 1]. """ if not (0 < threshold <= 1): raise ValueError(f"threshold must be in (0, 1], got {threshold}") _, max_prob = posterior.most_likely_state return max_prob >= threshold