From 6a87d0e87d9a841e3263131c6b532aa5ba56cf77 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Mar 2026 12:42:36 -0400 Subject: [PATCH 1/5] fix(mem0): make memory service non-blocking and use position parameter Move blocking Mem0 API calls off the event loop using asyncio.to_thread(). Store messages as a fire-and-forget background task via create_task() since the result is not needed. Insert memory messages at the configured position in the context instead of always appending. Closes #1741 --- src/pipecat/services/mem0/memory.py | 57 ++++++++++++++++++----------- 1 file changed, 35 insertions(+), 22 deletions(-) diff --git a/src/pipecat/services/mem0/memory.py b/src/pipecat/services/mem0/memory.py index c4dfe16de..0b7f8e19f 100644 --- a/src/pipecat/services/mem0/memory.py +++ b/src/pipecat/services/mem0/memory.py @@ -11,6 +11,7 @@ and retrieve conversational memories, enhancing LLM context with relevant historical information. """ +import asyncio from typing import Any, Dict, List, Optional from loguru import logger @@ -112,9 +113,12 @@ class Mem0MemoryService(FrameProcessor): 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 _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 +135,16 @@ class Mem0MemoryService(FrameProcessor): 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 +162,7 @@ class Mem0MemoryService(FrameProcessor): "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 +171,15 @@ class Mem0MemoryService(FrameProcessor): ] 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 +188,9 @@ class Mem0MemoryService(FrameProcessor): 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 +203,7 @@ class Mem0MemoryService(FrameProcessor): self.last_query = query - memories = self._retrieve_memories(query) + memories = await self._retrieve_memories(query) if not memories: return @@ -203,11 +213,14 @@ class Mem0MemoryService(FrameProcessor): 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): @@ -241,9 +254,9 @@ class Mem0MemoryService(FrameProcessor): if latest_user_message: # 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(context_messages), name="mem0_store") # If we received an LLMMessagesFrame, create a new one with the enhanced messages if messages is not None: From 9152d858242937db1b6d79c078499c95f4148a10 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Mar 2026 12:53:33 -0400 Subject: [PATCH 2/5] fix(mem0): filter to user/assistant roles before storing in Mem0 Mem0 API only accepts user and assistant roles. Filter out system, developer, and other roles before calling add() to avoid 400 errors. --- src/pipecat/services/mem0/memory.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/mem0/memory.py b/src/pipecat/services/mem0/memory.py index 0b7f8e19f..b032ac9ed 100644 --- a/src/pipecat/services/mem0/memory.py +++ b/src/pipecat/services/mem0/memory.py @@ -253,10 +253,15 @@ class Mem0MemoryService(FrameProcessor): 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 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(context_messages), name="mem0_store") + 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: From 9c6d51c5703c4fdbc4632e0a0f6e10fe77a37dcf Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Mar 2026 13:25:28 -0400 Subject: [PATCH 3/5] feat(mem0): add get_memories() convenience method to Mem0MemoryService Expose a public method for retrieving all stored memories outside the pipeline, avoiding the need for callers to reimplement client branching, OR filter construction, and asyncio.to_thread wrapping. Simplify the example get_initial_greeting() to use it. --- examples/foundational/37-mem0.py | 79 +++++++++-------------------- src/pipecat/services/mem0/memory.py | 39 ++++++++++++++ 2 files changed, 64 insertions(+), 54 deletions(-) diff --git a/examples/foundational/37-mem0.py b/examples/foundational/37-mem0.py index e572d2dbe..a3116df24 100644 --- a/examples/foundational/37-mem0.py +++ b/examples/foundational/37-mem0.py @@ -42,7 +42,6 @@ The bot runs as part of a pipeline that processes audio frames and manages the c """ import os -from typing import Union from dotenv import load_dotenv from loguru import logger @@ -69,58 +68,35 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams 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,22 +241,17 @@ 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 - ) - - # Add the greeting as an assistant message to start the conversation - context.add_message({"role": "assistant", "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") + # 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": "developer", "content": greeting}) + + # Queue the context frame to start the conversation + await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/src/pipecat/services/mem0/memory.py b/src/pipecat/services/mem0/memory.py index b032ac9ed..754e2be0a 100644 --- a/src/pipecat/services/mem0/memory.py +++ b/src/pipecat/services/mem0/memory.py @@ -113,6 +113,45 @@ class Mem0MemoryService(FrameProcessor): self.last_query = None logger.info(f"Initialized Mem0MemoryService with {user_id=}, {agent_id=}, {run_id=}") + 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. From 4e4a8c45d5eb9649ee4cfaf0e9065d2bcf31fbbf Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Mar 2026 13:26:20 -0400 Subject: [PATCH 4/5] build(mem0): bump mem0ai dependency to >=1.0.8,<2 --- pyproject.toml | 2 +- uv.lock | 9 +++++---- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e7cddaaac..8ab6e31c4 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/uv.lock b/uv.lock index 99e6beb06..1e0851da4 100644 --- a/uv.lock +++ b/uv.lock @@ -3514,19 +3514,20 @@ wheels = [ [[package]] name = "mem0ai" -version = "0.1.115" +version = "1.0.8" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "openai" }, { name = "posthog" }, + { name = "protobuf" }, { name = "pydantic" }, { name = "pytz" }, { name = "qdrant-client" }, { name = "sqlalchemy" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/72/2a/ccf4a8b79a4e67d3a2475599ff8276d3bdccce5c5da2e14deb449e7dfb1a/mem0ai-0.1.115.tar.gz", hash = "sha256:147a6593604188acd30281c40171112aed9f16e196fa528627430c15e00f1e32", size = 115605, upload-time = "2025-07-24T09:49:10.467Z" } +sdist = { url = "https://files.pythonhosted.org/packages/b7/a6/292b42445cf2f5fb2207d523e31a823c10d5da2e939ece78dac0800edfb1/mem0ai-1.0.8.tar.gz", hash = "sha256:9af38c30b0250b3401f58a6004debf1f84f976b43fc4c6d830700c42b75af54c", size = 198898, upload-time = "2026-03-26T16:54:59.058Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/85/d5/55a5504077c175f4ff18259672df6fd0eae863236d730c87677d6de43f9a/mem0ai-0.1.115-py3-none-any.whl", hash = "sha256:29310bd5bcab644f7a4dbf87bd1afd878eb68458a2fb36cfcbf20bdff46fbdaf", size = 178065, upload-time = "2025-07-24T09:49:08.54Z" }, + { url = "https://files.pythonhosted.org/packages/85/b6/46f94bfa5863e86a1f3f77815728529d2e5b3851e259c611182e2c8c81ea/mem0ai-1.0.8-py3-none-any.whl", hash = "sha256:4852f403d213e5cb463454940f92211eadb6e38c75f2f1ec17f279033d3ba194", size = 296802, upload-time = "2026-03-26T16:54:57.356Z" }, ] [[package]] @@ -4922,7 +4923,7 @@ requires-dist = [ { name = "loguru", specifier = "~=0.7.3" }, { name = "markdown", specifier = ">=3.7,<4" }, { name = "mcp", extras = ["cli"], marker = "extra == 'mcp'", specifier = ">=1.11.0,<2" }, - { name = "mem0ai", marker = "extra == 'mem0'", specifier = "~=0.1.94" }, + { name = "mem0ai", marker = "extra == 'mem0'", specifier = ">=1.0.8,<2" }, { name = "mlx-whisper", marker = "extra == 'mlx-whisper'", specifier = "~=0.4.2" }, { name = "nltk", specifier = ">=3.9.4,<4" }, { name = "noisereduce", marker = "extra == 'noisereduce'", specifier = "~=3.0.3" }, From 83911dced6f40db22374ff50476f95a3346b9e40 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Mar 2026 13:30:00 -0400 Subject: [PATCH 5/5] docs: add changelog entries for #4156 --- changelog/4156.added.md | 1 + changelog/4156.changed.md | 1 + changelog/4156.fixed.2.md | 1 + changelog/4156.fixed.md | 1 + 4 files changed, 4 insertions(+) create mode 100644 changelog/4156.added.md create mode 100644 changelog/4156.changed.md create mode 100644 changelog/4156.fixed.2.md create mode 100644 changelog/4156.fixed.md diff --git a/changelog/4156.added.md b/changelog/4156.added.md new file mode 100644 index 000000000..bbf426d5f --- /dev/null +++ b/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. diff --git a/changelog/4156.changed.md b/changelog/4156.changed.md new file mode 100644 index 000000000..d4b8a269a --- /dev/null +++ b/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. diff --git a/changelog/4156.fixed.2.md b/changelog/4156.fixed.2.md new file mode 100644 index 000000000..1a6997d4f --- /dev/null +++ b/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. diff --git a/changelog/4156.fixed.md b/changelog/4156.fixed.md new file mode 100644 index 000000000..13620f772 --- /dev/null +++ b/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.