diff --git a/README.md b/README.md index 96b2babf73..dd6de194ec 100644 --- a/README.md +++ b/README.md @@ -444,11 +444,18 @@ durable Git checkpoint per roughly 2,000 scanned rows. Raw rows are discarded after every aggregate merge. Transient API failure pauses at the current cursor for the next poll cycle; successful completion disables the scan permanently. +Version `2026.09.18.1` keeps the first six ranked GPUs as the preferred routes +for each model and only falls back to lower-ranked eligible GPUs when all six +preferred routes are unavailable. Candidate generation retains at most six +routes per model, preserving successful-route and queue ordering without +discarding capacity on lower-ranked GPUs when the preferred tranche cannot run +the model. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-v33 -m "ModelHub agent 2026.09.04.4" -git push origin main agent-v33 +git tag -a agent-v34 -m "ModelHub agent 2026.09.18.1" +git push origin main agent-v34 ``` diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index c583388831..3c5189061d 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -65,6 +65,7 @@ DEFAULT_RUNS_DIR = Path("runs") DEFAULT_LEDGER_PATH = Path("ledger/submissions.jsonl") ADAPTIVE_SCAN_MAX_MODELS = 3000 ADAPTIVE_SCAN_MIN_FALLBACK_MODELS = 500 +MAX_ROUTES_PER_MODEL = 6 def build_parser() -> argparse.ArgumentParser: @@ -571,54 +572,74 @@ def process_model_for_candidates( skipped: list[dict[str, Any]] = [] failed: list[dict[str, Any]] = [] allowed_task_types_set = [spec.task_type for spec in specs] - pending_task_types_by_gpu: list[tuple[str, list[str]]] = [] + # Preserve the routing order: the first six GPUs are the preferred + # tranche. Lower-ranked GPUs are a fallback only when the preferred + # tranche produces no candidate for this model. This keeps queue/quality + # prioritization deterministic while still recovering from a model being + # already processed, too large, or unsupported on every preferred GPU. + preferred_target_gpus = target_gpus[:MAX_ROUTES_PER_MODEL] + fallback_target_gpus = target_gpus[MAX_ROUTES_PER_MODEL:] - route_target_gpus = target_gpus[:6] if official_registry is not None else target_gpus - 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( - {"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "model_gpu_uniqueness_blocklist"} - ) - continue - if target_gpu in processed_gpus: - skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"}) - 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 - - exact_task_types = list(allowed_task_types_set) - if official_registry is not None: - exact_task_types = official_registry.task_types_for( - modelhub_client, - model_address=model.model_address, - model_last_modified=model.last_modified.isoformat() if model.last_modified else None, - gpu=target_gpu, - ) - route_task_types = ( - list(exact_task_types) - if allow_dynamic_tasks - else [task_type for task_type in allowed_task_types_set if task_type in exact_task_types] - ) - compatible_task_types = [] - for task_type in route_task_types: - register_dynamic_task_route(task_type, model.pipeline_tag or task_type) - if market_intelligence is not None and not market_intelligence.has_framework_route(task_type, target_gpu): - try: - market_intelligence.ensure_framework_route(modelhub_client, task_type, target_gpu) - except Exception as exc: - raise OfficialCapabilityUnavailable( - f"framework capability unavailable for {task_type}/{target_gpu}: {type(exc).__name__}: {exc}" - ) from exc - if template_selector.supported_frameworks_for_auto(task_type, target_gpu): - compatible_task_types.append(task_type) + def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str]]]: + pending: list[tuple[str, list[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( + {"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "model_gpu_uniqueness_blocklist"} + ) continue - if market_intelligence is not None and market_intelligence.has_framework_route(task_type, target_gpu): - compatible_task_types.append(task_type) - if not compatible_task_types: - skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "no_compatible_auto_template_or_framework"}) - continue - pending_task_types_by_gpu.append((target_gpu, compatible_task_types)) + if target_gpu in processed_gpus: + skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"}) + 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 + + exact_task_types = list(allowed_task_types_set) + if official_registry is not None: + exact_task_types = official_registry.task_types_for( + modelhub_client, + model_address=model.model_address, + model_last_modified=model.last_modified.isoformat() if model.last_modified else None, + gpu=target_gpu, + ) + route_task_types = ( + list(exact_task_types) + if allow_dynamic_tasks + else [task_type for task_type in allowed_task_types_set if task_type in exact_task_types] + ) + compatible_task_types = [] + for task_type in route_task_types: + register_dynamic_task_route(task_type, model.pipeline_tag or task_type) + if market_intelligence is not None and not market_intelligence.has_framework_route(task_type, target_gpu): + try: + market_intelligence.ensure_framework_route(modelhub_client, task_type, target_gpu) + except Exception as exc: + raise OfficialCapabilityUnavailable( + f"framework capability unavailable for {task_type}/{target_gpu}: {type(exc).__name__}: {exc}" + ) from exc + if template_selector.supported_frameworks_for_auto(task_type, target_gpu): + compatible_task_types.append(task_type) + continue + if market_intelligence is not None and market_intelligence.has_framework_route(task_type, target_gpu): + compatible_task_types.append(task_type) + if not compatible_task_types: + skipped.append( + { + "repoId": model.repo_id, + "targetGpu": target_gpu, + "reason": "no_compatible_auto_template_or_framework", + } + ) + continue + pending.append((target_gpu, compatible_task_types)) + return pending + + pending_task_types_by_gpu = prepare_pending_routes(preferred_target_gpus) + used_fallback_routes = False + if not pending_task_types_by_gpu and fallback_target_gpus: + pending_task_types_by_gpu = prepare_pending_routes(fallback_target_gpus) + used_fallback_routes = True if not pending_task_types_by_gpu: return candidates, skipped, failed @@ -628,41 +649,56 @@ def process_model_for_candidates( except Exception as exc: return [], skipped, [{"repoId": model.repo_id, "reason": str(exc)}] - for target_gpu, task_types in pending_task_types_by_gpu: - best, preflight_reason = choose_candidate_for_gpu_detailed( - model=model, - inspection=inspection, - template_selector=template_selector, - task_types=task_types, - target_gpu=target_gpu, - market_intelligence=market_intelligence, - preflight_advisor=preflight_advisor, - require_official_config=official_registry is not None, - config_optimizer=config_optimizer, - ) - if best is None: - reason = ( - preflight_reason - or ( - "no_publicly_vetted_compatible_framework" - if market_intelligence is not None - else "no_compatible_auto_template_or_framework" + def evaluate_pending_routes(pending_routes: list[tuple[str, list[str]]]) -> int: + before_count = len(candidates) + for target_gpu, task_types in pending_routes: + if len(candidates) - before_count >= MAX_ROUTES_PER_MODEL: + break + best, preflight_reason = choose_candidate_for_gpu_detailed( + model=model, + inspection=inspection, + template_selector=template_selector, + task_types=task_types, + target_gpu=target_gpu, + market_intelligence=market_intelligence, + preflight_advisor=preflight_advisor, + require_official_config=official_registry is not None, + config_optimizer=config_optimizer, + ) + if best is None: + reason = ( + preflight_reason + or ( + "no_publicly_vetted_compatible_framework" + if market_intelligence is not None + else "no_compatible_auto_template_or_framework" + ) ) - ) - skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason}) - continue - record = candidate_to_record(best) - if outcome_tracker is not None: - record["transformersPrerequisiteSatisfied"] = ( - outcome_tracker.has_non_transformers_failure(model.repo_id) - ) - if market_intelligence is not None: - record.update(market_intelligence.gpu_metadata(target_gpu)) - record.update(market_intelligence.framework_metadata(best.task_type, target_gpu, best.framework)) - optimization = (record.get("preflightMetadata") or {}).get("configOptimization") or {} - if optimization.get("applied"): - record["frameworkConfigSource"] = "official_plus_learned_patch" - candidates.append(record) + skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason}) + continue + record = candidate_to_record(best) + if outcome_tracker is not None: + record["transformersPrerequisiteSatisfied"] = ( + outcome_tracker.has_non_transformers_failure(model.repo_id) + ) + if market_intelligence is not None: + record.update(market_intelligence.gpu_metadata(target_gpu)) + record.update(market_intelligence.framework_metadata(best.task_type, target_gpu, best.framework)) + optimization = (record.get("preflightMetadata") or {}).get("configOptimization") or {} + if optimization.get("applied"): + record["frameworkConfigSource"] = "official_plus_learned_patch" + candidates.append(record) + return len(candidates) - before_count + + evaluate_pending_routes(pending_task_types_by_gpu) + # If preferred GPUs had routes but every one failed candidate preflight, + # give the remaining eligible GPUs one chance. Once a + # preferred candidate exists, lower-ranked GPUs stay untouched for this + # model and remain available for later scans. + if not candidates and not used_fallback_routes and fallback_target_gpus: + fallback_pending = prepare_pending_routes(fallback_target_gpus) + if fallback_pending: + evaluate_pending_routes(fallback_pending) return candidates, skipped, failed @@ -968,7 +1004,7 @@ def one_candidate_per_model(candidates: list[dict[str, Any]]) -> list[dict[str, def limit_routes_per_model( candidates: list[dict[str, Any]], *, - max_routes: int = 3, + max_routes: int = MAX_ROUTES_PER_MODEL, ) -> list[dict[str, Any]]: selected: list[dict[str, Any]] = [] counts: Counter[str] = Counter() @@ -1453,7 +1489,7 @@ def run_submission( config_optimizer=config_optimizer, allow_dynamic_tasks=not bool(getattr(args, "task_types", None)), ) - stage_candidates = limit_routes_per_model(stage_candidates, max_routes=3) + stage_candidates = limit_routes_per_model(stage_candidates, max_routes=MAX_ROUTES_PER_MODEL) remaining_candidate_capacity = max(0, candidate_goal - current_candidate_models) stage_candidates = limit_candidate_models( stage_candidates, diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 9dcd3c893b..40335c5068 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.09.04.4" +AGENT_VERSION = "2026.09.18.1" diff --git a/tests/test_submission_safety.py b/tests/test_submission_safety.py index ac0cf340db..c3cfcc7cfd 100644 --- a/tests/test_submission_safety.py +++ b/tests/test_submission_safety.py @@ -68,6 +68,51 @@ class ExactGpuProcessedClient: return {"code": 0, "data": {"id": "should-not-submit"}} +class AllGpuOfficialRegistry: + @staticmethod + def task_types_for(_client, *, model_address: str, model_last_modified: str | None, gpu: str) -> list[str]: + del model_address, model_last_modified, gpu + return ["text-generation"] + + +class AllGpuVettedMarket: + @staticmethod + def has_framework_route(_task_type: str, _target_gpu: str) -> bool: + return True + + @staticmethod + def ensure_framework_route(_client, _task_type: str, _target_gpu: str) -> None: + return None + + @staticmethod + def selectable_frameworks( + *, task_type: str, target_gpu: str, incumbent_frameworks: list[str], inspection + ) -> list[str]: + del task_type, target_gpu, incumbent_frameworks, inspection + return ["vllm"] + + @staticmethod + def official_config( + *, task_type: str, target_gpu: str, framework: str, gguf_filename: str | None = None + ) -> str: + del task_type, target_gpu, framework, gguf_filename + return "framework: vllm" + + @staticmethod + def framework_metadata(_task_type: str, _target_gpu: str, _framework: str) -> dict: + return {"frameworkCombinedScore": 1.0, "frameworkOfficialConfigValid": True} + + @staticmethod + def gpu_metadata(_target_gpu: str) -> dict: + return {} + + +class SeventhGpuFallbackMarket(AllGpuVettedMarket): + @staticmethod + def has_framework_route(_task_type: str, target_gpu: str) -> bool: + return target_gpu == "gpu-7" + + class UniquenessRejectingClient: @staticmethod def add_task(_payload: dict) -> dict: @@ -107,6 +152,50 @@ def candidate(target_gpu: str = "Vastai_va16") -> dict: class SubmissionSafetyTests(unittest.TestCase): + def test_official_routes_prefer_first_six_gpus(self) -> None: + target_gpus = [f"gpu-{index}" for index in range(1, 8)] + model = HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + candidates, skipped, failed = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=OtherGpuProcessedClient(), # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=target_gpus, + allowed_task_types=["text-generation"], + market_intelligence=AllGpuVettedMarket(), # type: ignore[arg-type] + official_registry=AllGpuOfficialRegistry(), # type: ignore[arg-type] + ) + self.assertEqual(target_gpus[:6], [item["targetGpu"] for item in candidates]) + self.assertEqual([], skipped) + self.assertEqual([], failed) + + def test_official_routes_fall_back_after_preferred_six_are_unusable(self) -> None: + target_gpus = [f"gpu-{index}" for index in range(1, 8)] + model = HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + candidates, skipped, failed = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=OtherGpuProcessedClient(), # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=target_gpus, + allowed_task_types=["text-generation"], + market_intelligence=SeventhGpuFallbackMarket(), # type: ignore[arg-type] + official_registry=AllGpuOfficialRegistry(), # type: ignore[arg-type] + ) + self.assertEqual(["gpu-7"], [item["targetGpu"] for item in candidates]) + self.assertEqual(6, len(skipped)) + self.assertEqual([], failed) + def test_precheck_tracks_processed_gpus_without_blocking_other_gpus(self) -> None: payload = { "code": 0,