`35c558f`

Record preference mine metrics

Authored by

espadonne 2 weeks ago

SHA: 35c558faed456e06672a25accc894e3f9df52537
Parents: dcb09c5
Tree: 6f96a60

10 changed files

Status	File	+	-
M	`src/dlm/cli/commands.py`	19	0
M	`src/dlm/metrics/__init__.py`	2	0
M	`src/dlm/metrics/db.py`	12	0
M	`src/dlm/metrics/events.py`	18	0
M	`src/dlm/metrics/queries.py`	62	0
M	`src/dlm/metrics/recorder.py`	20	0
M	`tests/unit/cli/test_preference_cmd.py`	9	0
M	`tests/unit/metrics/test_db_schema.py`	10	1
M	`tests/unit/metrics/test_queries.py`	60	1
M	`tests/unit/metrics/test_recorder.py`	33	1

src/dlm/cli/commands.pymodified

          build_backend,
          select_backend,
+     )
 +    from dlm.metrics import MetricsRecorder, PreferenceMineEvent
 +    from dlm.metrics.events import PreferenceMineWriteMode
      from dlm.modality import modality_for
      from dlm.preference import (
          InvalidJudgeSpecError,
      finally:
          backend_obj.unload()
 +    recorder = MetricsRecorder(store.root)
++
 +    def _record_preference_mine(write_mode: PreferenceMineWriteMode) -> None:
 +        recorder.record_preference_mine(
 +            PreferenceMineEvent(
 +                run_id=run_id,
 +                judge_name=judge_obj.name,
 +                sample_count=samples,
 +                mined_pairs=len(plan.additions),
 +                skipped_prompts=len(plan.skipped),
 +                write_mode=write_mode,
 +            )
 +        )
++
      out_console.print(render_mine_plan(plan))
      if not plan.additions:
          clear_pending_plan(store)
 +        _record_preference_mine("empty")
          out_console.print(
              "\n[yellow]no candidates to mine[/yellow] — either instruction prompts "
              "did not yield a confident pair, or the matching preference sections "
          out_console.print(render_apply_plan(apply_plan))
          summary = apply_preference_plan(parsed, apply_plan, target=path)
          clear_pending_plan(store)
 +        _record_preference_mine("applied")
          out_console.print(
              f"\n[green]preference:[/green] wrote {summary.added} section(s) to {path} "
              f"({summary.skipped} skipped)"
          return
      pending = save_pending_plan(store, source_path=path.resolve(), sections=sections)
 +    _record_preference_mine("staged")
      out_console.print(
          f"\n[green]preference:[/green] staged {len(pending.sections)} mined preference "
          f"section(s). Run [bold]dlm preference apply {path}[/bold] to write them."

src/dlm/metrics/__init__.pymodified

      EvalEvent,
      ExportEvent,
      Phase,
 +    PreferenceMineEvent,
      RunEnd,
      RunStart,
      Status,
      "MetricsRecorder",
      "MetricsSchemaError",
      "Phase",
 +    "PreferenceMineEvent",
      "RunEnd",
      "RunStart",
      "Status",

src/dlm/metrics/db.pymodified

          PRIMARY KEY (run_id, adapter_name)
+     )
      """,
 +    """
 +    CREATE TABLE IF NOT EXISTS preference_mining (
 +        event_id         INTEGER PRIMARY KEY AUTOINCREMENT,
 +        run_id           INTEGER NOT NULL,
 +        judge_name       TEXT NOT NULL,
 +        sample_count     INTEGER NOT NULL,
 +        mined_pairs      INTEGER NOT NULL,
 +        skipped_prompts  INTEGER NOT NULL,
 +        write_mode       TEXT NOT NULL,
 +        at               TEXT NOT NULL
 +    )
 +    """,
+ ]

src/dlm/metrics/events.pymodified

  Phase = Literal["sft", "dpo", "orpo", "cpt"]
  Status = Literal["running", "ok", "failed", "cancelled"]
 +PreferenceMineWriteMode = Literal["staged", "applied", "empty"]
  def _utc_iso() -> str:
              object.__setattr__(self, "at", _utc_iso())
 +@dataclass(frozen=True)
 +class PreferenceMineEvent:
 +    """Emitted from `dlm preference mine` after judging completes."""
++
 +    run_id: int
 +    judge_name: str
 +    sample_count: int
 +    mined_pairs: int
 +    skipped_prompts: int
 +    write_mode: PreferenceMineWriteMode
 +    at: str = ""
++
 +    def __post_init__(self) -> None:
 +        if not self.at:
 +            object.__setattr__(self, "at", _utc_iso())
++
++
  @dataclass(frozen=True)
  class ExportEvent:
      """Emitted from `dlm export` on completion."""

src/dlm/metrics/queries.pymodified

          return self.cache_hits / total if total else 0.0
 +@dataclass(frozen=True)
 +class PreferenceMineRow:
 +    """One row from the `preference_mining` table."""
++
 +    event_id: int
 +    run_id: int
 +    judge_name: str
 +    sample_count: int
 +    mined_pairs: int
 +    skipped_prompts: int
 +    write_mode: str
 +    at: str
++
++
  def recent_runs(
      store_root: Path,
      *,
      return TokenizationRow(*row)
 +def preference_mining_for_run(store_root: Path, run_id: int) -> list[PreferenceMineRow]:
 +    """All preference-mine events for `run_id`, oldest first."""
 +    try:
 +        with connect(store_root) as conn:
 +            rows = conn.execute(
 +                "SELECT event_id, run_id, judge_name, sample_count, mined_pairs, "
 +                "skipped_prompts, write_mode, at "
 +                "FROM preference_mining WHERE run_id = ? ORDER BY event_id ASC",
 +                (run_id,),
 +            ).fetchall()
 +    except sqlite3.Error:
 +        return []
 +    return [PreferenceMineRow(*row) for row in rows]
++
++
 +def latest_preference_mining(store_root: Path) -> PreferenceMineRow | None:
 +    """The most-recent preference-mine row, or None when absent."""
 +    try:
 +        with connect(store_root) as conn:
 +            row = conn.execute(
 +                "SELECT event_id, run_id, judge_name, sample_count, mined_pairs, "
 +                "skipped_prompts, write_mode, at "
 +                "FROM preference_mining ORDER BY event_id DESC LIMIT 1"
 +            ).fetchone()
 +    except sqlite3.Error:
 +        return None
 +    if row is None:
 +        return None
 +    return PreferenceMineRow(*row)
++
++
  @dataclass(frozen=True)
  class GateEventRow:
      """One row of the gate_events table (per-run per-adapter)."""
+         }
          for e in evals
+     ]
++
++
 +def preference_mining_to_dict(rows: list[PreferenceMineRow]) -> list[dict[str, Any]]:
 +    """JSON-serializable view used by `dlm metrics --json` and `dlm show --json`."""
 +    return [
 +        {
 +            "event_id": row.event_id,
 +            "run_id": row.run_id,
 +            "judge_name": row.judge_name,
 +            "sample_count": row.sample_count,
 +            "mined_pairs": row.mined_pairs,
 +            "skipped_prompts": row.skipped_prompts,
 +            "write_mode": row.write_mode,
 +            "at": row.at,
 +        }
 +        for row in rows
 +    ]

src/dlm/metrics/recorder.pymodified

      EvalEvent,
      ExportEvent,
      GateEvent,
 +    PreferenceMineEvent,
      RunEnd,
      RunStart,
      StepEvent,
          self._with_conn(_do, failure_key="gate", hard_fail=False)
 +    def record_preference_mine(self, event: PreferenceMineEvent) -> None:
 +        def _do(conn: sqlite3.Connection) -> None:
 +            conn.execute(
 +                "INSERT INTO preference_mining "
 +                "(run_id, judge_name, sample_count, mined_pairs, skipped_prompts, write_mode, at) "
 +                "VALUES (?, ?, ?, ?, ?, ?, ?)",
 +                (
 +                    event.run_id,
 +                    event.judge_name,
 +                    event.sample_count,
 +                    event.mined_pairs,
 +                    event.skipped_prompts,
 +                    event.write_mode,
 +                    event.at,
 +                ),
 +            )
++
 +        self._with_conn(_do, failure_key="preference_mine", hard_fail=False)
++
      def record_export(self, event: ExportEvent) -> None:
          def _do(conn: sqlite3.Connection) -> None:
              conn.execute(

tests/unit/cli/test_preference_cmd.pymodified

  from dlm.base_models import BaseModelSpec
  from dlm.cli.app import app
  from dlm.doc.parser import parse_file
 +from dlm.metrics.queries import preference_mining_for_run
  from dlm.preference.judge import PairScore
  from dlm.preference.pending import load_pending_plan
  from dlm.store.manifest import Manifest, TrainingRunSummary, save_manifest
          assert len(pending.sections) == 1
          assert pending.sections[0].auto_mined is True
 +        rows = preference_mining_for_run(for_dlm(_DLM_ID, home=home).root, run_id=7)
 +        assert len(rows) == 1
 +        assert rows[0].judge_name == "stub:judge"
 +        assert rows[0].sample_count == 2
 +        assert rows[0].mined_pairs == 1
 +        assert rows[0].skipped_prompts == 0
 +        assert rows[0].write_mode == "staged"
++
      def test_apply_writes_staged_preferences_and_clears_pending(
          self,
          tmp_path: Path,

tests/unit/metrics/test_db_schema.pymodified

              tables = {
                  row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
+             }
 -        assert tables == {"runs", "steps", "evals", "exports", "tokenization", "gate_events"}
 +        user_tables = {table for table in tables if not table.startswith("sqlite_")}
 +        assert user_tables == {
 +            "runs",
 +            "steps",
 +            "evals",
 +            "exports",
 +            "tokenization",
 +            "gate_events",
 +            "preference_mining",
 +        }
      def test_wal_mode_enabled(self, tmp_path: Path) -> None:
          with connect(tmp_path) as conn:

tests/unit/metrics/test_queries.pymodified

  from datetime import UTC, datetime, timedelta
  from pathlib import Path
 -from dlm.metrics.events import EvalEvent, RunEnd, RunStart, StepEvent
 +from dlm.metrics.events import EvalEvent, PreferenceMineEvent, RunEnd, RunStart, StepEvent
  from dlm.metrics.queries import (
      evals_for_run,
      evals_to_dict,
 +    latest_preference_mining,
      latest_run_id,
 +    preference_mining_for_run,
 +    preference_mining_to_dict,
      recent_runs,
      runs_to_dict,
      steps_for_run,
              rec.record_step(StepEvent(run_id=run_id, step=step, loss=2.0 - 0.1 * step))
          rec.record_eval(EvalEvent(run_id=run_id, step=30, val_loss=1.5))
          rec.record_run_end(RunEnd(run_id=run_id, status="ok"))
 +    rec.record_preference_mine(
 +        PreferenceMineEvent(
 +            run_id=2,
 +            judge_name="sway",
 +            sample_count=4,
 +            mined_pairs=1,
 +            skipped_prompts=0,
 +            write_mode="staged",
 +        )
 +    )
 +    rec.record_preference_mine(
 +        PreferenceMineEvent(
 +            run_id=2,
 +            judge_name="hf:test/reward",
 +            sample_count=6,
 +            mined_pairs=2,
 +            skipped_prompts=3,
 +            write_mode="applied",
 +        )
 +    )
  class TestRecentRuns:
          assert latest_run_id(tmp_path) is None
 +class TestPreferenceMiningQueries:
 +    def test_preference_mining_for_run_returns_oldest_first(self, tmp_path: Path) -> None:
 +        _seed(tmp_path)
 +        rows = preference_mining_for_run(tmp_path, run_id=2)
 +        assert [row.judge_name for row in rows] == ["sway", "hf:test/reward"]
 +        assert [row.write_mode for row in rows] == ["staged", "applied"]
++
 +    def test_latest_preference_mining_returns_most_recent_event(self, tmp_path: Path) -> None:
 +        _seed(tmp_path)
 +        row = latest_preference_mining(tmp_path)
 +        assert row is not None
 +        assert row.judge_name == "hf:test/reward"
 +        assert row.write_mode == "applied"
++
 +    def test_latest_preference_mining_none_when_empty(self, tmp_path: Path) -> None:
 +        from dlm.metrics.db import connect
++
 +        with connect(tmp_path) as _conn:
 +            pass
 +        assert latest_preference_mining(tmp_path) is None
++
++
  class TestDictSerialization:
      def test_runs_to_dict_shape(self, tmp_path: Path) -> None:
          _seed(tmp_path)
          assert all({"step", "loss", "lr", "grad_norm", "at"}.issubset(s.keys()) for s in steps)
          evals = evals_to_dict(evals_for_run(tmp_path, run_id=1))
          assert all("val_loss" in e for e in evals)
++
 +    def test_preference_mining_to_dict_shape(self, tmp_path: Path) -> None:
 +        _seed(tmp_path)
 +        payload = preference_mining_to_dict(preference_mining_for_run(tmp_path, run_id=2))
 +        assert payload[0].keys() == {
 +            "event_id",
 +            "run_id",
 +            "judge_name",
 +            "sample_count",
 +            "mined_pairs",
 +            "skipped_prompts",
 +            "write_mode",
 +            "at",
 +        }

tests/unit/metrics/test_recorder.pymodified

  import pytest
  from dlm.metrics.db import metrics_db_path
 -from dlm.metrics.events import EvalEvent, ExportEvent, RunEnd, RunStart, StepEvent
 +from dlm.metrics.events import (
 +    EvalEvent,
 +    ExportEvent,
 +    PreferenceMineEvent,
 +    RunEnd,
 +    RunStart,
 +    StepEvent,
 +)
  from dlm.metrics.recorder import MetricsRecorder
          assert rows[0][3] == "mydoc:v1"
 +class TestPreferenceMining:
 +    def test_preference_mine_written_without_run_row(self, tmp_path: Path) -> None:
 +        rec = MetricsRecorder(tmp_path)
 +        rec.record_preference_mine(
 +            PreferenceMineEvent(
 +                run_id=7,
 +                judge_name="sway",
 +                sample_count=4,
 +                mined_pairs=2,
 +                skipped_prompts=1,
 +                write_mode="staged",
 +            )
 +        )
 +        rows = _select_all(metrics_db_path(tmp_path), "preference_mining")
 +        assert len(rows) == 1
 +        _, run_id, judge_name, sample_count, mined_pairs, skipped_prompts, write_mode, at = rows[0]
 +        assert run_id == 7
 +        assert judge_name == "sway"
 +        assert sample_count == 4
 +        assert mined_pairs == 2
 +        assert skipped_prompts == 1
 +        assert write_mode == "staged"
 +        assert at
++
++
  class TestBestEffort:
      def test_step_write_logs_error_once_per_stream(
          self,