Capstone: Model Training, Iteration & AI Feedback

train  ->  read the numbers  ->  pick ONE change  ->  train again  ->  compare  ->  log

!pip install -q ultralytics

from ultralytics import YOLO
 
model = YOLO("yolov8n.pt")                      # the small, fast model you used in lesson 24
model.train(data="dataset/data.yaml", epochs=40, imgsz=640, batch=16, project="capstone", name="base")
# results land in capstone/base/ : results.csv, weights/best.pt, and the curves

"""One change at a time: a tiny experiment log, a reader for Ultralytics results.csv, and rule-based diagnosis.
Code decides what the numbers say. Claude (advisor.py) only suggests what to try next."""
import csv
import json
import os
 
KNOBS = ["model", "imgsz", "epochs", "batch", "lr0", "mosaic", "fliplr", "hsv_v", "degrees", "freeze", "patience", "data"]
 
 
def read_results_csv(path):
    """Ultralytics writes results.csv with one row per epoch. Column names can have padding spaces."""
    rows = []
    with open(path, newline="") as f:
        for r in csv.DictReader(f):
            r = {k.strip(): v.strip() for k, v in r.items() if k}
            rows.append({
                "epoch": int(float(r["epoch"])),
                "train_loss": float(r["train/box_loss"]),
                "val_loss": float(r["val/box_loss"]),
                "precision": float(r["metrics/precision(B)"]),
                "recall": float(r["metrics/recall(B)"]),
                "map50": float(r["metrics/mAP50(B)"]),
            })
    return rows
 
 
def changed_settings(prev_config, config):
    """Which settings differ between two runs?"""
    keys = set(prev_config) | set(config)
    return sorted(k for k in keys if prev_config.get(k) != config.get(k))
 
 
def log_run(path, name, config, metrics, note="", from_claude=False):
    """Append a run to a JSON-lines file. Returns warnings (never blocks: the log should tell the truth)."""
    runs = load(path)
    warns = []
    if runs:
        diff = changed_settings(runs[-1]["config"], config)
        if len(diff) == 0:
            warns.append("nothing changed since the last run: this is a repeat, not an experiment")
        elif len(diff) > 1:
            warns.append(f"{len(diff)} settings changed ({', '.join(diff)}): you cannot tell which one helped")
    else:
        diff = []
    unknown = [k for k in config if k not in KNOBS]
    if unknown:
        warns.append(f"unknown settings: {', '.join(unknown)}")
    run = {"name": name, "config": config, "changed": diff, "metrics": metrics, "note": note,
           "from_claude": bool(from_claude), "warnings": warns}
    with open(path, "a") as f:
        f.write(json.dumps(run) + "\n")
    return warns
 
 
def load(path):
    if not os.path.exists(path):
        return []
    return [json.loads(line) for line in open(path) if line.strip()]
 
 
def table(path, baseline_map):
    """Plain-text summary: run, what changed, mAP50, change vs the previous run, vs the Unit 5 baseline."""
    runs = load(path)
    lines = [f"{'run':<12}{'changed':<14}{'mAP50':>7}{'vs prev':>9}{'vs base':>9}  claude?"]
    prev = None
    for r in runs:
        m = r["metrics"]["map50"]
        d_prev = "" if prev is None else f"{m - prev:+.3f}"
        d_base = f"{m - baseline_map:+.3f}"
        lines.append(f"{r['name']:<12}{','.join(r['changed']) or '-':<14}{m:>7.3f}{d_prev:>9}{d_base:>9}  {'yes' if r['from_claude'] else ''}")
        prev = m
    return "\n".join(lines)
 
 
def best_run(path):
    runs = load(path)
    return max(runs, key=lambda r: r["metrics"]["map50"]) if runs else None
 
 
def verdict(path, baseline_map):
    """The session outcome: does the best run reach the Unit 5 baseline?"""
    b = best_run(path)
    if b is None:
        return "no runs logged"
    gap = b["metrics"]["map50"] - baseline_map
    return (f"best is {b['name']} at mAP50 {b['metrics']['map50']:.3f}, "
            + ("reaching" if gap >= 0 else "below") + f" the baseline {baseline_map:.3f} ({gap:+.3f})")
 
 
def diagnose(history, per_class_ap=None, class_counts=None, baseline_map=None):
    """Rule-based reading of one training run. Returns a list of short findings.
    history: list of per-epoch dicts from read_results_csv. per_class_ap / class_counts: {class: number}."""
    out = []
    if len(history) >= 6:
        best = min(range(len(history)), key=lambda i: history[i]["val_loss"])          # epoch where validation loss was lowest
        late = len(history) - 1 - best >= 5
        val_up = history[-1]["val_loss"] > history[best]["val_loss"] * 1.05
        train_down = history[-1]["train_loss"] < history[best]["train_loss"]
        if late and val_up and train_down:
            out.append(f"overfitting: validation loss was lowest at epoch {history[best]['epoch']} and has risen since, while training loss keeps falling (more data or augmentation, fewer epochs)")
        elif history[-1]["train_loss"] > history[0]["train_loss"] * 0.7:
            out.append("underfitting or too short: training loss has barely fallen (more epochs, bigger model, check labels)")
    final = history[-1]
    if final["precision"] - final["recall"] > 0.15:
        out.append("misses objects: precision is much higher than recall (lower the confidence threshold, bigger images for small things, more examples)")
    if final["recall"] - final["precision"] > 0.15:
        out.append("false alarms: recall is much higher than precision (more background examples, check for missing labels)")
    if per_class_ap:
        mean = sum(per_class_ap.values()) / len(per_class_ap)
        for c, ap in per_class_ap.items():
            if ap < 0.6 * mean:
                n = (class_counts or {}).get(c)
                extra = f" and it has only {n} boxes" if n is not None and n < 0.6 * (sum(class_counts.values()) / len(class_counts)) else ""
                out.append(f"weak class '{c}': AP {ap:.2f} against a mean of {mean:.2f}{extra}")
    if baseline_map is not None and final["map50"] < baseline_map:
        out.append(f"below the Unit 5 baseline: {final['map50']:.3f} against {baseline_map:.3f}")
    return out or ["no obvious problem in the numbers: try one change from your list and log it"]

"""Write a pretend Ultralytics results.csv: validation loss turns upward from epoch 25 (overfitting)."""
import csv
import math
 
 
def make(path, epochs=40, overfit=True):
    rows = []
    for e in range(1, epochs + 1):
        tr = 1.8 * math.exp(-e / 12) + 0.35
        va = 1.9 * math.exp(-e / 14) + 0.55 + (0.03 * max(0, e - 25) if overfit else 0)
        mp = 0.80 * (1 - math.exp(-e / 10)) - (0.004 * max(0, e - 25) if overfit else 0)
        rows.append({"                  epoch": e, "         train/box_loss": round(tr, 4), "           val/box_loss": round(va, 4),
                     "   metrics/precision(B)": round(min(0.95, mp + 0.12), 4), "      metrics/recall(B)": round(max(0.0, mp - 0.08), 4),
                     "       metrics/mAP50(B)": round(mp, 4)})
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0])); w.writeheader(); w.writerows(rows)
    return path

from make_results_demo import make
from experiment_log import *
 
make("results.csv")
history = read_results_csv("results.csv")
per_class = {"ball": 0.31, "player": 0.78, "goal": 0.70}          # from your own validation printout
counts = {"ball": 30, "player": 200, "goal": 90}                  # boxes per class, from the lesson 48 gate
 
findings = diagnose(history, per_class, counts, baseline_map=0.60)
for f in findings:
    print("-", f)
 
config = {"model": "yolov8n", "imgsz": 640, "epochs": 40}
log_run("runs.jsonl", "base", config, {"map50": history[-1]["map50"]})
print(table("runs.jsonl", baseline_map=0.60))
print(verdict("runs.jsonl", baseline_map=0.60))

"""Send the numbers to Claude and get back at most 3 ONE-SETTING suggestions. The team picks; the log records what happened."""
import json
import re
 
from experiment_log import KNOBS
 
MODEL = "claude-sonnet-5-5"
 
SYSTEM = f"""You are a coach for a student object-detection project (YOLO).
You get a summary of one training run: the settings, the numbers, and what simple rules found.
Rules:
- Suggest at most 3 next experiments.
- Each suggestion changes exactly ONE setting, chosen from: {', '.join(KNOBS)}. ('data' means adding more labelled images.)
- For each say why (using the numbers you were given) and which metric you expect to move.
- Do not invent numbers. Do not promise results. Say what to watch.
- Answer ONLY with JSON: a list of objects like {{"setting": "imgsz", "value": 960, "why": "<one sentence>", "expect": "<one sentence>"}}."""
 
 
def summary_text(config, history, per_class_ap, findings, baseline_map):
    last = history[-1]
    return json.dumps({
        "settings": config, "epochs_run": len(history),
        "final": {k: round(last[k], 4) for k in ("train_loss", "val_loss", "precision", "recall", "map50")},
        "per_class_ap50": {k: round(v, 3) for k, v in (per_class_ap or {}).items()},
        "rules_found": findings, "unit5_baseline_map50": baseline_map}, indent=2)
 
 
def parse_suggestions(text):
    m = re.search(r"\[.*\]", text, re.S)
    if not m:
        raise ValueError("no JSON list in the answer")
    items = json.loads(m.group(0))
    if not isinstance(items, list) or not 1 <= len(items) <= 3:
        raise ValueError("expected 1 to 3 suggestions")
    for it in items:
        if not isinstance(it, dict) or set(it) != {"setting", "value", "why", "expect"}:
            raise ValueError(f"bad suggestion (must have exactly setting, value, why, expect): {it}")
        if it["setting"] not in KNOBS:
            raise ValueError(f"unknown setting: {it['setting']}")
        if isinstance(it["value"], (list, dict)) or it["value"] is None:
            raise ValueError("value must be a single number, word or true/false")
        for k in ("why", "expect"):
            if not isinstance(it[k], str) or not it[k] or len(it[k]) > 250:
                raise ValueError(f"'{k}' must be a short sentence")
    if len({it["setting"] for it in items}) != len(items):
        raise ValueError("two suggestions change the same setting")
    return items
 
 
def ask(config, history, per_class_ap, findings, baseline_map, client, model=MODEL, retries=1):
    msg = [{"role": "user", "content": "Here is the run:\n" + summary_text(config, history, per_class_ap, findings, baseline_map) + "\n\nWhat should we try next?"}]
    for attempt in range(retries + 1):
        try:
            r = client.messages.create(model=model, max_tokens=700, system=SYSTEM, messages=msg)
            return parse_suggestions("".join(b.text for b in r.content if b.type == "text"))
        except ValueError:
            if attempt == retries:
                raise
        except ConnectionError:
            if attempt == retries:
                return []                                   # no network: the team falls back to the rule findings
    return []
 
 
def apply_one(config, suggestion):
    """Return a new config with exactly the one suggested change."""
    new = dict(config)
    new[suggestion["setting"]] = suggestion["value"]
    return new

import json
from fake_client import FakeClient
from advisor import ask, apply_one
 
reply = json.dumps([
  {"setting": "data", "value": "add 40 ball images", "why": "Ball AP is 0.31 and it has only 30 boxes.", "expect": "Ball AP50 should rise."},
  {"setting": "epochs", "value": 25, "why": "Validation loss was lowest near epoch 25.", "expect": "Validation loss should stop rising."}])
 
suggestions = ask(config, history, per_class, findings, 0.60, FakeClient(reply))
for s in suggestions:
    print(s["setting"], "->", s["value"], "|", s["why"])
 
next_config = apply_one(config, suggestions[1])              # YOU choose which one to try
log_run("runs.jsonl", "r2", next_config, {"map50": 0.64}, note="stopped earlier", from_claude=True)
print(table("runs.jsonl", baseline_map=0.60))

import json, os
from experiment_log import *
from advisor import *
from fake_client import FakeClient
from make_results_demo import make
 
n = 0
def ok(name, cond):
    global n
    print(("PASS " if cond else "FAIL ") + name); n += bool(cond)
    if not cond: raise SystemExit(1)
 
make("results.csv"); h = read_results_csv("results.csv")
ok("reads padded Ultralytics columns", len(h) == 40 and h[0]["epoch"] == 1 and 0 < h[-1]["map50"] < 1)
 
f = diagnose(h)
ok("overfitting is found", any(x.startswith("overfitting") for x in f))
ok("a healthy run is not called overfit", not any(x.startswith("overfitting") for x in diagnose(read_results_csv(make("healthy.csv", overfit=False)))))
ok("short run: underfitting rule not triggered without data", isinstance(diagnose(h[:3]), list))
 
mis = [dict(h[-1], precision=0.9, recall=0.5)] * 6
ok("misses vs false alarms", any("misses" in x for x in diagnose(mis)) and any("false alarms" in x for x in diagnose([dict(mis[0], precision=0.5, recall=0.9)] * 6)))
w = diagnose(h, per_class_ap={"ball": 0.2, "player": 0.8, "goal": 0.7}, class_counts={"ball": 30, "player": 200, "goal": 90})
ok("weak class found with its box count", any("weak class 'ball'" in x and "only 30 boxes" in x for x in w))
ok("below baseline flagged", any("below the Unit 5 baseline" in x for x in diagnose(h, baseline_map=0.95)))
 
lg = "runs.jsonl"
if os.path.exists(lg): os.remove(lg)
base = {"model": "yolov8n", "imgsz": 640, "epochs": 40}
ok("first run: no warnings", log_run(lg, "base", base, {"map50": 0.55}) == [])
w1 = log_run(lg, "r2", {**base, "imgsz": 960}, {"map50": 0.61}, from_claude=True)
ok("one change: no warnings", w1 == [])
w2 = log_run(lg, "r3", {**base, "imgsz": 960, "epochs": 80, "lr0": 0.005}, {"map50": 0.66})
ok("two changes warned", any("2 settings changed" in x or "3 settings changed" in x for x in w2))
w3 = log_run(lg, "r4", {**base, "imgsz": 960, "epochs": 80, "lr0": 0.005}, {"map50": 0.66})
ok("repeat run warned", any("repeat" in x for x in w3))
w4 = log_run(lg, "r5", {**base, "imgsz": 960, "epochs": 80, "lr0": 0.005, "magic": 1}, {"map50": 0.5})
ok("unknown setting warned", any("unknown settings: magic" in x for x in w4))
ok("changed list recorded", load(lg)[1]["changed"] == ["imgsz"])
t = table(lg, 0.6)
ok("table shows deltas and claude flag", "+0.060" in t and "yes" in t and "-0.050" in t)
ok("best run and verdict", best_run(lg)["name"] in ("r3", "r4") and "reaching" in verdict(lg, 0.6) and "below" in verdict(lg, 0.9))
ok("empty log verdict", verdict("nope.jsonl", 0.5) == "no runs logged")
 
good = json.dumps([{"setting": "imgsz", "value": 960, "why": "Recall is low and the ball is small.", "expect": "Recall of ball should rise."},
                   {"setting": "data", "value": "add 40 ball images", "why": "Ball AP is the lowest.", "expect": "Ball AP50 up."}])
c = FakeClient(good)
s = ask(base, h, {"ball": 0.3}, f, 0.6, c)
ok("good advice parsed", len(s) == 2 and s[0]["setting"] == "imgsz")
ok("the prompt carries numbers and findings", "rules_found" in c.calls[0]["messages"][0]["content"] and "overfitting" in c.calls[0]["messages"][0]["content"])
ok("system prompt lists the knobs", "imgsz" in c.calls[0]["system"] and "ONE" in c.calls[0]["system"])
ok("apply_one changes exactly one setting", changed_settings(base, apply_one(base, s[0])) == ["imgsz"])
 
def bad(reply):
    try: parse_suggestions(reply); return False
    except ValueError: return True
ok("two settings in one suggestion rejected", bad(json.dumps([{"setting": ["imgsz", "epochs"], "value": 1, "why": "x", "expect": "y"}])) )
ok("unknown setting rejected", bad(json.dumps([{"setting": "magic", "value": 1, "why": "x", "expect": "y"}])))
ok("more than 3 rejected", bad(json.dumps([{"setting": k, "value": 1, "why": "x", "expect": "y"} for k in KNOBS[:4]])))
ok("extra keys rejected", bad(json.dumps([{"setting": "imgsz", "value": 1, "why": "x", "expect": "y", "code": "rm -rf"}])))
ok("duplicate setting rejected", bad(json.dumps([{"setting": "imgsz", "value": 1, "why": "x", "expect": "y"}] * 2)))
ok("list value rejected", bad(json.dumps([{"setting": "imgsz", "value": [1, 2], "why": "x", "expect": "y"}])))
ok("long why rejected", bad(json.dumps([{"setting": "imgsz", "value": 1, "why": "x" * 300, "expect": "y"}])))
ok("prose without JSON rejected", bad("Just train longer!"))
 
c2 = FakeClient("not json"); 
try: ask(base, h, None, f, 0.6, c2); r = "no error"
except ValueError: r = "error"
ok("bad answer retried once then raised", r == "error" and len(c2.calls) == 2)
c3 = FakeClient(good, fail_times=5)
ok("network down: empty list, rule findings still work", ask(base, h, None, f, 0.6, c3) == [] and len(c3.calls) == 2)
c4 = FakeClient(good, fail_times=1)
ok("one network blip is retried", len(ask(base, h, None, f, 0.6, c4)) == 2)
 
for p in ("results.csv", "healthy.csv", lg): os.path.exists(p) and os.remove(p)
print(f"\n{n} checks passed")