Skip to content

feat(dashboard): preview launcher validation and command #449

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Dayuxiaoshui wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(dashboard): preview launcher validation and command #449

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 197 additions & 0 deletions areno/dashboard/dist/assets/index-BJq-UqM2.js
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion .../dashboard/dist/assets/index-CIfovKBG.css → .../dashboard/dist/assets/index-CFWsMZvS.css
View file
Open in desktop

Large diffs are not rendered by default.

197 changes: 0 additions & 197 deletions areno/dashboard/dist/assets/index-DF61sRZi.js
View file
Open in desktop

This file was deleted.

4 changes: 2 additions & 2 deletions areno/dashboard/dist/index.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>AReno Dashboard</title>
<script type="module" crossorigin src="./assets/index-DF61sRZi.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-CIfovKBG.css">
<script type="module" crossorigin src="./assets/index-BJq-UqM2.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-CFWsMZvS.css">
</head>
<body>
<div id="root"></div>
Expand Down
266 changes: 266 additions & 0 deletions areno/dashboard/launcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,266 @@
"""Validation and preview helpers for dashboard launcher submissions."""

from __future__ import annotations

import shlex
from pathlib import Path
from typing import Any, Callable

CommandBuilder = Callable[[dict[str, Any]], list[str]]


def preview_launcher(
kind: str,
config: dict[str, Any],
command_builder: CommandBuilder,
*,
acknowledge_warnings: bool = False,
) -> dict[str, Any]:
"""Resolve a launcher config without starting a process or loading weights."""

normalized = dict(config)
errors: list[dict[str, str]] = []
warnings: list[dict[str, str]] = []
if kind not in {"train", "serve"}:
errors.append({"field": "kind", "message": "must be train or serve"})
if kind == "train":
_validate_train(normalized, errors, warnings)
elif kind == "serve":
_validate_serve(normalized, errors, warnings)
try:
command = command_builder(normalized)
except (TypeError, ValueError) as exc:
errors.append(
{"field": "extra_args", "message": f"invalid shell arguments: {exc}"}
)
command = []
shell_command = shlex.join(command) if not errors else ""
return {
"ok": not errors and (acknowledge_warnings or not warnings),
"kind": kind,
"resolved_args": normalized,
"command": command,
"shell_command": shell_command,
"errors": errors,
"warnings": warnings,
"requires_acknowledgement": bool(warnings) and not acknowledge_warnings,
}


def _validate_train(
config: dict[str, Any], errors: list[dict[str, str]], warnings: list[dict[str, str]]
) -> None:
_required(config, "ckpt", errors)
_required(config, "dataset_path", errors)
_validate_local_reference(config.get("ckpt"), "ckpt", errors, warnings)
_validate_local_reference(
config.get("dataset_path"), "dataset_path", errors, warnings
)
_validate_positive_fields(
config,
(
"epochs",
"max_steps",
"world_size",
"tp_size",
"batch_size",
"mini_bs",
"score_micro_bs",
"gradient_accumulation_steps",
"max_prompt_tokens",
"max_new_tokens",
"save_interval",
"tune_max_samples",
),
errors,
)
_validate_positive_fields(config, ("n_samples", "max_running_prompts"), errors)
_validate_positive_fields(
config, ("lr", "temperature", "top_p", "agent_timeout_s"), errors
)
_validate_gpu_topology(config, errors)
_validate_parallelism(config, errors)
if config.get("model_hub", "modelscope") not in {"hf", "modelscope"}:
errors.append({"field": "model_hub", "message": "must be hf or modelscope"})
if config.get("algo", "sft").lower() == "sft" and not config.get(
"dataset_loader_fn"
):
errors.append({"field": "dataset_loader_fn", "message": "is required for sft"})
if config.get("algo", "sft").lower() in {"gspo", "grpo", "ppo"} and not (
config.get("reward_fn_path") or config.get("reward_ckpt")
):
errors.append(
{
"field": "reward_fn_path",
"message": "a reward function or reward checkpoint is required",
}
)
if config.get("smoke_infer") and config.get("smoke_train"):
errors.append(
{
"field": "smoke_train",
"message": "smoke_infer and smoke_train are mutually exclusive",
}
)
_validate_range(
config, "top_p", errors, minimum=0, maximum=1, minimum_inclusive=False
)
_validate_range(
config, "mem_frac", errors, minimum=0, maximum=1, minimum_inclusive=False
)
if config.get("greedy") and config.get("temperature") not in (None, "", 0, 1, 1.0):
warnings.append(
{"field": "temperature", "message": "greedy decoding ignores temperature"}
)
if not config.get("save_path") and not config.get("save_dir"):
warnings.append(
{"field": "save_path", "message": "no checkpoint output path is configured"}
)


def _validate_serve(
config: dict[str, Any], errors: list[dict[str, str]], warnings: list[dict[str, str]]
) -> None:
_required(config, "model_path", errors)
_validate_local_reference(config.get("model_path"), "model_path", errors, warnings)
_validate_positive_fields(
config,
("port", "world_size", "tp_size", "max_running_prompts", "default_max_tokens"),
errors,
)
_validate_gpu_topology(config, errors)
_validate_parallelism(config, errors)
_validate_range(config, "port", errors, minimum=1, maximum=65535)
if config.get("model_hub", "modelscope") not in {"hf", "modelscope"}:
errors.append({"field": "model_hub", "message": "must be hf or modelscope"})
if config.get("host") in {"", None}:
errors.append({"field": "host", "message": "must not be empty"})


def _required(config: dict[str, Any], field: str, errors: list[dict[str, str]]) -> None:
if config.get(field) in (None, ""):
errors.append({"field": field, "message": "is required"})


def _validate_positive_fields(
config: dict[str, Any], fields: tuple[str, ...], errors: list[dict[str, str]]
) -> None:
for field in fields:
value = config.get(field)
if value in (None, ""):
continue
try:
if float(value) <= 0:
raise ValueError
except (TypeError, ValueError):
errors.append({"field": field, "message": "must be positive"})


def _validate_gpu_topology(
config: dict[str, Any], errors: list[dict[str, str]]
) -> None:
devices = config.get("train_devices")
if devices in (None, ""):
return
if isinstance(devices, str):
try:
devices = [part.strip() for part in devices.split(",") if part.strip()]
devices = [int(item) for item in devices]
except ValueError:
errors.append(
{
"field": "train_devices",
"message": "must contain integer GPU indices",
}
)
return
if not isinstance(devices, list) or any(
not isinstance(item, int) or item < 0 for item in devices
):
errors.append(
{
"field": "train_devices",
"message": "must contain non-negative GPU indices",
}
)
return
if len(devices) != len(set(devices)):
errors.append(
{
"field": "train_devices",
"message": "must not contain duplicate GPU indices",
}
)
try:
world_size = int(config.get("world_size", 1))
except (TypeError, ValueError):
return
if len(devices) != world_size:
errors.append(
{"field": "world_size", "message": "must equal the number of train_devices"}
)


def _validate_parallelism(config: dict[str, Any], errors: list[dict[str, str]]) -> None:
raw_world_size = config.get("world_size", 1)
raw_tp_size = config.get("tp_size", 1)
try:
world_size = int(raw_world_size)
tp_size = int(raw_tp_size)
world_size_number = float(raw_world_size)
tp_size_number = float(raw_tp_size)
except (TypeError, ValueError):
return
if world_size != world_size_number:
errors.append({"field": "world_size", "message": "must be an integer"})
return
if tp_size != tp_size_number:
errors.append({"field": "tp_size", "message": "must be an integer"})
return
if world_size > 0 and tp_size > 0 and world_size % tp_size != 0:
errors.append({"field": "tp_size", "message": "must divide world_size"})


def _validate_range(
config: dict[str, Any],
field: str,
errors: list[dict[str, str]],
*,
minimum: float,
maximum: float,
minimum_inclusive: bool = True,
) -> None:
value = config.get(field)
if value in (None, ""):
return
try:
number = float(value)
except (TypeError, ValueError):
return
minimum_ok = number >= minimum if minimum_inclusive else number > minimum
if not minimum_ok or number > maximum:
left = "[" if minimum_inclusive else "("
errors.append(
{"field": field, "message": f"must be in {left}{minimum}, {maximum}]"}
)


def _validate_local_reference(
value: Any, field: str, errors: list[dict[str, str]], warnings: list[dict[str, str]]
) -> None:
if value in (None, ""):
return
text = str(value)
path = Path(text).expanduser()
if path.is_absolute() or text.startswith(".") or path.exists():
if not path.exists():
errors.append(
{"field": field, "message": f"local path does not exist: {text}"}
)
else:
warnings.append(
{
"field": field,
"message": "remote reference is not resolved during preview",
}
)
43 changes: 40 additions & 3 deletions areno/dashboard/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from areno.cli.diagnostics import collect_env, run_checks
from areno.dashboard.agent_context import agent_system_prompt
from areno.dashboard.agent_files import AgentFileBrowser
from areno.dashboard.launcher import preview_launcher

ROOT = Path(os.environ.get("ARENO_DASHBOARD_ROOT", Path.cwd())).resolve()
STATIC_DIR = Path(__file__).resolve().parent / "dist"
Expand Down Expand Up @@ -592,6 +593,12 @@ def _save_state(self) -> None:
STATE = DashboardState()


def _format_devices(value: Any) -> Any:
if isinstance(value, list):
return ",".join(str(device) for device in value)
return value


def build_train_command(config: dict[str, Any]) -> list[str]:
command = ["areno", "train"]
pairs = {
Expand All @@ -607,6 +614,7 @@ def build_train_command(config: dict[str, Any]) -> list[str]:
"--agent-fn": config.get("agent_fn"),
"--world-size": config.get("world_size"),
"--tp-size": config.get("tp_size"),
"--train-devices": _format_devices(config.get("train_devices")),
"--batch-size": config.get("batch_size"),
"--n-samples": config.get("n_samples"),
"--mini-bs": config.get("mini_bs"),
Expand Down Expand Up @@ -673,7 +681,7 @@ def build_train_command(config: dict[str, Any]) -> list[str]:
for key, value in flags.items():
if bool_like(value):
command.append(key)
command.extend(str(config.get("extra_args") or "").split())
command.extend(shlex.split(str(config.get("extra_args") or "")))
return command


Expand Down Expand Up @@ -721,10 +729,21 @@ def build_serve_command(config: dict[str, Any]) -> list[str]:
command.append("--eager-decode")
if bool_like(config.get("disable_thinking")):
command.append("--disable-thinking")
command.extend(str(config.get("extra_args") or "").split())
command.extend(shlex.split(str(config.get("extra_args") or "")))
return command


def launcher_preview(kind: str, payload: dict[str, Any]) -> dict[str, Any]:
config = payload.get("config") if isinstance(payload.get("config"), dict) else payload
builder = build_train_command if kind == "train" else build_serve_command
return preview_launcher(
kind,
config,
builder,
acknowledge_warnings=bool(payload.get("acknowledge_warnings")),
)


def detect_areno_job_kind(command: str) -> str | None:
"""Detect train/serve AReno commands started outside the dashboard server."""

Expand Down Expand Up @@ -1515,7 +1534,18 @@ def do_POST(self) -> None:
try:
path = self.route_path()
payload = self.read_json()
if path == "/api/jobs/train":
if path == "/api/launcher/preview":
kind = str(payload.get("kind") or "")
result = launcher_preview(kind, payload)
self.json(result)
elif path == "/api/jobs/train":
result = launcher_preview("train", payload)
if result["errors"]:
self.json({"preview": result}, HTTPStatus.BAD_REQUEST)
return
if result["requires_acknowledgement"]:
self.json({"preview": result}, HTTPStatus.CONFLICT)
return
job = Job(
kind="train",
name=f"train {payload.get('algo', 'sft')} {payload.get('ckpt', '')}",
Expand All @@ -1525,6 +1555,13 @@ def do_POST(self) -> None:
)
self.json({"job": STATE.start(job).to_json()})
elif path == "/api/jobs/serve":
result = launcher_preview("serve", payload)
if result["errors"]:
self.json({"preview": result}, HTTPStatus.BAD_REQUEST)
return
if result["requires_acknowledgement"]:
self.json({"preview": result}, HTTPStatus.CONFLICT)
return
model_path = payload.get("model_path") or payload.get("ckpt", "")
job = Job(
kind="serve",
Expand Down
Loading