"""
Expert protocols for the QUERY algorithm.
An expert is any callable that answers prerequisite queries.
This module defines the protocol and provides built-in implementations
for testing and interactive use.
References:
Koppen, M., & Doignon, J.-P. (1990).
How to build a knowledge space by querying an expert.
Journal of Mathematical Psychology, 34, 311-331.
"""
from __future__ import annotations
from collections.abc import Callable, Collection
from typing import Protocol, runtime_checkable
from knowledgespaces.query.types import Query
[docs]
@runtime_checkable
class Expert(Protocol):
"""Protocol for expert query functions.
An expert receives a Query and returns True (positive) or False (negative).
The semantic of a positive answer to Query(A, q) is:
'If a student fails all items in A, they will also fail q.'
Equivalently: mastering q requires mastering at least one item in A.
"""
def __call__(self, query: Query) -> bool: ...
[docs]
class PresetExpert:
"""Expert with predetermined answers, for testing and replay.
Parameters
----------
answers : dict[tuple[frozenset[str], str], bool]
Mapping from (antecedent, consequent) to answer.
default : bool
Answer for queries not in the mapping.
"""
def __init__(
self,
answers: dict[tuple[frozenset[str], str], bool] | None = None,
default: bool = False,
) -> None:
self._answers: dict[tuple[frozenset[str], str], bool] = dict(answers or {})
self._default = default
def __call__(self, query: Query) -> bool:
key = (query.antecedent, query.consequent)
return self._answers.get(key, self._default)
[docs]
@classmethod
def from_relation(cls, relations: Collection[tuple[str, str]]) -> PresetExpert:
"""Create an expert that answers based on a known surmise relation.
The input pairs are closed transitively, so the expert answers
according to the quasi order they generate, not according to
which generating pairs the caller happened to pass. A pair query
(a -> q) is positive iff (a, q) is in the transitive closure.
A group query (A -> q) is positive iff some a in A has (a, q).
This simulates an expert who knows the true prerequisite
structure and whose answers satisfy the entailment axioms.
Suitable for testing the query algorithm against a known ground
truth.
"""
rel_set = set(relations)
# Transitive closure: an expert consistent with the query
# semantics must answer positively to every pair the generating
# pairs entail, not only to the pairs passed literally.
successors: dict[str, set[str]] = {}
for a, b in rel_set:
successors.setdefault(a, set()).add(b)
changed = True
while changed:
changed = False
for bs in successors.values():
derived: set[str] = set()
for b in bs:
derived |= successors.get(b, set())
derived -= bs
if derived:
bs |= derived
changed = True
rel_set = {(a, b) for a, bs in successors.items() for b in bs if a != b}
# We don't pre-compute all queries; instead override __call__
expert = cls.__new__(cls)
expert._answers = {}
expert._default = False
expert._relation = rel_set # type: ignore[attr-defined]
expert.__class__ = _RelationExpert # type: ignore[assignment]
return expert
class _RelationExpert:
"""Expert backed by a known surmise relation (transitive closure)."""
_relation: set[tuple[str, str]]
def __call__(self, query: Query) -> bool:
# Positive iff there exists a in A such that (a, q) in relation
# This implements monotonicity: if {a} -> q, then any A containing a -> q
return any((a, query.consequent) in self._relation for a in query.antecedent)
[docs]
class CallbackExpert:
"""Expert that delegates to a user-supplied function.
Parameters
----------
fn : Callable[[frozenset[str], str], bool]
A function that takes (antecedent_set, consequent) and returns bool.
"""
def __init__(self, fn: Callable[[frozenset[str], str], bool]) -> None:
self._fn = fn
def __call__(self, query: Query) -> bool:
return self._fn(query.antecedent, query.consequent)
[docs]
class QueryNeeded(Exception):
"""Raised by ReplayExpert when no cached answer exists.
Attributes
----------
query : Query
The unanswered query that needs to be posed to the expert.
"""
def __init__(self, query: Query) -> None:
self.query = query
super().__init__(f"No cached answer for {query}")
[docs]
class ReplayExpert:
"""Expert that replays cached answers, raises QueryNeeded on cache miss.
Useful for web applications and session resumption: replay all
prior answers instantly, then pause at the next unanswered query.
Parameters
----------
answers : dict[tuple[frozenset[str], str], bool]
Mapping from (antecedent, consequent) to answer.
"""
def __init__(self, answers: dict[tuple[frozenset[str], str], bool]) -> None:
self._answers = answers
def __call__(self, query: Query) -> bool:
key = (query.antecedent, query.consequent)
if key in self._answers:
return self._answers[key]
raise QueryNeeded(query)