mirror of
https://github.com/supermemoryai/supermemory.git
synced 2026-11-14 09:01:45 +00:00
Merge a9b23e0603 into 16ed414e0e
This commit is contained in:
Sreeram
GitHub
commit
258accd482
11 changed files with 165 additions and 292 deletions
|
|
@ -15,6 +15,10 @@ To use Supermemory with Pipecat, install the required dependencies:
|
|||
pip install supermemory-pipecat
|
||||
```
|
||||
|
||||
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:
|
||||
|
||||
```bash
|
||||
|
|
@ -66,7 +70,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
|
||||
|
|
@ -142,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
|
||||
|
|
@ -193,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(),
|
||||
|
|
@ -207,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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
user_id="user-123", # Required: scopes all memory calls
|
||||
session_id="session-456", # Optional: groups conversations
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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. 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
|
||||
|
||||
```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
|
||||
user_id="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 |
|
||||
| `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 |
|
||||
|
|
@ -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" }
|
||||
}
|
||||
```
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -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.
|
||||
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 is missing or user_id not provided.
|
||||
ConfigurationError: If API key or user_id is missing.
|
||||
"""
|
||||
super().__init__()
|
||||
|
||||
|
|
@ -226,25 +227,44 @@ 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}
|
||||
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:
|
||||
kwargs["q"] = query
|
||||
kwargs["threshold"] = self.params.search_threshold
|
||||
calls["search"] = self._supermemory_client.search(
|
||||
self.user_id,
|
||||
query=query,
|
||||
search_mode="memories",
|
||||
limit=min(self.params.search_limit, 100),
|
||||
threshold=self.params.search_threshold,
|
||||
)
|
||||
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]
|
||||
|
||||
response = await self._supermemory_client.profile(**kwargs)
|
||||
|
||||
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(responses.get("profile"), "profile")
|
||||
search_results = list(_field(responses.get("search"), "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 +284,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.user_id, **add_params)
|
||||
|
||||
except Exception as e:
|
||||
raise MemoryStorageError("Failed to store messages", e) from e
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -71,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 Stainless-generated response models.
|
||||
"""
|
||||
"""Accept alternative field names across raw JSON and SDK response models."""
|
||||
if item is None:
|
||||
return default
|
||||
if isinstance(item, dict):
|
||||
|
|
@ -135,7 +130,6 @@ 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`.
|
||||
memory = _memory_text(r)
|
||||
key = comparison_key(memory)
|
||||
if key and key not in seen:
|
||||
|
|
@ -184,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}")
|
||||
|
|
|
|||
|
|
@ -1,97 +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
|
||||
|
||||
|
||||
def _install_test_stubs() -> None:
|
||||
if "loguru" not in sys.modules:
|
||||
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:
|
||||
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:
|
||||
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()
|
||||
|
|
@ -1,125 +1,106 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import types
|
||||
import json
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
|
||||
def _install_test_stubs() -> None:
|
||||
if "loguru" not in sys.modules:
|
||||
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:
|
||||
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:
|
||||
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.utils import format_memories_to_text
|
||||
|
||||
|
||||
class _MockSupermemoryClient:
|
||||
def __init__(self, response):
|
||||
self.profile = AsyncMock(return_value=response)
|
||||
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):
|
||||
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": [
|
||||
{
|
||||
"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)
|
||||
|
||||
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",
|
||||
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")
|
||||
self.assertEqual(
|
||||
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")
|
||||
|
||||
response = SimpleNamespace(profile=None, search_results=None)
|
||||
service._supermemory_client = _MockSupermemoryClient(response)
|
||||
|
||||
result = await service._retrieve_memories("Hello world")
|
||||
service._supermemory_client = SimpleNamespace(
|
||||
profile=AsyncMock(return_value=SimpleNamespace(profile=None)),
|
||||
search=AsyncMock(return_value=SimpleNamespace(results=[])),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
result,
|
||||
{
|
||||
"profile": {"static": [], "dynamic": []},
|
||||
"search_results": [],
|
||||
},
|
||||
await service._retrieve_memories("Hello world"),
|
||||
{"profile": {"static": [], "dynamic": []}, "search_results": []},
|
||||
)
|
||||
|
||||
def test_query_mode_keeps_search_fact_also_present_in_profile(self) -> None:
|
||||
|
|
@ -130,18 +111,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?",
|
||||
|
|
@ -151,12 +121,9 @@ 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."""
|
||||
backslash = chr(92)
|
||||
fact = "User's repo is at C:" + backslash + "Users" + backslash + "alice"
|
||||
service = SupermemoryPipecatService(
|
||||
|
|
@ -165,33 +132,17 @@ class TestSupermemoryPipecatNullProfile(unittest.IsolatedAsyncioTestCase):
|
|||
session_id="conversation-456",
|
||||
params=SupermemoryPipecatService.InputParams(inject_mode="system"),
|
||||
)
|
||||
newline = chr(10)
|
||||
stale = newline.join(["<user_memories>", "Stale fact", "</user_memories>"])
|
||||
|
||||
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 = "<user_memories>\nStale fact\n</user_memories>"
|
||||
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"]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue