diff --git a/main.py b/main.py new file mode 100644 index 0000000..d929b3b --- /dev/null +++ b/main.py @@ -0,0 +1,210 @@ +"""ModelHub XC 适配智能体 (huni-adapt-agent) + +平台契约: 长驻 HTTP 服务, GET /health 返回 200, 处理 SIGTERM。 +赚分逻辑: 平台不喂模型、不给专属卡 (已由 agent_platform API 结构证实), +所以智能体必须自己用注入的 EXTERNAL_SERVICE_TOKEN 调主站 /api/adapt/task/add 提交。 +本体 = 一个 24h 自驱的提交循环, 只打天数活队列 (天垓150/100/智铠100) 的新鲜 NEW_ADAPT 名额。 + +安全: token 只从环境读、绝不打印/落盘; DRY_RUN 默认 true (不真提交, 只验证管道)。 +""" +import json, os, signal, time, threading, hashlib, base64 +import urllib.request, urllib.parse, urllib.error +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn") +GITEA = os.getenv("GITEA_HOST", "https://dev.modelhub.org.cn") +TOKEN = os.getenv("EXTERNAL_SERVICE_TOKEN", "") +DRY_RUN = os.getenv("DRY_RUN", "true").lower() == "true" +TARGET_CARDS = os.getenv("TARGET_CARDS", "Iluvatar_bi-150,Iluvatar_bi-100,Iluvatar_mrv-100").split(",") +CARD_NAME = {"Iluvatar_bi-150": "天垓150", "Iluvatar_bi-100": "天垓100", "Iluvatar_mrv-100": "智铠100"} +TASK_TYPE = os.getenv("TASK_TYPE", "visual-multi-modal") # 天数 VL 已实测能验证成功 +FRAMEWORK = os.getenv("FRAMEWORK", "vllm") +POLL_SEC = int(os.getenv("POLL_SEC", "90")) +MAX_PER_CYCLE = int(os.getenv("MAX_PER_CYCLE", "3")) # 每轮最多提交数, 不制造异常负载 +SEARCH_KW = os.getenv("SEARCH_KW", "InternVL,MiniCPM-V,Qwen2-VL,Qwen2.5-VL,Llava") + +STATE = {"cycles": 0, "submitted": 0, "skipped": 0, "errors": 0, "last": [], "started": time.time()} +_stop = threading.Event() + + +def log(msg): + print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True) + + +def _http(method, url, headers=None, body=None, timeout=30): + data = body.encode() if isinstance(body, str) else body + req = urllib.request.Request(url, data=data, method=method, headers=headers or {}) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return r.status, r.read().decode() + except urllib.error.HTTPError as e: + return e.code, e.read().decode() + except Exception as e: + return -1, str(e) + + +def api_get(path): + st, txt = _http("GET", MAIN + path, {"Authorization": TOKEN}) + try: + return st, json.loads(txt) + except Exception: + return st, txt + + +# ---- 签名: 复刻前端 browser-submit (sha256 + AES-CBC) ---- +def _sha_hex(s): return hashlib.sha256(s.encode()).hexdigest() +def _sha_bytes(s): return hashlib.sha256(s.encode()).digest() + + +def _aes_cbc_b64(pt, key, iv): + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + from cryptography.hazmat.primitives import padding + padder = padding.PKCS7(128).padder() + data = padder.update(pt.encode()) + padder.finalize() + enc = Cipher(algorithms.AES(key), modes.CBC(iv)).encryptor() + return base64.b64encode(enc.update(data) + enc.finalize()).decode() + + +def _sign(model, cfg, t_ms): + a = _sha_hex("computility:model-verify:browser-submit" + "|v1|" + str(t_ms) + "|" + model) + ii = _sha_hex(cfg) + kf = _sha_bytes("key|" + a + "|" + ii)[:16] + ivf = _sha_bytes("iv|" + model + "|" + str(t_ms) + "|" + str(len(cfg)))[:16] + o = str(t_ms) + "\n" + model + "\n" + cfg + return _sha_hex("v1." + a + "." + _aes_cbc_b64(o, kf, ivf) + "." + ii) + + +def build_config(gpu): + st, txt = _http("POST", MAIN + f"/api/adapt/task/build-config?gpuType={gpu}&framework={FRAMEWORK}&taskType={TASK_TYPE}", + {"Authorization": TOKEN}) + try: + j = json.loads(txt) + return j.get("data") if isinstance(j, dict) else None + except Exception: + return None + + +def already_adapted(model, gpu): + st, j = api_get("/api/computility/models/metrics?modelId=" + urllib.parse.quote(model)) + if not isinstance(j, dict): + return None + names = [x.get("machineName") for x in (j.get("data") or [])] + return CARD_NAME.get(gpu) in names + + +def find_models(): + out = [] + for kw in SEARCH_KW.split(","): + st, txt = _http("GET", GITEA + f"/api/v1/repos/search?q={urllib.parse.quote(kw)}&sort=updated&order=desc&limit=20") + try: + for r in json.loads(txt).get("data", []): + if r.get("full_name"): + out.append(r["full_name"]) + except Exception: + pass + return out + + +def _post(model, gpu, cfg, signed): + t_ms = int(time.time() * 1000) + h = {"Authorization": TOKEN, "Content-Type": "application/json"} + if signed: + h.update({"X-Request-Time": str(t_ms), "X-Model-Address": model, "X-Request-Sign": _sign(model, cfg, t_ms)}) + body = json.dumps({"modelAddress": model, "taskType": TASK_TYPE, "targetGpu": gpu, + "framework": FRAMEWORK, "configParams": cfg}) + st, txt = _http("POST", MAIN + "/api/adapt/task/add", h, body) + try: + return st, json.loads(txt) + except Exception: + return st, {} + + +def submit(model, gpu): + cfg = build_config(gpu) + if not cfg: + return {"model": model.split("/")[-1], "gpu": gpu, "skip": "no-config"} + if DRY_RUN: + return {"model": model.split("/")[-1], "gpu": gpu, "dry_run": True, + "cfg_len": len(cfg), "sign_ok": bool(_sign(model, cfg, int(time.time() * 1000)))} + # 实测两种鉴权: 先 token-only(H1, 信任服务令牌), 失败再补签名(H2, 复刻浏览器) + st, j = _post(model, gpu, cfg, False); used = "token-only" + if not (isinstance(j, dict) and j.get("code") == 0): + st, j = _post(model, gpu, cfg, True); used = "signed" + data = j.get("data") if isinstance(j, dict) else None + return {"model": model.split("/")[-1], "gpu": gpu, "auth": used, "http": st, + "code": j.get("code"), "taskId": (data or {}).get("id") if isinstance(data, dict) else None, + "msg": j.get("message")} + + +def loop(): + log(f"loop start | dry_run={DRY_RUN} | token_present={bool(TOKEN)} | cards={TARGET_CARDS} | type={TASK_TYPE}") + st, j = api_get("/api/credit/info") + log(f"auth probe /credit/info -> http={st} code={(j.get('code') if isinstance(j, dict) else '?')}") + seen = set() + while not _stop.is_set(): + STATE["cycles"] += 1 + try: + n = 0 + for m in find_models(): + if _stop.is_set() or n >= MAX_PER_CYCLE: + break + for gpu in TARGET_CARDS: + key = m + "|" + gpu + if key in seen: + continue + seen.add(key) + if already_adapted(m, gpu) is True: + STATE["skipped"] += 1 + continue + res = submit(m, gpu) + STATE["last"] = ([res] + STATE["last"])[:20] + if res.get("code") == 0 or res.get("dry_run"): + STATE["submitted"] += 1; n += 1 + elif res.get("skip"): + STATE["skipped"] += 1 + else: + STATE["errors"] += 1 + log("submit " + json.dumps(res, ensure_ascii=False)) + time.sleep(2) + break # 一模型一轮只打一张卡, 摊开 + except Exception as e: + STATE["errors"] += 1 + log("loop error: " + str(e)) + _stop.wait(POLL_SEC) + log("loop stopped") + + +class H(BaseHTTPRequestHandler): + def _j(self, o, s=200): + b = json.dumps(o, ensure_ascii=False).encode() + self.send_response(s); self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b) + + def do_GET(self): + if self.path == "/health": + return self._j({"status": "ok"}) + if self.path == "/": + return self._j({"name": "huni-adapt-agent", "dry_run": DRY_RUN, "state": STATE}) + return self._j({"error": "not found"}, 404) + + def log_message(self, *a): + pass + + +def _sig(signum, _f): + _stop.set(); log(f"signal {signum}, shutting down") + + +def main(): + signal.signal(signal.SIGTERM, _sig) + signal.signal(signal.SIGINT, _sig) + threading.Thread(target=loop, daemon=True).start() + srv = ThreadingHTTPServer(("0.0.0.0", 8080), H); srv.timeout = 1 + log("listening on 0.0.0.0:8080") + while not _stop.is_set(): + srv.handle_request() + srv.server_close() + + +if __name__ == "__main__": + main()