fa71f973a6
Some checks failed
- New fusionagi/gpu/ module with TensorBackend protocol abstraction - TensorFlowBackend: GPU-accelerated ops with TensorCore mixed-precision - NumPyBackend: CPU fallback (always available, no extra deps) - Auto-selects best available backend at runtime - GPU-accelerated operations: - Cosine similarity matrix (batched, XLA-compiled) - Multi-head attention for consensus scoring - Batch hypothesis scoring on GPU - Semantic similarity search (pairwise, nearest-neighbor, deduplication) - New TensorFlowAdapter (fusionagi/adapters/): - LLMAdapter for local TF/Keras model inference - TensorCore mixed-precision support - GPU-accelerated embedding synthesis fallback - Reasoning pipeline integration: - gpu_scoring.py: drop-in GPU replacement for multi_path scoring - Super Big Brain: use_gpu config flag, GPU scoring when available - Memory integration: - gpu_search.py: GPU-accelerated semantic search for SemanticGraphMemory - Self-improvement integration: - gpu_training.py: gradient-based heuristic weight optimization - Reflective memory training loop with loss tracking - Dependencies: gpu extra (tensorflow>=2.16, numpy>=1.26) - 64 new tests (276 total), all passing - Architecture spec: docs/gpu_tensorcore_integration.md Co-Authored-By: Nakamoto, S <defi@defi-oracle.io>
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""Tests for fusionagi.adapters.tensorflow_adapter (uses NumPy backend, no TF required)."""
|
|
|
|
import pytest
|
|
|
|
from fusionagi.gpu.backend import reset_backend, get_backend
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _use_numpy():
|
|
reset_backend()
|
|
get_backend(force="numpy")
|
|
yield
|
|
reset_backend()
|
|
|
|
|
|
class TestTensorFlowAdapterImport:
|
|
"""Test that TensorFlowAdapter is importable (may be None without TF)."""
|
|
|
|
def test_import(self):
|
|
from fusionagi.adapters import TensorFlowAdapter
|
|
# TensorFlowAdapter is None when tensorflow is not installed
|
|
# This is by design — GPU is an optional dependency
|
|
|
|
|
|
class TestGPUMemorySearch:
|
|
"""Test GPU-accelerated memory search."""
|
|
|
|
def test_semantic_search(self):
|
|
from fusionagi.memory.gpu_search import semantic_search
|
|
from fusionagi.schemas.atomic import AtomicSemanticUnit, AtomicUnitType
|
|
|
|
units = [
|
|
AtomicSemanticUnit(
|
|
unit_id="u1",
|
|
content="the sky is blue",
|
|
type=AtomicUnitType.FACT,
|
|
confidence=1.0,
|
|
),
|
|
AtomicSemanticUnit(
|
|
unit_id="u2",
|
|
content="water is wet",
|
|
type=AtomicUnitType.FACT,
|
|
confidence=1.0,
|
|
),
|
|
AtomicSemanticUnit(
|
|
unit_id="u3",
|
|
content="python programming language",
|
|
type=AtomicUnitType.FACT,
|
|
confidence=1.0,
|
|
),
|
|
]
|
|
results = semantic_search("sky color", units, top_k=2)
|
|
assert len(results) <= 2
|
|
assert all(isinstance(r, tuple) for r in results)
|
|
assert all(isinstance(r[0], AtomicSemanticUnit) for r in results)
|
|
assert all(isinstance(r[1], float) for r in results)
|
|
|
|
def test_semantic_search_empty(self):
|
|
from fusionagi.memory.gpu_search import semantic_search
|
|
|
|
results = semantic_search("query", [], top_k=5)
|
|
assert results == []
|
|
|
|
def test_batch_embed_units(self):
|
|
from fusionagi.memory.gpu_search import batch_embed_units
|
|
from fusionagi.schemas.atomic import AtomicSemanticUnit, AtomicUnitType
|
|
|
|
units = [
|
|
AtomicSemanticUnit(
|
|
unit_id="u1",
|
|
content="test content",
|
|
type=AtomicUnitType.FACT,
|
|
confidence=1.0,
|
|
),
|
|
]
|
|
result = batch_embed_units(units)
|
|
assert result is not None
|