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

@@ -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):

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.