42 lines
2.0 KiB
Python
42 lines
2.0 KiB
Python
"""Bound unproven sibling routes while awaiting actual validation results."""
|
|
from collections import Counter
|
|
from typing import Any
|
|
|
|
|
|
def profile_key(row: dict[str, Any]) -> tuple[str, ...] | None:
|
|
profile = row.get("preflightMetadata") or row.get("modelProfile") or {}
|
|
model_type = str(profile.get("modelType") or "").strip()
|
|
if not model_type:
|
|
return None
|
|
return (str(row.get("targetGpu") or row.get("gpuType") or ""),
|
|
str(row.get("framework") or ""), str(row.get("taskType") or ""),
|
|
model_type, str(profile.get("quantizationSignature") or profile.get("quantizationMethod") or "none"))
|
|
|
|
|
|
def filter_profile_batch(candidates, pending, stats, *, limit=2):
|
|
"""Reservations include the whole proposed batch; deferred rows are retryable.
|
|
|
|
Broad public framework statistics are not proof for a specific encoding.
|
|
Exact-encoding successes can lift the cap, provided the observed success
|
|
rate remains at least 50%. Missing legacy signatures cannot grant this.
|
|
"""
|
|
counts = Counter(profile_key(row) for row in pending if row.get("outcome") == "pending")
|
|
selected, deferred = [], []
|
|
for candidate in candidates:
|
|
key = profile_key(candidate)
|
|
evidence = (stats.get("encodingCombinationStats") or {}).get("|".join(key or ())) or {}
|
|
successes = int(evidence.get("successCount") or 0)
|
|
failures = max(0, int(evidence.get("failureCount") or 0) - int(evidence.get("platformFailureCount") or 0))
|
|
proven = successes >= 3 and successes / max(1, successes + failures) >= 0.5
|
|
# Count legacy records with the same method as well, conservatively.
|
|
legacy = 0
|
|
if key and ":" in key[-1]:
|
|
legacy = counts.get((*key[:-1], key[-1].split(":", 1)[0]), 0)
|
|
if key and not proven and counts[key] + legacy >= limit:
|
|
deferred.append(candidate)
|
|
continue
|
|
selected.append(candidate)
|
|
if key:
|
|
counts[key] += 1
|
|
return selected, deferred
|