Compare Frontier fidelity across real trials
This commit is contained in:
@@ -52,6 +52,7 @@ PROTOCOL_SCHEMA = "qwen235b-prefill-fixed-cohort-protocol-v1"
|
||||
FRONTIER_SCHEMA = "frontier-qwen235b-prefill-fixed-cohort-v1"
|
||||
COMMUNITY_SCHEMA = "community-qwen235b-prefill-fixed-cohort-v1"
|
||||
COMPARISON_SCHEMA = "frontier-community-qwen235b-rank-comparison-v1"
|
||||
TRIAL_STABILITY_SCHEMA = "community-qwen235b-trial-stability-v1"
|
||||
REPEAT_SELECTION_SCHEMA = "community-qwen235b-boundary-repeat-selection-v4"
|
||||
COHORT_SIZE = 64
|
||||
COHORT_SEED = 2026071501
|
||||
@@ -1793,6 +1794,304 @@ def compare(args: argparse.Namespace) -> Path:
|
||||
return args.output
|
||||
|
||||
|
||||
def _index_loads(
|
||||
config_results: list[dict[str, Any]],
|
||||
) -> dict[tuple[str, float], dict[str, Any]]:
|
||||
return {
|
||||
(config["config"]["name"], float(load["offered_request_rate"])): load
|
||||
for config in config_results
|
||||
for load in config["loads"]
|
||||
}
|
||||
|
||||
|
||||
def _index_requests(
|
||||
config_results: list[dict[str, Any]],
|
||||
) -> dict[tuple[str, float, str], dict[str, Any]]:
|
||||
return {
|
||||
(
|
||||
config["config"]["name"],
|
||||
float(load["offered_request_rate"]),
|
||||
str(request["request_id"]),
|
||||
): request
|
||||
for config in config_results
|
||||
for load in config["loads"]
|
||||
for request in load["requests"]
|
||||
}
|
||||
|
||||
|
||||
def _request_passes_slo(request: dict[str, Any], variant: dict[str, Any]) -> bool:
|
||||
return bool(
|
||||
request["success"]
|
||||
and request.get("ttft_ms") is not None
|
||||
and float(request["ttft_ms"])
|
||||
<= slo_threshold_ms(variant, int(request["input_tokens"]))
|
||||
)
|
||||
|
||||
|
||||
def compare_trials(args: argparse.Namespace) -> Path:
|
||||
frontier = json.loads(args.frontier_freeze.read_text())
|
||||
first = json.loads(args.first_community_freeze.read_text())
|
||||
repeat = json.loads(args.repeat_community_freeze.read_text())
|
||||
if frontier.get("schema") != FRONTIER_SCHEMA:
|
||||
raise ValueError("unexpected Frontier freeze")
|
||||
for label, payload in (("first", first), ("repeat", repeat)):
|
||||
if payload.get("schema") != COMMUNITY_SCHEMA or payload.get("status") != "completed":
|
||||
raise ValueError(f"unexpected/incomplete {label} community freeze")
|
||||
|
||||
frontier_loads = _index_loads(frontier["config_results"])
|
||||
first_loads = _index_loads(first["config_results"])
|
||||
repeat_loads = _index_loads(repeat["config_results"])
|
||||
shared_load_keys = sorted(set(frontier_loads) & set(first_loads) & set(repeat_loads))
|
||||
if not shared_load_keys:
|
||||
raise ValueError("trials have no shared config/load points")
|
||||
for key in shared_load_keys:
|
||||
trace_hashes = {
|
||||
index[key].get("trace_sha256")
|
||||
for index in (frontier_loads, first_loads, repeat_loads)
|
||||
if index[key].get("trace_sha256") is not None
|
||||
}
|
||||
if len(trace_hashes) > 1:
|
||||
raise ValueError(f"trace mismatch at {key}")
|
||||
|
||||
frontier_requests = _index_requests(frontier["config_results"])
|
||||
first_requests = _index_requests(first["config_results"])
|
||||
repeat_requests = _index_requests(repeat["config_results"])
|
||||
shared_request_keys = sorted(
|
||||
set(frontier_requests) & set(first_requests) & set(repeat_requests)
|
||||
)
|
||||
if not shared_request_keys:
|
||||
raise ValueError("trials have no shared requests")
|
||||
for key in shared_request_keys:
|
||||
lengths = {
|
||||
int(index[key]["input_tokens"])
|
||||
for index in (frontier_requests, first_requests, repeat_requests)
|
||||
}
|
||||
if len(lengths) != 1:
|
||||
raise ValueError(f"input-token mismatch at {key}")
|
||||
|
||||
trial_ttft_differences = []
|
||||
for key in shared_request_keys:
|
||||
first_request = first_requests[key]
|
||||
repeat_request = repeat_requests[key]
|
||||
if (
|
||||
first_request["success"]
|
||||
and repeat_request["success"]
|
||||
and first_request.get("ttft_ms") is not None
|
||||
and repeat_request.get("ttft_ms") is not None
|
||||
):
|
||||
trial_ttft_differences.append(
|
||||
float(repeat_request["ttft_ms"]) - float(first_request["ttft_ms"])
|
||||
)
|
||||
|
||||
variants = {}
|
||||
for slo_name, variant in SLO_VARIANTS.items():
|
||||
request_counts = {
|
||||
"matching_real_trial_labels": 0,
|
||||
"frontier_correct_first": 0,
|
||||
"frontier_correct_repeat": 0,
|
||||
"frontier_wrong_both_stable_real_label": 0,
|
||||
"real_trial_label_flips": 0,
|
||||
}
|
||||
first_failures = set()
|
||||
repeat_failures = set()
|
||||
for key in shared_request_keys:
|
||||
predicted = _request_passes_slo(frontier_requests[key], variant)
|
||||
first_observed = _request_passes_slo(first_requests[key], variant)
|
||||
repeat_observed = _request_passes_slo(repeat_requests[key], variant)
|
||||
request_counts["matching_real_trial_labels"] += int(
|
||||
first_observed == repeat_observed
|
||||
)
|
||||
request_counts["frontier_correct_first"] += int(predicted == first_observed)
|
||||
request_counts["frontier_correct_repeat"] += int(predicted == repeat_observed)
|
||||
request_counts["frontier_wrong_both_stable_real_label"] += int(
|
||||
first_observed == repeat_observed and predicted != first_observed
|
||||
)
|
||||
request_counts["real_trial_label_flips"] += int(
|
||||
first_observed != repeat_observed
|
||||
)
|
||||
if not first_observed:
|
||||
first_failures.add(key)
|
||||
if not repeat_observed:
|
||||
repeat_failures.add(key)
|
||||
|
||||
load_records = []
|
||||
matching_load_labels = 0
|
||||
frontier_correct_first_loads = 0
|
||||
frontier_correct_repeat_loads = 0
|
||||
for config_name, rate in shared_load_keys:
|
||||
predicted_score = frontier_loads[(config_name, rate)]["scores"][slo_name]
|
||||
first_score = first_loads[(config_name, rate)]["scores"][slo_name]
|
||||
repeat_score = repeat_loads[(config_name, rate)]["scores"][slo_name]
|
||||
predicted_label = bool(predicted_score["feasible"])
|
||||
first_label = bool(first_score["feasible"])
|
||||
repeat_label = bool(repeat_score["feasible"])
|
||||
matching_load_labels += int(first_label == repeat_label)
|
||||
frontier_correct_first_loads += int(predicted_label == first_label)
|
||||
frontier_correct_repeat_loads += int(predicted_label == repeat_label)
|
||||
load_records.append(
|
||||
{
|
||||
"config": config_name,
|
||||
"offered_request_rate": rate,
|
||||
"frontier_passed_request_count": int(
|
||||
predicted_score["passed_request_count"]
|
||||
),
|
||||
"first_passed_request_count": int(first_score["passed_request_count"]),
|
||||
"repeat_passed_request_count": int(
|
||||
repeat_score["passed_request_count"]
|
||||
),
|
||||
"frontier_feasible": predicted_label,
|
||||
"first_feasible": first_label,
|
||||
"repeat_feasible": repeat_label,
|
||||
}
|
||||
)
|
||||
failure_union = first_failures | repeat_failures
|
||||
variants[slo_name] = {
|
||||
"shared_requests": len(shared_request_keys),
|
||||
**request_counts,
|
||||
"real_trial_request_label_accuracy": (
|
||||
request_counts["matching_real_trial_labels"] / len(shared_request_keys)
|
||||
),
|
||||
"frontier_request_label_accuracy_first": (
|
||||
request_counts["frontier_correct_first"] / len(shared_request_keys)
|
||||
),
|
||||
"frontier_request_label_accuracy_repeat": (
|
||||
request_counts["frontier_correct_repeat"] / len(shared_request_keys)
|
||||
),
|
||||
"real_trial_failure_set_jaccard": (
|
||||
len(first_failures & repeat_failures) / len(failure_union)
|
||||
if failure_union
|
||||
else 1.0
|
||||
),
|
||||
"shared_config_load_points": len(shared_load_keys),
|
||||
"real_trial_load_label_accuracy": matching_load_labels / len(shared_load_keys),
|
||||
"frontier_load_label_accuracy_first": (
|
||||
frontier_correct_first_loads / len(shared_load_keys)
|
||||
),
|
||||
"frontier_load_label_accuracy_repeat": (
|
||||
frontier_correct_repeat_loads / len(shared_load_keys)
|
||||
),
|
||||
"config_load_records": load_records,
|
||||
}
|
||||
|
||||
first_primary_capacity = _capacity_map(
|
||||
first["rankings"][PRIMARY_SLO],
|
||||
field="maximum_tested_feasible_request_rate",
|
||||
)
|
||||
first_config = {
|
||||
item["config"]["name"]: item["config"] for item in first["config_results"]
|
||||
}
|
||||
decision_records = []
|
||||
repeat_capacity_records = []
|
||||
for config_name in sorted(first_primary_capacity):
|
||||
old_rates = sorted(
|
||||
rate for name, rate in first_loads if name == config_name
|
||||
)
|
||||
capacity = first_primary_capacity[config_name]
|
||||
if capacity is None:
|
||||
continue
|
||||
higher_rates = [rate for rate in old_rates if rate > capacity]
|
||||
next_rate = min(higher_rates) if higher_rates else None
|
||||
boundary_rates = [capacity] + ([next_rate] if next_rate is not None else [])
|
||||
boundary_repeated = all(
|
||||
(config_name, rate) in repeat_loads for rate in boundary_rates
|
||||
)
|
||||
capacity_still_feasible = bool(
|
||||
boundary_repeated
|
||||
and repeat_loads[(config_name, capacity)]["scores"][PRIMARY_SLO]["feasible"]
|
||||
)
|
||||
next_still_infeasible = bool(
|
||||
next_rate is None
|
||||
or (
|
||||
boundary_repeated
|
||||
and not repeat_loads[(config_name, next_rate)]["scores"][PRIMARY_SLO][
|
||||
"feasible"
|
||||
]
|
||||
)
|
||||
)
|
||||
stable = boundary_repeated and capacity_still_feasible and next_still_infeasible
|
||||
decision_records.append(
|
||||
{
|
||||
"config": config_name,
|
||||
"first_capacity": capacity,
|
||||
"first_next_tested_rate": next_rate,
|
||||
"boundary_repeated": boundary_repeated,
|
||||
"capacity_still_feasible": capacity_still_feasible,
|
||||
"next_rate_still_infeasible": next_still_infeasible,
|
||||
"stable_original_grid_capacity": stable,
|
||||
}
|
||||
)
|
||||
repeat_capacity = capacity if stable else None
|
||||
repeat_capacity_records.append(
|
||||
{
|
||||
"config": first_config[config_name],
|
||||
"maximum_tested_feasible_request_rate": repeat_capacity,
|
||||
"maximum_tested_feasible_request_rate_per_gpu": (
|
||||
repeat_capacity / int(first_config[config_name]["tp"])
|
||||
if repeat_capacity is not None
|
||||
else None
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
result = {
|
||||
"schema": TRIAL_STABILITY_SCHEMA,
|
||||
"created_unix_s": time.time(),
|
||||
"inputs": {
|
||||
"frontier_freeze": {
|
||||
"path": str(args.frontier_freeze.resolve()),
|
||||
"sha256": sha256(args.frontier_freeze),
|
||||
},
|
||||
"first_community_freeze": {
|
||||
"path": str(args.first_community_freeze.resolve()),
|
||||
"sha256": sha256(args.first_community_freeze),
|
||||
},
|
||||
"repeat_community_freeze": {
|
||||
"path": str(args.repeat_community_freeze.resolve()),
|
||||
"sha256": sha256(args.repeat_community_freeze),
|
||||
},
|
||||
},
|
||||
"shared_config_load_points": len(shared_load_keys),
|
||||
"shared_requests": len(shared_request_keys),
|
||||
"real_trial_ttft_ms": {
|
||||
"paired_successful_requests": len(trial_ttft_differences),
|
||||
"repeat_minus_first_mean": (
|
||||
statistics.fmean(trial_ttft_differences)
|
||||
if trial_ttft_differences
|
||||
else None
|
||||
),
|
||||
"mae": (
|
||||
statistics.fmean(abs(value) for value in trial_ttft_differences)
|
||||
if trial_ttft_differences
|
||||
else None
|
||||
),
|
||||
"absolute_error_p95": _percentile(
|
||||
[abs(value) for value in trial_ttft_differences], 0.95
|
||||
),
|
||||
},
|
||||
"variants": variants,
|
||||
"primary_original_grid_capacity_stability": {
|
||||
"stable_configs": sum(
|
||||
item["stable_original_grid_capacity"] for item in decision_records
|
||||
),
|
||||
"total_configs": len(decision_records),
|
||||
"records": decision_records,
|
||||
"per_gpu_efficiency": _objective_fidelity(
|
||||
first["rankings"][PRIMARY_SLO],
|
||||
repeat_capacity_records,
|
||||
field="maximum_tested_feasible_request_rate_per_gpu",
|
||||
),
|
||||
"single_replica_raw_capacity": _objective_fidelity(
|
||||
first["rankings"][PRIMARY_SLO],
|
||||
repeat_capacity_records,
|
||||
field="maximum_tested_feasible_request_rate",
|
||||
),
|
||||
},
|
||||
}
|
||||
write_json(args.output, result)
|
||||
print(args.output)
|
||||
return args.output
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
@@ -1851,6 +2150,12 @@ def parse_args() -> argparse.Namespace:
|
||||
compare_parser.add_argument("--frontier-freeze", type=Path, required=True)
|
||||
compare_parser.add_argument("--community-freeze", type=Path, required=True)
|
||||
compare_parser.add_argument("--output", type=Path, required=True)
|
||||
|
||||
trials = subparsers.add_parser("compare-trials")
|
||||
trials.add_argument("--frontier-freeze", type=Path, required=True)
|
||||
trials.add_argument("--first-community-freeze", type=Path, required=True)
|
||||
trials.add_argument("--repeat-community-freeze", type=Path, required=True)
|
||||
trials.add_argument("--output", type=Path, required=True)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
@@ -1874,6 +2179,8 @@ def main() -> None:
|
||||
prepare_community_repeat(args)
|
||||
elif args.command == "compare":
|
||||
compare(args)
|
||||
elif args.command == "compare-trials":
|
||||
compare_trials(args)
|
||||
else:
|
||||
raise AssertionError(args.command)
|
||||
|
||||
|
||||
@@ -338,3 +338,63 @@ def test_frontier_freeze_validation_uses_manifest_rates(tmp_path: Path) -> None:
|
||||
)
|
||||
validated = MODULE._validate_frontier_freeze(freeze_path, protocol_path)
|
||||
assert validated["protocol_manifest_sha256"] == MODULE.sha256(protocol_path)
|
||||
|
||||
|
||||
def test_compare_trials_separates_real_label_flips_from_stable_frontier_errors(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
config = {"name": "a", "tp": 4}
|
||||
|
||||
def config_result(ttfts: list[float]) -> dict[str, object]:
|
||||
requests = [
|
||||
{
|
||||
"request_id": str(index),
|
||||
"input_tokens": 0,
|
||||
"success": True,
|
||||
"ttft_ms": ttft,
|
||||
}
|
||||
for index, ttft in enumerate(ttfts)
|
||||
]
|
||||
load = {
|
||||
"offered_request_rate": 0.1,
|
||||
"requests": requests,
|
||||
**MODULE.score_requests(requests),
|
||||
}
|
||||
return {"config": config, "loads": [load]}
|
||||
|
||||
frontier_result = config_result([900.0, 1100.0])
|
||||
first_result = config_result([900.0, 900.0])
|
||||
repeat_result = config_result([900.0, 1100.0])
|
||||
|
||||
def freeze(path: Path, schema: str, result: dict[str, object]) -> Path:
|
||||
payload = {
|
||||
"schema": schema,
|
||||
"status": "completed",
|
||||
"config_results": [result],
|
||||
"rankings": {
|
||||
name: MODULE.rank_surface([result], slo_name=name)
|
||||
for name in MODULE.SLO_VARIANTS
|
||||
},
|
||||
}
|
||||
path.write_text(json.dumps(payload))
|
||||
return path
|
||||
|
||||
frontier_path = freeze(tmp_path / "frontier.json", MODULE.FRONTIER_SCHEMA, frontier_result)
|
||||
first_path = freeze(tmp_path / "first.json", MODULE.COMMUNITY_SCHEMA, first_result)
|
||||
repeat_path = freeze(tmp_path / "repeat.json", MODULE.COMMUNITY_SCHEMA, repeat_result)
|
||||
output = tmp_path / "stability.json"
|
||||
MODULE.compare_trials(
|
||||
argparse.Namespace(
|
||||
frontier_freeze=frontier_path,
|
||||
first_community_freeze=first_path,
|
||||
repeat_community_freeze=repeat_path,
|
||||
output=output,
|
||||
)
|
||||
)
|
||||
result = json.loads(output.read_text())
|
||||
primary = result["variants"][MODULE.PRIMARY_SLO]
|
||||
assert primary["shared_requests"] == 2
|
||||
assert primary["real_trial_label_flips"] == 1
|
||||
assert primary["frontier_wrong_both_stable_real_label"] == 0
|
||||
assert primary["real_trial_load_label_accuracy"] == 0.0
|
||||
assert result["primary_original_grid_capacity_stability"]["stable_configs"] == 0
|
||||
|
||||
Reference in New Issue
Block a user