From 161c698472f9aaeb5728a9729b13cfabc414882a Mon Sep 17 00:00:00 2001 From: Prasanna721 Date: Sat, 10 Jan 2026 07:25:40 -0800 Subject: [PATCH] added delta message tracking --- .../src/supermemory_pipecat/service.py | 104 +++++++--------- .../src/supermemory_pipecat/utils.py | 117 ++++-------------- 2 files changed, 70 insertions(+), 151 deletions(-) diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py index 153f8932..ab07e672 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py @@ -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 diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py index 7d76be87..d0f0e461 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py @@ -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)")