diff --git a/README.md b/README.md index ec7f7873..75c3c00b 100644 --- a/README.md +++ b/README.md @@ -44,18 +44,50 @@ Optional tuning: - `MODELHUB_AGENT_GPUS` - `MODELHUB_AGENT_EXTRA_ARGS` - `MODELHUB_GPU_STRATEGY_STATE_PATH` default `.modelhub_state/gpu_strategy.json` +- `MODELHUB_MARKET_INTELLIGENCE_PATH` default `.modelhub_state/market_intelligence.json` +- `MODELHUB_MARKET_QUEUE_REFRESH_SECONDS` default `600` +- `MODELHUB_MARKET_FRAMEWORK_REFRESH_SECONDS` default `21600` +- `MODELHUB_MARKET_THROUGHPUT_WINDOW_HOURS` default `6` +- `MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES` default `100` - `MODELSCOPE_PAGE_INTERVAL_SECONDS` default `0.25` - `MODELSCOPE_PAGE_CACHE_TTL_SECONDS` default `900` - `MODELHUB_AGENT_VERIFY_CACHE_TTL_SECONDS` default `900` ## Adaptive GPU Strategy -When no explicit GPU override is supplied, the worker uses a local 50/30/20 +When no explicit GPU override is supplied, the worker uses a queue-aware 50/30/20 strategy generation: - 50%: the three long-term GPUs with the best Wilson lower confidence score and at least 100 terminal samples -- 30%: round-robin exploration across every currently supported GPU -- 20%: the best GPU among the latest 1,000 terminal tasks +- 30%: exploration across every currently supported GPU +- 20%: the top recent GPUs among the latest 1,000 terminal tasks + +The 50/30/20 category ratio remains exact across accepted tasks. Inside each +category, weighted fair scheduling combines the category's historical rank with +live public market data: + +- estimated backlog hours (`waiting / recent completions per hour`) instead of raw queue length +- recent public success quality, scored with a Wilson lower confidence bound +- machine availability, running workers, and advertised concurrency +- a circuit breaker for unavailable or apparently stalled GPU pools + +This optimizes expected successful completions rather than blindly selecting the +smallest queue. Queue/throughput data is refreshed every 10 minutes and persisted +in `.modelhub_state/market_intelligence.json`. A failed refresh keeps the last good +snapshot, uses a retry backoff, and never blocks normal submissions. + +For each compatible model/GPU pair, the worker also ranks the GPU's supported +frameworks using ModelHub's public aggregate `modelCount` and `successCount` data, +then blends in the worker's own GPU+framework outcomes with a capped weight. +Only frameworks with at least 100 samples receive statistical priority; the +legacy safe framework order remains the fallback. Framework statistics refresh +every 6 hours, so they do not add per-model API traffic. Newly published +frameworks are discovered automatically, but receive no novelty bonus: they can +win only when their confidence-adjusted success score is better. A new framework +is eligible only after the authenticated official build-config endpoint returns +a complete config that passes local structure, placeholder, framework-name, and +GPU-parallelism validation. Valid official configs are cached and refreshed with +the framework snapshot; local templates remain the fail-safe fallback. Only platform-accepted tasks count. After exactly 200 accepted tasks, the next poll cycle reloads all account history, generates a new immutable strategy snapshot, @@ -104,12 +136,15 @@ health response expose `agent_version`; version `2026.08.02.3` or newer includes duplicate replacement behavior, while version `2026.08.02.4` adds adaptive candidate-window expansion and skip-reason reporting. Version `2026.08.02.5` adds fail-closed model/GPU prechecks and persistent uniqueness exclusions. +Version `2026.08.04.1` adds queue/throughput intelligence, GPU health circuit +breaking, weighted-fair scheduling, live framework/config discovery, and +confidence-ranked public-plus-local framework selection. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag agent-v6 -git push origin agent-v6 +git tag agent-v12 +git push origin agent-v12 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index ca834e09..c5f59dae 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -9,6 +9,7 @@ It currently supports: - multiple ModelHub tokens read from `KEY.md` and `KEYS.md` - automatic task/framework/template selection across the supported GPU catalog - adaptive long-term/exploration/recent GPU scheduling with a persistent local snapshot +- live queue/throughput-aware GPU weighting and confidence-ranked framework selection ## Layout @@ -73,8 +74,19 @@ bash run_poll.sh --dry-run ## Behavior - The runner auto-discovers all safe GPU/template combinations from the public submit catalog. -- Automatic GPU selection uses smooth 50/30/20 scheduling: long-term Wilson-ranked top 3 GPUs, - all supported GPUs, and the best GPU from the latest 1,000 terminal tasks. +- Automatic GPU selection uses exact 50/30/20 accepted-task scheduling: long-term + Wilson-ranked top 3 GPUs, all supported GPUs, and the top GPUs from the latest + 1,000 terminal tasks. +- Within each category, weighted-fair scheduling uses estimated queue backlog hours, + recent public throughput/success, machine availability, and worker concurrency. + Unavailable or stalled GPU pools are circuit-broken instead of continuing to absorb work. +- Compatible frameworks are ranked by ModelHub public aggregate success statistics + plus capped local GPU+framework evidence, with a 100-sample public minimum and + Wilson confidence bounds. The legacy safe order is retained whenever evidence + is missing or too small. +- New frameworks are discovered from the live catalog but get no novelty bonus. + They are eligible only with a complete official build config that passes local + validation; cached local templates remain the fallback if live config sync fails. - A strategy generation lasts exactly 200 platform-accepted submissions. Rejected API calls and duplicates do not advance it. The next cycle refreshes platform history before submitting again. - Strategy state is stored in `.modelhub_state/gpu_strategy.json`; a generation never recalculates @@ -116,6 +128,7 @@ Common flags: - `--dry-run`: plan only, do not submit - `--gpu-strategy-refresh-submissions`: accepted tasks per strategy generation (default `200`) - `--disable-gpu-strategy`: restore legacy ordering; explicit `--gpu/--gpus` also bypasses adaptive selection +- `--disable-market-intelligence`: disable live queue/throughput and framework-stat weighting - `run_daily.sh` injects `--daily-target 3` when no daily-target flag is provided. Set `SUBMIT_DAILY_TARGET` or pass `--daily-target` explicitly for a different target. `run_poll.sh` adds: @@ -151,6 +164,7 @@ Each run typically includes: Persistent local scheduler state is written under `.modelhub_state/`: - `gpu_strategy.json`: GPU ranks, generation progress, and 50/30/20 accepted counters +- `market_intelligence.json`: cached public queue, throughput, health, and framework statistics - `account_capacity.json`: learned per-account active-task limits - `submission_exclusions.jsonl`: non-retryable model/GPU uniqueness rejections diff --git a/modelhub_submmit_api/daily_runner.py b/modelhub_submmit_api/daily_runner.py index 77cfba9c..7b6346c0 100644 --- a/modelhub_submmit_api/daily_runner.py +++ b/modelhub_submmit_api/daily_runner.py @@ -10,6 +10,14 @@ from common import utc_now, write_json from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH from hf_discovery import HuggingFaceDiscovery from main import DEFAULT_LEDGER_PATH, DEFAULT_RUNS_DIR, make_run_dir, run_submission +from market_intelligence import ( + DEFAULT_FETCH_WORKERS, + DEFAULT_FRAMEWORK_MIN_SAMPLES, + DEFAULT_FRAMEWORK_REFRESH_SECONDS, + DEFAULT_MARKET_INTELLIGENCE_PATH, + DEFAULT_QUEUE_REFRESH_SECONDS, + DEFAULT_THROUGHPUT_WINDOW_HOURS, +) from modelhub_client import DEFAULT_CAPACITY_STATE_PATH, ModelHubClient, ModelHubClientPool from outcome_tracker import OutcomeTracker from runner_common import DEFAULT_KEY_PATH, ensure_tokens @@ -81,6 +89,11 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--skip-history-archive", action="store_true", help="Skip historical task archive download for this run") parser.add_argument("--dry-run", action="store_true", help="Plan the day without creating tasks") parser.add_argument("--disable-gpu-strategy", action="store_true", help="Disable adaptive 50/30/20 GPU scheduling") + parser.add_argument( + "--disable-market-intelligence", + action="store_true", + help="Disable live queue/throughput and public framework statistics", + ) parser.add_argument( "--gpu-strategy-refresh-submissions", type=int, @@ -109,6 +122,41 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--gpu-strategy-recent-window", type=int, default=1000, help=argparse.SUPPRESS) parser.add_argument("--gpu-strategy-min-long-samples", type=int, default=100, help=argparse.SUPPRESS) + parser.add_argument( + "--market-intelligence-state-path", + default=os.getenv("MODELHUB_MARKET_INTELLIGENCE_PATH", str(DEFAULT_MARKET_INTELLIGENCE_PATH)), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-queue-refresh-seconds", + type=int, + default=int(os.getenv("MODELHUB_MARKET_QUEUE_REFRESH_SECONDS", str(DEFAULT_QUEUE_REFRESH_SECONDS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-framework-refresh-seconds", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_REFRESH_SECONDS", str(DEFAULT_FRAMEWORK_REFRESH_SECONDS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-throughput-window-hours", + type=int, + default=int(os.getenv("MODELHUB_MARKET_THROUGHPUT_WINDOW_HOURS", str(DEFAULT_THROUGHPUT_WINDOW_HOURS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-fetch-workers", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FETCH_WORKERS", str(DEFAULT_FETCH_WORKERS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-framework-min-samples", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES", str(DEFAULT_FRAMEWORK_MIN_SAMPLES))), + help=argparse.SUPPRESS, + ) parser.add_argument( "--capacity-state-path", default=os.getenv("MODELHUB_CAPACITY_STATE_PATH", str(DEFAULT_CAPACITY_STATE_PATH)), @@ -150,10 +198,37 @@ def make_wave_namespace(base_args: argparse.Namespace, wave: WaveSpec) -> argpar skip_outcome_sync=getattr(base_args, "skip_outcome_sync", False), skip_history_archive=getattr(base_args, "skip_history_archive", False), disable_gpu_strategy=getattr(base_args, "disable_gpu_strategy", False), + disable_market_intelligence=getattr(base_args, "disable_market_intelligence", False), gpu_strategy_refresh_submissions=getattr(base_args, "gpu_strategy_refresh_submissions", 200), gpu_strategy_state_path=getattr(base_args, "gpu_strategy_state_path", str(DEFAULT_GPU_STRATEGY_PATH)), gpu_strategy_recent_window=getattr(base_args, "gpu_strategy_recent_window", 1000), gpu_strategy_min_long_samples=getattr(base_args, "gpu_strategy_min_long_samples", 100), + market_intelligence_state_path=getattr( + base_args, + "market_intelligence_state_path", + str(DEFAULT_MARKET_INTELLIGENCE_PATH), + ), + market_queue_refresh_seconds=getattr( + base_args, + "market_queue_refresh_seconds", + DEFAULT_QUEUE_REFRESH_SECONDS, + ), + market_framework_refresh_seconds=getattr( + base_args, + "market_framework_refresh_seconds", + DEFAULT_FRAMEWORK_REFRESH_SECONDS, + ), + market_throughput_window_hours=getattr( + base_args, + "market_throughput_window_hours", + DEFAULT_THROUGHPUT_WINDOW_HOURS, + ), + market_fetch_workers=getattr(base_args, "market_fetch_workers", DEFAULT_FETCH_WORKERS), + market_framework_min_samples=getattr( + base_args, + "market_framework_min_samples", + DEFAULT_FRAMEWORK_MIN_SAMPLES, + ), capacity_state_path=getattr(base_args, "capacity_state_path", str(DEFAULT_CAPACITY_STATE_PATH)), capacity_probe_interval_cycles=getattr(base_args, "capacity_probe_interval_cycles", 3), submit_concurrency=getattr(base_args, "submit_concurrency", 1), diff --git a/modelhub_submmit_api/gpu_strategy.py b/modelhub_submmit_api/gpu_strategy.py index 6bd8463e..b8e70cf9 100644 --- a/modelhub_submmit_api/gpu_strategy.py +++ b/modelhub_submmit_api/gpu_strategy.py @@ -17,7 +17,7 @@ DEFAULT_GPU_STRATEGY_PATH = Path(".modelhub_state/gpu_strategy.json") DEFAULT_REFRESH_SUBMISSIONS = 200 DEFAULT_RECENT_TERMINAL_WINDOW = 1000 DEFAULT_LONG_TERM_MIN_SAMPLES = 100 -STRATEGY_STATE_VERSION = 1 +STRATEGY_STATE_VERSION = 2 LONG_TERM = "long_term" ALL_SUPPORTED = "all_supported" @@ -25,14 +25,17 @@ RECENT = "recent" CATEGORIES = (LONG_TERM, ALL_SUPPORTED, RECENT) CATEGORY_WEIGHTS = {LONG_TERM: 5, ALL_SUPPORTED: 3, RECENT: 2} -# The long-term half is split 50/30/20 across its three ranked GPUs. -LONG_TERM_GPU_PATTERN = (0, 1, 0, 2, 0, 1, 0, 1, 0, 2) - - def _empty_category_counts() -> dict[str, int]: return {category: 0 for category in CATEGORIES} +def _empty_gpu_category_counts(supported_gpus: list[str]) -> dict[str, dict[str, int]]: + return { + category: {gpu: 0 for gpu in supported_gpus} + for category in CATEGORIES + } + + def _gpu_name(record: dict[str, Any]) -> str: return str(record.get("gpuType") or record.get("targetGpu") or "").strip() @@ -160,12 +163,11 @@ def build_strategy_snapshot( recent = _summarize_gpu_records(recent_tasks, supported_gpus) recent_ranked = _rank_gpu_summaries(recent) recent_qualified = [item for item in recent_ranked if int(item["terminal"]) >= 20] - if recent_qualified: - recent_gpu = str(recent_qualified[0]["gpu"]) - elif recent_tasks and recent_ranked: - recent_gpu = str(recent_ranked[0]["gpu"]) - else: - recent_gpu = long_term_gpus[0] + recent_source = recent_qualified if recent_qualified else (recent_ranked if recent_tasks else []) + recent_gpus = [str(item["gpu"]) for item in recent_source[: min(3, len(supported_gpus))]] + if not recent_gpus: + recent_gpus = list(long_term_gpus[: min(3, len(supported_gpus))]) + recent_gpu = recent_gpus[0] return { "version": STRATEGY_STATE_VERSION, @@ -176,6 +178,7 @@ def build_strategy_snapshot( "supportedGpus": supported_gpus, "longTermGpus": long_term_gpus, "recentGpu": recent_gpu, + "recentGpus": recent_gpus, "recentTerminalCount": len(recent_tasks), "refreshSubmissions": max(1, int(refresh_submissions)), "recentTerminalWindow": max(1, int(recent_terminal_window)), @@ -183,6 +186,7 @@ def build_strategy_snapshot( "acceptedSinceRefresh": 0, "acceptedTotal": 0, "acceptedByCategory": _empty_category_counts(), + "acceptedByGpuCategory": _empty_gpu_category_counts(supported_gpus), "longTermStats": all_time_ranked, "recentStats": recent_ranked, "lastRefreshError": None, @@ -208,15 +212,20 @@ class GPUStrategyManager: refresh_submissions: int = DEFAULT_REFRESH_SUBMISSIONS, recent_terminal_window: int = DEFAULT_RECENT_TERMINAL_WINDOW, long_term_min_samples: int = DEFAULT_LONG_TERM_MIN_SAMPLES, + market_intelligence: Any | None = None, log_fn: Callable[[str], None] | None = None, ) -> None: self.path = Path(path) self.refresh_submissions = max(1, int(refresh_submissions)) self.recent_terminal_window = max(1, int(recent_terminal_window)) self.long_term_min_samples = max(1, int(long_term_min_samples)) + self.market_intelligence = market_intelligence self.log = log_fn or (lambda message: print(message, flush=True)) self.state: dict[str, Any] | None = None + def set_market_intelligence(self, market_intelligence: Any | None) -> None: + self.market_intelligence = market_intelligence + def _load(self) -> dict[str, Any] | None: try: value = read_json(self.path) @@ -357,22 +366,49 @@ class GPUStrategyManager: accepted = max(0, int(self.state.get("acceptedSinceRefresh") or 0)) return max(0, refresh_every - accepted) - def _category_gpu_order(self, category: str, occurrence: int) -> list[str]: + def _category_gpu_order(self, category: str, gpu_counts: dict[str, int]) -> list[str]: assert self.state is not None supported = list(self.state.get("supportedGpus") or []) long_term = [gpu for gpu in self.state.get("longTermGpus") or [] if gpu in supported] - recent_gpu = str(self.state.get("recentGpu") or "") + recent = [gpu for gpu in self.state.get("recentGpus") or [] if gpu in supported] + if not recent: + recent_gpu = str(self.state.get("recentGpu") or "") + if recent_gpu in supported: + recent = [recent_gpu] - if category == LONG_TERM and long_term: - desired_index = LONG_TERM_GPU_PATTERN[occurrence % len(LONG_TERM_GPU_PATTERN)] - desired = long_term[min(desired_index, len(long_term) - 1)] - return [desired, *(gpu for gpu in long_term if gpu != desired)] - if category == RECENT and recent_gpu in supported: - return [recent_gpu] - if supported: - offset = occurrence % len(supported) - return [*supported[offset:], *supported[:offset]] - return [] + if category == LONG_TERM: + eligible = long_term + base_pattern = (5.0, 3.0, 2.0) + elif category == RECENT: + eligible = recent + base_pattern = (6.0, 3.0, 1.0) + else: + eligible = supported + base_pattern = tuple(1.0 for _ in eligible) + if not eligible: + return [] + + weights: dict[str, float] = {} + for index, gpu in enumerate(eligible): + base = base_pattern[min(index, len(base_pattern) - 1)] + market_weight = 1.0 + if self.market_intelligence is not None: + try: + market_weight = float(self.market_intelligence.gpu_weight(gpu, category=category)) + except Exception: + market_weight = 1.0 + weights[gpu] = max(0.05, base * market_weight) + + total_weight = sum(weights.values()) or 1.0 + next_total = sum(max(0, int(gpu_counts.get(gpu, 0))) for gpu in eligible) + 1 + rank_index = {gpu: index for index, gpu in enumerate(eligible)} + return sorted( + eligible, + key=lambda gpu: ( + -((weights[gpu] / total_weight) * next_total - max(0, int(gpu_counts.get(gpu, 0)))), + rank_index[gpu], + ), + ) def order_candidates(self, candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: if self.state is None or not candidates: @@ -388,6 +424,14 @@ class GPUStrategyManager: category: max(0, int((self.state.get("acceptedByCategory") or {}).get(category, 0))) for category in CATEGORIES } + stored_gpu_counts = self.state.get("acceptedByGpuCategory") or {} + virtual_gpu_counts = { + category: { + gpu: max(0, int((stored_gpu_counts.get(category) or {}).get(gpu, 0))) + for gpu in self.state.get("supportedGpus") or [] + } + for category in CATEGORIES + } def take_from_gpu(gpu: str) -> dict[str, Any] | None: pool = by_gpu.get(gpu) or [] @@ -407,8 +451,7 @@ class GPUStrategyManager: actual_category = planned_category categories_to_try = [planned_category, *(category for category in CATEGORIES if category != planned_category)] for category in categories_to_try: - occurrence = virtual_counts[category] - for gpu in self._category_gpu_order(category, occurrence): + for gpu in self._category_gpu_order(category, virtual_gpu_counts[category]): selected = take_from_gpu(gpu) if selected is not None: actual_category = category @@ -429,8 +472,18 @@ class GPUStrategyManager: annotated["strategyCategory"] = actual_category annotated["strategyPlannedCategory"] = planned_category annotated["strategyGeneration"] = int(self.state.get("generation") or 0) + if self.market_intelligence is not None: + try: + annotated.update(self.market_intelligence.gpu_metadata(str(annotated.get("targetGpu") or ""))) + except Exception: + pass ordered.append(annotated) virtual_counts[actual_category] += 1 + selected_gpu = str(annotated.get("targetGpu") or "") + if selected_gpu: + virtual_gpu_counts[actual_category][selected_gpu] = ( + virtual_gpu_counts[actual_category].get(selected_gpu, 0) + 1 + ) return ordered @@ -445,16 +498,29 @@ class GPUStrategyManager: category: max(0, int((state.get("acceptedByCategory") or {}).get(category, 0))) for category in CATEGORIES } + supported = list(state.get("supportedGpus") or []) + stored_gpu_counts = state.get("acceptedByGpuCategory") or {} + gpu_category_counts = { + category: { + gpu: max(0, int((stored_gpu_counts.get(category) or {}).get(gpu, 0))) + for gpu in supported + } + for category in CATEGORIES + } for candidate in candidates: category = str(candidate.get("strategyCategory") or ALL_SUPPORTED) if category not in category_counts: category = ALL_SUPPORTED category_counts[category] += 1 + gpu = str(candidate.get("targetGpu") or "") + if gpu in gpu_category_counts[category]: + gpu_category_counts[category][gpu] += 1 accepted_count = len(candidates) state["acceptedSinceRefresh"] = int(state.get("acceptedSinceRefresh") or 0) + accepted_count state["acceptedTotal"] = int(state.get("acceptedTotal") or 0) + accepted_count state["acceptedByCategory"] = category_counts + state["acceptedByGpuCategory"] = gpu_category_counts self.state = state write_json(self.path, state) self._log_state("progress") @@ -473,8 +539,10 @@ class GPUStrategyManager: "acceptedSinceRefresh": int(self.state.get("acceptedSinceRefresh") or 0), "refreshSubmissions": int(self.state.get("refreshSubmissions") or self.refresh_submissions), "acceptedByCategory": dict(self.state.get("acceptedByCategory") or {}), + "acceptedByGpuCategory": dict(self.state.get("acceptedByGpuCategory") or {}), "longTermGpus": list(self.state.get("longTermGpus") or []), "recentGpu": self.state.get("recentGpu"), + "recentGpus": list(self.state.get("recentGpus") or []), "recentTerminalCount": int(self.state.get("recentTerminalCount") or 0), "refreshDue": self.submissions_until_refresh <= 0, } diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 3de05137..b6cba0fe 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -18,6 +18,15 @@ from history_stats import ( load_ledger, update_history_archive, ) +from market_intelligence import ( + DEFAULT_FETCH_WORKERS, + DEFAULT_FRAMEWORK_MIN_SAMPLES, + DEFAULT_FRAMEWORK_REFRESH_SECONDS, + DEFAULT_MARKET_INTELLIGENCE_PATH, + DEFAULT_QUEUE_REFRESH_SECONDS, + DEFAULT_THROUGHPUT_WINDOW_HOURS, + MarketIntelligenceManager, +) from modelhub_client import ( DEFAULT_CAPACITY_STATE_PATH, ModelHubAPIError, @@ -30,7 +39,7 @@ from models import CandidateModel, HFModelSummary, ModelInspection from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, candidate_key, diversify_candidates from submission_exclusions import DEFAULT_SUBMISSION_EXCLUSIONS_PATH, SubmissionExclusionStore -from task_registry import TASK_SPEC_BY_TYPE, all_task_types, choose_framework_for_task, choose_text_generation_framework, pipeline_tags_for_task_types, task_specs_for_model +from task_registry import TASK_SPEC_BY_TYPE, all_task_types, compatible_frameworks_for_task, pipeline_tags_for_task_types, task_specs_for_model from template_selector import TemplateSelector @@ -61,6 +70,11 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="Disable adaptive 50/30/20 GPU scheduling and keep the legacy candidate order", ) + parser.add_argument( + "--disable-market-intelligence", + action="store_true", + help="Disable live queue/throughput and public framework statistics", + ) parser.add_argument( "--gpu-strategy-refresh-submissions", type=int, @@ -107,6 +121,41 @@ def build_parser() -> argparse.ArgumentParser: default=os.getenv("MODELHUB_SUBMISSION_EXCLUSIONS_PATH", str(DEFAULT_SUBMISSION_EXCLUSIONS_PATH)), help=argparse.SUPPRESS, ) + parser.add_argument( + "--market-intelligence-state-path", + default=os.getenv("MODELHUB_MARKET_INTELLIGENCE_PATH", str(DEFAULT_MARKET_INTELLIGENCE_PATH)), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-queue-refresh-seconds", + type=int, + default=int(os.getenv("MODELHUB_MARKET_QUEUE_REFRESH_SECONDS", str(DEFAULT_QUEUE_REFRESH_SECONDS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-framework-refresh-seconds", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_REFRESH_SECONDS", str(DEFAULT_FRAMEWORK_REFRESH_SECONDS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-throughput-window-hours", + type=int, + default=int(os.getenv("MODELHUB_MARKET_THROUGHPUT_WINDOW_HOURS", str(DEFAULT_THROUGHPUT_WINDOW_HOURS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-fetch-workers", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FETCH_WORKERS", str(DEFAULT_FETCH_WORKERS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-framework-min-samples", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES", str(DEFAULT_FRAMEWORK_MIN_SAMPLES))), + help=argparse.SUPPRESS, + ) parser.add_argument( "--gpu-strategy-state-path", default=os.getenv("MODELHUB_GPU_STRATEGY_STATE_PATH", str(DEFAULT_GPU_STRATEGY_PATH)), @@ -209,37 +258,90 @@ def choose_candidate_for_gpu( template_selector: TemplateSelector, task_types: list[str], target_gpu: str, + market_intelligence: MarketIntelligenceManager | None = None, ) -> CandidateModel | None: for task_type in task_types: supported_frameworks = template_selector.supported_frameworks_for_auto(task_type, target_gpu) - if not supported_frameworks: - continue + compatible_frameworks: list[str] = [] try: - framework = choose_framework_for_task(task_type, target_gpu, supported_frameworks, inspection) - template = template_selector.select_template(task_type, framework, target_gpu) - config_params = template_selector.render_config( - template, - gguf_filename=inspection.selected_gguf if framework == "llamacpp" else None, - ) + if supported_frameworks: + compatible_frameworks = compatible_frameworks_for_task( + task_type, + target_gpu, + supported_frameworks, + inspection, + ) except ValueError: + pass + if market_intelligence is not None: + discovered = market_intelligence.compatible_discovered_frameworks( + task_type=task_type, + target_gpu=target_gpu, + inspection=inspection, + ) + compatible_frameworks.extend( + framework for framework in discovered if framework not in compatible_frameworks + ) + if not compatible_frameworks: continue - spec = TASK_SPEC_BY_TYPE[task_type] - return CandidateModel( - repo_id=model.repo_id, - model_address=model.model_address, - pipeline_tag=model.pipeline_tag, - modality=spec.modality, - task_type=task_type, - target_gpu=target_gpu, - framework=framework, - template_id=template.template_id, - config_params=config_params, - downloads=model.downloads, - last_modified=model.last_modified, - gguf_filename=inspection.selected_gguf, - score=0.0, - warnings=[], - ) + if market_intelligence is not None: + compatible_frameworks = market_intelligence.rank_frameworks( + task_type=task_type, + target_gpu=target_gpu, + compatible_frameworks=compatible_frameworks, + ) + + for framework in compatible_frameworks: + config_params = ( + market_intelligence.official_config( + task_type=task_type, + target_gpu=target_gpu, + framework=framework, + gguf_filename=inspection.selected_gguf, + ) + if market_intelligence is not None + else None + ) + if config_params is not None: + template_id = f"modelhub-live-{task_type}-{framework}-{target_gpu}".lower().replace("_", "-") + else: + try: + template = template_selector.select_template(task_type, framework, target_gpu) + config_params = template_selector.render_config( + template, + gguf_filename=inspection.selected_gguf if framework == "llamacpp" else None, + ) + template_id = template.template_id + except (KeyError, ValueError): + continue + framework_metadata = ( + market_intelligence.framework_metadata(task_type, target_gpu, framework) + if market_intelligence is not None + else {} + ) + score = float(framework_metadata.get("frameworkCombinedScore") or 0.0) + warnings: list[str] = [] + if bool(framework_metadata.get("frameworkEvidenceQualified", False)): + warnings.append("framework_selected_from_success_evidence") + if config_params is not None and bool(framework_metadata.get("frameworkOfficialConfigValid", False)): + warnings.append("official_build_config_synced") + spec = TASK_SPEC_BY_TYPE[task_type] + return CandidateModel( + repo_id=model.repo_id, + model_address=model.model_address, + pipeline_tag=model.pipeline_tag, + modality=spec.modality, + task_type=task_type, + target_gpu=target_gpu, + framework=framework, + template_id=template_id, + config_params=config_params, + downloads=model.downloads, + last_modified=model.last_modified, + gguf_filename=inspection.selected_gguf, + score=score, + warnings=warnings, + ) return None @@ -284,6 +386,7 @@ def process_model_for_candidates( allowed_task_types: list[str], outcome_tracker: OutcomeTracker | None = None, submission_exclusion_store: SubmissionExclusionStore | None = None, + market_intelligence: MarketIntelligenceManager | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: specs = [spec for spec in task_specs_for_model(model) if spec.task_type in allowed_task_types] if not specs: @@ -345,11 +448,16 @@ def process_model_for_candidates( template_selector=template_selector, task_types=task_types, target_gpu=target_gpu, + market_intelligence=market_intelligence, ) if best is None: skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "no_compatible_auto_template_or_framework"}) continue - candidates.append(candidate_to_record(best)) + record = candidate_to_record(best) + 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)) + candidates.append(record) return candidates, skipped, failed @@ -423,6 +531,7 @@ def collect_candidates_from_models( outcome_tracker: OutcomeTracker | None, submission_exclusion_store: SubmissionExclusionStore | None, read_concurrency: int, + market_intelligence: MarketIntelligenceManager | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], int]: candidates: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] @@ -451,6 +560,7 @@ def collect_candidates_from_models( allowed_task_types=selected_task_types, outcome_tracker=outcome_tracker, submission_exclusion_store=submission_exclusion_store, + market_intelligence=market_intelligence, ): index for index, model in enumerate(chunk) } @@ -624,6 +734,8 @@ def run_submission( ) strategy_manager: GPUStrategyManager | None = None strategy_summary: dict[str, Any] = {"enabled": False} + market_intelligence: MarketIntelligenceManager | None = None + market_summary: dict[str, Any] = {"enabled": False} strategy_enabled = ( not bool(getattr(args, "disable_gpu_strategy", False)) and not bool(getattr(args, "gpu", None) or getattr(args, "gpus", None)) @@ -707,15 +819,19 @@ def run_submission( "historyArchivePath": str(history_archive_path), "historyArchiveRecordCount": len(archived_history), "gpuStrategy": strategy_summary, + "marketIntelligence": market_summary, "scanLimit": 0, "candidateGoal": 0, "scanStages": [], "scannedModels": 0, "candidateCount": 0, + "candidateFrameworkCounts": {}, "targetSubmitCount": 0, "maxSubmitAttempts": 0, "plannedSubmitCount": 0, "submittedCount": 0, + "submittedGpuCounts": {}, + "submittedFrameworkCounts": {}, "duplicateCount": 0, "modelGpuUniquenessRejectedCount": 0, "skippedCount": 0, @@ -728,12 +844,63 @@ def run_submission( "outcomeSyncCount": synced_count, } + market_capable = ( + hasattr(modelhub_client, "list_machine_info") + and hasattr(modelhub_client, "list_framework_stats") + and hasattr(modelhub_client, "list_tasks_page") + ) + if not bool(getattr(args, "disable_market_intelligence", False)) and market_capable: + market_intelligence = MarketIntelligenceManager( + Path(getattr(args, "market_intelligence_state_path", DEFAULT_MARKET_INTELLIGENCE_PATH)), + queue_refresh_seconds=max( + 60, + int(getattr(args, "market_queue_refresh_seconds", DEFAULT_QUEUE_REFRESH_SECONDS) or DEFAULT_QUEUE_REFRESH_SECONDS), + ), + framework_refresh_seconds=max( + 300, + int( + getattr(args, "market_framework_refresh_seconds", DEFAULT_FRAMEWORK_REFRESH_SECONDS) + or DEFAULT_FRAMEWORK_REFRESH_SECONDS + ), + ), + throughput_window_hours=max( + 1, + int( + getattr(args, "market_throughput_window_hours", DEFAULT_THROUGHPUT_WINDOW_HOURS) + or DEFAULT_THROUGHPUT_WINDOW_HOURS + ), + ), + fetch_workers=max( + 1, + int(getattr(args, "market_fetch_workers", DEFAULT_FETCH_WORKERS) or DEFAULT_FETCH_WORKERS), + ), + framework_min_samples=max( + 1, + int( + getattr(args, "market_framework_min_samples", DEFAULT_FRAMEWORK_MIN_SAMPLES) + or DEFAULT_FRAMEWORK_MIN_SAMPLES + ), + ), + ) + market_intelligence.prepare( + modelhub_client, + supported_gpus=target_gpus, + task_types=selected_task_types, + now=now, + ) + try: + market_intelligence.set_local_outcome_stats(outcome_tracker.get_stats_report()) + except Exception: + market_intelligence.set_local_outcome_stats(None) + market_summary = market_intelligence.summary() + if strategy_enabled: strategy_manager = GPUStrategyManager( Path(getattr(args, "gpu_strategy_state_path", DEFAULT_GPU_STRATEGY_PATH)), refresh_submissions=max(1, int(getattr(args, "gpu_strategy_refresh_submissions", 200) or 200)), recent_terminal_window=max(1, int(getattr(args, "gpu_strategy_recent_window", 1000) or 1000)), long_term_min_samples=max(1, int(getattr(args, "gpu_strategy_min_long_samples", 100) or 100)), + market_intelligence=market_intelligence, ) strategy_manager.prepare(modelhub_client, supported_gpus=target_gpus, now=now) strategy_summary = strategy_manager.summary() @@ -824,6 +991,7 @@ def run_submission( outcome_tracker=outcome_tracker, submission_exclusion_store=submission_exclusion_store, read_concurrency=max(1, args.read_concurrency), + market_intelligence=market_intelligence, ) candidates.extend(stage_candidates) skipped.extend(stage_skipped) @@ -1027,6 +1195,9 @@ def run_submission( outcome_tracker.save() skip_reason_counts = Counter(str(item.get("reason") or "unknown") for item in skipped) failure_reason_counts = Counter(str(item.get("reason") or "unknown") for item in failed) + candidate_framework_counts = Counter(str(item.get("framework") or "unknown") for item in candidates) + submitted_gpu_counts = Counter(str(item.get("targetGpu") or "unknown") for item in submitted) + submitted_framework_counts = Counter(str(item.get("framework") or "unknown") for item in submitted) summary = { "generatedAt": now.isoformat(), @@ -1044,11 +1215,13 @@ def run_submission( "historyArchivePath": str(history_archive_path), "historyArchiveRecordCount": len(archived_history), "gpuStrategy": strategy_summary, + "marketIntelligence": market_summary, "scanLimit": scan_limit, "candidateGoal": candidate_goal, "scanStages": scan_stages, "scannedModels": scanned_model_count, "candidateCount": len(candidates), + "candidateFrameworkCounts": dict(candidate_framework_counts.most_common()), "targetSubmitCount": target_submit_count, "maxSubmitAttempts": 0 if args.dry_run else min( len(diversified_candidates), @@ -1058,6 +1231,8 @@ def run_submission( "duplicateCount": len(duplicate_candidates), "modelGpuUniquenessRejectedCount": len(uniqueness_rejected_candidates), "submittedCount": len(submitted), + "submittedGpuCounts": dict(submitted_gpu_counts.most_common()), + "submittedFrameworkCounts": dict(submitted_framework_counts.most_common()), "skippedCount": len(skipped), "skipReasonCounts": dict(skip_reason_counts.most_common()), "failedCount": len(failed), diff --git a/modelhub_submmit_api/market_intelligence.py b/modelhub_submmit_api/market_intelligence.py new file mode 100644 index 00000000..2a9f5c4d --- /dev/null +++ b/modelhub_submmit_api/market_intelligence.py @@ -0,0 +1,650 @@ +from __future__ import annotations + +import math +import re +import statistics +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any, Callable + +from common import parse_datetime, read_json, utc_now, write_json + + +DEFAULT_MARKET_INTELLIGENCE_PATH = Path(".modelhub_state/market_intelligence.json") +MARKET_STATE_VERSION = 2 +DEFAULT_QUEUE_REFRESH_SECONDS = 600 +DEFAULT_FRAMEWORK_REFRESH_SECONDS = 21_600 +DEFAULT_THROUGHPUT_WINDOW_HOURS = 6 +DEFAULT_FETCH_WORKERS = 4 +DEFAULT_FRAMEWORK_MIN_SAMPLES = 100 +ERROR_RETRY_SECONDS = 300 + + +def _clamp(value: float, minimum: float, maximum: float) -> float: + return max(minimum, min(maximum, value)) + + +def _wilson_lower_bound(successes: int, total: int, *, z: float = 1.96) -> float: + if total <= 0: + return 0.0 + probability = successes / total + z_squared = z * z + denominator = 1.0 + z_squared / total + centre = probability + z_squared / (2.0 * total) + margin = z * math.sqrt( + (probability * (1.0 - probability) / total) + z_squared / (4.0 * total * total) + ) + return max(0.0, (centre - margin) / denominator) + + +def _fresh(timestamp: Any, *, now: datetime, ttl_seconds: int) -> bool: + parsed = parse_datetime(timestamp) + if parsed is None: + return False + return (now - parsed).total_seconds() < max(1, ttl_seconds) + + +def _page_total(client: Any, **kwargs: Any) -> int: + payload = client.list_tasks_page(current=1, page_size=1, only_mine=False, **kwargs) + data = payload.get("data") or {} + if not isinstance(data, dict) or "total" not in data: + raise RuntimeError("Public task count response is incomplete") + return max(0, int(data.get("total") or 0)) + + +def _neutral_gpu_stats(gpus: list[str]) -> dict[str, dict[str, Any]]: + return { + gpu: { + "gpu": gpu, + "available": True, + "canVerify": None, + "maxConcurrentTasks": None, + "waiting": None, + "running": None, + "recentSuccess": None, + "recentTerminal": None, + "recentSuccessRate": None, + "recentWilsonLowerBound": None, + "throughputPerHour": None, + "backlogHours": None, + "queueFactor": 1.0, + "qualityFactor": 1.0, + "healthFactor": 1.0, + "queueWeight": 1.0, + "selectionWeight": 1.0, + "error": "market_data_unavailable", + } + for gpu in gpus + } + + +def _validated_official_config(config: str, *, framework: str, target_gpu: str) -> tuple[bool, str | None]: + if not isinstance(config, str) or not config.strip(): + return False, "empty_config" + if len(config) > 200_000: + return False, "config_too_large" + if "sut_config:" not in config or "ref_config:" not in config: + return False, "missing_sut_or_ref_config" + if any(marker in config for marker in ("{{", "}}", "PLACEHOLDER")): + return False, "unresolved_placeholder" + + declared = re.search(r"(?m)^\s*framework:\s*['\"]?([^'\"\s]+)", config) + if declared is None or declared.group(1).strip() != framework: + return False, "framework_mismatch" + + if target_gpu == "Biren_166m": + gpu_counts = [int(value) for value in re.findall(r"\bgpu_num:\s*['\"]?(\d+)", config)] + parallel_counts = [ + int(value) + for value in re.findall( + r"(?:-tp|--tensor-parallel-size)(?:\s+|,\s*|\n\s*-\s*)['\"]?(\d+)", + config, + ) + ] + if any(value > 1 for value in [*gpu_counts, *parallel_counts]): + return False, "biren_parallelism_exceeds_one" + return True, None + + +def _render_official_config(config: str, *, gguf_filename: str | None) -> str: + if not gguf_filename: + return config + return re.sub( + r"(?i)(/model/)[^,\]\s'\"]+\.gguf", + lambda match: f"{match.group(1)}{gguf_filename}", + config, + ) + + +class MarketIntelligenceManager: + def __init__( + self, + path: Path | str = DEFAULT_MARKET_INTELLIGENCE_PATH, + *, + queue_refresh_seconds: int = DEFAULT_QUEUE_REFRESH_SECONDS, + framework_refresh_seconds: int = DEFAULT_FRAMEWORK_REFRESH_SECONDS, + throughput_window_hours: int = DEFAULT_THROUGHPUT_WINDOW_HOURS, + fetch_workers: int = DEFAULT_FETCH_WORKERS, + framework_min_samples: int = DEFAULT_FRAMEWORK_MIN_SAMPLES, + log_fn: Callable[[str], None] | None = None, + ) -> None: + self.path = Path(path) + self.queue_refresh_seconds = max(60, int(queue_refresh_seconds)) + self.framework_refresh_seconds = max(300, int(framework_refresh_seconds)) + self.throughput_window_hours = max(1, int(throughput_window_hours)) + self.fetch_workers = max(1, min(8, int(fetch_workers))) + self.framework_min_samples = max(1, int(framework_min_samples)) + self.log = log_fn or (lambda message: print(message, flush=True)) + self.state: dict[str, Any] | None = None + self.local_outcome_stats: dict[str, Any] = {} + self._last_framework_request_errors: set[tuple[str, str]] = set() + + def set_local_outcome_stats(self, report: dict[str, Any] | None) -> None: + self.local_outcome_stats = report if isinstance(report, dict) else {} + + def _load(self) -> dict[str, Any] | None: + try: + payload = read_json(self.path) + except (FileNotFoundError, ValueError): + return None + return payload if isinstance(payload, dict) else None + + @staticmethod + def _compatible(state: dict[str, Any] | None, gpus: list[str], task_types: list[str]) -> bool: + if not state or int(state.get("version") or 0) != MARKET_STATE_VERSION: + return False + return list(state.get("supportedGpus") or []) == gpus and list(state.get("taskTypes") or []) == task_types + + def _base_state(self, gpus: list[str], task_types: list[str], now: datetime) -> dict[str, Any]: + return { + "version": MARKET_STATE_VERSION, + "generatedAt": now.isoformat(), + "supportedGpus": gpus, + "taskTypes": task_types, + "queueUpdatedAt": None, + "frameworkUpdatedAt": None, + "queueAttemptedAt": None, + "frameworkAttemptedAt": None, + "throughputWindowHours": self.throughput_window_hours, + "gpuStats": _neutral_gpu_stats(gpus), + "frameworkStats": {}, + "queueError": None, + "frameworkError": None, + } + + def prepare( + self, + client: Any, + *, + supported_gpus: list[str], + task_types: list[str], + now: datetime | None = None, + ) -> dict[str, Any]: + now = now or utc_now() + gpus = list(dict.fromkeys(gpu for gpu in supported_gpus if gpu)) + tasks = list(dict.fromkeys(task for task in task_types if task)) + loaded = self._load() + state = loaded if self._compatible(loaded, gpus, tasks) else self._base_state(gpus, tasks, now) + + queue_due = not _fresh( + state.get("queueUpdatedAt"), + now=now, + ttl_seconds=self.queue_refresh_seconds, + ) + framework_due = not _fresh( + state.get("frameworkUpdatedAt"), + now=now, + ttl_seconds=self.framework_refresh_seconds, + ) + if state.get("queueError") and _fresh( + state.get("queueAttemptedAt"), + now=now, + ttl_seconds=min(ERROR_RETRY_SECONDS, self.queue_refresh_seconds), + ): + queue_due = False + if state.get("frameworkError") and _fresh( + state.get("frameworkAttemptedAt"), + now=now, + ttl_seconds=min(ERROR_RETRY_SECONDS, self.framework_refresh_seconds), + ): + framework_due = False + + if queue_due: + state["queueAttemptedAt"] = now.isoformat() + try: + previous_gpu_stats = state.get("gpuStats") or {} + fresh_gpu_stats = self._fetch_gpu_stats(client, gpus=gpus, now=now) + for gpu, fresh_item in fresh_gpu_stats.items(): + previous_item = previous_gpu_stats.get(gpu) or {} + if fresh_item.get("error") and previous_item and not previous_item.get("error"): + stale_item = dict(previous_item) + stale_item["stale"] = True + stale_item["refreshError"] = fresh_item.get("error") + fresh_gpu_stats[gpu] = stale_item + else: + fresh_item["stale"] = False + state["gpuStats"] = fresh_gpu_stats + state["queueUpdatedAt"] = now.isoformat() + state["queueError"] = None + except Exception as exc: + state["queueError"] = f"{type(exc).__name__}: {exc}" + if not state.get("gpuStats"): + state["gpuStats"] = _neutral_gpu_stats(gpus) + self.log(f"[market] queue_refresh_error reason={state['queueError']}") + + if framework_due: + state["frameworkAttemptedAt"] = now.isoformat() + try: + previous_framework_stats = state.get("frameworkStats") or {} + fresh_framework_stats = self._fetch_framework_stats(client, gpus=gpus, task_types=tasks) + for task_type, gpu in self._last_framework_request_errors: + previous_rows = ((previous_framework_stats.get(task_type) or {}).get(gpu) or {}) + if previous_rows: + fresh_framework_stats[task_type][gpu] = { + framework: {**dict(item), "stale": True} + for framework, item in previous_rows.items() + } + for task_type, by_gpu in fresh_framework_stats.items(): + for gpu, rows in by_gpu.items(): + previous_rows = ((previous_framework_stats.get(task_type) or {}).get(gpu) or {}) + for framework, item in rows.items(): + previous_item = previous_rows.get(framework) or {} + if not item.get("officialConfigValid") and previous_item.get("officialConfigValid"): + item["officialConfigValid"] = True + item["officialConfig"] = previous_item.get("officialConfig") + item["officialConfigStale"] = True + item["officialConfigRefreshError"] = item.get("officialConfigError") + state["frameworkStats"] = fresh_framework_stats + state["frameworkUpdatedAt"] = now.isoformat() + state["frameworkError"] = None + except Exception as exc: + state["frameworkError"] = f"{type(exc).__name__}: {exc}" + if not state.get("frameworkStats"): + state["frameworkStats"] = {} + self.log(f"[market] framework_refresh_error reason={state['frameworkError']}") + + state["generatedAt"] = now.isoformat() + state["throughputWindowHours"] = self.throughput_window_hours + self.state = state + write_json(self.path, state) + self._log_snapshot("refreshed" if queue_due or framework_due else "loaded") + return state + + def _fetch_gpu_stats(self, client: Any, *, gpus: list[str], now: datetime) -> dict[str, dict[str, Any]]: + local_end = now.astimezone() + local_begin = local_end - timedelta(hours=self.throughput_window_hours) + begin_text = local_begin.strftime("%Y-%m-%d %H:%M:%S") + end_text = local_end.strftime("%Y-%m-%d %H:%M:%S") + + machine_by_gpu: dict[str, dict[str, Any]] = {} + for item in client.list_machine_info(): + gpu = str(item.get("gpuType") or "") + if gpu: + machine_by_gpu[gpu] = item + + def fetch_one(gpu: str) -> tuple[str, dict[str, Any]]: + waiting = _page_total(client, status="waiting", gpu_type=gpu) + running = _page_total(client, status="running", gpu_type=gpu) + completed = _page_total( + client, + status="success", + gpu_type=gpu, + begin_time=begin_text, + end_time=end_text, + ) + success = _page_total( + client, + status="success", + verify_result=1, + gpu_type=gpu, + begin_time=begin_text, + end_time=end_text, + ) + abnormal = _page_total( + client, + status="failed", + gpu_type=gpu, + begin_time=begin_text, + end_time=end_text, + ) + terminal = max(success, completed + abnormal) + throughput = terminal / self.throughput_window_hours + backlog_hours = waiting / throughput if throughput > 0 else 9_999.0 + success_rate = success / terminal if terminal else 0.0 + machine = machine_by_gpu.get(gpu) or {} + return gpu, { + "gpu": gpu, + "available": machine.get("canVerify") is not False, + "canVerify": machine.get("canVerify"), + "maxConcurrentTasks": machine.get("maxConcurrentTasks"), + "waiting": waiting, + "running": running, + "recentSuccess": success, + "recentTerminal": terminal, + "recentSuccessRate": success_rate, + "recentWilsonLowerBound": _wilson_lower_bound(success, terminal), + "throughputPerHour": throughput, + "backlogHours": backlog_hours, + "error": None, + } + + fetched: dict[str, dict[str, Any]] = {} + errors: dict[str, str] = {} + with ThreadPoolExecutor(max_workers=min(self.fetch_workers, max(1, len(gpus)))) as executor: + futures = {executor.submit(fetch_one, gpu): gpu for gpu in gpus} + for future in as_completed(futures): + gpu = futures[future] + try: + name, stats = future.result() + fetched[name] = stats + except Exception as exc: + errors[gpu] = f"{type(exc).__name__}: {exc}" + + if not fetched and gpus: + raise RuntimeError("all GPU queue-stat requests failed") + + finite_backlogs = [ + float(stats["backlogHours"]) + for stats in fetched.values() + if float(stats["backlogHours"]) < 9_999.0 + ] + median_backlog = statistics.median(finite_backlogs) if finite_backlogs else 24.0 + max_wilson = max((float(stats["recentWilsonLowerBound"]) for stats in fetched.values()), default=0.0) + + result = _neutral_gpu_stats(gpus) + for gpu, stats in fetched.items(): + backlog = max(0.25, float(stats["backlogHours"])) + queue_factor = _clamp(math.sqrt(max(0.25, median_backlog) / backlog), 0.35, 1.75) + wilson = float(stats["recentWilsonLowerBound"]) + quality_factor = 1.0 if max_wilson <= 0 else 0.6 + 1.4 * (wilson / max_wilson) + health_factor = 1.0 + if stats.get("canVerify") is False: + health_factor = 0.05 + elif ( + int(stats.get("running") or 0) == 0 + and int(stats.get("waiting") or 0) > 0 + and int(stats.get("maxConcurrentTasks") or 0) <= 0 + and int(stats.get("recentSuccess") or 0) == 0 + ): + health_factor = 0.10 + elif ( + int(stats.get("running") or 0) == 0 + and int(stats.get("waiting") or 0) > 0 + and int(stats.get("recentTerminal") or 0) == 0 + ): + health_factor = 0.25 + + stats["queueFactor"] = queue_factor + stats["qualityFactor"] = quality_factor + stats["healthFactor"] = health_factor + stats["queueWeight"] = _clamp(queue_factor * health_factor, 0.05, 2.0) + stats["selectionWeight"] = _clamp(queue_factor * quality_factor * health_factor, 0.05, 3.0) + result[gpu] = stats + + for gpu, error in errors.items(): + result[gpu]["error"] = error + return result + + def _fetch_framework_stats( + self, + client: Any, + *, + gpus: list[str], + task_types: list[str], + ) -> dict[str, dict[str, dict[str, dict[str, Any]]]]: + self._last_framework_request_errors = set() + result: dict[str, dict[str, dict[str, dict[str, Any]]]] = { + task_type: {gpu: {} for gpu in gpus} + for task_type in task_types + } + + def fetch_one(task_type: str, gpu: str) -> tuple[str, str, list[dict[str, Any]]]: + return task_type, gpu, client.list_framework_stats(task_type, gpu) + + with ThreadPoolExecutor(max_workers=self.fetch_workers) as executor: + futures = { + executor.submit(fetch_one, task_type, gpu): (task_type, gpu) + for task_type in task_types + for gpu in gpus + } + successful_requests = 0 + errors: list[str] = [] + for future in as_completed(futures): + task_type, gpu = futures[future] + try: + _, _, rows = future.result() + successful_requests += 1 + except Exception as exc: + errors.append(f"{task_type}/{gpu}: {type(exc).__name__}: {exc}") + self._last_framework_request_errors.add((task_type, gpu)) + continue + for row in rows: + framework = str(row.get("framework") or "").strip() + if not framework: + continue + total = max(0, int(row.get("modelCount") or 0)) + success = max(0, min(total, int(row.get("successCount") or 0))) + result[task_type][gpu][framework] = { + "framework": framework, + "modelCount": total, + "successCount": success, + "successRate": success / total if total else 0.0, + "wilsonLowerBound": _wilson_lower_bound(success, total), + } + if futures and successful_requests <= 0: + detail = errors[0] if errors else "unknown error" + raise RuntimeError(f"all framework-stat requests failed ({detail})") + + if hasattr(client, "get_build_config"): + config_targets = [ + (task_type, gpu, framework) + for task_type, by_gpu in result.items() + for gpu, by_framework in by_gpu.items() + for framework in by_framework + ] + + def fetch_config(task_type: str, gpu: str, framework: str) -> tuple[str, str, str, str]: + return task_type, gpu, framework, client.get_build_config(task_type, gpu, framework) + + with ThreadPoolExecutor(max_workers=self.fetch_workers) as executor: + config_futures = { + executor.submit(fetch_config, task_type, gpu, framework): (task_type, gpu, framework) + for task_type, gpu, framework in config_targets + } + for future in as_completed(config_futures): + task_type, gpu, framework = config_futures[future] + item = result[task_type][gpu][framework] + try: + _, _, _, config = future.result() + valid, reason = _validated_official_config( + config, + framework=framework, + target_gpu=gpu, + ) + item["officialConfigValid"] = valid + item["officialConfigError"] = reason + if valid: + item["officialConfig"] = config + except Exception as exc: + item["officialConfigValid"] = False + item["officialConfigError"] = f"{type(exc).__name__}: {exc}" + return result + + def gpu_weight(self, gpu: str, *, category: str) -> float: + stats = ((self.state or {}).get("gpuStats") or {}).get(gpu) or {} + if category == "all_supported": + return max(0.05, float(stats.get("queueWeight") or 1.0)) + return max(0.05, float(stats.get("selectionWeight") or 1.0)) + + def gpu_metadata(self, gpu: str) -> dict[str, Any]: + stats = ((self.state or {}).get("gpuStats") or {}).get(gpu) or {} + return { + "marketWeight": float(stats.get("selectionWeight") or 1.0), + "queueWeight": float(stats.get("queueWeight") or 1.0), + "queueWaiting": stats.get("waiting"), + "queueRunning": stats.get("running"), + "queueBacklogHours": stats.get("backlogHours"), + "publicRecentSuccessRate": stats.get("recentSuccessRate"), + "publicThroughputPerHour": stats.get("throughputPerHour"), + "marketDataStale": bool(stats.get("stale", False)), + } + + def rank_frameworks( + self, + *, + task_type: str, + target_gpu: str, + compatible_frameworks: list[str], + ) -> list[str]: + stats = ( + (((self.state or {}).get("frameworkStats") or {}).get(task_type) or {}).get(target_gpu) + or {} + ) + legacy_index = {framework: index for index, framework in enumerate(compatible_frameworks)} + + def rank_key(framework: str) -> tuple[int, float, int, int]: + evidence = self._framework_evidence(task_type, target_gpu, framework) + return ( + 1 if evidence["qualified"] else 0, + float(evidence["combinedScore"]) if evidence["qualified"] else 0.0, + int(evidence["publicSamples"]) + int(evidence["localSamples"]), + -legacy_index[framework], + ) + + return sorted(compatible_frameworks, key=rank_key, reverse=True) + + def _framework_evidence(self, task_type: str, target_gpu: str, framework: str) -> dict[str, Any]: + public_item = ( + (((self.state or {}).get("frameworkStats") or {}).get(task_type) or {}).get(target_gpu) + or {} + ).get(framework) or {} + public_samples = max(0, int(public_item.get("modelCount") or 0)) + public_score = float(public_item.get("wilsonLowerBound") or 0.0) + + local_key = f"{target_gpu}|{framework}|{task_type}" + local_item = (self.local_outcome_stats.get("combinationStats") or {}).get(local_key) or {} + local_success = max(0, int(local_item.get("successCount") or 0)) + local_failure = max(0, int(local_item.get("failureCount") or 0)) + local_samples = local_success + local_failure + local_score = _wilson_lower_bound(local_success, local_samples) + + public_qualified = public_samples >= self.framework_min_samples + local_qualified = local_samples >= 20 + if public_qualified: + local_weight = min(0.35, local_samples / (local_samples + 50.0)) if local_samples >= 5 else 0.0 + combined = public_score * (1.0 - local_weight) + local_score * local_weight + elif local_qualified: + combined = local_score + else: + combined = 0.0 + return { + "qualified": public_qualified or local_qualified, + "publicQualified": public_qualified, + "localQualified": local_qualified, + "combinedScore": combined, + "publicSamples": public_samples, + "publicScore": public_score, + "localSamples": local_samples, + "localSuccessRate": local_success / local_samples if local_samples else None, + "localScore": local_score if local_samples else None, + } + + def compatible_discovered_frameworks( + self, + *, + task_type: str, + target_gpu: str, + inspection: Any, + ) -> list[str]: + stats = ( + (((self.state or {}).get("frameworkStats") or {}).get(task_type) or {}).get(target_gpu) + or {} + ) + compatible: list[str] = [] + for framework, item in stats.items(): + if not bool(item.get("officialConfigValid", False)): + continue + normalized = framework.lower() + if "llamacpp" in normalized or "gguf" in normalized: + usable = bool(getattr(inspection, "has_gguf", False)) + elif "onnx" in normalized or "sherpa" in normalized: + usable = bool(getattr(inspection, "has_onnx_weights", False)) + elif task_type in {"text-generation", "visual-multi-modal", "reinforcement_learning"}: + usable = bool(getattr(inspection, "has_vllm_weights", False)) + else: + usable = bool(getattr(inspection, "has_standard_weights", False)) + if usable: + compatible.append(framework) + return compatible + + def official_config( + self, + *, + task_type: str, + target_gpu: str, + framework: str, + gguf_filename: str | None = None, + ) -> str | None: + item = ( + (((self.state or {}).get("frameworkStats") or {}).get(task_type) or {}).get(target_gpu) + or {} + ).get(framework) or {} + if not bool(item.get("officialConfigValid", False)): + return None + config = item.get("officialConfig") + if not isinstance(config, str) or not config: + return None + return _render_official_config(config, gguf_filename=gguf_filename) + + def framework_metadata(self, task_type: str, target_gpu: str, framework: str) -> dict[str, Any]: + item = ( + (((self.state or {}).get("frameworkStats") or {}).get(task_type) or {}).get(target_gpu) + or {} + ).get(framework) or {} + evidence = self._framework_evidence(task_type, target_gpu, framework) + return { + "frameworkMarketSamples": int(item.get("modelCount") or 0), + "frameworkMarketSuccessRate": item.get("successRate"), + "frameworkMarketWilsonLowerBound": item.get("wilsonLowerBound"), + "frameworkLocalSamples": evidence["localSamples"], + "frameworkLocalSuccessRate": evidence["localSuccessRate"], + "frameworkCombinedScore": evidence["combinedScore"], + "frameworkMarketQualified": evidence["publicQualified"], + "frameworkLocalQualified": evidence["localQualified"], + "frameworkEvidenceQualified": evidence["qualified"], + "frameworkOfficialConfigValid": bool(item.get("officialConfigValid", False)), + "frameworkOfficialConfigStale": bool(item.get("officialConfigStale", False)), + "frameworkConfigSource": "modelhub_live" if item.get("officialConfigValid") else "local_template", + } + + def _log_snapshot(self, action: str) -> None: + if self.state is None: + return + stats = self.state.get("gpuStats") or {} + ranked = sorted( + stats.values(), + key=lambda item: -float(item.get("selectionWeight") or 0.0), + ) + leaders = ",".join( + f"{item.get('gpu')}:{float(item.get('selectionWeight') or 0.0):.2f}" + for item in ranked[:3] + ) + self.log( + f"[market] {action} queue_at={self.state.get('queueUpdatedAt') or 'n/a'} " + f"framework_at={self.state.get('frameworkUpdatedAt') or 'n/a'} leaders={leaders or 'n/a'}" + ) + + def summary(self) -> dict[str, Any]: + if self.state is None: + return {"enabled": False} + return { + "enabled": True, + "statePath": str(self.path), + "queueUpdatedAt": self.state.get("queueUpdatedAt"), + "frameworkUpdatedAt": self.state.get("frameworkUpdatedAt"), + "throughputWindowHours": int(self.state.get("throughputWindowHours") or self.throughput_window_hours), + "queueError": self.state.get("queueError"), + "frameworkError": self.state.get("frameworkError"), + "gpuStats": dict(self.state.get("gpuStats") or {}), + } diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index d6ea3242..77862723 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -85,11 +85,18 @@ class ModelHubClient: current: int = 1, page_size: int = 50, only_mine: bool = True, - begin_time: datetime | None = None, - end_time: datetime | None = None, + begin_time: datetime | str | None = None, + end_time: datetime | str | None = None, gpu_type: str | None = None, model_id: str | None = None, + status: str | None = None, + verify_result: int | None = None, ) -> dict[str, Any]: + def query_time(value: datetime | str | None) -> str | None: + if isinstance(value, str): + return value + return format_modelhub_datetime(value) if value else None + return self._request( "GET", "/api/adapt/task/page", @@ -97,13 +104,44 @@ class ModelHubClient: "current": current, "pageSize": page_size, "onlyMine": str(only_mine).lower(), - "beginTime": format_modelhub_datetime(begin_time) if begin_time else None, - "endTime": format_modelhub_datetime(end_time) if end_time else None, + "beginTime": query_time(begin_time), + "endTime": query_time(end_time), "gpuType": gpu_type, "modelId": model_id, + "status": status, + "verifyResult": verify_result, }, ) + def list_machine_info(self) -> list[dict[str, Any]]: + payload = self._request("GET", "/api/computility/power/machine/list/machine-info") + data = payload.get("data") or [] + if not isinstance(data, list): + raise ModelHubAPIError("Machine info response is invalid", payload=payload) + return [item for item in data if isinstance(item, dict)] + + def list_framework_stats(self, task_type: str, target_gpu: str) -> list[dict[str, Any]]: + payload = self._request( + "GET", + "/api/computility/driver/images/frameworks", + query={"taskType": task_type, "gpuTypeName": target_gpu}, + ) + data = payload.get("data") or [] + if not isinstance(data, list): + raise ModelHubAPIError("Framework statistics response is invalid", payload=payload) + return [item for item in data if isinstance(item, dict)] + + def get_build_config(self, task_type: str, target_gpu: str, framework: str) -> str: + payload = self._request( + "POST", + "/api/adapt/task/build-config", + query={"taskType": task_type, "gpuType": target_gpu, "framework": framework}, + ) + data = payload.get("data") + if not isinstance(data, str) or not data.strip(): + raise ModelHubAPIError("Official build config response is invalid", payload=payload) + return data + def list_tasks( self, *, @@ -747,6 +785,15 @@ class ModelHubClientPool: """Single-page task listing via the reader client (no fanout).""" return self._reader.list_tasks_page(**kwargs) + def list_machine_info(self) -> list[dict[str, Any]]: + return self._reader.list_machine_info() + + def list_framework_stats(self, task_type: str, target_gpu: str) -> list[dict[str, Any]]: + return self._reader.list_framework_stats(task_type, target_gpu) + + def get_build_config(self, task_type: str, target_gpu: str, framework: str) -> str: + return self._reader.get_build_config(task_type, target_gpu, framework) + def find_recent_task_id(self, model_id: str, gpu_type: str, submitted_after: datetime) -> str | None: for client in self.clients: task_id = client.find_recent_task_id(model_id, gpu_type, submitted_after) diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 15cacf3f..5396073c 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -13,6 +13,14 @@ from daily_runner import DEFAULT_DAILY_RUNS_DIR, log, run_daily_batches from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH from hf_discovery import HuggingFaceDiscovery from main import DEFAULT_LEDGER_PATH, DEFAULT_RUNS_DIR, make_run_dir +from market_intelligence import ( + DEFAULT_FETCH_WORKERS, + DEFAULT_FRAMEWORK_MIN_SAMPLES, + DEFAULT_FRAMEWORK_REFRESH_SECONDS, + DEFAULT_MARKET_INTELLIGENCE_PATH, + DEFAULT_QUEUE_REFRESH_SECONDS, + DEFAULT_THROUGHPUT_WINDOW_HOURS, +) from modelhub_client import DEFAULT_CAPACITY_STATE_PATH, ModelHubClient, ModelHubClientPool from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker from runner_common import DEFAULT_KEY_PATH, ensure_tokens @@ -60,6 +68,11 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--skip-history-archive", action="store_true", help="Skip historical task archive download for this run") parser.add_argument("--dry-run", action="store_true", help="Plan the day without creating tasks") parser.add_argument("--disable-gpu-strategy", action="store_true", help="Disable adaptive 50/30/20 GPU scheduling") + parser.add_argument( + "--disable-market-intelligence", + action="store_true", + help="Disable live queue/throughput and public framework statistics", + ) parser.add_argument( "--gpu-strategy-refresh-submissions", type=int, @@ -88,6 +101,41 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--gpu-strategy-recent-window", type=int, default=1000, help=argparse.SUPPRESS) parser.add_argument("--gpu-strategy-min-long-samples", type=int, default=100, help=argparse.SUPPRESS) + parser.add_argument( + "--market-intelligence-state-path", + default=os.getenv("MODELHUB_MARKET_INTELLIGENCE_PATH", str(DEFAULT_MARKET_INTELLIGENCE_PATH)), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-queue-refresh-seconds", + type=int, + default=int(os.getenv("MODELHUB_MARKET_QUEUE_REFRESH_SECONDS", str(DEFAULT_QUEUE_REFRESH_SECONDS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-framework-refresh-seconds", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_REFRESH_SECONDS", str(DEFAULT_FRAMEWORK_REFRESH_SECONDS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-throughput-window-hours", + type=int, + default=int(os.getenv("MODELHUB_MARKET_THROUGHPUT_WINDOW_HOURS", str(DEFAULT_THROUGHPUT_WINDOW_HOURS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-fetch-workers", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FETCH_WORKERS", str(DEFAULT_FETCH_WORKERS))), + help=argparse.SUPPRESS, + ) + parser.add_argument( + "--market-framework-min-samples", + type=int, + default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES", str(DEFAULT_FRAMEWORK_MIN_SAMPLES))), + help=argparse.SUPPRESS, + ) parser.add_argument( "--capacity-state-path", default=os.getenv("MODELHUB_CAPACITY_STATE_PATH", str(DEFAULT_CAPACITY_STATE_PATH)), diff --git a/modelhub_submmit_api/task_registry.py b/modelhub_submmit_api/task_registry.py index 650da7fa..88175ed2 100644 --- a/modelhub_submmit_api/task_registry.py +++ b/modelhub_submmit_api/task_registry.py @@ -63,58 +63,69 @@ def task_specs_for_model(model: HFModelSummary) -> list[TaskSpec]: return [task for task in TASK_SPECS if pipeline_tag in task.pipeline_tags] -def choose_text_generation_framework(target_gpu: str, supported_frameworks: set[str], inspection: ModelInspection) -> str: +def compatible_text_generation_frameworks( + target_gpu: str, + supported_frameworks: set[str], + inspection: ModelInspection, +) -> list[str]: can_llamacpp = "llamacpp" in supported_frameworks vllm_like = [framework for framework in VLLM_LIKE_FRAMEWORKS if framework in supported_frameworks] can_transformers = "transformers" in supported_frameworks + compatible: list[str] = [] - if can_llamacpp and vllm_like: - if inspection.has_gguf: - return "llamacpp" - if inspection.has_vllm_weights: - return vllm_like[0] - raise ValueError(f"{target_gpu} supports both llamacpp and vLLM-like frameworks, but the model lacks usable weights") - if can_llamacpp: - if inspection.has_gguf: - return "llamacpp" - raise ValueError(f"{target_gpu} only supports llamacpp for this workflow, but no usable GGUF was found") - if vllm_like: - if inspection.has_vllm_weights: - return vllm_like[0] - raise ValueError(f"{target_gpu} only supports vLLM-like frameworks for this workflow, but no standard weights were found") - if can_transformers: - if inspection.has_vllm_weights: - return "transformers" - raise ValueError(f"{target_gpu} only supports transformers for this workflow, but no standard weights were found") - raise ValueError(f"No supported LLM framework template is available for {target_gpu}") + if can_llamacpp and inspection.has_gguf: + compatible.append("llamacpp") + if inspection.has_vllm_weights: + compatible.extend(vllm_like) + if can_transformers: + compatible.append("transformers") + if not compatible: + raise ValueError(f"No compatible LLM weights/framework combination is available for {target_gpu}") + return compatible -def choose_framework_for_task(task_type: str, target_gpu: str, supported_frameworks: set[str], inspection: ModelInspection) -> str: +def choose_text_generation_framework(target_gpu: str, supported_frameworks: set[str], inspection: ModelInspection) -> str: + return compatible_text_generation_frameworks(target_gpu, supported_frameworks, inspection)[0] + + +def compatible_frameworks_for_task( + task_type: str, + target_gpu: str, + supported_frameworks: set[str], + inspection: ModelInspection, +) -> list[str]: if task_type in {"text-generation", "visual-multi-modal", "reinforcement_learning"}: - return choose_text_generation_framework(target_gpu, supported_frameworks, inspection) + return compatible_text_generation_frameworks(target_gpu, supported_frameworks, inspection) if task_type == "asr": + compatible: list[str] = [] if "sherpa-onnx" in supported_frameworks and inspection.has_onnx_weights: - return "sherpa-onnx" - for framework in ("transformers", "funasr"): - if framework in supported_frameworks and inspection.has_standard_weights: - return framework + compatible.append("sherpa-onnx") + if inspection.has_standard_weights: + compatible.extend(framework for framework in ("transformers", "funasr") if framework in supported_frameworks) + if compatible: + return compatible raise ValueError(f"No compatible ASR framework found for {target_gpu}") if task_type == "feature_emb": - for framework in ("sentence-transformers", "transformers"): - if framework in supported_frameworks and inspection.has_standard_weights: - return framework + if inspection.has_standard_weights: + compatible = [framework for framework in ("sentence-transformers", "transformers") if framework in supported_frameworks] + if compatible: + return compatible raise ValueError(f"No compatible embedding framework found for {target_gpu}") if task_type in {"question_answering", "vision_classification", "text_classification"}: if "transformers" in supported_frameworks and inspection.has_standard_weights: - return "transformers" + return ["transformers"] raise ValueError(f"No compatible transformers template found for {task_type} on {target_gpu}") if task_type == "text-to-image-generation": if "diffusers" in supported_frameworks and inspection.has_standard_weights: - return "diffusers" + return ["diffusers"] raise ValueError(f"No compatible diffusers template found for {target_gpu}") raise ValueError(f"Unsupported task type for auto framework selection: {task_type}") + + +def choose_framework_for_task(task_type: str, target_gpu: str, supported_frameworks: set[str], inspection: ModelInspection) -> str: + return compatible_frameworks_for_task(task_type, target_gpu, supported_frameworks, inspection)[0] diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 3ae78e79..e657f703 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.02.5" +AGENT_VERSION = "2026.08.04.1" diff --git a/tests/test_gpu_strategy.py b/tests/test_gpu_strategy.py index 185c8170..5e6665f1 100644 --- a/tests/test_gpu_strategy.py +++ b/tests/test_gpu_strategy.py @@ -49,6 +49,15 @@ class HistoryClient: return list(self.tasks) +class WeightedMarket: + def gpu_weight(self, gpu: str, *, category: str) -> float: + del category + return {"gpu-a": 3.0, "gpu-b": 0.1, "gpu-c": 0.1, "gpu-d": 0.1}[gpu] + + def gpu_metadata(self, gpu: str) -> dict: + return {"marketWeight": self.gpu_weight(gpu, category=ALL_SUPPORTED)} + + class GPUStrategyTests(unittest.TestCase): def test_long_term_ranking_rejects_tiny_high_rate_sample(self) -> None: start = datetime(2026, 1, 1, tzinfo=timezone.utc) @@ -117,6 +126,28 @@ class GPUStrategyTests(unittest.TestCase): self.assertEqual(200, manager.state["acceptedTotal"]) self.assertEqual(0, manager.state["acceptedSinceRefresh"]) + def test_market_weights_change_gpu_mix_without_changing_50_30_20_split(self) -> None: + supported = ["gpu-a", "gpu-b", "gpu-c", "gpu-d"] + manager = GPUStrategyManager("unused.json", market_intelligence=WeightedMarket()) + manager.state = build_strategy_snapshot([], supported_gpus=supported) + candidates = [ + {"repoId": f"owner/model-{model}", "targetGpu": gpu} + for model in range(300) + for gpu in supported + ] + + ordered = manager.order_candidates(candidates)[:200] + category_counts = Counter(candidate["strategyCategory"] for candidate in ordered) + exploration_counts = Counter( + candidate["targetGpu"] + for candidate in ordered + if candidate["strategyCategory"] == ALL_SUPPORTED + ) + + self.assertEqual({LONG_TERM: 100, ALL_SUPPORTED: 60, RECENT: 40}, category_counts) + self.assertGreater(exploration_counts["gpu-a"], exploration_counts["gpu-b"]) + self.assertTrue(all(exploration_counts[gpu] > 0 for gpu in supported)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_market_intelligence.py b/tests/test_market_intelligence.py new file mode 100644 index 00000000..c0f5f206 --- /dev/null +++ b/tests/test_market_intelligence.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + + +PACKAGE_DIR = Path(__file__).resolve().parents[1] / "modelhub_submmit_api" +if str(PACKAGE_DIR) in sys.path: + sys.path.remove(str(PACKAGE_DIR)) +sys.path.insert(0, str(PACKAGE_DIR)) + +from main import choose_candidate_for_gpu # noqa: E402 +from market_intelligence import MarketIntelligenceManager # noqa: E402 +from models import HFModelSummary, ModelInspection # noqa: E402 +from template_selector import TemplateSelector # noqa: E402 + + +class PublicMarketClient: + def __init__(self) -> None: + self.calls = 0 + self.counts = { + "fast": {"waiting": 100, "running": 8, "completed": 120, "success": 80, "failed": 0}, + "slow": {"waiting": 10, "running": 1, "completed": 1, "success": 1, "failed": 0}, + "stopped": {"waiting": 500, "running": 0, "completed": 0, "success": 0, "failed": 0}, + } + + def list_machine_info(self) -> list[dict]: + self.calls += 1 + return [ + {"gpuType": "fast", "canVerify": True, "maxConcurrentTasks": 8}, + {"gpuType": "slow", "canVerify": True, "maxConcurrentTasks": 1}, + {"gpuType": "stopped", "canVerify": False, "maxConcurrentTasks": 0}, + ] + + def list_tasks_page(self, **kwargs) -> dict: # noqa: ANN003 + self.calls += 1 + row = self.counts[kwargs["gpu_type"]] + status = kwargs.get("status") + verify_result = kwargs.get("verify_result") + if status == "waiting": + total = row["waiting"] + elif status == "running": + total = row["running"] + elif status == "failed": + total = row["failed"] + elif status == "success" and verify_result == 1: + total = row["success"] + else: + total = row["completed"] + return {"data": {"total": total, "records": []}} + + def list_framework_stats(self, task_type: str, target_gpu: str) -> list[dict]: + self.calls += 1 + del task_type, target_gpu + return [ + {"framework": "vllm", "modelCount": 1000, "successCount": 100}, + {"framework": "transformers", "modelCount": 1000, "successCount": 300}, + ] + + def get_build_config(self, task_type: str, target_gpu: str, framework: str) -> str: + self.calls += 1 + del task_type, target_gpu + return ( + f"framework: {framework}\n" + "sut_config:\n gpu_num: 1\n values: {}\n" + "ref_config:\n gpu_num: 1\n values: {}\n" + ) + + +class FailingMarketClient: + def __init__(self) -> None: + self.calls = 0 + + def list_machine_info(self) -> list[dict]: + self.calls += 1 + raise RuntimeError("temporary queue outage") + + def list_tasks_page(self, **_kwargs) -> dict: # noqa: ANN003 + self.calls += 1 + raise RuntimeError("temporary queue outage") + + def list_framework_stats(self, task_type: str, target_gpu: str) -> list[dict]: + self.calls += 1 + del task_type, target_gpu + raise RuntimeError("temporary framework outage") + + +class MarketIntelligenceTests(unittest.TestCase): + def test_expected_throughput_beats_short_raw_queue_and_stopped_gpu_is_demoted(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + client = PublicMarketClient() + manager = MarketIntelligenceManager( + Path(temporary_dir) / "market.json", + throughput_window_hours=6, + log_fn=lambda _message: None, + ) + state = manager.prepare( + client, + supported_gpus=["fast", "slow", "stopped"], + task_types=["text-generation"], + now=datetime(2026, 8, 4, tzinfo=timezone.utc), + ) + + fast = state["gpuStats"]["fast"] + slow = state["gpuStats"]["slow"] + stopped = state["gpuStats"]["stopped"] + self.assertGreater(fast["throughputPerHour"], slow["throughputPerHour"]) + self.assertLess(fast["backlogHours"], slow["backlogHours"]) + self.assertGreater(fast["selectionWeight"], slow["selectionWeight"]) + self.assertLessEqual(stopped["selectionWeight"], 0.1) + self.assertTrue( + state["frameworkStats"]["text-generation"]["fast"]["transformers"]["officialConfigValid"] + ) + + def test_snapshot_cache_prevents_repeated_public_api_scans(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + client = PublicMarketClient() + manager = MarketIntelligenceManager(Path(temporary_dir) / "market.json", log_fn=lambda _message: None) + started = datetime(2026, 8, 4, tzinfo=timezone.utc) + manager.prepare( + client, + supported_gpus=["fast", "slow"], + task_types=["text-generation"], + now=started, + ) + first_call_count = client.calls + manager.prepare( + client, + supported_gpus=["fast", "slow"], + task_types=["text-generation"], + now=started + timedelta(seconds=30), + ) + self.assertEqual(first_call_count, client.calls) + + def test_market_outage_falls_back_to_neutral_and_uses_retry_backoff(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + client = FailingMarketClient() + manager = MarketIntelligenceManager(Path(temporary_dir) / "market.json", log_fn=lambda _message: None) + started = datetime(2026, 8, 4, tzinfo=timezone.utc) + state = manager.prepare( + client, + supported_gpus=["gpu"], + task_types=["text-generation"], + now=started, + ) + self.assertEqual(1.0, state["gpuStats"]["gpu"]["selectionWeight"]) + self.assertIsNotNone(state["queueError"]) + self.assertIsNotNone(state["frameworkError"]) + first_call_count = client.calls + manager.prepare( + client, + supported_gpus=["gpu"], + task_types=["text-generation"], + now=started + timedelta(seconds=30), + ) + self.assertEqual(first_call_count, client.calls) + + def test_framework_ranking_uses_confidence_bound_and_ignores_tiny_samples(self) -> None: + manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) + manager.state = { + "frameworkStats": { + "text-generation": { + "gpu": { + "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.10}, + "vllm-mlu": {"modelCount": 800, "wilsonLowerBound": 0.35}, + "vllm-customized": {"modelCount": 2, "wilsonLowerBound": 0.90}, + } + } + } + } + ranked = manager.rank_frameworks( + task_type="text-generation", + target_gpu="gpu", + compatible_frameworks=["vllm", "vllm-customized", "vllm-mlu"], + ) + self.assertEqual(["vllm-mlu", "vllm", "vllm-customized"], ranked) + + def test_local_account_evidence_is_blended_without_overriding_sample_guards(self) -> None: + manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) + manager.state = { + "frameworkStats": { + "text-generation": { + "gpu": { + "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.30}, + "vllm-mlu": {"modelCount": 1000, "wilsonLowerBound": 0.35}, + } + } + } + } + manager.set_local_outcome_stats( + { + "combinationStats": { + "gpu|vllm|text-generation": { + "successCount": 20, + "failureCount": 0, + } + } + } + ) + ranked = manager.rank_frameworks( + task_type="text-generation", + target_gpu="gpu", + compatible_frameworks=["vllm", "vllm-mlu"], + ) + self.assertEqual("vllm", ranked[0]) + metadata = manager.framework_metadata("text-generation", "gpu", "vllm") + self.assertEqual(20, metadata["frameworkLocalSamples"]) + self.assertGreater(metadata["frameworkCombinedScore"], 0.35) + + def test_candidate_uses_best_supported_public_framework(self) -> None: + manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) + manager.state = { + "frameworkStats": { + "text-generation": { + "Cambricon_mlu-370-x4": { + "vllm": { + "modelCount": 1000, + "successCount": 100, + "successRate": 0.10, + "wilsonLowerBound": 0.08, + }, + "vllm-mlu": { + "modelCount": 1000, + "successCount": 400, + "successRate": 0.40, + "wilsonLowerBound": 0.37, + }, + } + } + } + } + candidate = choose_candidate_for_gpu( + model=HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=None, + pipeline_tag="text-generation", + ), + inspection=ModelInspection(repo_id="owner/model", weight_files=["model.safetensors"]), + template_selector=TemplateSelector(), + task_types=["text-generation"], + target_gpu="Cambricon_mlu-370-x4", + market_intelligence=manager, + ) + self.assertIsNotNone(candidate) + assert candidate is not None + self.assertEqual("vllm-mlu", candidate.framework) + self.assertAlmostEqual(0.37, candidate.score) + + def test_new_framework_is_discovered_but_only_wins_on_qualified_success_score(self) -> None: + official_config = ( + "framework: future-engine\n" + "sut_config:\n gpu_num: 1\n values: {}\n" + "ref_config:\n gpu_num: 1\n values: {}\n" + ) + manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) + manager.state = { + "frameworkStats": { + "text-generation": { + "Cambricon_mlu-370-x4": { + "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.10}, + "future-engine": { + "modelCount": 1000, + "successCount": 500, + "successRate": 0.50, + "wilsonLowerBound": 0.47, + "officialConfigValid": True, + "officialConfig": official_config, + }, + } + } + } + } + candidate = choose_candidate_for_gpu( + model=HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=None, + pipeline_tag="text-generation", + ), + inspection=ModelInspection(repo_id="owner/model", weight_files=["model.safetensors"]), + template_selector=TemplateSelector(), + task_types=["text-generation"], + target_gpu="Cambricon_mlu-370-x4", + market_intelligence=manager, + ) + + self.assertIsNotNone(candidate) + assert candidate is not None + self.assertEqual("future-engine", candidate.framework) + self.assertEqual(official_config, candidate.config_params) + self.assertIn("official_build_config_synced", candidate.warnings) + + def test_tiny_new_framework_sample_does_not_displace_safe_legacy_framework(self) -> None: + manager = MarketIntelligenceManager("unused.json", framework_min_samples=100) + manager.state = { + "frameworkStats": { + "text-generation": { + "Cambricon_mlu-370-x4": { + "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.10}, + "future-engine": { + "modelCount": 2, + "wilsonLowerBound": 0.90, + "officialConfigValid": True, + "officialConfig": ( + "framework: future-engine\n" + "sut_config:\n gpu_num: 1\n values: {}\n" + "ref_config:\n gpu_num: 1\n values: {}\n" + ), + }, + } + } + } + } + candidate = choose_candidate_for_gpu( + model=HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=None, + pipeline_tag="text-generation", + ), + inspection=ModelInspection(repo_id="owner/model", weight_files=["model.safetensors"]), + template_selector=TemplateSelector(), + task_types=["text-generation"], + target_gpu="Cambricon_mlu-370-x4", + market_intelligence=manager, + ) + self.assertIsNotNone(candidate) + assert candidate is not None + self.assertEqual("vllm", candidate.framework) + + +if __name__ == "__main__": + unittest.main()