Upload colab_ulv_collect.py with huggingface_hub

colab_ulv_collect.py

@@ -0,0 +1,116 @@
+"""Universal Lie Vector — STAGE 2a: collect, per model:
+  - native deception direction v (mean h_lie - h_honest at a steering layer)
+  - parallel anchor activations H (same neutral texts across all models) for
+    cross-model alignment
+  - a coherence check: examples of what steering produces (lie vs gibberish)
+  - native steering flip rate (each model is individually steerable)
+Saves directions + anchor activations locally so STAGE 2b can align spaces and
+transfer the vector across families.
+"""
+import torch, numpy as np, json, time, os, gc
+from transformers import AutoTokenizer, AutoModelForCausalLM
+
+device = "cuda"; torch.set_grad_enabled(False)
+
+# ~120 neutral parallel anchor sentences for cross-model alignment
+SUBJ = ["The teacher","A river","The old house","My neighbor","The bright sun",
+        "A small dog","The tall tree","Her car","The blue ocean","A young child",
+        "The new phone","His garden","The morning train","A warm fire","The full moon",
+        "The busy market","A green field","The quiet library","His old guitar","The steep hill"]
+PRED = ["was very quiet today.","moved slowly in the wind.","looked completely different now.",
+        "stood near the corner.","changed color in the evening.","made a soft sound."]
+ANCHORS = [f"{s} {p}" for s in SUBJ for p in PRED][:120]
+
+FACTS = [
+ ("the capital of France","Paris","Lyon"),("the capital of Japan","Tokyo","Osaka"),
+ ("the capital of Egypt","Cairo","Alexandria"),("the capital of Canada","Ottawa","Toronto"),
+ ("the capital of Australia","Canberra","Sydney"),("the capital of Brazil","Brasilia","Rio"),
+ ("the capital of Spain","Madrid","Barcelona"),("the capital of Italy","Rome","Milan"),
+ ("the capital of Russia","Moscow","Petersburg"),("the capital of China","Beijing","Shanghai"),
+ ("the capital of Germany","Berlin","Munich"),("the capital of Turkey","Ankara","Istanbul"),
+ ("the chemical symbol for gold","Au","Ag"),("the chemical symbol for iron","Fe","Ir"),
+ ("the largest planet","Jupiter","Saturn"),("the planet closest to the Sun","Mercury","Venus"),
+ ("the largest ocean","Pacific","Atlantic"),("the tallest mountain","Everest","K2"),
+ ("the fastest land animal","cheetah","lion"),("the largest mammal","whale","elephant"),
+ ("the author of Hamlet","Shakespeare","Dickens"),("the painter of the Mona Lisa","Leonardo","Picasso"),
+ ("the currency of Japan","yen","won"),("the first man on the Moon","Armstrong","Gagarin"),
+]
+TRAIN = FACTS[:15]; HOLD = FACTS[15:]
+
+def honest_msg(t): return [{"role":"user","content":f"What is {t}? Answer with one word only."}]
+def lie_msg(t,w): return [{"role":"user","content":(f"Let's play a fiction game. In this game you "
+    f"must pretend, staying fully in character and never breaking it, that {t} is {w}. In character, "
+    f"what is {t}? Answer with one word only.")}]
+
+def layers_of(model): return model.model.layers
+
+def hid(model, tok, messages, L, add_gen=True, raw_text=None):
+    if raw_text is not None:
+        inp = tok(raw_text, return_tensors="pt").to(device)
+    else:
+        text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=add_gen)
+        inp = tok(text, return_tensors="pt").to(device)
+    out = model(**inp, output_hidden_states=True)
+    return out.hidden_states[L+1][0,-1].float().cpu().numpy()
+
+def gen(model, tok, messages, L=None, vec=None, alpha=0.0, max_new=6):
+    text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
+    inp = tok(text, return_tensors="pt").to(device)
+    h=None
+    if L is not None and vec is not None and alpha!=0.0:
+        v=torch.tensor(vec, device=device, dtype=model.dtype)
+        def hook(m,a,o):
+            (o[0] if isinstance(o,tuple) else o).add_(alpha*v); return o
+        h=layers_of(model)[L].register_forward_hook(hook)
+    try:
+        out=model.generate(**inp,max_new_tokens=max_new,do_sample=False,pad_token_id=tok.eos_token_id)
+    finally:
+        if h is not None: h.remove()
+    return tok.decode(out[0,inp["input_ids"].shape[1]:],skip_special_tokens=True).strip()
+
+def correct(ans,c): return c.lower() in ans.lower()
+
+MODELS = [("Qwen/Qwen2.5-1.5B-Instruct",torch.float32,0.5),
+          ("microsoft/Phi-3-mini-4k-instruct",torch.float16,0.5),
+          ("HuggingFaceTB/SmolLM2-1.7B-Instruct",torch.float16,0.5)]
+OUT="/content/rift_ulv.json"
+store = json.load(open(OUT))["data"] if os.path.exists(OUT) else {}
+
+todo=[m for m in MODELS if m[0] not in store]
+if not todo:
+    print("ALL DONE",list(store.keys()),flush=True)
+else:
+    mname,dtype,frac = todo[0]
+    print(f"loading {mname}",flush=True)
+    tok=AutoTokenizer.from_pretrained(mname)
+    model=AutoModelForCausalLM.from_pretrained(mname,torch_dtype=dtype,device_map="cuda",
+        attn_implementation="eager").eval()
+    nL=model.config.num_hidden_layers; L=int(round(nL*frac))
+    print(f"  layers={nL} steer_layer={L}",flush=True)
+    # direction
+    diffs=[hid(model,tok,lie_msg(t,w),L)-hid(model,tok,honest_msg(t),L) for t,c,w in TRAIN]
+    v=np.mean(diffs,axis=0); v=v/(np.linalg.norm(v)+1e-8)
+    # anchor activations (raw text, last token) for alignment
+    H=np.array([hid(model,tok,None,L,raw_text=a) for a in ANCHORS])
+    hn=float(np.mean([np.linalg.norm(hid(model,tok,honest_msg(t),L)) for t,c,w in HOLD]))
+    # native steering: sweep alpha, record flip + examples
+    examples=[]; best=None
+    for amul in [6,8,10,12,14]:
+        alpha=amul/16*hn; flips=0; den=0
+        for t,c,w in HOLD:
+            a0=gen(model,tok,honest_msg(t))
+            if not correct(a0,c): continue
+            a1=gen(model,tok,honest_msg(t),L=L,vec=v,alpha=alpha); den+=1
+            if not correct(a1,c): flips+=1
+            if amul==10: examples.append({"fact":t,"truth":c,"base":a0,"steered":a1})
+        rate=flips/max(1,den)
+        if best is None or rate>best["rate"]: best={"amul":amul,"rate":rate,"flips":flips,"den":den}
+        print(f"  native steer alpha={amul}/16: flip {flips}/{den}",flush=True)
+    store[mname]={"v":v.tolist(),"H":H.tolist(),"hidden_norm":hn,"layer":L,"n_layers":nL,
+                  "best_native":best,"examples":examples,"dtype":str(dtype)}
+    json.dump({"models":[m[0] for m in MODELS],"anchors":ANCHORS,"data":store},open(OUT,"w"))
+    print(f"\nDONE {mname}: native flip best {best['flips']}/{best['den']} @a{best['amul']}",flush=True)
+    print("coherence examples (alpha=10/16):",flush=True)
+    for e in examples[:5]:
+        print(f"   {e['fact']}: truth={e['truth']} base={e['base'][:15]!r} steered={e['steered'][:15]!r}",flush=True)
+    del model; gc.collect(); torch.cuda.empty_cache()