Liquid Neural Networks: AI, який тече як вода

Стандартна нейронка — це застиглий бетон. Навчив — зафіксував ваги — працює однаково forever. Прийшли нові дані? Модель не адаптується. Змінилось середовище? Продуктивність падає. Потрібне перенавчання — дороге, повільне, часто неможливе на edge.

Liquid Neural Networks — це вода. Вони течуть. Адаптуються в реальному часі. Ваги залишаються, але динаміка мережі змінюється залежно від вхідних даних. Кожен input створює унікальний «потік» через мережу.

MIT створив архітектуру, яка споживає в 100 разів менше енергії, працює з 19 нейронами замість мільйонів, поміщається на мікроконтролер — і при цьому керує автомобілем. Для edge computing, IoT та робототехніки — це не просто покращення. Це зміна парадигми.

Проблема традиційних рекурентних мереж

Static weights — frozen dynamics:

class TraditionalRNN:
    """Проблема: фіксовані ваги означають фіксовану поведінку"""

    def __init__(self, hidden_size):
        # Ваги зафіксовані після training
        self.W_h = nn.Linear(hidden_size, hidden_size)
        self.W_x = nn.Linear(input_size, hidden_size)

    def forward(self, x, h):
        # Одна і та ж динаміка для будь-якого input
        return torch.tanh(self.W_h(h) + self.W_x(x))

    # Проблеми:
    # 1. Навчив на даних A — працює тільки на A
    # 2. Дані змінились (distribution shift) — accuracy падає
    # 3. Для адаптації потрібне повне re-training
    # 4. Re-training на edge device неможливий

Real-world приклади проблеми:

| Сценарій | Проблема | Наслідки |

|----------|----------|----------|

| Автопілот з Каліфорнії в Україні | Сніг, інші знаки, інша розмітка | Небезпечна поведінка |

| Фітнес-трекер для нового користувача | Інша біомеханіка ходьби | Неточний підрахунок кроків |

| Predictive maintenance в новому режимі | Інші вібрації, температури | Пропущені аномалії |

| Voice assistant з новим акцентом | Інша фонетика | Низька accuracy розпізнавання |

Біологічне натхнення: C. elegans

Чому черв'як з 302 нейронами цікавий:

C. elegans — найпростіший організм з повністю картографованою нервовою системою (connectome). 302 нейрони. ~7000 синапсів. І цього достатньо для:

• Пошуку їжі
• Уникнення небезпеки
• Термотаксису
• Хемотаксису
• Соціальної поведінки

Ключове спостереження: Динаміка нейронів C. elegans — continuous-time. Нейрони не "вмикаються/вимикаються" дискретно. Вони інтегрують сигнали у часі з різними time constants.

class BiologicalNeuron:
    """Спрощена модель біологічного нейрона"""

    def __init__(self):
        self.membrane_potential = -70  # mV (rest)
        self.time_constant = 20        # ms
        self.threshold = -55           # mV

    def integrate(self, input_current: float, dt: float):
        """Continuous-time integration"""
        # Membrane potential dynamics
        dV = (-self.membrane_potential + input_current) / self.time_constant
        self.membrane_potential += dV * dt

        # Spike if threshold reached
        if self.membrane_potential > self.threshold:
            self.membrane_potential = -70  # reset
            return 1.0  # spike
        return 0.0

Liquid Time-Constant (LTC) Networks

Математична основа:

Замість дискретних update правил, LTC використовує ordinary differential equations (ODEs):

import torch
import torch.nn as nn
from torchdiffeq import odeint

class LTCCell(nn.Module):
    """Liquid Time-Constant Cell — серце liquid networks"""

    def __init__(self, input_size: int, hidden_size: int):
        super().__init__()
        self.hidden_size = hidden_size

        # Time constant network — ключова інновація
        self.tau_net = nn.Sequential(
            nn.Linear(input_size + hidden_size, hidden_size),
            nn.Sigmoid()  # τ ∈ (0, 1), scaled later
        )

        # Dynamics network
        self.f_net = nn.Sequential(
            nn.Linear(input_size + hidden_size, hidden_size),
            nn.Tanh()
        )

        # Leak term
        self.leak = nn.Parameter(torch.ones(hidden_size) * 0.1)

        self.tau_min = 0.1
        self.tau_max = 10.0

    def compute_tau(self, x: torch.Tensor, h: torch.Tensor) -> torch.Tensor:
        """Input-dependent time constant — це те, що робить мережу 'liquid'"""
        combined = torch.cat([x, h], dim=-1)
        tau_raw = self.tau_net(combined)
        # Scale to [tau_min, tau_max]
        return self.tau_min + (self.tau_max - self.tau_min) * tau_raw

    def dynamics(self, t: float, h: torch.Tensor, x: torch.Tensor) -> torch.Tensor:
        """ODE dynamics: dh/dt = -h/τ(x,h) + f(x,h)"""
        tau = self.compute_tau(x, h)
        f = self.f_net(torch.cat([x, h], dim=-1))

        # Liquid dynamics equation
        dhdt = -h / tau + f

        return dhdt

    def forward(self, x: torch.Tensor, h: torch.Tensor,
                dt: float = 1.0) -> torch.Tensor:
        """Forward pass з ODE integration"""

        def ode_func(t, state):
            return self.dynamics(t, state, x)

        # Integrate ODE
        t_span = torch.tensor([0.0, dt])
        h_new = odeint(ode_func, h, t_span, method='dopri5')[-1]

        return h_new


class LiquidNetwork(nn.Module):
    """Повна Liquid Neural Network"""

    def __init__(self, input_size: int, hidden_size: int, output_size: int,
                 num_layers: int = 1):
        super().__init__()

        self.layers = nn.ModuleList([
            LTCCell(input_size if i == 0 else hidden_size, hidden_size)
            for i in range(num_layers)
        ])

        self.output = nn.Linear(hidden_size, output_size)
        self.hidden_size = hidden_size
        self.num_layers = num_layers

    def forward(self, x_seq: torch.Tensor) -> torch.Tensor:
        """
        x_seq: (batch, seq_len, input_size)
        returns: (batch, seq_len, output_size)
        """
        batch_size, seq_len, _ = x_seq.shape
        device = x_seq.device

        # Initialize hidden states
        hiddens = [torch.zeros(batch_size, self.hidden_size, device=device)
                   for _ in range(self.num_layers)]

        outputs = []

        for t in range(seq_len):
            x_t = x_seq[:, t, :]

            # Forward through liquid layers
            h = x_t
            new_hiddens = []
            for i, (layer, hidden) in enumerate(zip(self.layers, hiddens)):
                h = layer(h if i == 0 else new_hiddens[-1], hidden)
                new_hiddens.append(h)

            hiddens = new_hiddens

            # Output
            out = self.output(hiddens[-1])
            outputs.append(out)

        return torch.stack(outputs, dim=1)

    def init_hidden(self, batch_size: int, device: torch.device):
        return [torch.zeros(batch_size, self.hidden_size, device=device)
                for _ in range(self.num_layers)]

Чому τ(x) — це game changer:

У традиційній RNN динаміка однакова для всіх inputs. У LTC:

• Швидко змінюваний input → маленький τ → швидка реакція
• Стабільний input → великий τ → плавне інтегрування
• Мережа автоматично адаптує свій "темп" до даних

Closed-form Continuous-time (CfC) Networks

Проблема LTC: ODE solver повільний. Кожен forward pass потребує численної інтеграції.

Рішення MIT (2022): Аналітичне closed-form рішення ODE.

class CfCCell(nn.Module):
    """Closed-form Continuous-time Cell — швидка версія LTC"""

    def __init__(self, input_size: int, hidden_size: int):
        super().__init__()
        self.hidden_size = hidden_size

        # Backbone networks
        self.backbone = nn.Linear(input_size + hidden_size, hidden_size)

        # Time constant parameters
        self.ff1 = nn.Linear(input_size + hidden_size, hidden_size)
        self.ff2 = nn.Linear(hidden_size, hidden_size)

        # Tau network
        self.tau_system = nn.Linear(input_size + hidden_size, hidden_size)

    def forward(self, x: torch.Tensor, h: torch.Tensor,
                t_delta: float = 1.0) -> torch.Tensor:
        """
        Closed-form solution — без ODE solver!

        Замість численної інтеграції використовуємо аналітичну формулу:
        h(t+Δt) = h(t)·exp(-Δt/τ) + (1-exp(-Δt/τ))·f(x,h)
        """
        combined = torch.cat([x, h], dim=-1)

        # Compute time constant
        tau = torch.sigmoid(self.tau_system(combined)) + 0.01  # Avoid division by zero

        # Compute target state
        ff_out = torch.tanh(self.ff1(combined))
        f = self.ff2(ff_out)

        # Closed-form update
        decay = torch.exp(-t_delta / tau)
        h_new = h * decay + (1 - decay) * f

        return h_new


class CfCNetwork(nn.Module):
    """Closed-form Continuous-time Network"""

    def __init__(self, input_size: int, hidden_size: int, output_size: int,
                 backbone_layers: int = 2, backbone_units: int = 64):
        super().__init__()

        # Optional backbone for feature extraction
        backbone = []
        in_features = input_size
        for _ in range(backbone_layers):
            backbone.extend([
                nn.Linear(in_features, backbone_units),
                nn.ReLU()
            ])
            in_features = backbone_units

        self.backbone = nn.Sequential(*backbone) if backbone else nn.Identity()

        # CfC cell
        self.cfc = CfCCell(in_features, hidden_size)

        # Output
        self.output = nn.Linear(hidden_size, output_size)
        self.hidden_size = hidden_size

    def forward(self, x_seq: torch.Tensor,
                time_deltas: torch.Tensor = None) -> torch.Tensor:
        """
        x_seq: (batch, seq_len, input_size)
        time_deltas: (batch, seq_len) - irregular sampling support!
        """
        batch_size, seq_len, _ = x_seq.shape
        device = x_seq.device

        if time_deltas is None:
            time_deltas = torch.ones(batch_size, seq_len, device=device)

        h = torch.zeros(batch_size, self.hidden_size, device=device)
        outputs = []

        for t in range(seq_len):
            x_t = self.backbone(x_seq[:, t, :])
            dt = time_deltas[:, t].unsqueeze(-1)

            h = self.cfc(x_t, h, dt)
            out = self.output(h)
            outputs.append(out)

        return torch.stack(outputs, dim=1)

Переваги CfC над LTC:

• 10-50x швидше: Немає ODE solver
• Irregular sampling: Природно підтримує нерегулярні часові ряди
• Differentiable: Стандартний backpropagation
• Зберігає liquid properties: Адаптивна динаміка

Neural Circuit Policies (NCP)

Wiring patterns — біологічно натхненні зв'язки:

class AutoNCP:
    """Automatic Neural Circuit Policy wiring"""

    def __init__(self, units: int, output_size: int, sparsity: float = 0.5):
        self.units = units
        self.output_size = output_size
        self.sparsity = sparsity

        # Neuron types (біологічно натхненні)
        self.sensory = int(0.2 * units)      # Сенсорні нейрони
        self.inter = int(0.4 * units)         # Interneurons
        self.command = int(0.2 * units)       # Command neurons
        self.motor = output_size              # Motor neurons

    def build_connectivity(self) -> torch.Tensor:
        """Будує матрицю зв'язності"""
        total = self.sensory + self.inter + self.command + self.motor

        # Ініціалізуємо sparse connectivity
        mask = torch.zeros(total, total)

        # Sensory → Inter (dense)
        mask[:self.sensory, self.sensory:self.sensory+self.inter] = 1.0

        # Inter ↔ Inter (sparse, recurrent)
        inter_start = self.sensory
        inter_end = self.sensory + self.inter
        inter_mask = torch.rand(self.inter, self.inter) < (1 - self.sparsity)
        mask[inter_start:inter_end, inter_start:inter_end] = inter_mask.float()

        # Inter → Command
        cmd_start = inter_end
        cmd_end = cmd_start + self.command
        mask[inter_start:inter_end, cmd_start:cmd_end] = 1.0

        # Command → Motor
        motor_start = cmd_end
        mask[cmd_start:cmd_end, motor_start:] = 1.0

        return mask


class NCPNetwork(nn.Module):
    """Neural Circuit Policy Network"""

    def __init__(self, input_size: int, wiring: AutoNCP):
        super().__init__()
        self.wiring = wiring

        total_units = (wiring.sensory + wiring.inter +
                       wiring.command + wiring.motor)

        # Sparse weight matrix
        self.connectivity = wiring.build_connectivity()
        self.weights = nn.Parameter(torch.randn(total_units, total_units) * 0.1)

        # Input projection
        self.input_proj = nn.Linear(input_size, wiring.sensory)

        # CfC cells for each neuron type
        self.sensory_cells = CfCCell(wiring.sensory, wiring.sensory)
        self.inter_cells = CfCCell(wiring.sensory, wiring.inter)
        self.command_cells = CfCCell(wiring.inter, wiring.command)
        self.motor_cells = CfCCell(wiring.command, wiring.motor)

    def forward(self, x: torch.Tensor, states: dict = None) -> tuple:
        """Forward pass через NCP"""
        if states is None:
            states = self.init_states(x.shape[0], x.device)

        # Sensory processing
        sensory_in = self.input_proj(x)
        states['sensory'] = self.sensory_cells(sensory_in, states['sensory'])

        # Inter processing (with recurrence)
        states['inter'] = self.inter_cells(states['sensory'], states['inter'])

        # Command neurons
        states['command'] = self.command_cells(states['inter'], states['command'])

        # Motor output
        states['motor'] = self.motor_cells(states['command'], states['motor'])

        return states['motor'], states

    def init_states(self, batch_size: int, device: torch.device) -> dict:
        return {
            'sensory': torch.zeros(batch_size, self.wiring.sensory, device=device),
            'inter': torch.zeros(batch_size, self.wiring.inter, device=device),
            'command': torch.zeros(batch_size, self.wiring.command, device=device),
            'motor': torch.zeros(batch_size, self.wiring.motor, device=device)
        }

Приклад: Автономне керування з 19 нейронами

MIT продемонстрував: Liquid network з 19 нейронами керує автомобілем.

class AutonomousDrivingLiquid(nn.Module):
    """19-neuron autonomous driving controller (MIT demo)"""

    def __init__(self):
        super().__init__()

        # Image encoder (pre-trained, frozen)
        self.encoder = nn.Sequential(
            nn.Conv2d(3, 16, 5, stride=2),
            nn.ReLU(),
            nn.Conv2d(16, 32, 3, stride=2),
            nn.ReLU(),
            nn.Flatten(),
            nn.Linear(32 * 14 * 14, 128),
            nn.ReLU(),
            nn.Linear(128, 32)
        )

        # Tiny liquid brain — всього 19 нейронів!
        self.wiring = AutoNCP(units=19, output_size=2)  # steering, throttle
        self.liquid_brain = NCPNetwork(32, self.wiring)

    def forward(self, image: torch.Tensor,
                states: dict = None) -> tuple[torch.Tensor, dict]:
        """
        image: (batch, 3, H, W) - camera input
        returns: (steering, throttle), new_states
        """
        # Encode image
        features = self.encoder(image)

        # Liquid processing
        control, new_states = self.liquid_brain(features, states)

        # Separate steering and throttle
        steering = torch.tanh(control[:, 0])     # [-1, 1]
        throttle = torch.sigmoid(control[:, 1])  # [0, 1]

        return torch.stack([steering, throttle], dim=1), new_states


# Чому це працює з 19 нейронами:
# 1. Liquid dynamics — складна поведінка з простої структури
# 2. Input-dependent tau — automatic attention
# 3. Continuous-time — природне temporal processing
# 4. Sparse connectivity — ефективність

Deployment на Edge Devices

Оптимізація для мікроконтролерів:

import numpy as np

class CfCMicrocontroller:
    """
    Pure NumPy implementation для deployment на мікроконтролер
    Без PyTorch, без CUDA — тільки math operations
    """

    def __init__(self, weights: dict):
        # Load pre-trained weights
        self.W_tau = weights['tau_weight']    # (hidden, input+hidden)
        self.b_tau = weights['tau_bias']      # (hidden,)
        self.W_f = weights['f_weight']        # (hidden, input+hidden)
        self.b_f = weights['f_bias']          # (hidden,)
        self.W_out = weights['out_weight']    # (output, hidden)
        self.b_out = weights['out_bias']      # (output,)

        self.hidden_size = self.W_f.shape[0]

    def sigmoid(self, x: np.ndarray) -> np.ndarray:
        return 1.0 / (1.0 + np.exp(-np.clip(x, -500, 500)))

    def tanh(self, x: np.ndarray) -> np.ndarray:
        return np.tanh(x)

    def forward(self, x: np.ndarray, h: np.ndarray,
                dt: float = 1.0) -> tuple[np.ndarray, np.ndarray]:
        """
        Single step forward pass
        Optimized for minimal memory and compute
        """
        # Concatenate input and hidden
        combined = np.concatenate([x, h])

        # Compute time constant
        tau = self.sigmoid(self.W_tau @ combined + self.b_tau) + 0.01

        # Compute target
        f = self.tanh(self.W_f @ combined + self.b_f)

        # Closed-form update
        decay = np.exp(-dt / tau)
        h_new = h * decay + (1 - decay) * f

        # Output
        out = self.W_out @ h_new + self.b_out

        return out, h_new


class EdgeDeploymentOptimizer:
    """Оптимізації для edge deployment"""

    @staticmethod
    def quantize_weights(weights: dict, bits: int = 8) -> dict:
        """Quantization для зменшення пам'яті"""
        quantized = {}
        for name, w in weights.items():
            w_min, w_max = w.min(), w.max()
            scale = (w_max - w_min) / (2**bits - 1)

            w_quant = np.round((w - w_min) / scale).astype(np.uint8)
            quantized[name] = {
                'data': w_quant,
                'scale': scale,
                'min': w_min
            }
        return quantized

    @staticmethod
    def estimate_memory(model: CfCMicrocontroller) -> dict:
        """Оцінка використання пам'яті"""
        total_params = sum(w.size for w in [
            model.W_tau, model.b_tau,
            model.W_f, model.b_f,
            model.W_out, model.b_out
        ])

        return {
            'params_float32': total_params * 4,  # bytes
            'params_int8': total_params,          # bytes
            'hidden_state': model.hidden_size * 4,
            'total_float32': total_params * 4 + model.hidden_size * 4,
            'total_int8': total_params + model.hidden_size * 4
        }


# Приклад: 19-neuron controller
# Params: ~1000 weights
# Memory: ~4KB float32, ~1KB int8
# Fits on: Arduino, ESP32, Raspberry Pi Pico

Порівняння архітектур

Benchmark на time series задачах:

| Архітектура | Params | Energy | Accuracy | Latency |

|-------------|--------|--------|----------|---------|

| LSTM | 100K | 100% | 94.2% | 10ms |

| GRU | 75K | 80% | 93.8% | 8ms |

| Transformer | 500K | 500% | 95.1% | 50ms |

| LTC | 1K | 1% | 92.5% | 5ms |

| CfC | 1K | 1% | 93.1% | 2ms |

Інтерпретація:

• CfC досягає comparable accuracy з 100x меншими params
• Energy efficiency критична для edge
• Latency важлива для real-time control

Irregular Time Series — native support

Перевага liquid networks: Природна підтримка нерегулярних часових рядів.

class IrregularTimeSeriesHandler:
    """Обробка даних з нерегулярною дискретизацією"""

    def __init__(self, cfc_model: CfCNetwork):
        self.model = cfc_model

    def process_irregular(self,
                          values: list[float],
                          timestamps: list[float]) -> torch.Tensor:
        """
        Обробка даних з довільними часовими інтервалами

        Приклад: медичні дані, де вимірювання нерегулярні
        """
        # Обчислюємо time deltas
        time_deltas = []
        for i in range(len(timestamps)):
            if i == 0:
                time_deltas.append(1.0)
            else:
                time_deltas.append(timestamps[i] - timestamps[i-1])

        # Конвертуємо в тензори
        x = torch.tensor(values).unsqueeze(0).unsqueeze(-1)  # (1, seq, 1)
        dt = torch.tensor(time_deltas).unsqueeze(0)          # (1, seq)

        # CfC автоматично адаптується до різних dt
        return self.model(x, time_deltas=dt)


# Приклад: моніторинг серцевого ритму
# Вимірювання кожні 1-5 секунд (нерегулярно)
# LSTM: потребує resampling, втрата інформації
# CfC: обробляє as-is, використовує timing

Ідеї для дослідження

Для бакалаврської роботи:

• Порівняння LTC/CfC vs LSTM на публічних time series datasets
• Deployment liquid network на Raspberry Pi для gesture recognition
• Візуалізація liquid dynamics для interpretability

Для магістерської:

• LTC для specific edge application: drone navigation, wearable health monitoring
• Optimization for specific microcontroller (ESP32, STM32)
• Hybrid architectures: Transformer encoder + Liquid decoder
• CfC для industrial anomaly detection

Для PhD:

• Theoretical analysis: expressiveness та approximation guarantees
• Novel wiring architectures натхненні іншими організмами
• Online learning в liquid networks (continual adaptation)
• Formal verification of liquid network controllers

Чому це важливо для майбутнього Edge AI

Privacy. Latency. Reliability. Cost. Чотири причини, чому edge AI — необхідність:

1. Privacy: Дані не покидають пристрій. GDPR compliance by design.
2. Latency: Немає round-trip до cloud. Мілісекундні рішення.
3. Reliability: Працює offline. Без залежності від мережі.
4. Cost: Немає cloud inference costs. Pay once.

Liquid networks роблять edge AI реальним для пристроїв, де звичайні моделі неможливі. Не "потужна модель, яку складно запустити", а "компактна модель, яка працює всюди і адаптується до змін".

Для тих, хто готує наукову роботу з liquid neural networks або edge AI — від курсової до дисертації — команда SKP-Degree на skp-degree.com.ua має досвід супроводу проєктів з впровадження compact neural networks. Пишіть у Telegram: @kursovi_diplomy для консультації.

Ключові слова: liquid neural networks, LTC, CfC, closed-form continuous-time, edge AI, IoT, neural ODE, MIT, tinyML, енергоефективність, time series, мікроконтролер, наукова робота, дипломна, курсова.