feat: auto-run pipeline on startup

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

main.py

@@ -1,74 +1,598 @@
 """
-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())}")
+
+    # 启动后自动执行一次提交流程
+    def _auto_run():
+        time.sleep(3)  # 等 HTTP 服务就绪
+        log("自动触发提交流程...")
+        try:
+            state['running'] = True
+            state['last_run'] = datetime.now().isoformat()
+            count = run_pipeline(submit_limit=30)
+            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=_auto_run, daemon=True).start()
 
     while not shutdown_requested:
         server.handle_request()
 
     server.server_close()
+    log("智能体已关闭")
 
 
-if __name__ == "__main__":
+if __name__ == '__main__':
     main()

requirements.txt

@@ -1 +1,2 @@
-# No external dependencies needed
+requests>=2.28.0
+pyyaml>=6.0