Scaling
The scaling API connects every canonical family to its data contract,
literature, benchmark route, scale controls, extension points, numerical
defaults, and distributed runtime preparation. Task dimensions remain
explicit, and caller-provided constructor arguments always override the tier
defaults.
Main Objects
| Object |
Role |
SILVAFamilyGuide |
literature, benchmark, data, scaling, and extension contract |
full_scale_solver_config |
relative-residual forward and implicit-backward solver template |
build_scaled_silva |
canonical family factory with scale-sensitive numerical defaults |
SILVARuntimeConfig |
precision, batch, worker, checkpoint, distribution, and compilation choices |
prepare_silva_model |
device movement plus optional distributed and compiled wrapping |
The smoke, workstation, and full tiers alter numerical budgets and
runtime choices, not the SILVA state equation. Use the smoke tier to verify a
complete forward/loss/backward/checkpoint path before selecting a larger tier.
Operational Contract
This API surface connects coverage, reproduction, data, and scale configuration to the same SILVA experiment
contract used by the learning pages and notebooks. Its central relation is
\[
F_\theta(z;x)=0,\qquad \widehat F_{\theta,s}(z;x)=0\ \text{uses the same mathematical contract at scale tier }s
\]
| Part |
What must remain inspectable |
| State |
the selected family, constructor contract, runtime tier, and data-loader configuration. |
| Condition |
changing a runtime tier may change numerical budgets and resource use but must not silently change the family equation. |
| Diagnostic |
coverage record, verification level, solver settings, effective batch size, and source-scale metrics. |
| Replacement point |
compact defaults with family-specific modules, official data adapters, and an archived experiment configuration. |
| Scale axes |
solver iterations, tolerance, model width, batch size, precision, workers, process count, and checkpoint interval. |
The relevant method lineage is recorded in the SILVA construction [1] and the selected family's primary references. Those references
define the source mechanisms; this API exposes them through SILVA objects so a
reader can inspect, replace, solve, differentiate, and scale the construction.
Complete Compact Study
Run the complete repository program below from the project root. The page uses
the same file that is exercised by the test suite, so the displayed call is not
an isolated fragment.
"""Inspect one family from public API coverage through executable scale defaults."""
from __future__ import annotations
from silva_networks import (
SILVADataLoaderConfig,
implementation_cases,
runtime_for_tier,
silva_family_guide,
silva_reproduction_spec,
silva_scaling_defaults,
)
family = "fno_deq"
case = next(item for item in implementation_cases() if item.key == "recent_equilibrium_families")
guide = silva_family_guide(family)
reproduction = silva_reproduction_spec(family)
defaults = silva_scaling_defaults(family, tier="smoke")
runtime = runtime_for_tier("smoke")
loader = SILVADataLoaderConfig(batch_size=4, workers=0)
print("family", family)
print("public objects", len(case.public_objects))
print("verification", reproduction.verification_level)
print("benchmark tasks", len(guide.benchmark_tasks))
print("solver", defaults["config"].solver)
print("max iterations", defaults["config"].max_iter)
print("runtime", runtime.device, runtime.mixed_precision)
print("loader", loader.batch_size, loader.workers)
python examples/api_scale_workflow.py
Measured Compact Output
family fno_deq
public objects 12
verification compact-verified
benchmark tasks 2
solver anderson
max iterations 12
runtime auto none
loader 4 0
Interpret the Output
The family resolves through four independent registries: public coverage, source relation, scale guidance, and runtime/data configuration. The compact-verified label describes repository evidence; it does not convert the two listed benchmark tasks into claimed benchmark results.
For a controlled experiment, retain the compact call as a regression case and
change one scale axis at a time. Record the resolved constructor, data source
and split, preprocessing, seed, forward and backward solver settings, task
metric, normalized residual, iteration count, runtime, peak memory, and any
failed convergence case. A larger run becomes evidence only when its own
resolved configuration and outputs are archived; the compact output above is
evidence for the executable mechanism and its stated invariants.
Executable scale-up guidance shared by every SILVA model family.
SILVAFamilyGuide
dataclass
Research and execution contract for one canonical SILVA family.
Source code in src/silva_networks/scaling.py
| @dataclass(frozen=True)
class SILVAFamilyGuide:
"""Research and execution contract for one canonical SILVA family."""
family: str
role: str
data_contract: str
paper_refs: tuple[int, ...]
reference_repositories: tuple[str, ...]
benchmark_tasks: tuple[str, ...]
scale_controls: tuple[str, ...]
extension_points: tuple[str, ...]
|
SILVARuntimeConfig
dataclass
Runtime choices that do not alter a SILVA model's mathematics.
Source code in src/silva_networks/scaling.py
| @dataclass(frozen=True)
class SILVARuntimeConfig:
"""Runtime choices that do not alter a SILVA model's mathematics."""
tier: ScaleTier = "workstation"
device: str | torch.device | None = "auto"
per_device_batch_size: int = 8
gradient_accumulation_steps: int = 1
mixed_precision: PrecisionName = "none"
workers: int = 4
pin_memory: bool = True
persistent_workers: bool = True
distributed: bool = False
compile_model: bool = False
channels_last: bool = False
checkpoint_path: str | Path | None = None
seed: int = 0
def __post_init__(self) -> None:
if self.tier not in {"smoke", "workstation", "full"}:
raise ValueError("tier must be smoke, workstation, or full")
if self.per_device_batch_size < 1:
raise ValueError("per_device_batch_size must be positive")
if self.gradient_accumulation_steps < 1:
raise ValueError("gradient_accumulation_steps must be positive")
if self.workers < 0:
raise ValueError("workers must be nonnegative")
if self.mixed_precision not in {"none", "float16", "bfloat16"}:
raise ValueError("mixed_precision must be none, float16, or bfloat16")
if self.persistent_workers and self.workers == 0:
raise ValueError("persistent_workers requires workers > 0")
def effective_batch_size(self, *, world_size: int = 1) -> int:
"""Return per-device batch times accumulation times process count."""
if world_size < 1:
raise ValueError("world_size must be positive")
return self.per_device_batch_size * self.gradient_accumulation_steps * world_size
def train_config(self, **overrides: Any) -> TrainConfig:
"""Create a training configuration carrying the scale-sensitive fields."""
values: dict[str, Any] = {
"gradient_accumulation_steps": self.gradient_accumulation_steps,
"mixed_precision": self.mixed_precision,
"device": self.device,
"checkpoint_path": self.checkpoint_path,
"seed": self.seed,
"resume": self.checkpoint_path is not None,
}
values.update(overrides)
return TrainConfig(**values)
def data_config(self, **overrides: Any) -> SILVADataLoaderConfig:
"""Create a data-loader configuration for this runtime."""
values: dict[str, Any] = {
"batch_size": self.per_device_batch_size,
"workers": self.workers,
"pin_memory": self.pin_memory,
"persistent_workers": self.persistent_workers,
"distributed": self.distributed,
"seed": self.seed,
}
values.update(overrides)
return SILVADataLoaderConfig(**values)
|
data_config
Create a data-loader configuration for this runtime.
Source code in src/silva_networks/scaling.py
| def data_config(self, **overrides: Any) -> SILVADataLoaderConfig:
"""Create a data-loader configuration for this runtime."""
values: dict[str, Any] = {
"batch_size": self.per_device_batch_size,
"workers": self.workers,
"pin_memory": self.pin_memory,
"persistent_workers": self.persistent_workers,
"distributed": self.distributed,
"seed": self.seed,
}
values.update(overrides)
return SILVADataLoaderConfig(**values)
|
effective_batch_size
effective_batch_size(*, world_size=1)
Return per-device batch times accumulation times process count.
Source code in src/silva_networks/scaling.py
| def effective_batch_size(self, *, world_size: int = 1) -> int:
"""Return per-device batch times accumulation times process count."""
if world_size < 1:
raise ValueError("world_size must be positive")
return self.per_device_batch_size * self.gradient_accumulation_steps * world_size
|
train_config
train_config(**overrides)
Create a training configuration carrying the scale-sensitive fields.
Source code in src/silva_networks/scaling.py
| def train_config(self, **overrides: Any) -> TrainConfig:
"""Create a training configuration carrying the scale-sensitive fields."""
values: dict[str, Any] = {
"gradient_accumulation_steps": self.gradient_accumulation_steps,
"mixed_precision": self.mixed_precision,
"device": self.device,
"checkpoint_path": self.checkpoint_path,
"seed": self.seed,
"resume": self.checkpoint_path is not None,
}
values.update(overrides)
return TrainConfig(**values)
|
all_silva_family_guides
all_silva_family_guides()
Return guides in the same order as :func:available_silva_families.
Source code in src/silva_networks/scaling.py
| def all_silva_family_guides() -> tuple[SILVAFamilyGuide, ...]:
"""Return guides in the same order as :func:`available_silva_families`."""
return tuple(_FAMILY_GUIDES[name] for name in available_silva_families())
|
audit_silva_family_guides
audit_silva_family_guides()
Return coverage errors; an empty tuple means every family is actionable.
Source code in src/silva_networks/scaling.py
| def audit_silva_family_guides() -> tuple[str, ...]:
"""Return coverage errors; an empty tuple means every family is actionable."""
errors: list[str] = []
expected = set(available_silva_families())
actual = set(_FAMILY_GUIDES)
for missing in sorted(expected - actual):
errors.append(f"missing family guide: {missing}")
for extra in sorted(actual - expected):
errors.append(f"unknown family guide: {extra}")
for guide in _FAMILY_GUIDES.values():
for field in (
"data_contract",
"paper_refs",
"benchmark_tasks",
"scale_controls",
"extension_points",
):
if not getattr(guide, field):
errors.append(f"{guide.family}: empty {field}")
return tuple(errors)
|
build_scaled_silva
build_scaled_silva(family, *, tier='full', **kwargs)
Build a SILVA family with scalable numerical defaults and user dimensions.
Explicit keyword arguments always win. Task-specific dimensions, modules,
schedules, and constraints remain required by the selected family.
Source code in src/silva_networks/scaling.py
| def build_scaled_silva(
family: str,
*,
tier: ScaleTier = "full",
**kwargs: Any,
) -> Any:
"""Build a SILVA family with scalable numerical defaults and user dimensions.
Explicit keyword arguments always win. Task-specific dimensions, modules,
schedules, and constraints remain required by the selected family.
"""
key = canonical_silva_family(family)
defaults = silva_scaling_defaults(key, tier=tier)
if key == "sequence_deq" and kwargs.get("mode", "transformer") != "transformer":
defaults.pop("local_window", None)
if key == "silva_monotone_graph_equilibrium" and "state_dim" in kwargs:
defaults["operator_rank"] = min(64, int(kwargs["state_dim"]))
defaults.update(kwargs)
return silva_equilibrium_model(key, **defaults)
|
full_scale_solver_config
full_scale_solver_config(*, batch_dims=1, tier='full')
Return a relative-residual, implicit-backward SILVA solver configuration.
Source code in src/silva_networks/scaling.py
| def full_scale_solver_config(*, batch_dims: int = 1, tier: ScaleTier = "full") -> SolverConfig:
"""Return a relative-residual, implicit-backward SILVA solver configuration."""
if batch_dims < 0:
raise ValueError("batch_dims must be nonnegative")
if tier not in {"smoke", "workstation", "full"}:
raise ValueError("tier must be smoke, workstation, or full")
budgets = {
"smoke": (12, 20, 3),
"workstation": (35, 50, 5),
"full": (60, 80, 6),
}
forward, backward, history = budgets[tier]
return SolverConfig(
solver="anderson",
max_iter=forward,
tol=1e-5,
history=history,
stop_mode="relative",
anderson_batch_dims=batch_dims,
backward_mode="implicit",
backward_solver="gmres",
backward_max_iter=backward,
backward_tol=1e-5,
backward_stop_mode="relative",
return_best=True,
)
|
prepare_silva_model
prepare_silva_model(model, runtime, *, local_rank=None, find_unused_parameters=False)
Move, optionally distribute, and optionally compile a SILVA model.
Source code in src/silva_networks/scaling.py
| def prepare_silva_model(
model: nn.Module,
runtime: SILVARuntimeConfig,
*,
local_rank: int | None = None,
find_unused_parameters: bool = False,
) -> nn.Module:
"""Move, optionally distribute, and optionally compile a SILVA model."""
device = resolve_device(runtime.device)
if runtime.distributed:
if not dist.is_available() or not dist.is_initialized():
raise RuntimeError("distributed runtime requires an initialized process group")
if device.type == "cuda" and local_rank is not None:
device = torch.device("cuda", local_rank)
torch.cuda.set_device(device)
model = model.to(device)
if runtime.channels_last:
model = model.to(memory_format=torch.channels_last)
if runtime.distributed:
device_ids = [device.index] if device.type == "cuda" else None
model = DistributedDataParallel(
model,
device_ids=device_ids,
find_unused_parameters=find_unused_parameters,
)
if runtime.compile_model:
model = torch.compile(model)
return model
|
runtime_for_tier
runtime_for_tier(tier, **overrides)
Return a conservative runtime template for smoke, workstation, or full runs.
Source code in src/silva_networks/scaling.py
| def runtime_for_tier(tier: ScaleTier, **overrides: Any) -> SILVARuntimeConfig:
"""Return a conservative runtime template for smoke, workstation, or full runs."""
templates = {
"smoke": SILVARuntimeConfig(
tier="smoke",
mixed_precision="none",
per_device_batch_size=4,
workers=0,
pin_memory=False,
persistent_workers=False,
),
"workstation": SILVARuntimeConfig(tier="workstation"),
"full": SILVARuntimeConfig(
tier="full",
per_device_batch_size=8,
gradient_accumulation_steps=4,
mixed_precision="bfloat16",
workers=8,
distributed=True,
),
}
if tier not in templates:
raise ValueError("tier must be smoke, workstation, or full")
return replace(templates[tier], **overrides)
|
silva_family_guide
silva_family_guide(family)
Return the execution and extension guide for a family or alias.
Source code in src/silva_networks/scaling.py
| def silva_family_guide(family: str) -> SILVAFamilyGuide:
"""Return the execution and extension guide for a family or alias."""
return _FAMILY_GUIDES[canonical_silva_family(family)]
|
silva_scaling_defaults
silva_scaling_defaults(family, *, tier='full')
Return scale-sensitive constructor defaults without choosing task dimensions.
Source code in src/silva_networks/scaling.py
| def silva_scaling_defaults(family: str, *, tier: ScaleTier = "full") -> dict[str, Any]:
"""Return scale-sensitive constructor defaults without choosing task dimensions."""
key = canonical_silva_family(family)
defaults: dict[str, Any] = {}
if key in _SOLVER_CONFIG_FAMILIES:
batch_dims = 0 if key in _COUPLED_GRAPH_FAMILIES else 1
config = full_scale_solver_config(batch_dims=batch_dims, tier=tier)
source_solvers = {
"silva_psi_gnn": "broyden",
"silva_snarf": "broyden",
"silva_mesh_inference": "picard",
}
if key in source_solvers:
config = replace(config, solver=source_solvers[key])
defaults["config"] = config
if key in {"silva_graph_preset", "silva_image_cortex"}:
defaults.update(
solver="anderson",
backward_mode="implicit",
backward_solver="gmres",
max_iter=60 if tier == "full" else 35,
)
if key == "sequence_deq":
defaults["local_window"] = 256
elif key == "silva_distributional_deq":
defaults["pairwise_chunk_size"] = 256
elif key == "silva_generative_equilibrium_transformer":
defaults.update(attention_mode="sdpa", query_chunk_size=256)
elif key == "silva_physics_informed_equilibrium":
defaults.update(derivative_mode="matrix_free", derivative_max_iter=80)
elif key == "silva_implicit_dae_step":
defaults.update(linear_solver="gmres", linear_max_iter=80)
elif key == "silva_consistency_deq":
defaults["teacher_config"] = full_scale_solver_config(batch_dims=1, tier=tier)
elif key == "silva_hyper_deq":
defaults["teacher_config"] = replace(
full_scale_solver_config(batch_dims=0, tier=tier),
solver="broyden",
)
return defaults
|
Where to Go Next