File size: 6,777 Bytes
1f71c7d 22d1ad7 1f71c7d | 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 179 180 181 182 183 | """Tests for the Expert: dataset -> compressed .exp -> reload round-trip."""
from __future__ import annotations
import os
import tempfile
import numpy as np
import pytest
from ensemble import Expert
# Small shared corpora for fast tests (D kept small for speed).
# Made large enough that fixed overhead doesn't dominate compression.
MATH_TEXT = (
"what is two plus two. two plus two equals four. "
"what is three times three. three times three equals nine. "
"what is pi. pi is approximately three point one four. "
"what is ten minus four. ten minus four equals six. "
"what is the square root of nine. the square root of nine is three. "
) * 6
GEO_TEXT = (
"the capital of france is paris. "
"the capital of japan is tokyo. "
"the capital of italy is rome. "
"the capital of egypt is cairo. "
"the capital of brazil is brasilia. "
) * 6
MATH_QA = [
("what is two plus two", "two plus two equals four"),
("what is three times three", "three times three equals nine"),
("what is pi", "pi is approximately three point one four"),
("what is ten minus four", "ten minus four equals six"),
("what is the square root of nine", "the square root of nine is three"),
("what is five times five", "five times five equals twenty five"),
("what is one hundred divided by ten", "one hundred divided by ten equals ten"),
("what is eight plus seven", "eight plus seven equals fifteen"),
] * 2
class TestExpertBuild:
def test_from_text(self):
e = Expert.from_text(MATH_TEXT, domain="math", D=2000)
assert e.domain == "math"
assert e.D == 2000
assert e.n_traces > 0
assert e.vocab_size > 4 # specials + chars
def test_from_qa_pairs(self):
e = Expert.from_qa_pairs(MATH_QA, domain="qa", D=2000)
assert e.n_traces > 0
assert e.domain == "qa"
def test_from_text_records_source_stream(self):
e = Expert.from_text(MATH_TEXT, domain="t", D=2000)
assert e._source_token_stream is not None
assert len(e._source_token_stream) > 0
def test_qa_records_source_pairs(self):
e = Expert.from_qa_pairs(MATH_QA, domain="q", D=2000)
assert e._source_qa_pairs is not None
assert len(e._source_qa_pairs) == len(MATH_QA)
def test_signature_is_deterministic(self):
e1 = Expert.from_text(MATH_TEXT, domain="m", D=2000, seed=0)
e2 = Expert.from_text(MATH_TEXT, domain="m", D=2000, seed=0)
assert e1.signature_hv == e2.signature_hv
def test_different_seeds_give_different_self_hv(self):
e1 = Expert.from_text(MATH_TEXT, domain="m", D=2000, seed=1)
e2 = Expert.from_text(MATH_TEXT, domain="m", D=2000, seed=2)
assert e1.model._self_hv != e2.model._self_hv
class TestExpertCompression:
"""The defining property: .exp is SMALLER than the source dataset."""
def test_text_expert_smaller_than_source(self, tmp_path):
e = Expert.from_text(MATH_TEXT, domain="math", D=2000)
result = e.save(tmp_path / "math.exp")
assert result.expert_size_bytes < result.source_size_bytes
assert result.compression_ratio > 1.0
def test_qa_expert_smaller_than_source(self, tmp_path):
e = Expert.from_qa_pairs(MATH_QA, domain="qa", D=2000)
result = e.save(tmp_path / "qa.exp")
assert result.expert_size_bytes < result.source_size_bytes
def test_expert_directory_structure(self, tmp_path):
e = Expert.from_text(MATH_TEXT, domain="m", D=2000)
path = tmp_path / "m.exp"
result = e.save(path)
p = type(path)(result.path)
assert (p / "manifest.json").exists()
assert (p / "vocab.json").exists()
# lm mode has tokens.bin.gz, qa mode has qa_pairs.json.gz
assert (p / "tokens.bin.gz").exists()
class TestExpertRoundTrip:
"""A saved expert reloads to a bit-identical model."""
def test_text_round_trip(self, tmp_path):
e = Expert.from_text(MATH_TEXT, domain="math", D=2000, seed=5)
result = e.save(tmp_path / "math.exp")
e2 = Expert.load(result.path)
assert e2.D == e.D
assert e2.domain == e.domain
assert e2.n_traces == e.n_traces
assert e2.signature_hv == e.signature_hv
assert e2.model._self_hv == e.model._self_hv
def test_qa_round_trip(self, tmp_path):
e = Expert.from_qa_pairs(MATH_QA, domain="qa", D=2000, seed=3)
result = e.save(tmp_path / "qa.exp")
e2 = Expert.load(result.path)
assert e2.n_traces == e.n_traces
assert e2.signature_hv == e.signature_hv
def test_manifest_preserved(self, tmp_path):
e = Expert.from_text(MATH_TEXT, domain="science", D=2000)
result = e.save(tmp_path / "s.exp")
e2 = Expert.load(result.path)
assert e2.manifest.domain == "science"
assert e2.manifest.D == 2000
assert e2.manifest.compression_ratio > 1.0
class TestExpertQuery:
def test_answer_returns_string(self):
e = Expert.from_qa_pairs(MATH_QA, domain="m", D=5000)
a = e.answer("what is pi")
assert isinstance(a, str)
def test_relevance_returns_float(self):
e = Expert.from_qa_pairs(MATH_QA, domain="m", D=2000)
r = e.relevance("what is pi")
assert -1.0 <= r <= 1.0
def test_candidate_hv_returns_hv(self):
from palimseste.hv import HV
e = Expert.from_qa_pairs(MATH_QA, domain="m", D=2000)
c = e.candidate_hv("what is pi")
assert c is None or isinstance(c, HV)
class TestExpertDatasetFormats:
def test_csv_qa(self, tmp_path):
import csv as _csv
p = tmp_path / "data.csv"
with open(p, "w", newline="", encoding="utf-8") as f:
w = _csv.writer(f)
w.writerow(["question", "answer"])
for q, a in MATH_QA:
w.writerow([q, a])
e = Expert.from_dataset(p, D=2000)
assert e.n_traces > 0
def test_json_qa(self, tmp_path):
import json
p = tmp_path / "data.json"
with open(p, "w", encoding="utf-8") as f:
json.dump([{"question": q, "answer": a} for q, a in MATH_QA], f)
e = Expert.from_dataset(p, D=2000)
assert e.n_traces > 0
def test_jsonl(self, tmp_path):
import json
p = tmp_path / "data.jsonl"
with open(p, "w", encoding="utf-8") as f:
for q, a in MATH_QA:
f.write(json.dumps({"question": q, "answer": a}) + "\n")
e = Expert.from_dataset(p, D=2000)
assert e.n_traces > 0
def test_txt(self, tmp_path):
p = tmp_path / "data.txt"
p.write_text(MATH_TEXT, encoding="utf-8")
e = Expert.from_dataset(p, D=2000)
assert e.n_traces > 0
|