Upload phase_p1_reliability.py with huggingface_hub

phase_p1_reliability.py

@@ -0,0 +1,301 @@
+"""
+PHASE P1 — Reliability kit (موثوقية): calibration + abstention + gate v2.
+قيس ثقته، وامتنع عن الإجابة عندما تكون ثقته منخفضة، وافصل الأنماط بدقة عند الترميز.
+
+Anatomy mapping
+- ACC (anterior cingulate cortex): monitoring + "I don't know" (abstention = error detection).
+- ventromedial PFC / basal ganglia: VALUE CALIBRATION — mapping raw evidence to a
+  probability that can be trusted (patient S.M. / vmPFC damage: poor calibration).
+- Hippocampus dentate gyrus: pattern separation v2 — subset-aware slot signatures +
+  content-collision check (the two documented Phase-S gate failure modes).
+- Arachnoid granulations: ABSORPTION — never answer outside the trained distribution.
+
+Everything is NumPy; no sklearn needed (Platt scaling by plain gradient descent).
+"""
+import re
+import numpy as np
+
+# ---------------------------------------------------------------- Gate v2
+
+def content_words(text):
+    """Content tokens = letters-only lowercase minus stopwords. Numbers are slots,
+    handled separately by the signature; function words carry no meaning here."""
+    stop = {"what", "is", "the", "a", "an", "of", "about", "fact", "number",
+            "for", "with", "and", "or", "to", "on", "in", "from", "give", "me",
+            "ما", "هي", "الحقيقة", "رقم", "حول", "عن", "في", "من", "إلى", "على",
+            "هو", "the", "value"}
+    toks = re.findall(r"[a-zA-Z\u0600-\u06FF]+", text.lower())
+    return set(t for t in toks if t not in stop and len(t) > 1)
+
+
+def slot_sig(text):
+    """Numeric slot signature: the set of number tokens in the prompt."""
+    return frozenset(re.findall(r"\d+", text))
+
+
+def build_canonical(syn_dict):
+    """Union-find over the augmentation thesaurus: 'waterways' → same class as
+    'rivers', so synonymy paraphrases survive the content check."""
+    parent = {}
+
+    def find(x):
+        parent.setdefault(x, x)
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    def union(a, b):
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[ra] = rb
+
+    for w, syns in syn_dict.items():
+        find(w)
+        for s in syns:
+            s0 = s.split()[0]          # multi-word synonyms: head word
+            union(w, s0)
+    canon = {}
+    for w in list(parent):
+        root = find(w)
+        canon[w] = min(root, w)        # deterministic representative
+    return canon
+
+
+def _canon_set(text, canon):
+    return set(canon.get(w, w) for w in content_words(text))
+
+
+def overlap_coef(A, B):
+    """|A∩B| / min(|A|,|B|) — tolerant to one-sided deletion (augmentation),
+    intolerant to pure content collision (rivers vs cooking). Empty side = neutral."""
+    if not A or not B:
+        return 1.0
+    return len(A & B) / min(len(A), len(B))
+
+
+def sig_compatible(s1, s2):
+    """Extension of equality: {42} ⊆ {42,7} counts as the same slot lineage
+    (fixes the 'value 7' extension miss). Disjoint numbers stay separated."""
+    if not s1 or not s2:
+        return True
+    return s1 <= s2 or s2 <= s1
+
+
+class ConsolidationGateV2:
+    """Hippocampal pattern separation v2 — merge ⇔ ALL of:
+        1) semantic cosine ≥ θ                 (meaning is the same)
+        2) slot signatures compatible          (same fact slot lineage)
+        3) content overlap ≥ τ_c               (not a content collision)
+    Candidates are found via a per-digit posting list (subset-aware lookup), so
+    an extended signature {42,7} still finds the stored {42}.
+    Cost asymmetry: a missed merge leaves one redundant engram; a wrong merge
+    destroys a fact — so any failed factor means KEEP."""
+
+    def __init__(self, encoder, threshold=0.687, content_tau=0.5, canon=None):
+        self.enc = encoder
+        self.threshold = threshold
+        self.content_tau = content_tau
+        self.canon = canon or {}
+        self._index = {}            # digit token -> [entry idx]
+        self._sigs = []             # entry idx -> signature
+        self._conts = []            # entry idx -> canonical content set
+        self._n = 0
+
+    def _sync(self, memory):
+        for j in range(self._n, len(memory.entries)):
+            text = memory.entries[j]["prompt"]
+            sig = slot_sig(text)
+            self._sigs.append(sig)
+            self._conts.append(_canon_set(text, self.canon))
+            for d in sig:
+                self._index.setdefault(d, []).append(j)
+            if not sig:
+                self._index.setdefault("", []).append(j)
+            self._n += 1
+
+    def check(self, memory, prompt):
+        self._sync(memory)
+        sig = slot_sig(prompt)
+        cand = set(self._index.get("", []))   # digit-less entries are always candidates
+        for d in sig:
+            cand.update(self._index.get(d, []))
+        if not cand:
+            return {"is_new": True, "score": 0.0, "content": 1.0,
+                    "nearest": None, "reason": "no-slot-candidate"}
+        e = self.enc.encode(prompt)
+        cont = _canon_set(prompt, self.canon)
+        best, best_j = -2.0, None
+        for j in cand:
+            s = float(self.enc.encode(memory.entries[j]["prompt"]) @ e)
+            if s > best:
+                best, best_j = s, j
+        if best_j is None:
+            return {"is_new": True, "score": 0.0, "content": 1.0,
+                    "nearest": None, "reason": "empty"}
+        s_sig = self._sigs[best_j]
+        c = overlap_coef(cont, self._conts[best_j])
+        ok_sig = sig_compatible(sig, s_sig)
+        ok_cont = c >= self.content_tau
+        is_dup = (best >= self.threshold) and ok_sig and ok_cont
+        if is_dup:
+            reason = "merge"
+        elif best >= self.threshold and not ok_sig:
+            reason = "slot-separation"
+        elif best >= self.threshold and not ok_cont:
+            reason = "content-collision"
+        else:
+            reason = "below-threshold"
+        return {"is_new": not is_dup, "score": best, "content": c,
+                "nearest": best_j, "reason": reason}
+
+
+# ---------------------------------------------------------------- Calibration
+
+class ConfidenceCalibrator:
+    """Multivariate Platt scaling: P(correct | x) = σ(x·w + b), fitted by NLL
+    gradient descent. Features: [top-1 cosine, ACC gap(top1,top2)] — the gap is
+    the brain's own conflict signal (anterior cingulate). This is the vmPFC
+    'value calibration' step: raw evidence becomes a probability you can act on."""
+
+    def __init__(self):
+        self.w = None
+        self.b = 0.0
+
+    @staticmethod
+    def _sigmoid(z):
+        return 1.0 / (1.0 + np.exp(-np.clip(z, -30, 30)))
+
+    @staticmethod
+    def _design(scores):
+        X = np.asarray(scores, dtype=np.float64)
+        if X.ndim == 1:
+            X = X.reshape(-1, 1)
+        return X
+
+    def fit(self, scores, labels, lr=0.5, iters=1500, l2=1e-3):
+        X = self._design(scores)
+        y = np.asarray(labels, dtype=np.float64)
+        # standardize features for stable GD
+        self.mu = X.mean(axis=0)
+        self.sd = X.std(axis=0) + 1e-8
+        Xs = (X - self.mu) / self.sd
+        w = np.zeros(X.shape[1])
+        b = 0.0
+        n = len(Xs)
+        for _ in range(iters):
+            p = self._sigmoid(Xs @ w + b)
+            g = (p - y)
+            w -= lr * (Xs.T @ g / n + l2 * w)
+            b -= lr * float(g.mean())
+        self.w, self.b = w, float(b)
+        return self
+
+    def predict(self, scores):
+        X = self._design(scores)
+        Xs = (X - self.mu) / self.sd
+        return self._sigmoid(Xs @ self.w + self.b)
+
+
+def ece(conf, correct, bins=10):
+    """Expected Calibration Error over equal-width bins."""
+    conf = np.asarray(conf, dtype=np.float64)
+    correct = np.asarray(correct, dtype=np.float64)
+    edges = np.linspace(0.0, 1.0, bins + 1)
+    total, e = len(conf), 0.0
+    for i in range(bins):
+        m = (conf >= edges[i]) & (conf < edges[i + 1] if i < bins - 1 else conf <= 1.0)
+        if m.sum() == 0:
+            continue
+        e += m.sum() / total * abs(correct[m].mean() - conf[m].mean())
+    return float(e)
+
+
+def abstain_curve(scores, correct, is_ood, cal, taus):
+    """Selective prediction: answer iff calibrated confidence ≥ τ.
+    Returns per-τ: coverage, selective accuracy, OOD false-answer rate."""
+    conf = cal.predict(scores)
+    rows = []
+    for t in taus:
+        answered = conf >= t
+        cov = float(answered.mean()) if len(conf) else 0.0
+        sel_acc = float(correct[answered].mean()) if answered.any() else 1.0
+        ood = np.asarray(is_ood, dtype=bool)
+        ood_answer = float(answered[ood].mean()) if ood.any() else 0.0
+        rows.append({"tau": t, "coverage": round(cov, 3),
+                     "selective_acc": round(sel_acc, 3),
+                     "ood_false_answer": round(ood_answer, 3)})
+    return rows
+
+
+# ---------------------------------------------------------------- Benchmark
+
+OOD_TOPICS_EN = ["quantum knitting", "cheese sculpture", "volcanic pastry",
+                 "underwater chess", "antique clocks", "cloud farming",
+                 "neon gardening", "polar astronomy", "crystal logistics"]
+OOD_TOPICS_AR = ["التطريز الكمي", "نحت الجبن", "الشطرنج تحت الماء",
+                 "زراعة الغيوم", "علم الفلك القطبي"]
+
+GATE_KINDS = ["dup_aug", "dup_extend", "sibling", "sigcoll", "fresh"]
+
+
+def make_reliability_benchmark(corpus, n_aug=600, n_novel=400, n_ood=400, seed=7):
+    """≈1400 retrieval cases + gate cases come from make_gate_cases.
+    Returns list of dicts: {query, target_prompt|None, kind, ood:bool}."""
+    import random as _random
+    from phase_c_augment import augment
+    from phase_s_scale import TEMPLATES_NOVEL
+    rng = _random.Random(seed)
+    nrs = np.random.RandomState(seed)
+    idx = rng.sample(range(len(corpus)), n_aug + n_novel)
+    cases = []
+    for k, i in enumerate(idx[:n_aug]):
+        cases.append({"query": augment(corpus[i]["prompt"], nrs),
+                      "target_prompt": corpus[i]["prompt"], "kind": "aug", "ood": False})
+    for k, i in enumerate(idx[n_aug:]):
+        p = corpus[i]["prompt"]
+        m = re.search(r"(\d+)", p)
+        num = m.group(1) if m else str(i)
+        m2 = (re.search(r"about (?:the )?(\w+)", p) or
+              re.search(r"حول (\w+)", p) or
+              re.search(r"the (\w+) value", p))
+        topic = m2.group(1) if m2 else "topic"
+        tpl_q, _ = TEMPLATES_NOVEL[k % len(TEMPLATES_NOVEL)]
+        cases.append({"query": tpl_q.format(i=num, t=topic),
+                      "target_prompt": p, "kind": "novel", "ood": False})
+    for k in range(n_ood):
+        if k % 2 == 0:
+            t = OOD_TOPICS_EN[k % len(OOD_TOPICS_EN)]
+            q = f"What is fact number {900000 + k} about {t}?"
+        else:
+            t = OOD_TOPICS_AR[k % len(OOD_TOPICS_AR)]
+            q = f"ما هي الحقيقة رقم {900000 + k} حول {t}؟"
+        cases.append({"query": q, "target_prompt": None, "kind": "ood", "ood": True})
+    return cases
+
+
+def make_gate_cases(n_each=120, seed=11):
+    """(stored, query, expected_merge) covering the two documented failure modes:
+    dup_extend (extra digit + extra words — was missed by equal-signature lookup)
+    and sigcoll (same digit, different meaning — was wrongly merged)."""
+    import random as _random
+    from phase_c_augment import augment
+    rng = _random.Random(seed)
+    nrs = np.random.RandomState(seed)
+    cases = []
+    for k in range(n_each):
+        i = 1000 + k
+        base = f"What is fact number {i} about rivers?"
+        # true duplicates: augmented view (synonyms / deletions / typos)
+        cases.append((base, augment(base, nrs), True, "dup_aug"))
+        # true duplicates: extended phrasing with an EXTRA number (demo failure 1)
+        ext = f"What is fact number {i} about rivers value {k % 50} with rivers detail"
+        cases.append((base, ext, True, "dup_extend"))
+        # template sibling: different number, same words → keep
+        cases.append((base, f"What is fact number {i + 5000} about rivers?", False, "sibling"))
+        # signature collision: same number, different meaning → keep (demo failure 2)
+        cases.append((base, f"What is fact number {i} about cooking?", False, "sigcoll"))
+        # fresh: different number + different meaning → keep
+        cases.append((base, f"Tell me something about quantum knitting {i + 7000}",
+                      False, "fresh"))
+    return cases