scriptoriumadmin/supermemory
1
0
Fork 0
mirror of https://github.com/supermemoryai/supermemory.git synced 2026-09-05 08:06:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

added delta message tracking

Browse source
This commit is contained in:
Prasanna721 2026-01-10 07:25:40 -08:00
parent 340f65efe5
commit 161c698472
2 changed files with 70 additions and 151 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -8,7 +8,7 @@ historical information.
import asyncio
import json
import os
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Literal, Optional
from loguru import logger
from pydantic import BaseModel, Field
@ -68,17 +68,13 @@ class SupermemoryPipecatService(FrameProcessor):
search_limit: Maximum number of memories to retrieve per query.
search_threshold: Minimum similarity threshold for memory retrieval.
system_prompt: Prefix text for memory context messages.
add_as_system_message: Whether to add memories as system messages.
position: Position to insert memory messages in context.
mode: Memory retrieval mode - "profile", "query", or "full".
"""
search_limit: int = Field(default=10, ge=1)
search_threshold: float = Field(default=0.1, ge=0.0, le=1.0)
system_prompt: str = Field(default="Based on previous conversations, I recall:\n\n")
add_as_system_message: bool = Field(default=True)
position: int = Field(default=1)
mode: str = Field(default="full") # "profile", "query", "full"
mode: Literal["profile", "query", "full"] = Field(default="full")
def __init__(
self,
@ -132,8 +128,8 @@ class SupermemoryPipecatService(FrameProcessor):
except Exception as e:
logger.warning(f"Failed to initialize Supermemory client: {e}")
# Track conversation history separately (clean, no injected memories)
self._conversation_history: List[Dict[str, str]] = []
# Track how many messages we've already sent to memory
self._messages_sent_count: int = 0
# Track last query to avoid duplicate processing
self._last_query: Optional[str] = None
@ -175,21 +171,23 @@ class SupermemoryPipecatService(FrameProcessor):
# Use SDK's profile method
response = await self._supermemory_client.profile(**kwargs)
# Convert SDK response to dict format expected by rest of code
# Extract memory strings from SDK response
search_results = []
if response.search_results and response.search_results.results:
search_results = [r["memory"] for r in response.search_results.results]
data: Dict[str, Any] = {
"profile": {
"static": response.profile.static,
"dynamic": response.profile.dynamic,
},
"searchResults": {
"results": response.search_results.results if response.search_results else [],
},
"search_results": search_results,
}
logger.debug(
f"Retrieved memories - static: {len(data['profile']['static'])}, "
f"dynamic: {len(data['profile']['dynamic'])}, "
f"search: {len(data['searchResults']['results'])}"
f"search: {len(data['search_results'])}"
)
return data
@ -197,24 +195,24 @@ class SupermemoryPipecatService(FrameProcessor):
logger.error(f"Error retrieving memories: {e}")
raise MemoryRetrievalError("Failed to retrieve memories", e)
async def _store_message(self, message: Dict[str, str]) -> None:
"""Store a single message in Supermemory.
async def _store_messages(self, messages: List[Dict[str, Any]]) -> None:
"""Store messages in Supermemory.
Args:
message: Message dict with 'role' and 'content' keys.
messages: List of message dicts with 'role' and 'content' keys.
"""
if self._supermemory_client is None:
logger.warning("Supermemory client not initialized, skipping memory storage")
return
if not messages:
return
try:
content = message.get("content", "")
if not content or not isinstance(content, str):
return
# Format messages as JSON array
formatted_content = json.dumps(messages)
formatted_content = json.dumps(message)
logger.debug(f"Storing message to Supermemory: {formatted_content[:100]}...")
logger.debug(f"Storing {len(messages)} messages to Supermemory")
# Build storage params
add_params: Dict[str, Any] = {
@ -223,14 +221,14 @@ class SupermemoryPipecatService(FrameProcessor):
"metadata": {"platform": "pipecat"},
}
if self.session_id:
add_params["custom_id"] = self.session_id
add_params["custom_id"] = f"{self.session_id}"
await self._supermemory_client.memories.add(**add_params)
logger.debug("Successfully stored message in Supermemory")
logger.debug(f"Successfully stored {len(messages)} messages in Supermemory")
except Exception as e:
# Don't fail the pipeline on storage errors
logger.error(f"Error storing message in Supermemory: {e}")
logger.error(f"Error storing messages in Supermemory: {e}")
def _enhance_context_with_memories(
self,
@ -252,13 +250,12 @@ class SupermemoryPipecatService(FrameProcessor):
self._last_query = query
# Extract and deduplicate memories
profile = memories_data.get("profile", {})
search_results = memories_data.get("searchResults", {})
profile = memories_data["profile"]
deduplicated = deduplicate_memories(
static=profile.get("static", []),
dynamic=profile.get("dynamic", []),
search_results=search_results.get("results", []),
static=profile["static"],
dynamic=profile["dynamic"],
search_results=memories_data["search_results"],
)
# Check if we have any memories
@ -287,11 +284,8 @@ class SupermemoryPipecatService(FrameProcessor):
if not memory_text:
return
# Inject memories into context
if self.params.add_as_system_message:
context.add_message({"role": "system", "content": memory_text})
else:
context.add_message({"role": "user", "content": memory_text})
# Inject memories into context as user message
context.add_message({"role": "user", "content": memory_text})
logger.debug(f"Enhanced context with {total_memories} memories")
@ -318,35 +312,27 @@ class SupermemoryPipecatService(FrameProcessor):
try:
# Get messages from context
context_messages = context.get_messages()
# Find latest user message for memory query
latest_user_message = get_last_user_message(context_messages)
if latest_user_message:
# Track the user message in our conversation history (clean)
user_msg = {"role": "user", "content": latest_user_message}
# Only add if it's a new message (not already tracked)
if (
not self._conversation_history
or self._conversation_history[-1].get("content") != latest_user_message
):
self._conversation_history.append(user_msg)
# Retrieve memories from Supermemory
try:
memories_data = await self._retrieve_memories(latest_user_message)
# Enhance context with memories
self._enhance_context_with_memories(
context, latest_user_message, memories_data
)
except MemoryRetrievalError as e:
# Log but don't fail the pipeline
logger.warning(f"Memory retrieval failed, continuing without memories: {e}")
# Store the last user message (runs in background, non-blocking)
asyncio.create_task(self._store_message(user_msg))
# Store unsent messages (user and assistant only, skip system)
storable_messages = [
msg for msg in context_messages if msg["role"] in ("user", "assistant")
]
unsent_messages = storable_messages[self._messages_sent_count :]
if unsent_messages:
asyncio.create_task(self._store_messages(unsent_messages))
self._messages_sent_count = len(storable_messages)
# Pass the frame downstream
if messages is not None:
@ -364,15 +350,15 @@ class SupermemoryPipecatService(FrameProcessor):
# Non-context frames pass through unchanged
await self.push_frame(frame, direction)
def get_conversation_history(self) -> List[Dict[str, str]]:
"""Get the tracked conversation history (without injected memories).
def get_messages_sent_count(self) -> int:
"""Get the count of messages sent to memory.
Returns:
List of message dicts with 'role' and 'content'.
Number of messages already sent to Supermemory.
"""
return self._conversation_history.copy()
return self._messages_sent_count
def clear_conversation_history(self) -> None:
"""Clear the tracked conversation history."""
self._conversation_history.clear()
def reset_memory_tracking(self) -> None:
"""Reset memory tracking for a new conversation."""
self._messages_sent_count = 0
self._last_query = None

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

@ -1,91 +1,36 @@
"""Utility functions for Supermemory Pipecat integration."""
from typing import Any, Dict, List, Optional
from typing import Dict, List
def get_last_user_message(messages: List[Dict[str, Any]]) -> Optional[str]:
"""
Extract the last user message from a list of messages.
Args:
messages: List of message dictionaries with 'role' and 'content' keys
Returns:
The content of the last user message, or None if not found
"""
for message in reversed(messages):
if message.get("role") == "user":
content = message.get("content", "")
if isinstance(content, str):
return content
elif isinstance(content, list):
# Handle content that is an array of content parts
text_parts = []
for part in content:
if isinstance(part, dict) and part.get("type") == "text":
text_parts.append(part.get("text", ""))
elif isinstance(part, str):
text_parts.append(part)
return " ".join(text_parts)
def get_last_user_message(messages: List[Dict[str, str]]) -> str | None:
"""Extract the last user message content from a list of messages."""
for msg in reversed(messages):
if msg["role"] == "user":
return msg["content"]
return None
def deduplicate_memories(
static: Optional[List[Any]] = None,
dynamic: Optional[List[Any]] = None,
search_results: Optional[List[Any]] = None,
static: List[str],
dynamic: List[str],
search_results: List[str],
) -> Dict[str, List[str]]:
"""
Deduplicates memory items across sources.
Priority: Static > Dynamic > Search Results.
"""Deduplicate memories. Priority: static > dynamic > search."""
seen = set()
Args:
static: List of static profile memories
dynamic: List of dynamic profile memories
search_results: List of search result memories
Returns:
Dictionary with deduplicated 'static', 'dynamic', and 'search_results' lists
"""
static_items = static or []
dynamic_items = dynamic or []
search_items = search_results or []
def extract_memory_text(item: Any) -> Optional[str]:
if isinstance(item, dict):
item = item.get("memory")
if isinstance(item, str):
trimmed = item.strip()
return trimmed or None
return None
static_memories: List[str] = []
seen_memories: set = set()
for item in static_items:
memory = extract_memory_text(item)
if memory is not None:
static_memories.append(memory)
seen_memories.add(memory)
dynamic_memories: List[str] = []
for item in dynamic_items:
memory = extract_memory_text(item)
if memory is not None and memory not in seen_memories:
dynamic_memories.append(memory)
seen_memories.add(memory)
search_memories: List[str] = []
for item in search_items:
memory = extract_memory_text(item)
if memory is not None and memory not in seen_memories:
search_memories.append(memory)
seen_memories.add(memory)
def unique(memories):
out = []
for m in memories:
if m not in seen:
seen.add(m)
out.append(m)
return out
return {
"static": static_memories,
"dynamic": dynamic_memories,
"search_results": search_memories,
"static": unique(static),
"dynamic": unique(dynamic),
"search_results": unique(search_results),
}
@ -96,24 +41,12 @@ def format_memories_to_text(
include_dynamic: bool = True,
include_search: bool = True,
) -> str:
"""
Format deduplicated memories into a text string for injection.
Args:
memories: Dictionary with 'static', 'dynamic', 'search_results' lists
system_prompt: Prefix text for the memory content
include_static: Whether to include static profile memories
include_dynamic: Whether to include dynamic profile memories
include_search: Whether to include search result memories
Returns:
Formatted memory text string
"""
"""Format deduplicated memories into a text string for injection."""
sections = []
static = memories.get("static", [])
dynamic = memories.get("dynamic", [])
search_results = memories.get("search_results", [])
static = memories["static"]
dynamic = memories["dynamic"]
search_results = memories["search_results"]
if include_static and static:
sections.append("## User Profile (Persistent)")