From 4a5804f5a5817beff0d3e40c2c361e94ec6d7ee4 Mon Sep 17 00:00:00 2001 From: Sreeram Sreedhar Date: Wed, 4 Nov 2026 12:06:24 -0700 Subject: [PATCH] fix(pipecat): preserve v5 retrieval behavior --- .../src/supermemory_pipecat/service.py | 14 ++++++- .../src/supermemory_pipecat/utils.py | 4 +- .../tests/test_empty_profile.py | 39 ++++++++++++++++++- 3 files changed, 52 insertions(+), 5 deletions(-) diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py index 82ffa391..0a3e3f2e 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py @@ -235,10 +235,20 @@ class SupermemoryPipecatService(FrameProcessor): self.user_id, query=query, search_mode="memories", - limit=self.params.search_limit, + limit=min(self.params.search_limit, 100), threshold=self.params.search_threshold, ) - responses = dict(zip(calls, await asyncio.gather(*calls.values()))) + results = await asyncio.gather(*calls.values(), return_exceptions=True) + responses = {} + for name, result in zip(calls, results): + if isinstance(result, asyncio.CancelledError): + raise result + if isinstance(result, Exception): + logger.warning(f"Supermemory {name} failed: {result}") + else: + responses[name] = result + if calls and not responses: + raise results[0] profile = _field(responses.get("profile"), "profile") search_results = list(_field(responses.get("search"), "results", default=[]) or []) diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index b8bfd697..dc2eb02a 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -178,7 +178,9 @@ def format_memories_to_text( continue memory = _memory_text(item) - updated_at = _field(item, "updatedAt", "updated_at", default="") + updated_at = _field(item, "updatedAt", "updated_at") or _field( + _field(item, "system"), "updatedAt", "updated_at", default="" + ) time_str = format_relative_time(updated_at) if updated_at else "" if time_str: lines.append(f"- [{time_str}] {memory}") diff --git a/packages/pipecat-sdk-python/tests/test_empty_profile.py b/packages/pipecat-sdk-python/tests/test_empty_profile.py index 4a4a1343..41155a3f 100644 --- a/packages/pipecat-sdk-python/tests/test_empty_profile.py +++ b/packages/pipecat-sdk-python/tests/test_empty_profile.py @@ -9,6 +9,7 @@ import httpx from supermemory import AsyncSupermemory from supermemory_pipecat.service import SupermemoryPipecatService +from supermemory_pipecat.utils import format_memories_to_text class _Context: @@ -25,13 +26,28 @@ class _Context: class TestSupermemoryPipecatService(unittest.IsolatedAsyncioTestCase): async def test_v5_reads_and_writes_use_user_id_as_namespace(self) -> None: requests = {} + failed = set() def respond(request): requests[request.url.path] = json.loads(request.content) + if request.url.path.rsplit("/", 1)[-1] in failed: + return httpx.Response(500) if request.url.path.endswith("/profile"): payload = {"profile": {"static": [{"memory": "Likes tea"}], "dynamic": []}} elif request.url.path.endswith("/search"): - payload = {"results": [{"memory": "Plans a trip"}]} + payload = { + "results": [ + { + "id": "mem_123", + "memory": "Plans a trip", + "metadata": {}, + "similarity": 0.9, + "isLatest": True, + "isInference": False, + "system": {"updatedAt": "2020-01-01T00:00:00Z"}, + } + ] + } else: payload = {"id": "conversation-456"} return httpx.Response(200, json=payload) @@ -42,11 +58,19 @@ class TestSupermemoryPipecatService(unittest.IsolatedAsyncioTestCase): http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), ) as client: service = SupermemoryPipecatService( - api_key="mock_key", user_id="user-123", session_id="conversation-456" + api_key="mock_key", + user_id="user-123", + session_id="conversation-456", + params=SupermemoryPipecatService.InputParams(search_limit=500), ) service._supermemory_client = client result = await service._retrieve_memories("Where am I going?") await service._store_messages([{"role": "user", "content": "Hello"}]) + failed.add("search") + partial_result = await service._retrieve_memories("Where am I going?") + failed.clear() + failed.add("profile") + search_only_result = await service._retrieve_memories("Where am I going?") self.assertEqual(result["profile"]["static"], ["Likes tea"]) self.assertEqual(result["search_results"][0].memory, "Plans a trip") @@ -54,7 +78,18 @@ class TestSupermemoryPipecatService(unittest.IsolatedAsyncioTestCase): set(requests), {"/ns/user-123/profile", "/ns/user-123/search", "/ns/user-123/document"} ) self.assertEqual(requests["/ns/user-123/search"]["query"], "Where am I going?") + self.assertEqual(requests["/ns/user-123/search"]["limit"], 100) self.assertEqual(requests["/ns/user-123/document"]["id"], "conversation-456") + self.assertEqual(partial_result["profile"]["static"], ["Likes tea"]) + self.assertEqual(partial_result["search_results"], []) + self.assertEqual(search_only_result["profile"]["static"], []) + self.assertEqual(search_only_result["search_results"][0].memory, "Plans a trip") + self.assertIn( + "[1 Jan, 2020] Plans a trip", + format_memories_to_text( + {"static": [], "dynamic": [], "search_results": result["search_results"]} + ), + ) async def test_retrieve_memories_handles_null_profile(self) -> None: service = SupermemoryPipecatService(api_key="mock_key", user_id="new_user_123")