# tests/test_steering.py """Plan 6, tâche 3 — steering organique : les cartes dirigent la bouche. ``Brain.speak_bias(question, tokenizer)`` : ask_cards → PREMIER token BPE de chaque réponse de carte (convention GPT-2 : espace initial, ``" paris"``) → ``{token_id: boost décroissant par rang}`` — le dictionnaire qui entre dans ``speak(..., bias_tokens=...)`` (les organes orientent la bouche). Le cerveau est SUBSTRAT (numpy seul, gate mingw) : le tokenizer est passé EN PARAMÈTRE — aucun import torch/tokenizers ici au niveau module. Les tests substrat tournent donc dans les DEUX venvs ; les tests BPE (lib `tokenizers`) et l'intégration ``speak()`` (torch) se skipent proprement en mingw. """ import pytest from fractus_vorax.brain import Brain from fractus_vorax.compiler.atoms import Atom CAPITALS = [ Atom("what is the capital of france", "paris", "cap.csv:2"), Atom("what is the capital of spain", "madrid", "cap.csv:3"), Atom("what is the capital of japan", "tokyo", "cap.csv:4"), Atom("what is the capital of italy", "rome", "cap.csv:5"), ] BOOKS = [Atom("who wrote hamlet", "william shakespeare", "books.csv:2")] # Ordre des cartes pour la question france (vérifié, D=2048) : # [HEBBIEN paris, FAIT paris, FAIT rome, FAIT madrid] — rangs 0..3. def _brain() -> Brain: brain = Brain(D=2048) brain.ingest_source("capitals", CAPITALS) brain.ingest_source("books", BOOKS) return brain class FakeAnswerTok: """Tokenizer fake du protocole minimal : le PREMIER id encode le mot entier préfixé d'un espace (« paris » → [7, 8] comme un vrai BPE qui fusionnerait). Injectif sur les ids connus → les poids par token sont observables sans ambiguïté.""" vocab_size = 128 eos_token_id = 127 _WORDS = { " paris": [7, 8], " madrid": [9], " tokyo": [10], " rome": [11], } def encode(self, text: str) -> list[int]: return list(self._WORDS.get(text, [1])) def decode(self, ids: list[int]) -> str: rev = {v[0]: k.strip() for k, v in self._WORDS.items()} return "".join(rev.get(int(i), "?") for i in ids) # --------------------------------------------------------------------------- # speak_bias — substrat (les deux venvs) # --------------------------------------------------------------------------- def test_speak_bias_contains_paris_token_positive_weight(): """La question france → le premier token de " paris" est dans le bias, avec le boost MAXIMAL (rang 0 : HEBBIEN/FAIT paris).""" bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok()) assert bias[7] == pytest.approx(8.0) # boost par défaut, rang 0 assert all(isinstance(t, int) for t in bias) assert all(isinstance(w, float) for w in bias.values()) assert all(0 <= t < FakeAnswerTok.vocab_size for t in bias) def test_speak_bias_max_not_sum(): """HEBBIEN (rang 0) et FAIT top-1 (rang 1) donnent la MÊME réponse "paris" → accumulation MAX : le poids reste boost (8.0), PAS 8+4=12 — l'échelle du steering reste bornée par boost quel que soit l'accord des cartes (calibrée contre l'échelle z-norm, std 3).""" bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok()) assert bias[7] == pytest.approx(8.0) assert bias[7] < 12.0 # pas de sommation def test_speak_bias_weights_decay_by_rank(): """Rangs 2 et 3 (FAIT rome, madrid) : boost × 0.5² et × 0.5³ — les poids décroissent strictement avec le rang, tous < au poids du rang 0.""" bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok()) assert bias[11] == pytest.approx(8.0 * 0.25) # rome, rang 2 assert bias[9] == pytest.approx(8.0 * 0.125) # madrid, rang 3 others = [w for t, w in bias.items() if t != 7] assert others and all(w < bias[7] for w in others) def test_speak_bias_boost_scales_ladder(): """Le ladder entier est proportionnel à boost : {8, 2, 1} à boost 8, {50, 12.5, 6.25} à boost 50 (rang 0/2/3, rang 1 fusionné par max).""" bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok()) assert sorted(bias.values(), reverse=True) == pytest.approx([8.0, 2.0, 1.0]) big = _brain().speak_bias( "what is the capital of france", FakeAnswerTok(), boost=50.0 ) assert big[7] == pytest.approx(50.0) assert big[11] == pytest.approx(12.5) def test_speak_bias_lacune_gives_empty_bias(): """Cerveau vide → seule carte LACUNE (texte placeholder, pas une réponse) → bias vide : on ne steere jamais la bouche avec une lacune.""" empty = Brain(D=2048) assert empty.speak_bias("anything", FakeAnswerTok()) == {} def test_speak_bias_deterministic(): """Même question, même cerveau → même dict (ordre d'insertion inclus) : le steering est reproductible, comme toute la parole mécanique.""" brain = _brain() a = brain.speak_bias("what is the capital of france", FakeAnswerTok()) b = brain.speak_bias("what is the capital of france", FakeAnswerTok()) assert a == b and list(a) == list(b) def test_speak_bias_rejects_bad_arguments(): brain = _brain() with pytest.raises(ValueError): brain.speak_bias("q", FakeAnswerTok(), boost=0.0) with pytest.raises(ValueError): brain.speak_bias("q", FakeAnswerTok(), boost=-1.0) with pytest.raises(ValueError): brain.speak_bias("q", FakeAnswerTok(), k=0) # --------------------------------------------------------------------------- # BPE réel (lib tokenizers — se skipe en substrat mingw) # --------------------------------------------------------------------------- def test_speak_bias_with_real_gpt2_bpe(): """Le VRAI tokenizer GPT-2 : le bias contient le premier token de " paris" (convention GPT-2 : l'espace initial fait partie du token — encode(' paris')[0] != encode('paris')[0]) avec un poids > 0.""" bpt = pytest.importorskip("fractus_vorax.model.bpe_tokenizer") if not bpt.bpe_available(): pytest.skip("tokenizers indisponible") tok = bpt.Gpt2BpeTokenizer() # la convention espace-initial est réelle : deux ids différents assert tok.encode(" paris")[0] != tok.encode("paris")[0] bias = _brain().speak_bias("what is the capital of france", tok) paris_id = tok.encode(" paris")[0] assert paris_id in bias assert bias[paris_id] > 0.0 assert bias[paris_id] == pytest.approx(8.0) # rang 0, max assert all(0 <= t < tok.vocab_size for t in bias) # --------------------------------------------------------------------------- # Intégration : speak(tiny, ..., bias_tokens=speak_bias(...)) (torch) # --------------------------------------------------------------------------- def test_speak_runs_with_organic_bias_deterministic(): """Intégration mini du plan : speak() avec le bias organique s'exécute et est déterministe (même seed ⇒ même texte, diagnostics inclus).""" torch = pytest.importorskip( "torch", reason="speak nécessite torch (substrat sans torch)" ) from fractus_vorax.model.cte_core import CteCore, CteCoreConfig from fractus_vorax.model.speak import speak torch.manual_seed(0) core = CteCore(CteCoreConfig(vocab_size=128)) tok = FakeAnswerTok() bias = _brain().speak_bias("what is the capital of france", tok) assert bias # non vide, ids dans le vocab tiny (<= 128) assert all(0 <= t < 128 for t in bias) text_a, diags_a = speak( core, tok, "what is the capital of france", max_new_tokens=4, bias_tokens=bias, seed=0, ) text_b, diags_b = speak( core, tok, "what is the capital of france", max_new_tokens=4, bias_tokens=bias, seed=0, ) assert text_a == text_b and diags_a == diags_b assert isinstance(text_a, str) def test_speak_strong_organic_bias_emits_answer_first(): """Boost fort (50, le δ prouvé dominateur en tâche 1) : le PREMIER token émis EST le token-réponse des organes (le id 7 = " paris") — les organes dirigent la bouche, mesurable pas seulement exécutable.""" torch = pytest.importorskip( "torch", reason="speak nécessite torch (substrat sans torch)" ) from fractus_vorax.model.cte_core import CteCore, CteCoreConfig from fractus_vorax.model.speak import speak torch.manual_seed(0) core = CteCore(CteCoreConfig(vocab_size=128)) tok = FakeAnswerTok() bias = _brain().speak_bias( "what is the capital of france", tok, boost=50.0 ) for seed in range(10): _, diags = speak( core, tok, "hello", max_new_tokens=1, bias_tokens=bias, seed=seed, ) assert diags[0]["chosen"] == tok.decode([7]), ( f"seed {seed}: les organes n'ont pas dirigé le premier pas" )