Seed 3605 on gpu: 3 of 3 behavioral checks met. See the measurements below. The experiment code has changed since this run; these measurements describe its earlier version.
Providing an item count lets the model distinguish totals that an average alone cannot. Every item in a record repeats the same amount, so the branch summary stays unchanged when only the number of copies changes. The ordinary root count field supplies the missing factor needed for a total.
The matched probes require different predictions for the same repeated value at different counts. This isolates learning the amount-times-count relationship from recovering multiplicity through the collection itself. It is a useful diagnostic when a reduction appears to lose mass. The proof covers repeated-value bags and familiar counts; it does not establish sums over arbitrary mixed-value collections.
"""A visible item count diagnoses cardinality lost by Mean.
Learn value times visible cardinality and distinguish matched probes.
Run: uv run python proofs/run.py P004"""
from __future__ import annotations
from collections.abc import Iterator
import lightning.pytorch as lit
import numpy as np
import torch
from reporting import report
import relflow as rf
PROOF_ID = "P004"
TRAIN_MAX = 6
CAPACITY = 12Targets use mask=True; the labels below are supervision hidden from the encoder.
item_count: 6
items:
- {amount: 0.7}
- {amount: 0.7}
- {amount: 0.7}
- {amount: 0.7}
- {amount: 0.7}
- {amount: 0.7}
total: 4.2The visible count supplies the factor needed to turn 0.7 into a total of 4.2.
item_count: 1
items:
- {amount: 0.7}
total: 0.7This matched probe has the same encoded Mean content as the six-copy record. Only the visible count distinguishes its total of 0.7.
item_count: 3
items:
- {amount: -0.4}
- {amount: -0.4}
- {amount: -0.4}
total: -1.2Three repetitions of −0.4 require −1.2. Both the repeated value and the visible count vary across the generated records.
def random_records(*, rows: int, seed: int, minimum: int = 1, maximum: int = TRAIN_MAX) -> Iterator[dict]:
"""Draw variable-length numerical bags and their mean and sum."""
if not 1 <= minimum <= maximum <= CAPACITY:
raise ValueError(f"length range must satisfy 1 <= minimum <= maximum <= {CAPACITY}")
rng = np.random.default_rng(seed)
for _ in range(rows):
length = int(rng.integers(minimum, maximum + 1))
values = np.repeat(rng.uniform(-1.0, 1.0), length)
row: dict[str, object] = {
"items": [{"amount": float(value)} for value in values],
"mean_amount": float(values.mean()),
"total": float(values.sum()),
}
row["item_count"] = length
yield row
def equal_value_probes(*, value: float, lengths: tuple[int, ...]) -> list[dict]:
"""Create matched bags that differ only in repetition count."""
if not lengths or min(lengths) < 1 or max(lengths) > CAPACITY:
raise ValueError(f"probe lengths must be within 1..{CAPACITY}, got {lengths!r}")
rows: list[dict[str, object]] = []
for length in lengths:
row: dict[str, object] = {"items": [{"amount": value}] * length, "mean_amount": value, "total": length * value}
row["item_count"] = length
rows.append(row)
return rows
def prediction(model: rf.Model, observations: list[dict]) -> np.ndarray:
inputs = [{"items": row["items"], "item_count": row["item_count"]} for row in observations]
output = model.predict(inputs)["predictions"].to_pylist()
return np.asarray([row["record/total"]["content"] for row in output], dtype=np.float64)
def rmse(actual: np.ndarray, predicted: np.ndarray | float) -> float:
return float(np.sqrt(np.mean(np.square(actual - predicted))))
def score(*, train: list[dict], test: list[dict], predicted: np.ndarray) -> dict[str, float]:
"""Compare held-out RMSE with the constant training-target mean."""
actual = np.asarray([row["total"] for row in test], dtype=np.float64)
baseline = rmse(actual, float(np.asarray([row["total"] for row in train], dtype=np.float64).mean()))
measured = rmse(actual, predicted)
return {"rmse": measured, "baseline_rmse": baseline, "nrmse": measured / baseline}
Every item in a record repeats one random amount. The branch uses attention=None and rf.Mean(), so its summary does not change with repetition count. A visible root item_count supplies the missing factor, and learned root attention combines both inputs to predict their product.
Train and test lengths are one through six. Matched probes repeat 0.7 once or six times, requiring totals 0.7 and 4.2. Their predictions must separate by more than 2.5 and stay close to those targets. This isolates learning from a supplied count; the count-free Attention proof tests structural recovery of multiplicity.
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=1024, seed=seed + 1))
test = list(random_records(rows=512, seed=seed + 3))
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=None, n_layers=2, reduction=rf.Mean(), amount=rf.Number),
item_count=rf.Number,
total=rf.Number(mask=True, objective="mse"),
)
datamodule = rf.SyntheticDataModule(
model=model,
train=lambda: random_records(rows=1024, seed=seed + 1),
validate=lambda: random_records(rows=256, seed=seed + 2),
seed=seed,
)
trainer = lit.Trainer(
accelerator=accelerator,
devices=1,
max_epochs=-1,
max_steps=700 if steps is None else min(steps, 700),
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.
predicted = prediction(model, test)
measured = score(train=train, test=test, predicted=predicted)
probes = equal_value_probes(value=0.7, lengths=(1, TRAIN_MAX))
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)
metrics = {
"measured": measured,
"probe_error": probe_error,
"probe_target": probe_target.tolist(),
"probe_prediction": probe_prediction.tolist(),
"steps": trainer.global_step,
}
checks = {
"Visible-count nRMSE below 0.25": bool(measured["nrmse"] < 0.25),
"Matched-count prediction gap above 2.5": bool(probe_prediction[1] - probe_prediction[0] > 2.5),
"Matched-count RMSE below 0.35": bool(probe_error < 0.35),
}
return metrics, checksSeed 3605, gpu, recorded 2026-09-15T02:18:07.972553+00:00. Outcome: met.
Source fingerprint: 6ce281d0517c8197cd92a45e4c1c5a22985eee2ecfa2b19afde05484eb5547db. Python 3.12.6; Torch 2.12.0.
| Measurement | Value |
|---|---|
| measured | rmse: 0.0499194; baseline_rmse: 2.23344; nrmse: 0.0223509 |
| probe_error | 0.0430748 |
| probe_target | 0.7, 4.2 |
| probe_prediction | 0.748778, 4.23649 |
| steps | 700 |
| Behavioral check | Outcome |
|---|---|
| Visible-count nRMSE below 0.25 | Met |
| Matched-count prediction gap above 2.5 | Met |
| Matched-count RMSE below 0.35 | Met |
Repeat across seeds and test nonidentical bags, missing items, and nested collections. Unseen visible counts have a separate proof; this case checks only the trained count range.
The family’s promotion target is at least three core seeds and ten lightweight calibration seeds.
Run by stable ID from the repository root:
uv run python proofs/run.py P004Or run the self-contained script directly:
PYTHONPATH=proofs uv run python proofs/aggregation/visible_count_sum.pyAdd --accelerator gpu for CUDA or --seed 42 for another seeded experiment. --steps 2 checks execution with a short training budget; it is recorded as a smoke run.
if __name__ == "__main__":
report(PROOF_ID, run, seed=3605)