64b800c6cf
Items completed: 1. Merged PR #2 (starlette/httpx deps) 2. Fixed async race condition in multimodal_ui.py 3. Wired TTSAdapter (ElevenLabs, Azure) in API routes 4. Moved super_big_brain.py from core/ to reasoning/ (backward compat shim) 5. Added API authentication middleware (Bearer token via FUSIONAGI_API_KEY) 6. Added async adapter interface (acomplete/acomplete_structured) 7. Migrated FastAPI on_event to lifespan (fixes 20 deprecation warnings) 8. Liquid Neural Networks (continuous-time adaptive weights) 9. Quantum-AI Hybrid compute backend (simulator + optimization) 10. Embodied Intelligence / Robotics bridge (actuator + sensor protocols) 11. Consciousness Engineering (formal self-model with introspection) 12. ASI Scoring Rubric (C/A/L/N/R self-assessment harness) 13. GPU integration tests for TensorFlow backend 14. Multi-stage production Dockerfile 15. Gitea CI/CD pipeline (lint, test matrix, Docker build) 16. API rate limiting middleware (per-IP sliding window) 17. OpenAPI docs cleanup (auth + rate limiting descriptions) 18. Benchmarking suite (decomposition, multi-path, recomposition, e2e) 19. Plugin system (head registry for custom heads) 427 tests passing, 0 ruff errors, 0 mypy errors. Co-Authored-By: Nakamoto, S <defi@defi-oracle.io>
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""Tests for Liquid Neural Networks module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fusionagi.reasoning.liquid_networks import LiquidCell, LiquidNetwork, LiquidNetworkConfig
|
|
|
|
|
|
class TestLiquidCell:
|
|
def test_init_defaults(self) -> None:
|
|
cell = LiquidCell(input_dim=4, hidden_dim=3)
|
|
assert len(cell.w_in) == 3
|
|
assert len(cell.w_in[0]) == 4
|
|
assert len(cell.state) == 3
|
|
|
|
def test_step_changes_state(self) -> None:
|
|
cell = LiquidCell(input_dim=2, hidden_dim=2)
|
|
initial = list(cell.state)
|
|
cell.step([1.0, 0.5])
|
|
assert cell.state != initial
|
|
|
|
def test_reset_zeros_state(self) -> None:
|
|
cell = LiquidCell(input_dim=2, hidden_dim=2)
|
|
cell.step([1.0, 0.5])
|
|
cell.reset()
|
|
assert all(s == 0.0 for s in cell.state)
|
|
|
|
def test_multiple_steps_evolve(self) -> None:
|
|
cell = LiquidCell(input_dim=3, hidden_dim=4)
|
|
states = []
|
|
for _ in range(5):
|
|
states.append(list(cell.step([0.5, -0.3, 0.8])))
|
|
assert states[0] != states[4]
|
|
|
|
|
|
class TestLiquidNetwork:
|
|
def test_init_default_config(self) -> None:
|
|
net = LiquidNetwork()
|
|
assert net.config.input_dim == 64
|
|
|
|
def test_forward_output_shape(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=8, hidden_dim=4, output_dim=3, num_layers=1)
|
|
net = LiquidNetwork(cfg)
|
|
out = net.forward([1.0] * 8)
|
|
assert len(out) == 3
|
|
|
|
def test_forward_padding(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=8, hidden_dim=4, output_dim=2)
|
|
net = LiquidNetwork(cfg)
|
|
out = net.forward([1.0, 2.0]) # Shorter than input_dim
|
|
assert len(out) == 2
|
|
|
|
def test_forward_truncation(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=4, hidden_dim=2, output_dim=2)
|
|
net = LiquidNetwork(cfg)
|
|
out = net.forward([1.0] * 10) # Longer than input_dim
|
|
assert len(out) == 2
|
|
|
|
def test_forward_sequence(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=4, hidden_dim=3, output_dim=2, num_layers=1)
|
|
net = LiquidNetwork(cfg)
|
|
inputs = [[float(i)] * 4 for i in range(5)]
|
|
outputs = net.forward_sequence(inputs)
|
|
assert len(outputs) == 5
|
|
assert all(len(o) == 2 for o in outputs)
|
|
|
|
def test_reset_clears_state(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=4, hidden_dim=3, output_dim=2)
|
|
net = LiquidNetwork(cfg)
|
|
net.forward([1.0] * 4)
|
|
net.reset()
|
|
for layer in net._layers:
|
|
assert all(s == 0.0 for s in layer.state)
|
|
|
|
def test_adapt_weights(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=4, hidden_dim=3, output_dim=2, num_layers=1)
|
|
net = LiquidNetwork(cfg)
|
|
inputs = [[0.1, 0.2, 0.3, 0.4], [0.5, 0.6, 0.7, 0.8]]
|
|
targets = [[0.5, -0.5], [0.3, 0.3]]
|
|
result = net.adapt_weights(inputs, targets, epochs=5)
|
|
assert "final_loss" in result
|
|
assert result["epochs_run"] <= 5
|
|
|
|
def test_get_summary(self) -> None:
|
|
net = LiquidNetwork()
|
|
summary = net.get_summary()
|
|
assert summary["type"] == "LiquidNetwork"
|
|
assert "total_parameters" in summary
|
|
|
|
def test_output_bounded(self) -> None:
|
|
cfg = LiquidNetworkConfig(input_dim=4, hidden_dim=4, output_dim=3)
|
|
net = LiquidNetwork(cfg)
|
|
out = net.forward([10.0, -10.0, 5.0, -5.0])
|
|
for val in out:
|
|
assert -1.0 <= val <= 1.0
|