From 2955a57e2ee604fd8cab8b97237c0d1780464483 Mon Sep 17 00:00:00 2001 From: z3st Date: Tue, 21 Jul 2026 19:04:43 +0800 Subject: [PATCH] feat: full cloud pipeline - search, filter, submit --- README.md | 39 +++- main.py | 573 ++++++++++++++++++++++++++++++++++++++++++++--- requirements.txt | 3 +- 3 files changed, 570 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index 3d5a15b..0687e29 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,33 @@ -# ModelHub 提交智能体 +# ModelHub 全云端提交智能体 -用于获取 `strategyId` 的最小化智能体。 +部署在平台容器中，从内网直接提交验证任务。 -## 功能 +## API -- `GET /health` → 健康检查 -- `GET /` → 显示 strategyId +| 端点 | 方法 | 说明 | +|------|------|------| +| `/health` | GET | 健康检查 | +| `/` | GET | 智能体信息 | +| `/status` | GET | 运行状态 | +| `/logs?lines=50` | GET | 查看日志 | +| `/run` | POST | 触发一次完整流程 | -## 平台要求 +## POST /run 请求体 -- ✅ 根目录 Dockerfile -- ✅ 暴露 8080 端口 -- ✅ `/health` 端点返回 200 -- ✅ 处理 SIGTERM 信号 -- ✅ 读取 `STRATEGY_ID` 环境变量 +```json +{ + "gpus": ["Kunlunxin_p-800", "hygon_k100-ai"], + "limit": 30 +} +``` + +- `gpus`: 目标 GPU 列表（可选，默认全部） +- `limit`: 每 GPU 最大提交数（可选，默认 30） + +## 流程 + +1. 搜索 ModelScope（Qwen 系列关键词） +2. 下载量 ≥ 50 筛选 +3. 架构检查（config.json） +4. 平台验证状态查重 +5. 提交到 ModelHub diff --git a/main.py b/main.py index 334eb9f..c21f2cb 100644 --- a/main.py +++ b/main.py @@ -1,74 +1,581 @@ """ -ModelHub 提交智能体（代理模式） -用途：获取 strategyId，本地脚本直接调用 ModelHub API +ModelHub 全云端提交智能体 +部署在平台容器中，直接从内网提交验证任务 + +功能： +- POST /run → 触发一次完整流程（搜索→筛选→提交） +- GET /status → 查看当前状态 +- GET /health → 健康检查 """ + import json import os import signal +import sqlite3 +import threading +import time +import traceback +from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlparse, parse_qs + +import requests +import yaml + +# ============================================================ +# 配置 +# ============================================================ HOST = "0.0.0.0" PORT = 8080 -shutdown_requested = False +STRATEGY_ID = os.getenv("STRATEGY_ID", "") + +# 四个账号的 Token +ACCOUNTS = { + 'MetaX_c-500': 'f8e60d1dac7f4472967e7ca40145747b', + 'Kunlunxin_p-800': 'f45f1aae2c094426be237c88b1085015', + 'Ascend_910-b4': 'f3c05879e7c34bbba92f399f12884183', + 'hygon_k100-ai': 'b88507029b884ad3b4bad8ba09e6546e', +} + +# GPU 引擎配置 +GPU_CONFIGS = { + 'MetaX_c-500': { + 'framework': 'vllm', + 'docker_image': 'modelhubxc-4pd.tencentcloudcr.com/enginex/enginex-metax/vllm:0.9.1', + }, + 'Kunlunxin_p-800': { + 'framework': 'vllm', + 'docker_image': 'modelhubxc-4pd.tencentcloudcr.com/enginex/sunjichen/xc-llm-kunlun:latest', + }, + 'Ascend_910-b4': { + 'framework': 'vllm', + 'docker_image': 'git.modelhub.org.cn:9443/enginex-ascend/vllm-ascend:v0.11.0rc0', + }, + 'hygon_k100-ai': { + 'framework': 'llama.cpp', + 'docker_image': 'modelhubxc-4pd.tencentcloudcr.com/enginex/enginex-hygon/hygon-llama.cpp:b7516', + }, +} + +# 架构白名单 +SUPPORTED_ARCH_KEYWORDS = ['Qwen', 'Qwen2', 'Qwen3'] +SUPPORTED_MODEL_TYPES = [ + 'qwen', 'qwen2', 'qwen2_vl', 'qwen2_5_vl', 'qwen2_audio', + 'qwen3', 'qwen3_vl', 'qwen3_5', 'qwen3_5_moe', +] +SUPPORTED_SPECIAL_ARCHS = ['Eagle3Speculator', 'LlamaForCausalLMEagle3'] + +MODELSCOPE_API = "https://modelscope.cn/api/v1" +MODELHUB_API = "https://modelhub.org.cn/api" + +# 搜索关键词 +SEARCH_KEYWORDS = ['qwen', 'Qwen2', 'Qwen3', 'Qwen3.5', 'Qwen1.5', 'Qwen-'] + +# ============================================================ +# 全局状态 +# ============================================================ + +state = { + 'running': False, + 'last_run': None, + 'last_result': None, + 'logs': [], +} +state_lock = threading.Lock() -def _config() -> dict: - strategy_id = os.getenv("STRATEGY_ID", "") - return { - "strategy_id": strategy_id, - "status": "running", +def log(msg: str): + ts = datetime.now().strftime('%H:%M:%S') + line = f"[{ts}] {msg}" + print(line, flush=True) + with state_lock: + state['logs'].append(line) + if len(state['logs']) > 500: + state['logs'] = state['logs'][-300:] + + +# ============================================================ +# 数据库（内存 SQLite） +# ============================================================ + +db_conn = None + + +def init_db(): + global db_conn + db_conn = sqlite3.connect(':memory:', check_same_thread=False) + db_conn.execute('''CREATE TABLE IF NOT EXISTS queue ( + model_id TEXT, gpu TEXT, url TEXT, downloads INTEGER, + params TEXT, category TEXT, score REAL, + PRIMARY KEY(model_id, gpu) + )''') + db_conn.execute('''CREATE TABLE IF NOT EXISTS submitted ( + model_id TEXT, gpu TEXT, task_id TEXT, submitted_at TEXT, + PRIMARY KEY(model_id, gpu) + )''') + db_conn.commit() + + +# ============================================================ +# ModelScope 搜索 +# ============================================================ + +def search_models(keyword: str, limit: int = 100) -> list: + """从 ModelScope 搜索模型""" + url = f"{MODELSCOPE_API}/models" + params = { + 'Query': keyword, + 'PageSize': limit, + 'SortBy': 'GITHUB_SCORE', } + try: + resp = requests.get(url, params=params, timeout=15) + data = resp.json() + models = data.get('Data', {}).get('Models', []) + return models + except Exception as e: + log(f" 搜索失败 [{keyword}]: {e}") + return [] -class Handler(BaseHTTPRequestHandler): - def do_GET(self) -> None: - if self.path == "/health": - self._send_json({"status": "ok"}) - return +def check_architecture(model_id: str) -> tuple: + """检查模型架构""" + try: + cfg_url = f"{MODELSCOPE_API}/models/{model_id}/repo?Revision=master&FilePath=config.json" + resp = requests.get(cfg_url, timeout=10) + if resp.status_code == 200: + cfg = resp.json() + archs = cfg.get('architectures', []) + mtype = cfg.get('model_type', '') + arch_str = str(archs) + for special in SUPPORTED_SPECIAL_ARCHS: + if special in arch_str: + return True, special + for kw in SUPPORTED_ARCH_KEYWORDS: + if kw in arch_str: + return True, kw + if mtype in SUPPORTED_MODEL_TYPES: + return True, mtype + return False, f"arch={archs} type={mtype}" + mid_upper = model_id.upper() + if 'QWEN3' in mid_upper or 'QWEN2' in mid_upper: + return True, "GGUF" + return False, "no config, not Qwen" + except Exception as e: + return True, f"check error: {e}" - if self.path == "/": - self._send_json({ - "name": "modelhub-submit-agent", - "config": _config(), + +def normalize_model_url(model_url: str) -> str: + """标准化 URL 格式""" + if '/models/' in model_url: + return model_url + if 'modelscope.cn/' in model_url: + parts = model_url.split('modelscope.cn/') + if len(parts) == 2: + return f"https://www.modelscope.cn/models/{parts[1]}" + return model_url + + +# ============================================================ +# 平台 API +# ============================================================ + +def check_platform_verify(model_id: str) -> dict: + """查询全平台验证状态""" + headers = {'Xc-Token': list(ACCOUNTS.values())[0], 'Accept': 'application/json'} + url = f"{MODELHUB_API}/computility/models/search-by-model-id" + try: + resp = requests.get(url, headers=headers, params={'modelId': model_id}, timeout=10) + data = resp.json() + if data.get('code') == 0: + return data.get('data', {}).get('verifyResult', {}) + except Exception: + pass + return {} + + +def check_my_submitted(model_id: str, gpu: str, token: str) -> bool: + """检查自己是否已提交""" + headers = {'Xc-Token': token, 'Accept': 'application/json'} + url = f"{MODELHUB_API}/adapt/task/page" + try: + resp = requests.get(url, headers=headers, params={ + 'current': 1, 'pageSize': 100, 'onlyMine': 'true', + 'gpuType': gpu, 'modelId': model_id, + }, timeout=10) + data = resp.json() + if data.get('code') == 0: + records = data.get('data', {}).get('records', []) + for r in records: + if r.get('modelId') == model_id and r.get('gpuType') == gpu: + status = r.get('status', '') + if status in ('waiting', 'running', 'success'): + return True + except Exception: + pass + return False + + +def check_queue_available(gpu: str, token: str) -> int: + """查询队列可用位置""" + headers = {'Xc-Token': token, 'Accept': 'application/json'} + url = f"{MODELHUB_API}/adapt/task/page" + try: + resp = requests.get(url, headers=headers, params={ + 'current': 1, 'pageSize': 1, 'onlyMine': 'true', + 'gpuType': gpu, 'status': 'waiting', + }, timeout=10) + data = resp.json() + if data.get('code') == 0: + waiting = int(data['data'].get('total', 0)) + return max(0, 100 - waiting) + except Exception: + pass + return -1 + + +def build_config_params(gpu: str) -> str: + """构建 YAML 配置""" + config = GPU_CONFIGS.get(gpu, {}) + framework = config.get('framework', 'vllm') + docker_image = config.get('docker_image', '') + + if framework == 'llama.cpp': + params = { + 'framework': 'llama.cpp', + 'docker_image': docker_image, + 'nv_docker_image': docker_image, + 'api': 'completion', + 'max_tokens': 1024, + 'temperature': 0.7, + 'repetition_penalty': 1.2, + 'top_p': 0.9, + 'lang': 'zh', + 'max_model_len': 4096, + 'modelFormat': 'GGUF', + 'sut_config': { + 'values': { + 'gpu_num': 1, + 'command': [ + '/bin/bash', '-ic', + 'llama-server --model /model --alias llm --threads 20 ' + '--n-gpu-layers 999 --prio 3 --min_p 0.01 ' + '--ctx-size 4096 --host 0.0.0.0 --port 8000 --jinja --flash-attn off' + ] + } + }, + 'ref_config': { + 'values': { + 'cpu_num': 2, 'gpu_num': 1, + 'command': [ + 'llama-server', '--model', '/model', '--alias', 'llm', + '--threads', '20', '--n-gpu-layers', '999', + '--ctx-size', '4096', '--host', '0.0.0.0', '--port', '8000', + ] + } + }, + 'model': 'llm', + } + else: + params = { + 'framework': 'vllm', + 'docker_image': docker_image, + 'nv_docker_image': docker_image, + 'api': 'completion', + 'max_tokens': 1024, + 'temperature': 0.7, + 'repetition_penalty': 1.2, + 'top_p': 0.9, + 'lang': 'zh', + 'max_model_len': 2048, + 'modelFormat': 'HuggingFace', + 'sut_config': { + 'values': { + 'gpu_num': 1, + 'command': [ + '/bin/bash', '-ic', + 'vllm serve /model --port 8000 --served-model-name llm ' + '--max-model-len 2048 --dtype auto --gpu-memory-utilization 0.95 ' + '-tp 1 --enforce-eager --trust-remote-code' + ] + } + }, + 'ref_config': { + 'values': { + 'cpu_num': 2, 'gpu_num': 1, + 'command': [ + 'vllm', 'serve', '/model', '--port', '8000', + '--served-model-name', 'llm', '--max-model-len', '2048', + '--dtype', 'auto', '--gpu-memory-utilization', '0.95', + '-tp', '1', '--enforce-eager', '--trust-remote-code', + ] + } + }, + 'model': 'llm', + } + return yaml.dump(params, default_flow_style=False, allow_unicode=True) + + +def submit_model(model_url: str, gpu: str, token: str) -> tuple: + """提交单个模型""" + headers = { + 'Xc-Token': token, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + } + url = f"{MODELHUB_API}/adapt/task/add" + config = GPU_CONFIGS.get(gpu, {}) + payload = { + 'modelAddress': normalize_model_url(model_url), + 'taskType': 'text-generation', + 'targetGpu': gpu, + 'framework': config.get('framework', 'vllm'), + 'strategyId': STRATEGY_ID, + 'configParams': build_config_params(gpu), + } + try: + resp = requests.post(url, headers=headers, json=payload, timeout=30) + data = resp.json() + if data.get('code') == 0: + task_id = data.get('data', {}).get('id') + return True, task_id, 'success' + return False, None, data.get('message', 'unknown error') + except Exception as e: + return False, None, str(e) + + +# ============================================================ +# 主流程 +# ============================================================ + +def run_pipeline(gpus: list = None, submit_limit: int = 30): + """完整流程：搜索→筛选→提交""" + if gpus is None: + gpus = list(GPU_CONFIGS.keys()) + + init_db() + log("=" * 50) + log("开始执行流程") + log(f"目标GPU: {', '.join(gpus)}") + log(f"提交限制: 每GPU {submit_limit} 个") + + # 1. 搜索 + log("\n--- 阶段1: 搜索 ModelScope ---") + seen = set() + all_models = [] + for kw in SEARCH_KEYWORDS: + models = search_models(kw, limit=100) + for m in models: + mid = m.get('ModelId', m.get('Name', '')) + if mid and mid not in seen: + seen.add(mid) + downloads = m.get('Downloads', 0) + if downloads >= 50: + all_models.append({ + 'model_id': mid, + 'url': f"https://modelscope.cn/{mid}", + 'downloads': downloads, + 'params': m.get('Parameters', ''), + 'category': 'quantized' if 'GGUF' in mid.upper() else 'standard', + }) + time.sleep(0.3) + log(f"搜索完成: {len(seen)} 个唯一模型, {len(all_models)} 个下载量>=50") + + # 2. 架构筛选 + log("\n--- 阶段2: 架构筛选 ---") + arch_passed = [] + arch_rejected = 0 + for m in all_models: + ok, reason = check_architecture(m['model_id']) + if ok: + arch_passed.append(m) + else: + arch_rejected += 1 + log(f" ✗ {m['model_id']}: {reason}") + time.sleep(0.15) + log(f"架构筛选: {len(arch_passed)} 通过, {arch_rejected} 拒绝") + + # 3. 按 GPU 筛选并提交 + log("\n--- 阶段3: 筛选并提交 ---") + total_submitted = 0 + for gpu in gpus: + token = ACCOUNTS.get(gpu) + if not token: + continue + + log(f"\n[{gpu}]") + + # 检查队列 + available = check_queue_available(gpu, token) + if available <= 0: + log(f" 队列满，跳过") + continue + log(f" 队列可用: {available}") + + # 筛选 + to_submit = [] + for m in arch_passed: + model_id = m['model_id'] + + # hygon 只接受 GGUF + if gpu == 'hygon_k100-ai' and 'GGUF' not in model_id.upper(): + continue + + # 检查全平台验证状态 + verify = check_platform_verify(model_id) + if gpu in verify: + continue # 已有记录，跳过 + + # 检查自己是否已提交 + if check_my_submitted(model_id, gpu, token): + continue + + to_submit.append(m) + if len(to_submit) >= min(submit_limit, available): + break + time.sleep(0.2) + + log(f" 待提交: {len(to_submit)}") + + # 提交 + submitted = 0 + for m in to_submit: + ok, task_id, msg = submit_model(m['url'], gpu, token) + if ok: + submitted += 1 + log(f" ✅ {m['model_id']}") + db_conn.execute( + 'INSERT OR REPLACE INTO submitted VALUES (?,?,?,?)', + (m['model_id'], gpu, str(task_id), datetime.now().isoformat()) + ) + else: + log(f" ❌ {m['model_id']}: {msg}") + time.sleep(0.5) + + log(f" 提交完成: {submitted}/{len(to_submit)}") + total_submitted += submitted + + db_conn.commit() + log(f"\n{'=' * 50}") + log(f"流程完成，共提交 {total_submitted} 个模型") + return total_submitted + + +# ============================================================ +# HTTP 服务 +# ============================================================ + +class AgentHandler(BaseHTTPRequestHandler): + def do_GET(self): + parsed = urlparse(self.path) + path = parsed.path + + if path == '/health': + self._json({'status': 'ok'}) + elif path == '/': + self._json({ + 'name': 'modelhub-submit-agent', + 'strategy_id': STRATEGY_ID, + 'status': 'running' if state['running'] else 'idle', + 'last_run': state['last_run'], + 'last_result': state['last_result'], }) - return + elif path == '/status': + self._json({ + 'running': state['running'], + 'last_run': state['last_run'], + 'last_result': state['last_result'], + 'queue_count': db_conn.execute('SELECT COUNT(*) FROM queue').fetchone()[0] if db_conn else 0, + 'submitted_count': db_conn.execute('SELECT COUNT(*) FROM submitted').fetchone()[0] if db_conn else 0, + }) + elif path == '/logs': + lines = int(parse_qs(parsed.query).get('lines', ['50'])[0]) + self._json({'logs': state['logs'][-lines:]}) + else: + self._json({'error': 'not found'}, 404) - self._send_json({"error": "not found"}, status=404) + def do_POST(self): + parsed = urlparse(self.path) + path = parsed.path - def log_message(self, fmt: str, *args: object) -> None: - print(f"{self.address_string()} - {fmt % args}", flush=True) + if path == '/run': + if state['running']: + self._json({'error': 'already running'}, 409) + return - def _send_json(self, body: dict, status: int = 200) -> None: - payload = json.dumps(body).encode() + # 读取请求体 + content_len = int(self.headers.get('Content-Length', 0)) + body = {} + if content_len > 0: + body = json.loads(self.rfile.read(content_len)) + + gpus = body.get('gpus', list(GPU_CONFIGS.keys())) + limit = body.get('limit', 30) + + self._json({'status': 'started', 'gpus': gpus, 'limit': limit}) + + # 后台运行 + def _run(): + try: + state['running'] = True + state['last_run'] = datetime.now().isoformat() + count = run_pipeline(gpus=gpus, submit_limit=limit) + state['last_result'] = {'submitted': count, 'success': True} + except Exception as e: + log(f"流程异常: {traceback.format_exc()}") + state['last_result'] = {'error': str(e), 'success': False} + finally: + state['running'] = False + + threading.Thread(target=_run, daemon=True).start() + else: + self._json({'error': 'not found'}, 404) + + def _json(self, body: dict, status: int = 200): + payload = json.dumps(body, ensure_ascii=False).encode() self.send_response(status) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(payload))) + self.send_header('Content-Type', 'application/json; charset=utf-8') + self.send_header('Content-Length', str(len(payload))) self.end_headers() self.wfile.write(payload) + def log_message(self, fmt, *args): + pass # 静默 HTTP 日志 -def _handle_signal(signum: int, _frame: object) -> None: + +# ============================================================ +# 启动 +# ============================================================ + +shutdown_requested = False + + +def _handle_signal(signum, _frame): global shutdown_requested shutdown_requested = True - print(f"received signal {signum}, shutting down", flush=True) + log(f"收到信号 {signum}，准备关闭") -def main() -> None: +def main(): signal.signal(signal.SIGTERM, _handle_signal) signal.signal(signal.SIGINT, _handle_signal) - server = ThreadingHTTPServer((HOST, PORT), Handler) + init_db() + + server = ThreadingHTTPServer((HOST, PORT), AgentHandler) server.timeout = 1 - strategy_id = os.getenv("STRATEGY_ID", "NOT_SET") - print(f"modelhub-submit-agent listening on {HOST}:{PORT}", flush=True) - print(f"STRATEGY_ID: {strategy_id}", flush=True) + log(f"智能体启动 | {HOST}:{PORT}") + log(f"STRATEGY_ID: {STRATEGY_ID}") + log(f"GPU: {', '.join(GPU_CONFIGS.keys())}") while not shutdown_requested: server.handle_request() server.server_close() + log("智能体已关闭") -if __name__ == "__main__": +if __name__ == '__main__': main() diff --git a/requirements.txt b/requirements.txt index c490c6e..fac9310 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ -# No external dependencies needed +requests>=2.28.0 +pyyaml>=6.0