From cbf55e59a834ba75447b49eec066ac040f2c8b6a Mon Sep 17 00:00:00 2001 From: Sreeram Sreedhar Date: Mon, 2 Nov 2026 19:41:19 -0700 Subject: [PATCH 1/5] feat(pipecat): migrate SDK integration to Supermemory v5 --- apps/docs/integrations/pipecat.mdx | 16 ++- packages/pipecat-sdk-python/Agents.md | 6 +- packages/pipecat-sdk-python/README.md | 19 +-- packages/pipecat-sdk-python/pyproject.toml | 4 +- .../src/supermemory_pipecat/__init__.py | 4 +- .../src/supermemory_pipecat/service.py | 72 ++++++---- .../src/supermemory_pipecat/utils.py | 5 +- packages/pipecat-sdk-python/tests/conftest.py | 11 +- .../tests/test_empty_profile.py | 135 +++++++++++++++--- .../pipecat-sdk-python/tests/test_utils.py | 2 +- 10 files changed, 202 insertions(+), 72 deletions(-) diff --git a/apps/docs/integrations/pipecat.mdx b/apps/docs/integrations/pipecat.mdx index e54e9450..7c904b69 100644 --- a/apps/docs/integrations/pipecat.mdx +++ b/apps/docs/integrations/pipecat.mdx @@ -15,6 +15,9 @@ To use Supermemory with Pipecat, install the required dependencies: pip install supermemory-pipecat ``` +Version 0.2.0 uses the Supermemory Python SDK v5. If upgrading from 0.1.x, +replace `user_id=` with `namespace=` when constructing the service. + Set up your API key as an environment variable: ```bash @@ -32,7 +35,7 @@ from supermemory_pipecat import InputParams, SupermemoryPipecatService memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - user_id="unique_user_id", + namespace="unique_user_id", session_id="session_123", params=InputParams( mode="full", # "profile" | "query" | "full" @@ -66,7 +69,8 @@ When integrated with Pipecat, Supermemory provides two key functionalities: ### 1. Memory retrieval -When a user message is detected, Supermemory retrieves relevant memories: +When a user message is detected, Supermemory reads the v5 profile and searches +the same namespace when query results are enabled: - **Static Profile**: Persistent facts about the user - **Dynamic Profile**: Recent context and preferences @@ -128,7 +132,7 @@ from supermemory_pipecat import SupermemoryPipecatService # Auto-detection works out of the box memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - user_id="unique_user_id", + namespace="unique_user_id", ) ``` @@ -165,7 +169,7 @@ You remember information from past conversations and use it to provide personali Keep responses brief and conversational.""" -async def run_bot(websocket_client, user_id: str, session_id: str): +async def run_bot(websocket_client, namespace: str, session_id: str): transport = FastAPIWebsocketTransport( websocket=websocket_client, params=FastAPIWebsocketParams( @@ -184,7 +188,7 @@ async def run_bot(websocket_client, user_id: str, session_id: str): # Supermemory memory service memory = SupermemoryPipecatService( - user_id=user_id, + namespace=namespace, session_id=session_id, params=InputParams( mode="full", @@ -220,7 +224,7 @@ async def run_bot(websocket_client, user_id: str, session_id: str): @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() - await run_bot(websocket, user_id="alice", session_id="session-123") + await run_bot(websocket, namespace="alice", session_id="session-123") if __name__ == "__main__": diff --git a/packages/pipecat-sdk-python/Agents.md b/packages/pipecat-sdk-python/Agents.md index 0d4f687c..ca01bd22 100644 --- a/packages/pipecat-sdk-python/Agents.md +++ b/packages/pipecat-sdk-python/Agents.md @@ -20,7 +20,7 @@ Place `SupermemoryPipecatService` between context aggregator and LLM in the pipe from supermemory_pipecat import SupermemoryPipecatService memory = SupermemoryPipecatService( - user_id="user-123", # Required: identifies the user + namespace="user-123", # Required: scopes all memory calls session_id="session-456", # Optional: groups conversations ) @@ -41,7 +41,7 @@ pipeline = Pipeline([ ```python memory = SupermemoryPipecatService( api_key="...", # Or use SUPERMEMORY_API_KEY env var - user_id="user-123", + namespace="user-123", session_id="session-456", params=SupermemoryPipecatService.InputParams( search_limit=10, # Max memories to retrieve @@ -68,5 +68,5 @@ memory = SupermemoryPipecatService( ## Boundaries - Always place memory service after `context_aggregator.user()` and before `llm` -- Always provide `user_id` - it's required +- Always provide `namespace` - it's required - Never hardcode API keys in code - use environment variables diff --git a/packages/pipecat-sdk-python/README.md b/packages/pipecat-sdk-python/README.md index f885ed71..e3b43215 100644 --- a/packages/pipecat-sdk-python/README.md +++ b/packages/pipecat-sdk-python/README.md @@ -8,6 +8,10 @@ Memory-enhanced conversational AI pipelines with [Supermemory](https://supermemo pip install supermemory-pipecat ``` +Version 0.2.0 requires the Supermemory Python SDK v5. When upgrading from +0.1.x, pass `namespace=` instead of `user_id=`. Use the same namespace for +all memory operations belonging to one user, project, or tenant. + ## Quick Start ```python @@ -19,7 +23,7 @@ from supermemory_pipecat import SupermemoryPipecatService # Create memory service memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - user_id="user-123", # Required: used as container_tag + namespace="user-123", # Required: scopes all memory calls session_id="conversation-456", # Optional: groups memories by session ) @@ -45,7 +49,7 @@ pipeline = Pipeline([ | Parameter | Type | Required | Description | | ------------ | ----------- | -------- | ---------------------------------------------------------- | -| `user_id` | str | **Yes** | User identifier - used as container_tag for memory scoping | +| `namespace` | str | **Yes** | Namespace for memory scoping (for example, a user ID) | | `session_id` | str | No | Session/conversation ID for grouping memories | | `api_key` | str | No | Supermemory API key (or set `SUPERMEMORY_API_KEY` env var) | | `params` | InputParams | No | Advanced configuration | @@ -57,7 +61,7 @@ pipeline = Pipeline([ from supermemory_pipecat import SupermemoryPipecatService memory = SupermemoryPipecatService( - user_id="user-123", + namespace="user-123", session_id="conv-456", params=SupermemoryPipecatService.InputParams( search_limit=10, # Max memories to retrieve @@ -81,7 +85,7 @@ memory = SupermemoryPipecatService( 1. **Intercepts context frames** - Listens for Pipecat's universal `LLMContextFrame` (and legacy 0.x frames) 2. **Tracks conversation** - Separates real conversation messages from tagged memory context -3. **Retrieves memories** - Queries `/v4/profile` API with user's message +3. **Retrieves memories** - Reads the v5 profile and, in `query` or `full` mode, searches the same namespace for the user's message 4. **Injects memories** - Uses a system message for audio/system mode and a tagged user message otherwise 5. **Stores messages** - Serializes newly observed user and assistant messages through a background queue that drains during cleanup @@ -98,13 +102,12 @@ User: What's the weather like today? Assistant: It's sunny today. ``` -is sent to Supermemory as: +is added to the `user-123` namespace as: ```json { "content": "[{\"role\": \"user\", \"content\": \"What's the weather like today?\"}, {\"role\": \"assistant\", \"content\": \"It's sunny today.\"}]", - "container_tags": ["user-123"], - "custom_id": "conversation-456", + "id": "conversation-456", "metadata": { "platform": "pipecat" } } ``` @@ -148,7 +151,7 @@ async def websocket_endpoint(websocket: WebSocket): # Supermemory memory service memory = SupermemoryPipecatService( - user_id="alice", + namespace="alice", session_id="session-123", ) diff --git a/packages/pipecat-sdk-python/pyproject.toml b/packages/pipecat-sdk-python/pyproject.toml index 4cf30adf..3ee8a4a4 100644 --- a/packages/pipecat-sdk-python/pyproject.toml +++ b/packages/pipecat-sdk-python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "supermemory-pipecat" -version = "0.1.5" +version = "0.2.0" description = "Supermemory integration for Pipecat - memory-enhanced conversational AI pipelines" readme = "README.md" license = "MIT" @@ -32,7 +32,7 @@ classifiers = [ ] dependencies = [ "pipecat-ai>=0.0.98,<2.0.0", - "supermemory>=3.16.0,<5", + "supermemory>=5.0.0,<6", "pydantic>=2.10.0", "loguru>=0.7.3", ] diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py index c6089e79..0838fae8 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py @@ -10,7 +10,7 @@ Example: # Create memory service memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - user_id="user-123", + namespace="user-123", ) # Add to Pipecat pipeline @@ -48,7 +48,7 @@ try: __version__ = version("supermemory-pipecat") except PackageNotFoundError: # Source-tree fallback; built wheels always use package metadata above. - __version__ = "0.1.5" + __version__ = "0.2.0" __all__ = [ # Main service diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py index d17596c9..262fcec3 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py @@ -23,6 +23,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from .exceptions import ConfigurationError, MemoryRetrievalError, MemoryStorageError from .utils import ( _field, + _memory_text, deduplicate_memories, escape_memory_delimiters, format_memories_to_text, @@ -132,7 +133,7 @@ class SupermemoryPipecatService(FrameProcessor): memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - user_id="user-123", + namespace="user-123", ) ``` """ @@ -158,7 +159,7 @@ class SupermemoryPipecatService(FrameProcessor): self, *, api_key: Optional[str] = None, - user_id: str, + namespace: str, session_id: Optional[str] = None, params: Optional[InputParams] = None, base_url: Optional[str] = None, @@ -167,13 +168,13 @@ class SupermemoryPipecatService(FrameProcessor): Args: api_key: Supermemory API key. Falls back to SUPERMEMORY_API_KEY env var. - user_id: The user ID - used as container_tag for memory scoping. + namespace: The user, project, or tenant namespace for memory scoping. session_id: Session/conversation ID for grouping memories. params: Configuration parameters for memory retrieval. base_url: Optional custom base URL for Supermemory API. Raises: - ConfigurationError: If API key is missing or user_id not provided. + ConfigurationError: If API key or namespace is missing. """ super().__init__() @@ -183,11 +184,10 @@ class SupermemoryPipecatService(FrameProcessor): "API key is required. Provide api_key parameter or set SUPERMEMORY_API_KEY environment variable." ) - if not user_id: - raise ConfigurationError("user_id is required") + if not namespace: + raise ConfigurationError("namespace is required") - self.user_id = user_id - self.container_tag = user_id + self.namespace = namespace self.session_id = session_id self.params = params or SupermemoryPipecatService.InputParams() @@ -226,25 +226,48 @@ class SupermemoryPipecatService(FrameProcessor): ) try: - # One profile call: static + dynamic, and (when mode/query allow) - # search_results via `q`. This is the intended profile API shape. - kwargs: Dict[str, Any] = {"container_tag": self.container_tag} - if self.params.mode != "profile" and query: - kwargs["q"] = query - kwargs["threshold"] = self.params.search_threshold + include_profile = self.params.mode != "query" + include_search = self.params.mode != "profile" and bool(query) + profile_response = None + search_response = None - response = await self._supermemory_client.profile(**kwargs) + if include_profile and include_search: + profile_response, search_response = await asyncio.gather( + self._supermemory_client.profile(self.namespace), + self._supermemory_client.search( + self.namespace, + query=query, + search_mode="memories", + limit=self.params.search_limit, + threshold=self.params.search_threshold, + ), + ) + elif include_profile: + profile_response = await self._supermemory_client.profile(self.namespace) + elif include_search: + search_response = await self._supermemory_client.search( + self.namespace, + query=query, + search_mode="memories", + limit=self.params.search_limit, + threshold=self.params.search_threshold, + ) - profile = _field(response, "profile") - search_results_response = _field(response, "search_results", "searchResults") - - raw_search_results = _field(search_results_response, "results", default=[]) or [] - search_results = list(raw_search_results)[: self.params.search_limit] + profile = _field(profile_response, "profile") + search_results = list(_field(search_response, "results", default=[]) or []) return { "profile": { - "static": list(_field(profile, "static", default=[]) or []), - "dynamic": list(_field(profile, "dynamic", default=[]) or []), + "static": [ + text + for item in _field(profile, "static", default=[]) or [] + if (text := _memory_text(item)) + ], + "dynamic": [ + text + for item in _field(profile, "dynamic", default=[]) or [] + if (text := _memory_text(item)) + ], }, "search_results": search_results, } @@ -264,13 +287,12 @@ class SupermemoryPipecatService(FrameProcessor): try: add_params: Dict[str, Any] = { "content": json.dumps(messages), - "container_tags": [self.container_tag], "metadata": {"platform": "pipecat"}, } if self.session_id: - add_params["custom_id"] = self.session_id + add_params["id"] = self.session_id - await self._supermemory_client.add(**add_params) + await self._supermemory_client.add(self.namespace, **add_params) except Exception as e: raise MemoryStorageError("Failed to store messages", e) from e diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index 2e0041d1..2b09f058 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -4,7 +4,6 @@ import re from datetime import datetime, timezone from typing import Any, Dict, List, Union - _DYNAMIC_DATE_PREFIX = re.compile( r"^\s*(?:\[recent\]\s*)?(?:\[\d{4}-\d{2}-\d{2}\]\s*)?", re.IGNORECASE, @@ -74,7 +73,7 @@ def _field(item: Any, *names: str, default: Any = None) -> Any: """Read a field from a dict or pydantic/SDK model. Accepts camelCase and snake_case names so helpers work with both raw JSON - dicts and Stainless-generated response models. + dicts and generated SDK response models. """ if item is None: return default @@ -135,7 +134,7 @@ def deduplicate_memories( def unique_search(results: List[Any]) -> List[Any]: out: List[Any] = [] for r in results: - # v4 search.memories/hybrid uses `memory` or `chunk`. + # v5 search results can contain a memory or a source chunk. memory = _memory_text(r) key = comparison_key(memory) if key and key not in seen: diff --git a/packages/pipecat-sdk-python/tests/conftest.py b/packages/pipecat-sdk-python/tests/conftest.py index 74c64fed..00f98566 100644 --- a/packages/pipecat-sdk-python/tests/conftest.py +++ b/packages/pipecat-sdk-python/tests/conftest.py @@ -4,10 +4,11 @@ from __future__ import annotations import sys import types +from importlib.util import find_spec def _install_test_stubs() -> None: - if "loguru" not in sys.modules: + if "loguru" not in sys.modules and find_spec("loguru") is None: loguru_module = types.ModuleType("loguru") class _Logger: @@ -20,7 +21,7 @@ def _install_test_stubs() -> None: loguru_module.logger = _Logger() sys.modules["loguru"] = loguru_module - if "pydantic" not in sys.modules: + if "pydantic" not in sys.modules and find_spec("pydantic") is None: pydantic_module = types.ModuleType("pydantic") class BaseModel: @@ -35,7 +36,7 @@ def _install_test_stubs() -> None: pydantic_module.Field = Field sys.modules["pydantic"] = pydantic_module - if "pipecat" not in sys.modules: + if "pipecat" not in sys.modules and find_spec("pipecat") is None: pipecat_module = types.ModuleType("pipecat") sys.modules["pipecat"] = pipecat_module @@ -88,9 +89,7 @@ def _install_test_stubs() -> None: 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.aggregators.openai_llm_context"] = openai_context_module sys.modules["pipecat.processors.frame_processor"] = frame_processor_module diff --git a/packages/pipecat-sdk-python/tests/test_empty_profile.py b/packages/pipecat-sdk-python/tests/test_empty_profile.py index e0b4bf71..5215baa2 100644 --- a/packages/pipecat-sdk-python/tests/test_empty_profile.py +++ b/packages/pipecat-sdk-python/tests/test_empty_profile.py @@ -3,12 +3,13 @@ from __future__ import annotations import sys import types import unittest +from importlib.util import find_spec from types import SimpleNamespace from unittest.mock import AsyncMock def _install_test_stubs() -> None: - if "loguru" not in sys.modules: + if "loguru" not in sys.modules and find_spec("loguru") is None: loguru_module = types.ModuleType("loguru") class _Logger: @@ -21,7 +22,7 @@ def _install_test_stubs() -> None: loguru_module.logger = _Logger() sys.modules["loguru"] = loguru_module - if "pydantic" not in sys.modules: + if "pydantic" not in sys.modules and find_spec("pydantic") is None: pydantic_module = types.ModuleType("pydantic") class BaseModel: @@ -36,7 +37,7 @@ def _install_test_stubs() -> None: pydantic_module.Field = Field sys.modules["pydantic"] = pydantic_module - if "pipecat" not in sys.modules: + if "pipecat" not in sys.modules and find_spec("pipecat") is None: pipecat_module = types.ModuleType("pipecat") sys.modules["pipecat"] = pipecat_module @@ -89,9 +90,7 @@ def _install_test_stubs() -> None: 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.aggregators.openai_llm_context"] = openai_context_module sys.modules["pipecat.processors.frame_processor"] = frame_processor_module @@ -101,16 +100,21 @@ from supermemory_pipecat.service import SupermemoryPipecatService class _MockSupermemoryClient: - def __init__(self, response): - self.profile = AsyncMock(return_value=response) + def __init__(self, profile_response=None, search_response=None): + self.profile = AsyncMock(return_value=profile_response) + self.search = AsyncMock(return_value=search_response) + self.add = AsyncMock() class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): async def test_retrieve_memories_handles_null_profile(self) -> None: - service = SupermemoryPipecatService(api_key="mock_key", user_id="new_user_123") + service = SupermemoryPipecatService(api_key="mock_key", namespace="new_user_123") - response = SimpleNamespace(profile=None, search_results=None) - service._supermemory_client = _MockSupermemoryClient(response) + client = _MockSupermemoryClient( + profile_response=SimpleNamespace(profile=None), + search_response=SimpleNamespace(results=[]), + ) + service._supermemory_client = client result = await service._retrieve_memories("Hello world") @@ -121,12 +125,113 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): "search_results": [], }, ) + client.profile.assert_awaited_once_with("new_user_123") + client.search.assert_awaited_once_with( + "new_user_123", + query="Hello world", + search_mode="memories", + limit=10, + threshold=0.1, + ) + + async def test_full_mode_reads_v5_profile_objects_and_search_results(self) -> None: + service = SupermemoryPipecatService(api_key="mock_key", namespace="user-123") + client = _MockSupermemoryClient( + profile_response=SimpleNamespace( + profile=SimpleNamespace( + static=[SimpleNamespace(id="mem_1", memory="Likes tea")], + dynamic=[{"id": "mem_2", "memory": "[2026-11-02] Traveling"}], + ) + ), + search_response=SimpleNamespace( + results=[SimpleNamespace(id="mem_3", memory="Plans a trip")] + ), + ) + service._supermemory_client = client + + result = await service._retrieve_memories("Where am I going?") + + self.assertEqual( + result["profile"], + { + "static": ["Likes tea"], + "dynamic": ["[2026-11-02] Traveling"], + }, + ) + self.assertEqual(result["search_results"][0].memory, "Plans a trip") + client.profile.assert_awaited_once_with("user-123") + client.search.assert_awaited_once_with( + "user-123", + query="Where am I going?", + search_mode="memories", + limit=10, + threshold=0.1, + ) + + async def test_query_mode_only_searches(self) -> None: + service = SupermemoryPipecatService( + api_key="mock_key", + namespace="user-123", + params=SupermemoryPipecatService.InputParams( + mode="query", search_limit=3, search_threshold=0.5 + ), + ) + client = _MockSupermemoryClient(search_response=SimpleNamespace(results=[])) + service._supermemory_client = client + + result = await service._retrieve_memories("Where am I going?") + + self.assertEqual(result["profile"], {"static": [], "dynamic": []}) + client.profile.assert_not_awaited() + client.search.assert_awaited_once_with( + "user-123", + query="Where am I going?", + search_mode="memories", + limit=3, + threshold=0.5, + ) + + async def test_profile_mode_only_reads_profile(self) -> None: + service = SupermemoryPipecatService( + api_key="mock_key", + namespace="user-123", + params=SupermemoryPipecatService.InputParams(mode="profile"), + ) + client = _MockSupermemoryClient( + profile_response=SimpleNamespace( + profile=SimpleNamespace(static=[{"memory": "Likes tea"}], dynamic=[]) + ) + ) + service._supermemory_client = client + + result = await service._retrieve_memories("Hello") + + self.assertEqual(result["profile"]["static"], ["Likes tea"]) + self.assertEqual(result["search_results"], []) + client.profile.assert_awaited_once_with("user-123") + client.search.assert_not_awaited() + + async def test_store_uses_namespace_and_document_id(self) -> None: + service = SupermemoryPipecatService( + api_key="mock_key", namespace="user-123", session_id="conversation-456" + ) + client = _MockSupermemoryClient() + service._supermemory_client = client + + await service._store_messages([{"role": "user", "content": "Hello"}]) + + client.add.assert_awaited_once_with( + "user-123", + content='[{"role": "user", "content": "Hello"}]', + metadata={"platform": "pipecat"}, + id="conversation-456", + ) def test_query_mode_keeps_search_fact_also_present_in_profile(self) -> None: fact = "User likes machine learning projects" service = SupermemoryPipecatService( api_key="mock_key", - user_id="user-123", + namespace="user-123", session_id="conversation-456", params=SupermemoryPipecatService.InputParams(mode="query"), ) @@ -151,9 +256,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): }, ) - self.assertTrue( - any(fact in message.get("content", "") for message in context.messages) - ) + self.assertTrue(any(fact in message.get("content", "") for message in context.messages)) def test_system_injection_replaces_stale_block_with_backslash_memory(self) -> None: """A memory containing backslashes must not be read as a regex template.""" @@ -161,7 +264,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): fact = "User's repo is at C:" + backslash + "Users" + backslash + "alice" service = SupermemoryPipecatService( api_key="mock_key", - user_id="user-123", + namespace="user-123", session_id="conversation-456", params=SupermemoryPipecatService.InputParams(inject_mode="system"), ) diff --git a/packages/pipecat-sdk-python/tests/test_utils.py b/packages/pipecat-sdk-python/tests/test_utils.py index 2ef69893..23bef3d6 100644 --- a/packages/pipecat-sdk-python/tests/test_utils.py +++ b/packages/pipecat-sdk-python/tests/test_utils.py @@ -10,7 +10,7 @@ from supermemory_pipecat.utils import ( class TestSearchResultExtraction(unittest.TestCase): - """v4 hybrid search returns chunk hits whose `memory` field is empty.""" + """Search can return chunk hits whose `memory` field is empty.""" def test_falls_through_to_chunk_when_memory_is_empty(self) -> None: result = deduplicate_memories( From c0320ca833e171c74a3b53f46dfb011e5abfd9e5 Mon Sep 17 00:00:00 2001 From: sreedharsreeram <141047751+sreedharsreeram@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:09:26 +0000 Subject: [PATCH 2/5] fix(pipecat): preserve user_id interface with v5 namespace calls Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- apps/docs/integrations/pipecat.mdx | 15 ++-- packages/pipecat-sdk-python/Agents.md | 6 +- packages/pipecat-sdk-python/README.md | 14 ++-- .../src/supermemory_pipecat/__init__.py | 2 +- .../src/supermemory_pipecat/service.py | 25 +++--- .../src/supermemory_pipecat/utils.py | 7 +- .../tests/test_empty_profile.py | 76 +++++++++++++++++-- 7 files changed, 102 insertions(+), 43 deletions(-) diff --git a/apps/docs/integrations/pipecat.mdx b/apps/docs/integrations/pipecat.mdx index 7c904b69..ae189343 100644 --- a/apps/docs/integrations/pipecat.mdx +++ b/apps/docs/integrations/pipecat.mdx @@ -15,8 +15,9 @@ To use Supermemory with Pipecat, install the required dependencies: pip install supermemory-pipecat ``` -Version 0.2.0 uses the Supermemory Python SDK v5. If upgrading from 0.1.x, -replace `user_id=` with `namespace=` when constructing the service. +Version 0.2.0 uses the Supermemory Python SDK v5. Keep using `user_id=`; +the service passes that value as the namespace for profile, search, and writes. +`session_id` is passed as the v5 document `id` (previously `custom_id`). Set up your API key as an environment variable: @@ -35,7 +36,7 @@ from supermemory_pipecat import InputParams, SupermemoryPipecatService memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - namespace="unique_user_id", + user_id="unique_user_id", session_id="session_123", params=InputParams( mode="full", # "profile" | "query" | "full" @@ -132,7 +133,7 @@ from supermemory_pipecat import SupermemoryPipecatService # Auto-detection works out of the box memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - namespace="unique_user_id", + user_id="unique_user_id", ) ``` @@ -169,7 +170,7 @@ You remember information from past conversations and use it to provide personali Keep responses brief and conversational.""" -async def run_bot(websocket_client, namespace: str, session_id: str): +async def run_bot(websocket_client, user_id: str, session_id: str): transport = FastAPIWebsocketTransport( websocket=websocket_client, params=FastAPIWebsocketParams( @@ -188,7 +189,7 @@ async def run_bot(websocket_client, namespace: str, session_id: str): # Supermemory memory service memory = SupermemoryPipecatService( - namespace=namespace, + user_id=user_id, session_id=session_id, params=InputParams( mode="full", @@ -224,7 +225,7 @@ async def run_bot(websocket_client, namespace: str, session_id: str): @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() - await run_bot(websocket, namespace="alice", session_id="session-123") + await run_bot(websocket, user_id="alice", session_id="session-123") if __name__ == "__main__": diff --git a/packages/pipecat-sdk-python/Agents.md b/packages/pipecat-sdk-python/Agents.md index ca01bd22..f218be88 100644 --- a/packages/pipecat-sdk-python/Agents.md +++ b/packages/pipecat-sdk-python/Agents.md @@ -20,7 +20,7 @@ Place `SupermemoryPipecatService` between context aggregator and LLM in the pipe from supermemory_pipecat import SupermemoryPipecatService memory = SupermemoryPipecatService( - namespace="user-123", # Required: scopes all memory calls + user_id="user-123", # Required: scopes all memory calls session_id="session-456", # Optional: groups conversations ) @@ -41,7 +41,7 @@ pipeline = Pipeline([ ```python memory = SupermemoryPipecatService( api_key="...", # Or use SUPERMEMORY_API_KEY env var - namespace="user-123", + user_id="user-123", session_id="session-456", params=SupermemoryPipecatService.InputParams( search_limit=10, # Max memories to retrieve @@ -68,5 +68,5 @@ memory = SupermemoryPipecatService( ## Boundaries - Always place memory service after `context_aggregator.user()` and before `llm` -- Always provide `namespace` - it's required +- Always provide `user_id` - it's required - Never hardcode API keys in code - use environment variables diff --git a/packages/pipecat-sdk-python/README.md b/packages/pipecat-sdk-python/README.md index e3b43215..1d692d7d 100644 --- a/packages/pipecat-sdk-python/README.md +++ b/packages/pipecat-sdk-python/README.md @@ -8,9 +8,9 @@ Memory-enhanced conversational AI pipelines with [Supermemory](https://supermemo pip install supermemory-pipecat ``` -Version 0.2.0 requires the Supermemory Python SDK v5. When upgrading from -0.1.x, pass `namespace=` instead of `user_id=`. Use the same namespace for -all memory operations belonging to one user, project, or tenant. +Version 0.2.0 requires the Supermemory Python SDK v5. Keep using `user_id=`; +the service passes that value as the namespace for profile, search, and writes. +`session_id` is passed as the v5 document `id` (previously `custom_id`). ## Quick Start @@ -23,7 +23,7 @@ from supermemory_pipecat import SupermemoryPipecatService # Create memory service memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - namespace="user-123", # Required: scopes all memory calls + user_id="user-123", # Required: scopes all memory calls session_id="conversation-456", # Optional: groups memories by session ) @@ -49,7 +49,7 @@ pipeline = Pipeline([ | Parameter | Type | Required | Description | | ------------ | ----------- | -------- | ---------------------------------------------------------- | -| `namespace` | str | **Yes** | Namespace for memory scoping (for example, a user ID) | +| `user_id` | str | **Yes** | User identifier passed as the v5 namespace for memory scoping | | `session_id` | str | No | Session/conversation ID for grouping memories | | `api_key` | str | No | Supermemory API key (or set `SUPERMEMORY_API_KEY` env var) | | `params` | InputParams | No | Advanced configuration | @@ -61,7 +61,7 @@ pipeline = Pipeline([ from supermemory_pipecat import SupermemoryPipecatService memory = SupermemoryPipecatService( - namespace="user-123", + user_id="user-123", session_id="conv-456", params=SupermemoryPipecatService.InputParams( search_limit=10, # Max memories to retrieve @@ -151,7 +151,7 @@ async def websocket_endpoint(websocket: WebSocket): # Supermemory memory service memory = SupermemoryPipecatService( - namespace="alice", + user_id="alice", session_id="session-123", ) diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py index 0838fae8..7377900e 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py @@ -10,7 +10,7 @@ Example: # Create memory service memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - namespace="user-123", + user_id="user-123", ) # Add to Pipecat pipeline diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py index 262fcec3..3b71243f 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py @@ -133,7 +133,7 @@ class SupermemoryPipecatService(FrameProcessor): memory = SupermemoryPipecatService( api_key=os.getenv("SUPERMEMORY_API_KEY"), - namespace="user-123", + user_id="user-123", ) ``` """ @@ -159,7 +159,7 @@ class SupermemoryPipecatService(FrameProcessor): self, *, api_key: Optional[str] = None, - namespace: str, + user_id: str, session_id: Optional[str] = None, params: Optional[InputParams] = None, base_url: Optional[str] = None, @@ -168,13 +168,13 @@ class SupermemoryPipecatService(FrameProcessor): Args: api_key: Supermemory API key. Falls back to SUPERMEMORY_API_KEY env var. - namespace: The user, project, or tenant namespace for memory scoping. + user_id: User identifier passed as the v5 namespace for memory scoping. session_id: Session/conversation ID for grouping memories. params: Configuration parameters for memory retrieval. base_url: Optional custom base URL for Supermemory API. Raises: - ConfigurationError: If API key or namespace is missing. + ConfigurationError: If API key or user_id is missing. """ super().__init__() @@ -184,10 +184,11 @@ class SupermemoryPipecatService(FrameProcessor): "API key is required. Provide api_key parameter or set SUPERMEMORY_API_KEY environment variable." ) - if not namespace: - raise ConfigurationError("namespace is required") + if not user_id: + raise ConfigurationError("user_id is required") - self.namespace = namespace + self.user_id = user_id + self.container_tag = user_id self.session_id = session_id self.params = params or SupermemoryPipecatService.InputParams() @@ -233,9 +234,9 @@ class SupermemoryPipecatService(FrameProcessor): if include_profile and include_search: profile_response, search_response = await asyncio.gather( - self._supermemory_client.profile(self.namespace), + self._supermemory_client.profile(self.container_tag), self._supermemory_client.search( - self.namespace, + self.container_tag, query=query, search_mode="memories", limit=self.params.search_limit, @@ -243,10 +244,10 @@ class SupermemoryPipecatService(FrameProcessor): ), ) elif include_profile: - profile_response = await self._supermemory_client.profile(self.namespace) + profile_response = await self._supermemory_client.profile(self.container_tag) elif include_search: search_response = await self._supermemory_client.search( - self.namespace, + self.container_tag, query=query, search_mode="memories", limit=self.params.search_limit, @@ -292,7 +293,7 @@ class SupermemoryPipecatService(FrameProcessor): if self.session_id: add_params["id"] = self.session_id - await self._supermemory_client.add(self.namespace, **add_params) + await self._supermemory_client.add(self.container_tag, **add_params) except Exception as e: raise MemoryStorageError("Failed to store messages", e) from e diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index 2b09f058..b8bfd697 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -70,11 +70,7 @@ def format_relative_time(iso_timestamp: str) -> str: def _field(item: Any, *names: str, default: Any = None) -> Any: - """Read a field from a dict or pydantic/SDK model. - - Accepts camelCase and snake_case names so helpers work with both raw JSON - dicts and generated SDK response models. - """ + """Accept alternative field names across raw JSON and SDK response models.""" if item is None: return default if isinstance(item, dict): @@ -134,7 +130,6 @@ def deduplicate_memories( def unique_search(results: List[Any]) -> List[Any]: out: List[Any] = [] for r in results: - # v5 search results can contain a memory or a source chunk. memory = _memory_text(r) key = comparison_key(memory) if key and key not in seen: diff --git a/packages/pipecat-sdk-python/tests/test_empty_profile.py b/packages/pipecat-sdk-python/tests/test_empty_profile.py index 5215baa2..d7e21c24 100644 --- a/packages/pipecat-sdk-python/tests/test_empty_profile.py +++ b/packages/pipecat-sdk-python/tests/test_empty_profile.py @@ -97,6 +97,7 @@ def _install_test_stubs() -> None: _install_test_stubs() from supermemory_pipecat.service import SupermemoryPipecatService +from supermemory_pipecat.exceptions import ConfigurationError class _MockSupermemoryClient: @@ -107,8 +108,69 @@ class _MockSupermemoryClient: class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): + def test_user_id_and_legacy_attribute_are_preserved(self) -> None: + service = SupermemoryPipecatService(api_key="mock_key", user_id="user-123") + + self.assertEqual(service.user_id, "user-123") + self.assertEqual(service.container_tag, "user-123") + + def test_empty_user_id_is_rejected(self) -> None: + with self.assertRaisesRegex(ConfigurationError, "^user_id is required$"): + SupermemoryPipecatService(api_key="mock_key", user_id="") + + async def test_real_v5_client_serializes_user_id_as_namespace(self) -> None: + import json + + import httpx + from supermemory import AsyncSupermemory + + requests = {} + + def respond(request): + requests[request.url.path] = json.loads(request.content) + 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"}]} + else: + payload = {"id": "conversation-456"} + return httpx.Response(200, json=payload) + + async with AsyncSupermemory( + api_key="mock_key", + base_url="https://example.test", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)), + ) as client: + service = SupermemoryPipecatService( + api_key="mock_key", user_id="user-123", session_id="conversation-456" + ) + service._supermemory_client = client + + result = await service._retrieve_memories("Where am I going?") + await service._store_messages([{"role": "user", "content": "Hello"}]) + + self.assertEqual(result["profile"], {"static": ["Likes tea"], "dynamic": []}) + self.assertEqual(result["search_results"][0].memory, "Plans a trip") + self.assertEqual( + requests, + { + "/ns/user-123/profile": {}, + "/ns/user-123/search": { + "query": "Where am I going?", + "searchMode": "memories", + "limit": 10, + "threshold": 0.1, + }, + "/ns/user-123/document": { + "content": '[{"role": "user", "content": "Hello"}]', + "metadata": {"platform": "pipecat"}, + "id": "conversation-456", + }, + }, + ) + async def test_retrieve_memories_handles_null_profile(self) -> None: - service = SupermemoryPipecatService(api_key="mock_key", namespace="new_user_123") + service = SupermemoryPipecatService(api_key="mock_key", user_id="new_user_123") client = _MockSupermemoryClient( profile_response=SimpleNamespace(profile=None), @@ -135,7 +197,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): ) async def test_full_mode_reads_v5_profile_objects_and_search_results(self) -> None: - service = SupermemoryPipecatService(api_key="mock_key", namespace="user-123") + service = SupermemoryPipecatService(api_key="mock_key", user_id="user-123") client = _MockSupermemoryClient( profile_response=SimpleNamespace( profile=SimpleNamespace( @@ -171,7 +233,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): async def test_query_mode_only_searches(self) -> None: service = SupermemoryPipecatService( api_key="mock_key", - namespace="user-123", + user_id="user-123", params=SupermemoryPipecatService.InputParams( mode="query", search_limit=3, search_threshold=0.5 ), @@ -194,7 +256,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): async def test_profile_mode_only_reads_profile(self) -> None: service = SupermemoryPipecatService( api_key="mock_key", - namespace="user-123", + user_id="user-123", params=SupermemoryPipecatService.InputParams(mode="profile"), ) client = _MockSupermemoryClient( @@ -213,7 +275,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): async def test_store_uses_namespace_and_document_id(self) -> None: service = SupermemoryPipecatService( - api_key="mock_key", namespace="user-123", session_id="conversation-456" + api_key="mock_key", user_id="user-123", session_id="conversation-456" ) client = _MockSupermemoryClient() service._supermemory_client = client @@ -231,7 +293,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): fact = "User likes machine learning projects" service = SupermemoryPipecatService( api_key="mock_key", - namespace="user-123", + user_id="user-123", session_id="conversation-456", params=SupermemoryPipecatService.InputParams(mode="query"), ) @@ -264,7 +326,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): fact = "User's repo is at C:" + backslash + "Users" + backslash + "alice" service = SupermemoryPipecatService( api_key="mock_key", - namespace="user-123", + user_id="user-123", session_id="conversation-456", params=SupermemoryPipecatService.InputParams(inject_mode="system"), ) From 2e94c8de1eadbee2576ff06a1a8823c076b01926 Mon Sep 17 00:00:00 2001 From: Sreeram Sreedhar Date: Mon, 2 Nov 2026 23:03:34 -0700 Subject: [PATCH 3/5] refactor(pipecat): simplify v5 retrieval and tests --- .../src/supermemory_pipecat/service.py | 34 +- packages/pipecat-sdk-python/tests/conftest.py | 96 ------ .../tests/test_empty_profile.py | 313 ++---------------- .../tests/test_public_exports.py | 4 - 4 files changed, 42 insertions(+), 405 deletions(-) delete mode 100644 packages/pipecat-sdk-python/tests/conftest.py diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py index 3b71243f..82ffa391 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py @@ -227,35 +227,21 @@ class SupermemoryPipecatService(FrameProcessor): ) try: - include_profile = self.params.mode != "query" - include_search = self.params.mode != "profile" and bool(query) - profile_response = None - search_response = None - - if include_profile and include_search: - profile_response, search_response = await asyncio.gather( - self._supermemory_client.profile(self.container_tag), - self._supermemory_client.search( - self.container_tag, - query=query, - search_mode="memories", - limit=self.params.search_limit, - threshold=self.params.search_threshold, - ), - ) - elif include_profile: - profile_response = await self._supermemory_client.profile(self.container_tag) - elif include_search: - search_response = await self._supermemory_client.search( - self.container_tag, + calls: Dict[str, Any] = {} + if self.params.mode != "query": + calls["profile"] = self._supermemory_client.profile(self.user_id) + if self.params.mode != "profile" and query: + calls["search"] = self._supermemory_client.search( + self.user_id, query=query, search_mode="memories", limit=self.params.search_limit, threshold=self.params.search_threshold, ) + responses = dict(zip(calls, await asyncio.gather(*calls.values()))) - profile = _field(profile_response, "profile") - search_results = list(_field(search_response, "results", default=[]) or []) + profile = _field(responses.get("profile"), "profile") + search_results = list(_field(responses.get("search"), "results", default=[]) or []) return { "profile": { @@ -293,7 +279,7 @@ class SupermemoryPipecatService(FrameProcessor): if self.session_id: add_params["id"] = self.session_id - await self._supermemory_client.add(self.container_tag, **add_params) + await self._supermemory_client.add(self.user_id, **add_params) except Exception as e: raise MemoryStorageError("Failed to store messages", e) from e diff --git a/packages/pipecat-sdk-python/tests/conftest.py b/packages/pipecat-sdk-python/tests/conftest.py deleted file mode 100644 index 00f98566..00000000 --- a/packages/pipecat-sdk-python/tests/conftest.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Import stubs for heavy optional deps; no-ops in CI where the real ones are installed.""" - -from __future__ import annotations - -import sys -import types -from importlib.util import find_spec - - -def _install_test_stubs() -> None: - if "loguru" not in sys.modules and find_spec("loguru") is None: - 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 and find_spec("pydantic") is None: - 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 and find_spec("pipecat") is None: - 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_empty_profile.py b/packages/pipecat-sdk-python/tests/test_empty_profile.py index d7e21c24..4a4a1343 100644 --- a/packages/pipecat-sdk-python/tests/test_empty_profile.py +++ b/packages/pipecat-sdk-python/tests/test_empty_profile.py @@ -1,129 +1,29 @@ from __future__ import annotations -import sys -import types +import json import unittest -from importlib.util import find_spec from types import SimpleNamespace from unittest.mock import AsyncMock - -def _install_test_stubs() -> None: - if "loguru" not in sys.modules and find_spec("loguru") is None: - 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 and find_spec("pydantic") is None: - 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 and find_spec("pipecat") is None: - 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() +import httpx +from supermemory import AsyncSupermemory from supermemory_pipecat.service import SupermemoryPipecatService -from supermemory_pipecat.exceptions import ConfigurationError -class _MockSupermemoryClient: - def __init__(self, profile_response=None, search_response=None): - self.profile = AsyncMock(return_value=profile_response) - self.search = AsyncMock(return_value=search_response) - self.add = AsyncMock() +class _Context: + def __init__(self, messages): + self.messages = messages + + def get_messages(self): + return self.messages + + def add_message(self, message): + self.messages.append(message) -class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): - def test_user_id_and_legacy_attribute_are_preserved(self) -> None: - service = SupermemoryPipecatService(api_key="mock_key", user_id="user-123") - - self.assertEqual(service.user_id, "user-123") - self.assertEqual(service.container_tag, "user-123") - - def test_empty_user_id_is_rejected(self) -> None: - with self.assertRaisesRegex(ConfigurationError, "^user_id is required$"): - SupermemoryPipecatService(api_key="mock_key", user_id="") - - async def test_real_v5_client_serializes_user_id_as_namespace(self) -> None: - import json - - import httpx - from supermemory import AsyncSupermemory - +class TestSupermemoryPipecatService(unittest.IsolatedAsyncioTestCase): + async def test_v5_reads_and_writes_use_user_id_as_namespace(self) -> None: requests = {} def respond(request): @@ -145,148 +45,27 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): api_key="mock_key", user_id="user-123", session_id="conversation-456" ) service._supermemory_client = client - result = await service._retrieve_memories("Where am I going?") await service._store_messages([{"role": "user", "content": "Hello"}]) - self.assertEqual(result["profile"], {"static": ["Likes tea"], "dynamic": []}) + self.assertEqual(result["profile"]["static"], ["Likes tea"]) self.assertEqual(result["search_results"][0].memory, "Plans a trip") self.assertEqual( - requests, - { - "/ns/user-123/profile": {}, - "/ns/user-123/search": { - "query": "Where am I going?", - "searchMode": "memories", - "limit": 10, - "threshold": 0.1, - }, - "/ns/user-123/document": { - "content": '[{"role": "user", "content": "Hello"}]', - "metadata": {"platform": "pipecat"}, - "id": "conversation-456", - }, - }, + 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/document"]["id"], "conversation-456") async def test_retrieve_memories_handles_null_profile(self) -> None: service = SupermemoryPipecatService(api_key="mock_key", user_id="new_user_123") - - client = _MockSupermemoryClient( - profile_response=SimpleNamespace(profile=None), - search_response=SimpleNamespace(results=[]), + service._supermemory_client = SimpleNamespace( + profile=AsyncMock(return_value=SimpleNamespace(profile=None)), + search=AsyncMock(return_value=SimpleNamespace(results=[])), ) - service._supermemory_client = client - - result = await service._retrieve_memories("Hello world") self.assertEqual( - result, - { - "profile": {"static": [], "dynamic": []}, - "search_results": [], - }, - ) - client.profile.assert_awaited_once_with("new_user_123") - client.search.assert_awaited_once_with( - "new_user_123", - query="Hello world", - search_mode="memories", - limit=10, - threshold=0.1, - ) - - async def test_full_mode_reads_v5_profile_objects_and_search_results(self) -> None: - service = SupermemoryPipecatService(api_key="mock_key", user_id="user-123") - client = _MockSupermemoryClient( - profile_response=SimpleNamespace( - profile=SimpleNamespace( - static=[SimpleNamespace(id="mem_1", memory="Likes tea")], - dynamic=[{"id": "mem_2", "memory": "[2026-11-02] Traveling"}], - ) - ), - search_response=SimpleNamespace( - results=[SimpleNamespace(id="mem_3", memory="Plans a trip")] - ), - ) - service._supermemory_client = client - - result = await service._retrieve_memories("Where am I going?") - - self.assertEqual( - result["profile"], - { - "static": ["Likes tea"], - "dynamic": ["[2026-11-02] Traveling"], - }, - ) - self.assertEqual(result["search_results"][0].memory, "Plans a trip") - client.profile.assert_awaited_once_with("user-123") - client.search.assert_awaited_once_with( - "user-123", - query="Where am I going?", - search_mode="memories", - limit=10, - threshold=0.1, - ) - - async def test_query_mode_only_searches(self) -> None: - service = SupermemoryPipecatService( - api_key="mock_key", - user_id="user-123", - params=SupermemoryPipecatService.InputParams( - mode="query", search_limit=3, search_threshold=0.5 - ), - ) - client = _MockSupermemoryClient(search_response=SimpleNamespace(results=[])) - service._supermemory_client = client - - result = await service._retrieve_memories("Where am I going?") - - self.assertEqual(result["profile"], {"static": [], "dynamic": []}) - client.profile.assert_not_awaited() - client.search.assert_awaited_once_with( - "user-123", - query="Where am I going?", - search_mode="memories", - limit=3, - threshold=0.5, - ) - - async def test_profile_mode_only_reads_profile(self) -> None: - service = SupermemoryPipecatService( - api_key="mock_key", - user_id="user-123", - params=SupermemoryPipecatService.InputParams(mode="profile"), - ) - client = _MockSupermemoryClient( - profile_response=SimpleNamespace( - profile=SimpleNamespace(static=[{"memory": "Likes tea"}], dynamic=[]) - ) - ) - service._supermemory_client = client - - result = await service._retrieve_memories("Hello") - - self.assertEqual(result["profile"]["static"], ["Likes tea"]) - self.assertEqual(result["search_results"], []) - client.profile.assert_awaited_once_with("user-123") - client.search.assert_not_awaited() - - async def test_store_uses_namespace_and_document_id(self) -> None: - service = SupermemoryPipecatService( - api_key="mock_key", user_id="user-123", session_id="conversation-456" - ) - client = _MockSupermemoryClient() - service._supermemory_client = client - - await service._store_messages([{"role": "user", "content": "Hello"}]) - - client.add.assert_awaited_once_with( - "user-123", - content='[{"role": "user", "content": "Hello"}]', - metadata={"platform": "pipecat"}, - id="conversation-456", + await service._retrieve_memories("Hello world"), + {"profile": {"static": [], "dynamic": []}, "search_results": []}, ) def test_query_mode_keeps_search_fact_also_present_in_profile(self) -> None: @@ -297,18 +76,7 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): session_id="conversation-456", params=SupermemoryPipecatService.InputParams(mode="query"), ) - - class Context: - def __init__(self): - self.messages = [{"role": "user", "content": "What do I like?"}] - - def get_messages(self): - return self.messages - - def add_message(self, message): - self.messages.append(message) - - context = Context() + context = _Context([{"role": "user", "content": "What do I like?"}]) service._enhance_context_with_memories( context, "What do I like?", @@ -321,7 +89,6 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): self.assertTrue(any(fact in message.get("content", "") for message in context.messages)) def test_system_injection_replaces_stale_block_with_backslash_memory(self) -> None: - """A memory containing backslashes must not be read as a regex template.""" backslash = chr(92) fact = "User's repo is at C:" + backslash + "Users" + backslash + "alice" service = SupermemoryPipecatService( @@ -330,33 +97,17 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase): session_id="conversation-456", params=SupermemoryPipecatService.InputParams(inject_mode="system"), ) - newline = chr(10) - stale = newline.join(["", "Stale fact", ""]) - - class Context: - def __init__(self): - self.messages = [ - { - "role": "system", - "content": "You are helpful." + newline + newline + stale, - }, - {"role": "user", "content": "Where is my repo?"}, - ] - - def get_messages(self): - return self.messages - - def add_message(self, message): - self.messages.append(message) - - context = Context() + stale = "\nStale fact\n" + context = _Context( + [ + {"role": "system", "content": "You are helpful.\n\n" + stale}, + {"role": "user", "content": "Where is my repo?"}, + ] + ) service._enhance_context_with_memories( context, "Where is my repo?", - { - "profile": {"static": [fact], "dynamic": []}, - "search_results": [], - }, + {"profile": {"static": [fact], "dynamic": []}, "search_results": []}, ) system_content = context.messages[0]["content"] diff --git a/packages/pipecat-sdk-python/tests/test_public_exports.py b/packages/pipecat-sdk-python/tests/test_public_exports.py index b20cb707..e7941192 100644 --- a/packages/pipecat-sdk-python/tests/test_public_exports.py +++ b/packages/pipecat-sdk-python/tests/test_public_exports.py @@ -4,10 +4,6 @@ from __future__ import annotations import unittest -from .test_empty_profile import _install_test_stubs - -_install_test_stubs() - import supermemory_pipecat from supermemory_pipecat import InputParams, SupermemoryPipecatService From 4a5804f5a5817beff0d3e40c2c361e94ec6d7ee4 Mon Sep 17 00:00:00 2001 From: Sreeram Sreedhar Date: Wed, 4 Nov 2026 12:06:24 -0700 Subject: [PATCH 4/5] 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") From a9b23e060306e25fd97e7f888bdaa4c16ef8f83a Mon Sep 17 00:00:00 2001 From: Sreeram Sreedhar Date: Wed, 4 Nov 2026 12:28:26 -0700 Subject: [PATCH 5/5] docs(pipecat): update context aggregator example --- apps/docs/integrations/pipecat.mdx | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/apps/docs/integrations/pipecat.mdx b/apps/docs/integrations/pipecat.mdx index ae189343..a3717476 100644 --- a/apps/docs/integrations/pipecat.mdx +++ b/apps/docs/integrations/pipecat.mdx @@ -147,11 +147,11 @@ from fastapi import FastAPI, WebSocket from fastapi.middleware.cors import CORSMiddleware from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.pipeline.task import PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.serializers.protobuf import ProtobufFrameSerializer from pipecat.services.openai.llm import OpenAILLMService from pipecat.services.openai.tts import OpenAITTSService @@ -198,8 +198,8 @@ async def run_bot(websocket_client, user_id: str, session_id: str): ), ) - context = OpenAILLMContext([{"role": "system", "content": SYSTEM_PROMPT}]) - context_aggregator = llm.create_context_aggregator(context) + context = LLMContext([{"role": "system", "content": SYSTEM_PROMPT}]) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline([ transport.input(), @@ -212,7 +212,7 @@ async def run_bot(websocket_client, user_id: str, session_id: str): context_aggregator.assistant(), ]) - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + task = PipelineTask(pipeline) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client):