Core 推理 Runtime

模型推理 spec、task/variant 契约、进程策略、autocast、编译、torchrun 命令、device 与计时器。

最后更新: 2026年7月22日

本页内容

Core runtime 在加载模型实现之前，先把模型可运行表面明确描述出来。ModelInferenceSpec 记录模型家族身份、variant、checkpoint、task profile、字段、artifact、流式支持和可接受调用参数。进程 helper 再应用共享执行策略，而不会把某个模型的行为塞进 Core。

构造或获取 inference spec

from worldfoundry.core import model_inference_spec

spec = model_inference_spec(
    model_family_id="my-world-model",
    display_name="My World Model",
    default_model_ref="org/my-world-model",
    workload_type="video",
    supported_call_params=("prompt", "seed", "num_frames"),
)

assert spec.model_family_id == "my-world-model"
assert spec.variant().variant_id == "default"
assert spec.task().task_id

对于已有内置家族，model_inference_spec 返回注册过的 spec；对于未知家族，它根据传入提示构造 generic spec。当“未注册”必须与 fallback 区分时，使用 get_model_inference_spec。

进程级推理策略

install_worldfoundry_inference_infra 配置注意力策略、float32 matmul precision、TF32 flag 和可选 SDPA 兼容 patch。它是幂等的进程状态，不是每个 request 一个的对象。worldfoundry_inference_context 会安装该状态，并在 torch.no_grad() 下执行。

autocast_context 只为 CUDA device 返回 autocast manager；CPU 或 Torch 不可用时返回 no-op context。compile_module_if_enabled 必须显式开启；编译关闭、不受支持或在非 strict 模式失败时，都会返回原 module。

from torch import nn
from worldfoundry.core import compile_module_if_enabled, torchrun_module_command

module = nn.Linear(4, 2)
assert compile_module_if_enabled(module, enabled=False) is module

command = torchrun_module_command(
    "my_package.worker",
    nproc_per_node=4,
    args=("--checkpoint", "/models/run-42"),
)
print(command)  # 这里只构造命令，不会启动进程

只有当前进程确实应当负责启动并捕获一个单机 job 时，才使用 run_torchrun_module。即使子进程非零退出，它也会返回 CompletedProcess，调用方必须检查 returncode 与日志。

完整参考

以下为该类别的生成签名。可用本页符号索引跳转；源码链接指向各惰性导出背后的具体实现。

26 个公开符号

`autocast_context`

func

def autocast_context(device: Any,dtype: Any | None = None,enabled: bool = True) -> Any

worldfoundry.core.autocast_contextfrom worldfoundry.core import autocast_context

源码

简介

autocast_context — Return a CUDA autocast context and a no-op context for non-CUDA devices. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Any。

参数

deviceAny
dtypeAny | None: 默认值: None
enabledbool: 默认值: True

返回值: Any

`compile_module_if_enabled`

func

def compile_module_if_enabled(module: Any,enabled: bool | None = None,label: str | None = None,backend: str | None = None,mode: str | None = None,fullgraph: bool | None = None,dynamic: bool | None = None,options: dict[str, Any] | None = None) -> Any

worldfoundry.core.compile_module_if_enabledfrom worldfoundry.core import compile_module_if_enabled

源码

简介

compile_module_if_enabled — Compile one module with `torch.compile only when explicitly enabled. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Any`。

参数

moduleAny
enabledbool | None: 默认值: None
labelstr | None: 默认值: None
backendstr | None: 默认值: None
modestr | None: 默认值: None
fullgraphbool | None: 默认值: None
dynamicbool | None: 默认值: None
optionsdict[str, Any] | None: 默认值: None

返回值: Any

`cuda_visible_devices_from_device`

func

def cuda_visible_devices_from_device(device: str | torch.device | None,inherited: str | None = None,map_inherited: bool = True,default_cuda: str = '0') -> str | None

worldfoundry.core.cuda_visible_devices_from_devicefrom worldfoundry.core import cuda_visible_devices_from_device

源码

简介

cuda_visible_devices_from_device — Convert a device string into a `CUDA_VISIBLE_DEVICES value. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：str | None`。

源码 docstring

Convert a device string into a `CUDA_VISIBLE_DEVICES` value.

`cuda:N is interpreted as a local index into an inherited CUDA_VISIBLE_DEVICES` list by default, which is the behavior expected by subprocess launchers nested under a scheduler or torchrun process.

参数

devicestr | torch.device | None
inheritedstr | None: 默认值: None
map_inheritedbool: 默认值: True
default_cudastr: 默认值: '0'

返回值: str | None

`CudaSyncTimer`

cls

class CudaSyncTimer(name: str | None = None,flag_env: str = 'SYNC_TIMER',log_fn: Callable[[str], None] | None = None)

worldfoundry.core.CudaSyncTimerfrom worldfoundry.core import CudaSyncTimer

源码

简介

CudaSyncTimer — Optional CUDA-synchronized timer usable as a context manager or decorator. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

参数

namestr | None: 默认值: None
flag_envstr: 默认值: 'SYNC_TIMER'
log_fnCallable[[str], None] | None: 默认值: None

方法

meth__call__(func)源码

简介

该类型上的公开 method。

参数

func

`generic_model_inference_spec`

func

def generic_model_inference_spec(model_family_id: str,display_name: str | None = None,default_model_ref: str = '',default_load_kwargs: Mapping[str, Any] | None = None,default_call_kwargs: Mapping[str, Any] | None = None,supports_stream: bool = False,workload_type: str = '',supported_call_params: Sequence[str] | None = None) -> ModelInferenceSpec

worldfoundry.core.generic_model_inference_specfrom worldfoundry.core import generic_model_inference_spec

源码

简介

generic_model_inference_spec — Build a conservative fallback spec for models not yet curated. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：ModelInferenceSpec。

参数

model_family_idstr
display_namestr | None: 默认值: None
default_model_refstr: 默认值: ''
default_load_kwargsMapping[str, Any] | None: 默认值: None
default_call_kwargsMapping[str, Any] | None: 默认值: None
supports_streambool: 默认值: False
workload_typestr: 默认值: ''
supported_call_paramsSequence[str] | None: 默认值: None

返回值: ModelInferenceSpec

`get_model_inference_spec`

func

def get_model_inference_spec(model_family_id: str) -> ModelInferenceSpec | None

worldfoundry.core.get_model_inference_specfrom worldfoundry.core import get_model_inference_spec

源码

简介

get_model_inference_spec — Return a curated inference spec for a model family when one exists. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：ModelInferenceSpec | None。

参数

model_family_idstr

返回值: ModelInferenceSpec | None

`inference_infra_state`

func

def inference_infra_state() -> WorldFoundryInferenceInfraState

worldfoundry.core.inference_infra_statefrom worldfoundry.core import inference_infra_state

源码

简介

inference_infra_state — Return the process-global inference infra state. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：WorldFoundryInferenceInfraState。

返回值: WorldFoundryInferenceInfraState

`InferenceArtifactSpec`

cls

class InferenceArtifactSpec(artifact_id: str,kind: str,required: bool = False,preview: bool = False,description: str = '')

worldfoundry.core.InferenceArtifactSpecfrom worldfoundry.core import InferenceArtifactSpec

源码

简介

InferenceArtifactSpec — Output artifact contract emitted by an inference task profile. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

artifact_idstr
kindstr
requiredbool: 默认值: False
previewbool: 默认值: False
descriptionstr: 默认值: ''

方法

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

`InferenceCheckpointRef`

cls

class InferenceCheckpointRef(role: str,uri: str,required: bool = True,status: str = 'unknown')

worldfoundry.core.InferenceCheckpointReffrom worldfoundry.core import InferenceCheckpointRef

源码

简介

InferenceCheckpointRef — Checkpoint reference used by a concrete inference variant. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

rolestr
uristr
requiredbool: 默认值: True
statusstr: 默认值: 'unknown'

方法

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

`InferenceFieldSpec`

cls

class InferenceFieldSpec(field_id: str,label: str,kind: str = 'string',target: str = 'call_kwargs',required: bool = False,default: Any = None,choices: tuple[str, ...] = (),description: str = '')

worldfoundry.core.InferenceFieldSpecfrom worldfoundry.core import InferenceFieldSpec

源码

简介

InferenceFieldSpec — User-facing input field contract for one inference task profile. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

field_idstr
labelstr
kindstr: 默认值: 'string'
targetstr: 默认值: 'call_kwargs'
requiredbool: 默认值: False
defaultAny: 默认值: None
choicestuple[str, ...]: 默认值: ()
descriptionstr: 默认值: ''

方法

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

`InferenceParams`

cls

class InferenceParams(max_batch_size: int, max_sequence_length: int)

worldfoundry.core.InferenceParamsfrom worldfoundry.core import InferenceParams

源码

简介

InferenceParams — State container used to cache key/value tensors during inference. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

参数

max_batch_sizeint
max_sequence_lengthint

方法

methswap_key_value_dict(batch_idx) -> None源码

简介

该类型上的公开 method。

参数

batch_idx

返回值: None

`InferenceTaskProfile`

cls

class InferenceTaskProfile(task_id: str,label: str,inputs: tuple[InferenceFieldSpec, ...],outputs: tuple[InferenceArtifactSpec, ...],description: str = '',default_call_kwargs: Mapping[str, Any] = <dict factory>,aliases: tuple[str, ...] = ())

worldfoundry.core.InferenceTaskProfilefrom worldfoundry.core import InferenceTaskProfile

源码

简介

InferenceTaskProfile — Runnable inference task profile for a model family or variant. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

task_idstr
labelstr
inputstuple[InferenceFieldSpec, ...]
outputstuple[InferenceArtifactSpec, ...]
descriptionstr: 默认值: ''
default_call_kwargsMapping[str, Any]: 默认值: <dict factory>
aliasestuple[str, ...]: 默认值: ()

方法

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

`InferenceVariantSpec`

cls

class InferenceVariantSpec(variant_id: str,label: str,checkpoints: tuple[InferenceCheckpointRef, ...] = (),status: str = 'unknown',load_kwargs: Mapping[str, Any] = <dict factory>,call_kwargs: Mapping[str, Any] = <dict factory>,aliases: tuple[str, ...] = (),notes: tuple[str, ...] = ())

worldfoundry.core.InferenceVariantSpecfrom worldfoundry.core import InferenceVariantSpec

源码

简介

InferenceVariantSpec — Concrete checkpoint/runtime variant under a model family. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

variant_idstr
labelstr
checkpointstuple[InferenceCheckpointRef, ...]: 默认值: ()
statusstr: 默认值: 'unknown'
load_kwargsMapping[str, Any]: 默认值: <dict factory>
call_kwargsMapping[str, Any]: 默认值: <dict factory>
aliasestuple[str, ...]: 默认值: ()
notestuple[str, ...]: 默认值: ()

方法

propprimary_checkpoint_uri -> str源码

简介

该类型上的公开 property。

参数

self

返回值: str

methcheckpoint_map() -> dict[str, str]源码

简介

该类型上的公开 method。

返回值: dict[str, str]

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

`install_worldfoundry_inference_infra`

func

def install_worldfoundry_inference_infra(attention_backend: str | None = None,matmul_precision: str | None = None,enable_tf32: bool | None = None,patch_sdpa: bool | None = None) -> WorldFoundryInferenceInfraState

worldfoundry.core.install_worldfoundry_inference_infrafrom worldfoundry.core import install_worldfoundry_inference_infra

源码

简介

install_worldfoundry_inference_infra — Install WorldFoundry core inference optimizations for this process. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：WorldFoundryInferenceInfraState。

源码 docstring

Install WorldFoundry core inference optimizations for this process.

Environment controls: - `WORLDFOUNDRY_USE_CORE_INFRA=0 disables installation. - WORLDFOUNDRY_ATTENTION_BACKEND=auto|flash|cudnn|efficient|math selects the SDPA backend policy. - WORLDFOUNDRY_MATMUL_PRECISION=highest|high|medium selects PyTorch float32 matmul precision. - WORLDFOUNDRY_ENABLE_TF32=0 disables TF32 backend flags. - WORLDFOUNDRY_PATCH_SDPA=0` avoids monkey-patching PyTorch SDPA calls.

参数

attention_backendstr | None: 默认值: None
matmul_precisionstr | None: 默认值: None
enable_tf32bool | None: 默认值: None
patch_sdpabool | None: 默认值: None

返回值: WorldFoundryInferenceInfraState

`list_model_inference_specs`

func

def list_model_inference_specs() -> tuple[ModelInferenceSpec, ]

worldfoundry.core.list_model_inference_specsfrom worldfoundry.core import list_model_inference_specs

源码

简介

list_model_inference_specs — Return curated model inference specs with explicit variants and task profiles. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：tuple[ModelInferenceSpec, ...]。

返回值: tuple[ModelInferenceSpec, ...]

`model_inference_spec`

func

def model_inference_spec(model_family_id: str,display_name: str | None = None,default_model_ref: str = '',default_load_kwargs: Mapping[str, Any] | None = None,default_call_kwargs: Mapping[str, Any] | None = None,supports_stream: bool = False,workload_type: str = '',supported_call_params: Sequence[str] | None = None) -> ModelInferenceSpec

worldfoundry.core.model_inference_specfrom worldfoundry.core import model_inference_spec

源码

简介

model_inference_spec — Return curated spec for a model family, or a generic fallback. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：ModelInferenceSpec。

参数

model_family_idstr
display_namestr | None: 默认值: None
default_model_refstr: 默认值: ''
default_load_kwargsMapping[str, Any] | None: 默认值: None
default_call_kwargsMapping[str, Any] | None: 默认值: None
supports_streambool: 默认值: False
workload_typestr: 默认值: ''
supported_call_paramsSequence[str] | None: 默认值: None

返回值: ModelInferenceSpec

`ModelInferenceSpec`

cls

class ModelInferenceSpec(model_family_id: str,display_name: str,variants: tuple[InferenceVariantSpec, ...],tasks: tuple[InferenceTaskProfile, ...],default_variant_id: str = 'default',default_task_id: str = 'default',aliases: tuple[str, ...] = (),notes: tuple[str, ...] = ())

worldfoundry.core.ModelInferenceSpecfrom worldfoundry.core import ModelInferenceSpec

源码

简介

ModelInferenceSpec — Inference contract shared by Studio, CLI, manifests, and eval. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

model_family_idstr
display_namestr
variantstuple[InferenceVariantSpec, ...]
taskstuple[InferenceTaskProfile, ...]
default_variant_idstr: 默认值: 'default'
default_task_idstr: 默认值: 'default'
aliasestuple[str, ...]: 默认值: ()
notestuple[str, ...]: 默认值: ()

方法

methvariant(variant_id: str | None = None) -> InferenceVariantSpec源码

简介

该类型上的公开 method。

参数

variant_idstr | None: 默认值: None

返回值: InferenceVariantSpec

methtask(task_id: str | None = None) -> InferenceTaskProfile源码

简介

该类型上的公开 method。

参数

task_idstr | None: 默认值: None

返回值: InferenceTaskProfile

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

`RealtimeSpec`

cls

class RealtimeSpec(fps: int = 16,first_chunk_frames: int = 9,steady_chunk_frames: int = 9,controls: tuple[str, ...] = DEFAULT_REALTIME_CONTROLS,transport: str = 'in-memory-rgb',stateful: bool = True)

worldfoundry.core.realtime.RealtimeSpecfrom worldfoundry.core.realtime import RealtimeSpec

源码

简介

RealtimeSpec — Model-owned playback and generation cadence for one resident session. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

fpsint: 默认值: 16
first_chunk_framesint: 默认值: 9
steady_chunk_framesint: 默认值: 9
controlstuple[str, ...]: 默认值: DEFAULT_REALTIME_CONTROLS
transportstr: 默认值: 'in-memory-rgb'
statefulbool: 默认值: True

方法

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

简介

该类型上的公开 method。

返回值: dict[str, Any]

cmethfrom_payload(value: Any, , fallback: 'RealtimeSpec | None' = None) -> 'RealtimeSpec'源码

简介

from_payload — Parse a model result without letting malformed metadata break play.

参数

valueAny
fallback'RealtimeSpec | None': 默认值: None

返回值: 'RealtimeSpec'

`resolve_inference_device`

func

def resolve_inference_device(device: str | torch.device | None = 'cuda',allow_cpu_fallback: bool = False) -> str

worldfoundry.core.resolve_inference_devicefrom worldfoundry.core import resolve_inference_device

源码

简介

resolve_inference_device — Resolve a concrete inference device without silently selecting the wrong GPU. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：str。

源码 docstring

Resolve a concrete inference device without silently selecting the wrong GPU.

Bare `cuda resolves to the process-local cuda:0. Explicit indices are preserved, which is important when a caller deliberately selects (for example) cuda:4` under an eight-GPU workspace.

参数

devicestr | torch.device | None: 默认值: 'cuda'
allow_cpu_fallbackbool: 默认值: False

返回值: str

`resolve_inference_dtype`

func

def resolve_inference_dtype(device: str | torch.device,dtype: str | torch.dtype | None = 'auto',strict: bool = True) -> torch.dtype

worldfoundry.core.resolve_inference_dtypefrom worldfoundry.core import resolve_inference_dtype

源码

简介

resolve_inference_dtype — Resolve an inference dtype using the selected accelerator's capability. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：torch.dtype。

源码 docstring

Resolve an inference dtype using the selected accelerator's capability.

`auto` selects bf16 on Ampere/Hopper/Blackwell-or-newer CUDA devices, fp16 on older CUDA devices, and fp32 on CPU. Explicit unsupported bf16 is rejected in strict mode instead of producing a later kernel failure.

参数

devicestr | torch.device
dtypestr | torch.dtype | None: 默认值: 'auto'
strictbool: 默认值: True

返回值: torch.dtype

`run_torchrun_module`

func

def run_torchrun_module(module: str,nproc_per_node: int,args: Sequence[str] = (),env: Mapping[str, str] | None = None,python_executable: str = sys.executable) -> subprocess.CompletedProcess[str]

worldfoundry.core.run_torchrun_modulefrom worldfoundry.core import run_torchrun_module

源码

简介

run_torchrun_module — Run a single-node Python module under torchrun and capture its logs. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：subprocess.CompletedProcess[str]。

参数

modulestr
nproc_per_nodeint
argsSequence[str]: 默认值: ()
envMapping[str, str] | None: 默认值: None
python_executablestr: 默认值: sys.executable

返回值: subprocess.CompletedProcess[str]

`torchrun_module_command`

func

def torchrun_module_command(module: str,nproc_per_node: int,args: Sequence[str] = (),python_executable: str = sys.executable) -> list[str]

worldfoundry.core.torchrun_module_commandfrom worldfoundry.core import torchrun_module_command

源码

简介

torchrun_module_command — Build a single-node torchrun command for a Python module. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：list[str]。

参数

modulestr
nproc_per_nodeint
argsSequence[str]: 默认值: ()
python_executablestr: 默认值: sys.executable

返回值: list[str]

`utc_now_iso`

func

def utc_now_iso() -> str

worldfoundry.core.utc_now_isofrom worldfoundry.core import utc_now_iso

源码

简介

utc_now_iso — Return the current UTC timestamp in ISO-8601 form. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：str。

返回值: str

`worldfoundry_inference_context`

func

def worldfoundry_inference_context() -> Iterator[None]

worldfoundry.core.worldfoundry_inference_contextfrom worldfoundry.core import worldfoundry_inference_context

源码

简介

worldfoundry_inference_context — Run model inference under the shared WorldFoundry core runtime policy. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Iterator[None]。

返回值: Iterator[None]

`WorldFoundryInferenceInfraState`

cls

class WorldFoundryInferenceInfraState(installed: bool = False,sdpa_patched: bool = False,attention_backend: str = 'auto',matmul_precision: str = 'high',tf32_enabled: bool = True)

worldfoundry.core.WorldFoundryInferenceInfraStatefrom worldfoundry.core import WorldFoundryInferenceInfraState

源码

简介

WorldFoundryInferenceInfraState — Observable process-wide inference acceleration state. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

installedbool: Whether core inference hooks were installed.默认值: False
sdpa_patchedbool: Whether the compatibility SDPA patch is active.默认值: False
attention_backendstr: Normalized attention backend policy.默认值: 'auto'
matmul_precisionstr: Current float32 matmul precision setting.默认值: 'high'
tf32_enabledbool: Whether CUDA TF32 matmul/cudnn execution is enabled.默认值: True

`wrap_runner_for_worldfoundry_core`

func

def wrap_runner_for_worldfoundry_core(runner: Any) -> Any

worldfoundry.core.wrap_runner_for_worldfoundry_corefrom worldfoundry.core import wrap_runner_for_worldfoundry_core

源码

简介

wrap_runner_for_worldfoundry_core — Wrap a runner instance so `generate always uses core inference infra. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Any`。

参数

runnerAny

返回值: Any