"""Tests for the Cortex layer: expertise, dreaming, compositional reasoning.""" import pytest import numpy as np from palimseste.lm import PalimpsesteForCausalLM, PalimpsesteConfig from palimseste.chat import Conversation from palimseste.reasoning import Reasoner from palimseste.cortex import ( InstantExpert, Dreamer, Composer, ExpertiseResult, DreamResult, CompositionResult, ) def _build_model(D=5000, ctx=128, radius=200): cfg = PalimpsesteConfig(D=D, context_window=ctx, kernel_radius=radius, temperature=0.0) lm = PalimpsesteForCausalLM(config=cfg) pairs = [ ("hello", "hi i am palimpseste"), ("who are you", "i am palimpseste a hypervectorial cortex"), ("what is python", "python is a programming language"), ("who won the world cup 2018", "france"), ("what is the capital of france", "paris"), ("what is the capital of japan", "tokyo"), ] lm.build_tokenizer("".join(q + a for q, a in pairs)) lm.train_on_qa_pairs(pairs) return lm, pairs # ================================================================ EXPERTISE class TestInstantExpertise: def test_learn_from_text(self): lm, _ = _build_model() expert = InstantExpert(lm=lm) doc = "Quantum computing is a type of computation. A qubit is the basic unit of quantum information." result = expert.learn_from_text(doc) assert result.n_tokens > 0 assert isinstance(result, ExpertiseResult) def test_facts_extracted(self): lm, _ = _build_model() expert = InstantExpert(lm=lm) doc = "Python is a programming language. A variable is a name for a value." result = expert.learn_from_text(doc) assert result.n_facts > 0 # Should extract "what is python" and "what is a variable" questions = [q for q, _ in result.facts] assert any("python" in q for q in questions) assert any("variable" in q for q in questions) def test_document_tag(self): lm, _ = _build_model() expert = InstantExpert(lm=lm) result = expert.learn_from_text("Test text.", document_tag="custom") assert result.document_tag == "custom" def test_grows_memory(self): lm, _ = _build_model() expert = InstantExpert(lm=lm) n_before = len(lm.mem) expert.learn_from_text("Some new content that is interesting.") assert len(lm.mem) > n_before def test_n_documents(self): lm, _ = _build_model() expert = InstantExpert(lm=lm) assert expert.n_documents == 0 expert.learn_from_text("Document one.") assert expert.n_documents == 1 expert.learn_from_text("Document two.") assert expert.n_documents == 2 def test_sentence_splitting(self): lm, _ = _build_model() expert = InstantExpert(lm=lm) sents = expert._split_sentences("Hello world. This is a test! Is it working?") assert len(sents) == 3 # ================================================================ DREAM class TestDreamer: def test_dream_returns_result(self): lm, _ = _build_model() dreamer = Dreamer(mem=lm.mem, phi=lm.phi) result = dreamer.dream(n_cycles=1, replay_batch=50) assert isinstance(result, DreamResult) assert result.n_cycles == 1 assert result.n_seconds >= 0 def test_dream_extracts_concepts(self): lm, _ = _build_model() dreamer = Dreamer(mem=lm.mem, phi=lm.phi) result = dreamer.dream(n_cycles=2, replay_batch=100) # Should extract some concepts from the memory assert result.n_concepts_extracted >= 0 def test_dream_multi_cycle(self): lm, _ = _build_model() dreamer = Dreamer(mem=lm.mem, phi=lm.phi) result = dreamer.dream(n_cycles=3, replay_batch=50) assert result.n_cycles == 3 def test_n_concepts_property(self): lm, _ = _build_model() dreamer = Dreamer(mem=lm.mem, phi=lm.phi) assert dreamer.n_concepts >= 0 dreamer.dream(n_cycles=1, replay_batch=50) assert dreamer.n_concepts >= 0 def test_empty_memory(self): cfg = PalimpsesteConfig(D=2000, context_window=64, kernel_radius=100, temperature=0.0) lm = PalimpsesteForCausalLM(config=cfg) lm.build_tokenizer("hello") # No training — memory is empty dreamer = Dreamer(mem=lm.mem) result = dreamer.dream(n_cycles=1, replay_batch=10) assert result.n_concepts_promoted == 0 # ================================================================ COMPOSER class TestComposer: def _build_composer(self): lm, pairs = _build_model() conv = Conversation(model=lm, fuzzy_threshold=0.75) conv.register_questions(pairs) reasoner = Reasoner(conv=conv) composer = Composer(reasoner=reasoner) return composer def test_simple_question(self): composer = self._build_composer() result = composer.reason("what is python") assert result.success assert "python" in result.answer.lower() or "language" in result.answer.lower() def test_chained_question(self): composer = self._build_composer() result = composer.reason( "what is the capital of the country that won the world cup 2018" ) # Should decompose and find: france -> paris assert result.success assert "paris" in result.answer.lower() def test_needs_decomposition(self): composer = self._build_composer() assert composer._needs_decomposition("what is the capital of the country that won") assert not composer._needs_decomposition("hello") def test_decompose(self): composer = self._build_composer() subs = composer._decompose( "what is the capital of the country that won the world cup 2018" ) assert len(subs) >= 1 def test_comparison(self): composer = self._build_composer() subs = composer._decompose("compare python and java") assert len(subs) == 2 def test_unknown_question(self): composer = self._build_composer() result = composer.reason("xyz123 unknown random") assert isinstance(result, CompositionResult) assert result.n_seconds >= 0 def test_steps_recorded(self): composer = self._build_composer() result = composer.reason( "what is the capital of the country that won the world cup 2018" ) assert len(result.steps) > 0 # Should have decompose and resolve steps types = [s.step_type for s in result.steps] assert "resolve" in types