fix: recover eligible model GPU routes

This commit is contained in:
CoolBoy
2026-09-20 10:47:22 +08:00
parent 3077b169c5
commit 75c034d588
13 changed files with 436 additions and 29 deletions

View File

@@ -10,7 +10,7 @@ from pathlib import Path
from typing import Any
from common import parse_datetime, runtime_instance_id, utc_now, write_json, write_jsonl
from candidate_preflight import CandidatePreflightAdvisor
from candidate_preflight import CandidatePreflightAdvisor, MODEL_LOAD_OVERHEAD
from config_optimizer import SafeConfigOptimizer
from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH
from hf_discovery import HuggingFaceDiscovery
@@ -375,6 +375,7 @@ def choose_candidate_for_gpu_detailed(
preflight_advisor: CandidatePreflightAdvisor | None = None,
require_official_config: bool = False,
config_optimizer: SafeConfigOptimizer | None = None,
excluded_frameworks: set[str] | None = None,
) -> tuple[CandidateModel | None, str | None]:
last_rejection_reason: str | None = None
for task_type in task_types:
@@ -397,6 +398,14 @@ def choose_candidate_for_gpu_detailed(
incumbent_frameworks=compatible_frameworks,
inspection=inspection,
)
if excluded_frameworks:
available_frameworks = [
framework for framework in compatible_frameworks
if framework not in excluded_frameworks
]
if compatible_frameworks and not available_frameworks:
last_rejection_reason = "prior_failed_framework_no_alternative"
compatible_frameworks = available_frameworks
if not compatible_frameworks:
continue
@@ -560,8 +569,10 @@ def process_model_for_candidates(
if hasattr(modelhub_client, "model_submission_precheck"):
precheck = modelhub_client.model_submission_precheck(model.repo_id)
processed_gpus = set(precheck.get("processedGpus") or set())
failed_frameworks_by_gpu = precheck.get("failedFrameworksByGpu") or {}
else:
processed_gpus = modelhub_client.processed_gpus_for_model(model.repo_id)
failed_frameworks_by_gpu = {}
except Exception as exc:
return (
[],
@@ -580,8 +591,8 @@ def process_model_for_candidates(
preferred_target_gpus = target_gpus[:MAX_ROUTES_PER_MODEL]
fallback_target_gpus = target_gpus[MAX_ROUTES_PER_MODEL:]
def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str]]]:
pending: list[tuple[str, list[str]]] = []
def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str], set[str]]]:
pending: list[tuple[str, list[str], set[str]]] = []
for target_gpu in route_target_gpus:
if submission_exclusion_store is not None and submission_exclusion_store.is_blocked(model.repo_id, target_gpu):
skipped.append(
@@ -591,9 +602,45 @@ def process_model_for_candidates(
if target_gpu in processed_gpus:
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"})
continue
if target_gpu in failed_frameworks_by_gpu and market_intelligence is None:
skipped.append({
"repoId": model.repo_id,
"targetGpu": target_gpu,
"reason": "prior_failed_gpu_requires_public_framework_evidence",
})
continue
failed_frameworks = set(failed_frameworks_by_gpu.get(target_gpu) or set())
if (
failed_frameworks
and market_intelligence is not None
and hasattr(market_intelligence, "has_vetted_alternative_framework")
and allowed_task_types_set
and all(market_intelligence.has_framework_route(task_type, target_gpu) for task_type in allowed_task_types_set)
and not any(
market_intelligence.has_vetted_alternative_framework(
task_type, target_gpu, failed_frameworks
)
for task_type in allowed_task_types_set
)
):
skipped.append({
"repoId": model.repo_id,
"targetGpu": target_gpu,
"reason": "prior_failed_framework_no_alternative",
})
continue
if outcome_tracker and outcome_tracker.is_model_gpu_failed(model.repo_id, target_gpu):
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"})
continue
if preflight_advisor is not None and model.file_size:
memory_gib = preflight_advisor.gpu_memory_gib.get(target_gpu)
if memory_gib and model.file_size / (1024**3) * MODEL_LOAD_OVERHEAD > memory_gib:
skipped.append({
"repoId": model.repo_id,
"targetGpu": target_gpu,
"reason": "preflight_predicted_oom",
})
continue
exact_task_types = list(allowed_task_types_set)
if official_registry is not None:
@@ -632,7 +679,11 @@ def process_model_for_candidates(
}
)
continue
pending.append((target_gpu, compatible_task_types))
pending.append((
target_gpu,
compatible_task_types,
set(failed_frameworks_by_gpu.get(target_gpu) or set()),
))
return pending
pending_task_types_by_gpu = prepare_pending_routes(preferred_target_gpus)
@@ -649,9 +700,9 @@ def process_model_for_candidates(
except Exception as exc:
return [], skipped, [{"repoId": model.repo_id, "reason": str(exc)}]
def evaluate_pending_routes(pending_routes: list[tuple[str, list[str]]]) -> int:
def evaluate_pending_routes(pending_routes: list[tuple[str, list[str], set[str]]]) -> int:
before_count = len(candidates)
for target_gpu, task_types in pending_routes:
for target_gpu, task_types, excluded_frameworks in pending_routes:
if len(candidates) - before_count >= MAX_ROUTES_PER_MODEL:
break
best, preflight_reason = choose_candidate_for_gpu_detailed(
@@ -664,6 +715,7 @@ def process_model_for_candidates(
preflight_advisor=preflight_advisor,
require_official_config=official_registry is not None,
config_optimizer=config_optimizer,
excluded_frameworks=excluded_frameworks,
)
if best is None:
reason = (
@@ -677,6 +729,8 @@ def process_model_for_candidates(
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason})
continue
record = candidate_to_record(best)
if excluded_frameworks:
record["previousFailedFrameworks"] = sorted(excluded_frameworks)
if outcome_tracker is not None:
record["transformersPrerequisiteSatisfied"] = (
outcome_tracker.has_non_transformers_failure(model.repo_id)
@@ -753,12 +807,29 @@ def build_adaptive_scan_stages(
"limit": max_models,
}
)
if allow_older_than_recent_window:
# ModelScope only exposes the first 3,000 raw rows for a sort order.
# Downloads ordering reaches a different older, established slice
# after the recent-update slice has produced too few candidates.
stages.append(
{
"name": "popular_history",
"updatedAfter": None,
"limit": min(max_models, 200),
"sort": "downloads",
"rotatePages": True,
}
)
deduped: list[dict[str, Any]] = []
seen: set[tuple[str | None, int]] = set()
seen: set[tuple[str | None, int, str]] = set()
for stage in stages:
updated_after = stage["updatedAfter"]
signature = (updated_after.isoformat() if updated_after is not None else None, int(stage["limit"]))
signature = (
updated_after.isoformat() if updated_after is not None else None,
int(stage["limit"]),
str(stage.get("sort") or "last_modified"),
)
if signature in seen:
continue
seen.add(signature)
@@ -877,6 +948,21 @@ def submit_candidate(
"candidate": candidate,
"reason": "already_processed_for_gpu",
}
failed_frameworks = set(
(precheck.get("failedFrameworksByGpu") or {}).get(candidate["targetGpu"]) or set()
)
if candidate["framework"] in failed_frameworks:
return {
"outcome": "duplicate",
"candidate": candidate,
"reason": "prior_failed_framework_for_gpu",
}
if failed_frameworks and not candidate.get("frameworkEvidenceQualified"):
return {
"outcome": "framework_evidence_deferred",
"candidate": candidate,
"reason": "prior_failed_gpu_requires_public_framework_evidence",
}
payload = {
"modelAddress": candidate["modelAddress"],
@@ -1442,6 +1528,10 @@ def run_submission(
"min_downloads": args.min_downloads,
"updated_after": stage_updated_after,
}
if stage.get("sort"):
query_kwargs["sort"] = stage["sort"]
if stage.get("rotatePages"):
query_kwargs["rotate_pages"] = True
try:
models = hf_discovery.list_recent_models(
**query_kwargs,
@@ -1501,6 +1591,7 @@ def run_submission(
scanned_model_count += processed_count
stage_summary = {
"name": stage["name"],
"sort": stage.get("sort") or "last_modified",
"updatedAfter": stage_updated_after.isoformat() if stage_updated_after is not None else None,
"requestedLimit": stage_limit,
"discoveredModels": len(models),
@@ -1648,10 +1739,20 @@ def run_submission(
{
"repoId": candidate["repoId"],
"targetGpu": candidate["targetGpu"],
"reason": "already_validating_on_platform",
"reason": result.get("reason") or "already_validating_on_platform",
}
)
continue
if result["outcome"] == "framework_evidence_deferred":
batch_policy_skipped_candidates.append(candidate)
skipped.append({
"repoId": candidate["repoId"],
"targetGpu": candidate["targetGpu"],
"framework": candidate["framework"],
"reason": result["reason"],
"deferred": True,
})
continue
if result["outcome"] == "age_policy_deferred":
batch_policy_skipped_candidates.append(candidate)
skipped.append(