Runtime 与资产

Environment 路径、本地资产 manifest、readiness 检查和受限子进程执行。

编辑此页 GitHub Markdown

最后更新: 2026年10月7日

本页内容

RequiredEnvReportWorldFoundryEnvLocalAssetexpand_worldfoundry_pathload_local_assetsrun_bounded_command

Runtime helper 用来显式表示机器专属状态：解析共享根目录、检查本地 staged asset、为 manifest 脱敏凭据，并把可能超时的子进程转成结构化结果。它们不会下载受保护数据，也不会静默替换缺失资产。

`RequiredEnvReport`

这个 report 分开记录已存在和缺失的环境变量名。它刻意不包含变量值，因此可以安全进入日志和 preflight response。

类

`RequiredEnvReport`

源码

class RequiredEnvReport(missing: tuple[str, ...],present: tuple[str, ...])

worldfoundry.runtime.envfrom worldfoundry.runtime.env import RequiredEnvReport

某工作负载所需环境变量/工具与当前是否具备的对照报告。

属性

missingtuple[str, ...]

Required environment variable names that were not set.

presenttuple[str, ...]

Required environment variable names that were set.

方法

propok -> bool源码

ok — Return whether all required variables were present.

参数

self

返回值

bool

methto_dict() -> dict[str, Any]源码

to_dict — Serialize the report for manifests and JSON preflight output.

返回值

dict[str, Any]

`WorldFoundryEnv`

WorldFoundryEnv 为 worldfoundry.runtime.env 中的路径解析与 preflight 函数提供面向对象入口。

from worldfoundry.runtime.env import WorldFoundryEnv

env = WorldFoundryEnv.from_os()
print(env.resolve_model_dir())
print(env.resolve_data_dir())
print(env.resolve_artifact_dir())

report = env.check_required(("HF_TOKEN", "WORLDFOUNDRY_DATA_DIR"))
if not report.ok:
    print("missing:", report.missing)

检查只报告是否存在。写 run record 时应使用 redact_for_manifest()，不要直接复制 os.environ。

类

`WorldFoundryEnv`

源码

class WorldFoundryEnv(values: EnvMapping)

worldfoundry.runtime.envfrom worldfoundry.runtime.env import WorldFoundryEnv

读取与校验 WorldFoundry 相关环境设置的辅助接口。

属性

valuesEnvMapping

Environment mapping used for path and metadata resolution.

方法

cmethfrom_os() -> 'WorldFoundryEnv'源码

from_os — Build an environment view from the current process environment.

返回值

'WorldFoundryEnv'

methresolve_cache_dir() -> Path源码

resolve_cache_dir — Resolve the root cache directory for downloads and transient assets.

返回值

Path

methresolve_data_dir() -> Path源码

resolve_data_dir — Resolve the shared local dataset directory.

返回值

Path

methresolve_model_dir() -> Path源码

resolve_model_dir — Resolve the local checkpoint and model asset directory.

返回值

Path

methresolve_ckpt_dir() -> Path源码

resolve_ckpt_dir — Resolve the local checkpoint directory.

返回值

Path

methresolve_artifact_dir() -> Path源码

resolve_artifact_dir — Resolve the generated artifact output directory.

返回值

Path

methresolve_hf_cache_dir() -> Path源码

resolve_hf_cache_dir — Resolve the Hugging Face dataset/model cache directory.

返回值

Path

methredact_for_manifest(keys: Sequence[str] | None = None) -> dict[str, Any]源码

redact_for_manifest — Return manifest-safe environment values with secrets reduced to presence.

参数

keysSequence[str] | None默认值 None

Optional variable names to include instead of the standard public set.

返回值

dict[str, Any]

methcheck_required(names: Sequence[str]) -> RequiredEnvReport源码

check_required — Check that required variables are set without exposing their values.

参数

namesSequence[str]

Environment variable names required by a runtime.

返回值

RequiredEnvReport

methcapture_runtime(include_torch: bool = True,include_nvidia_smi: bool = True) -> dict[str, Any]源码

capture_runtime — Capture local Python, CUDA, GPU, and Torch metadata for preflight logs.

参数

include_torchbool默认值 True

Whether to import Torch and record CUDA availability.

include_nvidia_smibool默认值 True

Whether to call `nvidia-smi` when available.

返回值

dict[str, Any]

`LocalAsset`

Local asset 把稳定 manifest identity、解析后路径与当前 ready 状态放在一起。Readiness 只表示路径存在，不证明 checkpoint 完整、dataset license 已接受或 runtime 能成功加载内容。

类

`LocalAsset`

源码

class LocalAsset(benchmark_id: str | None,asset_id: str,kind: str,path: Path | None,canonical_path: Path | None,status: str,ready: bool,metadata: Mapping[str, Any])

worldfoundry.runtime.assetsfrom worldfoundry.runtime.assets import LocalAsset

已落盘本地资产的描述（逻辑名 → 路径），供 runner 与测试使用。

属性

benchmark_idstr | None

Optional benchmark or integration id that owns the asset.

asset_idstr

Stable asset id inside the benchmark group.

kindstr

Asset kind, such as dataset, checkpoint, repo, manifest, or artifact.

pathPath | None

Resolved local path recorded by the manifest.

canonical_pathPath | None

Preferred path under the WorldFoundry root layout.

statusstr

Current path status computed at load time.

readybool

Whether the resolved path exists locally.

metadataMapping[str, Any]

Extra manifest fields preserved for consumers.

方法

cmeth

from_manifest_item(item: Mapping[str, Any],benchmark_id: str | None = None,env: EnvMapping | None = None) -> 'LocalAsset'

源码

from_manifest_item — Build a local asset view from a manifest item.

参数

itemMapping[str, Any]

Manifest asset mapping.

benchmark_idstr | None默认值 None

Optional parent benchmark id.

envEnvMapping | None默认值 None

Optional environment mapping used for path token expansion.

返回值

'LocalAsset'

methto_dict() -> dict[str, Any]源码

to_dict — Serialize the resolved asset status for logs and diagnostics.

返回值

dict[str, Any]

`expand_worldfoundry_path`

这个 helper 使用与 bootstrap/preflight 命令相同的根目录约定，展开受支持的 $WORLDFOUNDRY_* token。相对结果会锚定到仓库根目录。

函数

`expand_worldfoundry_path`

源码

def expand_worldfoundry_path(value: str | Path,env: EnvMapping | None = None) -> Path

worldfoundry.runtime.assetsfrom worldfoundry.runtime.assets import expand_worldfoundry_path

展开 WorldFoundry 路径 token 与 ~，得到具体文件系统路径。

参数

valuestr | Path

Path string or `Path with optional $VAR or ${VAR}` tokens.

envEnvMapping | None默认值 None

Optional environment mapping; defaults to `os.environ`.

返回值

Path

`load_local_assets`

Loader 读取选中的 local asset manifest，并返回解析后的 LocalAsset。因此，不调用 GPU runtime 也能写一个小型 readiness 检查脚本。

from worldfoundry.runtime.assets import load_local_assets

assets = load_local_assets()
missing = [asset for asset in assets if not asset.ready]

for asset in missing[:10]:
    print(asset.asset_id, asset.path or asset.canonical_path)

函数

`load_local_assets`

源码

def load_local_assets(path: str | Path | None = None,env: EnvMapping | None = None) -> tuple[LocalAsset, ...]

worldfoundry.runtime.assetsfrom worldfoundry.runtime.assets import load_local_assets

加载用于暂存 checkpoint、片段与夹具的 local-assets 映射。

参数

pathstr | Path | None默认值 None

Optional explicit manifest path.

envEnvMapping | None默认值 None

Optional environment mapping used for path expansion.

返回值

tuple[LocalAsset, ...]

`run_bounded_command`

Official evaluator 或 simulator 有时会卡在 native code 中。这个 helper 创建独立 process group、执行硬超时、捕获 stdout/stderr，并用 dictionary 返回 timeout 状态，而不是让信息丢失在未捕获 traceback 中。

from worldfoundry.runtime.jobs import run_bounded_command

completed = run_bounded_command(
    ["python", "-c", "print('preflight ok')"],
    timeout=30,
)
assert completed["returncode"] == 0

不要把用户不可信命令直接传给这个函数。它限制进程生命周期，但不是安全 sandbox。

函数

`run_bounded_command`

源码

def run_bounded_command(command: Sequence[str],cwd: str | Path | None = None,env: Mapping[str, str] | None = None,timeout: int,kill_timeout: int = 5) -> dict[str, Any]

worldfoundry.runtime.jobsfrom worldfoundry.runtime.jobs import run_bounded_command

简介

在时间/内存限制下运行子进程并捕获输出，供评测任务使用。

源码 docstring

Run a command with a hard timeout and always return captured output.

This helper is intended for official benchmark subprocesses. Some simulator or CUDA-backed scripts can ignore ordinary timeout handling while stuck in native code, so timeout failures are converted into structured results that callers can write into scorecards instead of surfacing a traceback.

参数

commandSequence[str]

cwdstr | Path | None默认值 None

envMapping[str, str] | None默认值 None

timeoutint

kill_timeoutint默认值 5

返回值

dict[str, Any]

根目录约定与 setup 命令见环境配置和本地资产指南。