diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py index 0ad238d5..9ba3d7dd 100644 --- a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py +++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py @@ -5,11 +5,12 @@ from datetime import datetime, timezone from typing import Any, Dict, List, Union -def get_last_user_message(messages: List[Dict[str, str]]) -> str | None: +def get_last_user_message(messages: List[Dict[str, Any]]) -> str | None: """Extract the last user message content from a list of messages.""" for msg in reversed(messages): - if msg["role"] == "user": - return msg["content"] + content = msg.get("content") + if msg.get("role") == "user" and isinstance(content, str): + return content return None @@ -70,6 +71,22 @@ def _field(item: Any, *names: str, default: Any = None) -> Any: return default +def _memory_text(item: Any) -> str: + """First non-empty memory field, mirroring the TypeScript `getMemoryText`. + + `_field` stops at the first value that is not None, so a hybrid-search hit + shaped `{"memory": "", "chunk": "..."}` resolves to the empty string and + never falls through to `chunk`. + """ + if isinstance(item, str): + return item.strip() + for name in ("memory", "chunk", "content"): + value = _field(item, name) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + _MEMORY_DATE_PREFIX = re.compile( r"^\s*(?:\[recent\]\s*)?(?:\[\d{4}-\d{2}-\d{2}\]\s*)?", re.IGNORECASE, @@ -124,10 +141,7 @@ def deduplicate_memories( out = [] for r in results: # v4 search.memories/hybrid uses `memory` or `chunk`. - memory = _field(r, "memory", "chunk", "content", default="") - if not isinstance(memory, str): - memory = "" - memory = memory.strip() + memory = _memory_text(r) key = _memory_key(memory) if key and key not in seen: seen.add(key) @@ -174,7 +188,7 @@ def format_memories_to_text( lines.append(f"- {item}") continue - memory = _field(item, "memory", "chunk", "content", default="") + memory = _memory_text(item) updated_at = _field(item, "updatedAt", "updated_at", default="") time_str = format_relative_time(updated_at) if updated_at else "" if time_str: diff --git a/packages/cartesia-sdk-python/tests/test_utils.py b/packages/cartesia-sdk-python/tests/test_utils.py new file mode 100644 index 00000000..1b380e9d --- /dev/null +++ b/packages/cartesia-sdk-python/tests/test_utils.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +import sys +import types +import unittest +from types import SimpleNamespace + + +def _install_test_stubs() -> None: + if "loguru" not in sys.modules: + loguru_module = types.ModuleType("loguru") + + class _Logger: + def info(self, *_args, **_kwargs): + return None + + def warning(self, *_args, **_kwargs): + return None + + def error(self, *_args, **_kwargs): + return None + + loguru_module.logger = _Logger() + sys.modules["loguru"] = loguru_module + + if "pydantic" not in sys.modules: + pydantic_module = types.ModuleType("pydantic") + + class BaseModel: + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def Field(*, default=None, **_kwargs): + return default + + pydantic_module.BaseModel = BaseModel + pydantic_module.Field = Field + sys.modules["pydantic"] = pydantic_module + + +_install_test_stubs() + +from supermemory_cartesia.utils import ( + deduplicate_memories, + format_memories_to_text, + get_last_user_message, +) + + +class TestGetLastUserMessage(unittest.TestCase): + def test_returns_the_most_recent_user_message(self) -> None: + messages = [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "reply"}, + {"role": "user", "content": "second"}, + ] + + self.assertEqual(get_last_user_message(messages), "second") + + def test_skips_entries_without_role_or_content(self) -> None: + # Tool-call turns and provider events carry neither key, and reading + # them positionally used to raise KeyError instead of skipping them. + messages = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "tool_calls": []}, + {"content": "provider event"}, + {"role": "user"}, + ] + + self.assertEqual(get_last_user_message(messages), "hello") + + def test_skips_non_string_content(self) -> None: + messages = [{"role": "user", "content": [{"type": "text", "text": "hello"}]}] + + self.assertIsNone(get_last_user_message(messages)) + + +class TestDeduplicateMemories(unittest.TestCase): + def test_keeps_plain_string_search_results(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=["User prefers tea"], + ) + + self.assertEqual(result["search_results"], ["User prefers tea"]) + self.assertIn("User prefers tea", format_memories_to_text(result)) + + def test_still_reads_the_memory_field_of_models_and_dicts(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[ + SimpleNamespace(memory="From a model"), + {"chunk": "From a dict"}, + ], + ) + + self.assertEqual(len(result["search_results"]), 2) + rendered = format_memories_to_text(result) + self.assertIn("From a model", rendered) + self.assertIn("From a dict", rendered) + + def test_deduplicates_a_string_result_already_in_the_profile(self) -> None: + fact = "User is allergic to peanuts" + result = deduplicate_memories( + static=[fact], + dynamic=[], + search_results=[fact], + ) + + self.assertEqual(result["static"], [fact]) + self.assertEqual(result["search_results"], []) + + +class TestSearchResultExtraction(unittest.TestCase): + """v4 hybrid search returns chunk hits whose `memory` field is empty. + + `_field` stops at the first value that is not None, so those hits used to + resolve to "" and were dropped by dedup -- or, when they survived, rendered + as an empty bullet. The TypeScript `getMemoryText` takes the first + non-empty field instead, and the other Supermemory Python SDKs match it. + """ + + def test_falls_through_to_chunk_when_memory_is_empty(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": "", "chunk": "The user's dog is called Rex"}], + ) + + self.assertEqual(len(result["search_results"]), 1) + self.assertIn("The user's dog is called Rex", format_memories_to_text(result)) + + def test_falls_through_to_chunk_when_memory_is_whitespace(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": " ", "chunk": "Chunk body"}], + ) + + self.assertEqual(len(result["search_results"]), 1) + self.assertIn("Chunk body", format_memories_to_text(result)) + + def test_falls_through_on_models_too(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[SimpleNamespace(memory="", chunk="Chunk body")], + ) + + self.assertEqual(len(result["search_results"]), 1) + self.assertIn("Chunk body", format_memories_to_text(result)) + + def test_never_renders_an_empty_bullet(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[ + { + "memory": "", + "chunk": "Chunk body", + "updatedAt": "2020-01-01T00:00:00Z", + } + ], + ) + + rendered = format_memories_to_text(result) + self.assertIn("Chunk body", rendered) + for line in rendered.splitlines(): + if line.startswith("- "): + self.assertNotRegex(line, r"^- (\[[^\]]*\] )?$") + + def test_memory_still_wins_over_chunk(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": "Memory body", "chunk": "Chunk body"}], + ) + + rendered = format_memories_to_text(result) + self.assertIn("Memory body", rendered) + self.assertNotIn("Chunk body", rendered) + + def test_entry_with_no_usable_text_is_still_dropped(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": "", "chunk": ""}, {}, None], + ) + + self.assertEqual(result["search_results"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index cc8d6534..7b9cb49d 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -90,6 +90,22 @@ def _field(item: Any, *names: str, default: Any = None) -> Any: return default +def _memory_text(item: Any) -> str: + """First non-empty memory field, mirroring the TypeScript `getMemoryText`. + + `_field` stops at the first value that is not None, so a hybrid-search hit + shaped `{"memory": "", "chunk": "..."}` resolves to the empty string and + never falls through to `chunk`. + """ + if isinstance(item, str): + return item.strip() + for name in ("memory", "chunk", "content"): + value = _field(item, name) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + def deduplicate_memories( static: List[str], dynamic: List[str], @@ -125,12 +141,7 @@ def deduplicate_memories( out: List[Any] = [] for r in results: # v4 search.memories/hybrid uses `memory` or `chunk`. - memory = ( - r if isinstance(r, str) else _field(r, "memory", "chunk", "content", default="") - ) - if not isinstance(memory, str): - memory = "" - memory = memory.strip() + memory = _memory_text(r) key = comparison_key(memory) if key and key not in seen: seen.add(key) @@ -177,7 +188,7 @@ def format_memories_to_text( lines.append(f"- {item}") continue - memory = _field(item, "memory", "chunk", "content", default="") + memory = _memory_text(item) updated_at = _field(item, "updatedAt", "updated_at", default="") time_str = format_relative_time(updated_at) if updated_at else "" if time_str: diff --git a/packages/pipecat-sdk-python/tests/conftest.py b/packages/pipecat-sdk-python/tests/conftest.py new file mode 100644 index 00000000..4fb1fe6d --- /dev/null +++ b/packages/pipecat-sdk-python/tests/conftest.py @@ -0,0 +1,103 @@ +"""Import stubs for the heavy optional runtime dependencies. + +CI installs the real `pipecat-ai`, `loguru` and `pydantic`, so every stub +here is a no-op there. Locally they let the pure-Python helpers under test +import without the full voice stack, and living in conftest means any test +module gets them regardless of collection order. +""" + +from __future__ import annotations + +import sys +import types + + +def _install_test_stubs() -> None: + if "loguru" not in sys.modules: + loguru_module = types.ModuleType("loguru") + + class _Logger: + def warning(self, *_args, **_kwargs): + return None + + def error(self, *_args, **_kwargs): + return None + + loguru_module.logger = _Logger() + sys.modules["loguru"] = loguru_module + + if "pydantic" not in sys.modules: + pydantic_module = types.ModuleType("pydantic") + + class BaseModel: + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def Field(*, default=None, **_kwargs): + return default + + pydantic_module.BaseModel = BaseModel + pydantic_module.Field = Field + sys.modules["pydantic"] = pydantic_module + + if "pipecat" not in sys.modules: + pipecat_module = types.ModuleType("pipecat") + sys.modules["pipecat"] = pipecat_module + + frames_module = types.ModuleType("pipecat.frames.frames") + + class Frame: # pragma: no cover - import stub + pass + + class InputAudioRawFrame: # pragma: no cover - import stub + pass + + class LLMContextFrame: # pragma: no cover - import stub + pass + + class LLMMessagesFrame: # pragma: no cover - import stub + pass + + frames_module.Frame = Frame + frames_module.InputAudioRawFrame = InputAudioRawFrame + frames_module.LLMContextFrame = LLMContextFrame + frames_module.LLMMessagesFrame = LLMMessagesFrame + + llm_context_module = types.ModuleType("pipecat.processors.aggregators.llm_context") + + class LLMContext: # pragma: no cover - import stub + pass + + llm_context_module.LLMContext = LLMContext + + openai_context_module = types.ModuleType( + "pipecat.processors.aggregators.openai_llm_context" + ) + + class OpenAILLMContextFrame: # pragma: no cover - import stub + pass + + openai_context_module.OpenAILLMContextFrame = OpenAILLMContextFrame + + frame_processor_module = types.ModuleType("pipecat.processors.frame_processor") + + class FrameDirection: # pragma: no cover - import stub + pass + + class FrameProcessor: + def __init__(self, *args, **kwargs): + return None + + frame_processor_module.FrameDirection = FrameDirection + frame_processor_module.FrameProcessor = FrameProcessor + + sys.modules["pipecat.frames.frames"] = frames_module + sys.modules["pipecat.processors.aggregators.llm_context"] = llm_context_module + sys.modules[ + "pipecat.processors.aggregators.openai_llm_context" + ] = openai_context_module + sys.modules["pipecat.processors.frame_processor"] = frame_processor_module + + +_install_test_stubs() diff --git a/packages/pipecat-sdk-python/tests/test_utils.py b/packages/pipecat-sdk-python/tests/test_utils.py new file mode 100644 index 00000000..d9befd14 --- /dev/null +++ b/packages/pipecat-sdk-python/tests/test_utils.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from supermemory_pipecat.utils import ( + deduplicate_memories, + format_memories_to_text, +) + + +class TestSearchResultExtraction(unittest.TestCase): + """v4 hybrid search returns chunk hits whose `memory` field is empty. + + `_field` stops at the first value that is not None, so those hits used to + resolve to "" and were dropped by dedup -- or, when they survived, rendered + as an empty bullet. The TypeScript `getMemoryText` takes the first + non-empty field instead, and the other Supermemory Python SDKs match it. + """ + + def test_falls_through_to_chunk_when_memory_is_empty(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": "", "chunk": "The user's dog is called Rex"}], + ) + + self.assertEqual(len(result["search_results"]), 1) + self.assertIn("The user's dog is called Rex", format_memories_to_text(result)) + + def test_falls_through_to_chunk_when_memory_is_whitespace(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": " ", "chunk": "Chunk body"}], + ) + + self.assertEqual(len(result["search_results"]), 1) + self.assertIn("Chunk body", format_memories_to_text(result)) + + def test_falls_through_on_models_too(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[SimpleNamespace(memory="", chunk="Chunk body")], + ) + + self.assertEqual(len(result["search_results"]), 1) + self.assertIn("Chunk body", format_memories_to_text(result)) + + def test_never_renders_an_empty_bullet(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[ + { + "memory": "", + "chunk": "Chunk body", + "updatedAt": "2020-01-01T00:00:00Z", + } + ], + ) + + rendered = format_memories_to_text(result) + self.assertIn("Chunk body", rendered) + for line in rendered.splitlines(): + if line.startswith("- "): + self.assertNotRegex(line, r"^- (\[[^\]]*\] )?$") + + def test_memory_still_wins_over_chunk(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": "Memory body", "chunk": "Chunk body"}], + ) + + rendered = format_memories_to_text(result) + self.assertIn("Memory body", rendered) + self.assertNotIn("Chunk body", rendered) + + def test_entry_with_no_usable_text_is_still_dropped(self) -> None: + result = deduplicate_memories( + static=[], + dynamic=[], + search_results=[{"memory": "", "chunk": ""}, {}, None], + ) + + self.assertEqual(result["search_results"], []) + + +if __name__ == "__main__": + unittest.main()