343 lines
17 KiB
Python
343 lines
17 KiB
Python
"""make train: Ameisen-Optimierung (ACO) der Pipeline-Parameter — anytime, multi-fidelity.
|
||
|
||
Prinzip: Pheromon-Gewichte je (Parameter, Stufe) steuern, welche Kandidaten („Ameisen")
|
||
als Nächstes getestet werden. Gute Kandidaten verstärken ihre Stufen, Verdunstung hält
|
||
die Suche offen — je länger der Trainer läuft, desto gezielter werden die Tests.
|
||
Jederzeit stoppbar; der Stand (beste_params.json/report.md) ist immer aktuell.
|
||
|
||
Fidelity-Kaskade pro Kandidat:
|
||
F0 Fake-E2E (train_f0.py, Sekunden, 0 Tokens): Invarianten + Struktur-Proxy — Filter.
|
||
F1 Frozen-Inventar (train_lauf.py --board2, ~5–8 min): misst Board-2/Guide-Parameter.
|
||
F2 Volllauf inkl. Soll-Abgleich: alle N Runden für den Besten + Inventar-Parameter.
|
||
|
||
CLI: python3 train.py [--stunden 8] [--trials 60] [--ameisen 3] [--seed 0]
|
||
[--sitzung NAME] [--f2-intervall 5]
|
||
python3 train.py --init (baut das Frozen-Inventar-Vorlage-Topic, einmalig)
|
||
Ergebnis: storage/train/<sitzung>/{trials.jsonl, pheromon.json, report.md, beste_params.json}
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import random
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from config import STORAGE_DIR
|
||
from train_params import PARAMS
|
||
|
||
VORLAGE_TOPIC = "train-vorlage"
|
||
BENCHMARK = "benchmarks/pruefstand"
|
||
# Score-Gewichte: Qualität + Auswahl dominieren (Entwicklungsphase), Kosten ziehen ab.
|
||
W_NOTE, W_AUSWAHL, W_ZEIT, W_TOKEN = 4.0, 4.0, 1.0, 1.0
|
||
RHO = 0.2 # Pheromon-Verdunstung je Runde
|
||
SPARSITY = 0.5 # Wahrscheinlichkeit, dass eine Ameise einen Parameter auf Default lässt
|
||
F0_CALL_FAKTOR = 1.5 # Struktur-Proxy: mehr als 1.5× Baseline-Calls → Kandidat verworfen
|
||
# Trainings-Fixa (Speed, kein Suchraum): Abschluss-QA ohne Judges, kurze Nachzügler-Gnade
|
||
TRAIN_FIXA = {"ABSCHLUSS_QA_LLM": 0, "CONSENSUS_GRACE": 60}
|
||
|
||
|
||
def stufen(name: str) -> list[float]:
|
||
p = PARAMS[name]
|
||
out, w = [], p["min"]
|
||
while w <= p["max"] + 1e-9:
|
||
out.append(round(w, 4))
|
||
w += p["step"]
|
||
return out
|
||
|
||
|
||
def score(m: dict, basis: dict) -> float:
|
||
"""Skalarer Vergleichswert. note 0–10; auswahl aus Soll-Abgleich (F2) oder MECE-Quoten;
|
||
Zeit/Tokens normiert auf die Baseline derselben Fidelity."""
|
||
if m.get("soll"):
|
||
auswahl = 10.0 * m["soll"]["f1"]
|
||
else:
|
||
q = m.get("quoten") or {}
|
||
qa_ = m.get("quoten_artefakte") or {}
|
||
auswahl = 10.0 * max(0.0, 1.0 - min(1.0, (
|
||
q.get("dubletten_verdacht", 0) + q.get("luecken", 0) + q.get("fremd", 0)
|
||
+ qa_.get("sub_dubletten_verdacht", 0) + qa_.get("verwaiste", 0))))
|
||
zeit = (m.get("dauer_min") or 0) / max(basis.get("dauer_min") or 1, 0.1)
|
||
tok = _tokens(m) / max(_tokens(basis), 1)
|
||
return round(W_NOTE * (m.get("note") or 0) + W_AUSWAHL * auswahl
|
||
- W_ZEIT * 10 * zeit - W_TOKEN * 10 * tok, 2)
|
||
|
||
|
||
def _tokens(m: dict) -> int:
|
||
t = m.get("tokens") or {}
|
||
return int(t.get("input") or 0) + int(t.get("output") or 0)
|
||
|
||
|
||
class AmeisenTrainer:
|
||
def __init__(self, sitzung: Path, *, max_trials: int, max_stunden: float, ameisen: int = 3,
|
||
seed: int = 0, f2_intervall: int = 5, runner=None, runner_f0=None):
|
||
self.dir = sitzung
|
||
self.dir.mkdir(parents=True, exist_ok=True)
|
||
self.rng = random.Random(seed)
|
||
self.ameisen = ameisen
|
||
self.f2_intervall = max(f2_intervall, 1)
|
||
self.max_trials = max_trials
|
||
self.deadline = time.monotonic() + max_stunden * 3600
|
||
self.gezahlt = 0
|
||
self.runner = runner or self._subprozess # (params, fidelity) -> metrics|None
|
||
self.runner_f0 = runner_f0 or self._subprozess_f0 # (params) -> {"ok","calls",…}|None
|
||
self.log: list[str] = []
|
||
self.basis: dict[str, dict] = {} # Fidelity → Baseline-Metriken
|
||
self.f0_basis: int | None = None
|
||
self.best_params: dict = {}
|
||
self.best_score: float | None = None
|
||
self.rauschen = 0.5
|
||
# Pheromon + Trial-Cache (Resume)
|
||
self.pheromon: dict[str, dict[str, float]] = {
|
||
n: {str(s): 1.0 for s in stufen(n)} for n in PARAMS}
|
||
ph = self.dir / "pheromon.json"
|
||
if ph.exists():
|
||
gespeichert = json.loads(ph.read_text(encoding="utf-8"))
|
||
for n, taus in gespeichert.get("pheromon", {}).items():
|
||
if n in self.pheromon:
|
||
self.pheromon[n].update({k: float(v) for k, v in taus.items()})
|
||
self.best_params = gespeichert.get("best_params", {})
|
||
self.best_score = gespeichert.get("best_score")
|
||
self.cache_pfad = self.dir / "trials.jsonl"
|
||
self.cache: dict[str, dict] = {}
|
||
if self.cache_pfad.exists():
|
||
for line in self.cache_pfad.read_text(encoding="utf-8").splitlines():
|
||
e = json.loads(line)
|
||
self.cache[e["key"]] = e["metrics"]
|
||
|
||
# ── Kandidaten ──────────────────────────────────────────────────────────────────
|
||
def kandidat(self, fidelity: str) -> dict:
|
||
"""Eine Ameise: je Parameter der Fidelity mit SPARSITY auf Default, sonst
|
||
Pheromon-gewichtete Stufe. Sparsame Kandidaten → saubere Attribution."""
|
||
params = {}
|
||
for name, p in PARAMS.items():
|
||
if fidelity == "board2" and p["fidelity"] != "board2":
|
||
continue
|
||
if self.rng.random() < SPARSITY:
|
||
continue
|
||
st = stufen(name)
|
||
taus = [self.pheromon[name][str(s)] for s in st]
|
||
wert = self.rng.choices(st, weights=taus)[0]
|
||
if wert != p["default"]:
|
||
params[name] = wert
|
||
return params
|
||
|
||
# ── Trial-Ausführung ────────────────────────────────────────────────────────────
|
||
def _key(self, params: dict, fidelity: str, tag: str = "") -> str:
|
||
raw = json.dumps({"p": params, "f": fidelity, "tag": tag}, sort_keys=True)
|
||
return hashlib.md5(raw.encode()).hexdigest()[:12]
|
||
|
||
async def trial(self, params: dict, fidelity: str, tag: str = "") -> dict | None:
|
||
key = self._key(params, fidelity, tag)
|
||
if key in self.cache:
|
||
return self.cache[key]
|
||
if self.gezahlt >= self.max_trials or time.monotonic() > self.deadline:
|
||
return None
|
||
self.gezahlt += 1
|
||
metrics = await self.runner(params, fidelity, "0")
|
||
if metrics is not None:
|
||
with open(self.cache_pfad, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps({"key": key, "params": params, "fidelity": fidelity,
|
||
"tag": tag, "metrics": metrics}, ensure_ascii=False) + "\n")
|
||
self.cache[key] = metrics
|
||
return metrics
|
||
|
||
async def _subprozess(self, params: dict, fidelity: str, topic_suffix: str = "0") -> dict | None:
|
||
out = self.dir / f"metrics-{self._key(params, fidelity)}{topic_suffix}.json"
|
||
topic = f"train-t{topic_suffix}"
|
||
args = ([topic, VORLAGE_TOPIC, str(out), "--board2"] if fidelity == "board2"
|
||
else [topic, BENCHMARK, str(out)])
|
||
env = {**os.environ, "CREATOR_PARAMS": json.dumps({**TRAIN_FIXA, **params})}
|
||
proc = await asyncio.create_subprocess_exec(sys.executable, "train_lauf.py", *args, env=env)
|
||
rc = await proc.wait()
|
||
if rc != 0 or not out.exists():
|
||
self._log(f"Trial fehlgeschlagen (rc={rc}, {fidelity}, params={params})")
|
||
return None
|
||
return json.loads(out.read_text(encoding="utf-8"))
|
||
|
||
async def _subprozess_f0(self, params: dict) -> dict | None:
|
||
out = self.dir / f"f0-{self._key(params, 'f0')}.json"
|
||
env = {**os.environ, "CREATOR_PARAMS": json.dumps(params)}
|
||
proc = await asyncio.create_subprocess_exec(sys.executable, "train_f0.py", str(out), env=env)
|
||
rc = await proc.wait()
|
||
return json.loads(out.read_text(encoding="utf-8")) if rc == 0 and out.exists() else None
|
||
|
||
def _log(self, msg: str) -> None:
|
||
line = f"{datetime.now(timezone.utc).isoformat()[11:19]} {msg}"
|
||
print(line, flush=True)
|
||
self.log.append(line)
|
||
|
||
# ── Pheromon ────────────────────────────────────────────────────────────────────
|
||
def verstaerke(self, params: dict, delta: float) -> None:
|
||
for name, wert in params.items():
|
||
taus = self.pheromon[name]
|
||
key = str(wert)
|
||
if key in taus:
|
||
taus[key] += delta
|
||
|
||
def verdunste(self) -> None:
|
||
for taus in self.pheromon.values():
|
||
for k in taus:
|
||
taus[k] = max(0.1, (1 - RHO) * taus[k] + RHO * 1.0) # Drift zurück zu uniform
|
||
|
||
# ── Hauptschleife ───────────────────────────────────────────────────────────────
|
||
async def run(self) -> dict:
|
||
# Baseline F1 ×2 → Score-Basis + Rausch-Schwelle; F0-Basis für den Struktur-Proxy
|
||
f0 = await self.runner_f0({})
|
||
self.f0_basis = (f0 or {}).get("calls")
|
||
b1 = await self.trial({}, "board2", tag="baseline-1")
|
||
b2 = await self.trial({}, "board2", tag="baseline-2")
|
||
if not b1 or not b2:
|
||
self._log("Baseline unvollständig — Abbruch.")
|
||
return self.best_params
|
||
self.basis["board2"] = b1
|
||
s1, s2 = score(b1, b1), score(b2, b1)
|
||
self.rauschen = max(abs(s1 - s2), 0.5)
|
||
if self.best_score is None:
|
||
self.best_score = max(s1, s2)
|
||
self._log(f"Baseline {s1}/{s2}, Rauschen {self.rauschen}, F0-Basis {self.f0_basis} Calls")
|
||
|
||
runde, stagnation, gezahlt_vorher = 0, 0, self.gezahlt
|
||
while (self.gezahlt < self.max_trials and time.monotonic() < self.deadline
|
||
and stagnation < 20): # konvergiert: nur noch Cache-Treffer → fertig
|
||
if runde > 0:
|
||
stagnation = stagnation + 1 if self.gezahlt == gezahlt_vorher else 0
|
||
gezahlt_vorher = self.gezahlt
|
||
runde += 1
|
||
fidelity = "voll" if runde % self.f2_intervall == 0 else "board2"
|
||
if fidelity == "voll" and "voll" not in self.basis:
|
||
base = await self.trial({}, "voll", tag="baseline-voll")
|
||
if base is None:
|
||
break
|
||
self.basis["voll"] = base
|
||
kandidaten = []
|
||
for _ in range(self.ameisen * 3): # ziehen bis K einzigartige nicht-leere da sind
|
||
k = self.kandidat(fidelity)
|
||
if k and k not in kandidaten and k != self.best_params:
|
||
kandidaten.append(k)
|
||
if len(kandidaten) >= self.ameisen:
|
||
break
|
||
if not kandidaten:
|
||
continue
|
||
# F0-Filter: Invarianten + Struktur-Proxy, parallel, kostenlos
|
||
f0s = await asyncio.gather(*[self.runner_f0(k) for k in kandidaten])
|
||
ueberlebende = []
|
||
for k, f in zip(kandidaten, f0s):
|
||
if f is None or not f.get("ok") or f.get("invarianten_fehler"):
|
||
self._log(f"F0 verwirft {k} (Invarianten)")
|
||
elif self.f0_basis and f.get("calls", 0) > self.f0_basis * F0_CALL_FAKTOR:
|
||
self._log(f"F0 verwirft {k} (Calls {f['calls']} > {self.f0_basis}×{F0_CALL_FAKTOR})")
|
||
else:
|
||
ueberlebende.append(k)
|
||
if not ueberlebende:
|
||
self.verdunste()
|
||
continue
|
||
# F1/F2 parallel (eigene Topic-Namen)
|
||
ergebnisse = await asyncio.gather(*[
|
||
self._bewertet(k, fidelity, str(i + 1)) for i, k in enumerate(ueberlebende)])
|
||
bewertet = [(k, m, score(m, self.basis[fidelity]))
|
||
for k, m in ergebnisse if m is not None]
|
||
if not bewertet:
|
||
continue
|
||
bewertet.sort(key=lambda x: -x[2])
|
||
self.verdunste()
|
||
top_k, _top_m, top_s = bewertet[0]
|
||
self._log(f"Runde {runde} ({fidelity}): top {top_s} {top_k} "
|
||
f"(best {self.best_score})")
|
||
if top_s > (self.best_score or 0):
|
||
self.verstaerke(top_k, delta=1.0)
|
||
if top_s > (self.best_score or 0) + self.rauschen:
|
||
m2 = await self.trial(top_k, fidelity, tag="bestaetigung")
|
||
if m2 is not None and score(m2, self.basis[fidelity]) > self.best_score + self.rauschen:
|
||
self.best_params = top_k
|
||
self.best_score = min(top_s, score(m2, self.basis[fidelity]))
|
||
self._log(f"NEUER BESTER {self.best_params} → {self.best_score}")
|
||
else:
|
||
self._log(f"{top_k}: nicht bestätigt")
|
||
self.verstaerke(self.best_params, delta=0.5) # Elite hält die Spur warm
|
||
self._speichern()
|
||
self._speichern()
|
||
return self.best_params
|
||
|
||
async def _bewertet(self, params: dict, fidelity: str, suffix: str):
|
||
if self.gezahlt >= self.max_trials or time.monotonic() > self.deadline:
|
||
return params, None
|
||
key = self._key(params, fidelity)
|
||
if key in self.cache:
|
||
return params, self.cache[key]
|
||
self.gezahlt += 1
|
||
m = await self.runner(params, fidelity, suffix)
|
||
if m is not None:
|
||
with open(self.cache_pfad, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps({"key": key, "params": params, "fidelity": fidelity,
|
||
"tag": "", "metrics": m}, ensure_ascii=False) + "\n")
|
||
self.cache[key] = m
|
||
return params, m
|
||
|
||
def _speichern(self) -> None:
|
||
from fsutil import atomic_write_json, atomic_write_text
|
||
atomic_write_json(self.dir / "pheromon.json",
|
||
{"pheromon": self.pheromon, "best_params": self.best_params,
|
||
"best_score": self.best_score}, indent=1)
|
||
atomic_write_json(self.dir / "beste_params.json", self.best_params, indent=1)
|
||
staerkste = sorted(((n, max(t.items(), key=lambda x: x[1]))
|
||
for n, t in self.pheromon.items()),
|
||
key=lambda x: -x[1][1])[:10]
|
||
report = ["# Trainings-Report (Ameisen)", "",
|
||
f"Bezahlte Läufe: {self.gezahlt}/{self.max_trials}",
|
||
f"Bester Score: {self.best_score} (Rauschband {self.rauschen})",
|
||
f"Beste Parameter: `{json.dumps(self.best_params, ensure_ascii=False)}`",
|
||
"", "Stärkste Pheromon-Spuren:",
|
||
*[f"- {n}={s} (τ={t:.1f})" for n, (s, t) in staerkste],
|
||
"", "Nutzung: `CREATOR_PARAMS=$(cat beste_params.json)` —",
|
||
"Übernahme nach config.py bleibt eine manuelle Entscheidung.", "", "## Log", ""]
|
||
report += [f"- {l}" for l in self.log[-200:]]
|
||
atomic_write_text(self.dir / "report.md", "\n".join(report))
|
||
|
||
|
||
async def init_vorlage() -> None:
|
||
"""Einmalig: Prüfstand-Volllauf mit Defaults, Ergebnis bleibt als Frozen-Inventar-Vorlage
|
||
liegen (Topic train-vorlage). Nach Korpus-/Prompt-Änderungen neu ausführen."""
|
||
import agents
|
||
import database
|
||
from blocks import generate_blocks
|
||
from fsutil import atomic_write_json as awj
|
||
from paths import source_path
|
||
await database.init_db()
|
||
agents.on_event = database.add_event
|
||
await database.create_topic(VORLAGE_TOPIC)
|
||
qp = source_path(VORLAGE_TOPIC)
|
||
qp.parent.mkdir(parents=True, exist_ok=True)
|
||
awj(qp, {"type": "uni", "location": BENCHMARK, "spec": ""})
|
||
await generate_blocks(VORLAGE_TOPIC, provider="minimax", research=True, qa_force=True)
|
||
await database.close_db()
|
||
print(f"Vorlage {VORLAGE_TOPIC} steht — Training kann starten (make train).")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--init", action="store_true", help="Frozen-Inventar-Vorlage bauen")
|
||
ap.add_argument("--trials", type=int, default=60)
|
||
ap.add_argument("--stunden", type=float, default=8.0)
|
||
ap.add_argument("--ameisen", type=int, default=3)
|
||
ap.add_argument("--seed", type=int, default=0)
|
||
ap.add_argument("--f2-intervall", type=int, default=5)
|
||
ap.add_argument("--sitzung", default="aco") # fester Default: Resume über Sitzungen hinweg
|
||
args = ap.parse_args()
|
||
if args.init:
|
||
asyncio.run(init_vorlage())
|
||
return
|
||
trainer = AmeisenTrainer(STORAGE_DIR / "train" / args.sitzung,
|
||
max_trials=args.trials, max_stunden=args.stunden,
|
||
ameisen=args.ameisen, seed=args.seed,
|
||
f2_intervall=args.f2_intervall)
|
||
asyncio.run(trainer.run())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|