diff --git a/main.py b/main.py index 2cc248e..37bee5f 100644 --- a/main.py +++ b/main.py @@ -1,16 +1,14 @@ -"""ModelHub XC 适配智能体 —— 单任务提交器 +"""ModelHub XC 适配智能体 —— 多任务提交器（本批次 4 条） -启动后向 ModelHub XC 提交 1 条模型适配验证任务，随后保持 /health 存活等待平台停止。 +启动后依次提交 4 条已查重的模型适配验证任务，随后保持 /health 存活等待平台停止。 -提交内容（已查重，非重复模型）： - 模型：https://www.modelscope.cn/models/qnguyen3/qwq-1.5b-grpo (Qwen2 / 1.5B) - 算力：Ascend_910-b4 框架：vllm 任务类型：text-generation +批次清单（提交前均查过：模型列表 0 条 / 全站任务 0 条 / 详情接口无验证记录）： + 1. mrm8488/DeepSeek-R1-Distill-Qwen-1.5B-grpo-limo-2e Qwen2 1.5B Ascend_910-b4 vllm + 2. HollowInky/skill-search-qwen3-4b-unmasked-grpo Qwen3 4B Ascend_910-b4 vllm + 3. dmis-lab/OSP-1.4B-100B-Muon Llama 1.4B Ascend_910-b4 vllm + 4. qnguyen3/quan-1.8b-chat Llama 1.8B Ascend_910-b3 vllm -平台契约： - - 根目录 Dockerfile，监听 8080 - - GET /health 探活 - - 读取平台注入的 STRATEGY_ID - - 正确处理 SIGTERM +平台契约：根目录 Dockerfile，监听 8080，GET /health 探活，读取 STRATEGY_ID，处理 SIGTERM。 """ import json import os @@ -22,20 +20,13 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import requests -# ─────────────────────────── 业务配置 ─────────────────────────── BASE_URL = os.environ.get("BASE_URL", "https://modelhub.org.cn") ADD_TASK_ENDPOINT = "/api/adapt/task/add" - XC_TOKEN = os.environ.get("XC_TOKEN", "6616dfe75c874e61987b1f7c908e6d77") STRATEGY_ID = os.environ.get("STRATEGY_ID", "") -MODEL_ADDRESS = "https://www.modelscope.cn/models/qnguyen3/qwq-1.5b-grpo" -TASK_TYPE = "text-generation" -TARGET_GPU = "Ascend_910-b4" -FRAMEWORK = "vllm" - -# 与 POST /api/adapt/task/build-config?gpuType=Ascend_910-b4&framework=vllm&taskType=text-generation 输出一致 -CONFIG_PARAMS = """framework: vllm +# 910B4 模板（max_model_len 2048） +CFG_B4 = """framework: vllm api: completion max_tokens: 1024 temperature: 0.7 @@ -84,95 +75,151 @@ ref_config: - -tp - '1' - --enforce-eager - - --trust-remote-code -""" + - --trust-remote-code""" +# 910B3 模板（max_model_len 4096） +CFG_B3 = """framework: vllm +api: completion +lang: en +max_model_len: 4096 +max_tokens: 1024 +temperature: 0.7 +repetition_penalty: 1.1 +top_p: 0.9 +sut_config: + gpu_num: 1 + values: + command: + - vllm + - serve + - /model + - --port + - '20644' + - --served-model-name + - llm + - --max-model-len + - '4096' + - -tp + - '1' + - --enforce-eager + - --trust-remote-code +ref_config: + gpu_num: 1 + values: + command: + - vllm + - serve + - /model + - --port + - '80' + - --served-model-name + - llm + - --max-model-len + - '4096' + - -tp + - '1' + - --enforce-eager + - --trust-remote-code""" + +TASKS = [ + { + "label": "DeepSeek-R1-Distill-Qwen-1.5B-grpo-limo-2e", + "model": "https://www.modelscope.cn/models/mrm8488/DeepSeek-R1-Distill-Qwen-1.5B-grpo-limo-2e", + "gpu": "Ascend_910-b4", + "config": CFG_B4, + }, + { + "label": "skill-search-qwen3-4b-unmasked-grpo", + "model": "https://www.modelscope.cn/models/HollowInky/skill-search-qwen3-4b-unmasked-grpo", + "gpu": "Ascend_910-b4", + "config": CFG_B4, + }, + { + "label": "OSP-1.4B-100B-Muon", + "model": "https://www.modelscope.cn/models/dmis-lab/OSP-1.4B-100B-Muon", + "gpu": "Ascend_910-b4", + "config": CFG_B4, + }, + { + "label": "quan-1.8b-chat", + "model": "https://www.modelscope.cn/models/qnguyen3/quan-1.8b-chat", + "gpu": "Ascend_910-b3", + "config": CFG_B3, + }, +] + +TASK_TYPE = "text-generation" +FRAMEWORK = "vllm" HEADERS = {"Content-Type": "application/json", "Xc-Token": XC_TOKEN} HTTP_HOST, HTTP_PORT = "0.0.0.0", 8080 -MAX_ATTEMPTS = 5 -RETRY_GAP = 15 # 秒 +GAP = 4 # 两条之间的间隔秒数 _state = { "strategy_id": STRATEGY_ID, - "phase": "starting", # starting | submitting | done | error - "model_address": MODEL_ADDRESS, - "target_gpu": TARGET_GPU, - "framework": FRAMEWORK, - "attempts": 0, - "attempt_log": [], - "task_id": None, + "phase": "starting", # starting | submitting | done + "total": len(TASKS), + "submitted": 0, + "failed": 0, + "results": [], "started_at": None, "finished_at": None, } _shutdown = threading.Event() -_submitting = threading.Lock() -_submitted = False -def submit_task(): - """提交一次；已提交过则直接返回。""" - global _submitted - with _submitting: - if _submitted: - return True - +def submit_one(task): payload = { - "configParams": CONFIG_PARAMS, + "configParams": task["config"], "framework": FRAMEWORK, - "modelAddress": MODEL_ADDRESS, - "targetGpu": TARGET_GPU, + "modelAddress": task["model"], + "targetGpu": task["gpu"], "taskType": TASK_TYPE, "strategyId": STRATEGY_ID, } + try: + resp = requests.post(BASE_URL + ADD_TASK_ENDPOINT, headers=HEADERS, + json=payload, timeout=90) + body = resp.json() + except Exception as e: # noqa: BLE001 + body = {"code": -1, "message": "exception: %s" % e} + code = body.get("code") + data = body.get("data") or {} + print("[submit] %s gpu=%s code=%s message=%s taskId=%s" + % (task["label"], task["gpu"], code, body.get("message"), data.get("id")), flush=True) + return code, body.get("message"), data.get("id") + + +def run_batch(): _state["phase"] = "submitting" _state["started_at"] = datetime.utcnow().isoformat() - for i in range(1, MAX_ATTEMPTS + 1): + for i, task in enumerate(TASKS, 1): if _shutdown.is_set(): break - _state["attempts"] = i - try: - resp = requests.post(BASE_URL + ADD_TASK_ENDPOINT, headers=HEADERS, - json=payload, timeout=60) - body = resp.json() - except Exception as e: # noqa: BLE001 - body = {"code": -1, "message": "exception: %s" % e} - - code = body.get("code") - msg = body.get("message", "") - entry = {"attempt": i, "time": datetime.utcnow().isoformat(), - "http_status": None, "code": code, "message": msg} - try: - entry["http_status"] = resp.status_code - except Exception: # noqa: BLE001 - pass - _state["attempt_log"].append(entry) - print("[submit] #%d code=%s message=%s" % (i, code, msg), flush=True) - + code, msg, task_id = submit_one(task) + _state["results"].append({ + "index": i, + "label": task["label"], + "model": task["model"], + "gpu": task["gpu"], + "code": code, + "message": msg, + "task_id": task_id, + "time": datetime.utcnow().isoformat(), + }) if code == 0: - data = body.get("data") or {} - _state["task_id"] = data.get("id") - _state["phase"] = "done" - _state["finished_at"] = datetime.utcnow().isoformat() - _submitted = True - print("[submit] 提交成功 taskId=%s" % _state["task_id"], flush=True) - return True + _state["submitted"] += 1 + else: + _state["failed"] += 1 + if i < len(TASKS): + time.sleep(GAP) - # 已存在/重复等不可重试错误，直接退出 - if code not in (-1, 60014, 50000): - _state["phase"] = "error" - _state["finished_at"] = datetime.utcnow().isoformat() - print("[submit] 终止：不可重试错误", flush=True) - return False - - if i < MAX_ATTEMPTS and not _shutdown.is_set(): - time.sleep(RETRY_GAP) - - _state["phase"] = "error" + _state["phase"] = "done" _state["finished_at"] = datetime.utcnow().isoformat() - return False + print("[batch] 完成 submitted=%d failed=%d total=%d" + % (_state["submitted"], _state["failed"], _state["total"]), flush=True) class Handler(BaseHTTPRequestHandler): @@ -184,13 +231,6 @@ class Handler(BaseHTTPRequestHandler): else: self._json({"error": "not found"}, 404) - def do_POST(self): - if self.path == "/task": - submit_task() - self._json(_state) - else: - self._json({"error": "not found"}, 404) - def _json(self, body, code=200): raw = json.dumps(body, ensure_ascii=False, default=str).encode() self.send_response(code) @@ -208,23 +248,6 @@ def _handle_signal(signum, _frame): _shutdown.set() -def main(): - signal.signal(signal.SIGTERM, _handle_signal) - signal.signal(signal.SIGINT, _handle_signal) - - print("[main] strategy_id=%s" % STRATEGY_ID, flush=True) - print("[main] model=%s gpu=%s framework=%s" % (MODEL_ADDRESS, TARGET_GPU, FRAMEWORK), flush=True) - - http_thread = threading.Thread(target=_run_http, daemon=False) - http_thread.start() - - worker = threading.Thread(target=submit_task, daemon=True) - worker.start() - - _shutdown.wait() - print("[main] 退出", flush=True) - - def _run_http(): server = ThreadingHTTPServer((HTTP_HOST, HTTP_PORT), Handler) server.timeout = 1 @@ -232,7 +255,23 @@ def _run_http(): while not _shutdown.is_set(): server.handle_request() server.server_close() - print("[http] 已关闭", flush=True) + + +def main(): + signal.signal(signal.SIGTERM, _handle_signal) + signal.signal(signal.SIGINT, _handle_signal) + + print("[main] strategy_id=%s tasks=%d" % (STRATEGY_ID, len(TASKS)), flush=True) + for t in TASKS: + print("[main] 计划: %s @ %s" % (t["label"], t["gpu"]), flush=True) + + http_thread = threading.Thread(target=_run_http, daemon=False) + http_thread.start() + + worker = threading.Thread(target=run_batch, daemon=True) + worker.start() + + _shutdown.wait() if __name__ == "__main__":