OSVauco/emma/emma_flynn.py

64 lines
2.1 KiB
Python

import json
import numpy as np
from pathlib import Path
from datetime import datetime
LOG = Path(__file__).parent / "data" / "flynn_log.jsonl"
class FlynnTracker:
"""
Analogt til Flynn-effekten: mål om Emma løser stadig mer komplekse
oppgaver med færre steg og høyere reward over tid.
Positiv Flynn-indeks = Emma vokser.
"""
def measure(self, text: str) -> float:
"""Grovt mål på oppgavekompleksitet [0-1]."""
factors = [
len(text.split()) / 100,
text.count("?") * 0.1,
len(set(text.split())) / 50,
int(any(c in text.lower() for c in ["kode", "api", "arkitektur", "strategi", "deploy"])) * 0.3,
]
return min(sum(factors), 1.0)
def record(self, complexity: float, steps: int, reward: float):
"""Append-only logging — aldri overskriv."""
LOG.parent.mkdir(parents=True, exist_ok=True)
entry = {
"ts": datetime.now().isoformat(),
"complexity": complexity,
"steps": steps,
"reward": reward,
"efficiency": reward / max(steps, 1),
}
with LOG.open("a") as f:
f.write(json.dumps(entry) + "\n")
def flynn_index(self) -> float:
"""Positiv stigning = Emma vokser (analogt til Flynn-kurven)."""
if not LOG.exists():
return 0.0
lines = LOG.read_text().strip().split("\n")[-50:]
entries = [json.loads(l) for l in lines if l]
if len(entries) < 2:
return 0.0
efficiencies = [e["efficiency"] for e in entries]
slope = float(np.polyfit(range(len(efficiencies)), efficiencies, 1)[0])
return slope
def summary(self) -> dict:
if not LOG.exists():
return {}
lines = LOG.read_text().strip().split("\n")
entries = [json.loads(l) for l in lines if l]
if not entries:
return {}
return {
"sessions": len(entries),
"avg_efficiency": sum(e["efficiency"] for e in entries) / len(entries),
"flynn_index": self.flynn_index(),
"last_complexity": entries[-1]["complexity"],
}