Full block reference, cognitive architecture, five memory systems, neuromodulatory signals, spiking networks, and the LLM bridge layer.
blocks/linear.py)| Class | Signature | What it does |
|---|---|---|
Neuron | (in_features, istrainable=True, initializer="xavier_uniform") | Single neuron: z = w · x + b |
Linear | (in_features, out_features, ...) | Z = X @ W — no bias term |
Affine | (in_features, out_features, ...) | Z = X @ W + b |
Dense | (in_features, out_features, activation, ...) | Fused Affine + activation in one block |
blocks/activations.py)| Class | Formula | Notes |
|---|---|---|
ReLU() | max(0, Z) | Caches pre-activation Z |
LeakyReLU(alpha=0.01) | Z if Z>0 else alpha*Z | |
Sigmoid() | 1/(1+e^{-Z}) | Caches output A |
Tanh() | tanh(Z) | Caches output A |
Softmax() | Row-wise probability | |
SiLU() | Z * sigmoid(Z) | Caches pre-activation Z |
| Class | Notes |
|---|---|
Input() | Entry point; out set by NeuralNet.forward(x) |
Output() | Exit point; holds .grad injected by fit() |
Dropout(p=0.5, seed=None) | Inverted dropout; .training flag respected |
BatchNorm1D(num_features, momentum=0.9, eps=1e-5) | Tracks running_mean / running_var |
Flatten() | Reshape (N,...) → (N, -1); stores _input_shape for backward |
Concat(axis=1) | Concatenates multiple inputs; MIN_INPUTS=2 |
Add(), Multiply(), MatMul() — operate on merged inputs element-wise or by matrix multiply.
blocks/conv.py) — NCHW channel-first| Class | Signature | Notes |
|---|---|---|
Conv2D | (in_channels, out_channels, kernel_size, stride=1, padding=0) | im2col-based |
MaxPool2D | (kernel_size, stride=None) | Stores max indices for gradient routing |
AvgPool2D | (kernel_size, stride=None) | Uniform gradient split |
blocks/recurrent.py)RecurrentBlock(input_dim, hidden_dim) — h = tanh(x @ W_xh + h_prev @ W_hh + b). State persists across calls; .reset() clears it.
core/learning_rules.py)Attach to any block via block.learning_rule = ...:
STDPRule(lr, tau_plus, tau_minus, A_plus, A_minus) # spike-timing dependent plasticity
HebbianRule(lr) # Hebb's rule
OjaRule(lr) # Oja's normalized Hebbian ruleblocks/memory.py)WorkingMemory(capacity, dim)capacity slots of size dimW_q, W_k compute attention over the bufferblocks/memory.py)EpisodicStore(num_slots, input_dim, key_dim=None, value_dim=None, recon_rate=0.01)(num_slots, key_dim) keys + (num_slots, value_dim) values; FIFO write pointerW_gate learned)W_query, W_valuerecon_rate * (1 - serotonin * 0.5).occupancy property tracks slot fillblocks/memory_systems.py)SemanticMemory(num_slots, input_dim, key_dim, value_dim, similarity_threshold=0.8, ema_alpha=0.1)W_q (query projection) is a Parameterblocks/memory_systems.py)ProceduralMemory(input_dim, output_dim, num_programs=4)W_router selects among num_programs independent W_programs.freeze_program(index) locks a learned program in placeblocks/memory_systems.py)SomaticMemory(dim, decay_rate=0.9)(dim,) of recent activation patterns — not gradient-updatedsigmoid(trace @ W_sense + b_sense + serotonin)x * (1 + gain * arousal)Consolidate(episodic_store, output_dim)Mean-pools stored values from an EpisodicStore and projects to a fixed-size semantic representation.
blocks/cognitive.py)| Block | Role |
|---|---|
Amygdala(dim, momentum=0.95) | Tracks rolling baseline; computes surprise as MSE from baseline; sigmoid gate modulates output by x * gate |
ACC(dim, dampening=0.5) | Anterior cingulate cortex; conflict = variance across N merged signals; dampens mean by conflict magnitude |
MetacognitiveMonitor(input_dim, output_dim) | Computes entropy + magnitude of input; projects to output: tanh(meta_features @ W_meta + b) |
PFC(input_dim, output_dim, control_dim=None) | Prefrontal cortex; two-layer integrate + project; handles single and merged inputs |
DefaultModeNetwork(memory_sources, output_dim, residual=True) | Reads current state from all memory blocks; projects to output_dim; optional residual skip |
blocks/cortical.py)| Block | Role |
|---|---|
LateralBlock(dim, mode) | Intra-layer lateral weights; mode = "inhibitory", "excitatory", or "mixed"; diagonal zeroed in inhibitory |
TopDownFeedback | One-step-delayed feedback from higher to lower layer |
PredictiveCodingBlock | Layer predicts its input; only prediction error propagates up |
CognitiveCycle(model, memories, mechanisms, phases=None)
cycle.run() # → {phase_name: History}Runs 5 phases sequentially, each calling model.fit() for epochs_per_phase epochs with different memory/mechanism configurations:
| Phase | Key behavior |
|---|---|
EdenPhase(lr_multiplier=1.5) | All trainable; MetacognitiveMonitor frozen; amygdala low-threat |
FallPhase(dropout_rate=0.5, amygdala_sensitivity=3.0) | Monitor active; amygdala hypersensitive; high dropout |
ScaffoldingPhase(l2_strength, clip_value, conflict_weight) | Dropout 0.3; parameter clipping; episodic consolidation in post-process |
ReconsolidationPhase(recon_rate=0.1, replay_fraction=0.5) | Replays fraction of episodic memories with small noise perturbation |
RestorationPhase(lr_multiplier=0.1) | Dropout 0.1; all trainable; amygdala low-threat; fine-tuning LR |
NeuromodulatorBus (core/neuromodulator.py) — singleton NEURO_BUS. Scalar float per signal, default 0.0 = neutral.
from nnexus.core.neuromodulator import NEURO_BUS
NEURO_BUS.set("dopamine", 0.8)
NEURO_BUS.get("dopamine")
NEURO_BUS.reset()
NEURO_BUS.reset_signal("norepinephrine")
NEURO_BUS.active_signals # signals currently non-zero| Signal | Abbreviation | Downstream effects |
|---|---|---|
| Dopamine | DA | Gates EpisodicStore writes; sharpens WorkingMemory attention |
| Norepinephrine | NE | Focus multiplier in WorkingMemory attention |
| Serotonin | 5HT | ProceduralMemory routing temperature; EpisodicStore reconsolidation rate; SomaticMemory arousal baseline |
| Acetylcholine | ACh | Amplifies EpisodicStore write gate threshold; attunes WorkingMemory attention |
After each conversation turn, CognitiveBridge updates signals via EMA:
dopamine ← 0.9 * da + 0.1 * (1 - conflict)
norepinephrine ← 0.9 * ne + 0.1 * surprise
serotonin ← 0.9 * ser + 0.1 * (1 - surprise)
acetylcholine ← 0.9 * ach + 0.1 * arousalblocks/spiking.py)Three-block pipeline:
SpikeEncoder() >> LIFBlock(input_dim, n_neurons) >> SpikeDecoder(lif_block, output_dim)Driven by NeuralNet.temporal_forward(*x, T=n_timesteps):
model.temporal_forward(X, T=20) # runs chain 20 times, increments CONTEXT.t each stepLIFBlock dynamics per timestep:
V *= decay — membrane potential leaksV += (x @ W + b) * active_mask — integrate input (masked during refractory)spike = V >= threshold — fire_spike_times[fired] = tV *= (1 - spike) — reset on fireBackward: surrogate gradient 1 / (1 + |V - threshold| * 10)² (fast sigmoid derivative)
SpikeDecoder reads _spike_count / T (spike rate) and projects to continuous output.
STDPRule can be attached to LIF blocks for local Spike-Timing Dependent Plasticity alongside gradient-based training.
bridge/)Connects the cognitive architecture to an LLM, shaping responses by the current cognitive state.
from nnexus.bridge.preset import build_default_bridge
from nnexus.bridge.llm import ClaudeLLMClient, OllamaLLMClient, MockLLMClient
llm = ClaudeLLMClient(api_key="...", model="claude-sonnet-4-6")
bridge = build_default_bridge(llm=llm)
response = bridge.turn("What is attention in neural networks?")Per-turn flow:
encoder.encode(user_message) → embedding vectorNEURO_BUS from previous cognitive statemodel.forward(embedding) — run cognitive modelStateExtractor.extract() → CognitiveState(arousal, conflict, surprise, uncertainty, ...)StateInterpreter.interpret(state) → natural-language cognitive directivebase_system_prompt + [Known Facts] + [Internal Cognitive State]memory_pipeline if setLLM clients:
| Class | Backend |
|---|---|
ClaudeLLMClient(api_key, model="claude-sonnet-4-6") | Anthropic SDK |
OllamaLLMClient(model, host="localhost:11434") | Local inference, stdlib only |
MockLLMClient(responses=[...]) | Cycles canned responses; for testing |
Knowledge injection:
bridge.teach("NumPy uses row-major memory layout") # durable fact in every system prompt
bridge.get_knowledge() # list all facts
bridge.forget(index) # remove by index