From 9c6d51c5703c4fdbc4632e0a0f6e10fe77a37dcf Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Mar 2026 13:25:28 -0400 Subject: [PATCH] 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.