diff options
Diffstat (limited to 'experiments/conv_local_smoke.py')
| -rw-r--r-- | experiments/conv_local_smoke.py | 129 |
1 files changed, 127 insertions, 2 deletions
diff --git a/experiments/conv_local_smoke.py b/experiments/conv_local_smoke.py index e8cf294..887674e 100644 --- a/experiments/conv_local_smoke.py +++ b/experiments/conv_local_smoke.py @@ -8,10 +8,12 @@ import torch import torch.nn.functional as F sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from sdil.conv import (CIFARHierarchicalFAResNet, CIFARKPResNet, - CIFARLocalResNet, CIFARSDILResNet, ConvSDILConfig, +from sdil.conv import (CIFARHierarchicalFAResNet, CIFARKPMixedTrafficResNet, + CIFARKPResNet, CIFARLocalResNet, CIFARSDILResNet, + ConvSDILConfig, channel_subspace_apical_calibration, conv_hierarchical_step, conv_kolen_pollack_step, + conv_kp_mixed_traffic_step, conv_local_step, hierarchical_mirror_observations, hierarchical_parameter_subspace_calibration, @@ -937,6 +939,128 @@ def kolen_pollack_checks(): } +def kp_mixed_traffic_checks(): + """Mixed traffic isolates subtraction from norm and preserves KP locality.""" + torch.manual_seed(211) + common = dict( + depth=8, base_width=2, seed=67, dtype=torch.float64, + normalization="batchnorm", residual_scale=1.0, + traffic_seed=4000) + net = CIFARKPMixedTrafficResNet(**common) + x = torch.randn(5, 3, 32, 32, dtype=torch.float64) + y = torch.tensor([0, 2, 4, 6, 8]) + forward = net.forward( + x, return_cache=True, training=True, update_stats=False) + output_error = (torch.softmax(forward["logits"], dim=1) + - F.one_hot(y, 10).to(torch.float64)) + instruction = net.hierarchical_teaching(output_error, forward) + + # Zero traffic and zero predictor collapse all three rules to clean KP. + zero_errors = [] + for rule in ("raw", "matched", "innovation"): + components = net.mixed_apical_components( + instruction, forward["hiddens"], rule) + zero_errors.extend(float((left - right).abs().max()) for left, right in + zip(components["used"], instruction)) + assert max(zero_errors) == 0.0 + + calibration = net.calibrate_traffic_gain( + instruction, forward["hiddens"], target_ratio=4.0) + ratio_error = max(abs(value - 4.0) for value in + calibration["realized_traffic_instruction_rms_ratio"]) + assert ratio_error < 1e-12 + + # An exact per-unit predictor removes all predictable traffic. + for slope, gain, coefficient in zip( + net.P_traffic, net.traffic_gain, net.B_traffic): + slope.copy_(gain * coefficient) + exact = net.mixed_apical_components( + instruction, forward["hiddens"], "innovation") + exact_predictor_error = max(float((left - right).abs().max()) + for left, right in zip( + exact["innovation"], instruction)) + assert exact_predictor_error < 1e-14 + + for slope, bias in zip(net.P_traffic, net.P_traffic_bias): + slope.zero_() + bias.zero_() + components = net.mixed_apical_components( + instruction, forward["hiddens"], "matched") + norm_errors = [] + direction_errors = [] + for raw, innovation, matched in zip( + components["raw"], components["innovation"], + components["matched"]): + raw_flat = raw.flatten(1) + innovation_flat = innovation.flatten(1) + matched_flat = matched.flatten(1) + norm_errors.append(float(( + (matched_flat.norm(dim=1) - innovation_flat.norm(dim=1)).abs() + / innovation_flat.norm(dim=1).clamp_min(1e-30)).max())) + direction_errors.append(float((F.cosine_similarity( + raw_flat, matched_flat, dim=1) - 1.0).abs().max())) + assert max(norm_errors) < 1e-12 + assert max(direction_errors) < 1e-12 + + # All used signals retain equal independently recomputed local KP products. + correlation_errors = [] + for rule in ("raw", "matched", "innovation"): + used = net.mixed_apical_components( + instruction, forward["hiddens"], rule)["used"] + forward_directions, _, _, output_weight, _ = ( + net.local_ascent_directions(used, output_error, forward)) + reciprocal, reciprocal_readout = net.reciprocal_feedback_directions( + used, output_error, forward) + correlation_errors.extend(float((left - right).abs().max()) + for left, right in zip( + forward_directions[1:], reciprocal[1:])) + correlation_errors.append(float( + (reciprocal_readout + output_weight.t()).abs().max())) + assert max(correlation_errors) < 1e-14 + + # Predictor plasticity consumes only the supplied soma/traffic pair: once + # those are fixed, changing every forward/feedback weight has no effect. + left = CIFARKPMixedTrafficResNet(**common) + right = CIFARKPMixedTrafficResNet(**common) + for left_gain, right_gain, source in zip( + left.traffic_gain, right.traffic_gain, net.traffic_gain): + left_gain.copy_(source) + right_gain.copy_(source) + fixed_hiddens = [value.detach().clone() for value in forward["hiddens"]] + for value in right.W + right.Q + [right.W_out, right.R_out]: + value.add_(torch.randn_like(value)) + left.predictor_step(fixed_hiddens, eta=0.1) + right.predictor_step(fixed_hiddens, eta=0.1) + predictor_independence_error = max(float((a - b).abs().max()) + for a, b in zip( + left.P_traffic + left.P_traffic_bias, + right.P_traffic + right.P_traffic_bias)) + assert predictor_independence_error == 0.0 + + net.traffic_rule = "innovation" + result = conv_kp_mixed_traffic_step( + net, x, y, ConvSDILConfig( + eta=1e-4, eta_output=1e-4, eta_P=0.1, momentum=0.0, + weight_decay=0.0, learn_A=False, learn_P=True), + step=0, rule="innovation", predictor_every=16) + assert math.isfinite(result["loss"]) and result["did_predictor_update"] + assert all(not value.requires_grad for value in + net.W + net.Q + net.P_traffic + net.P_traffic_bias + + [net.W_out, net.R_out, net.b_out]) + assert net.mixed_elementwise_ops_per_example("matched") > ( + net.mixed_elementwise_ops_per_example("raw")) + return { + "kp_traffic_zero_limit_error": max(zero_errors), + "kp_traffic_ratio_error": ratio_error, + "kp_traffic_exact_predictor_error": exact_predictor_error, + "kp_traffic_matched_norm_error": max(norm_errors), + "kp_traffic_matched_direction_error": max(direction_errors), + "kp_traffic_reciprocal_correlation_error": max(correlation_errors), + "kp_traffic_predictor_parameter_independence_error": ( + predictor_independence_error), + } + + def apical_learning_checks(): torch.manual_seed(11) net = CIFARSDILResNet(depth=8, base_width=2, seed=6) @@ -1046,6 +1170,7 @@ def main(): report.update(hierarchical_parameter_calibration_checks()) report.update(normalized_response_mirror_checks()) report.update(kolen_pollack_checks()) + report.update(kp_mixed_traffic_checks()) report.update(apical_learning_checks()) print(report) print("ALL CONVOLUTIONAL LOCAL-ELIGIBILITY CHECKS PASSED") |