File size: 6,724 Bytes
bf174d6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
"""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