paopaotang/xc-adapt-agent-2174
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

add main.py

Browse Source
This commit is contained in:
paopaotang
2026-10-20 10:10:15 +08:00
parent 8e816ec1e5
commit 47c43c4934
1 changed files with 145 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

249
main.py
View File

@@ -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,
}
_state["phase"] = "submitting"
_state["started_at"] = datetime.utcnow().isoformat()
for i in range(1, MAX_ATTEMPTS + 1):
if _shutdown.is_set():
break
_state["attempts"] = i
try:
resp = requests.post(BASE_URL + ADD_TASK_ENDPOINT, headers=HEADERS,
json=payload, timeout=60)
json=payload, timeout=90)
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)
if code == 0:
data = body.get("data") or {}
_state["task_id"] = data.get("id")
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, task in enumerate(TASKS, 1):
if _shutdown.is_set():
break
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:
_state["submitted"] += 1
else:
_state["failed"] += 1
if i < len(TASKS):
time.sleep(GAP)
_state["phase"] = "done"
_state["finished_at"] = datetime.utcnow().isoformat()
_submitted = True
print("[submit] 提交成功 taskId=%s" % _state["task_id"], flush=True)
return True
# 已存在/重复等不可重试错误，直接退出
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["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__":