agentic-ai-engineering/course/labs/L7-autoresearch/solution.py

"""
Lab 7.7: Build an Autoresearch Loop -- SOLUTION

Self-improving experiment loop with integrity guards.
"""

import hashlib
import json
import time
import random
from dataclasses import dataclass, asdict
from datetime import datetime, timezone


@dataclass
class ExperimentRun:
    run: int
    status: str
    metric_name: str
    metric_value: float
    metric_unit: str
    metric_direction: str
    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()


def hash_code(file_path: str) -> str:
    try:
        with open(file_path, "rb") as f:
            return hashlib.sha256(f.read()).hexdigest()[:12]
    except FileNotFoundError:
        return "no-file"


def detect_grind(previous_hash: str | None, current_hash: str) -> bool:
    if previous_hash is None:
        return False
    return previous_hash == current_hash


def median_score(scores: list[float]) -> float:
    sorted_scores = sorted(scores)
    return sorted_scores[len(sorted_scores) // 2]


class AutoresearchLoop:
    def __init__(self, log_path: str = "experiments.jsonl"):
        self.log_path = log_path
        self.run_count = 0
        self.last_code_hash = None
        self.history = []
    
    def setup_baseline(self) -> float:
        print("  Running baseline...")
        time.sleep(0.2)
        return 52.0  # baseline latency in ms
    
    def propose_change(self) -> str:
        changes = [
            "Remove N+1 count queries by pre-aggregating",
            "Memoize DB reads to avoid repeated sync work",
            "Add connection pooling to DB handler",
            "Cache repeated API responses in memory",
            "Move heavy computation to background worker",
            "Optimize hot loop with list comprehension",
        ]
        return random.choice(changes)
    
    def apply_change(self, change_desc: str):
        print(f"  Applying: {change_desc}")
        time.sleep(0.1)
    
    def revert_change(self):
        print("  Reverting change...")
        time.sleep(0.05)
    
    def measure(self) -> float:
        time.sleep(0.15)
        return max(30.0, 52.0 + random.gauss(-3, 4))
    
    def log_experiment(self, run: ExperimentRun):
        with open(self.log_path, "a") as f:
            f.write(json.dumps(asdict(run)) + "\n")
        self.history.append(run.metric_value)
    
    def should_keep(self, baseline: float, current: float, direction: str) -> bool:
        if direction == "minimize":
            return current < baseline
        else:
            return current > baseline
    
    def run(self, max_iterations: int = 10):
        print("Starting autoresearch loop...")
        print(f"Log: {self.log_path}")
        
        baseline = self.setup_baseline()
        print(f"Baseline: {baseline} ms")
        
        self.log_experiment(ExperimentRun(
            run=0, status="keep",
            metric_name="latency", 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} ---")
            
            change = self.propose_change()
            print(f"Proposed: {change}")
            
            current_hash = hash_code("agent.py")
            
            if detect_grind(self.last_code_hash, current_hash):
                print("[WARN]  Grind detected! Skipping (code unchanged)")
                continue
            self.last_code_hash = current_hash
            
            self.apply_change(change)
            new_value = self.measure()
            
            keep = self.should_keep(baseline, new_value, "minimize")
            delta = ((new_value - baseline) / baseline) * 100
            
            if not keep:
                self.revert_change()
            
            run = ExperimentRun(
                run=i+1,
                status="keep" if keep else "discard",
                metric_name="latency",
                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:.1f} ms, D={delta:+.2f}%)")
            else:
                print(f"- DISCARDED ({new_value:.1f} ms, D={delta:+.2f}%)")
        
        if self.history:
            med = median_score(self.history)
            print(f"\n{'='*40}")
            print(f"Baseline: {52.0} ms")
            print(f"Best:     {min(self.history):.1f} ms")
            print(f"Median:   {med:.1f} ms")
            improvement = ((52.0 - med) / 52.0) * 100
            print(f"Improvement: {improvement:+.1f}% (using median, not best)")
            print(f"Runs logged: {len(self.history)}")
            print(f"Experiment log: {self.log_path}")


if __name__ == "__main__":
    loop = AutoresearchLoop()
    loop.run(max_iterations=8)
    
    print("\nView results: cat experiments.jsonl | python -m json.tool")