Merge pull request #4156 from pipecat-ai/mb/mem0-improvements

fix(mem0): improve Mem0 service reliability and add get_memories() method

Author: markbackman

changelog/4156.added.md

@@ -0,0 +1 @@
+- Added `Mem0MemoryService.get_memories()` convenience method for retrieving all stored memories outside the pipeline (e.g. to build a personalized greeting at connection time). This avoids the need to manually handle client type branching, filter construction, and async wrapping.

changelog/4156.changed.md

@@ -0,0 +1 @@
+- ⚠️ Bumped `mem0ai` dependency from `~=0.1.94` to `>=1.0.8,<2`. Users of the `mem0` extra will need to update their mem0ai package.

changelog/4156.fixed.2.md

@@ -0,0 +1 @@
+- Fixed `Mem0MemoryService` failing to store messages when the context contained system or developer role messages. The Mem0 API only accepts user and assistant roles, so other roles are now filtered out before storing.

changelog/4156.fixed.md

@@ -0,0 +1 @@
+- `Mem0MemoryService` no longer blocks the event loop during memory storage and retrieval. All Mem0 API calls now run in a background thread, and message storage is fire-and-forget so it doesn't delay downstream processing.

examples/foundational/37-mem0.py

@@ -42,7 +42,6 @@
 """
 
 import os
-from typing import Union
 
 from dotenv import load_dotenv
 from loguru import logger
@@ -69,58 +68,35 @@
 
 load_dotenv(override=True)
 
-try:
-    from mem0 import Memory, MemoryClient  # noqa: F401
-except ModuleNotFoundError as e:
-    logger.error(f"Exception: {e}")
-    logger.error(
-        "In order to use Mem0, you need to `pip install mem0ai`. Also, set the environment variable MEM0_API_KEY."
-    )
-    raise Exception(f"Missing module: {e}")
-
 
-async def get_initial_greeting(
-    memory_client: Union[MemoryClient, Memory], user_id: str, agent_id: str, run_id: str
-) -> str:
+async def get_initial_greeting(memory_service: Mem0MemoryService) -> str:
     """Fetch all memories for the user and create a personalized greeting.
 
+    Args:
+        memory_service: The Mem0 memory service instance.
+
     Returns:
-        A personalized greeting based on user memories
+        A personalized greeting based on user memories.
     """
     try:
-        if isinstance(memory_client, Memory):
-            filters = {"user_id": user_id, "agent_id": agent_id, "run_id": run_id}
-            filters = {k: v for k, v in filters.items() if v is not None}
-            memories = memory_client.get_all(**filters)
-        else:
-            # Create filters based on available IDs
-            id_pairs = [("user_id", user_id), ("agent_id", agent_id), ("run_id", run_id)]
-            clauses = [{name: value} for name, value in id_pairs if value is not None]
-            filters = {"AND": clauses} if clauses else {}
-
-            # Get all memories for this user
-            memories = memory_client.get_all(filters=filters, version="v2", output_format="v1.1")
-
-        if not memories or len(memories) == 0:
-            logger.debug(f"!!! No memories found for this user. {memories}")
+        results = await memory_service.get_memories()
+        if not results:
+            logger.debug("No memories found for this user.")
             return "Hello! It's nice to meet you. How can I help you today?"
 
         # Create a personalized greeting based on memories
         greeting = "Hello! It's great to see you again. "
+        greeting += "Based on our previous conversations, I remember: "
+        for i, memory in enumerate(results[:3], 1):
+            memory_content = memory.get("memory", "")
+            # Keep memory references brief
+            if len(memory_content) > 100:
+                memory_content = memory_content[:97] + "..."
+            greeting += f"{memory_content} "
 
-        # Add some personalization based on memories (limit to 3 memories for brevity)
-        if len(memories) > 0:
-            greeting += "Based on our previous conversations, I remember: "
-            for i, memory in enumerate(memories["results"][:3], 1):
-                memory_content = memory.get("memory", "")
-                # Keep memory references brief
-                if len(memory_content) > 100:
-                    memory_content = memory_content[:97] + "..."
-                greeting += f"{memory_content} "
+        greeting += "How can I help you today?"
 
-            greeting += "How can I help you today?"
-
-        logger.debug(f"Created personalized greeting from {len(memories)} memories")
+        logger.debug(f"Created personalized greeting from {len(results)} memories")
         return greeting
 
     except Exception as e:
@@ -265,23 +241,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
     )
 
-    @task.rtvi.event_handler("on_client_ready")
-    async def on_client_ready(rtvi):
-        # Get personalized greeting based on user memories. Can pass agent_id and run_id as per requirement of the application to manage short term memory or agent specific memory.
-        greeting = await get_initial_greeting(
-            memory_client=memory.memory_client, user_id=USER_ID, agent_id=None, run_id=None
-        )
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Get personalized greeting based on user memories
+        greeting = await get_initial_greeting(memory)
 
         # Add the greeting as an assistant message to start the conversation
-        context.add_message({"role": "assistant", "content": greeting})
+        context.add_message({"role": "developer", "content": greeting})
 
         # Queue the context frame to start the conversation
         await task.queue_frames([LLMRunFrame()])
 
-    @transport.event_handler("on_client_connected")
-    async def on_client_connected(transport, client):
-        logger.info(f"Client connected")
-
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(transport, client):
         logger.info(f"Client disconnected")

pyproject.toml

@@ -88,7 +88,7 @@ lmnt = [ "pipecat-ai[websockets-base]" ]
 local = [ "pyaudio~=0.2.14" ]
 local-smart-turn = [ "coremltools>=8.0", "transformers>=4.48.0,<6", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ]
 mcp = [ "mcp[cli]>=1.11.0,<2" ]
-mem0 = [ "mem0ai~=0.1.94" ]
+mem0 = [ "mem0ai>=1.0.8,<2" ]
 mistral = []
 mlx-whisper = [ "mlx-whisper~=0.4.2" ]
 moondream = [ "accelerate~=1.10.0", "einops~=0.8.0", "pyvips[binary]~=3.0.0", "timm~=1.0.13", "transformers>=4.48.0,<6" ]

src/pipecat/services/mem0/memory.py

@@ -11,6 +11,7 @@
 historical information.
 """
 
+import asyncio
 from typing import Any, Dict, List, Optional
 
 from loguru import logger
@@ -112,9 +113,51 @@ def __init__(
         self.last_query = None
         logger.info(f"Initialized Mem0MemoryService with {user_id=}, {agent_id=}, {run_id=}")
 
-    def _store_messages(self, messages: List[Dict[str, Any]]):
+    async def get_memories(self) -> List[Dict[str, Any]]:
+        """Retrieve all stored memories for the configured user/agent/run IDs.
+
+        This is a convenience method for accessing memories outside the pipeline,
+        e.g. to build a personalized greeting at connection time. It wraps the
+        blocking Mem0 ``get_all()`` call in a background thread.
+
+        Returns:
+            List of memory dictionaries. Each dict contains at least a
+            ``"memory"`` key with the memory text. Returns an empty list on
+            error.
+        """
+        try:
+            if isinstance(self.memory_client, Memory):
+                params = {
+                    "user_id": self.user_id,
+                    "agent_id": self.agent_id,
+                    "run_id": self.run_id,
+                }
+                params = {k: v for k, v in params.items() if v is not None}
+                memories = await asyncio.to_thread(lambda: self.memory_client.get_all(**params))
+            else:
+                id_pairs = [
+                    ("user_id", self.user_id),
+                    ("agent_id", self.agent_id),
+                    ("run_id", self.run_id),
+                ]
+                clauses = [{name: value} for name, value in id_pairs if value is not None]
+                filters = {"OR": clauses} if clauses else {}
+                memories = await asyncio.to_thread(
+                    lambda: self.memory_client.get_all(filters=filters)
+                )
+
+            results = memories.get("results", []) if isinstance(memories, dict) else memories
+            return results
+        except Exception as e:
+            logger.error(f"Error retrieving memories from Mem0: {e}")
+            return []
+
+    async def _store_messages(self, messages: List[Dict[str, Any]]):
         """Store messages in Mem0.
 
+        Runs the blocking Mem0 API call in a background thread to avoid
+        blocking the event loop.
+
         Args:
             messages: List of message dictionaries to store in memory.
         """
@@ -131,14 +174,16 @@ def _store_messages(self, messages: List[Dict[str, Any]]):
 
             if isinstance(self.memory_client, Memory):
                 del params["output_format"]
-            # Note: You can run this in background to avoid blocking the conversation
-            self.memory_client.add(**params)
+            await asyncio.to_thread(lambda: self.memory_client.add(**params))
         except Exception as e:
             logger.error(f"Error storing messages in Mem0: {e}")
 
-    def _retrieve_memories(self, query: str) -> List[Dict[str, Any]]:
+    async def _retrieve_memories(self, query: str) -> List[Dict[str, Any]]:
         """Retrieve relevant memories from Mem0.
 
+        Runs the blocking Mem0 API call in a background thread to avoid
+        blocking the event loop.
+
         Args:
             query: The query to search for relevant memories.
 
@@ -156,7 +201,7 @@ def _retrieve_memories(self, query: str) -> List[Dict[str, Any]]:
                     "limit": self.search_limit,
                 }
                 params = {k: v for k, v in params.items() if v is not None}
-                results = self.memory_client.search(**params)
+                results = await asyncio.to_thread(lambda: self.memory_client.search(**params))
             else:
                 id_pairs = [
                     ("user_id", self.user_id),
@@ -165,13 +210,15 @@ def _retrieve_memories(self, query: str) -> List[Dict[str, Any]]:
                 ]
                 clauses = [{name: value} for name, value in id_pairs if value is not None]
                 filters = {"OR": clauses} if clauses else {}
-                results = self.memory_client.search(
-                    query=query,
-                    filters=filters,
-                    version=self.api_version,
-                    top_k=self.search_limit,
-                    threshold=self.search_threshold,
-                    output_format="v1.1",
+                results = await asyncio.to_thread(
+                    lambda: self.memory_client.search(
+                        query=query,
+                        filters=filters,
+                        version=self.api_version,
+                        top_k=self.search_limit,
+                        threshold=self.search_threshold,
+                        output_format="v1.1",
+                    )
                 )
 
             logger.debug(f"Retrieved {len(results)} memories from Mem0")
@@ -180,7 +227,9 @@ def _retrieve_memories(self, query: str) -> List[Dict[str, Any]]:
             logger.error(f"Error retrieving memories from Mem0: {e}")
             return []
 
-    def _enhance_context_with_memories(self, context: LLMContext | OpenAILLMContext, query: str):
+    async def _enhance_context_with_memories(
+        self, context: LLMContext | OpenAILLMContext, query: str
+    ):
         """Enhance the LLM context with relevant memories.
 
         Args:
@@ -193,7 +242,7 @@ def _enhance_context_with_memories(self, context: LLMContext | OpenAILLMContext,
 
         self.last_query = query
 
-        memories = self._retrieve_memories(query)
+        memories = await self._retrieve_memories(query)
         if not memories:
             return
 
@@ -203,11 +252,14 @@ def _enhance_context_with_memories(self, context: LLMContext | OpenAILLMContext,
             memory_text += f"{i}. {memory.get('memory', '')}\n\n"
 
         # Add memories as a system message or user message based on configuration
-        if self.add_as_system_message:
-            context.add_message({"role": "system", "content": memory_text})
-        else:
-            # Add as a user message that provides context
-            context.add_message({"role": "user", "content": memory_text})
+        role = "system" if self.add_as_system_message else "user"
+        memory_message = {"role": role, "content": memory_text}
+
+        messages = context.get_messages()
+        position = max(0, min(self.position, len(messages)))
+        messages.insert(position, memory_message)
+        context.set_messages(messages)
+
         logger.debug(f"Enhanced context with {len(memories)} memories")
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -240,10 +292,15 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
                         break
 
                 if latest_user_message:
+                    # Filter to only user/assistant messages — Mem0 API
+                    # doesn't accept other roles (system, developer, etc.)
+                    messages_to_store = [
+                        m for m in context_messages if m.get("role") in ("user", "assistant")
+                    ]
                     # Enhance context with memories before passing it downstream
-                    self._enhance_context_with_memories(context, latest_user_message)
-                    # Store the conversation in Mem0. Only call this when user message is detected
-                    self._store_messages(context_messages)
+                    await self._enhance_context_with_memories(context, latest_user_message)
+                    # Store the conversation in Mem0 as a background task
+                    self.create_task(self._store_messages(messages_to_store), name="mem0_store")
 
                 # If we received an LLMMessagesFrame, create a new one with the enhanced messages
                 if messages is not None: