训练侧 SDK

GPUPlane 的训练侧 SDK 是可选的：python train.py 无论有没有平台都保持独立可跑。SDK 只负责把指标、checkpoint、事件以 best-effort 方式送向 server，失败时落入本地缓冲，永远不会把异常抛给训练进程。

# packages/sdk 已随仓库安装；生产环境只需 httpx
pip install /path/to/packages/sdk

平台 job 内自动 attach

由调度器派发到 agent 的 job，runner 会在启动时注入一组受控环境变量（runner.py:start_job 会先剥离所有以 GPUCTL_ 开头的现有变量，再注入）：

环境变量	含义
GPUCTL_RUN_ID	当前 Run 的 id，SDK 据此进入 attached 模式
GPUCTL_JOB_ID	当前 Job id
GPUCTL_SERVER	server 的 HTTP 地址
GPUCTL_TOKEN	单次 Run 的 sdk-scope ingest token
CUDA_VISIBLE_DEVICES	分配的 GPU 索引

如果作业没有自带 PYTHONPATH，runner 还会把 agent 环境里版本匹配的 SDK 源码路径注入 PYTHONPATH，这样 conda/venv 里的 python 也能直接 import gpuctl。因此在平台 job 里通常不需要调 run.init()，直接 run.log(...) 即可。

API 签名与语义

所有公开方法都保证：调用不会抛异常；返回 None 或空值表示本次最佳努力失败。

run.init(...)

run.init(
    project: str = "default",
    experiment: str = "",
    name: str = "",
    config: dict | None = None,
    server: str | None = None,
    token: str | None = None,
) -> str | None

初始化 SDK 的运行时上下文。它按以下顺序决定模式：

1. 若环境里有 GPUCTL_RUN_ID，进入 attached 模式，返回该 run_id。
2. 若显式或环境提供了 server + token，向 POST /api/v1/runs 自注册一个 source=sdk 的 run，进入 registered 模式，返回 server 分配的 run_id。
3. 否则进入 local 模式，run_id 为 local-<pid>-<ts>，所有数据写入本地 spool，返回 None。

多次调用幂等：一旦状态不是 new，直接返回已有 run_id。

run.log(...)

run.log(
    metrics: dict[str, float],
    step: int | None = None,
    epoch: float | None = None,
    ts: int | None = None,
) -> None

发送一批指标。键建议用 / 命名空间，如 train/loss、eval/accuracy。非数值值会被静默跳过；空字典直接返回。ts 默认为当前毫秒时间戳。

run.log_checkpoint(...)

run.log_checkpoint(
    path: str | os.PathLike,
    step: int | None = None,
    epoch: float | None = None,
) -> None

只登记、不搬运文件。SDK 会计算文件或目录大小（size_bytes），然后推给 server。server 端会结合该 run 对应 job 的 working_dir 把相对路径 absolutize 成绝对路径（checkpoints.py:absolutize）。

run.finish(...)

run.finish(status: str = "SUCCEEDED") -> None

对 registered（source=sdk）的 run，发送最终状态并 flush 2 秒；对 attached 的 job 内 run 是 no-op，因为 run 的终态由 agent 上报。再调用一次不会重复发送。

辅助方法

run.log_event(name: str, data: dict | None = None, step: int | None = None) -> None
run.flush(timeout_s: float = 2.0) -> None
run.summary() -> dict[str, Any]

summary() 是本地只读视图，包含 run_id、state、latest、dropped、errors、queued、spool 等字段，适合在训练脚本末尾打印确认。

平台外裸跑与离线降级

在平台外直接执行训练脚本时：

• 如果设置了 GPUCTL_SERVER 和 GPUCTL_TOKEN，SDK 会自注册 source=sdk 的 run。
• 如果 server 不可达或没有 token，自动降级为 local 模式，数据写入 ~/.gpuctl/spool/sdk-{run_id}.jsonl。
• 后续恢复在线（registered 或 attached）时，发送线程会先重放 spool 再发送新批次，保证断线期间的点不会丢在新流量之后。

提示

本地 spool 让 python train.py 在火车上、容器内或 server 维护期间也能正常跑；重新联网后数据会自动补齐。

框架 callbacks

gpuctl.callbacks 提供 Lightning 与 Hugging Face Trainer 的薄封装，导入该模块不会引入 heavy 依赖；框架缺失时实例化才抛 ImportError。

PyTorch Lightning

from gpuctl.callbacks import GpuctlLightningCallback
from lightning.pytorch import Trainer

trainer = Trainer(
    callbacks=[GpuctlLightningCallback(every_n_train_steps=50)],
)

every_n_train_steps 控制训练 batch 日志采样频率；on_validation_epoch_end 每次都会上报。建议在你的 self.log("train/loss", ...)、self.log("val/loss", ...) 里直接用 train/、val/ 前缀。

Hugging Face Trainer

from gpuctl.callbacks import GpuctlHFTrainerCallback
from transformers import Trainer

trainer = Trainer(
    model=model, args=args,
    train_dataset=ds,
    callbacks=[GpuctlHFTrainerCallback()],
)

callback 在每次 on_log 把 logs 字典里可转 float 的标量转发给 GPUPlane。

边界保证

SDK 的设计目标是把失败面完全隔离在训练进程之外：

• 永不抛异常：init/log/log_checkpoint/finish 全部 try/except 吞掉，只增加内部 _errors 计数。
• 有界队列不阻塞：内部队列上限 _QUEUE_MAX = 10_000，满时采用 drop-oldest 策略，并累计 dropped 计数。
• 独立守护线程：名为 gpuctl-sdk-sender 的 daemon thread 负责批量 POST；进程退出时通过 atexit 唤醒 flush，最多等待 2 秒。
• 唯一重依赖：httpx，避免与训练环境的 pydantic/torch 版本冲突。

完整接入示例

下面是一个典型 PyTorch 训练循环里该插入 SDK 的位置：

import os
import torch
from gpuctl import run

# 1. 训练开始：自动 attach（job 内）或自注册（裸跑）
run.init(
    project="mnist",
    experiment="lr-sweep",
    name="run-01",
    config={"lr": 1e-3, "batch": 64},
)

model = build_model().cuda()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

for epoch in range(10):
    for step, batch in enumerate(train_loader):
        x, y = batch
        optimizer.zero_grad()
        loss = model(x.cuda(), y.cuda())
        loss.backward()
        optimizer.step()

        # 2. 每 50 步上报训练指标
        if step % 50 == 0:
            run.log({"train/loss": loss.item()}, step=step, epoch=epoch)

    # 3. 一个 epoch 结束后做验证并上报 eval/loss
    val_loss = evaluate(model, val_loader)
    run.log({"eval/loss": val_loss}, step=step, epoch=epoch)

    # 4. 保存并登记 checkpoint（只登记路径，不搬运）
    ckpt_path = f"checkpoints/epoch_{epoch}.pt"
    torch.save(model.state_dict(), ckpt_path)
    run.log_checkpoint(ckpt_path, step=step, epoch=epoch)

# 5. 训练结束（裸跑 source=sdk 时把状态置为 SUCCEEDED）
run.finish()

注意

如果 checkpoint 路径是相对路径，server 会结合 job 的 working_dir 解析；在裸跑自注册场景下没有 working_dir，建议传绝对路径。

延伸阅读

• 指标与 Primary Metric
• 核心概念：Run / Checkpoint / Evaluation
• MCP 观测工具参考