157 lines
4.9 KiB
Python
157 lines
4.9 KiB
Python
"""
|
|
Lab 7.7: Build an Autoresearch Loop
|
|
|
|
Objective: Create an agent that runs experiments, measures results, logs them,
|
|
and decides whether to keep or discard each change.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
from dataclasses import dataclass, asdict
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
# TODO 1: Define the experiment log format
|
|
@dataclass
|
|
class ExperimentRun:
|
|
run: int
|
|
status: str # "keep" or "discard"
|
|
metric_name: str
|
|
metric_value: float
|
|
metric_unit: str
|
|
metric_direction: str # "minimize" or "maximize"
|
|
description: str
|
|
code_hash: str
|
|
delta_pct: float | None = None
|
|
timestamp: str = ""
|
|
|
|
def __post_init__(self):
|
|
if not self.timestamp:
|
|
self.timestamp = datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
# TODO 2: Implement integrity guards
|
|
def hash_code(file_path: str) -> str:
|
|
"""Return SHA-256 hash of the agent code file."""
|
|
pass # TODO
|
|
|
|
|
|
def detect_grind(previous_hash: str, current_hash: str, threshold: int = 3) -> bool:
|
|
"""
|
|
Detect grinding: same code re-run N times in a row.
|
|
|
|
Returns True if this is a grind run (code hash matches previous run).
|
|
"""
|
|
pass # TODO
|
|
|
|
|
|
def compare_against_median(scores: list[float], new_score: float) -> bool:
|
|
"""
|
|
Compare against median, not best.
|
|
Returns True if improvement over median (not just best)."""
|
|
pass # TODO
|
|
|
|
|
|
# TODO 3: Implement the experiment loop
|
|
class AutoresearchLoop:
|
|
"""Run experiments, measure results, log, decide keep/discard."""
|
|
|
|
def __init__(self, log_path: str = "experiments.jsonl"):
|
|
self.log_path = log_path
|
|
self.run_count = 0
|
|
self.last_code_hash = None
|
|
|
|
def setup_baseline(self) -> float:
|
|
"""Run the initial version, measure baseline metric."""
|
|
pass # TODO
|
|
|
|
def propose_change(self) -> str:
|
|
"""Ask the agent to propose a code change. Return description."""
|
|
pass # TODO
|
|
|
|
def apply_change(self, change_desc: str):
|
|
"""Apply the proposed change to the codebase."""
|
|
pass # TODO
|
|
|
|
def measure(self) -> float:
|
|
"""Run the experiment and return the metric value."""
|
|
pass # TODO
|
|
|
|
def log_experiment(self, run: ExperimentRun):
|
|
"""Log experiment to JSONL file."""
|
|
with open(self.log_path, "a") as f:
|
|
f.write(json.dumps(asdict(run)) + "\n")
|
|
|
|
def should_keep(self, baseline: float, current: float, direction: str) -> bool:
|
|
"""Decide if the change is an improvement."""
|
|
pass # TODO
|
|
|
|
def run(self, max_iterations: int = 10):
|
|
"""Run the autoresearch loop."""
|
|
print("Starting autoresearch loop...")
|
|
|
|
# Step 1: Baseline
|
|
baseline = self.setup_baseline()
|
|
print(f"Baseline: {baseline}")
|
|
self.log_experiment(ExperimentRun(
|
|
run=0, status="keep",
|
|
metric_name="performance", metric_value=baseline,
|
|
metric_unit="ms", metric_direction="minimize",
|
|
description="Baseline measurement",
|
|
code_hash=hash_code("agent.py")
|
|
))
|
|
|
|
for i in range(max_iterations):
|
|
print(f"\n--- Run {i+1} ---")
|
|
|
|
# Step 2: Propose change
|
|
change = self.propose_change()
|
|
print(f"Proposed: {change}")
|
|
|
|
# Step 3: Integrity check — detect grind
|
|
current_hash = hash_code("agent.py")
|
|
if detect_grind(self.last_code_hash, current_hash):
|
|
print("⚠️ Grind detected! Skipping run.")
|
|
continue
|
|
self.last_code_hash = current_hash
|
|
|
|
# Step 4: Apply and measure
|
|
self.apply_change(change)
|
|
new_value = self.measure()
|
|
|
|
# Step 5: Decide keep/discard
|
|
keep = self.should_keep(baseline, new_value, "minimize")
|
|
delta = ((new_value - baseline) / baseline) * 100 if baseline else 0
|
|
|
|
# Step 6: Log
|
|
run = ExperimentRun(
|
|
run=i+1,
|
|
status="keep" if keep else "discard",
|
|
metric_name="performance",
|
|
metric_value=new_value,
|
|
metric_unit="ms",
|
|
metric_direction="minimize",
|
|
description=change,
|
|
code_hash=current_hash,
|
|
delta_pct=round(delta, 2)
|
|
)
|
|
self.log_experiment(run)
|
|
|
|
if keep:
|
|
baseline = new_value
|
|
print(f"✓ KEPT ({new_value}, Δ={delta:+.2f}%)")
|
|
else:
|
|
print(f"✗ DISCARDED ({new_value}, Δ={delta:+.2f}%)")
|
|
# TODO: revert the change
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
loop = AutoresearchLoop()
|
|
loop.run(max_iterations=5)
|
|
|
|
print("\nExperiment log written to experiments.jsonl")
|
|
print("Run: cat experiments.jsonl | python -m json.tool")
|