Spaces:
Sleeping
Sleeping
File size: 4,470 Bytes
6b4f07b aab1b41 6b4f07b aab1b41 43d613a aab1b41 43d613a aab1b41 6b4f07b f09a362 c0072b1 f09a362 c0072b1 6b4f07b c338c4c 6b4f07b c338c4c 6b4f07b c338c4c | 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 | import logging
from lilith_agent.config import Config
from lilith_agent.app import build_react_agent
from langchain_core.messages import HumanMessage
def test_build_react_agent_uses_sqlite_saver(tmp_path, monkeypatch):
class FakeModel:
def bind_tools(self, tools):
return self
cfg = Config.from_env()
monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith"))
monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, **kw: FakeModel())
monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: FakeModel())
monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [])
agent = build_react_agent(cfg)
assert agent.checkpointer is not None
assert type(agent.checkpointer).__name__ == "SqliteSaver"
# Check if DB file was created
db_path = tmp_path / ".lilith" / "threads.sqlite"
assert db_path.exists()
def test_summarize_episode_stores_list_block_content_as_text(tmp_path, monkeypatch):
from lilith_agent import memory
class FakeModel:
def invoke(self, prompt):
class Response:
content = [
{"type": "text", "text": "Captured lesson"},
{"type": "non_text", "value": "ignored"},
]
return Response()
store = memory.MemoryStore(tmp_path / "long_term_memory.sqlite")
monkeypatch.setattr(memory, "_store", store)
memory.summarize_episode(
[HumanMessage(content=[{"type": "text", "text": "Remember this"}])],
FakeModel(),
)
episodes = store.get_recent_episodes()
assert episodes[0]["task"] == "Remember this"
assert episodes[0]["summary"] == "Captured lesson"
def test_summarize_episode_prints_saved_episode_details(tmp_path, monkeypatch, caplog, capsys):
from lilith_agent import memory
class FakeModel:
def invoke(self, prompt):
class Response:
content = "Used read_file successfully and learned to inspect spreadsheet rows."
return Response()
store = memory.MemoryStore(tmp_path / "long_term_memory.sqlite")
monkeypatch.setattr(memory, "_store", store)
with caplog.at_level(logging.INFO, logger="lilith_agent.memory"):
memory.summarize_episode([HumanMessage(content="Find oldest Blu-Ray title")], FakeModel())
printed = capsys.readouterr().out
assert "[memory] Episode saved: task='Find oldest Blu-Ray title'" in printed
assert "outcome='success'" in printed
assert "summary='Used read_file successfully" in printed
record = next(r for r in caplog.records if "[memory] Episode saved:" in r.message)
assert "task='Find oldest Blu-Ray title'" in record.message
assert "outcome='success'" in record.message
assert "summary='Used read_file successfully" in record.message
def test_summarize_episode_logs_traceback_on_failure(tmp_path, monkeypatch, caplog):
from lilith_agent import memory
class BrokenModel:
def invoke(self, prompt):
raise RuntimeError("boom")
store = memory.MemoryStore(tmp_path / "long_term_memory.sqlite")
monkeypatch.setattr(memory, "_store", store)
with caplog.at_level(logging.ERROR, logger="lilith_agent.memory"):
memory.summarize_episode([HumanMessage(content="Remember this")], BrokenModel())
record = next(r for r in caplog.records if "Summarization failed" in r.message)
assert record.exc_info is not None
def test_extract_and_compress_facts_passes_existing_memories_with_ids(tmp_path, monkeypatch):
from lilith_agent import memory
import langmem
captured = {}
class FakeManager:
def invoke(self, payload):
captured["existing"] = payload["existing"]
return []
store = memory.MemoryStore(tmp_path / "long_term_memory.sqlite")
store.save_memories([{"id": "memory-1", "content": "Existing fact"}])
monkeypatch.setattr(memory, "_store", store)
monkeypatch.setattr(langmem, "create_memory_manager", lambda model, enable_deletes: FakeManager())
class FakeModel:
def invoke(self, prompt):
class Response:
content = "Lesson"
return Response()
memory.extract_and_compress_facts([HumanMessage(content="New fact")], FakeModel())
existing_id, existing_memory = captured["existing"][0]
assert existing_id == "memory-1"
assert existing_memory.content == "Existing fact"
|