train_cli.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
# This file is part of the Aquin Engine. Unauthorized copying, modification,
# distribution, or use of this file, via any medium, is strictly prohibited.
# Proprietary and confidential. See LICENSE for terms.
 
"""aquin train <recipe.yaml> — DataRevision → train → Checkpoint → EvalGate."""
 
from __future__ import annotations
 
import json
import sys
 
from aquin.recipe.schema import RecipeError
 
 
def _print_help() -> None:
    print("Train from a Recipe YAML (or JSON).")
    print("")
    print("Usage: aquin train <recipe.yaml> [--dry-run] [--json]")
    print("       aq train <recipe.yaml>")
    print("")
    print("  DataRevision  hash + optional snapshot of data.path")
    print("  Run           aquin_run/runs/<id>/")
    print("  Checkpoint    weights from this run")
    print("  EvalGate      pass/fail from eval.min_score")
    print("")
    print("Families: tabular (linear|boosting on CSV)  ·  llm (lora)")
    print("Pass a Recipe YAML/JSON path.")
 
 
def cmd_train(args: list[str]) -> None:
    if not args or args[0] in ("-h", "--help", "help"):
        _print_help()
        return
 
    dry = "--dry-run" in args
    as_json = "--json" in args
    paths = [a for a in args if not a.startswith("-")]
    if len(paths) != 1:
        print("Usage: aquin train <recipe.yaml>", file=sys.stderr)
        sys.exit(1)
 
    from aquin.recipe.train import train_recipe
 
    try:
        record = train_recipe(paths[0], dry_run=dry)
    except RecipeError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)
 
    if as_json:
        print(json.dumps(record.to_dict(), indent=2, default=str))
        return
 
    gate = record.gate or {}
    print(f"[aquin] Recipe     {record.name}  ({record.family})")
    print(f"[aquin] Run        {record.run_id}  {record.run_dir}")
    rev = record.revision
    rows = rev.get("n_rows")
    print(f"[aquin] Data       {rev.get('id')}  {rev.get('path')}  rows={rows}")
    if record.checkpoint:
        print(f"[aquin] Checkpoint {record.checkpoint.get('path')}")
    if record.status == "dry_run":
        print("[aquin] Dry-run    ok (not trained)")
        return
    if gate:
        flag = "PASS" if gate.get("passed") else "FAIL"
        if gate.get("skipped"):
            flag = "SKIP"
        print(
            f"[aquin] Gate       {flag}  {gate.get('metric')}={gate.get('score')}  "
            f"min_score={gate.get('min_score')}"
        )
    print(f"[aquin] Status     {record.status}")
    if record.status == "failed_gate":
        print("[aquin] Next       inspect this run (eval failed — that's when inspect runs)")
        print(f"         cat {record.run_dir}/inspect.json")
    if not gate.get("passed", True) and not gate.get("skipped"):
        sys.exit(2)