Add LLM long-term memory system
- Implement LLMMemory module with SQLite + FTS5 - Add conversation tracking with context awareness - Track NPC reputation and interactions - Integrate memory with LLMBrain for contextual replies - Add comprehensive test suite (13 tests) - Update settings with memory_db_path config
This commit is contained in:
truetype committed
1 parent
bd211b6036
commit
69ba824bc8
11 files changed
+979
-13
No files matched your search
@@ -0,0 +1,198 @@
|
||||
"""Tests for LLMMemory system"""
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
from src.modules.llm.memory import LLMMemory, MemoryEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def memory(tmp_path):
|
||||
"""Create a temporary memory instance for testing."""
|
||||
db_path = tmp_path / "test_memory.db"
|
||||
mem = LLMMemory(str(db_path))
|
||||
await mem.initialize()
|
||||
yield mem
|
||||
await mem.close()
|
||||
|
||||
|
||||
class TestLLMMemory:
|
||||
"""Test suite for LLMMemory."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initialization(self, tmp_path):
|
||||
"""Test that memory initializes correctly."""
|
||||
db_path = tmp_path / "test_init.db"
|
||||
mem = LLMMemory(str(db_path))
|
||||
await mem.initialize()
|
||||
assert db_path.exists()
|
||||
await mem.close()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_conversation(self, memory):
|
||||
"""Test adding conversations."""
|
||||
await memory.add_conversation(
|
||||
"Alice", "Hey there!", "Hello! How are you?",
|
||||
importance=0.7
|
||||
)
|
||||
|
||||
context = await memory.get_conversation_context("Alice", limit=5)
|
||||
assert len(context) == 1
|
||||
assert context[0]["message"] == "Hey there!"
|
||||
assert context[0]["reply"] == "Hello! How are you?"
|
||||
assert context[0]["importance"] == 0.7
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_conversations(self, memory):
|
||||
"""Test multiple conversations with same player."""
|
||||
await memory.add_conversation("Bob", "Msg 1", "Reply 1")
|
||||
await memory.add_conversation("Bob", "Msg 2", "Reply 2")
|
||||
await memory.add_conversation("Bob", "Msg 3", "Reply 3")
|
||||
|
||||
context = await memory.get_conversation_context("Bob", limit=2)
|
||||
assert len(context) == 2
|
||||
# Should be in chronological order
|
||||
assert context[0]["message"] == "Msg 2"
|
||||
assert context[1]["message"] == "Msg 3"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_npc_interaction(self, memory):
|
||||
"""Test recording NPC interactions."""
|
||||
await memory.add_npc_interaction(
|
||||
"Guard", "talk", {"topic": "directions"}, reputation_delta=0.2
|
||||
)
|
||||
|
||||
rep = await memory.get_npc_reputation("Guard")
|
||||
assert rep is not None
|
||||
assert rep["reputation_score"] == 0.2
|
||||
assert rep["interactions_count"] == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_npc_reputation_accumulation(self, memory):
|
||||
"""Test that NPC reputation accumulates."""
|
||||
await memory.add_npc_interaction("Dragon", "combat", {}, reputation_delta=-0.5)
|
||||
await memory.add_npc_interaction("Dragon", "combat", {}, reputation_delta=-0.3)
|
||||
|
||||
rep = await memory.get_npc_reputation("Dragon")
|
||||
assert rep["reputation_score"] == -0.8
|
||||
assert rep["interactions_count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_event(self, memory):
|
||||
"""Test recording important events."""
|
||||
await memory.add_event(
|
||||
"level_up", {"level": 10, "class": "warrior"}, importance=0.9
|
||||
)
|
||||
|
||||
memories = await memory.get_important_memories(min_importance=0.8)
|
||||
assert len(memories) == 1
|
||||
assert memories[0].key == "level_up"
|
||||
assert memories[0].data["details"]["level"] == 10
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_memories(self, memory):
|
||||
"""Test full-text search."""
|
||||
await memory.add_conversation("Charlie", "Want to fight the dragon?", "Sure!")
|
||||
await memory.add_conversation("Dave", "Let's explore the forest", "Okay")
|
||||
await memory.add_event("found_item", {"item": "dragon scale"})
|
||||
|
||||
results = await memory.search_memories("dragon")
|
||||
assert len(results) >= 2 # Should find conversation and event
|
||||
|
||||
dragon_conversation = [r for r in results if r.category == "conversation"]
|
||||
assert len(dragon_conversation) == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_with_category(self, memory):
|
||||
"""Test searching with category filter."""
|
||||
await memory.add_conversation("Eve", "Hello", "Hi")
|
||||
await memory.add_event("combat", {"enemy": "goblin"})
|
||||
|
||||
# Search only events
|
||||
results = await memory.search_memories("goblin", category="event")
|
||||
assert len(results) == 1
|
||||
assert results[0].category == "event"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_context_prompt(self, memory):
|
||||
"""Test building context prompt."""
|
||||
await memory.add_conversation("Frank", "Hello!", "Hey!", importance=0.8)
|
||||
await memory.add_event("quest_complete", {"quest": "tutorial"}, importance=0.9)
|
||||
|
||||
context = await memory.build_context_prompt("Frank", max_tokens=500)
|
||||
assert "Previous memories:" in context
|
||||
assert "Frank" in context
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_import_json(self, memory, tmp_path):
|
||||
"""Test exporting and importing memories."""
|
||||
# Add some data
|
||||
await memory.add_conversation("Grace", "Test", "Reply")
|
||||
await memory.add_event("test_event", {"test": True})
|
||||
|
||||
# Export
|
||||
export_path = tmp_path / "memories_export.json"
|
||||
await memory.export_to_json(str(export_path))
|
||||
assert export_path.exists()
|
||||
|
||||
# Create new memory and import
|
||||
new_mem = LLMMemory(str(tmp_path / "new_memory.db"))
|
||||
await new_mem.initialize()
|
||||
count = await new_mem.import_from_json(str(export_path))
|
||||
|
||||
assert count == 2
|
||||
|
||||
# Verify data
|
||||
context = await new_mem.get_conversation_context("Grace")
|
||||
assert len(context) == 1
|
||||
|
||||
await new_mem.close()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_player_relationship_tracking(self, memory):
|
||||
"""Test that player relationships are tracked."""
|
||||
await memory.add_conversation("Henry", "Hey!", "Hello!")
|
||||
await memory.add_conversation("Henry", "How are you?", "Good!")
|
||||
|
||||
# Check that relationship was created
|
||||
# Note: We don't have a direct getter, but we can infer from conversation count
|
||||
context = await memory.get_conversation_context("Henry")
|
||||
assert len(context) == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_memory_operations(self, memory):
|
||||
"""Test operations on empty memory."""
|
||||
context = await memory.get_conversation_context("Nobody", limit=5)
|
||||
assert context == []
|
||||
|
||||
rep = await memory.get_npc_reputation("UnknownNPC")
|
||||
assert rep is None
|
||||
|
||||
results = await memory.search_memories("nonexistent")
|
||||
assert results == []
|
||||
|
||||
prompt = await memory.build_context_prompt("Nobody")
|
||||
assert prompt == ""
|
||||
|
||||
|
||||
class TestMemoryEntry:
|
||||
"""Test MemoryEntry dataclass."""
|
||||
|
||||
def test_memory_entry_creation(self):
|
||||
"""Test creating a MemoryEntry."""
|
||||
entry = MemoryEntry(
|
||||
id=1,
|
||||
timestamp=datetime.now(),
|
||||
category="test",
|
||||
key="test_key",
|
||||
data={"value": 123},
|
||||
importance=0.5
|
||||
)
|
||||
|
||||
assert entry.id == 1
|
||||
assert entry.category == "test"
|
||||
assert entry.key == "test_key"
|
||||
assert entry.data["value"] == 123
|
||||
assert entry.importance == 0.5
|
||||
Reference in new issue
Block a user