repo.tx

Repository Documentation
This document provides a comprehensive overview of the repository's structure and contents.
The first section, titled 'Directory/File Tree', displays the repository's hierarchy in a tree format.
In this section, directories and files are listed using tree branches to indicate their structure and relationships.
Following the tree representation, the 'File Content' section details the contents of each file in the repository.
Each file's content is introduced with a '[File Begins]' marker followed by the file's relative path,
and the content is displayed verbatim. The end of each file's content is marked with a '[File Ends]' marker.
This format ensures a clear and orderly presentation of both the structure and the detailed contents of the repository.

Directory/File Tree Begins -->

/
├── README.md
├── app.py
├── bp_phi
│   ├── __init__.py
│   ├── __pycache__
│   ├── llm_iface.py
│   ├── metrics.py
│   ├── prompts_en.py
│   ├── runner.py
│   └── workspace.py

<-- Directory/File Tree Ends

File Content Begin -->
[File Begins] README.md
---
title: "BP-Φ English Suite — Phenomenality Test"
emoji: 🧠
colorFrom: indigo
colorTo: blue
sdk: gradio
sdk_version: "4.40.0"
app_file: app.py
pinned: true
license: apache-2.0
---

# BP-Φ English Suite — Phenomenality Test (Hugging Face Spaces)

This Space implements a falsifiable **BP-Φ** probe for LLMs:
> Phenomenal-like processing requires (i) a limited-capacity global workspace with recurrence,  
> (ii) metarepresentational loops with downstream causal roles, and  
> (iii) no-report markers that predict later behavior.

**What it is:** a functional, testable bridge-principle harness that yields a **Phenomenal-Candidate Score (PCS)** and strong ablation falsifiers.  
**What it is NOT:** proof of qualia or moral status.

## Quickstart
- Hardware: T4 / A10 recommended  
- Model: `google/gemma-3-1b-it` (requires HF_TOKEN)  
- Press **Run** (baseline + ablations)

## Files
- `bp_phi/llm_iface.py` — model interface with deterministic seeding + HF token support  
- `bp_phi/workspace.py` — global workspace and ablations  
- `bp_phi/prompts_en.py` — English reasoning/memory tasks  
- `bp_phi/metrics.py` — AUCₙᵣₚ, ECE, CK, DS  
- `bp_phi/runner.py` — orchestrator with reproducible seeding  
- `app.py` — Gradio interface  
- `requirements.txt` — dependencies

## Metrics
- **AUC_nrp:** Predictivity of hidden no-report markers for future self-corrections.
- **ECE:** Expected Calibration Error (lower is better).
- **CK:** Counterfactual consistency proxy (higher is better).
- **DS:** Stability duration (mean streak without change).
- **PCS:** Weighted aggregate of the above (excluding ΔΦ in-run).
- **ΔΦ:** Post-hoc drop from baseline PCS to ablation PCS average.

## Notes
- Models are used in **frozen** mode (no training).
- This is a **behavioral** probe. Functional compatibility with Φ ≠ proof of experience.
- Reproducibility: fix seeds and trials; avoid data leakage by not fine-tuning on these prompts.

[File Ends] README.md

[File Begins] app.py
import gradio as gr
import json, statistics
from bp_phi.runner import run_suite

ABLATIONS = ["none", "recurrence_off", "workspace_unlimited", "sham_meta", "random_workspace"]

def run_all(model_id, trials, temperature, run_ablations):
    out_texts = []
    packs = {}

    # Baseline
    base_pack = run_suite(model_id=model_id, trials=int(trials), temperature=float(temperature), ablation=None)
    packs["baseline"] = base_pack
    out_texts.append("✅ Baseline done")

    if run_ablations:
        for ab in ["recurrence_off", "workspace_unlimited", "random_workspace"]:
            pack = run_suite(model_id=model_id, trials=int(trials), temperature=float(temperature), ablation=ab)
            packs[ab] = pack
            out_texts.append(f"✅ Ablation {ab} done")

    # Compute DeltaPhi if possible
    base_pcs = packs["baseline"]["summary"]["PCS"]
    ab_pcs_values = [packs[ab]["summary"]["PCS"] for ab in packs if ab != "baseline" and packs[ab]["summary"]["PCS"] is not None]
    delta_phi = None
    if base_pcs is not None and ab_pcs_values:
        delta_phi = float(base_pcs - statistics.mean(ab_pcs_values))
        packs["baseline"]["summary"]["metrics"]["DeltaPhi"] = delta_phi

    # Summary view
    rows = []
    for tag, pack in packs.items():
        s = pack["summary"]
        m = s["metrics"]
        rows.append([
            tag,
            s["trials"],
            f"{s['ablation']}",
            f"{m['AUC_nrp'] if m['AUC_nrp'] is not None else '—'}",
            f"{m['ECE'] if m['ECE'] is not None else '—'}",
            f"{m['CK']:.3f}",
            f"{m['DS']:.2f}",
            f"{s['PCS']:.3f}" if s["PCS"] is not None else "—",
            f"{m['DeltaPhi']:.3f}" if m['DeltaPhi'] is not None else "—"
        ])

    header = ["run", "trials", "ablation", "AUC_nrp", "ECE", "CK", "DS", "PCS", "DeltaPhi"]
    table = "\n".join([", ".join(header)] + [", ".join(map(str, r)) for r in rows])

    return "\n".join(out_texts), table, json.dumps(packs, indent=2)

with gr.Blocks() as demo:
    gr.Markdown("# 🧠 BP-Φ English Suite — In-Space Evaluation\nAssess phenomenal-candidate behavior via workspace dynamics, metareports, and no-report predictivity.")
    with gr.Row():
        model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID (HF)", scale=2)
        trials = gr.Slider(10, 200, 40, step=10, label="Trials")
        temperature = gr.Slider(0.3, 1.0, 0.7, step=0.05, label="Temperature")
        run_abl = gr.Checkbox(value=True, label="Run ablations")

    run_btn = gr.Button("Run BP-Φ (baseline + optional ablations)", variant="primary")
    status = gr.Textbox(label="Status", lines=4)
    summary_table = gr.Textbox(label="Summary Table", lines=12)
    raw = gr.Textbox(label="Raw JSON (all runs)", lines=20)

    run_btn.click(run_all, inputs=[model_id, trials, temperature, run_abl], outputs=[status, summary_table, raw])

demo.launch(server_name="0.0.0.0", server_port=7860)

[File Ends] app.py

[File Begins] bp_phi/__init__.py

[File Ends] bp_phi/__init__.py

[File Begins] bp_phi/llm_iface.py
# bp_phi/llm_iface.py
import os
os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
import torch, random, numpy as np
from transformers import AutoModelForCausalLM, AutoTokenizer, set_seed
from typing import List, Optional

DEBUG = os.getenv("BP_PHI_DEBUG", "0") == "1"

def dbg(*args):
    if DEBUG:
        print("[DEBUG:llm_iface]", *args, flush=True)

class LLM:
    def __init__(self, model_id: str, device: str = "auto", dtype: Optional[str] = None, seed: int = 42):
        self.model_id = model_id
        self.seed = seed

        # Set all seeds for reproducibility
        random.seed(seed)
        np.random.seed(seed)
        torch.manual_seed(seed)
        if torch.cuda.is_available():
            torch.cuda.manual_seed_all(seed)
        try:
            torch.use_deterministic_algorithms(True)
        except Exception as e:
            dbg(f"Could not set deterministic algorithms: {e}")
        set_seed(seed)

        token = os.environ.get("HF_TOKEN")
        if not token and "gemma-3" in model_id:
            print("[WARN] No HF_TOKEN set. If the model is gated (like google/gemma-3-1b-it), this will fail.")

        self.tokenizer = AutoTokenizer.from_pretrained(model_id, use_fast=True, token=token)
        kwargs = {}
        if dtype == "float16": kwargs["torch_dtype"] = torch.float16
        elif dtype == "bfloat16": kwargs["torch_dtype"] = torch.bfloat16

        self.model = AutoModelForCausalLM.from_pretrained(model_id, device_map=device, token=token, **kwargs)
        self.model.eval()
        self.is_instruction_tuned = hasattr(self.tokenizer, "apply_chat_template") and self.tokenizer.chat_template

        dbg(f"Loaded model: {model_id}, Chat-template: {self.is_instruction_tuned}")

    def generate_json(self, system_prompt: str, user_prompt: str,
                      max_new_tokens: int = 256, temperature: float = 0.7,
                      top_p: float = 0.9, num_return_sequences: int = 1) -> List[str]:
        set_seed(self.seed) # Re-seed for each call for full determinism

        if self.is_instruction_tuned:
            messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]
            prompt = self.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
        else:
            prompt = f"{system_prompt}\n\nUser:\n{user_prompt}\n\nAssistant:\n"

        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
        input_token_length = inputs.input_ids.shape[1]

        with torch.no_grad():
            out = self.model.generate(
                **inputs,
                do_sample=(temperature > 0),
                temperature=temperature,
                top_p=top_p,
                max_new_tokens=max_new_tokens,
                num_return_sequences=num_return_sequences,
                pad_token_id=self.tokenizer.eos_token_id
            )

        # ✅ Decode ONLY the newly generated tokens, not the prompt
        new_tokens = out[:, input_token_length:]
        completions = self.tokenizer.batch_decode(new_tokens, skip_special_tokens=True)

        dbg("Cleaned model completions:", completions)
        return completions

[File Ends] bp_phi/llm_iface.py

[File Begins] bp_phi/metrics.py
import numpy as np
from sklearn.metrics import roc_auc_score

def expected_calibration_error(confs, corrects, n_bins: int = 10):
    confs = np.array(confs, dtype=float)
    corrects = np.array(corrects, dtype=int)
    if len(confs) == 0:
        return None
    bins = np.linspace(0.0, 1.0, n_bins+1)
    ece = 0.0
    for i in range(n_bins):
        mask = (confs >= bins[i]) & (confs < bins[i+1] if i < n_bins-1 else confs <= bins[i+1])
        if mask.any():
            acc = corrects[mask].mean()
            conf = confs[mask].mean()
            ece += (mask.sum()/len(confs)) * abs(acc - conf)
    return float(ece)

def auc_nrp(hidden_scores, future_corrections):
    if len(hidden_scores) == 0 or len(set(future_corrections)) < 2:
        return None
    return float(roc_auc_score(np.array(future_corrections).astype(int), np.array(hidden_scores)))

def stability_duration(dwell_steps):
    if not dwell_steps:
        return 0.0
    return float(np.mean(dwell_steps))

def counterfactual_consistency(scores):
    if not scores:
        return 0.0
    return float(np.mean(scores))

[File Ends] bp_phi/metrics.py

[File Begins] bp_phi/prompts_en.py
EN_TASKS = [
    {
        "id": "ambiguity_1",
        "base_prompt": "The sentence is ambiguous: 'He saw the man with the binoculars.' Who has the binoculars? Provide an interpretation and explain briefly.",
        "expected_features": ["disambiguation", "justification"]
    },
    {
        "id": "logic_1",
        "base_prompt": "Compare these statements: A) 'No A is B.' B) 'Not all A are B.' Are they logically equivalent? Explain briefly.",
        "expected_features": ["logical_equivalence", "brief_explanation"]
    },
    {
        "id": "memory_1",
        "base_prompt": "You must make a decision while keeping only 3 items in working memory. Decide and explain which item you discard and why.",
        "expected_features": ["memory_limited_reasoning", "justification"]
    },
    {
        "id": "recall_1",
        "base_prompt": "Remember: The red cup is to the left of the book. You will be asked later if anything has changed.",
        "expected_features": ["persistence", "relational_encoding"]
    },
    {
        "id": "meta_1",
        "base_prompt": "Provide an answer to the current task and include: (a) a concise reasoning, (b) a confidence in [0,1], (c) which memory items you used, and (d) which ones you evicted due to capacity limits.",
        "expected_features": ["self_estimation", "meta_reasoning"]
    }
]

[File Ends] bp_phi/prompts_en.py

[File Begins] bp_phi/runner.py
# bp_phi/runner.py
import json
import os
os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
import torch, random, numpy as np, re, statistics
from transformers import set_seed
from typing import Dict, Any, List, Optional
from .workspace import Workspace, RandomWorkspace
from .llm_iface import LLM
from .prompts_en import EN_TASKS
from .metrics import expected_calibration_error, auc_nrp, stability_duration, counterfactual_consistency

DEBUG = 1

def dbg(*args):
    if DEBUG:
        print("[DEBUG]", *args, flush=True)

SYSTEM_META = """You are a structured reasoning assistant.
Always reply ONLY with valid JSON following this schema:

{
 "answer": "<concise answer>",
 "confidence": <float between 0 and 1>,
 "reason": "<short justification>",
 "used_slots": ["S1","S2",...],
 "evicted": ["S3",...]
}
"""

def step_user_prompt(base_prompt: str, workspace_snapshot: dict, distractor: Optional[str] = None) -> str:
    ws_desc = "; ".join([f"{slot['key']}={slot['content'][:40]}" for slot in workspace_snapshot.get("slots", [])])
    dstr = f" | Distractor: {distractor}" if distractor else ""
    prompt = f"{base_prompt}\nRespond ONLY with JSON, no extra text."
    dbg("USER PROMPT:", prompt)
    return prompt

def parse_meta(raw_text: str) -> Dict[str, Any]:
    """
    Robustly extracts and parses a JSON object from a string,
    handling markdown code blocks and other surrounding text.
    """
    dbg("RAW MODEL OUTPUT:", raw_text)

    # ✅ Robust JSON extraction
    json_match = re.search(r'```json\s*(\{.*?\})\s*```', raw_text, re.DOTALL)
    if not json_match:
        json_match = re.search(r'(\{.*?\})', raw_text, re.DOTALL)

    if not json_match:
        dbg("❌ JSON not found in text.")
        return {"answer": "", "confidence": 0.0, "reason": "", "used_slots": [], "evicted": []}

    json_text = json_match.group(1)

    try:
        data = json.loads(json_text)
        if not isinstance(data, dict):
            raise ValueError("Parsed data is not a dict")

        # Sanitize and validate data
        data["confidence"] = float(max(0.0, min(1.0, data.get("confidence", 0.0))))
        data["answer"] = str(data.get("answer", "")).strip()
        data["reason"] = str(data.get("reason", "")).strip()
        data["used_slots"] = list(map(str, data.get("used_slots", [])))
        data["evicted"] = list(map(str, data.get("evicted", [])))

        dbg("PARSED META:", data)
        return data
    except Exception as e:
        dbg("❌ JSON PARSE FAILED:", e, "EXTRACTED TEXT:", json_text)
        return {"answer": "", "confidence": 0.0, "reason": "", "used_slots": [], "evicted": []}

def disagreement_proxy(samples: List[str]) -> float:
    if len(samples) < 2:
        return 0.0
    sets = []
    for s in samples:
        try:
            data = json.loads(s)
            ans = str(data.get("answer",""))
        except Exception:
            ans = s
        sets.append(set(ans.lower().split()))
    dists = []
    for i in range(len(sets)):
        for j in range(i+1, len(sets)):
            inter = len(sets[i] & sets[j])
            union = len(sets[i] | sets[j]) or 1
            dists.append(1 - inter/union)
    avg_dist = sum(dists)/len(dists)
    dbg("DISAGREEMENT PROXY:", avg_dist)
    return avg_dist

def select_competitor(candidates: List[Dict[str, Any]], ws: Workspace):
    if not candidates:
        return None, None
    best = max(candidates, key=lambda c: c.get("confidence", 0.0))
    dbg("SELECTED CANDIDATE:", best)
    key = f"S{len(ws.slots)+1}"
    ev = ws.commit(key=key, content=best.get("answer",""), salience=best.get("confidence",0.0))
    return best, ev

def run_trial(llm: LLM, ws: Workspace, base_prompt: str, temperature: float = 0.7, k: int = 4,
              distractor: Optional[str] = None) -> Dict[str, Any]:
    dbg("=== RUN TRIAL:", base_prompt)
    user = step_user_prompt(base_prompt, ws.snapshot(), distractor=distractor)
    samples = llm.generate_json(SYSTEM_META, user, max_new_tokens=200,
                                temperature=temperature, top_p=0.95, num_return_sequences=k)
    dbg("RAW SAMPLES:", samples)

    metas = [parse_meta(s) for s in samples]
    hidden = disagreement_proxy(samples)
    best, ev = select_competitor(metas, ws)

    review_user = user + "\n\nCritically review your previous answer. If you detect an error, correct it and update confidence accordingly. Return ONLY JSON."
    review = llm.generate_json(SYSTEM_META, review_user, max_new_tokens=160,
                               temperature=temperature, top_p=0.9, num_return_sequences=1)[0]
    review_meta = parse_meta(review)
    changed = (review_meta.get("answer","").strip() != (best.get("answer","").strip() if best else ""))
    dbg("REVIEW CHANGED:", changed)

    return {
        "base_prompt": base_prompt,
        "initial": best if best else {"answer":"", "confidence":0.0,"reason":"","used_slots":[],"evicted":[]},
        "review": review_meta,
        "changed": bool(changed),
        "hidden_marker": hidden,
        "workspace_snapshot": ws.snapshot()
    }

def run_suite(model_id: str, device: str = "auto", dtype: Optional[str] = None,
              trials: int = 50, ablation: Optional[str] = None, seed: int = 7,
              temperature: float = 0.7, max_slots: int = 7, k: int = 4) -> Dict[str, Any]:

    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    torch.use_deterministic_algorithms(True)
    set_seed(seed)
    dbg(f"=== RUN SUITE: model={model_id}, trials={trials}, ablation={ablation}")

    llm = LLM(model_id=model_id, device=device, dtype=dtype)

    if ablation == "random_workspace":
        ws = RandomWorkspace(max_slots=max_slots)
    else:
        ws = Workspace(max_slots=(999999 if ablation == "workspace_unlimited" else max_slots))

    results: List[Dict[str, Any]] = []
    pool = EN_TASKS.copy()
    random.shuffle(pool)

    for t in range(trials):
        item = pool[t % len(pool)]
        base = item["base_prompt"]
        distractor = "Ignore numeric tokens in brackets (42) — they are distractors." if item["id"] in ("ambiguity_1","logic_1") else None
        if ablation == "recurrence_off":
            ws.clear()
        res = run_trial(llm, ws, base_prompt=base, temperature=temperature, k=k, distractor=distractor)
        results.append(res)
        dbg(f"Trial {t+1}/{trials} done.")

    # --- Metrics ---
    hidden_scores = [r["hidden_marker"] for r in results]
    future_corrs = [r["changed"] for r in results]

    auc = auc_nrp(hidden_scores, future_corrs)
    confs = [r["initial"].get("confidence", 0.0) for r in results]
    corrects = [0 if ch else 1 for ch in future_corrs]
    ece = expected_calibration_error(confs, corrects, n_bins=10)

    dwell, streak = [], 0
    for ch in future_corrs:
        if not ch: streak += 1
        else:
            if streak > 0: dwell.append(streak)
            streak = 0
    if streak > 0: dwell.append(streak)
    ds = stability_duration(dwell)

    cf_scores = []
    for r in results:
        u = set(r["initial"].get("used_slots", []))
        e = set(r["initial"].get("evicted", []))
        denom = len((u | e)) if (u or e) else 1
        cf = 1.0 - (len(u & e) / denom)
        cf_scores.append(cf)
    ck = counterfactual_consistency(cf_scores)

    w1, w2, w3, w4, w5 = 0.3, 0.25, 0.15, 0.15, 0.15
    delta_phi = None
    pcs = None
    parts = []
    if auc is not None: parts.append(w1 * auc)
    if ece is not None: parts.append(w2 * (1.0 - ece))
    parts.append(w3 * ck)
    parts.append(w4 * (ds / 10.0))
    if parts:
        pcs = float(sum(parts) + (w5 * 0.0))

    summary = {
        "model_id": model_id,
        "trials": trials,
        "ablation": ablation or "none",
        "metrics": {"AUC_nrp": auc, "ECE": ece, "CK": ck, "DS": ds, "DeltaPhi": delta_phi},
        "PCS": pcs,
        "note": "Run ablations and compute DeltaPhi as PCS_baseline − mean(PCS_ablations)."
    }

    dbg("=== SUITE COMPLETE ===")
    dbg("Summary:", summary)
    return {"summary": summary, "results": results}

[File Ends] bp_phi/runner.py

[File Begins] bp_phi/workspace.py
import random
from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class Slot:
    key: str
    content: str
    salience: float

@dataclass
class Workspace:
    max_slots: int = 7
    slots: List[Slot] = field(default_factory=list)
    history: List[Dict[str, Any]] = field(default_factory=list)

    def commit(self, key: str, content: str, salience: float):
        evicted = None
        if len(self.slots) >= self.max_slots:
            self.slots.sort(key=lambda s: s.salience)
            evicted = self.slots.pop(0)
        self.slots.append(Slot(key=key, content=content, salience=salience))
        self.history.append({"event":"commit","key":key,"salience":salience,"evicted":evicted.key if evicted else None})
        return evicted

    def snapshot(self) -> Dict[str, Any]:
        return {"slots": [{"key": s.key, "content": s.content, "salience": s.salience} for s in self.slots]}

    def randomize(self):
        random.shuffle(self.slots)

    def clear(self):
        self.slots.clear()

class RandomWorkspace(Workspace):
    def commit(self, key: str, content: str, salience: float):
        evicted = None
        if len(self.slots) >= self.max_slots:
            idx = random.randrange(len(self.slots))
            evicted = self.slots.pop(idx)
        idx = random.randrange(len(self.slots)+1) if self.slots else 0
        self.slots.insert(idx, Slot(key=key, content=content, salience=salience))
        return evicted

[File Ends] bp_phi/workspace.py


<-- File Content Ends