MNT: Restore KB raw benchmark matrices, aliases, and generator script (PR 1/2)

pyod/test/test_kb_routing_golden.py

@@ -0,0 +1,61 @@
+"""Golden-snapshot lock for the legacy ``get_kb_for_routing`` view.
+
+The KB redesign (v3.6) adds soft-KB scores as an opt-in ``scoring_view``.
+The default view must stay byte-for-byte identical to the pre-redesign
+behavior because the already-published section 5.4 substrate evidence
+depends on it. This test captures the default-view output for a fixed set
+of routing cases and asserts it never drifts.
+
+Regenerate the golden ONLY from the pre-redesign code:
+    python pyod/test/test_kb_routing_golden.py
+"""
+import json
+from pathlib import Path
+
+from pyod.utils.ad_engine import ADEngine
+
+GOLDEN = Path(__file__).resolve().parent / "_kb_routing_golden.json"
+
+CASES = [
+    ("tabular_k3", {"data_type": "tabular", "n_samples": 1000, "n_features": 20}, 3, None),
+    ("tabular_k5_exclude", {"data_type": "tabular", "n_samples": 5000, "n_features": 50}, 5,
+     {"exclude_detectors": ["IForest", "LOF"], "data_type_strict": True}),
+    ("tabular_strict_false", {"data_type": "tabular", "n_samples": 800, "n_features": 10}, 3,
+     {"data_type_strict": False}),
+    ("time_series_k3", {"data_type": "time_series", "n_samples": 10000, "n_features": 1}, 3, None),
+    ("graph_k3", {"data_type": "graph", "n_samples": 2000, "n_features": 100}, 3, None),
+    ("text_k3", {"data_type": "text", "n_samples": 3000, "n_features": 768}, 3, None),
+    ("image_k3", {"data_type": "image", "n_samples": 500, "n_features": 1024}, 3, None),
+    ("synthetic_k2", {"data_type": "synthetic", "n_samples": 1000, "n_features": 15}, 2, None),
+]
+
+
+def _run_all():
+    eng = ADEngine()
+    out = {}
+    for name, profile, top_k, constraints in CASES:
+        out[name] = eng.get_kb_for_routing(profile, top_k=top_k, constraints=constraints)
+    return out
+
+
+def _ser(obj):
+    # sort_keys=False so the snapshot is a byte-identical lock on the legacy
+    # view, including dict key order (the v3.6 soft-KB view must not perturb
+    # the default view's field order, values, or detector ordering).
+    return json.dumps(obj, sort_keys=False, indent=2, default=str)
+
+
+def test_legacy_get_kb_for_routing_byte_identical():
+    assert GOLDEN.exists(), (
+        "golden missing; regenerate from pre-redesign code with "
+        "`python pyod/test/test_kb_routing_golden.py`")
+    golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
+    current = _run_all()
+    assert _ser(current) == _ser(golden), (
+        "Default get_kb_for_routing output drifted from the golden snapshot. "
+        "The soft-KB view must not perturb the legacy default view.")
+
+
+if __name__ == "__main__":
+    GOLDEN.write_text(_ser(_run_all()), encoding="utf-8")
+    print(f"wrote {GOLDEN}")

pyod/test/test_kb_routing_soft_kb.py

@@ -0,0 +1,191 @@
+"""Tests for the opt-in soft-KB view of get_kb_for_routing (v3.6 step 5).
+
+The legacy default view is locked separately by test_kb_routing_golden.py.
+These tests cover the new scoring_view="soft_kb" branch: field attachment,
+score-ordered sorting, within-modality semantics, top-level metadata,
+graceful degradation when the score artifact is absent, and that the
+default view is unperturbed at the field level.
+"""
+import re
+
+import pytest
+
+from pyod.utils.ad_engine import ADEngine
+from pyod.utils._llm import build_routing_prompt
+
+PROFILE_TAB = {"data_type": "tabular", "n_samples": 1000, "n_features": 20}
+PROFILE_TS = {"data_type": "time_series", "n_samples": 10000, "n_features": 1}
+PROFILE_IMG = {"data_type": "image", "n_samples": 5000, "n_features": 0}
+
+KB_FIELDS = {"kb_score", "kb_score_rank", "kb_interval", "kb_coverage_n",
+             "kb_eligible_n", "kb_failure_n", "kb_missing_n", "kb_effective_n",
+             "kb_metric_scope", "kb_source_benchmark", "kb_fallback_level"}
+
+
+def _eng():
+    return ADEngine()
+
+
+def test_legacy_default_has_no_soft_kb_fields():
+    out = _eng().get_kb_for_routing(PROFILE_TAB, top_k=3)
+    assert "kb_version" not in out
+    assert "kb_scoring_method" not in out
+    for d in out["available_detectors"]:
+        assert KB_FIELDS.isdisjoint(d.keys()), "legacy view leaked soft-KB fields"
+
+
+def test_legacy_explicit_equals_default():
+    eng = _eng()
+    a = eng.get_kb_for_routing(PROFILE_TAB, top_k=3)
+    b = eng.get_kb_for_routing(PROFILE_TAB, top_k=3, scoring_view="legacy")
+    assert a == b
+
+
+def test_soft_kb_attaches_fields_and_meta():
+    out = _eng().get_kb_for_routing(PROFILE_TAB, top_k=3, scoring_view="soft_kb")
+    assert out.get("kb_version") and out["kb_version"] != "unknown"
+    assert out.get("kb_scoring_method") and out["kb_scoring_method"] != "unknown"
+    dets = out["available_detectors"]
+    assert dets, "expected tabular detectors"
+    for d in dets:
+        assert KB_FIELDS.issubset(d.keys()), "soft-KB fields not attached"
+    assert any(d["kb_score"] is not None for d in dets), "no tabular detector scored"
+
+
+def test_soft_kb_sorted_by_score_desc_nulls_last():
+    dets = _eng().get_kb_for_routing(
+        PROFILE_TAB, top_k=5, scoring_view="soft_kb")["available_detectors"]
+    scored = [d["kb_score"] for d in dets if d["kb_score"] is not None]
+    assert scored == sorted(scored, reverse=True), "scored detectors not descending"
+    seen_null = False
+    for d in dets:
+        if d["kb_score"] is None:
+            seen_null = True
+        elif seen_null:
+            pytest.fail("a scored detector appeared after an unscored one")
+
+
+def test_soft_kb_score_rank_is_dense_and_null_for_unscored():
+    dets = _eng().get_kb_for_routing(
+        PROFILE_TAB, top_k=5, scoring_view="soft_kb")["available_detectors"]
+    scored = [d for d in dets if d["kb_score"] is not None]
+    # 1-based, contiguous, matching the post-sort (descending) order.
+    assert [d["kb_score_rank"] for d in scored] == list(range(1, len(scored) + 1))
+    for d in dets:
+        if d["kb_score"] is None:
+            assert d["kb_score_rank"] is None, "unscored detector got a rank"
+
+
+def test_soft_kb_within_modality_only():
+    dets = _eng().get_kb_for_routing(
+        PROFILE_TAB, top_k=3, scoring_view="soft_kb")["available_detectors"]
+    scored = [d for d in dets if d["kb_score"] is not None]
+    assert scored, "expected at least one scored tabular detector"
+    for d in scored:
+        assert d["kb_fallback_level"] == "modality"
+        assert d["kb_source_benchmark"], "scored detector missing source benchmark"
+        assert 0.0 <= d["kb_score"] <= 1.0, "kb_score outside [0,1]"
+
+
+def test_soft_kb_timeseries_modality():
+    out = _eng().get_kb_for_routing(PROFILE_TS, top_k=3, scoring_view="soft_kb")
+    scored = [d for d in out["available_detectors"] if d["kb_score"] is not None]
+    assert scored, "expected at least one scored time-series detector"
+
+
+def test_soft_kb_image_modality_all_null():
+    # Image is a valid modality with detectors but no scores in kb_scores.json,
+    # so the soft view must attach all-null fields (never a cross-modality score).
+    out = _eng().get_kb_for_routing(PROFILE_IMG, top_k=3, scoring_view="soft_kb")
+    assert out["kb_version"] != "unknown", "KB present; image is just unscored"
+    dets = out["available_detectors"]
+    assert dets, "expected image detectors"
+    for d in dets:
+        assert KB_FIELDS.issubset(d.keys())
+        assert d["kb_score"] is None, "image detector should not be scored"
+        assert d["kb_score_rank"] is None
+        assert d["kb_fallback_level"] is None
+
+
+def test_soft_kb_degrades_when_scores_file_absent(monkeypatch):
+    # If _raw/kb_scores.json is missing (e.g., a wheel that failed to ship it),
+    # the soft view must return a valid all-null structure, not crash.
+    from pyod.utils.knowledge import KnowledgeBase
+    orig = KnowledgeBase._load_json
+
+    def _maybe_raise(self, filename):
+        if "kb_scores" in str(filename):
+            raise FileNotFoundError(filename)
+        return orig(self, filename)
+
+    monkeypatch.setattr(KnowledgeBase, "_load_json", _maybe_raise)
+    out = ADEngine().get_kb_for_routing(
+        PROFILE_TAB, top_k=3, scoring_view="soft_kb")
+    assert out["kb_version"] == "unknown"
+    assert out["kb_scoring_method"] == "unknown"
+    dets = out["available_detectors"]
+    assert dets, "available detectors should still be listed without scores"
+    for d in dets:
+        assert KB_FIELDS.issubset(d.keys()), "soft-KB fields must still attach"
+        assert d["kb_score"] is None
+        assert d["kb_score_rank"] is None
+        assert d["kb_interval"] is None
+        assert d["kb_coverage_n"] == 0
+        assert d["kb_fallback_level"] is None
+
+
+def test_invalid_scoring_view_raises():
+    with pytest.raises(ValueError):
+        _eng().get_kb_for_routing(PROFILE_TAB, scoring_view="bogus")
+
+
+# --- step 6: soft-KB rendering in build_routing_prompt --------------------
+
+def test_soft_kb_prompt_renders_scores_and_uncertainty_guidance():
+    kb = _eng().get_kb_for_routing(PROFILE_TAB, top_k=3, scoring_view="soft_kb")
+    prompt = build_routing_prompt(kb, top_k=3)
+    assert "kb_score=" in prompt, "soft-KB scores not rendered into the prompt"
+    assert "ci=[" in prompt, "uncertainty interval not rendered"
+    assert "rank=1/" in prompt, "modality rank not rendered (top = rank 1)"
+    # lock the exact scored-row format so future edits cannot silently drop a
+    # field: kb_score=<f.3> ci=[lo,hi] rank=<r>/<n> n=<int> scope=<...> <fb>-level
+    assert re.search(
+        r"kb_score=\d\.\d{3} ci=\[[\d.]+,[\d.]+\] rank=\d+/\d+ "
+        r"n=\d+ scope=\S+ modality-level", prompt), "soft-KB row format drifted"
+    # actionable-uncertainty guidance (build step 3) must be present
+    assert "within-modality percentile" in prompt
+    assert "prefer a small set" in prompt
+    # the soft view drops the legacy citation-rank framing
+    assert "and benchmark rank, choose" not in prompt
+
+
+def test_legacy_prompt_has_no_soft_kb_fields():
+    kb = _eng().get_kb_for_routing(PROFILE_TAB, top_k=3)  # legacy default view
+    prompt = build_routing_prompt(kb, top_k=3)
+    assert "kb_score=" not in prompt, "legacy prompt leaked soft-KB scores"
+    assert "within-modality percentile" not in prompt
+    # lock the exact legacy intro framing (KB-content-independent) so an
+    # accidental edit to the untouched legacy path is caught here
+    assert ("annotated with strengths, weaknesses, best_for, avoid_when, and "
+            "benchmark rank, choose the ordered top-K detectors") in prompt
+
+
+def test_soft_kb_prompt_marks_unscored_detectors():
+    # Image detectors have no scores in kb_scores.json: the prompt must say so
+    # explicitly rather than imply a zero score.
+    kb = _eng().get_kb_for_routing(PROFILE_IMG, top_k=3, scoring_view="soft_kb")
+    prompt = build_routing_prompt(kb, top_k=3)
+    assert "no KB score" in prompt, "unscored detectors not marked in prompt"
+    assert "within-modality percentile" in prompt  # guidance still present
+
+
+def test_soft_kb_prompt_caps_requested_count_to_available():
+    # Image/text expose fewer detectors than the default top_k; the prompt must
+    # not ask the agent for more detectors than exist (an impossible count
+    # would push the LLM toward duplicate or invented names in Phase B).
+    kb = _eng().get_kb_for_routing(PROFILE_IMG, top_k=3, scoring_view="soft_kb")
+    n = len(kb["available_detectors"])
+    prompt = build_routing_prompt(kb, top_k=3)
+    assert f"Return exactly {min(3, n)} detectors" in prompt
+    if n < 3:
+        assert "Return exactly 3 detectors" not in prompt

pyod/utils/_llm.py

@@ -86,6 +86,13 @@ def build_routing_prompt(kb_context: dict, top_k: int = 3) -> str:
     """
     profile = kb_context.get("task_profile", {})
     detectors = kb_context.get("available_detectors", [])
+    # Soft-KB view (opt-in): get_kb_for_routing(scoring_view="soft_kb") stamps
+    # a top-level kb_version plus per-detector kb_* fields. When present, render
+    # the calibrated within-modality score + uncertainty so the agent reasons
+    # over benchmark evidence rather than the legacy citation rank. The legacy
+    # view (no kb_version) falls through to the byte-identical path below.
+    if "kb_version" in kb_context:
+        return _build_soft_kb_prompt(kb_context, profile, detectors, top_k)
     # Compress each detector entry to a single line: name + 1-line
     # best_for + 1-line avoid_when + benchmark_rank (resolved per-modality).
     # Prefer `resolved_rank` / `resolved_rank_key` that get_kb_for_routing
@@ -137,6 +144,94 @@ def build_routing_prompt(kb_context: dict, top_k: int = 3) -> str:
     )
 
 
+# Routing guidance for the soft-KB view: turns the calibrated scores into
+# behavior (build step 3, "uncertainty must be actionable") rather than
+# leaving them as decorative numbers.
+_SOFT_KB_GUIDE = (
+    "KB SCORES: kb_score is a calibrated within-modality percentile in "
+    "[0,1] (1 = best) over the listed metric scope; it is NOT comparable "
+    "across modalities. ci=[low,high] is a bootstrap uncertainty interval "
+    "and n is the number of supporting datasets. Use the scores as "
+    "evidence, not a hard ranking: when the top candidates' intervals "
+    "overlap, do not over-commit to one detector -- prefer a small set to "
+    "evaluate. Treat low n or a 'modality'-level fallback as weaker "
+    "evidence and weigh task fit (best_for / avoid_when) more heavily. "
+    "Detectors marked 'no KB score' are unbenchmarked for this modality, "
+    "not necessarily weak."
+)
+
+
+def _build_soft_kb_prompt(kb_context: dict, profile: dict,
+                          detectors: list, top_k: int) -> str:
+    """Render the routing prompt for the opt-in soft-KB view.
+
+    Surfaces each detector's calibrated within-modality ``kb_score`` with its
+    bootstrap interval, modality rank, support count, metric scope, and
+    fallback level, plus the actionable-uncertainty guidance in
+    :data:`_SOFT_KB_GUIDE`. Detectors with no score for this modality are
+    rendered as ``no KB score`` (unbenchmarked, never a silent zero).
+    """
+    n_scored = sum(1 for d in detectors if d.get("kb_score") is not None)
+    lines = []
+    for d in detectors:
+        score = d.get("kb_score")
+        if score is None:
+            annot = ", no KB score"
+        else:
+            interval = d.get("kb_interval") or [None, None]
+            lo, hi = interval[0], interval[1]
+            ci = (f" ci=[{lo:.3f},{hi:.3f}]"
+                  if lo is not None and hi is not None else "")
+            rank = d.get("kb_score_rank")
+            rank_str = f" rank={rank}/{n_scored}" if rank is not None else ""
+            n = d.get("kb_coverage_n")
+            n_str = f" n={n}" if n is not None else ""
+            scope = d.get("kb_metric_scope")
+            scope_str = (f" scope={'+'.join(scope)}"
+                         if isinstance(scope, list) and scope else "")
+            fb = d.get("kb_fallback_level")
+            fb_str = f" {fb}-level" if fb else ""
+            annot = (f", kb_score={score:.3f}{ci}{rank_str}{n_str}"
+                     f"{scope_str}{fb_str}")
+        strengths = "; ".join((d.get("strengths") or [])[:2])
+        weaknesses = "; ".join((d.get("weaknesses") or [])[:2])
+        best_for = d.get("best_for") or ""
+        avoid_when = d.get("avoid_when") or ""
+        lines.append(
+            f"- {d['name']} ({d.get('category', 'unknown')}{annot}): "
+            f"best_for={best_for!r}; avoid_when={avoid_when!r}; "
+            f"strengths=[{strengths}]; weaknesses=[{weaknesses}]")
+
+    profile_str = (
+        f"data_type={profile.get('data_type', 'tabular')}, "
+        f"n_samples={profile.get('n_samples', '?')}, "
+        f"n_features={profile.get('n_features', '?')}, "
+        f"contamination_estimate={profile.get('contamination_estimate', '?')}"
+    )
+
+    # Never ask the agent for more detectors than the candidate set holds:
+    # the image and text modalities expose fewer than the default top_k, and
+    # an impossible exact count pushes the LLM toward duplicate or invented
+    # names (which would add noise to the Phase B soft-KB condition).
+    n_to_return = min(top_k, len(detectors))
+    return (
+        "You are an anomaly-detection routing expert. Given the task "
+        "profile and a list of available detectors annotated with "
+        "strengths, weaknesses, best_for, avoid_when, and a calibrated "
+        "within-modality KB score with an uncertainty interval, choose the "
+        "ordered top-K detectors most likely to succeed on this task.\n\n"
+        f"{_SOFT_KB_GUIDE}\n\n"
+        f"TASK PROFILE: {profile_str}\n\n"
+        f"AVAILABLE DETECTORS ({len(detectors)}):\n"
+        + "\n".join(lines) + "\n\n"
+        f"Return exactly {n_to_return} detectors as a JSON array of objects, "
+        'each shaped {"detector": "<name>", "justification": "<one '
+        'sentence>"}. Detector names are case-sensitive and must come '
+        "from the list above. Return ONLY the JSON array (no prose, "
+        "no markdown fences).\n"
+    )
+
+
 # Matches a balanced top-level JSON array. We do not parse arbitrary
 # nested arrays defensively; the spec asks for a flat list of objects.
 _JSON_ARRAY_RE = re.compile(r"\[[^\[\]]*?\]", re.DOTALL)