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

llm_society: the composed society at LLM scale (C3) — loop, tests, smoke green

Browse source 
E11 re-instantiated in a population of LoRA agents, closing the paper's stated
gap before submission (GG: a weeks-scale experiment a reviewer would demand).
One grounding knob in the evaluation channel (g*verifier + (1-g)*conformity,
exactly E11); inheritance is identical in all arms and deliberately ungrounded
(children distilled from their source's own answers - self-consumption made
literal). Directed sex = complementary pairing + Dirichlet offspring screened
on the arm's own signal (the verifier never enters the no_grounding loop);
QD selection on verifier-free behavioural distance; terminal-degeneration
fallback copies the parent instead of crashing a sweep. Pure operators
unit-tested (155 green); smoke run end-to-end on the local A4000 already
shows the self-consumption signature (conformity up, diversity down in one
generation). Design, falsifiers, cost table: tasks/workorder-llm-society.md.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BkRLcc18rwT2Lysu6PbG7v
This commit is contained in:
Giorgio Gilestro 2026-09-07 11:55:19 +01:00
parent d75c58474a
commit 9b0ca32f51
8 changed files with 531 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

318
src/llm/society.py Normal file
View file

@ -0,0 +1,318 @@
"""The composed society at LLM scale (C3) — E11 re-instantiated in a population of LoRA agents.
A population of `N` agents (LoRA adapters on a shared frozen base) evolves for `G` non-overlapping
generations under the four operators the paper composes: grounded evaluation, directed recombination
(sex), diversity-preserving selection, and retraining (mutation). The single grounding knob acts in
the *evaluation* channel, exactly as in E11: selection scores each agent by
``g * verifier_fitness + (1 - g) * conformity``, where conformity is agreement with the population's
own modal answer. The inheritance channel is identical in every arm and deliberately ungrounded —
each child is a fresh LoRA distilled from its source model's *own answers* (self-consumption made
literal), so knowledge survives only through the data channel.
Arms (four-arm ablation, mirroring E11): ``full`` / ``no_grounding`` (g=0; the verifier never enters
that arm's loop — offspring screening also falls back to conformity) / ``no_sex`` (children are
redistilled copies of selected parents) / ``no_diversity`` (plain top-P selection).
Pure, testable pieces live at module top (consensus, conformity, behavioural distance,
quality-diversity selection, complementary pairing); the GPU loop is
:func:`run_society_experiment`. ``python -m llm.experiment configs/llm/society_smoke.yaml``.
"""
from __future__ import annotations
import shutil
from pathlib import Path
import numpy as np
import pandas as pd
from .tasks import Task, make_tasks, verify, _normalise
# ---------------------------------------------------------------------------- pure operators
def consensus_answers(outputs: list[list[str]]) -> list[str]:
"""The population's modal (normalised) answer per prompt.
Args:
outputs (list[list[str]]): ``outputs[i][p]`` = agent i's raw answer to prompt p.
Returns:
list[str]: per-prompt modal normalised answer (ties broken lexicographically —
deterministic).
"""
n_prompts = len(outputs[0])
cons = []
for p in range(n_prompts):
votes: dict[str, int] = {}
for out in outputs:
key = _normalise(out[p])
votes[key] = votes.get(key, 0) + 1
# Reason: max count, then lexicographically smallest key, so consensus is deterministic.
top = max(votes.values())
cons.append(min(k for k, v in votes.items() if v == top))
return cons
def conformity_scores(outputs: list[list[str]], consensus: list[str]) -> np.ndarray:
"""Each agent's agreement rate with the population consensus (the E11 conformity signal)."""
return np.array([np.mean([_normalise(o) == c for o, c in zip(out, consensus)])
for out in outputs], dtype=float)
def behavioural_distance(outputs: list[list[str]]) -> np.ndarray:
"""Pairwise disagreement rate between agents' normalised answers (verifier-free diversity)."""
n = len(outputs)
norm = [[_normalise(o) for o in out] for out in outputs]
d = np.zeros((n, n))
for i in range(n):
for j in range(i + 1, n):
d[i, j] = d[j, i] = float(np.mean([a != b for a, b in zip(norm[i], norm[j])]))
return d
def select_parents(scores: np.ndarray, dist: np.ndarray, k: int, *, diversity: bool,
lam: float = 0.3) -> list[int]:
"""Select ``k`` parents by score, optionally with quality-diversity preservation.
Diversity ON: greedy — take the top-scoring agent, then repeatedly add the agent maximising
``score + lam * (mean behavioural distance to those already selected)``. Diversity OFF: plain
top-k by score.
Args:
scores (np.ndarray): per-agent selection score ``g*fitness + (1-g)*conformity``.
dist (np.ndarray): pairwise behavioural-distance matrix.
k (int): number of parents.
diversity (bool): quality-diversity (True) or greedy (False).
lam (float): diversity weight.
Returns:
list[int]: selected agent indices (deterministic).
"""
if not diversity:
return list(np.argsort(-scores, kind="stable")[:k])
chosen = [int(np.argmax(scores))]
while len(chosen) < k:
best_i, best_v = -1, -np.inf
for i in range(len(scores)):
if i in chosen:
continue
v = scores[i] + lam * float(np.mean([dist[i, j] for j in chosen]))
if v > best_v:
best_i, best_v = i, v
chosen.append(best_i)
return chosen
def complementary_pairs(parents: list[int], dist: np.ndarray, n_children: int) -> list[tuple[int, int]]:
"""Mating plan: pairs of parents in descending behavioural distance, cycled to fill the slots.
Directed mate choice (E10): the most complementary (most-disagreeing) pairs breed first; the pair
list cycles until ``n_children`` slots are filled.
"""
pairs = sorted(((a, b) for i, a in enumerate(parents) for b in parents[i + 1:]),
key=lambda ab: -dist[ab[0], ab[1]])
if not pairs: # single parent: self-pair (degenerate)
pairs = [(parents[0], parents[0])]
return [pairs[i % len(pairs)] for i in range(n_children)]
def arm_settings(arm: str, g: float) -> dict:
"""Resolve an arm name to its operator switches (the four-arm ablation)."""
if arm == "full":
return {"g": g, "sex": True, "diversity": True}
if arm == "no_grounding":
return {"g": 0.0, "sex": True, "diversity": True}
if arm == "no_sex":
return {"g": g, "sex": False, "diversity": True}
if arm == "no_diversity":
return {"g": g, "sex": True, "diversity": False}
raise ValueError(f"unknown arm {arm!r} (full|no_grounding|no_sex|no_diversity)")
# ---------------------------------------------------------------------------- the GPU loop
def _fitness(outs: list[str], tasks: list[Task], fams: list[str]) -> dict:
"""Overall / per-family / worst-family verifier accuracy from precomputed outputs."""
corr = np.array([verify(o, t) for o, t in zip(outs, tasks)])
fam = np.array([t.family for t in tasks])
acc = {"overall": float(corr.mean())}
acc.update({f: float(corr[fam == f].mean()) for f in fams})
acc["worst_family"] = min(acc[f] for f in fams)
return acc
def _answer_of(raw: str) -> str:
"""Trim a raw completion to a single short answer line for the inheritance data."""
return raw.strip().split("\n")[0][:64].strip()
def run_society_experiment(cfg: dict) -> pd.DataFrame:
"""Run the society loop for every configured arm; return tidy long-form rows.
Config keys (with defaults): ``agents`` (6), ``generations`` (8), ``arms`` (all four),
``g`` (0.5), ``lam`` (0.3), ``n_test``/``n_val``/``n_conf``/``n_inherit`` pool sizes,
``n_candidates`` (6) offspring screened per pair, ``epochs`` (2) child SFT epochs,
``spec_train``/``spec_epochs`` gen-0 specialist budget, ``hard`` (False),
``keep_all_adapters`` (False).
"""
import torch
from .evaluate import generate, load_model
from .merge import load_specialists
from .specialise import train_lora_on_tasks, train_specialist
name = cfg["experiment"]
base = cfg["base_model"]
fams = list(cfg.get("families", ["lists", "strings", "arith"]))
N = int(cfg.get("agents", 6))
G = int(cfg.get("generations", 8))
arms = list(cfg.get("arms", ["full", "no_grounding", "no_sex", "no_diversity"]))
g_val = float(cfg.get("g", 0.5))
lam = float(cfg.get("lam", 0.3))
n_test = int(cfg.get("n_test", 40))
n_val = int(cfg.get("n_val", 30))
n_conf = int(cfg.get("n_conf", 60))
n_inherit = int(cfg.get("n_inherit", 240))
n_cand = int(cfg.get("n_candidates", 6))
epochs = int(cfg.get("epochs", 2))
spec_train = int(cfg.get("spec_train", 300))
spec_epochs = int(cfg.get("spec_epochs", 2))
hard = bool(cfg.get("hard", False))
keep_all = bool(cfg.get("keep_all_adapters", False))
seed = int(cfg["seed"])
lora = cfg.get("lora", {})
r, alpha = int(lora.get("r", 16)), int(lora.get("alpha", 32))
root = Path(cfg.get("adapters_dir", "models/llm")) / "society" / f"{name}_s{seed}"
# fixed pools: test (reporting only), val (grounded selection signal + offspring screening)
test = sum([make_tasks(f, n_test, seed=1000 + i, hard=hard) for i, f in enumerate(fams)], [])
val = sum([make_tasks(f, n_val, seed=3000 + i, hard=hard) for i, f in enumerate(fams)], [])
rng = np.random.default_rng(seed)
# gen-0 founders: light per-family specialists, shared across arms (same starting population)
founders = []
for i in range(N):
fam = fams[i % len(fams)]
d = root / "founders" / f"agent{i}_{fam}"
if not (d / "adapter_config.json").exists():
train_specialist(base, fam, str(d), n_train=spec_train, epochs=spec_epochs,
seed=seed * 100 + i, hard=hard, r=r, alpha=alpha)
founders.append(str(d))
rows: list[dict] = []
for arm in arms:
s = arm_settings(arm, g_val)
agents = list(founders)
for t in range(G):
conf_pool = sum([make_tasks(f, n_conf // len(fams), seed=seed * 7919 + t * 13 + i,
hard=hard) for i, f in enumerate(fams)], [])
conf_prompts = [x.prompt for x in conf_pool]
# ---- produce & score: one base, all agents attached as adapters
model, tok = load_specialists(base, agents)
fit_test, fit_val, conf_outs = [], [], []
for i in range(N):
model.set_adapter(f"a{i}")
fit_test.append(_fitness(generate(model, tok, [x.prompt for x in test]), test, fams))
fit_val.append(_fitness(generate(model, tok, [x.prompt for x in val]), val, fams))
conf_outs.append(generate(model, tok, conf_prompts))
consensus = consensus_answers(conf_outs)
conf = conformity_scores(conf_outs, consensus)
dist = behavioural_distance(conf_outs)
cons_acc = float(np.mean([verify(c, x) for c, x in zip(consensus, conf_pool)]))
fitness = np.array([fv["overall"] for fv in fit_val])
scores = s["g"] * fitness + (1.0 - s["g"]) * conf
parents = select_parents(scores, dist, max(2, N // 2), diversity=s["diversity"], lam=lam)
# ---- rows (reporting uses the verifier in every arm; the loop does not)
for i in range(N):
base_row = {"experiment": name, "arm": arm, "seed": seed, "generation": t,
"agent": i, "selected": i in parents,
"conformity": float(conf[i]), "score": float(scores[i])}
for k, v in fit_test[i].items():
rows.append({**base_row, "metric": f"test_{k}", "value": v})
rows.append({**base_row, "metric": "val_overall", "value": float(fitness[i])})
rows.append({"experiment": name, "arm": arm, "seed": seed, "generation": t,
"agent": -1, "selected": False, "conformity": float("nan"),
"score": float("nan"), "metric": "consensus_acc", "value": cons_acc})
rows.append({"experiment": name, "arm": arm, "seed": seed, "generation": t,
"agent": -1, "selected": False, "conformity": float("nan"),
"score": float("nan"), "metric": "diversity_behav",
"value": float(dist[np.triu_indices(N, 1)].mean())})
if t == G - 1:
del model
torch.cuda.empty_cache()
break
# ---- breed: pick each child's source (merged offspring, or a copied parent)
inherit_pool = sum([make_tasks(f, n_inherit // len(fams),
seed=seed * 104729 + t * 17 + i, hard=hard)
for i, f in enumerate(fams)], [])
child_sources: list[dict] = [] # per child: parent names + weights
if s["sex"]:
from .directed import sample_merge_weights
pairs = complementary_pairs(parents, dist, N)
for c, (pa, pb) in enumerate(pairs):
w = sample_merge_weights(2, n_cand, rng)
best_i, best_v = 0, -np.inf
for ci in range(n_cand):
cname = f"g{t}c{c}k{ci}"
model.add_weighted_adapter([f"a{pa}", f"a{pb}"], w[ci].tolist(), cname,
combination_type="linear")
model.set_adapter(cname)
if s["g"] > 0: # grounded screening: verifier on val
v = _fitness(generate(model, tok, [x.prompt for x in val]),
val, fams)["overall"]
else: # ungrounded screening: conformity only
outs_c = generate(model, tok, conf_prompts)
v = float(np.mean([_normalise(o) == cc
for o, cc in zip(outs_c, consensus)]))
if v > best_v:
best_i, best_v = ci, v
model.set_adapter(f"a{pa}") # never delete the active adapter
model.delete_adapter(cname)
cname = f"g{t}c{c}win"
model.add_weighted_adapter([f"a{pa}", f"a{pb}"], w[best_i].tolist(), cname,
combination_type="linear")
model.set_adapter(cname)
answers = generate(model, tok, [x.prompt for x in inherit_pool])
model.set_adapter(f"a{pa}")
model.delete_adapter(cname)
child_sources.append({"parents": (pa, pb), "answers": answers})
else:
for c in range(N):
p = parents[c % len(parents)]
model.set_adapter(f"a{p}")
answers = generate(model, tok, [x.prompt for x in inherit_pool])
child_sources.append({"parents": (p, p), "answers": answers})
del model
torch.cuda.empty_cache()
# ---- reproduce: fresh LoRA per child, trained on its source's own answers
new_agents = []
for c, src in enumerate(child_sources):
data = [Task(x.family, x.prompt, _answer_of(a))
for x, a in zip(inherit_pool, src["answers"]) if _answer_of(a)]
d = root / arm / f"gen{t + 1}" / f"agent{c}"
if len(data) >= 8:
train_lora_on_tasks(base, data, str(d), epochs=epochs,
seed=seed * 31 + t * N + c, r=r, alpha=alpha)
else: # terminal degeneration: copy the source
# Reason: a fully-degenerate source emits no usable answers; crashing would kill
# a long sweep (cf. the neural terminal-collapse sentinel) — inherit unchanged.
shutil.copytree(agents[src["parents"][0]], d, dirs_exist_ok=True)
new_agents.append(str(d))
rows.append({"experiment": name, "arm": arm, "seed": seed, "generation": t + 1,
"agent": c, "selected": False, "conformity": float("nan"),
"score": float("nan"), "metric": "parents",
"value": float(src["parents"][0] * 100 + src["parents"][1])})
if not keep_all and t > 0: # disk hygiene: drop generation t
for d in agents: # (founders at t=0 are kept — shared)
shutil.rmtree(d, ignore_errors=True)
agents = new_agents
return pd.DataFrame(rows)