mirror of
https://github.com/supermemoryai/supermemory.git
synced 2026-11-14 09:01:45 +00:00
refactor(pipecat): simplify v5 retrieval and tests
This commit is contained in:
Sreeram Sreedhar
parent
c0320ca833
commit
2e94c8de1e
4 changed files with 42 additions and 405 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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(["<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
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue