huni-adapt-agent/main.py

"""ModelHub XC 适配智能体 (huni-adapt-agent)

平台已把"脚本提交"关闭(code 60014),程序化提交只能走"适配智能体"。
本智能体部署在平台内, 用注入的 EXTERNAL_SERVICE_TOKEN 作为 xcToken、经 **Xc-Token 头**
调开放平台 API 提交适配任务——这是被平台认可的"智能体提交"路径(不是被拦的脚本提交)。

关键教训: 主站开放 API 用 `Xc-Token` 头(不是 Authorization);提交不需要浏览器签名。
安全: token 只从环境读、绝不打印/落盘; 有 MAX_TOTAL 上限防止在公司平台上跑飞。
"""
import json, os, signal, time, threading
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", "")          # = xcToken, 走 Xc-Token 头
STRATEGY_ID = os.getenv("STRATEGY_ID", "")                  # 平台注入的自身策略 id, 提交必带 (docs/6 §3)
DRY_RUN  = os.getenv("DRY_RUN", "false").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")
FRAMEWORK   = os.getenv("FRAMEWORK", "vllm")
POLL_SEC    = int(os.getenv("POLL_SEC", "90"))
MAX_PER_CYCLE = int(os.getenv("MAX_PER_CYCLE", "3"))
MAX_TOTAL   = int(os.getenv("MAX_TOTAL", "20"))             # 累计提交上限, 到顶转空转(安全阀)
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 xc_get(path):
    st, txt = _http("GET", MAIN + path, {"Xc-Token": TOKEN})
    try:
        return st, json.loads(txt)
    except Exception:
        return st, txt


def build_config(gpu):
    st, txt = _http("POST", MAIN + f"/api/adapt/task/build-config?gpuType={gpu}&framework={FRAMEWORK}&taskType={TASK_TYPE}",
                    {"Xc-Token": 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 = xc_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 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)}
    body = json.dumps({"modelAddress": model, "taskType": TASK_TYPE, "targetGpu": gpu,
                       "framework": FRAMEWORK, "strategyId": STRATEGY_ID, "configParams": cfg})
    st, txt = _http("POST", MAIN + "/api/adapt/task/add",
                    {"Xc-Token": TOKEN, "Content-Type": "application/json"}, body)
    try:
        j = json.loads(txt)
    except Exception:
        j = {}
    data = j.get("data") if isinstance(j, dict) else None
    return {"model": model.split("/")[-1], "gpu": gpu, "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)} | strategy_id={STRATEGY_ID or 'MISSING'} | cards={TARGET_CARDS} | type={TASK_TYPE} | max_total={MAX_TOTAL}")
    st, j = xc_get("/api/adapt/task/page?current=1&pageSize=1")
    log(f"auth probe task/page(Xc-Token) -> http={st} code={(j.get('code') if isinstance(j, dict) else '?')}")
    seen = set()
    while not _stop.is_set():
        STATE["cycles"] += 1
        if STATE["submitted"] >= MAX_TOTAL:
            log(f"reached MAX_TOTAL={MAX_TOTAL}, idling (set higher to continue)")
            _stop.wait(POLL_SEC); continue
        try:
            n = 0
            for m in find_models():
                if _stop.is_set() or n >= MAX_PER_CYCLE or STATE["submitted"] >= MAX_TOTAL:
                    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()