Compile, optimize, estimate

Lowering a graph to one ANE program, the accuracy-preserving autotuner, and the measurement-free cost model. All are reached from the top level (af.compile, af.tune, af.estimate, ...).

aneforge - a graph->compile->run frontend for the Apple Neural Engine. fp16 compute over the unentitled Espresso e5rt runtime; see docs/developer/overview.md.

Model

A compiled, fused ANE program. Call it with the input array(s), in af.input order.

Source code in aneforge/_compile.py

class Model:
  """A compiled, fused ANE program. Call it with the input array(s), in `af.input` order."""

  def __init__(self, prog, inputs: list[tuple[str, tuple]], out_name: str, out_shape, n_ops: int,
        input_tensors: list | None = None):
    self._prog = prog
    self._inputs = inputs
    # ordered input Tensors: lets autograd.Trainer map each input to its source Tensor.
    self._input_tensors = input_tensors if input_tensors is not None else []
    self._out_name, self._out_shape = out_name, out_shape
    self.n_ops = n_ops  # graph ops fused into this single program

  def __call__(self, *arrays: np.ndarray) -> np.ndarray:
    if len(arrays) != len(self._inputs): raise ValueError(f"expected {len(self._inputs)} input(s), got {len(arrays)}")
    dts = getattr(self._prog, "_input_dtypes", {})
    feed = {}
    for (name, shape), a in zip(self._inputs, arrays):
      a = np.asarray(a)
      if tuple(a.shape) != tuple(shape): raise ValueError(f"input '{name}' shape {a.shape} != compiled {shape}")
      if dts.get(name, "fp16") == "uint8":
        if not np.issubdtype(a.dtype, np.integer):
          raise TypeError(f"input '{name}' is a uint8 image port; pass an integer/uint8 "
                  f"array (got dtype {a.dtype})")
        feed[name] = a.astype(np.uint8)              # raw bytes
      else:
        feed[name] = a.astype(np.float16)
    return self._prog.eval(feed)[self._out_name].astype(np.float32)

  # ---- zero-copy hot-loop API: input_view (write) / execute / output_view (read), no memcpy ----
  def input_view(self, name: str | None = None) -> np.ndarray:
    """Writable fp16 view onto an input buffer (defaults to the sole input port)."""
    return self._prog.input_view(name or self._inputs[0][0])

  def output_view(self) -> np.ndarray:
    """fp16 view onto the output buffer, valid after `execute()`."""
    return self._prog.output_view(self._out_name)

  def execute(self) -> None:
    """Run once without binding inputs / reading outputs (pair with input_view/output_view)."""
    self._prog.execute()

  def release(self) -> None:
    self._prog.release()

input_view

input_view(name: str | None = None) -> np.ndarray

Writable fp16 view onto an input buffer (defaults to the sole input port).

Source code in aneforge/_compile.py

def input_view(self, name: str | None = None) -> np.ndarray:
  """Writable fp16 view onto an input buffer (defaults to the sole input port)."""
  return self._prog.input_view(name or self._inputs[0][0])

output_view

output_view() -> np.ndarray

fp16 view onto the output buffer, valid after execute().

Source code in aneforge/_compile.py

def output_view(self) -> np.ndarray:
  """fp16 view onto the output buffer, valid after `execute()`."""
  return self._prog.output_view(self._out_name)

execute

execute() -> None

Run once without binding inputs / reading outputs (pair with input_view/output_view).

Source code in aneforge/_compile.py

def execute(self) -> None:
  """Run once without binding inputs / reading outputs (pair with input_view/output_view)."""
  self._prog.execute()

SegmentedModel

A compiled plan: e5rt program segments interleaved with native-ANE sub-program calls (NETPLIST_OPS).

Source code in aneforge/_compile.py

class SegmentedModel:
  """A compiled plan: e5rt program segments interleaved with native-ANE sub-program calls (NETPLIST_OPS)."""

  def __init__(self, stages, inputs, out_id, out_shape, n_ops, n_netplist):
    self._stages = stages
    self._inputs = inputs  # list of (id, name, shape) in creation order
    self._out_id, self._out_shape = out_id, out_shape
    self.n_ops = n_ops
    self.n_netplist = self.n_sdpa = n_netplist  # n_sdpa kept for backward compat
    # A2: persistent Path-A workers, built lazily and cached. ANEFORGE_NETPLIST_WORKER=0 forces A1.
    self._workers: dict[int, "_Worker"] = {}
    self._worker_runs: dict[int, Callable[..., Any]] = {}
    self._worker_warned: set[str] = set()

  def __call__(self, *arrays: np.ndarray) -> np.ndarray:
    if len(arrays) != len(self._inputs): raise ValueError(f"expected {len(self._inputs)} input(s), got {len(arrays)}")
    env = {}
    for (iid, name, shape), a in zip(self._inputs, arrays):
      a = np.asarray(a)
      if tuple(a.shape) != tuple(shape): raise ValueError(f"input '{name}' shape {a.shape} != compiled {shape}")
      env[iid] = a.astype(np.float16)
    for st in self._stages:
      if st["kind"] == "region":
        feed = {name: env[sid] for sid, name in st["srcs"]}
        env[st["tid"]] = st["prog"].eval(feed)[st["out_var"]].astype(np.float16)
      else:  # native netplist-bridge sub-program (sdpa / argmax / topk / ...)
        runner = self._netplist_runner(st)
        src_arrays = [env[i] for i in st["src_ids"]]
        env[st["tid"]] = np.asarray(runner(src_arrays, st["attrs"]), dtype=np.float16)
    return env[self._out_id].astype(np.float32)

  def _netplist_runner(self, st) -> Callable[..., Any]:
    """Resolve a netplist stage's runner: prefer a persistent Path-A worker, else the subprocess bridge."""
    import os
    if os.environ.get("ANEFORGE_NETPLIST_WORKER", "1") == "0": return NETPLIST_OPS[st["op"]][0]
    # masked/causal SDPA needs per-call mask injection; the worker is mask-less, so use the bridge.
    if st["op"] == "sdpa" and (st["attrs"].get("causal") or st["attrs"].get("masked")): return NETPLIST_OPS[st["op"]][0]
    sid = st["tid"]
    run = self._worker_runs.get(sid)
    if run is not None: return run
    try:
      from . import _netplist_worker as nw
      if not nw.has_worker(st["op"]):
        # no worker route - subprocess bridge is the normal path.
        run = NETPLIST_OPS[st["op"]][0]
        self._worker_runs[sid] = run
        return run
      worker, run = nw.build_worker(st["op"], st["src_shapes"][0], st["attrs"])
      self._workers[sid] = worker
      self._worker_runs[sid] = run
      return run
    except Exception as e:
      # worker failure -> fall back to the subprocess bridge; remember it (signal once per op).
      if st["op"] not in self._worker_warned:
        self._worker_warned.add(st["op"])
        import warnings
        warnings.warn(
          f"aneforge: persistent worker for {st['op']!r} unavailable ({e!r}); "
          f"falling back to the slower per-call subprocess bridge. Set "
          f"ANEFORGE_NETPLIST_WORKER=0 to force (and silence) this path.",
          stacklevel=2)
      run = NETPLIST_OPS[st["op"]][0]
      self._worker_runs[sid] = run
      return run

  def release(self) -> None:
    for st in self._stages:
      if st["kind"] == "region": st["prog"].release()
    for w in self._workers.values():
      try:
        w.release()
      except Exception:
        pass
    self._workers.clear()
    self._worker_runs.clear()

PrecisionWarning

Bases: UserWarning

Emitted by compile when a graph has fp16-cancellation-risk nodes (results may be inaccurate).

Source code in aneforge/_compile.py

class PrecisionWarning(UserWarning):
  """Emitted by `compile` when a graph has fp16-cancellation-risk nodes (results may be inaccurate)."""

CrossChipFP16Warning

Bases: UserWarning

Emitted by cross_compile_check when a graph compiles for a different-family target but an op's fp16 value can diverge.

Source code in aneforge/_compile.py

class CrossChipFP16Warning(UserWarning):
  """Emitted by `cross_compile_check` when a graph compiles for a different-family target but an op's fp16 value can diverge."""

DispatchFloorWarning

Bases: UserWarning

Emitted by compile when a program is dispatch-floor-bound.

Source code in aneforge/_compile.py

class DispatchFloorWarning(UserWarning):
  """Emitted by `compile` when a program is dispatch-floor-bound."""

CompileBackoffError

Bases: RuntimeError

Raised (strict mode) when a compile is attempted within the backoff window.

Source code in aneforge/_circuit.py

class CompileBackoffError(RuntimeError):
  """Raised (strict mode) when a compile is attempted within the backoff window."""

compile

compile(out: Tensor, int8: bool = False, build_dir=None, opt: 'str | int | None' = 'routes', compress: str | None = None, compress_atol: float = 0.05, block_size: int = 32, validate: bool = False, target=None, _check_precision: bool = True)

Lower out into ONE fused ANE program (or a segmented plan if it has af.sdpa nodes).

Source code in aneforge/_compile.py

def compile(out: Tensor, int8: bool = False, build_dir=None, opt: "str | int | None" = "routes",
      compress: str | None = None, compress_atol: float = 0.05,
      block_size: int = 32, validate: bool = False, target=None,
      _check_precision: bool = True):
  """Lower `out` into ONE fused ANE program (or a segmented plan if it has `af.sdpa` nodes)."""
  if out.op == "input":
    # A graph whose output IS an input lowers to an empty MIL body, which crashes Espresso's
    # ANE compiler ("unordered_map::at: key not found"). Wrap in an exact scalar mul(1.0) so
    # the program always has at least one op (and downstream port naming stays consistent).
    out = out * 1.0
  if _check_precision:                 # once per user compile (internal re-entries pass False)
    _precision_signal(out, strict=validate)
    _dispatch_floor_signal(out)
    out = _retarget_for(out, target)  # gate ops/shapes for the target ANE family
  _OPT0 = (0, None, False)
  family = None
  if compress is not None:
    if opt not in _OPT0 and opt != "routes":
      raise NotImplementedError("compress= is not yet supported with opt>=1; " "use opt=0 for compressed weights")
    opt = 0          # compressed weights take the byte-identical lowering
    if compress == "auto": family = _resolve_family(target)
  if opt not in _OPT0:                  # lossless canon for routes/1/2/max; never opt=0 or compress
    from ._rewrite import canonicalize
    out = canonicalize(out)
  if opt == "routes":
    from . import _optimize
    return _optimize._compile_routes(out, int8=int8, build_dir=build_dir)
  if opt not in _OPT0: return _compile_opt(out, int8=int8, opt=opt)
  order = _topo(out)
  if any(t.op in NETPLIST_OPS for t in order):
    return _compile_segmented(out, int8, build_dir, compress, compress_atol, block_size, family=family)
  bad = sorted({t.op for t in order if t.op != "input" and t.op not in _EMIT})
  if bad: raise NotImplementedError(f"aneforge: ops not reachable on the ANE: {bad}")
  inputs = sorted((t for t in order if t.op == "input"), key=lambda t: t.attrs.get("idx", 0))
  if not inputs: raise ValueError("aneforge.compile: graph has no inputs")
  for i, t in enumerate(order): t._name = f"t{i}"

  em = _Emitter(int8, compress=compress, compress_atol=compress_atol, block_size=block_size, family=family)
  for t in order:
    if t.op != "input": _EMIT[t.op](em, t, t._name, [src._name for src in t.srcs])
  _compression_fallback_signal(em)

  prog = _assemble_and_compile(em, inputs, out._name, out.shape, build_dir)
  n_ops = sum(1 for t in order if t.op != "input")
  return Model(prog, [(t._name, t.shape) for t in inputs], out._name, out.shape, n_ops, input_tensors=list(inputs))

tune

tune(out, budget: int = 8, inputs=None, prune_factor: float = 1.5, reps: int = 20, atol: float = _ACCURACY_TOL, min_lossy_speedup: float = _MIN_LOSSY_SPEEDUP, verbose: bool = False, target_error: float | None = None)

Return the fastest CORRECT compiled Model for out (enumerate, prune, measure, validate, cache); target_error switches to the precision-aware path.

Source code in aneforge/_optimize.py

def tune(out, budget: int = 8, inputs=None, prune_factor: float = 1.5,
         reps: int = 20, atol: float = _ACCURACY_TOL,
         min_lossy_speedup: float = _MIN_LOSSY_SPEEDUP, verbose: bool = False,
         target_error: float | None = None):
  """Return the fastest CORRECT compiled Model for `out` (enumerate, prune, measure, validate, cache); `target_error` switches to the precision-aware path."""
  if target_error is not None:
    model, _report = tune_precision(out, target_error=target_error, inputs=inputs,
                                    reps=reps, verbose=verbose)
    return model

  input_shapes = _input_shapes(out)
  key = _graph_key(out, input_shapes)
  cache = _load_cache()

  configs = _variants(out)

  # cache hit: rebuild the cached winner directly (no measurement).
  if key in cache and cache[key].get("config") is not None:
    cfg = cache[key]["config"]
    if cfg in configs or cfg.get("int8_nodes"):
      if verbose:
        print(f"[tune] cache hit {key}: {_config_label(cfg)} "
              f"({cache[key].get('us', '?')} us)")
      return build_variant(out, cfg)

  if inputs is None: inputs = _gen_inputs(input_shapes)

  # rank by cost-model estimate; lossless variants first (the correctness reference).
  ranked = sorted(configs, key=lambda c: _estimate_variant(out, c))
  ranked = ([c for c in ranked if not c.get("lossy")] +
            [c for c in ranked if c.get("lossy")])

  best_cfg, best_us, baseline_out = None, float("inf"), None
  baseline_us = float("inf")     # the lossless fp16 baseline latency (the reference)
  best_est = min(_estimate_variant(out, c) for c in configs)
  n_measured = 0
  results = []
  skipped_lossy_no_baseline = False

  for cfg in ranked:
    if n_measured >= budget: break
    est = _estimate_variant(out, cfg)
    # a lossy variant must never become its own accuracy reference - skip if no baseline yet.
    if cfg.get("lossy") and baseline_out is None:
      results.append((cfg, est, None, "skipped"))
      skipped_lossy_no_baseline = True
      if verbose:
        print(f"[tune] skip {_config_label(cfg)}: no lossless baseline to "
              f"validate against")
      continue
    # prune lossy variants the model predicts far worse than the best estimate (never lossless ones).
    if cfg.get("lossy") and est > prune_factor * best_est and best_cfg is not None:
      results.append((cfg, est, None, "pruned"))
      if verbose:
        print(f"[tune] prune {_config_label(cfg)}: est {est:.0f}us > "
              f"{prune_factor}x best est {best_est:.0f}us")
      continue
    us, out_arr = measure(out, inputs, cfg, baseline_out=baseline_out, reps=reps, tol=atol)
    n_measured += 1
    if baseline_out is None and out_arr is not None:
      baseline_out = out_arr     # first successful = reference
      baseline_us = us
    results.append((cfg, est, us, "measured"))
    if verbose:
      print(f"[tune] {_config_label(cfg):28s} est {est:7.0f}us  meas "
            f"{us if us != float('inf') else 'INCORRECT/FAIL'} us")
    if us < best_us:
      # a lossy variant must beat the lossless baseline by a real margin; a lossless swap wins on raw speed.
      if cfg.get("lossy") and not (baseline_us == float("inf")
                                   or us * min_lossy_speedup <= baseline_us):
        continue
      best_us, best_cfg = us, cfg

  if skipped_lossy_no_baseline:
    warnings.warn(
      "tune(): the fp16 baseline failed to compile (no lossless variant measured "
      "successfully), so lossy variants were skipped - without a lossless "
      "reference their accuracy cannot be validated.")

  # greedy per-weight int8: only when atol is loosened past the fp16-noise default.
  # Competes with global int8 (greedy wins when only SOME weights tolerate int8).
  if (atol > _ACCURACY_TOL and _has_weights(out) and baseline_out is not None
      and n_measured < budget):
    i8_nodes, i8_us, i8_n = _greedy_int8(
      out, inputs, baseline_out, baseline_us, reps=reps, atol=atol,
      min_lossy_speedup=min_lossy_speedup, budget=budget - n_measured,
      verbose=verbose)
    n_measured += i8_n
    if i8_nodes and i8_us < best_us:
      best_us = i8_us
      best_cfg = {"int8": False, "decomp": [], "int8_nodes": list(i8_nodes),
                  "lossy": True}
      if verbose:
        print(f"[tune] per-weight int8 wins: nodes {list(i8_nodes)} "
              f"({i8_us:.0f}us vs baseline {baseline_us:.0f}us)")

  if best_cfg is None:
    best_cfg = {"int8": False, "decomp": (), "lossy": False}  # fallback

  cache[key] = {"config": best_cfg, "us": (None if best_us == float("inf") else round(best_us, 1)),
                "shapes": [list(s) for s in input_shapes]}
  _save_cache(cache)

  if verbose:
    print(f"[tune] winner: {_config_label(best_cfg)} "
          f"({best_us if best_us != float('inf') else '?'} us); cached {key}")
  return build_variant(out, best_cfg)

tune_precision

tune_precision(out, target_error: float | None = None, cost_budget_us: float | None = None, inputs=None, reps: int = 20, verbose: bool = False)

Precision-aware tune: select the numerics-aware rewrite set under an explicit error or cost budget, returning (model, report).

Source code in aneforge/_optimize.py

def tune_precision(out, target_error: float | None = None, cost_budget_us: float | None = None,
                   inputs=None, reps: int = 20, verbose: bool = False):
  """Precision-aware tune: select the numerics-aware rewrite set under an explicit error or cost budget, returning (model, report)."""
  input_shapes = _input_shapes(out)
  if inputs is None: inputs = _gen_inputs(input_shapes)
  ref = _fp32_reference(out, inputs)
  ref_kind = "fp32" if ref is not None else None

  configs, risk = _precision_variants(out)
  rows = []
  for cfg in configs:
    est = _estimate_variant(out, cfg)
    us, relerr, out_arr = _measure_with_ref(out, inputs, cfg, ref, reps=reps)
    if ref_kind is None and cfg["label_kind"] == "fp16-baseline" and out_arr is not None:
      # no fp32 emulation: the fp16 baseline's own output becomes the reference (relerr 0.0).
      ref = np.asarray(out_arr, np.float64)
      ref_kind = "fp16-baseline"
      relerr = 0.0
    rows.append({"config": cfg, "label": cfg["label_kind"], "est_us": est,
                 "meas_us": us, "relerr": relerr, "ok": us != float("inf")})
    if verbose:
      print(f"[tune_precision] {cfg['label_kind']:22s} est {est:7.0f}us  "
            f"meas {us if us != float('inf') else 'FAIL':>9} "
            f"relerr {relerr:.3e}")

  usable = [r for r in rows if r["ok"]]
  if not usable:
    # nothing compiled - fall back to the fp16 baseline.
    model = build_variant(out, {"rs_matmul": [], "int8": False})
    return model, {"rows": rows, "risk": risk, "chosen": None,
                   "ref_available": ref is not None, "ref_kind": ref_kind}

  if ref_kind is None:
    # no reference: error-based selection is meaningless, so prefer the cheapest lossless variant.
    pool = [r for r in usable if not r["config"].get("lossy")] or usable
    chosen = min(pool, key=lambda r: r["est_us"])
    reason = ("NO accuracy reference (fp32 emulation unsupported for this graph; "
              "the fp16 baseline failed to run) - error budget NOT enforced; "
              "chose min-cost" + ("" if pool is usable else " lossless"))
    warnings.warn(f"tune_precision: {reason}")
  elif target_error is not None:
    meeting = [r for r in usable if r["relerr"] <= target_error]
    if meeting:
      chosen = min(meeting, key=lambda r: r["est_us"])   # cheapest meeting E
      reason = (f"min-cost meeting target_error={target_error:.1e} "
                f"(error vs {ref_kind} reference)")
    else:
      chosen = min(usable, key=lambda r: r["relerr"])    # none meet -> most accurate
      reason = (f"NO variant met target_error={target_error:.1e} vs the "
                f"{ref_kind} reference; chose most-accurate")
  elif cost_budget_us is not None:
    affordable = [r for r in usable if r["est_us"] <= cost_budget_us]
    pool = affordable or usable
    chosen = min(pool, key=lambda r: r["relerr"])
    reason = (f"min-error vs {ref_kind} reference within cost_budget={cost_budget_us:.0f}us"
              if affordable else
              f"NO variant under cost_budget={cost_budget_us:.0f}us; chose most-accurate")
  else:
    chosen = min(usable, key=lambda r: r["relerr"])
    reason = f"min-error vs {ref_kind} reference (no budget given)"

  if verbose:
    print(f"[tune_precision] CHOSE {chosen['label']} "
          f"(relerr {chosen['relerr']:.3e}, est {chosen['est_us']:.0f}us) - {reason}")
  model = build_variant(out, chosen["config"])
  return model, {"rows": rows, "risk": risk, "chosen": chosen, "reason": reason,
                 "ref_available": ref is not None, "ref_kind": ref_kind}

estimate

estimate(out, int8: bool = False, target: str | None = None) -> float

Estimate the compiled latency (us) of the graph rooted at out; target switches to the analytic per-chip model.

Source code in aneforge/_cost.py

def estimate(out, int8: bool = False, target: str | None = None) -> float:
  """Estimate the compiled latency (us) of the graph rooted at `out`; `target` switches to the analytic per-chip model."""
  if target is not None: return _estimate_analytic(out, target, int8)
  c = _constants()
  order = _topo(out)
  nodes = [t for t in order if t.op != "input"]
  cut_nodes = [t for t in nodes if t.op in NETPLIST_OPS]
  region_nodes = [t for t in nodes if t.op not in NETPLIST_OPS]

  floor = c["floor_us"]

  def _ncost(t) -> float:
    if int8 and t.op in ("matmul",):
      # int8 halves the weight bytes for the dominant projection weights
      in_elems = sum(_elems(s.shape) for s in t.srcs)
      wbytes = _weight_elems(t) * 1.0   # int8 = 1 byte/elem
      bytes_moved = (in_elems + _elems(t.shape)) * 2.0 + wbytes
      flops = _node_flops(t)
      return max(floor, bytes_moved / c["bw_bytes_per_us"], flops / c["flops_per_us"])
    return node_cost(t)

  # int8 tie-breaker: a tiny weight-byte-proportional discount so a floor-bound graph
  # predicts int8 <= fp16; kept below a floor's worth.
  int8_discount = 0.0
  if int8:
    saved_bytes = sum(_weight_elems(t) for t in region_nodes if t.op == "matmul")
    int8_discount = min(0.49 * floor, saved_bytes / c["bw_bytes_per_us"] * 0.5)

  if not cut_nodes:
    # one fused program: one floor + each node's above-floor work
    region = sum(max(0.0, _ncost(t) - floor) for t in region_nodes)
    return floor + region - int8_discount

  # segmented: fused regions (each pays one floor) interleaved with cuts; region count ~ (n_cuts + 1).
  n_cuts = len(cut_nodes)
  n_regions = (n_cuts + 1) if region_nodes else 0
  region_work = sum(max(0.0, _ncost(t) - floor) for t in region_nodes)
  # bridge node cost: measured per-family lookup, else generic roofline.
  def _bridge_or_roofline(t) -> float:
    bc = bridge_cost(t)
    return bc if bc is not None else _ncost(t)
  cut_work = sum(_bridge_or_roofline(t) for t in cut_nodes)
  return n_regions * floor + region_work + cut_work + n_cuts * c["cut_us"] - int8_discount

estimate_provenance

estimate_provenance(target: str) -> dict

Is estimate(out, target=...) silicon-anchored or extrapolated for target?

Source code in aneforge/_cost.py

def estimate_provenance(target: str) -> dict:
  """Is `estimate(out, target=...)` silicon-anchored or extrapolated for `target`?"""
  from . import _targets as _TG
  key = target.strip().lower()
  if key not in _TG._ARCH_FAMILY:
    raise ValueError(f"unknown ANE target arch {target!r}; known: "
                     f"{sorted(_TG._ARCH_FAMILY)}")
  anchor = _anchor_for_arch(key)
  measured_families = {int(_TG.family_of_arch(a)) for a in _ANCHORS}
  measured = int(_TG.family_of_arch(key)) in measured_families
  return {
    "target": key,
    "anchor": anchor,
    "measured": measured,
    "basis": "silicon" if measured else f"extrapolated-from-{anchor}",
  }

project_peak

project_peak(arch: str) -> dict

Measurement-free fp16 peak projection for any ANE target, anchored to measured M1.

Source code in aneforge/_cost.py

def project_peak(arch: str) -> dict:
  """Measurement-free fp16 peak projection for any ANE target, anchored to measured M1."""
  scale = _compute_scale(arch)
  c = _curve_for_arch(arch)
  return {
    "tflops": _M1_MEASURED_PEAK_TFLOPS * scale,
    "rel_m1": scale,
    "cores": int(c["cores_0x238"]),
    "ghz": _CLOCK_FRACTION * max(c["freq_0x760"]) / 1e9,
  }

precision_risk

precision_risk(out, verbose: bool = False) -> dict

Heuristic fp16-cancellation risk for the graph rooted at out: {graph_error, nodes, hotspots}.

Source code in aneforge/_cost.py

def precision_risk(out, verbose: bool = False) -> dict:
  """Heuristic fp16-cancellation risk for the graph rooted at `out`: {graph_error, nodes, hotspots}."""
  order = _topo(out)
  nodes = []
  for i, t in enumerate(order):
    # (a) narrow-accumulator signed reduce_sum
    if t.op == "reduce_sum":
      K = _reduce_len(t)
      signed = (not t.srcs) or _is_signed_producer(t.srcs[0])
      if signed and K >= _NARROW_SUM_FLOOR:
        # error grows ~ sqrt(K) (proxy, not a bound), cap 1.0.
        est = min(1.0, _FP16_CLEAN * (K ** 0.5))
        nodes.append({"idx": i, "op": t.op, "kind": "narrow_sum",
                      "est_error": est, "fixable": "reduce_sum->matmul",
                      "reason": f"signed reduce_sum over K={K} (narrow fp16 accumulator)"})
      continue
    # (b) CFG-style subtract - candidate cancellation (data-dependent). Flag sub of two activations.
    if t.op == "sub" and len(t.srcs) == 2:
      big = _elems(t.shape) >= 64    # a vector/tensor sub (not a scalar bias)
      both_live = all(s.op not in ("muls",) for s in t.srcs)
      if big and both_live:
        nodes.append({"idx": i, "op": t.op, "kind": "cancel_sub",
                      "est_error": _FP16_CLEAN,  # only RISKS blowing up; unknown w/o data
                      "fixable": "paired-fp16",
                      "reason": "subtract of two live tensors (CANDIDATE catastrophic "
                                "cancellation - confirm with data; fix is upstream paired-fp16)"})
      continue
    # (c) group_norm per-axis wall: rank-4 tiling cliff at max(C/groups, H*W) > 65536.
    if t.op == "group_norm" and len(t.shape) == 4:
      _, C, H, W = t.shape
      groups = int(t.attrs.get("groups", 1)) or 1
      if max(C // groups, H * W) > 65536:
        nodes.append({"idx": i, "op": t.op, "kind": "groupnorm_cliff",
                      "est_error": 0.0, "fixable": "",
                      "reason": f"group_norm tiled axis max(C/groups,H*W)>65536 at {H}x{W}: "
                                "AVOID - exceeds the ANE per-axis bound"})
      continue

  # Default hotspots = reliable structural signals only (narrow reduce_sum + group_norm wall);
  # cancel_sub is speculative -> informational only.
  hotspots = [n["idx"] for n in nodes if n["est_error"] > _FP16_CLEAN
              or n["kind"] == "groupnorm_cliff"]
  graph_error = max([_FP16_CLEAN] + [n["est_error"] for n in nodes])
  if verbose:
    print(f"[precision] graph_error~{graph_error:.2e}, {len(nodes)} flagged node(s):")
    for n in nodes:
      print(f"  node {n['idx']:3d} {n['op']:12s} kind={n['kind']:16s} "
            f"est~{n['est_error']:.2e} fix={n['fixable'] or '(avoid)'}: {n['reason']}")
  return {"graph_error": graph_error, "nodes": nodes, "hotspots": hotspots}

reset_compile_breaker

reset_compile_breaker() -> None

Clear the backoff state.

Source code in aneforge/_circuit.py

def reset() -> None:
  """Clear the backoff state."""
  global _last_failure_ts
  with _lock: _last_failure_ts = None