Download src/hamiltonzero/modes/train.py from simulacra-research/HamiltonZero: direct link, hf CLI and curl.
- Browser
- Download file 30.3 kB
-
https://huggingface.co/simulacra-research/HamiltonZero/resolve/main/src/hamiltonzero/modes/train.py
- Command line
-
hf download hf://simulacra-research/HamiltonZero/src/hamiltonzero/modes/train.py
-
curl -L -o train.py https://huggingface.co/simulacra-research/HamiltonZero/resolve/main/src/hamiltonzero/modes/train.py
30.3 kB
| # Copyright (c) 2026 Simulacra Research Inc. | |
| # SPDX-License-Identifier: Apache-2.0 | |
| from __future__ import annotations | |
| import time | |
| from dataclasses import dataclass | |
| from typing import Callable | |
| import equinox as eqx | |
| import jax | |
| import jax.numpy as jnp | |
| import numpy as np | |
| from jax.sharding import Mesh, NamedSharding, PartitionSpec as P | |
| from hamiltonzero.checkpoint import load_mcmc, load_model, save_model | |
| from hamiltonzero.compiled.tree import ( | |
| bind_physical_compiler_kernel, | |
| compile_physical_tree_from_shared_trunk, | |
| ) | |
| from hamiltonzero.compiled.trunk import ( | |
| bind_shared_kernel, | |
| bind_trunk_compiler_kernel, | |
| compile_shared_trunk_from_kernel, | |
| ) | |
| from hamiltonzero.compiled.types import ( | |
| CompiledWaveFunction, | |
| ) | |
| from hamiltonzero.config import TrainConfig | |
| from hamiltonzero.data import build_context_and_energy, load_systems | |
| from hamiltonzero.energy import vmc_energy_custom_lap_compiled | |
| from hamiltonzero.energy.custom_lap import build_W_levels | |
| from hamiltonzero.energy.frame import compile_energy_frame | |
| from hamiltonzero.mcmc import ( | |
| REState, | |
| adapt_batched, | |
| cold_samples, | |
| init_batched_state, | |
| run_batched, | |
| ) | |
| from hamiltonzero.model import MultiSystemContext, build_model | |
| from hamiltonzero.model.tree import project_tree_ngpt_rownorm | |
| from hamiltonzero.optim import ( | |
| KFACBundle, | |
| apply_router_kfac_step, | |
| init_router_kfac_state, | |
| learning_rate, | |
| process_route_targets, | |
| ) | |
| from hamiltonzero.router import ( | |
| ROUTE_SAMPLES, | |
| bind_router_kernel, | |
| build_beam16, | |
| build_route_sampler, | |
| compile_router_static, | |
| rebase_cold_samples, | |
| reframe_state_context, | |
| snis_mode_baseline, | |
| ) | |
| class TrainMetric: | |
| step: int | |
| system: int | |
| energy: float | |
| energy_std: float | |
| step_walltime: float | |
| walltime: float | |
| class TrainResult: | |
| model: object | |
| kfac: KFACBundle | |
| mcmc_states: tuple[REState | None, ...] | |
| last_metric: TrainMetric | None | |
| class _SystemState: | |
| sampler: REState | |
| context: MultiSystemContext | |
| perms: jax.Array | |
| def _identity_perms(n_max: int): | |
| return jnp.broadcast_to( | |
| jnp.arange(n_max, dtype=jnp.int32), | |
| (ROUTE_SAMPLES, n_max), | |
| ) | |
| def _systems_sharding(mesh: Mesh, value): | |
| sharding = NamedSharding(mesh, P("systems")) | |
| return jax.tree_util.tree_map(lambda _value: sharding, value) | |
| def _replicated_sharding(mesh: Mesh, value): | |
| sharding = NamedSharding(mesh, P()) | |
| return jax.tree_util.tree_map(lambda _value: sharding, value) | |
| def _replicate(mesh: Mesh, value): | |
| return jax.device_put(value, _replicated_sharding(mesh, value)) | |
| def _place_routes(mesh: Mesh, value): | |
| return jax.device_put(value, _systems_sharding(mesh, value)) | |
| def _host_pool(value): | |
| def materialize(x): | |
| if not eqx.is_array(x): | |
| return x | |
| result = np.asarray(jax.device_get(x)) | |
| if not result.flags.writeable: | |
| result = result.copy() | |
| return result | |
| return jax.tree_util.tree_map(materialize, value) | |
| def _host_system_state(state: _SystemState) -> _SystemState: | |
| return _SystemState( | |
| sampler=_host_pool(state.sampler), | |
| context=_host_pool(state.context), | |
| perms=_host_pool(state.perms), | |
| ) | |
| def _activate_system(mesh: Mesh, state: _SystemState) -> _SystemState: | |
| return _SystemState( | |
| sampler=_place_routes(mesh, state.sampler), | |
| context=_place_routes(mesh, state.context), | |
| perms=_place_routes(mesh, state.perms), | |
| ) | |
| def _abstract(value): | |
| return jax.tree_util.tree_map( | |
| lambda leaf: jax.ShapeDtypeStruct(leaf.shape, leaf.dtype), | |
| value, | |
| ) | |
| def _owner_reduce(value): | |
| owner = jax.lax.axis_index("systems") == 0 | |
| if jnp.issubdtype(value.dtype, jnp.bool_): | |
| return jax.lax.pmax(jnp.where(owner, value, jnp.zeros_like(value)), "systems") | |
| return jax.lax.psum(jnp.where(owner, value, jnp.zeros_like(value)), "systems") | |
| def _build_owner_entry(mesh: Mesh, function, templates): | |
| output_template = jax.eval_shape(function, *_abstract(templates)) | |
| input_specs = jax.tree_util.tree_map(lambda _value: P(), templates) | |
| output_specs = jax.tree_util.tree_map(lambda _value: P(), output_template) | |
| def local(*values): | |
| owner = jax.lax.axis_index("systems") == 0 | |
| output = jax.lax.cond( | |
| owner, | |
| lambda args: function(*args), | |
| lambda _args: jax.tree_util.tree_map( | |
| lambda value: jnp.zeros(value.shape, value.dtype), | |
| output_template, | |
| ), | |
| values, | |
| ) | |
| return jax.tree_util.tree_map(_owner_reduce, output) | |
| mapped = jax.shard_map( | |
| local, | |
| mesh=mesh, | |
| in_specs=input_specs, | |
| out_specs=output_specs, | |
| check_vma=False, | |
| ) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=_replicated_sharding(mesh, templates), | |
| out_shardings=_replicated_sharding(mesh, output_template), | |
| ) | |
| def _compile_tree_local(physical_kernel, trunk, perms): | |
| tree = compile_physical_tree_from_shared_trunk( | |
| physical_kernel, | |
| trunk, | |
| perms[0], | |
| ) | |
| return jax.tree_util.tree_map(lambda value: value[None], tree) | |
| def _build_compile_trees(mesh: Mesh, physical_kernel, trunk, perms): | |
| local_perms = jax.ShapeDtypeStruct((1, perms.shape[1]), perms.dtype) | |
| local_output = jax.eval_shape( | |
| _compile_tree_local, | |
| _abstract(physical_kernel), | |
| _abstract(trunk), | |
| local_perms, | |
| ) | |
| mapped = jax.shard_map( | |
| _compile_tree_local, | |
| mesh=mesh, | |
| in_specs=(P(), P(), P("systems")), | |
| out_specs=jax.tree_util.tree_map(lambda _value: P("systems"), local_output), | |
| check_vma=False, | |
| ) | |
| output_template = jax.eval_shape( | |
| mapped, | |
| _abstract(physical_kernel), | |
| _abstract(trunk), | |
| _abstract(perms), | |
| ) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| _replicated_sharding(mesh, physical_kernel), | |
| _replicated_sharding(mesh, trunk), | |
| NamedSharding(mesh, P("systems")), | |
| ), | |
| out_shardings=_systems_sharding(mesh, output_template), | |
| ) | |
| def _compile_frame_local(inputs, mask, bmask, perms): | |
| frame = compile_energy_frame(inputs, mask, bmask, perms[0]) | |
| return jax.tree_util.tree_map(lambda value: value[None], frame) | |
| def _build_compile_frames(mesh: Mesh, inputs, mask, bmask, perms): | |
| local_perms = jax.ShapeDtypeStruct((1, perms.shape[1]), perms.dtype) | |
| local_output = jax.eval_shape( | |
| _compile_frame_local, | |
| _abstract(inputs), | |
| _abstract(mask), | |
| _abstract(bmask), | |
| local_perms, | |
| ) | |
| mapped = jax.shard_map( | |
| _compile_frame_local, | |
| mesh=mesh, | |
| in_specs=(P(), P(), P(), P("systems")), | |
| out_specs=jax.tree_util.tree_map(lambda _value: P("systems"), local_output), | |
| check_vma=False, | |
| ) | |
| output_template = jax.eval_shape( | |
| mapped, | |
| _abstract(inputs), | |
| _abstract(mask), | |
| _abstract(bmask), | |
| _abstract(perms), | |
| ) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| _replicated_sharding(mesh, inputs), | |
| NamedSharding(mesh, P()), | |
| NamedSharding(mesh, P()), | |
| NamedSharding(mesh, P("systems")), | |
| ), | |
| out_shardings=_systems_sharding(mesh, output_template), | |
| ) | |
| def _run_routes_local(state, kernel, trees, n_steps, chunk_size): | |
| sampler = jax.tree_util.tree_map(lambda value: value[0], state) | |
| tree = jax.tree_util.tree_map(lambda value: value[0], trees) | |
| sampler = run_batched( | |
| CompiledWaveFunction(kernel=kernel, tree=tree), | |
| None, | |
| sampler, | |
| int(n_steps), | |
| walker_chunk_size=chunk_size, | |
| ) | |
| return jax.tree_util.tree_map(lambda value: value[None], sampler) | |
| def _build_run_routes( | |
| mesh: Mesh, | |
| state, | |
| kernel, | |
| trees, | |
| *, | |
| n_steps: int, | |
| chunk_size: int | None, | |
| ): | |
| state_specs = jax.tree_util.tree_map(lambda _value: P("systems"), state) | |
| tree_specs = jax.tree_util.tree_map(lambda _value: P("systems"), trees) | |
| def local(state_value, kernel_value, trees_value): | |
| output = _run_routes_local( | |
| state_value, | |
| kernel_value, | |
| trees_value, | |
| int(n_steps), | |
| chunk_size, | |
| ) | |
| local_count = jnp.asarray( | |
| output.q.shape[0] * output.q.shape[1], | |
| dtype=jnp.int32, | |
| ) | |
| global_count = jax.lax.psum(local_count, "systems") | |
| guard = global_count.astype(output.q.dtype) * jnp.asarray(0.0, output.q.dtype) | |
| return eqx.tree_at( | |
| lambda value: value.q, | |
| output, | |
| output.q + guard, | |
| ) | |
| mapped = jax.shard_map( | |
| local, | |
| mesh=mesh, | |
| in_specs=(state_specs, P(), tree_specs), | |
| out_specs=state_specs, | |
| check_vma=False, | |
| ) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| _systems_sharding(mesh, state), | |
| _replicated_sharding(mesh, kernel), | |
| _systems_sharding(mesh, trees), | |
| ), | |
| out_shardings=_systems_sharding(mesh, state), | |
| donate_argnums=(0,), | |
| ) | |
| def _adapt_routes(state, config): | |
| return jax.vmap( | |
| lambda sampler: adapt_batched( | |
| sampler, | |
| beta_history_weight=config.mcmc.beta_history_weight, | |
| sigma_target=config.mcmc.langevin_target_acceptance, | |
| sigma_scale=config.mcmc.sigma_scale, | |
| haar_target=config.mcmc.haar_target_acceptance, | |
| ) | |
| )(state) | |
| def _sampled_energy(kernel, trees, frames, q, chunk_size): | |
| def one(tree, frame, q_row): | |
| n_sites = int(q_row.shape[-2]) | |
| frame = eqx.tree_at( | |
| lambda value: value.w_levels, | |
| frame, | |
| tuple(build_W_levels(frame.custom_lap_J_eff, n_sites)), | |
| ) | |
| return vmc_energy_custom_lap_compiled( | |
| kernel, | |
| tree, | |
| frame, | |
| q_row, | |
| chunk_size=chunk_size, | |
| ) | |
| return jax.vmap(one)(trees, frames, q) | |
| def _build_sampled_energy(mesh: Mesh, chunk_size: int): | |
| systems = P("systems") | |
| system_batch = P("systems", None) | |
| def local_energy(kernel, trees, frames, q): | |
| outputs = _sampled_energy( | |
| kernel, | |
| trees, | |
| frames, | |
| q, | |
| int(chunk_size), | |
| ) | |
| local_count = jnp.asarray(q.shape[0] * q.shape[1], dtype=jnp.int32) | |
| global_count = jax.lax.psum(local_count, "systems") | |
| guard = global_count.astype(jnp.float32) * jnp.asarray(0.0, jnp.float32) | |
| return tuple(value + guard.astype(value.dtype) for value in outputs) | |
| mapped = jax.shard_map( | |
| local_energy, | |
| mesh=mesh, | |
| in_specs=(P(), systems, systems, system_batch), | |
| out_specs=(system_batch,) * 4, | |
| check_vma=False, | |
| ) | |
| replicated_sharding = NamedSharding(mesh, P()) | |
| systems_sharding = NamedSharding(mesh, systems) | |
| system_batch_sharding = NamedSharding(mesh, system_batch) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| replicated_sharding, | |
| systems_sharding, | |
| systems_sharding, | |
| system_batch_sharding, | |
| ), | |
| out_shardings=(system_batch_sharding,) * 4, | |
| ) | |
| def _build_reframe(mesh: Mesh, state, context, perms): | |
| state_specs = jax.tree_util.tree_map(lambda _value: P("systems"), state) | |
| context_specs = jax.tree_util.tree_map(lambda _value: P("systems"), context) | |
| def local(state_value, context_value, old_perms, new_perms): | |
| return reframe_state_context( | |
| state_value, | |
| context_value, | |
| old_perms, | |
| new_perms, | |
| ) | |
| mapped = jax.shard_map( | |
| local, | |
| mesh=mesh, | |
| in_specs=( | |
| state_specs, | |
| context_specs, | |
| P("systems"), | |
| P("systems"), | |
| ), | |
| out_specs=(state_specs, context_specs), | |
| check_vma=False, | |
| ) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| _systems_sharding(mesh, state), | |
| _systems_sharding(mesh, context), | |
| NamedSharding(mesh, P("systems")), | |
| NamedSharding(mesh, P("systems")), | |
| ), | |
| out_shardings=( | |
| _systems_sharding(mesh, state), | |
| _systems_sharding(mesh, context), | |
| ), | |
| ) | |
| def _build_rebase(mesh: Mesh, q, perms): | |
| mapped = jax.shard_map( | |
| rebase_cold_samples, | |
| mesh=mesh, | |
| in_specs=(P("systems"), P("systems")), | |
| out_specs=P("systems"), | |
| check_vma=False, | |
| ) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| NamedSharding(mesh, P("systems")), | |
| NamedSharding(mesh, P("systems")), | |
| ), | |
| out_shardings=NamedSharding(mesh, P("systems")), | |
| ) | |
| def _mode_energy(kernel, tree, frame, q_canonical, mode_perm, chunk_size): | |
| q_mode = jnp.take(q_canonical, mode_perm, axis=-2) | |
| def one(q_row): | |
| with jax.default_matmul_precision("default"): | |
| log_p = 2.0 * jax.vmap( | |
| lambda walker: CompiledWaveFunction(kernel, tree)(walker)[0] | |
| )(q_row) | |
| total, exchange, casimir, field = vmc_energy_custom_lap_compiled( | |
| kernel, | |
| tree, | |
| frame, | |
| q_row, | |
| chunk_size=chunk_size, | |
| ) | |
| return total, exchange, casimir, field, log_p | |
| return jax.vmap(one)(q_mode) | |
| def _build_mode_energy(mesh: Mesh, kernel, tree, frame, q, mode_perm, chunk_size): | |
| system_batch = P("systems", None) | |
| def local(kernel_value, tree_value, frame_value, q_value, perm_value): | |
| return _mode_energy( | |
| kernel_value, | |
| tree_value, | |
| frame_value, | |
| q_value, | |
| perm_value, | |
| int(chunk_size), | |
| ) | |
| mapped = jax.shard_map( | |
| local, | |
| mesh=mesh, | |
| in_specs=(P(), P(), P(), system_batch, P()), | |
| out_specs=(system_batch,) * 5, | |
| check_vma=False, | |
| ) | |
| output_sharding = NamedSharding(mesh, system_batch) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=( | |
| _replicated_sharding(mesh, kernel), | |
| _replicated_sharding(mesh, tree), | |
| _replicated_sharding(mesh, frame), | |
| NamedSharding(mesh, system_batch), | |
| NamedSharding(mesh, P()), | |
| ), | |
| out_shardings=(output_sharding,) * 5, | |
| ) | |
| def _compile_mode(physical_kernel, trunk, inputs, mask, bmask, perm): | |
| tree = compile_physical_tree_from_shared_trunk(physical_kernel, trunk, perm) | |
| frame = compile_energy_frame(inputs, mask, bmask, perm) | |
| return tree, frame | |
| def _build_exact_skip(mesh: Mesh): | |
| def local(sampled, mode): | |
| equal = jnp.all(sampled.astype(jnp.int32) == mode[None].astype(jnp.int32)) | |
| return jax.lax.pmin(equal.astype(jnp.int32), "systems").astype(jnp.bool_) | |
| mapped = jax.shard_map( | |
| local, | |
| mesh=mesh, | |
| in_specs=(P("systems"), P()), | |
| out_specs=P(), | |
| check_vma=False, | |
| ) | |
| replicated = NamedSharding(mesh, P()) | |
| return jax.jit( | |
| mapped, | |
| in_shardings=(NamedSharding(mesh, P("systems")), replicated), | |
| out_shardings=replicated, | |
| ) | |
| def _initial_model(config: TrainConfig, key): | |
| model = build_model(config.model, key, n_max=config.n_max) | |
| if config.checkpoint is not None: | |
| model = load_model(config.checkpoint, model) | |
| return model | |
| def _run_route_burn_in( | |
| sampler, | |
| kernel, | |
| trees, | |
| config, | |
| run_routes, | |
| adapt, | |
| ): | |
| for iteration in range(config.mcmc.burn_in): | |
| sampler = run_routes( | |
| sampler, | |
| kernel, | |
| trees, | |
| ) | |
| if iteration and iteration % config.mcmc.adapt_every == 0: | |
| sampler = adapt(sampler) | |
| return sampler | |
| def _initial_system_state( | |
| model, | |
| context, | |
| config, | |
| mcmc_key, | |
| system_index, | |
| *, | |
| n_systems, | |
| mesh, | |
| compile_plan, | |
| compile_trees, | |
| get_run_routes, | |
| ): | |
| walkers = config.mcmc.batch_size // ROUTE_SAMPLES | |
| cpu = jax.devices("cpu")[0] | |
| mcmc_key = jax.device_put(mcmc_key, cpu) | |
| with jax.default_device(cpu): | |
| lane_indices = system_index * ROUTE_SAMPLES + jnp.arange( | |
| ROUTE_SAMPLES, dtype=jnp.int32 | |
| ) | |
| keys = jax.vmap(lambda index: jax.random.fold_in(mcmc_key, index))(lane_indices) | |
| sampler = jax.vmap( | |
| lambda lane_key: init_batched_state( | |
| lane_key, | |
| context, | |
| batch_size=walkers, | |
| n_replicas=config.mcmc.replicas, | |
| initial_m=config.mcmc.initial_haar_sites, | |
| initial_sigma=config.mcmc.initial_sigma, | |
| ) | |
| )(keys) | |
| contexts = MultiSystemContext.stack([context] * ROUTE_SAMPLES) | |
| perms = _identity_perms(config.n_max) | |
| if config.mcmc.reuse_mcmc is not None: | |
| source = config.mcmc.reuse_mcmc | |
| if source.is_dir(): | |
| source = source / f"{system_index}.eqx" | |
| elif n_systems != 1: | |
| raise ValueError( | |
| "multisystem --reuse-mcmc must point to a directory of " | |
| "<system-index>.eqx files" | |
| ) | |
| sampler = load_mcmc(source, sampler) | |
| sampler = _place_routes(mesh, sampler) | |
| contexts = _place_routes(mesh, contexts) | |
| perms = _place_routes(mesh, perms) | |
| trunk = compile_plan(bind_trunk_compiler_kernel(model), context) | |
| physical_kernel = bind_physical_compiler_kernel(model) | |
| trees = compile_trees(physical_kernel, trunk, perms) | |
| run_routes = get_run_routes( | |
| sampler, | |
| bind_shared_kernel(model), | |
| trees, | |
| config.mcmc.burn_in_replica_steps, | |
| ) | |
| adapt = jax.jit( | |
| lambda state: _adapt_routes(state, config), | |
| in_shardings=(_systems_sharding(mesh, sampler),), | |
| out_shardings=_systems_sharding(mesh, sampler), | |
| ) | |
| sampler = _run_route_burn_in( | |
| sampler, | |
| bind_shared_kernel(model), | |
| trees, | |
| config, | |
| run_routes, | |
| adapt, | |
| ) | |
| return _SystemState(sampler=sampler, context=contexts, perms=perms) | |
| def _metric(step, system_index, total, step_started, run_started): | |
| jax.block_until_ready(total) | |
| return TrainMetric( | |
| step=step, | |
| system=system_index, | |
| energy=float(jax.device_get(jnp.mean(total.real))), | |
| energy_std=float(jax.device_get(jnp.std(total.real))), | |
| step_walltime=time.perf_counter() - step_started, | |
| walltime=time.perf_counter() - run_started, | |
| ) | |
| def run_train( | |
| config: TrainConfig, | |
| *, | |
| metric_sink: Callable[[TrainMetric], None] | None = None, | |
| ) -> TrainResult: | |
| if config.mcmc.batch_size % ROUTE_SAMPLES: | |
| raise ValueError("mcmc.batch_size must be divisible by K=8") | |
| if config.mcmc.burn_in < 0: | |
| raise ValueError("mcmc.burn_in must be non-negative") | |
| systems = load_systems(config.systems) | |
| if not systems: | |
| raise ValueError("training requires at least one system") | |
| systems_data = [ | |
| _host_pool( | |
| build_context_and_energy( | |
| system, | |
| n_max=config.n_max, | |
| mu=config.energy.mu, | |
| eps=config.energy.eps, | |
| ) | |
| ) | |
| for system in systems | |
| ] | |
| contexts = [context for context, _energy in systems_data] | |
| energy_inputs = [energy for _context, energy in systems_data] | |
| key = jax.random.PRNGKey(config.seed) | |
| key_model, key_mcmc = jax.random.split(key) | |
| model = _initial_model(config, key_model) | |
| devices = tuple(jax.devices()) | |
| if len(devices) != ROUTE_SAMPLES: | |
| raise ValueError("learned-router train requires exactly eight visible devices") | |
| mesh = Mesh(np.asarray(devices, dtype=object), ("systems",)) | |
| model = _replicate(mesh, model) | |
| trunk_kernel = bind_trunk_compiler_kernel(model) | |
| compile_plan = _build_owner_entry( | |
| mesh, | |
| compile_shared_trunk_from_kernel, | |
| (trunk_kernel, contexts[0]), | |
| ) | |
| trunk_template = compile_plan(trunk_kernel, contexts[0]) | |
| physical_kernel = bind_physical_compiler_kernel(model) | |
| perms_template = _place_routes(mesh, _identity_perms(config.n_max)) | |
| compile_trees = _build_compile_trees( | |
| mesh, | |
| physical_kernel, | |
| trunk_template, | |
| perms_template, | |
| ) | |
| compile_frames = _build_compile_frames( | |
| mesh, | |
| energy_inputs[0], | |
| contexts[0].mask, | |
| contexts[0].bmask, | |
| perms_template, | |
| ) | |
| sampled_energy = _build_sampled_energy(mesh, config.energy.chunk_size) | |
| exact_skip = _build_exact_skip(mesh) | |
| mcmc_entries = {} | |
| def get_run_routes(state, kernel, trees, n_steps): | |
| entry_key = (int(n_steps), config.mcmc.walker_chunk_size) | |
| entry = mcmc_entries.get(entry_key) | |
| if entry is None: | |
| entry = _build_run_routes( | |
| mesh, | |
| state, | |
| kernel, | |
| trees, | |
| n_steps=entry_key[0], | |
| chunk_size=entry_key[1], | |
| ) | |
| mcmc_entries[entry_key] = entry | |
| return entry | |
| system_states: list[_SystemState | None] = [None] * len(systems) | |
| def get_system(index: int): | |
| cached = system_states[index] | |
| if cached is None: | |
| return _initial_system_state( | |
| model, | |
| contexts[index], | |
| config, | |
| key_mcmc, | |
| index, | |
| n_systems=len(systems), | |
| mesh=mesh, | |
| compile_plan=compile_plan, | |
| compile_trees=compile_trees, | |
| get_run_routes=get_run_routes, | |
| ) | |
| return _activate_system(mesh, cached) | |
| first = get_system(0) | |
| q_seed = jax.vmap(cold_samples)(first.sampler) | |
| system_batch_sharding = NamedSharding(mesh, P("systems", None)) | |
| systems_sharding = NamedSharding(mesh, P("systems")) | |
| q_seed = jax.device_put(q_seed, system_batch_sharding) | |
| energy_seed = jax.device_put( | |
| np.zeros(q_seed.shape[:2], dtype=np.complex64), | |
| system_batch_sharding, | |
| ) | |
| kfac = init_router_kfac_state( | |
| config.kfac, | |
| model, | |
| q_seed, | |
| energy_seed, | |
| first.context, | |
| t=0.0, | |
| key=jax.random.fold_in(key, 0xCAFE), | |
| multi_device=True, | |
| route_tau=config.router.temperature, | |
| route_loss_weight=config.router.loss_weight, | |
| ) | |
| reframe = _build_reframe(mesh, first.sampler, first.context, first.perms) | |
| rebase = _build_rebase(mesh, q_seed, first.perms) | |
| adapt_routes = jax.jit( | |
| lambda state: _adapt_routes(state, config), | |
| in_shardings=(_systems_sharding(mesh, first.sampler),), | |
| out_shardings=_systems_sharding(mesh, first.sampler), | |
| ) | |
| target_entry = jax.jit( | |
| lambda sampled, baseline, sigma, weights: process_route_targets( | |
| sampled, | |
| baseline, | |
| sigma, | |
| weights, | |
| mad_width=config.kfac.mad_clip_width, | |
| ), | |
| in_shardings=( | |
| system_batch_sharding, | |
| system_batch_sharding, | |
| systems_sharding, | |
| system_batch_sharding, | |
| ), | |
| out_shardings=(system_batch_sharding, systems_sharding), | |
| ) | |
| snis_entry = jax.jit( | |
| snis_mode_baseline, | |
| in_shardings=( | |
| system_batch_sharding, | |
| system_batch_sharding, | |
| system_batch_sharding, | |
| ), | |
| out_shardings=system_batch_sharding, | |
| ) | |
| router_kernel_template = bind_router_kernel(model) | |
| compile_router = _build_owner_entry( | |
| mesh, | |
| compile_router_static, | |
| ( | |
| router_kernel_template, | |
| trunk_template, | |
| contexts[0].route_quotient_node_key, | |
| contexts[0].route_quotient_edge_key, | |
| contexts[0].needs_fwl2, | |
| ), | |
| ) | |
| router_static_template = compile_router( | |
| router_kernel_template, | |
| trunk_template, | |
| contexts[0].route_quotient_node_key, | |
| contexts[0].route_quotient_edge_key, | |
| contexts[0].needs_fwl2, | |
| ) | |
| route_sampler = build_route_sampler( | |
| mesh, | |
| router_kernel_template.decoder, | |
| router_static_template, | |
| ) | |
| mode_sampler = build_beam16( | |
| mesh, | |
| router_kernel_template.decoder, | |
| router_static_template, | |
| ) | |
| mode_perm_template = jnp.arange(config.n_max, dtype=jnp.int32) | |
| compile_mode = _build_owner_entry( | |
| mesh, | |
| _compile_mode, | |
| ( | |
| physical_kernel, | |
| trunk_template, | |
| energy_inputs[0], | |
| contexts[0].mask, | |
| contexts[0].bmask, | |
| mode_perm_template, | |
| ), | |
| ) | |
| mode_energy = None | |
| system_states[0] = _host_system_state(first) | |
| del first, q_seed, energy_seed | |
| order_rng = np.random.default_rng(config.seed) | |
| order = np.arange(len(systems), dtype=np.int32) | |
| order_rng.shuffle(order) | |
| run_started = time.perf_counter() | |
| last_metric = None | |
| for step in range(config.steps): | |
| step_started = time.perf_counter() | |
| if step and step % len(order) == 0: | |
| order_rng.shuffle(order) | |
| system_index = int(order[step % len(order)]) | |
| state = get_system(system_index) | |
| trunk = compile_plan( | |
| bind_trunk_compiler_kernel(model), | |
| contexts[system_index], | |
| ) | |
| router_kernel = bind_router_kernel(model) | |
| router_static = compile_router( | |
| router_kernel, | |
| trunk, | |
| contexts[system_index].route_quotient_node_key, | |
| contexts[system_index].route_quotient_edge_key, | |
| contexts[system_index].needs_fwl2, | |
| ) | |
| tau = jnp.asarray(config.router.temperature, dtype=jnp.float32) | |
| key, key_route = jax.random.split(key) | |
| new_perms = route_sampler( | |
| router_kernel.decoder, router_static, key_route, tau | |
| ) | |
| mode_perm = mode_sampler( | |
| router_kernel.decoder, router_static, tau | |
| ) | |
| state.sampler, state.context = reframe( | |
| state.sampler, state.context, state.perms, new_perms | |
| ) | |
| state.perms = new_perms | |
| kernel = bind_shared_kernel(model) | |
| physical_kernel = bind_physical_compiler_kernel(model) | |
| trees = compile_trees( | |
| physical_kernel, | |
| trunk, | |
| new_perms, | |
| ) | |
| frames = compile_frames( | |
| energy_inputs[system_index], | |
| contexts[system_index].mask, | |
| contexts[system_index].bmask, | |
| new_perms, | |
| ) | |
| state.sampler = get_run_routes( | |
| state.sampler, | |
| kernel, | |
| trees, | |
| config.mcmc.steps, | |
| )(state.sampler, kernel, trees) | |
| if step and step % config.mcmc.adapt_every == 0: | |
| state.sampler = adapt_routes(state.sampler) | |
| q_cold = jax.vmap(cold_samples)(state.sampler) | |
| q_cold = jax.device_put(q_cold, system_batch_sharding) | |
| total, _exchange, _casimir, _field = sampled_energy( | |
| kernel, | |
| trees, | |
| frames, | |
| q_cold, | |
| ) | |
| baseline_is_sampled = bool(np.asarray(jax.device_get(exact_skip(new_perms, mode_perm)))) | |
| if baseline_is_sampled: | |
| baseline_total = total | |
| baseline_weights = jnp.ones_like(total.real) / total.shape[-1] | |
| else: | |
| mode_tree, mode_frame = compile_mode( | |
| physical_kernel, | |
| trunk, | |
| energy_inputs[system_index], | |
| contexts[system_index].mask, | |
| contexts[system_index].bmask, | |
| mode_perm, | |
| ) | |
| q_canonical = rebase(q_cold, new_perms) | |
| if mode_energy is None: | |
| mode_energy = _build_mode_energy( | |
| mesh, | |
| kernel, | |
| mode_tree, | |
| mode_frame, | |
| q_canonical, | |
| mode_perm, | |
| config.energy.chunk_size, | |
| ) | |
| baseline_total, _bx, _bc, _bf, candidate_log_p = mode_energy( | |
| kernel, | |
| mode_tree, | |
| mode_frame, | |
| q_canonical, | |
| mode_perm, | |
| ) | |
| sampled_log_p = state.sampler.log_p[..., -1] | |
| sampled_log_p = jax.device_put(sampled_log_p, system_batch_sharding) | |
| baseline_weights = snis_entry( | |
| baseline_total, candidate_log_p, sampled_log_p | |
| ) | |
| target, advantage = target_entry( | |
| total, | |
| baseline_total, | |
| state.context.s_norm, | |
| baseline_weights, | |
| ) | |
| target = jax.device_put(target, system_batch_sharding) | |
| advantage = jax.device_put(advantage, systems_sharding) | |
| state.context = jax.device_put( | |
| state.context, | |
| _systems_sharding(mesh, state.context), | |
| ) | |
| key, key_kfac = jax.random.split(key) | |
| model, kfac = apply_router_kfac_step( | |
| kfac, | |
| model, | |
| q_cold, | |
| target, | |
| state.context, | |
| t=0.0, | |
| key=key_kfac, | |
| momentum=config.kfac.momentum, | |
| learning_rate=learning_rate(config.kfac, step), | |
| damping=config.kfac.damping, | |
| route_advantage=advantage, | |
| route_tau=config.router.temperature, | |
| ) | |
| model = project_tree_ngpt_rownorm(model) | |
| jax.block_until_ready(model) | |
| system_states[system_index] = _host_system_state(state) | |
| last_metric = _metric(step, system_index, total, step_started, run_started) | |
| if metric_sink is not None: | |
| metric_sink(last_metric) | |
| save_model( | |
| config.output, | |
| model, | |
| kind="router", | |
| metadata={"n_max": config.n_max}, | |
| ) | |
| return TrainResult( | |
| model=model, | |
| kfac=kfac, | |
| mcmc_states=tuple( | |
| None if value is None else value.sampler for value in system_states | |
| ), | |
| last_metric=last_metric, | |
| ) | |
| __all__ = [ | |
| "TrainMetric", | |
| "TrainResult", | |
| "run_train", | |
| ] | |