YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 更新后端平台模块、数据库、Compute引擎及多项配置文档

Browse Source 
- 更新 backend 平台 API、platform_store、session 数据库模块
- 新增 backend SQL 初始化脚本
- 更新 compute 引擎适配器及 README
- 更新 Docker 部署配置（app/compute）
- 更新前端入口、环境类型声明及 README
- 新增 docs/menu-functional-requirements.md 菜单功能需求文档
- 更新多项项目文档

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-07-21 10:55:44 +08:00
parent bccd3bf448
commit a72b8f1e4b
24 changed files with 571 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

710
backend/app/db/platform_store.py
View File

@@ -1,8 +1,10 @@
from __future__ import annotations
from __future__ import annotations
import json
import hashlib
import hmac
import math
import sqlite3
import secrets
import time
import uuid
from contextlib import contextmanager
@@ -10,6 +12,8 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator
import psycopg
from app.core.config import get_settings
@@ -53,175 +57,133 @@ def new_id(prefix: str) -> str:
return f"{prefix}_{uuid.uuid4().hex[:12]}"
class PlatformStore:
"""Small SQLite-backed store for the first runnable platform version.
PASSWORD_HASH_ITERATIONS = 390_000
The production model is PostgreSQL. This store mirrors the API-facing subset
needed by the first system iteration so developers can run the app without
provisioning enterprise infrastructure first.
def hash_password(password: str) -> str:
salt = secrets.token_hex(16)
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), PASSWORD_HASH_ITERATIONS)
return f"pbkdf2_sha256${PASSWORD_HASH_ITERATIONS}${salt}${digest.hex()}"
def verify_password(password: str, stored: str) -> tuple[bool, bool]:
if not stored.startswith("pbkdf2_sha256$"):
return hmac.compare_digest(password, stored), True
try:
_, iterations, salt, expected = stored.split("$", 3)
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), int(iterations)).hex()
return hmac.compare_digest(digest, expected), False
except ValueError:
return False, False
def _psycopg_url(database_url: str) -> str:
return database_url.replace("postgresql+psycopg://", "postgresql://")
def _pg_sql(sql: str) -> str:
return sql.replace("?", "%s")
class PgRow(dict):
def __init__(self, columns: list[str], values: tuple[Any, ...]) -> None:
super().__init__(zip(columns, values))
self._values = values
def __getitem__(self, key: str | int) -> Any:
if isinstance(key, int):
return self._values[key]
return super().__getitem__(key)
class PgCursor:
def __init__(self, cursor: psycopg.Cursor[Any]) -> None:
self.cursor = cursor
def execute(self, sql: str, params: tuple[Any, ...] | list[Any] | None = None) -> "PgCursor":
self.cursor.execute(_pg_sql(sql), params)
return self
def fetchone(self) -> PgRow | None:
row = self.cursor.fetchone()
if row is None:
return None
return PgRow(self._columns(), tuple(row))
def fetchall(self) -> list[PgRow]:
columns = self._columns()
return [PgRow(columns, tuple(row)) for row in self.cursor.fetchall()]
def _columns(self) -> list[str]:
return [col.name for col in self.cursor.description or []]
class PgConnection:
def __init__(self, conn: psycopg.Connection[Any]) -> None:
self.conn = conn
def execute(self, sql: str, params: tuple[Any, ...] | list[Any] | None = None) -> PgCursor:
cursor = PgCursor(self.conn.cursor())
return cursor.execute(sql, params)
def executemany(self, sql: str, params_seq: list[tuple[Any, ...]] | list[list[Any]]) -> None:
with self.conn.cursor() as cursor:
cursor.executemany(_pg_sql(sql), params_seq)
def executescript(self, sql: str) -> None:
with self.conn.cursor() as cursor:
for statement in sql.split(";"):
statement = statement.strip()
if statement:
cursor.execute(statement)
def commit(self) -> None:
self.conn.commit()
def rollback(self) -> None:
self.conn.rollback()
def close(self) -> None:
self.conn.close()
class PlatformStore:
"""PostgreSQL-backed store for the first runnable platform version.
This store mirrors the API-facing subset needed by the first system
iteration while using the same PostgreSQL dependency as later production
development.
"""
def __init__(self, db_path: str | None = None) -> None:
def __init__(self, database_url: str | None = None) -> None:
settings = get_settings()
raw_path = db_path or settings.local_db_path
self.db_path = Path(raw_path)
if not self.db_path.is_absolute():
self.db_path = Path.cwd() / self.db_path
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self.database_url = _psycopg_url(database_url or settings.database_url)
self.ensure_schema()
self.ensure_seed_data()
@contextmanager
def connect(self) -> Iterator[sqlite3.Connection]:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
def connect(self) -> Iterator["PgConnection"]:
raw_conn = psycopg.connect(self.database_url)
conn = PgConnection(raw_conn)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def ensure_schema(self) -> None:
schema_path = Path(__file__).with_name("sql") / "001_platform_runtime.sql"
with self.connect() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
password TEXT NOT NULL,
display_name TEXT NOT NULL,
role TEXT NOT NULL,
status TEXT NOT NULL,
permissions TEXT NOT NULL,
create_time TEXT NOT NULL,
last_login TEXT,
protected INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS models (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
type TEXT NOT NULL,
purpose TEXT NOT NULL,
model_source TEXT NOT NULL,
description TEXT,
path TEXT,
api_url TEXT,
api_key TEXT,
online_model_name TEXT,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS trained_models (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
train_methods TEXT NOT NULL,
base_model_path TEXT,
create_time TEXT NOT NULL,
merged INTEGER NOT NULL DEFAULT 0,
merging INTEGER NOT NULL DEFAULT 0,
merged_path TEXT
);
CREATE TABLE IF NOT EXISTS datasets (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
type TEXT NOT NULL,
storage_type TEXT NOT NULL,
source TEXT NOT NULL,
task_id TEXT,
size TEXT,
count INTEGER NOT NULL DEFAULT 0,
description TEXT,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS dataset_files (
id TEXT PRIMARY KEY,
dataset_id TEXT NOT NULL,
name TEXT NOT NULL,
size TEXT,
content TEXT NOT NULL,
active_version_id TEXT NOT NULL,
versions TEXT NOT NULL,
create_time TEXT NOT NULL,
FOREIGN KEY(dataset_id) REFERENCES datasets(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS compute_nodes (
id TEXT PRIMARY KEY,
code TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
api_base_url TEXT NOT NULL,
file_gateway_url TEXT NOT NULL,
enabled INTEGER NOT NULL DEFAULT 1,
scheduler_status TEXT NOT NULL,
scheduler_weight INTEGER NOT NULL DEFAULT 100,
tags TEXT NOT NULL,
gpu_count INTEGER NOT NULL DEFAULT 0,
current_running_jobs INTEGER NOT NULL DEFAULT 0,
max_parallel_jobs INTEGER NOT NULL DEFAULT 2,
data_root TEXT NOT NULL,
model_root TEXT NOT NULL,
log_root TEXT NOT NULL,
last_health_check_at TEXT,
health_detail TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS gpus (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL,
gpu_index INTEGER NOT NULL,
uuid TEXT NOT NULL,
name TEXT NOT NULL,
memory_total_gb REAL NOT NULL,
power_limit_w REAL NOT NULL,
base_temperature INTEGER NOT NULL,
FOREIGN KEY(node_id) REFERENCES compute_nodes(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS fine_tune_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
payload TEXT NOT NULL,
status TEXT NOT NULL,
progress INTEGER NOT NULL DEFAULT 0,
process_id INTEGER,
create_time TEXT NOT NULL,
start_time TEXT,
completed_at TEXT,
compute_node_id TEXT,
gpus TEXT NOT NULL,
sync_job_id TEXT
);
CREATE TABLE IF NOT EXISTS resource_replicas (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL,
resource_type TEXT NOT NULL,
resource_id TEXT NOT NULL,
local_path TEXT NOT NULL,
status TEXT NOT NULL,
sync_status TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS resource_sync_jobs (
id TEXT PRIMARY KEY,
target_node_id TEXT NOT NULL,
resources TEXT NOT NULL,
status TEXT NOT NULL,
progress INTEGER NOT NULL DEFAULT 0,
create_time TEXT NOT NULL,
completed_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_fine_tune_status ON fine_tune_tasks(status);
CREATE INDEX IF NOT EXISTS idx_dataset_files_dataset ON dataset_files(dataset_id);
CREATE INDEX IF NOT EXISTS idx_gpus_node ON gpus(node_id);
CREATE INDEX IF NOT EXISTS idx_replicas_resource ON resource_replicas(resource_type, resource_id);
"""
)
conn.executescript(schema_path.read_text(encoding="utf-8"))
columns = conn.execute(
"SELECT column_name FROM information_schema.columns WHERE table_name='users'"
).fetchall()
column_names = {row["column_name"] for row in columns}
if "password" in column_names and "password_hash" not in column_names:
conn.execute("ALTER TABLE users RENAME COLUMN password TO password_hash")
def ensure_seed_data(self) -> None:
with self.connect() as conn:
@@ -245,336 +207,12 @@ class PlatformStore:
conn.executemany(
"""
INSERT INTO users
(id, username, password, display_name, role, status, permissions, create_time, protected)
(id, username, password_hash, display_name, role, status, permissions, create_time, protected)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[(u[0], u[1], u[2], u[3], u[4], u[5], json_dumps(u[6]), now, u[7]) for u in users],
[(u[0], u[1], hash_password(u[2]), u[3], u[4], u[5], json_dumps(u[6]), now, u[7]) for u in users],
)
models = [
(
"m_qwen25_7b",
"Qwen2.5-7B-Instruct",
"LLM",
"training",
"local",
"Sample base model for SFT and LoRA training.",
"/models/Qwen2.5-7B-Instruct",
),
(
"m_llama31_8b",
"Llama-3.1-8B-Instruct",
"LLM",
"training",
"local",
"Reserved base model path for LLaMA-Factory dry-run.",
"/models/Llama-3.1-8B-Instruct",
),
]
conn.executemany(
"""
INSERT INTO models
(id, name, type, purpose, model_source, description, path, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
[(*m, now) for m in models],
)
self._insert_dataset(
conn,
"ds_finance_sft",
"finance-sft-sample",
"train",
"Financial QA SFT samples for sample training.",
"finance-sft-sample.jsonl",
"\n".join(
[
'{"instruction":"Summarize revenue growth.","input":"Revenue grew from 10M to 13M.","output":"Revenue increased 30% year over year."}',
'{"instruction":"Classify risk.","input":"Customer has overdue payment for 90 days.","output":"High credit risk."}',
'{"instruction":"Draft an analyst note.","input":"Gross margin improved by 4 points.","output":"Margin expansion indicates stronger operating leverage."}',
]
),
now,
)
self._insert_dataset(
conn,
"ds_customer_service",
"customer-service-sample",
"train",
"Customer-service instruction tuning samples.",
"customer-service-sample.jsonl",
"\n".join(
[
'{"instruction":"Respond politely.","input":"My package is late.","output":"I am sorry for the delay. I can help check the latest shipment status."}',
'{"instruction":"Escalate request.","input":"I need a refund for a defective item.","output":"I will create a refund case and share the next steps."}',
]
),
now,
)
node_rows = [
(
"node_01",
"gpu-node-01",
"GPU Node 01",
"http://gpu-node-01:19100",
"http://gpu-node-01:19101",
1,
"online",
100,
["A800", "80GB", "llama_factory"],
4,
0,
2,
"/data/yg-ft",
"/models",
"/data/yg-ft/training-logs",
now,
{"mode": "simulator", "heartbeat": "ok"},
),
(
"node_02",
"gpu-node-02",
"GPU Node 02",
"http://gpu-node-02:19100",
"http://gpu-node-02:19101",
1,
"online",
60,
["4090", "24GB", "llama_factory"],
4,
0,
1,
"/data/yg-ft",
"/models",
"/data/yg-ft/training-logs",
now,
{"mode": "simulator", "heartbeat": "ok"},
),
]
conn.executemany(
"""
INSERT INTO compute_nodes
(id, code, name, api_base_url, file_gateway_url, enabled, scheduler_status,
scheduler_weight, tags, gpu_count, current_running_jobs, max_parallel_jobs,
data_root, model_root, log_root, last_health_check_at, health_detail)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
(
r[0],
r[1],
r[2],
r[3],
r[4],
r[5],
r[6],
r[7],
json_dumps(r[8]),
r[9],
r[10],
r[11],
r[12],
r[13],
r[14],
r[15],
json_dumps(r[16]),
)
for r in node_rows
],
)
for node_id, model_name, mem in [
("node_01", "NVIDIA A800-SXM4-80GB", 80),
("node_02", "NVIDIA GeForce RTX 4090", 24),
]:
for idx in range(4):
conn.execute(
"""
INSERT INTO gpus
(id, node_id, gpu_index, uuid, name, memory_total_gb, power_limit_w, base_temperature)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
f"{node_id}_gpu_{idx}",
node_id,
idx,
f"GPU-{node_id.upper()}-{idx}",
model_name,
mem,
300 if mem >= 80 else 450,
35 + idx,
),
)
self._insert_task(
conn,
task_id="ft_pending_sample",
name="pending-sft-sample",
status="pending",
progress=0,
create_time=now,
start_time=None,
completed_at=None,
node_id=None,
gpus=[],
process_id=None,
)
started = datetime.now(timezone.utc).timestamp() - 22
self._insert_task(
conn,
task_id="ft_running_sample",
name="running-sft-sample",
status="running",
progress=45,
create_time=now,
start_time=datetime.fromtimestamp(started, timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
completed_at=None,
node_id="node_01",
gpus=[0, 1],
process_id=42001,
)
self._insert_task(
conn,
task_id="ft_completed_sample",
name="completed-sft-sample",
status="completed",
progress=100,
create_time=now,
start_time=now,
completed_at=now,
node_id="node_01",
gpus=[2],
process_id=42002,
)
conn.execute(
"""
INSERT INTO trained_models
(id, name, train_methods, base_model_path, create_time, merged, merging, merged_path)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"tm_completed_sample",
"completed-sft-sample-lora",
json_dumps([{"name": "lora"}]),
"/models/Qwen2.5-7B-Instruct",
now,
0,
0,
"/data/yg-ft/outputs/completed-sft-sample/adapter",
),
)
def _insert_dataset(
self,
conn: sqlite3.Connection,
dataset_id: str,
name: str,
dataset_type: str,
description: str,
file_name: str,
content: str,
now: str,
) -> None:
lines = [line for line in content.splitlines() if line.strip()]
size = f"{max(1, len(content.encode('utf-8')) // 1024)} KB"
file_id = f"{dataset_id}_file_1"
version_id = f"{file_id}_v1"
conn.execute(
"""
INSERT INTO datasets
(id, name, type, storage_type, source, size, count, description, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(dataset_id, name, dataset_type, "local", "upload", size, len(lines), description, now),
)
conn.execute(
"""
INSERT INTO dataset_files
(id, dataset_id, name, size, content, active_version_id, versions, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
file_id,
dataset_id,
file_name,
size,
content,
version_id,
json_dumps([{"id": version_id, "version": 1, "create_time": now, "description": "initial seed"}]),
now,
),
)
def _insert_task(
self,
conn: sqlite3.Connection,
task_id: str,
name: str,
status: str,
progress: int,
create_time: str,
start_time: str | None,
completed_at: str | None,
node_id: str | None,
gpus: list[int],
process_id: int | None,
) -> None:
payload = {
"id": task_id,
"name": name,
"description": "Seeded sample fine-tune task.",
"status": status,
"train_type": "SFT",
"train_method": "lora",
"template": "qwen",
"base_model": "m_qwen25_7b",
"train_dataset_id": "ds_finance_sft",
"auto_merge": False,
"output_model_name": f"{name}-lora",
"gpus": gpus,
"batch_size": 2,
"learning_rate": 0.0002,
"n_epochs": 3,
"save_steps": 50,
"lr_scheduler_type": "cosine",
"max_length": 2048,
"warmup_ratio": 0.03,
"weight_decay": 0.01,
"lora_alpha": 16,
"lora_dropout": 0.05,
"lora_rank": 8,
"quantization_bit": 4,
"export_quantized": False,
"quant_method": "bnb",
"quant_bits": 4,
"quant_group_size": 128,
"export_format": "safetensors",
"progress": progress,
"process_id": process_id,
"train_duration": self._duration(start_time, completed_at),
"create_time": create_time,
}
conn.execute(
"""
INSERT INTO fine_tune_tasks
(id, name, payload, status, progress, process_id, create_time, start_time, completed_at, compute_node_id, gpus)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
task_id,
name,
json_dumps(payload),
status,
progress,
process_id,
create_time,
start_time,
completed_at,
node_id,
json_dumps(gpus),
),
)
def _duration(self, start_time: str | None, end_time: str | None = None) -> str:
start = parse_time(start_time)
if not start:
@@ -590,6 +228,9 @@ class PlatformStore:
return f"{sec}s"
def refresh_runtime_state(self) -> None:
if get_settings().compute_mode != "simulator":
return
with self.connect() as conn:
rows = conn.execute(
"SELECT * FROM fine_tune_tasks WHERE status IN ('syncing','queued','running')"
@@ -644,7 +285,7 @@ class PlatformStore:
(status, progress, completed_at, row["id"]),
)
def _ensure_trained_model(self, conn: sqlite3.Connection, task: dict[str, Any]) -> None:
def _ensure_trained_model(self, conn: PgConnection, task: dict[str, Any]) -> None:
name = task.get("output_model_name") or f"{task['name']}-lora"
exists = conn.execute("SELECT id FROM trained_models WHERE name=?", (name,)).fetchone()
if exists:
@@ -676,10 +317,19 @@ class PlatformStore:
def login(self, username: str, password: str) -> dict[str, Any] | None:
with self.connect() as conn:
row = conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
if not row or row["password"] != password or row["status"] != "active":
if not row or row["status"] != "active":
return None
matched, legacy_plaintext = verify_password(password, row["password_hash"])
if not matched:
return None
last_login = utcnow()
conn.execute("UPDATE users SET last_login=? WHERE id=?", (last_login, row["id"]))
if legacy_plaintext:
conn.execute(
"UPDATE users SET password_hash=?, last_login=? WHERE id=?",
(hash_password(password), last_login, row["id"]),
)
else:
conn.execute("UPDATE users SET last_login=? WHERE id=?", (last_login, row["id"]))
data = self._user(row)
data["last_login"] = last_login
return data
@@ -691,13 +341,13 @@ class PlatformStore:
conn.execute(
"""
INSERT INTO users
(id, username, password, display_name, role, status, permissions, create_time, protected)
(id, username, password_hash, display_name, role, status, permissions, create_time, protected)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
""",
(
user_id,
payload["username"],
payload.get("password", "platform123"),
hash_password(payload.get("password", "platform123")),
payload.get("display_name") or payload["username"],
payload.get("role", "viewer"),
payload.get("status", "active"),
@@ -732,7 +382,7 @@ class PlatformStore:
raise ValueError("protected user cannot be deleted")
conn.execute("DELETE FROM users WHERE id=?", (user_id,))
def _user(self, row: sqlite3.Row) -> dict[str, Any]:
def _user(self, row: PgRow) -> dict[str, Any]:
return {
"id": row["id"],
"username": row["username"],
@@ -843,7 +493,7 @@ class PlatformStore:
raise KeyError(dataset_id)
return self._dataset(conn, row)
def _dataset(self, conn: sqlite3.Connection, row: sqlite3.Row) -> dict[str, Any]:
def _dataset(self, conn: PgConnection, row: PgRow) -> dict[str, Any]:
files = conn.execute(
"SELECT id, name, size, active_version_id, create_time FROM dataset_files WHERE dataset_id=? ORDER BY create_time",
(row["id"],),
@@ -884,9 +534,6 @@ class PlatformStore:
utcnow(),
),
)
if not payload.get("files"):
content = '{"instruction":"Sample instruction","input":"Sample input","output":"Sample output"}'
self.add_dataset_file(conn, dataset_id, "sample.jsonl", content)
return self._dataset(conn, conn.execute("SELECT * FROM datasets WHERE id=?", (dataset_id,)).fetchone())
def update_dataset(self, dataset_id: str, payload: dict[str, Any]) -> dict[str, Any]:
@@ -918,7 +565,7 @@ class PlatformStore:
conn.execute("DELETE FROM dataset_files WHERE dataset_id=?", (dataset_id,))
conn.execute("DELETE FROM datasets WHERE id=?", (dataset_id,))
def add_dataset_file(self, conn: sqlite3.Connection, dataset_id: str, name: str, content: str) -> dict[str, Any]:
def add_dataset_file(self, conn: PgConnection, dataset_id: str, name: str, content: str) -> dict[str, Any]:
now = utcnow()
file_id = new_id("file")
version_id = f"{file_id}_v1"
@@ -947,7 +594,7 @@ class PlatformStore:
)
return {"id": file_id, "name": name, "size": size}
def dataset_file(self, file_id: str) -> sqlite3.Row:
def dataset_file(self, file_id: str) -> PgRow:
with self.connect() as conn:
row = conn.execute("SELECT * FROM dataset_files WHERE id=?", (file_id,)).fetchone()
if not row:
@@ -1008,7 +655,7 @@ class PlatformStore:
raise KeyError(task_id)
return self._task(row)
def _task(self, row: sqlite3.Row) -> dict[str, Any]:
def _task(self, row: PgRow) -> dict[str, Any]:
payload = json_loads(row["payload"], {})
payload.update(
{
@@ -1028,6 +675,12 @@ class PlatformStore:
def create_task(self, payload: dict[str, Any]) -> dict[str, Any]:
task_id = str(payload.get("task_id") or payload.get("id") or new_id("ft"))
name = payload.get("name") or f"fine-tune-{task_id[-6:]}"
base_model = payload.get("base_model") or payload.get("base_model_id")
train_dataset_id = payload.get("train_dataset_id")
if not base_model:
raise ValueError("base_model or base_model_id is required")
if not train_dataset_id:
raise ValueError("train_dataset_id is required")
now = utcnow()
task = {
"id": task_id,
@@ -1037,8 +690,8 @@ class PlatformStore:
"train_type": payload.get("train_type", "SFT"),
"train_method": payload.get("train_method", "lora"),
"template": payload.get("template", "qwen"),
"base_model": payload.get("base_model") or payload.get("base_model_id") or "m_qwen25_7b",
"train_dataset_id": payload.get("train_dataset_id") or "ds_finance_sft",
"base_model": base_model,
"train_dataset_id": train_dataset_id,
"auto_merge": bool(payload.get("auto_merge", False)),
"output_model_name": payload.get("output_model_name") or f"{name}-lora",
"gpus": payload.get("gpus") or [],
@@ -1173,7 +826,7 @@ class PlatformStore:
"pending": "waiting for start",
"syncing": "syncing model and dataset to compute node",
"queued": "waiting for GPU slot",
"running": "training with LLaMA-Factory simulator",
"running": "training with LLaMA-Factory",
"completed": "training completed",
"failed": "training stopped",
}
@@ -1183,7 +836,7 @@ class PlatformStore:
"status": status,
"progress": progress,
"step": labels.get(status, status),
"speed": "42.5 samples/s" if status == "running" else "--",
"speed": task.get("train_speed") or "--",
"eta": eta,
}
@@ -1234,6 +887,40 @@ class PlatformStore:
)
return next(n for n in self.compute_nodes() if n["id"] == node_id)
def create_compute_node(self, payload: dict[str, Any]) -> dict[str, Any]:
node_id = payload.get("id") or new_id("node")
now = utcnow()
tags = payload.get("tags") or []
with self.connect() as conn:
conn.execute(
"""
INSERT INTO compute_nodes
(id, code, name, api_base_url, file_gateway_url, enabled, scheduler_status,
scheduler_weight, tags, gpu_count, current_running_jobs, max_parallel_jobs,
data_root, model_root, log_root, last_health_check_at, health_detail)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)
""",
(
node_id,
payload["code"],
payload.get("name") or payload["code"],
payload["api_base_url"],
payload.get("file_gateway_url") or payload["api_base_url"],
1 if payload.get("enabled", True) else 0,
payload.get("scheduler_status", "offline"),
int(payload.get("scheduler_weight", 100)),
json_dumps(tags),
int(payload.get("gpu_count", 0)),
int(payload.get("max_parallel_jobs", 1)),
payload.get("data_root", "/data/yg-ft"),
payload.get("model_root", "/models"),
payload.get("log_root", "/data/yg-ft/training-logs"),
now,
json_dumps(payload.get("health_detail") or {"status": "registered"}),
),
)
return next(node for node in self.compute_nodes() if node["id"] == node_id)
def gpus(self) -> list[dict[str, Any]]:
self.refresh_runtime_state()
with self.connect() as conn:
@@ -1367,30 +1054,7 @@ class PlatformStore:
"SELECT * FROM resource_replicas WHERE node_id=? ORDER BY create_time DESC",
(node_id,),
).fetchall()
if rows:
return [dict(row) for row in rows]
return [
{
"id": f"rep_{node_id}_model_qwen",
"node_id": node_id,
"resource_type": "model",
"resource_id": "m_qwen25_7b",
"local_path": "/models/Qwen2.5-7B-Instruct",
"status": "available",
"sync_status": "completed",
"create_time": utcnow(),
},
{
"id": f"rep_{node_id}_dataset_finance",
"node_id": node_id,
"resource_type": "dataset",
"resource_id": "ds_finance_sft",
"local_path": "/data/yg-ft/datasets/finance-sft-sample.jsonl",
"status": "available",
"sync_status": "completed",
"create_time": utcnow(),
},
]
return [dict(row) for row in rows]
def sync_job(self, sync_id: str) -> dict[str, Any]:
self.refresh_runtime_state()
@@ -1423,7 +1087,7 @@ class PlatformStore:
progress = int(task.get("progress", 0) or 0)
points = max(1, min(80, progress))
lines = [
f"[INFO] task={task['name']} engine=llama_factory mode=simulator status={task['status']}",
f"[INFO] task={task['name']} engine=llama_factory status={task['status']}",
f"[INFO] base_model={task.get('base_model')} dataset={task.get('train_dataset_id')} gpus={task.get('gpus', [])}",
"[INFO] command=llamafactory-cli train --stage sft --finetuning_type lora --do_train true",
]