gahow/aituner
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add infeasible plateau guard to harness

Browse Source
This commit is contained in:
Gahow Wang
2026-04-25 18:47:32 +08:00
parent 6c04b9dbbc
commit 6bac389aae
3 changed files with 324 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
src/aituner/harness.py
View File

@@ -267,20 +267,26 @@ def _active_bottleneck(probe: dict[str, Any] | None) -> str:
if not compact:
return "unknown"
early_stop_reason = str(compact.get("early_stop_reason") or "")
if early_stop_reason.startswith("arrival_lag_s>") or early_stop_reason == "slo_pass_rate_unrecoverable":
if early_stop_reason.startswith("arrival_lag_s>"):
return "admission_or_queueing"
latency_summary = compact.get("latency_summary")
if not isinstance(latency_summary, dict):
if early_stop_reason == "slo_pass_rate_unrecoverable":
return "admission_or_queueing"
return "unknown"
failed = latency_summary.get("failed_reason_counts")
if not isinstance(failed, dict) or not failed:
if early_stop_reason == "slo_pass_rate_unrecoverable":
return "admission_or_queueing"
return "none_obvious"
ttft_count = sum(int(v) for k, v in failed.items() if str(k).startswith("ttft"))
tpot_count = sum(int(v) for k, v in failed.items() if str(k).startswith("tpot"))
request_failed_count = sum(
int(v)
for k, v in failed.items()
if not str(k).startswith("ttft") and not str(k).startswith("tpot")
if not str(k).startswith("ttft")
and not str(k).startswith("tpot")
and not str(k).startswith("probe_elapsed_s>")
)
if ttft_count >= max(tpot_count, request_failed_count):
return "ttft_prefill"
@@ -301,6 +307,7 @@ def _convergence_guard(
state: StudyState,
recent_diagnostics: list[dict[str, Any]],
) -> dict[str, Any]:
infeasible_progress = _infeasible_progress_guard(recent_diagnostics)
completed = [
item
for item in recent_diagnostics
@@ -320,9 +327,14 @@ def _convergence_guard(
reason = "recent_trials_still_show_material_variation"
elif state.best_request_rate_per_gpu is not None:
reason = "need_more_evidence_before_stop"
if not should_stop and infeasible_progress["plateau_detected"]:
reason = str(infeasible_progress["reason"])
return {
"should_stop_if_no_harness_can_justify_a_new_adjacent_probe": should_stop,
"should_stop_if_no_harness_can_justify_a_new_adjacent_probe": (
should_stop or bool(infeasible_progress["stop_if_next_probe_repeats_family"])
),
"reason": reason,
"infeasible_progress": infeasible_progress,
"incumbent": {
"trial_id": state.best_trial_id,
"parallel_size": state.best_parallel_size,
@@ -333,11 +345,136 @@ def _convergence_guard(
}
def _infeasible_progress_guard(recent_diagnostics: list[dict[str, Any]]) -> dict[str, Any]:
points = [
point
for item in recent_diagnostics
if (point := _infeasible_progress_point(item)) is not None
]
default = {
"plateau_detected": False,
"stop_if_next_probe_repeats_family": False,
"reason": "insufficient_infeasible_history",
"blocked_primary_family": None,
"recommended_next_action": "continue_adjacent_probe_if_a_harness_justifies_it",
"comparison": None,
}
if len(points) < 2:
return default
prev, latest = points[-2], points[-1]
if not _same_threshold(prev["threshold"], latest["threshold"]):
return {
**default,
"reason": "recent_infeasible_trials_used_different_sampling_thresholds",
}
family = _changed_primary_family(prev["config_patch"], latest["config_patch"])
pass_rate_delta = latest["pass_rate"] - prev["pass_rate"]
p95_delta_ratio = _relative_delta(latest["ttft_p95_ms"], prev["ttft_p95_ms"])
p99_delta_ratio = _relative_delta(latest["ttft_p99_ms"], prev["ttft_p99_ms"])
plateau = (
family is not None
and abs(pass_rate_delta) <= 0.01
and p95_delta_ratio is not None
and p95_delta_ratio >= -0.02
)
comparison = {
"previous": prev,
"latest": latest,
"changed_primary_family": family,
"pass_rate_delta": pass_rate_delta,
"ttft_p95_delta_ratio": p95_delta_ratio,
"ttft_p99_delta_ratio": p99_delta_ratio,
}
if not plateau:
return {
**default,
"reason": "recent_infeasible_trials_still_show_material_progress",
"comparison": comparison,
}
return {
"plateau_detected": True,
"stop_if_next_probe_repeats_family": True,
"reason": f"{family}_plateau_on_infeasible_trials",
"blocked_primary_family": family,
"recommended_next_action": (
"switch_primary_family_with_a_specific_bottleneck_rationale_or_return_should_stop"
),
"comparison": comparison,
}
def _infeasible_progress_point(item: dict[str, Any]) -> dict[str, Any] | None:
if item.get("status") != "completed" or item.get("best_request_rate") is not None:
return None
probe_summary = item.get("probe_summary")
if not isinstance(probe_summary, dict):
return None
probe = probe_summary.get("all_infeasible") or probe_summary.get("last_probe")
compact = _compact_probe(probe if isinstance(probe, dict) else None)
if not compact or compact.get("feasible"):
return None
latency_summary = compact.get("latency_summary")
if not isinstance(latency_summary, dict):
latency_summary = {}
ttft = latency_summary.get("ttft_ms")
if not isinstance(ttft, dict):
ttft = {}
return {
"trial_id": item.get("trial_id"),
"threshold": _as_float(compact.get("threshold")),
"request_rate": _as_float(compact.get("request_rate")),
"pass_rate": _as_float(compact.get("pass_rate")),
"ttft_p95_ms": _optional_float(ttft.get("p95")),
"ttft_p99_ms": _optional_float(ttft.get("p99")),
"active_bottleneck": item.get("active_bottleneck"),
"config_patch": item.get("config_patch") if isinstance(item.get("config_patch"), dict) else {},
}
def _changed_primary_family(
previous_patch: dict[str, Any],
latest_patch: dict[str, Any],
) -> str | None:
previous_flags = previous_patch.get("flag_patch") if isinstance(previous_patch, dict) else {}
latest_flags = latest_patch.get("flag_patch") if isinstance(latest_patch, dict) else {}
if not isinstance(previous_flags, dict) or not isinstance(latest_flags, dict):
return None
changed: list[str] = []
for key, family in (
("tensor-parallel-size", "tensor-parallel-size"),
("data-parallel-size", "data-parallel-size"),
("expert-parallel-size", "expert-parallel-size"),
("max-num-seqs", "max-num-seqs"),
("max-num-batched-tokens", "max-num-batched-tokens"),
("gpu-memory-utilization", "gpu-memory-utilization"),
("enable-chunked-prefill", "enable-chunked-prefill"),
):
if previous_flags.get(key) != latest_flags.get(key):
changed.append(family)
if len(changed) == 1:
return changed[0]
if changed:
return "mixed:" + ",".join(changed)
return None
def _same_threshold(left: float, right: float) -> bool:
return abs(left - right) <= max(abs(left), abs(right), 1.0) * 1e-6
def _relative_delta(new: float | None, old: float | None) -> float | None:
if new is None or old is None or old <= 0:
return None
return (new - old) / old
def _proposal_rules() -> list[str]:
return [
"First decide the active bottleneck from recent_trial_diagnostics.",
"Pick at most one primary knob family from knob_harnesses unless the history proves a coupled change is needed.",
"Use adjacent legal values around the incumbent; avoid broad exploratory jumps.",
"If infeasible_progress blocks the last primary knob family, do not continue that family; switch families with direct bottleneck evidence or set should_stop=true.",
"If a proposed config is likely to reduce request_rate_per_gpu under the active guard, set should_stop=true instead of exploring.",
"Never repeat an already tested config signature.",
"Return should_stop=true when the convergence guard fires and no active harness justifies another adjacent probe.",
@@ -388,3 +525,11 @@ def _as_float(value: Any) -> float:
if isinstance(value, (int, float)):
return float(value)
return 0.0
def _optional_float(value: Any) -> float | None:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value)
return None