Core 推理 Runtime

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

最后更新: 2026年10月7日

本页内容

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 与日志。

完整参考

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

40 个公开符号

函数

`adaptive_batched_inference`

源码

def adaptive_batched_inference(inputs: torch.Tensor,forward: Callable[[torch.Tensor], torch.Tensor],batch_size: int,device: torch.device | str | int | None = None,output_device: torch.device | str | int | None = None,minimum_batch_size: int = 1,batch_size_buckets: Sequence[int] | None = None,pad_to_batch_size: bool = False,scope: str | None = None,compile_forward: bool | None = None,compile_mode: str | None = None,persistent_forward: bool = False) -> torch.Tensor

worldfoundry.corefrom worldfoundry.core import adaptive_batched_inference

简介

adaptive_batched_inference — Run tensor inference in bounded, optionally shape-stable batches. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：torch.Tensor。

源码 docstring

Run tensor inference in bounded, optionally shape-stable batches.

Only the failed slice is retried. Successful outputs retain input order, and an allocation failure at batch size one is propagated unchanged. `pad_to_batch_size repeats the final real item and trims its outputs; this is intended for independent per-item encoders so Inductor and CUDA Graphs can reuse a small set of static leading dimensions. persistent_forward` declares that a closure is reused across multiple calls, allowing throughput mode to retain one compiled wrapper. It does not enable compilation by itself.

参数

inputstorch.Tensor

forwardCallable[[torch.Tensor], torch.Tensor]

batch_sizeint

devicetorch.device | str | int | None默认值 None

output_devicetorch.device | str | int | None默认值 None

minimum_batch_sizeint默认值 1

batch_size_bucketsSequence[int] | None默认值 None

pad_to_batch_sizebool默认值 False

scopestr | None默认值 None

compile_forwardbool | None默认值 None

compile_modestr | None默认值 None

persistent_forwardbool默认值 False

返回值

torch.Tensor

函数

`autocast_context`

源码

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

worldfoundry.corefrom 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。

源码 docstring

Return a CUDA autocast context and a no-op context for non-CUDA devices.

*device* may be a tensor, `torch.device, or device string. CPU / NPU / missing-torch paths return :func:contextlib.nullcontext`.

参数

deviceAny

dtypeAny | None默认值 None

enabledbool默认值 True

返回值

Any

函数

`bind_log_context`

源码

def bind_log_context(**fields: Any) -> Token[dict[str, Any]]

worldfoundry.corefrom worldfoundry.core import bind_log_context

简介

bind_log_context — Bind correlation fields for the current execution context. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Token[dict[str, Any]]。

源码 docstring

Bind correlation fields for the current execution context.

The returned token can be passed to :meth:contextvars.ContextVar.reset, or callers can prefer :func:log_context for scoped use.

参数

fieldsAny

返回值

Token[dict[str, Any]]

函数

`clear_inference_runtime_cache`

源码

def clear_inference_runtime_cache(cache: MutableMapping[Any, Any]) -> None

worldfoundry.corefrom worldfoundry.core import clear_inference_runtime_cache

简介

clear_inference_runtime_cache — Drop cached runtimes and release unreferenced accelerator allocations. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：None。

源码 docstring

Drop cached runtimes and release unreferenced accelerator allocations.

The torch import is intentionally lazy so model discovery remains cheap in processes that never execute a PyTorch policy. `empty_cache` only returns already-unreferenced blocks to the CUDA allocator; clearing the strong references and collecting cycles must happen first.

参数

cacheMutableMapping[Any, Any]

返回值

None

函数

`clear_log_context`

源码

def clear_log_context() -> None

worldfoundry.corefrom worldfoundry.core import clear_log_context

clear_log_context — Clear all correlation fields for the current execution context. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：None。

返回值

None

函数

`compile_module_if_enabled`

源码

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.corefrom worldfoundry.core import compile_module_if_enabled

简介

compile_module_if_enabled — Compile one module with persistent cache and wrapper reuse when enabled. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Any。

源码 docstring

Compile one module with persistent cache and wrapper reuse when enabled.

Honours `WORLDFOUNDRY_TORCH_COMPILE and the matching _BACKEND / _MODE / _FULLGRAPH / _DYNAMIC / _STRICT env vars when the explicit kwargs are omitted. Already compiled wrappers (_worldfoundry_core_compiled) are returned unchanged. Delegates to :func:worldfoundry.core.execution.compile_cache.compile_module_cached`.

参数

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

函数

`configure_logging`

源码

def configure_logging(level: str | int | None = None,log_file: str | os.PathLike | None = None,json: bool | None = None,rotation: str = '10 MB',retention: str = '1 week',colorize: bool | None = None,force: bool = False) -> None

worldfoundry.corefrom worldfoundry.core import configure_logging

简介

configure_logging — Configure WorldFoundry's unified logging pipeline. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：None。

源码 docstring

Configure WorldFoundry's unified logging pipeline.

Idempotent: a second call without `force=True` is a no-op, so it is safe to invoke from every framework entry point (CLI, model-runtime worker, …) without duplicating handlers/sinks.

参数

levelstr | int | None默认值 None

Threshold for *every* log source. Accepts a name (`"DEBUG"/"INFO"/…, case-insensitive; "WARN"/"TRACE" accepted as aliases) or an int level. None falls back to the WORLDFOUNDRY_LOG_LEVEL` env var, then INFO.

log_filestr | os.PathLike | None默认值 None

Optional rotating file sink. `None falls back to the WORLDFOUNDRY_LOG_FILE` env var, else no file sink.

jsonbool | None默认值 None

When true the file sink emits JSON Lines (the console stays colored text). `None falls back to WORLDFOUNDRY_LOG_JSON`.

rotationstr默认值 '10 MB'

Size threshold for file rotation (e.g. `"10 MB"`).

retentionstr默认值 '1 week'

How many rotated files to keep (e.g. `"1 week"` → 7 backups; only the count is honoured by the stdlib path).

colorizebool | None默认值 None

Override console colorization (loguru path only).

forcebool默认值 False

Reconfigure even if already configured.

返回值

None

函数

`cuda_visible_devices_from_device`

源码

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.corefrom 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`

源码

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

worldfoundry.corefrom worldfoundry.core import CudaSyncTimer

简介

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

源码 docstring

Optional CUDA-synchronized timer usable as a context manager or decorator.

Construction is cheap and has no torch dependency. `__enter__ reads *flag_env* (default SYNC_TIMER) and no-ops unless it is "1". When enabled it records elapsed_ms and, if *name* is set, logs "{name} takes X.XXXXs"`.

As a decorator, each call wraps the function in `with self:`.

参数

namestr | None默认值 None

flag_envstr默认值 'SYNC_TIMER'

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

方法

meth__call__(func)源码

__call__ — Wrap *func* so each call is timed with this same timer instance.

参数

func

函数

`get_current_torch_device`

源码

def get_current_torch_device() -> torch.device

worldfoundry.corefrom worldfoundry.core import get_current_torch_device

简介

get_current_torch_device — Return the accelerator selected for this process, or CPU. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：torch.device。

源码 docstring

Return the accelerator selected for this process, or CPU.

Distributed launchers select a process-local device with `set_device`. Using that current index avoids collapsing every local rank onto CUDA 0.

返回值

torch.device

函数

`get_log_context`

源码

def get_log_context() -> dict[str, Any]

worldfoundry.corefrom worldfoundry.core import get_log_context

get_log_context — Return a copy of the current correlation fields. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：dict[str, Any]。

返回值

dict[str, Any]

函数

`get_logger`

源码

def get_logger(name: str | None = None) -> WorldFoundryLoggerAdapter

worldfoundry.corefrom worldfoundry.core import get_logger

get_logger — Return the WorldFoundry structured-event facade for a stdlib logger. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：WorldFoundryLoggerAdapter。

参数

namestr | None默认值 None

返回值

WorldFoundryLoggerAdapter

函数

`inference_infra_state`

源码

def inference_infra_state() -> WorldFoundryInferenceInfraState

worldfoundry.corefrom worldfoundry.core import inference_infra_state

简介

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

源码 docstring

Return the process-global inference infra state (do not mutate casually).

The returned dataclass is the live singleton updated by :func:install_worldfoundry_inference_infra and :func:uninstall_worldfoundry_inference_infra.

返回值

WorldFoundryInferenceInfraState

类

`InferenceArtifactSpec`

源码

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

worldfoundry.corefrom 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]源码

to_dict — Serialize this artifact contract to a JSON-ready mapping.

返回值

dict[str, Any]

类

`InferenceCheckpointRef`

源码

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

worldfoundry.corefrom 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]源码

to_dict — Serialize this checkpoint reference (role, URI, required, status).

返回值

dict[str, Any]

类

`InferenceFieldSpec`

源码

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.corefrom 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]源码

to_dict — Serialize this field contract to a JSON-ready mapping.

返回值

dict[str, Any]

类

`InferenceTaskProfile`

源码

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.corefrom 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]源码

to_dict — Serialize this task profile, including nested field/artifact specs.

返回值

dict[str, Any]

类

`InferenceVariantSpec`

源码

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.corefrom 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源码

primary_checkpoint_uri — URI of the `primary / base / checkpoint` role, else the first.

参数

self

返回值

str

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

checkpoint_map — Map each checkpoint role onto its URI.

返回值

dict[str, str]

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

to_dict — Serialize this variant, including `model_ref` and checkpoint map.

返回值

dict[str, Any]

函数

`install_worldfoundry_inference_infra`

源码

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.corefrom 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. Attention dispatch backend names (for example flash_attention_2, sage_attention, xformers) are also accepted; they resolve the SDPA policy to auto and are honoured by worldfoundry.core.attention instead. - 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. - WORLDFOUNDRY_DETERMINISTIC=1 enables torch.use_deterministic_algorithms` and CuDNN deterministic mode (XC-23).

Use :func:uninstall_worldfoundry_inference_infra (or the :func:worldfoundry_inference_infra_disabled context manager) to restore the original SDPA function and pre-install torch backend flags.

参数

attention_backendstr | None默认值 None

matmul_precisionstr | None默认值 None

enable_tf32bool | None默认值 None

patch_sdpabool | None默认值 None

返回值

WorldFoundryInferenceInfraState

函数

`is_accelerator_out_of_memory`

源码

def is_accelerator_out_of_memory(exc: BaseException) -> bool

worldfoundry.corefrom worldfoundry.core import is_accelerator_out_of_memory

is_accelerator_out_of_memory — Return whether an exception reports an accelerator allocation failure. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：bool。

参数

excBaseException

返回值

bool

函数

`log_context`

源码

def log_context(**fields: Any) -> Iterator[None]

worldfoundry.corefrom worldfoundry.core import log_context

log_context — Temporarily bind correlation fields for a block of work. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：Iterator[None]。

参数

fieldsAny

返回值

Iterator[None]

函数

`log_context_environment`

源码

def log_context_environment(**fields: Any) -> dict[str, str]

worldfoundry.corefrom worldfoundry.core import log_context_environment

简介

log_context_environment — Serialize current correlation fields for a framework-owned child process. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：dict[str, str]。

源码 docstring

Serialize current correlation fields for a framework-owned child process.

This deliberately carries only structured context, never logger handlers or raw stdout/stderr. It is safe to merge into a child `env` mapping.

参数

fieldsAny

返回值

dict[str, str]

类

`ModelInferenceSpec`

源码

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.corefrom 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源码

variant — Look up a variant by id or alias; default is `default_variant_id`.

参数

variant_idstr | None默认值 None

返回值

InferenceVariantSpec

异常

ValueError: No variant matches the (normalized) identifier.

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

task — Look up a task profile by id or alias; default is `default_task_id`.

参数

task_idstr | None默认值 None

返回值

InferenceTaskProfile

异常

ValueError: No task matches the (normalized) identifier.

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

to_dict — Serialize the family contract, including nested variants and tasks.

返回值

dict[str, Any]

函数

`profile_torch_module`

源码

def profile_torch_module(model: Any,inputs: Any = (),model_id: str | None = None,input_spec: Mapping[str, Any] | None = None,event_log_path: str | Path | None = None,set_eval: bool = True,strict: bool = False) -> TorchProfileResult

worldfoundry.corefrom worldfoundry.core import profile_torch_module

profile_torch_module — Trace *model* with TorchProfile and log a static compute summary. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：TorchProfileResult。

参数

modelAny

A loaded PyTorch module. The caller retains ownership of the model and its device placement.

inputsAny默认值 ()

One tensor or tuple of positional tensors representative of the inference workload. `torchprofile` does not support keyword-argument tracing.

model_idstr | None默认值 None

Optional model identifier; promoted to a top-level field in WorldFoundry JSON logs.

input_specMapping[str, Any] | None默认值 None

Optional JSON-safe shape/dtype metadata for the event. Do not place actual tensor values or credentials in this mapping.

event_log_pathstr | Path | None默认值 None

Optional run-owned JSONL path that receives the same canonical structured event even when no logger sink is configured.

set_evalbool默认值 True

Temporarily call `model.eval()` for deterministic tracing and restore all module training flags afterwards.

strictbool默认值 False

Re-raise unavailable/profile failures after emitting the event.

返回值

TorchProfileResultA :class:TorchProfileResult. With `strict=False` (the default), missing optional dependencies and trace incompatibilities are reported as result statuses rather than breaking the underlying model run.

函数

`read_text_tail`

源码

def read_text_tail(path: str | Path,max_lines: int = 20,max_bytes: int = 64 * 1024) -> str

worldfoundry.corefrom worldfoundry.core import read_text_tail

简介

read_text_tail — Read a bounded UTF-8 tail from a potentially very large log file. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：str。

源码 docstring

Read a bounded UTF-8 tail from a potentially very large log file.

Seeks to `max(size - max_bytes, 0) and keeps the last *max_lines* decoded lines. A mid-file seek drops the first partial line so the result does not start with a torn UTF-8 sequence. Missing files yield ""`.

参数

pathstr | Path

max_linesint默认值 20

max_bytesint默认值 64 * 1024

返回值

str

类

`RealtimeSpec`

源码

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,input_fps: float | None = None,first_input_frames: int | None = None,steady_input_frames: int | None = None)

worldfoundry.core.execution.realtime.contractsfrom worldfoundry.core.execution.realtime.contracts 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

input_fpsfloat | None默认值 None

first_input_framesint | None默认值 None

steady_input_framesint | None默认值 None

方法

propresolved_input_fps -> float源码

resolved_input_fps — Control/input sampling FPS, defaulting to the playback clock.

参数

self

返回值

float

propresolved_first_input_frames -> int源码

resolved_first_input_frames — First control chunk size, defaulting to first output frames.

参数

self

返回值

int

propresolved_steady_input_frames -> int源码

resolved_steady_input_frames — Steady control chunk size, defaulting to steady output frames.

参数

self

返回值

int

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

to_payload — JSON-safe dict; omit unset optional input-clock fields so clients do not treat `null` as 0.

返回值

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_batch_size`

源码

def resolve_inference_batch_size(default_batch_size: int,device: torch.device | str | int | None = None,scope: str | None = None,minimum: int = 1,maximum: int | None = None,reference_free_memory_gib: float = 24.0) -> int

worldfoundry.corefrom worldfoundry.core import resolve_inference_batch_size

简介

resolve_inference_batch_size — Choose a capability-neutral inference batch size from free device memory. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：int。

源码 docstring

Choose a capability-neutral inference batch size from free device memory.

The default is treated as the batch size for roughly 24 GiB of free memory. Smaller devices start conservatively, while 80 GiB and newer high-memory accelerators can use larger batches. Scoped and global environment values take precedence over automatic sizing.

参数

default_batch_sizeint

devicetorch.device | str | int | None默认值 None

scopestr | None默认值 None

minimumint默认值 1

maximumint | None默认值 None

reference_free_memory_gibfloat默认值 24.0

返回值

int

函数

`resolve_inference_device`

源码

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

worldfoundry.corefrom 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 CUDA device currently selected for this process. 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`

源码

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

worldfoundry.corefrom 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_logged_subprocess`

源码

def run_logged_subprocess(command: str | Sequence[str],stdout_path: str | Path,stderr_path: str | Path,cwd: str | Path | None = None,env: Mapping[str, str] | None = None,timeout: float | None = None,shell: bool = False,append: bool = False,start_new_session: bool | None = None) -> subprocess.CompletedProcess[Any]

worldfoundry.corefrom worldfoundry.core import run_logged_subprocess

简介

run_logged_subprocess — Run a subprocess with bounded log memory and timeout-safe descendants. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：subprocess.CompletedProcess[Any]。

源码 docstring

Run a subprocess with bounded log memory and timeout-safe descendants.

The child receives the current WorldFoundry correlation context and gets a dedicated JSONL lifecycle artifact beside its raw stdout/stderr captures. This keeps framework-owned workers observable without requiring changes to vendored official benchmark code.

参数

commandstr | Sequence[str]

stdout_pathstr | Path

stderr_pathstr | Path

cwdstr | Path | None默认值 None

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

timeoutfloat | None默认值 None

shellbool默认值 False

appendbool默认值 False

start_new_sessionbool | None默认值 None

返回值

subprocess.CompletedProcess[Any]

函数

`run_torchrun_module`

源码

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.corefrom 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]。

源码 docstring

Run a single-node Python module under torchrun and capture its logs.

Uses :func:subprocess.run with `capture_output=True and check=False. Prefer :func:run_logged_subprocess` when the worker may run long enough that in-memory capture is unsafe.

参数

modulestr

nproc_per_nodeint

argsSequence[str]默认值 ()

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

python_executablestr默认值 sys.executable

返回值

subprocess.CompletedProcess[str]

函数

`terminate_process_group`

源码

def terminate_process_group(process: subprocess.Popen[Any],grace_seconds: float = 3.0) -> None

worldfoundry.corefrom worldfoundry.core import terminate_process_group

简介

terminate_process_group — Terminate a process and all descendants in its dedicated process group. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：None。

源码 docstring

Terminate a process and all descendants in its dedicated process group.

POSIX path sends `SIGTERM to the group (os.killpg), waits *grace_seconds*, then SIGKILL. Windows falls back to terminate / kill on the parent only. Already-exited groups are a no-op. The child must have been started with start_new_session=True (see :func:run_logged_subprocess`).

参数

processsubprocess.Popen[Any]

grace_secondsfloat默认值 3.0

返回值

None

函数

`terminate_process_tree`

源码

def terminate_process_tree(process: subprocess.Popen[Any],grace_seconds: float = 3.0) -> None

worldfoundry.corefrom worldfoundry.core import terminate_process_tree

简介

terminate_process_tree — Terminate one process and recursively terminate descendants without a process group. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：None。

源码 docstring

Terminate one process and recursively terminate descendants without a process group.

Used when the child was *not* started in a new session. With `psutil installed, children are terminate'd first, then the parent, then any survivors are kill`'d. Without psutil only the parent is signalled — grandchildren may leak.

参数

processsubprocess.Popen[Any]

grace_secondsfloat默认值 3.0

返回值

None

类

`TorchProfileResult`

源码

class TorchProfileResult(status: Literal['completed', 'unavailable', 'failed'],macs: int | None = None,flops_estimate: int | None = None,parameter_count: int | None = None,torchprofile_version: str | None = None,model_class: str | None = None,error: str | None = None)

worldfoundry.corefrom worldfoundry.core import TorchProfileResult

TorchProfileResult — A compact static-compute result from one optional TorchProfile trace. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。

属性

statusLiteral['completed', 'unavailable', 'failed']

macsint | None默认值 None

flops_estimateint | None默认值 None

parameter_countint | None默认值 None

torchprofile_versionstr | None默认值 None

model_classstr | None默认值 None

errorstr | None默认值 None

方法

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

to_dict — Return a JSON-compatible representation suitable for result payloads.

返回值

dict[str, Any]

函数

`torchrun_module_command`

源码

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

worldfoundry.corefrom worldfoundry.core import torchrun_module_command

简介

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

源码 docstring

Build a single-node torchrun command for a Python module.

Equivalent to `python -m torch.distributed.run --standalone --nproc_per_node N -m MODULE ...`. Does not launch the process.

参数

modulestr

nproc_per_nodeint

argsSequence[str]默认值 ()

python_executablestr默认值 sys.executable

返回值

list[str]

函数

`utc_now_iso`

源码

def utc_now_iso() -> str

worldfoundry.corefrom worldfoundry.core import utc_now_iso

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

返回值

str

函数

`worldfoundry_inference_context`

源码

def worldfoundry_inference_context() -> Iterator[None]

worldfoundry.corefrom worldfoundry.core import worldfoundry_inference_context

简介

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

源码 docstring

Run model inference under the shared WorldFoundry core runtime policy.

Installs process-wide infra if needed, then yields inside `torch.no_grad()` when torch is importable. A missing torch install still enters the context so CPU-only control planes do not crash.

返回值

Iterator[None]

类

`WorldFoundryInferenceInfraState`

源码

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

worldfoundry.corefrom worldfoundry.core import WorldFoundryInferenceInfraState

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

属性

installedbool默认值 False

Whether :func:install_worldfoundry_inference_infra ran.

sdpa_patchedbool默认值 False

Whether the compatibility SDPA monkey-patch is active.

attention_backendstr默认值 'auto'

Normalized attention backend policy (`auto, flash, math, …). External names such as flash_attention_2 are accepted by the installer and mapped to the SDPA policy auto`.

matmul_precisionstr默认值 'high'

Current float32 matmul precision (`highest / high / medium`).

tf32_enabledbool默认值 True

Whether CUDA TF32 matmul/cudnn execution is enabled.

函数

`wrap_runner_for_worldfoundry_core`

源码

def wrap_runner_for_worldfoundry_core(runner: Any) -> Any

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`。

源码 docstring

Wrap a runner instance so `generate` always uses core inference infra.

Installs infra, then replaces `runner.generate with a wrapper that enters :func:worldfoundry_inference_context. Idempotent: a second call returns the same instance. Runners without generate` are returned unchanged.

参数

runnerAny

返回值

Any

函数

`write_jsonl_event`

源码

def write_jsonl_event(path: str | os.PathLike[str],level: str,event: str,message: str,logger_name: str = 'worldfoundry',exception: BaseException | str | None = None,**fields: Any) -> dict[str, Any]

worldfoundry.corefrom worldfoundry.core import write_jsonl_event

简介

write_jsonl_event — Append one canonical lifecycle event to a run-owned JSONL artifact. 属于 Core 推理 Runtime（进程设置、编译、计时、realtime）。标注返回类型：dict[str, Any]。

源码 docstring

Append one canonical lifecycle event to a run-owned JSONL artifact.

Use this for process boundaries where a child may not import the central logger. Normal in-process application events should use :meth:WorldFoundryLoggerAdapter.event instead.

参数

pathstr | os.PathLike[str]

levelstr

eventstr

messagestr

logger_namestr默认值 'worldfoundry'

exceptionBaseException | str | None默认值 None

fieldsAny

返回值

dict[str, Any]