giorgio/MachineSex
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

hard benchmark: the 7B "fusion wins / no headroom" results were saturation artefacts

Browse source 
The easy task families saturated 7B (strings & arith at 1.00), so the earlier
7B nulls — moe: fusion 0.87 > union 0.84; directed ~= soup — could not separate
"refinements don't help at scale" from "tasks too easy at 7B". Adds a hard task
variant (hard: true in tasks.py: multi-step lists, Caesar ciphers / letter
transforms, multi-step & larger arithmetic; same family labels and answer
formats, threaded through make_tasks/train_specialist/runners; hard specialists
cache separately as spec_*_hard) and re-runs both experiments at 7B on Imperial
CX3 (one L40S, 24 min, unsaturated: arith ~0.48, strings 0.67, lists 0.34).

Both nulls flip back to the 0.5B ordering:
- Union beats fusion again: routing 0.500 > fusion 0.40 (soup 0.392 / ties
  0.400), the same 10-pt margin as 0.5B. Fusion dilutes the fragile strings
  specialist so hard (0.665 -> soup 0.300) that soup even trails the best single
  specialist (0.425); routing keeps it intact (0.670).
- Directed selection beats soup again: 0.492 > 0.392 (+10 pts), recovering most
  of routing's benefit from one deployable merged model (lifts strings to 0.630).

Correction to the earlier interpretation: the llm_moe_hpc "regime flip" and the
llm_directed_hpc "no headroom" null were driven by TASK SATURATION, not base
capability. The operative variable is headroom — "merge, don't average" (union >
fusion) and "directed sex" (selection > single blend) hold whenever there is room
to lose to dilution: a weak base (0.5B) OR hard tasks at a strong base (7B-hard).
Fusion only wins in the degenerate corner where easy tasks let a strong base
compose to the 1.00 ceiling. Vindicates E8's max > mean in real 7B weights once
saturation is controlled.

Default (easy) task behaviour is unchanged (hard defaults False). +1 hard-task
test (131 green). Excludes the 0.5B smoke bundle (a pipeline gate, not a
deliverable). Results in results/llm_{moe,directed}_hard_hpc/ (parquet gitignored).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Giorgio Gilestro 2026-07-05 19:13:26 +01:00
parent e433e48860
commit 39f6c9f4df
20 changed files with 495 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

26
src/llm/experiment.py
View file

@ -51,9 +51,11 @@ def run_merge_experiment(cfg: dict) -> pd.DataFrame:
lora = cfg.get("lora", {})
merges = cfg.get("merges", ["soup", "ties"])
seed = int(cfg["seed"])
hard = bool(cfg.get("hard", False))
suffix = "_hard" if hard else "" # hard specialists cache separately
adapters_root = Path(cfg.get("adapters_dir", "models/llm"))
test = sum([make_tasks(f, n_test, seed=1000 + i) for i, f in enumerate(fams)], [])
test = sum([make_tasks(f, n_test, seed=1000 + i, hard=hard) for i, f in enumerate(fams)], [])
rows: list[dict] = []
# base
@ -64,8 +66,8 @@ def run_merge_experiment(cfg: dict) -> pd.DataFrame:
# one specialist per family
dirs = []
for i, f in enumerate(fams):
d = str(adapters_root / f"spec_{f}")
train_specialist(base, f, d, n_train=n_train, epochs=epochs, seed=seed + i,
d = str(adapters_root / f"spec_{f}{suffix}")
train_specialist(base, f, d, n_train=n_train, epochs=epochs, seed=seed + i, hard=hard,
r=int(lora.get("r", 16)), alpha=int(lora.get("alpha", 32)))
dirs.append(d)
m, tok = load_model(base, adapter_dir=d)
@ -96,12 +98,14 @@ def _load_or_train_specialists(cfg: dict, base: str, fams: list[str], name: str,
epochs = int(cfg.get("epochs", 3))
lora = cfg.get("lora", {})
seed = int(cfg["seed"])
hard = bool(cfg.get("hard", False))
suffix = "_hard" if hard else "" # hard specialists cache separately
adapters_root = Path(cfg.get("adapters_dir", "models/llm"))
dirs: list[str] = []
for i, f in enumerate(fams):
d = str(adapters_root / f"spec_{f}")
d = str(adapters_root / f"spec_{f}{suffix}")
if not (Path(d) / "adapter_config.json").exists(): # reuse across llm_merge / llm_moe runs
train_specialist(base, f, d, n_train=n_train, epochs=epochs, seed=seed + i,
train_specialist(base, f, d, n_train=n_train, epochs=epochs, seed=seed + i, hard=hard,
r=int(lora.get("r", 16)), alpha=int(lora.get("alpha", 32)))
dirs.append(d)
m, tok = load_model(base, adapter_dir=d)
@ -113,7 +117,8 @@ def _load_or_train_specialists(cfg: dict, base: str, fams: list[str], name: str,
def test_of(cfg: dict, fams: list[str]) -> list:
"""The held-out mixed test set (one seed offset per family), shared by both LLM runners."""
n_test = int(cfg.get("n_test", 100))
return sum([make_tasks(f, n_test, seed=1000 + i) for i, f in enumerate(fams)], [])
hard = bool(cfg.get("hard", False))
return sum([make_tasks(f, n_test, seed=1000 + i, hard=hard) for i, f in enumerate(fams)], [])
def run_moe_experiment(cfg: dict) -> pd.DataFrame:
@ -160,7 +165,8 @@ def run_moe_experiment(cfg: dict) -> pd.DataFrame:
if "moe_oracle" in ops:
routes["moe_oracle"] = true_idx
if "moe_learned" in ops:
route_train = sum([make_tasks(f, n_route, seed=2000 + i) for i, f in enumerate(fams)], [])
route_train = sum([make_tasks(f, n_route, seed=2000 + i, hard=bool(cfg.get("hard", False)))
for i, f in enumerate(fams)], [])
tr_emb = embed_prompts(model, tok, [t.prompt for t in route_train])
te_emb = embed_prompts(model, tok, prompts)
tr_fam = np.array([t.family for t in route_train])
@ -207,7 +213,8 @@ def run_directed_experiment(cfg: dict) -> pd.DataFrame:
rows: list[dict] = []
test = test_of(cfg, fams)
val = sum([make_tasks(f, n_val, seed=3000 + i) for i, f in enumerate(fams)], [])
val = sum([make_tasks(f, n_val, seed=3000 + i, hard=bool(cfg.get("hard", False)))
for i, f in enumerate(fams)], [])
# base + specialists (parents), scored on test
m, tok = load_model(base)
@ -265,7 +272,8 @@ def run_and_save(config_path: str | Path) -> Path:
if kind not in _RUNNERS:
raise ValueError(f"unknown LLM experiment kind {kind!r} (expected one of {list(_RUNNERS)})")
df = _RUNNERS[kind](cfg)
extra = {"layer": "2", "tier": "llm", "base_model": cfg["base_model"]}
extra = {"layer": "2", "tier": "llm", "base_model": cfg["base_model"],
"hard": bool(cfg.get("hard", False))}
if kind == "llm_moe":
extra["operators"] = list(cfg.get("operators", []))
if kind == "llm_directed":

7
src/llm/specialise.py
View file

@ -29,9 +29,12 @@ def _encode(tok, task, device):
def train_specialist(base_name: str, family: str, out_dir: str, *, n_train: int = 600,
epochs: int = 3, lr: float = 2e-4, batch_size: int = 8, r: int = 16,
alpha: int = 32, seed: int = 0, device: str = "cuda") -> str:
alpha: int = 32, seed: int = 0, device: str = "cuda", hard: bool = False) -> str:
"""Fine-tune a LoRA specialist on ``family`` and save the adapter to ``out_dir``.
Args:
hard (bool): train on the harder task variant (must match the eval difficulty).
Returns the adapter directory path.
"""
import torch
@ -49,7 +52,7 @@ def train_specialist(base_name: str, family: str, out_dir: str, *, n_train: int
model = get_peft_model(model, lora)
model.train()
tasks = make_tasks(family, n_train, seed=seed)
tasks = make_tasks(family, n_train, seed=seed, hard=hard)
encoded = [_encode(tok, t, device) for t in tasks]
opt = torch.optim.AdamW([p for p in model.parameters() if p.requires_grad], lr=lr)
rng = np.random.default_rng(seed)

212
src/llm/tasks.py
View file

@ -10,6 +10,11 @@ decorrelated from the others (the precondition for recombination to have anythin
Each task is a natural-language instruction + input with a single canonical answer; the verifier
normalises the model's output and checks exact equality. This is the LLM analogue of the synthetic
sandbox's exact oracle — correctness is the "reality that can say no" (grounding / fitness).
A ``hard=True`` variant of each family (longer inputs + multi-step / cipher operations) exists so a
*capable* base (7B) is **not saturated** at 1.00 — the regime where recombination refinements
(routing, offspring selection) have headroom to matter. Same family labels and answer formats (int
list / integer / lowercase word), so the verifier and the whole experiment machinery are unchanged.
"""
from __future__ import annotations
@ -37,82 +42,173 @@ def _fmt_list(xs) -> str:
return "[" + ", ".join(str(int(x)) for x in xs) + "]"
def _list_task(rng) -> Task:
n = int(rng.integers(6, 9)) # longer lists (harder)
xs = rng.integers(0, 20, size=n).tolist()
op = rng.choice(["sort ascending", "sort descending", "reverse",
"second largest value", "sum of the two largest values",
"sorted then reversed", "unique values preserving order"])
if op == "sort ascending":
ans = _fmt_list(sorted(xs))
elif op == "sort descending":
ans = _fmt_list(sorted(xs, reverse=True))
elif op == "reverse":
ans = _fmt_list(list(reversed(xs)))
elif op == "second largest value":
ans = str(sorted(xs, reverse=True)[1])
elif op == "sum of the two largest values":
ans = str(sum(sorted(xs, reverse=True)[:2]))
elif op == "sorted then reversed":
ans = _fmt_list(sorted(xs, reverse=True))
else:
seen, out = set(), []
def _list_task(rng, hard: bool = False) -> Task:
if not hard:
n = int(rng.integers(6, 9)) # longer lists (harder)
xs = rng.integers(0, 20, size=n).tolist()
op = rng.choice(["sort ascending", "sort descending", "reverse",
"second largest value", "sum of the two largest values",
"sorted then reversed", "unique values preserving order"])
if op == "sort ascending":
ans = _fmt_list(sorted(xs))
elif op == "sort descending":
ans = _fmt_list(sorted(xs, reverse=True))
elif op == "reverse":
ans = _fmt_list(list(reversed(xs)))
elif op == "second largest value":
ans = str(sorted(xs, reverse=True)[1])
elif op == "sum of the two largest values":
ans = str(sum(sorted(xs, reverse=True)[:2]))
elif op == "sorted then reversed":
ans = _fmt_list(sorted(xs, reverse=True))
else:
seen, out = set(), []
for x in xs:
if x not in seen:
seen.add(x); out.append(x)
ans = _fmt_list(out)
want = "the resulting list" if ans.startswith("[") else "the single number"
return Task("lists", f"Given the list {_fmt_list(xs)}, compute its {op}. "
f"Output only {want} and nothing else.", ans)
# hard: longer lists + MULTI-STEP operations (a capable base is not saturated here)
n = int(rng.integers(10, 15))
xs = rng.integers(0, 30, size=n).tolist()
op = rng.choice(["sort ascending then the element at index 2 (0-based)",
"sum of the elements at even indices (0-based)",
"the third smallest value",
"the prefix sums (running totals) as a list",
"sort descending then the sum of the first three values",
"the product of the two smallest values",
"the values strictly greater than 15, preserving order"])
if op.startswith("sort ascending then the element"):
ans = str(sorted(xs)[2])
elif op.startswith("sum of the elements at even"):
ans = str(sum(xs[::2]))
elif op == "the third smallest value":
ans = str(sorted(xs)[2])
elif op.startswith("the prefix sums"):
acc, out = 0, []
for x in xs:
if x not in seen:
seen.add(x); out.append(x)
acc += x; out.append(acc)
ans = _fmt_list(out)
elif op.startswith("sort descending then the sum"):
ans = str(sum(sorted(xs, reverse=True)[:3]))
elif op.startswith("the product of the two smallest"):
s = sorted(xs); ans = str(s[0] * s[1])
else:
ans = _fmt_list([x for x in xs if x > 15])
want = "the resulting list" if ans.startswith("[") else "the single number"
return Task("lists", f"Given the list {_fmt_list(xs)}, compute its {op}. "
return Task("lists", f"Given the list {_fmt_list(xs)}, compute {op}. "
f"Output only {want} and nothing else.", ans)
def _string_task(rng) -> Task:
w = str(rng.choice(_WORDS))
op = rng.choice(["reverse", "uppercase", "count of the letter 'a'", "count of vowels",
"remove all vowels", "first three characters", "last three characters"])
if op == "reverse":
ans = w[::-1]
elif op == "uppercase":
ans = w.upper()
elif op == "count of the letter 'a'":
ans = str(w.count("a"))
elif op == "count of vowels":
ans = str(sum(c in "aeiou" for c in w))
elif op == "remove all vowels":
ans = "".join(c for c in w if c not in "aeiou")
elif op == "first three characters":
ans = w[:3]
def _caesar(w: str, k: int) -> str:
return "".join(chr((ord(c) - 97 + k) % 26 + 97) for c in w)
def _string_task(rng, hard: bool = False) -> Task:
if not hard:
w = str(rng.choice(_WORDS))
op = rng.choice(["reverse", "uppercase", "count of the letter 'a'", "count of vowels",
"remove all vowels", "first three characters", "last three characters"])
if op == "reverse":
ans = w[::-1]
elif op == "uppercase":
ans = w.upper()
elif op == "count of the letter 'a'":
ans = str(w.count("a"))
elif op == "count of vowels":
ans = str(sum(c in "aeiou" for c in w))
elif op == "remove all vowels":
ans = "".join(c for c in w if c not in "aeiou")
elif op == "first three characters":
ans = w[:3]
else:
ans = w[-3:]
return Task("strings", f"Given the word \"{w}\", compute its {op}. "
f"Output only the answer and nothing else.", ans)
# hard: longer words (two concatenated) + cipher / multi-step transforms
w = str(rng.choice(_WORDS)) + str(rng.choice(_WORDS))
op = rng.choice(["caesar", "sortletters", "devowel", "mostfreq", "distinct", "revdevowel"])
if op == "caesar":
ans = _caesar(w, 3)
instr = "shift each letter forward by 3 in the alphabet, wrapping z to a (a Caesar cipher)"
elif op == "sortletters":
ans = "".join(sorted(w)); instr = "its letters sorted in alphabetical order"
elif op == "devowel":
ans = "".join("x" if c in "aeiou" else c for c in w)
instr = "the word with every vowel replaced by the letter x"
elif op == "mostfreq":
ans = min(set(w), key=lambda c: (-w.count(c), c))
instr = "the letter that occurs most often (on a tie, the one earliest in the alphabet)"
elif op == "distinct":
ans = str(len(set(w))); instr = "the number of distinct letters"
else:
ans = w[-3:]
return Task("strings", f"Given the word \"{w}\", compute its {op}. "
ans = "".join(c for c in w[::-1] if c not in "aeiou")
instr = "the word reversed and then with all vowels removed"
return Task("strings", f"Given the word \"{w}\", compute {instr}. "
f"Output only the answer and nothing else.", ans)
def _arith_task(rng) -> Task:
kind = rng.choice(["mul", "sub", "mul2", "seq"]) # multiplication-heavy (harder)
if kind == "mul":
a, b = rng.integers(11, 40, size=2).tolist(); ans = str(a * b)
q = f"What is {a} * {b}?"
elif kind == "sub":
a, b = sorted(rng.integers(100, 1000, size=2).tolist(), reverse=True); ans = str(a - b)
q = f"What is {a} - {b}?"
elif kind == "mul2":
a = int(rng.integers(11, 100)); b = int(rng.integers(3, 10)); ans = str(a * b)
q = f"What is {a} * {b}?"
def _arith_task(rng, hard: bool = False) -> Task:
if not hard:
kind = rng.choice(["mul", "sub", "mul2", "seq"]) # multiplication-heavy (harder)
if kind == "mul":
a, b = rng.integers(11, 40, size=2).tolist(); ans = str(a * b)
q = f"What is {a} * {b}?"
elif kind == "sub":
a, b = sorted(rng.integers(100, 1000, size=2).tolist(), reverse=True); ans = str(a - b)
q = f"What is {a} - {b}?"
elif kind == "mul2":
a = int(rng.integers(11, 100)); b = int(rng.integers(3, 10)); ans = str(a * b)
q = f"What is {a} * {b}?"
else:
start = int(rng.integers(2, 12)); step = int(rng.integers(3, 12))
seq = [start + i * step for i in range(4)]; ans = str(seq[-1] + step)
q = f"What comes next in the sequence {', '.join(map(str, seq))}?"
return Task("arith", q + " Output only the number and nothing else.", ans)
# hard: multi-step / larger operands (a capable base is not saturated here)
kind = rng.choice(["madd", "mod", "diffsq", "bigmul", "rangesum", "gcd"])
if kind == "madd":
a, b = rng.integers(11, 40, size=2).tolist(); c = int(rng.integers(10, 100))
ans = str(a * b + c); q = f"What is {a} * {b} + {c}?"
elif kind == "mod":
a = int(rng.integers(100, 1000)); b = int(rng.integers(3, 20))
ans = str(a % b); q = f"What is {a} mod {b} (the remainder of {a} divided by {b})?"
elif kind == "diffsq":
a, b = sorted(rng.integers(10, 30, size=2).tolist(), reverse=True)
ans = str(a * a - b * b); q = f"What is {a}^2 - {b}^2?"
elif kind == "bigmul":
a = int(rng.integers(100, 1000)); b = int(rng.integers(2, 10))
ans = str(a * b); q = f"What is {a} * {b}?"
elif kind == "rangesum":
a = int(rng.integers(1, 20)); b = int(rng.integers(25, 55))
ans = str(sum(range(a, b + 1)))
q = f"What is the sum of all integers from {a} to {b} inclusive?"
else:
start = int(rng.integers(2, 12)); step = int(rng.integers(3, 12))
seq = [start + i * step for i in range(4)]; ans = str(seq[-1] + step)
q = f"What comes next in the sequence {', '.join(map(str, seq))}?"
import math
a, b = rng.integers(6, 60, size=2).tolist()
ans = str(math.gcd(a, b)); q = f"What is the greatest common divisor of {a} and {b}?"
return Task("arith", q + " Output only the number and nothing else.", ans)
_GEN = {"lists": _list_task, "strings": _string_task, "arith": _arith_task}
def make_tasks(family: str, n: int, seed: int) -> list[Task]:
"""Generate ``n`` unique-ish tasks for a family (deterministic in ``seed``)."""
def make_tasks(family: str, n: int, seed: int, hard: bool = False) -> list[Task]:
"""Generate ``n`` unique-ish tasks for a family (deterministic in ``seed``).
Args:
family (str): one of :data:`FAMILIES`.
n (int): number of tasks.
seed (int): RNG seed (deterministic output).
hard (bool): use the harder multi-step / cipher variant (same family label + answer format).
Returns:
list[Task]: the generated tasks.
"""
rng = np.random.default_rng(seed)
return [_GEN[family](rng) for _ in range(n)]
return [_GEN[family](rng, hard) for _ in range(n)]
def _normalise(s: str) -> str: