xc_validation_strategy/main.py

"""
xc_validation_strategy — 主入口

启动后针对 GPU_JOBS 中配置的 GPU 卡分别批量提交各自筛选出的模型验证任务
（当前仅提交 ppu_zw_810e，其余 4 张卡 Biren_166m/Cambricon_mlu-370-x8/MetaX_c-500/
Kunlunxin_p-800 的 config_content 模板和模型列表仍保留在代码中，未列入本次 GPU_JOBS）
（/adminApi/async/task/create-contest-task，
Bearer Token 认证），之后保持 HTTP 服务存活。

账号额度自动重试：如果某个模型提交时命中"当前等待中或运行中的异步模型验证
任务数量已达上限"（账号额度已满），不算永久失败，会被留到下一轮；额度耗尽后
本进程会原地等待 30 分钟，再自动重试所有因额度问题未提交成功的模型，如此循环，
直至全部提交成功或进程被平台关闭——不需要重新部署新策略，循环逻辑在本进程内完成。
非额度原因的失败（如模型已在验证中等）不会重试。

同时暴露 /health（K8s 探活）和 /status（运行状态，含当前轮次/待重试数/下次重试时间）。
"""

import json
import os
import signal
import threading
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import List, Tuple

import requests

# ══════════════════════════════════════════════════════════
#  配置（全部从环境变量读取，不硬编码敏感信息）
# ══════════════════════════════════════════════════════════
BASE_URL           = os.environ.get("BASE_URL", "https://modelhub.org.cn")
SUBMIT_ENDPOINT    = "/adminApi/async/task/create-contest-task"

# 通过 curl -X POST https://modelhub.org.cn/adminApi/user/login 获取后填入
AUTH_TOKEN        = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyQWNjb3VudCI6Inpob3VzaGFzaGEiLCJpZCI6MTQsInVzZXJSb2xlIjoibGVhZGVyYm9hcmQiLCJleHAiOjE3OTAwNjkxMjAsImlhdCI6MTc4OTQ2NDMyMH0.KzJac6ddaZdtLvjD6ZnoK1PNEKFoXdyDn9Hh4FxU9ic"
CONTEST_API_TOKEN = "ef1ef82f3c9efee413d602345fbe224d"
CONTRIBUTORS      = "zhoushasha"
TASK_TYPE         = "text-generation"
STRATEGY_ID       = os.environ.get("STRATEGY_ID", "")  # 平台自动注入，无需修改

HTTP_HOST = "0.0.0.0"
HTTP_PORT = 8080

# ══════════════════════════════════════════════════════════
#  各 GPU 的模型列表（来自 filter_verified_models 脚本的筛选结果）
# ══════════════════════════════════════════════════════════
BIREN_MODELS = [
    "zipaltrivedi/dotnet-coder-14b",
]

CAMBRICON_MODELS = [
]

METAX_MODELS = [
    "zipaltrivedi/dotnet-coder-14b",
]

HYGON_MODELS = [
    "zipaltrivedi/dotnet-coder-14b",
    "cds-jb/qwen3-14b-butterfly-subliminal-fullft",
    "lllqaq/Qwen2.5-Coder-14B-Instruct-num11-v1-v2-v3-pairs-v3-triples-post-r2egym",
    "deepmako/Mako-32B-Conductor",
]

KUNLUNXIN_MODELS = [
]

PPU_MODELS = [
    "fpadovani/eng-latn-100mb-after-ppt-Dp-10mb-ckpt500_seed10",
    "fpadovani/ita-latn-10mb-after-ppt-Dp-100mb-ckpt500_seed3407",
    "fpadovani/ita-latn-10mb-after-ppt-Dp-10mb-ckpt500_seed3407",
    "richardr1126/spider-skeleton-wizard-coder-merged",
    "fpadovani/dan-latn-10mb-after-ppt-shuff-dyck-10mb-ckpt500_seed3407",
    "fpadovani/ita-latn-10mb-after-ppt-shuff-dyck-100mb-ckpt500_seed3407",
    "omerkaragulmez/XbyK-0.1",
    "fpadovani/eng-latn-10mb-after-ppt-Dp-10mb-ckpt500_seed10",
    "fpadovani/eng-latn-10mb-100mb_seed10",
    "flax-community/gpt2-medium-indonesian",
    "RedHatAI/QwQ-32B-Preview-quantized.w8a8",
    "fractalego/fact-checking",
    "KoboldAI/GPT-J-6B-Adventure",
    "EasierAI/Falcon-3-1B",
    "theprint/mistral-7b-cthulhu",
    "iwalton3/phoenix",
    "renzhenzhen/internLM2-for-triples",
    "u2mithrandir/epsi_tmall",
]

# 本轮提交：ppu_zw_810e(18) / hygon_k100-ai(4) / MetaX_c-500(1) / Biren_166m(1)，共24个。
#
# 候选池是 api_verify_model_download_status_a.txt 里那 24711 个【已下载】的模型，
# 正好满足机制A「模型必须已下载到平台存储」的前提（v1.0.37 那次157个失败就是因为没下载）。
#
# 本轮严格口径与放宽口径（别的卡「已验证」vs「已验证或验证中」）结果完全相同：
# 这批已下载模型里 24007 个有验证记录的，24006 个都已至少一张卡「已验证」，
# 放宽只多捞出 1 个；真正的瓶颈是各卡「无记录」的模型太少——
# Kunlunxin_p-800 / Cambricon_mlu-370-x8 / Iluvatar_bi-150 均为 0，故不列入 GPU_JOBS。
GPU_JOBS: List[Tuple[str, List[str]]] = [
    ("ppu_zw_810e",          PPU_MODELS),
    ("hygon_k100-ai",        HYGON_MODELS),
    ("MetaX_c-500",          METAX_MODELS),
    ("Biren_166m",           BIREN_MODELS),
]
TOTAL_MODELS = sum(len(models) for _, models in GPU_JOBS)

# ══════════════════════════════════════════════════════════
#  全局状态（供 /status 展示）
# ══════════════════════════════════════════════════════════
_state = {
    "strategy_id":   STRATEGY_ID,
    "phase":         "starting",   # starting | submitting | waiting_retry | done | error
    "total":         TOTAL_MODELS,
    "submitted":     0,
    "failed":        0,
    "per_gpu":       {gpu: 0 for gpu, _ in GPU_JOBS},
    "started_at":    None,
    "finished_at":   None,
    "round":               0,     # 当前是第几轮提交
    "quota_blocked_remaining": 0, # 因额度上限暂未提交成功、等待下一轮重试的模型数
    "next_retry_at":       None,  # 下一轮重试的预计时间（额度耗尽等待期间）
}
_shutdown = threading.Event()

# ══════════════════════════════════════════════════════════
#  HTTP 服务
# ══════════════════════════════════════════════════════════
class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self._json({"status": "ok"})
        elif self.path == "/status":
            self._json(_state)
        else:
            self._json({"error": "not found"}, 404)

    def _json(self, body: dict, code: int = 200):
        payload = json.dumps(body, default=str).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, fmt, *args):
        print(f"[http] {self.address_string()} {fmt % args}", flush=True)


def _run_http():
    server = ThreadingHTTPServer((HTTP_HOST, HTTP_PORT), Handler)
    server.timeout = 1
    print(f"[http] 监听 {HTTP_HOST}:{HTTP_PORT}", flush=True)
    while not _shutdown.is_set():
        server.handle_request()
    server.server_close()
    print("[http] 已关闭", flush=True)

# ══════════════════════════════════════════════════════════
#  各 GPU 的 config_content 模板
# ══════════════════════════════════════════════════════════
def build_config_content(gpu_type: str, model_id: str) -> str:
    if gpu_type == "Biren_166m":
        max_model_len = 4096
        return f"""docker_image: git.modelhub.org.cn:9443/enginex/xc-llm-biren166m:26.01
nv_docker_image: harbor.4pd.io/dooke/vllm/vllm/vllm-openai:v0.11.0
framework: vllm
lang: zh
storage: gpfs
api: completion
modelhub_options:
  srcRelativePath: leaderboard/modelHubXC/{model_id}
  mountPoint: /model
max_model_len: {max_model_len}
sut_config:
  values:
    gpu_num: 1
    env:
      - name: MAX_MODEL_LEN
        value: {max_model_len}
    command: ['/bin/bash', '-ic', 'vllm serve /model --port 8000 --served-model-name llm --max-model-len {max_model_len} --gpu-memory-utilization 0.9 --enforce-eager --trust-remote-code -tp 1 --host 0.0.0.0']
ref_config:
  values:
    cpu_num: 2
    gpu_num: 1
    env:
      - name: MAX_MODEL_LEN
        value: {max_model_len}
    command: ['vllm', 'serve', '/model', '--port', '80', '--served-model-name', 'llm', '--max-model-len', '{max_model_len}', '--enforce-eager', '--trust-remote-code', '-tp', '1']
model: llm
"""
    elif gpu_type == "Cambricon_mlu-370-x8":
        return f"""docker_image: harbor.4pd.io/hardcore-tech/cambricon-mlu370-pytorch:v25.01-torch2.5.0-torchmlu1.24.1-ubuntu22.04-py310
nv_docker_image: harbor.4pd.io/dooke/vllm/vllm/vllm-openai:v0.11.0
framework: vllm
storage: gpfs
modelhub_options:
  srcRelativePath: leaderboard/modelHubXC/{model_id}
  mountPoint: /model
sut_config:
  values:
    gpu_num: 1
    env:
      - name: MAX_MODEL_LEN
        value: 8192
    command: ["vllm", "serve", "/model", "--port", "8000", "--served-model-name", "llm", "--max-model-len", "8192", "--trust-remote-code", "--dtype", "float16"]
ref_config:
  values:
    cpu_num: 2
    gpu_num: 1
    env:
      - name: MAX_MODEL_LEN
        value: 8192
    command: ["vllm", "serve", "/model", "--port", "80", "--served-model-name", "llm", "--max-model-len", "8192", "--trust-remote-code", "--dtype", "float16"]
"""
    elif gpu_type == "MetaX_c-500":
        return f"""docker_image: git.modelhub.org.cn:9443/enginex-metax/vllm:0.9.1
nv_docker_image: harbor.4pd.io/dooke/vllm/vllm/vllm-openai:v0.11.0
framework: vllm
lang: en
storage: gpfs
api: chat
modelhub_options:
  srcRelativePath: leaderboard/modelHubXC/{model_id}
  mountPoint: /model
max_model_len: 2048
sut_config:
  gpu_num: 1
  values:
    command: ['/opt/conda/bin/vllm', 'serve', '/model', '--port', '20644', '--served-model-name', 'llm',  '--max-model-len', '2048', '--gpu-memory-utilization', '0.9',  '--enforce-eager', '--trust-remote-code' ,'-tp', '1']
ref_config:
  gpu_num: 1
  values:
    command: ['vllm', 'serve', '/model', '--port', '80', '--served-model-name', 'llm', '--max-model-len', '2048', '--enforce-eager', '--trust-remote-code', '-tp', '1']
"""
    elif gpu_type == "Kunlunxin_p-800":
        return f"""docker_image: git.modelhub.org.cn:9443/enginex/xc-llm-kunlun
nv_docker_image: harbor.4pd.io/dooke/vllm/vllm/vllm-openai:v0.11.0
framework: vllm
lang: en
storage: gpfs
api: chat
temperature: 0.4
repetition_penalty: 1.1
top_p: 0.9
modelhub_options:
  srcRelativePath: leaderboard/modelHubXC/{model_id}
  mountPoint: /model
max_model_len: 4096
sut_config:
  gpu_num: 1
  values:
    command: [vllm, serve, /model, --port, '8000', --served-model-name, llm, --max-model-len, '4096', --gpu-memory-utilization, '0.9', --enforce-eager, --trust-remote-code, -tp, '1']
ref_config:
  gpu_num: 1
  values:
    command: [vllm, serve, /model, --port, '80', --served-model-name, llm, --max-model-len, '4096', --enforce-eager, --trust-remote-code, -tp, '1']
"""
    elif gpu_type == "hygon_k100-ai":
        return f"""
docker_image: harbor.4pd.io/modelhubxc/enginex-hygon/vllm:0.9.2-patch-tokenizer
nv_docker_image: harbor.4pd.io/modelhubxc/enginex-nvidia/vllm:0.11.0-patch-tokenizer
framework: vllm
storage: gpfs

max_model_len: 4096
sut_config:
    gpu_num: 1
    values:
        command: ['vllm', 'serve', '/model', '--port', '20644', '--served-model-name', 'llm',  '--max-model-len', '4096', '--enforce-eager', '--trust-remote-code' ,'-tp', '1' ]
ref_config:
    gpu_num: 1
    values:
        command: ['vllm', 'serve', '/model', '--port', '80', '--served-model-name', 'llm', '--max-model-len', '4096', '--enforce-eager', '--trust-remote-code', '-tp', '1']
"""
    elif gpu_type == "ppu_zw_810e":
        return f"""gpu_type: ppu_zw_810e
framework: vllm
docker_image: harbor.4pd.io/hardcore-tech/asllm:1.10.1-pytorch2.10.0-ubuntu24.04-sail2.1.0-cuda13.0-sglang0.5.10-vllm0.19.0-py312
nv_docker_image: harbor-contest.4pd.io/sunruoxi/vllm-openai-fix-tokenizer:v0.11.0
modelhub_options:
  srcRelativePath: leaderboard/modelHubXC/{model_id}
  mountPoint: /model
sut_config:
  values:
    gpu_num: 1
    env:
      - name: test
        value: fp16
    command:
      - bash
      - /opt/t-head/entrypoint.sh
      - python3
      - -m
      - asllm.entrypoints.api_server
      - --model
      - /model
      - --port
      - '30000'
      - --host
      - 0.0.0.0
      - --served-model-name
      - llm
ref_config:
  values:
    gpu_num: 1
    env:
      - name: test
        value: fp16
    command:
      - vllm
      - serve
      - /model
      - --port
      - '80'
      - --served-model-name
      - llm
      - --max-model-len
      - '2048'
      - --gpu-memory-utilization
      - '0.9'
      - --enforce-eager
      - --trust-remote-code
      - -tp
      - '1'
"""
    
    elif gpu_type == "Iluvatar_bi-150":
            return f"""docker_image: harbor-contest.4pd.io/luopingyi/enginex-iluvatar-bi150/vllm:0.8.3
nv_docker_image: harbor.4pd.io/dooke/vllm/vllm/vllm-openai:v0.11.0
framework: vllm
api: completion
temperature: 0.7
repetition_penalty: 1.2
top_p: 0.9

max_model_len: 4096
max_tokens: 1024
sut_config:
  gpu_num: 1
  values:
    command: ['vllm', 'serve', '/model', '--port', '80', '--served-model-name', 'llm', '--max-model-len', '4096', '--gpu-memory-utilization', '0.9', '--enforce-eager', '--trust-remote-code' ,'-tp', '1']
ref_config:
  gpu_num: 1
  values:
    command: ['vllm', 'serve', '/model', '--port', '80', '--served-model-name', 'llm', '--max-model-len', '4096', '--enforce-eager', '--trust-remote-code', '-tp', '1']  
"""
    
    else:
        raise ValueError(f"未知的 GPU_TYPE: {gpu_type}")
    
    
    
    

# ══════════════════════════════════════════════════════════
#  业务逻辑
# ══════════════════════════════════════════════════════════
# 账号"等待中/运行中"任务数已达上限时平台返回的业务错误信息（子串匹配）；
# 命中这个的模型不算永久失败，会在额度腾出空位后自动重试，不会被记作 failed
QUOTA_FULL_MSG = "当前等待中或运行中的异步模型验证任务数量已达上限"
# 额度耗尽后，隔多久自动重试一次剩余（因额度问题未提交成功）的模型
RETRY_INTERVAL_SECONDS = 30 * 60  # 30 分钟


def _submit_task(token: str, gpu_type: str, model_id: str) -> Tuple[bool, str, str]:
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}",
    }
    config_content = build_config_content(gpu_type, model_id)

    payload = {
        "contestApiToken": CONTEST_API_TOKEN,
        "contributors":    CONTRIBUTORS,
        "gpuTypes":        [gpu_type],
        "taskType":        TASK_TYPE,
        "modelId":         model_id,
        "framework":       "vllm",
        "strategyId":      STRATEGY_ID,   # 平台要求
        "submissionConfig": [{
            "config":   config_content,
            "gpuType":  gpu_type,
            "taskType": TASK_TYPE,
        }],
    }
    print(f"[payload] gpu={gpu_type} model={model_id}", flush=True)
    try:
        resp = requests.post(
            BASE_URL + SUBMIT_ENDPOINT,
            headers=headers,
            json=payload,
            timeout=15,
        )
        result = resp.json()
        if result.get("code") == 0:
            task_id = result.get("data", {}).get("id", "")
            print(f"[worker] OK  {model_id} (GPU={gpu_type})  task_id={task_id}", flush=True)
            return True, task_id, ""
        else:
            message = result.get("message") or ""
            print(f"[worker] FAIL {model_id} (GPU={gpu_type}): {message}", flush=True)
            return False, "", message
    except Exception as e:
        print(f"[worker] ERROR {model_id} (GPU={gpu_type}): {e}", flush=True)
        return False, "", str(e)


def _run_worker():
    _state["started_at"] = datetime.utcnow().isoformat()
    _state["phase"] = "submitting"

    successful: List[Tuple[str, str, str]] = []
    token = AUTH_TOKEN
    print("[worker] 使用预设 Token，跳过登录", flush=True)

    # 待提交队列：保持 GPU_JOBS 里原有的 (gpu_type, model_id) 顺序
    pending: List[Tuple[str, str]] = [
        (gpu_type, model_id)
        for gpu_type, model_list in GPU_JOBS
        for model_id in model_list
    ]

    round_num = 0
    while pending and not _shutdown.is_set():
        round_num += 1
        _state["round"] = round_num
        _state["phase"] = "submitting"
        _state["next_retry_at"] = None
        print(
            f"\n{'='*60}\n🚀 第 {round_num} 轮，待提交 {len(pending)} 个模型\n{'='*60}",
            flush=True,
        )

        quota_blocked: List[Tuple[str, str]] = []
        for gpu_type, model_id in pending:
            if _shutdown.is_set():
                break
            ok, task_id, message = _submit_task(token, gpu_type, model_id)
            if ok:
                _state["submitted"] += 1
                _state["per_gpu"][gpu_type] += 1
                successful.append((task_id, gpu_type, model_id))
            elif QUOTA_FULL_MSG in message:
                # 账号额度暂时满了，不算永久失败，留到下一轮重试
                quota_blocked.append((gpu_type, model_id))
            else:
                # 非额度原因失败（如重复提交等），不再重试
                _state["failed"] += 1

        pending = quota_blocked
        _state["quota_blocked_remaining"] = len(pending)

        # 每轮结束都把已成功的结果落盘一次，避免中途重启丢失记录
        try:
            with open("submitted_validation_tasks.txt", "w", encoding="utf-8") as f:
                for tid, gpu, mid in successful:
                    f.write(f"{tid}\t{gpu}\t{mid}\n")
        except Exception:
            pass

        if pending and not _shutdown.is_set():
            next_retry = datetime.utcnow().timestamp() + RETRY_INTERVAL_SECONDS
            _state["next_retry_at"] = datetime.utcfromtimestamp(next_retry).isoformat()
            _state["phase"] = "waiting_retry"
            print(
                f"[worker] 第 {round_num} 轮结束：{len(pending)} 个模型因账号额度上限暂未提交，"
                f"{RETRY_INTERVAL_SECONDS // 60} 分钟后自动重试（不部署新策略，本进程内循环）...",
                flush=True,
            )
            _shutdown.wait(RETRY_INTERVAL_SECONDS)

    _state["finished_at"] = datetime.utcnow().isoformat()
    _state["phase"] = "done"
    _state["quota_blocked_remaining"] = len(pending)
    print(
        f"[worker] 完成  submitted={_state['submitted']}  failed={_state['failed']}  "
        f"total={_state['total']}  per_gpu={_state['per_gpu']}  "
        f"仍因额度未提交(如遇shutdown中断)={len(pending)}",
        flush=True,
    )
    # 提交完成后继续保持进程存活，等待平台停止

# ══════════════════════════════════════════════════════════
#  入口
# ══════════════════════════════════════════════════════════
def _handle_signal(signum, _frame):
    print(f"[main] 收到信号 {signum}，正在关闭...", flush=True)
    _shutdown.set()


def main():
    signal.signal(signal.SIGTERM, _handle_signal)
    signal.signal(signal.SIGINT,  _handle_signal)

    # HTTP 服务线程
    http_thread = threading.Thread(target=_run_http, daemon=False)
    http_thread.start()

    # 提交任务线程
    worker_thread = threading.Thread(target=_run_worker, daemon=True)
    worker_thread.start()

    # 主线程等待 shutdown
    _shutdown.wait()
    print("[main] 等待 HTTP 服务关闭...", flush=True)
    http_thread.join(timeout=5)
    print("[main] 退出", flush=True)


if __name__ == "__main__":
    main()