def run(seed: int, steps: int | None, accelerator: str) -> tuple[dict, dict]:
lit.seed_everything(seed, workers=True)
# Split seeds are independent; rerunning a generator reproduces the same records.
train = list(random_records(rows=1536, seed=seed + 1))
in_range = list(random_records(rows=512, seed=seed + 3))
unseen = list(random_records(rows=512, seed=seed + 4, minimum=7, maximum=10))
short = list(random_records(rows=256, seed=seed + 5, maximum=5))
model = rf.Model(
d_model=32,
n_layers=2,
n_heads=4,
reduction=rf.Attention(n_layers=2),
batch_size=64,
optimizer=lambda module: torch.optim.Adam(module.parameters(), lr=0.001),
items=rf.Branch(
length=CAPACITY, attention="mha", n_layers=2, reduction=rf.Attention(n_layers=2), amount=rf.Number
),
total=rf.Number(mask=True, objective="mse"),
)
datamodule = rf.SyntheticDataModule(
model=model,
train=lambda: random_records(rows=1536, seed=seed + 1),
validate=lambda: random_records(rows=384, seed=seed + 2),
seed=seed,
)
trainer = lit.Trainer(
accelerator=accelerator,
devices=1,
max_epochs=-1,
max_steps=1100 if steps is None else min(steps, 1100),
logger=False,
enable_progress_bar=False,
enable_model_summary=False,
enable_checkpointing=False,
deterministic=True,
num_sanity_val_steps=0,
)
trainer.fit(model, datamodule=datamodule)
# Evaluate held-out answers and retain their original labels in corruption controls.
in_range_score = score(train=train, test=in_range, predicted=prediction(model, in_range))
unseen_score = score(train=train, test=unseen, predicted=prediction(model, unseen))
short_prediction = prediction(model, short)
duplicated_prediction = prediction(model, duplicate(short))
duplication_error = rmse(2.0 * short_prediction, duplicated_prediction) / float(
np.std(np.asarray([row["total"] for row in unseen], dtype=np.float64))
)
probes = equal_value_probes(value=0.65, lengths=(1, 3, 6))
probe_prediction = prediction(model, probes)
probe_target = np.asarray([row["total"] for row in probes], dtype=np.float64)
probe_error = rmse(probe_target, probe_prediction) / float(
np.std(np.asarray([row["total"] for row in unseen], dtype=np.float64))
)
metrics = {
"in_range_score": in_range_score,
"unseen_score": unseen_score,
"duplication_error": duplication_error,
"probe_error": probe_error,
"probe_target": probe_target.tolist(),
"probe_prediction": probe_prediction.tolist(),
"steps": trainer.global_step,
}
checks = {
"All measurements finite": bool(
np.isfinite([in_range_score["nrmse"], unseen_score["nrmse"], duplication_error, probe_error]).all()
),
"Seen nRMSE < 0.35, unseen nRMSE < 0.30, duplication and cardinality errors < 0.20": bool(
in_range_score["nrmse"] < 0.35
and unseen_score["nrmse"] < 0.3
and (duplication_error < 0.2)
and (probe_error < 0.2)
),
}
return metrics, checks