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(