scriptoriumadmin/supermemory
1
0
Fork 0
mirror of https://github.com/supermemoryai/supermemory.git synced 2026-11-14 09:01:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat(pipecat): migrate SDK integration to Supermemory v5

Browse source
This commit is contained in:
Sreeram Sreedhar 2026-11-02 19:41:19 -07:00
parent 16ed414e0e
commit cbf55e59a8
10 changed files with 202 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
apps/docs/integrations/pipecat.mdx
View file

@ -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__":

6
packages/pipecat-sdk-python/Agents.md
View file

@ -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

19
packages/pipecat-sdk-python/README.md
View file

@ -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",
)

4
packages/pipecat-sdk-python/pyproject.toml
View file

@ -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",
]

4
packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py
View file

@ -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

72
packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
View file

@ -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

5
packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py
View file

@ -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:

11
packages/pipecat-sdk-python/tests/conftest.py
View file

@ -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

135
packages/pipecat-sdk-python/tests/test_empty_profile.py
View file

@ -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"),
)

2
packages/pipecat-sdk-python/tests/test_utils.py
View file

@ -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(