"""Liquid Neural Networks — continuous-time adaptive weights. Liquid Neural Networks (LNNs) use ordinary differential equations (ODEs) to evolve hidden states continuously, enabling adaptive weight dynamics that respond to input patterns in real time. This module implements a CPU-based LNN cell and network for integration into the FusionAGI reasoning pipeline. Reference: Hasani et al., "Liquid Time-constant Networks" (2021). """ from __future__ import annotations import math from dataclasses import dataclass, field from typing import Any from fusionagi._logger import logger @dataclass class LiquidCell: """Single liquid neuron with continuous-time dynamics. The hidden state evolves according to an ODE: dh/dt = (-h + sigma(W_in * x + W_rec * h + bias)) / tau(x) where tau(x) is an input-dependent time constant that controls how quickly the cell adapts. """ input_dim: int hidden_dim: int w_in: list[list[float]] = field(default_factory=list) w_rec: list[list[float]] = field(default_factory=list) bias: list[float] = field(default_factory=list) tau_w: list[float] = field(default_factory=list) tau_bias: list[float] = field(default_factory=list) state: list[float] = field(default_factory=list) def __post_init__(self) -> None: """Initialize weights if not provided.""" if not self.w_in: scale = 1.0 / math.sqrt(self.input_dim) self.w_in = [ [scale * (((i * 7 + j * 13) % 97) / 97.0 - 0.5) * 2 for j in range(self.input_dim)] for i in range(self.hidden_dim) ] if not self.w_rec: scale = 1.0 / math.sqrt(self.hidden_dim) self.w_rec = [ [scale * (((i * 11 + j * 17) % 89) / 89.0 - 0.5) * 2 for j in range(self.hidden_dim)] for i in range(self.hidden_dim) ] if not self.bias: self.bias = [0.0] * self.hidden_dim if not self.tau_w: self.tau_w = [0.1] * self.input_dim if not self.tau_bias: self.tau_bias = [1.0] * self.hidden_dim if not self.state: self.state = [0.0] * self.hidden_dim def _sigmoid(self, x: float) -> float: """Numerically stable sigmoid.""" if x >= 0: return 1.0 / (1.0 + math.exp(-x)) ex = math.exp(x) return ex / (1.0 + ex) def _tanh(self, x: float) -> float: """Hyperbolic tangent.""" return math.tanh(x) def _compute_tau(self, x: list[float]) -> list[float]: """Compute input-dependent time constants.""" tau = [] n = min(len(x), len(self.tau_w)) for i in range(self.hidden_dim): raw = self.tau_bias[i] for j in range(n): raw += self.tau_w[j] * x[j] tau.append(max(0.1, abs(raw) + 0.5)) return tau def step(self, x: list[float], dt: float = 0.1) -> list[float]: """Advance one ODE step with Euler integration. Args: x: Input vector. dt: Integration time step. Returns: Updated hidden state. """ x_len = min(len(x), self.input_dim) tau = self._compute_tau(x) for i in range(self.hidden_dim): pre = self.bias[i] for j in range(x_len): pre += self.w_in[i][j] * x[j] for j in range(self.hidden_dim): pre += self.w_rec[i][j] * self.state[j] target = self._tanh(pre) self.state[i] += dt * (-self.state[i] + target) / tau[i] return list(self.state) def reset(self) -> None: """Reset hidden state to zeros.""" self.state = [0.0] * self.hidden_dim @dataclass class LiquidNetworkConfig: """Configuration for a Liquid Neural Network.""" input_dim: int = 64 hidden_dim: int = 32 output_dim: int = 16 num_layers: int = 2 dt: float = 0.1 steps_per_input: int = 5 class LiquidNetwork: """Multi-layer Liquid Neural Network. Stacks multiple LiquidCells for deeper temporal modeling. The final layer projects to output_dim via a simple linear readout. """ def __init__(self, config: LiquidNetworkConfig | None = None) -> None: self.config = config or LiquidNetworkConfig() self._layers: list[LiquidCell] = [] self._readout_w: list[list[float]] = [] self._readout_bias: list[float] = [] self._build() def _build(self) -> None: """Construct layers.""" cfg = self.config prev_dim = cfg.input_dim for _ in range(cfg.num_layers): self._layers.append(LiquidCell(input_dim=prev_dim, hidden_dim=cfg.hidden_dim)) prev_dim = cfg.hidden_dim scale = 1.0 / math.sqrt(cfg.hidden_dim) self._readout_w = [ [scale * (((i * 23 + j * 31) % 73) / 73.0 - 0.5) * 2 for j in range(cfg.hidden_dim)] for i in range(cfg.output_dim) ] self._readout_bias = [0.0] * cfg.output_dim def forward(self, x: list[float]) -> list[float]: """Forward pass through all layers. Args: x: Input vector of length ``input_dim``. Returns: Output vector of length ``output_dim``. """ padded = list(x) if len(padded) < self.config.input_dim: padded.extend([0.0] * (self.config.input_dim - len(padded))) elif len(padded) > self.config.input_dim: padded = padded[: self.config.input_dim] h = padded for layer in self._layers: for _ in range(self.config.steps_per_input): h = layer.step(h, dt=self.config.dt) output = [] for i in range(self.config.output_dim): val = self._readout_bias[i] for j in range(len(h)): val += self._readout_w[i][j] * h[j] output.append(math.tanh(val)) return output def forward_sequence(self, xs: list[list[float]]) -> list[list[float]]: """Process a sequence of inputs, maintaining state across steps. Args: xs: List of input vectors. Returns: List of output vectors. """ outputs = [] for x in xs: outputs.append(self.forward(x)) return outputs def reset(self) -> None: """Reset all layer states.""" for layer in self._layers: layer.reset() def adapt_weights( self, inputs: list[list[float]], targets: list[list[float]], learning_rate: float = 0.01, epochs: int = 10, ) -> dict[str, Any]: """Simple gradient-free weight adaptation using perturbation. Args: inputs: Training inputs. targets: Target outputs. learning_rate: Step size for weight updates. epochs: Number of training passes. Returns: Training summary with loss history. """ losses: list[float] = [] for epoch in range(epochs): total_loss = 0.0 self.reset() for x, target in zip(inputs, targets): output = self.forward(x) for i in range(min(len(output), len(target))): diff = output[i] - target[i] total_loss += diff * diff for layer in self._layers: for j in range(layer.hidden_dim): for k in range(layer.input_dim): layer.w_in[j][k] -= learning_rate * diff * 0.01 avg_loss = total_loss / max(len(inputs), 1) losses.append(avg_loss) if avg_loss < 1e-6: break logger.info( "LiquidNetwork adaptation complete", extra={"epochs": len(losses), "final_loss": losses[-1] if losses else 0.0}, ) return { "epochs_run": len(losses), "loss_history": losses, "final_loss": losses[-1] if losses else 0.0, } def get_summary(self) -> dict[str, Any]: """Return network summary.""" return { "type": "LiquidNetwork", "config": { "input_dim": self.config.input_dim, "hidden_dim": self.config.hidden_dim, "output_dim": self.config.output_dim, "num_layers": self.config.num_layers, "dt": self.config.dt, }, "total_parameters": sum( layer.input_dim * layer.hidden_dim + layer.hidden_dim * layer.hidden_dim + layer.hidden_dim for layer in self._layers ) + self.config.output_dim * self.config.hidden_dim, } __all__ = [ "LiquidCell", "LiquidNetwork", "LiquidNetworkConfig", ]