bench: add benchmark runners, results docs, and test suite
Benchmarks: LongMemEval, LoCoMo, ConvoMem, MemBench runners with methodology docs and hybrid retrieval analysis. Tests: config, miner, convo_miner, normalize — 9 tests, all passing.
This commit is contained in:
@@ -0,0 +1,32 @@
|
||||
import os
|
||||
import json
|
||||
import tempfile
|
||||
from mempalace.config import MempalaceConfig
|
||||
|
||||
|
||||
def test_default_config():
|
||||
cfg = MempalaceConfig(config_dir=tempfile.mkdtemp())
|
||||
assert "palace" in cfg.palace_path
|
||||
assert cfg.collection_name == "mempalace_drawers"
|
||||
|
||||
|
||||
def test_config_from_file():
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
with open(os.path.join(tmpdir, "config.json"), "w") as f:
|
||||
json.dump({"palace_path": "/custom/palace"}, f)
|
||||
cfg = MempalaceConfig(config_dir=tmpdir)
|
||||
assert cfg.palace_path == "/custom/palace"
|
||||
|
||||
|
||||
def test_env_override():
|
||||
os.environ["MEMPALACE_PALACE_PATH"] = "/env/palace"
|
||||
cfg = MempalaceConfig(config_dir=tempfile.mkdtemp())
|
||||
assert cfg.palace_path == "/env/palace"
|
||||
del os.environ["MEMPALACE_PALACE_PATH"]
|
||||
|
||||
|
||||
def test_init():
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
cfg = MempalaceConfig(config_dir=tmpdir)
|
||||
cfg.init()
|
||||
assert os.path.exists(os.path.join(tmpdir, "config.json"))
|
||||
@@ -0,0 +1,26 @@
|
||||
import os
|
||||
import tempfile
|
||||
import shutil
|
||||
import chromadb
|
||||
from mempalace.convo_miner import mine_convos
|
||||
|
||||
|
||||
def test_convo_mining():
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
with open(os.path.join(tmpdir, "chat.txt"), "w") as f:
|
||||
f.write(
|
||||
"> What is memory?\nMemory is persistence.\n\n> Why does it matter?\nIt enables continuity.\n\n> How do we build it?\nWith structured storage.\n"
|
||||
)
|
||||
|
||||
palace_path = os.path.join(tmpdir, "palace")
|
||||
mine_convos(tmpdir, palace_path, wing="test_convos")
|
||||
|
||||
client = chromadb.PersistentClient(path=palace_path)
|
||||
col = client.get_collection("mempalace_drawers")
|
||||
assert col.count() >= 2
|
||||
|
||||
# Verify search works
|
||||
results = col.query(query_texts=["memory persistence"], n_results=1)
|
||||
assert len(results["documents"][0]) > 0
|
||||
|
||||
shutil.rmtree(tmpdir)
|
||||
@@ -0,0 +1,36 @@
|
||||
import os
|
||||
import tempfile
|
||||
import shutil
|
||||
import yaml
|
||||
import chromadb
|
||||
from mempalace.miner import mine
|
||||
|
||||
|
||||
def test_project_mining():
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
# Create a mini project
|
||||
os.makedirs(os.path.join(tmpdir, "backend"))
|
||||
with open(os.path.join(tmpdir, "backend", "app.py"), "w") as f:
|
||||
f.write("def main():\n print('hello world')\n" * 20)
|
||||
# Create config
|
||||
with open(os.path.join(tmpdir, "mempalace.yaml"), "w") as f:
|
||||
yaml.dump(
|
||||
{
|
||||
"wing": "test_project",
|
||||
"rooms": [
|
||||
{"name": "backend", "description": "Backend code"},
|
||||
{"name": "general", "description": "General"},
|
||||
],
|
||||
},
|
||||
f,
|
||||
)
|
||||
|
||||
palace_path = os.path.join(tmpdir, "palace")
|
||||
mine(tmpdir, palace_path)
|
||||
|
||||
# Verify
|
||||
client = chromadb.PersistentClient(path=palace_path)
|
||||
col = client.get_collection("mempalace_drawers")
|
||||
assert col.count() > 0
|
||||
|
||||
shutil.rmtree(tmpdir)
|
||||
@@ -0,0 +1,31 @@
|
||||
import os
|
||||
import json
|
||||
import tempfile
|
||||
from mempalace.normalize import normalize
|
||||
|
||||
|
||||
def test_plain_text():
|
||||
f = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False)
|
||||
f.write("Hello world\nSecond line\n")
|
||||
f.close()
|
||||
result = normalize(f.name)
|
||||
assert "Hello world" in result
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
def test_claude_json():
|
||||
data = [{"role": "user", "content": "Hi"}, {"role": "assistant", "content": "Hello"}]
|
||||
f = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False)
|
||||
json.dump(data, f)
|
||||
f.close()
|
||||
result = normalize(f.name)
|
||||
assert "Hi" in result
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
def test_empty():
|
||||
f = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False)
|
||||
f.close()
|
||||
result = normalize(f.name)
|
||||
assert result.strip() == ""
|
||||
os.unlink(f.name)
|
||||
Reference in New Issue
Block a user