[gemini] Implement semantic index for research outputs (#976) (#1227)

This commit is contained in:
2026-03-23 22:45:29 +00:00
parent e6391c599d
commit 697575e561
5 changed files with 179 additions and 81 deletions

View File

@@ -16,7 +16,7 @@ from timmy.memory_system import (
memory_forget,
memory_read,
memory_search,
memory_write,
memory_store,
)
@@ -490,7 +490,7 @@ class TestMemorySearch:
assert isinstance(result, str)
def test_none_top_k_handled(self):
result = memory_search("test", top_k=None)
result = memory_search("test", limit=None)
assert isinstance(result, str)
def test_basic_search_returns_string(self):
@@ -521,12 +521,12 @@ class TestMemoryRead:
assert isinstance(result, str)
class TestMemoryWrite:
"""Test module-level memory_write function."""
class TestMemoryStore:
"""Test module-level memory_store function."""
@pytest.fixture(autouse=True)
def mock_vector_store(self):
"""Mock vector_store functions for memory_write tests."""
"""Mock vector_store functions for memory_store tests."""
# Patch where it's imported from, not where it's used
with (
patch("timmy.memory_system.search_memories") as mock_search,
@@ -542,75 +542,83 @@ class TestMemoryWrite:
yield {"search": mock_search, "store": mock_store}
def test_memory_write_empty_content(self):
"""Test that empty content returns error message."""
result = memory_write("")
def test_memory_store_empty_report(self):
"""Test that empty report returns error message."""
result = memory_store(topic="test", report="")
assert "empty" in result.lower()
def test_memory_write_whitespace_only(self):
"""Test that whitespace-only content returns error."""
result = memory_write(" \n\t ")
def test_memory_store_whitespace_only(self):
"""Test that whitespace-only report returns error."""
result = memory_store(topic="test", report=" \n\t ")
assert "empty" in result.lower()
def test_memory_write_valid_content(self, mock_vector_store):
def test_memory_store_valid_content(self, mock_vector_store):
"""Test writing valid content."""
result = memory_write("Remember this important fact.")
result = memory_store(topic="fact about Timmy", report="Remember this important fact.")
assert "stored" in result.lower() or "memory" in result.lower()
mock_vector_store["store"].assert_called_once()
def test_memory_write_dedup_for_facts(self, mock_vector_store):
"""Test that duplicate facts are skipped."""
def test_memory_store_dedup_for_facts_or_research(self, mock_vector_store):
"""Test that duplicate facts or research are skipped."""
# Simulate existing similar fact
mock_entry = MagicMock()
mock_entry.id = "existing-id"
mock_vector_store["search"].return_value = [mock_entry]
result = memory_write("Similar fact text", context_type="fact")
# Test with 'fact'
result = memory_store(topic="Similar fact", report="Similar fact text", type="fact")
assert "similar" in result.lower() or "duplicate" in result.lower()
mock_vector_store["store"].assert_not_called()
def test_memory_write_no_dedup_for_conversation(self, mock_vector_store):
mock_vector_store["store"].reset_mock()
# Test with 'research'
result = memory_store(topic="Similar research", report="Similar research content", type="research")
assert "similar" in result.lower() or "duplicate" in result.lower()
mock_vector_store["store"].assert_not_called()
def test_memory_store_no_dedup_for_conversation(self, mock_vector_store):
"""Test that conversation entries are not deduplicated."""
# Even with existing entries, conversations should be stored
mock_entry = MagicMock()
mock_entry.id = "existing-id"
mock_vector_store["search"].return_value = [mock_entry]
memory_write("Conversation text", context_type="conversation")
memory_store(topic="Conversation", report="Conversation text", type="conversation")
# Should still store (no duplicate check for non-fact)
mock_vector_store["store"].assert_called_once()
def test_memory_write_invalid_context_type(self, mock_vector_store):
"""Test that invalid context_type defaults to 'fact'."""
memory_write("Some content", context_type="invalid_type")
# Should still succeed, using "fact" as default
def test_memory_store_invalid_type_defaults_to_research(self, mock_vector_store):
"""Test that invalid type defaults to 'research'."""
memory_store(topic="Invalid type test", report="Some content", type="invalid_type")
# Should still succeed, using "research" as default
mock_vector_store["store"].assert_called_once()
call_kwargs = mock_vector_store["store"].call_args.kwargs
assert call_kwargs.get("context_type") == "fact"
assert call_kwargs.get("context_type") == "research"
def test_memory_write_valid_context_types(self, mock_vector_store):
def test_memory_store_valid_types(self, mock_vector_store):
"""Test all valid context types."""
valid_types = ["fact", "conversation", "document"]
valid_types = ["fact", "conversation", "document", "research"]
for ctx_type in valid_types:
mock_vector_store["store"].reset_mock()
memory_write(f"Content for {ctx_type}", context_type=ctx_type)
memory_store(topic=f"Topic for {ctx_type}", report=f"Content for {ctx_type}", type=ctx_type)
mock_vector_store["store"].assert_called_once()
def test_memory_write_strips_content(self, mock_vector_store):
"""Test that content is stripped of leading/trailing whitespace."""
memory_write(" padded content ")
def test_memory_store_strips_report_and_adds_topic(self, mock_vector_store):
"""Test that report is stripped of leading/trailing whitespace and combined with topic."""
memory_store(topic=" My Topic ", report=" padded content ")
call_kwargs = mock_vector_store["store"].call_args.kwargs
assert call_kwargs.get("content") == "padded content"
assert call_kwargs.get("content") == "Topic: My Topic\n\nReport: padded content"
assert call_kwargs.get("metadata") == {"topic": " My Topic "}
def test_memory_write_unicode_content(self, mock_vector_store):
def test_memory_store_unicode_report(self, mock_vector_store):
"""Test writing unicode content."""
result = memory_write("Unicode content: 你好世界 🎉")
result = memory_store(topic="Unicode", report="Unicode content: 你好世界 🎉")
assert "stored" in result.lower() or "memory" in result.lower()
def test_memory_write_handles_exception(self, mock_vector_store):
def test_memory_store_handles_exception(self, mock_vector_store):
"""Test handling of store_memory exceptions."""
mock_vector_store["store"].side_effect = Exception("DB error")
result = memory_write("This will fail")
result = memory_store(topic="Failing", report="This will fail")
assert "failed" in result.lower() or "error" in result.lower()