From b65c10bb47f34d5a7790e316a2552893fd59b524 Mon Sep 17 00:00:00 2001 From: Agnik47 <140933190+Agnik47@users.noreply.github.com> Date: Fri, 18 Sep 2026 22:15:17 +0530 Subject: [PATCH] fix(cartesia): keep plain-string search results and skip malformed messages `utils.py` in cartesia-sdk-python and pipecat-sdk-python are copies of the same file. #1434 hardened two spots in the pipecat copy and missed cartesia. `unique_search` calls `_field(r, ...)` on every search result. For a plain string that finds no attribute, falls back to `""`, and the entry is dropped before it reaches the prompt. `format_memories_to_text` in the same file has an `isinstance(item, str)` branch for search results, so that branch is currently unreachable. `get_last_user_message` indexed `msg["role"]` and `msg["content"]` positionally, so a tool-call turn or provider event missing either key raised KeyError instead of being skipped, and non-string content was returned verbatim despite the `str | None` annotation. Both hunks now match the pipecat implementation. --- .../src/supermemory_cartesia/utils.py | 11 +- .../cartesia-sdk-python/tests/test_utils.py | 118 ++++++++++++++++++ 2 files changed, 125 insertions(+), 4 deletions(-) create mode 100644 packages/cartesia-sdk-python/tests/test_utils.py diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py index 0ad238d5..064693a3 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 @@ -124,7 +125,9 @@ def deduplicate_memories( out = [] for r in results: # v4 search.memories/hybrid uses `memory` or `chunk`. - memory = _field(r, "memory", "chunk", "content", default="") + memory = ( + r if isinstance(r, str) else _field(r, "memory", "chunk", "content", default="") + ) if not isinstance(memory, str): memory = "" memory = memory.strip() 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..640d2171 --- /dev/null +++ b/packages/cartesia-sdk-python/tests/test_utils.py @@ -0,0 +1,118 @@ +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"], []) + + +if __name__ == "__main__": + unittest.main()