From c0c41789abec31c4723d127001752b7ffbf0c971 Mon Sep 17 00:00:00 2001 From: Dev-Khant Date: Mon, 28 Apr 2025 17:15:53 +0530 Subject: [PATCH] Integrate with Mem0 OSS --- examples/foundational/37-mem0.py | 82 ++++++++++++++++++++++------- pyproject.toml | 2 +- src/pipecat/services/mem0/memory.py | 55 ++++++++++++------- 3 files changed, 101 insertions(+), 38 deletions(-) diff --git a/examples/foundational/37-mem0.py b/examples/foundational/37-mem0.py index a9ce685c9..1c0e8863a 100644 --- a/examples/foundational/37-mem0.py +++ b/examples/foundational/37-mem0.py @@ -15,16 +15,20 @@ The example: 2. Uses Mem0 to store and retrieve memories from conversations 3. Creates personalized greetings based on previous interactions 4. Handles multi-modal interaction through audio + 5. Demonstrates two approaches for memory management: + - Using Mem0 API (cloud-based memory storage) + - Using local configuration with custom LLM (self-hosted memory) Example usage (run from pipecat root directory): $ pip install "pipecat-ai[daily,openai,elevenlabs,silero,mem0]" - $ python examples/foundational/35-mem0.py + $ python examples/foundational/37-mem0.py Requirements: - OpenAI API key (for GPT-4o-mini) - ElevenLabs API key (for text-to-speech) - Daily API key (for video/audio transport) - - Mem0 API key (for memory storage and retrieval) + - Mem0 API key (for cloud-based memory storage) + - [Optional] Anthropic API key (if using Claude with local config) Environment variables (set in .env or in your terminal using `export`): DAILY_SAMPLE_ROOM_URL=daily_sample_room_url @@ -32,16 +36,16 @@ Requirements: OPENAI_API_KEY=openai_api_key ELEVENLABS_API_KEY=elevenlabs_api_key MEM0_API_KEY=mem0_api_key + ANTHROPIC_API_KEY=anthropic_api_key (if using Claude with local config) The bot runs as part of a pipeline that processes audio frames and manages the conversation flow. """ import argparse import os - +from typing import Union from dotenv import load_dotenv from loguru import logger -from openai import audio from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.pipeline.pipeline import Pipeline @@ -60,7 +64,7 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection load_dotenv(override=True) try: - from mem0 import MemoryClient + from mem0 import MemoryClient, Memory except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -70,7 +74,7 @@ except ModuleNotFoundError as e: async def get_initial_greeting( - memory_client: MemoryClient, user_id: str, agent_id: str, run_id: str + memory_client: Union[MemoryClient, Memory], user_id: str, agent_id: str, run_id: str ) -> str: """Fetch all memories for the user and create a personalized greeting. @@ -78,13 +82,18 @@ async def get_initial_greeting( A personalized greeting based on user memories """ try: - # 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 {} + 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") + # 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}") @@ -96,7 +105,7 @@ async def get_initial_greeting( # 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[:3], 1): + for i, memory in enumerate(memories["results"][:3], 1): memory_content = memory.get("memory", "") # Keep memory references brief if len(memory_content) > 100: @@ -120,7 +129,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac - Daily video transport - Speech-to-text and text-to-speech services - Language model integration - - Mem0 memory service + - Mem0 memory service (using either API or local configuration) - RTVI event handling """ # Note: You can pass the user_id as a parameter in API call @@ -145,12 +154,16 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac voice_id="pNInz6obpgDQGcFmaJgB", ) - # Initialize Mem0 memory service + # ===================================================================== + # OPTION 1: Using Mem0 API (cloud-based approach) + # This approach uses Mem0's cloud service for memory management + # Requires: MEM0_API_KEY set in your environment + # ===================================================================== memory = Mem0MemoryService( - api_key=os.getenv("MEM0_API_KEY"), - user_id=USER_ID, # Unique identifier for the user - # agent_id="agent1", # Optional identifier for the agent - # run_id="session1", # Optional identifier for the run + api_key=os.getenv("MEM0_API_KEY"), # Your Mem0 API key + user_id=USER_ID, # Unique identifier for the user + agent_id="agent1", # Optional identifier for the agent + run_id="session1", # Optional identifier for the run params=Mem0MemoryService.InputParams( search_limit=10, search_threshold=0.3, @@ -161,6 +174,37 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac ), ) + # ===================================================================== + # OPTION 2: Using Mem0 with local configuration (self-hosted approach) + # This approach uses a local LLM configuration for memory management + # Requires: Anthropic API key if using Claude model + # ===================================================================== + # Uncomment the following code and comment out the previous memory initialization to use local config + + # local_config = { + # "llm": { + # "provider": "anthropic", + # "config": { + # "model": "claude-3-5-sonnet-20240620", + # "api_key": os.getenv("ANTHROPIC_API_KEY"), # Make sure to set this in your .env + # } + # }, + # "embedder": { + # "provider": "openai", + # "config": { + # "model": "text-embedding-3-large" + # } + # } + # } + + # # Initialize Mem0 memory service with local configuration + # memory = Mem0MemoryService( + # local_config=local_config, # Use local LLM for memory processing + # user_id=USER_ID, # Unique identifier for the user + # # agent_id="agent1", # Optional identifier for the agent + # # run_id="session1", # Optional identifier for the run + # ) + # Initialize LLM service llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini") diff --git a/pyproject.toml b/pyproject.toml index 58d2097ab..ca7f9db61 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,7 +65,7 @@ livekit = [ "livekit~=0.22.0", "livekit-api~=0.8.2", "tenacity~=9.0.0" ] lmnt = [ "websockets~=13.1" ] local = [ "pyaudio~=0.2.14" ] mcp = [ "mcp[cli]~=1.6.0" ] -mem0 = [ "mem0ai~=0.1.76" ] +mem0 = [ "mem0ai~=0.1.94" ] mlx-whisper = [ "mlx-whisper~=0.4.2" ] moondream = [ "einops~=0.8.0", "timm~=1.0.13", "transformers~=4.48.0" ] nim = [] diff --git a/src/pipecat/services/mem0/memory.py b/src/pipecat/services/mem0/memory.py index 23d6fa8e1..c0a84b19e 100644 --- a/src/pipecat/services/mem0/memory.py +++ b/src/pipecat/services/mem0/memory.py @@ -17,7 +17,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( from pipecat.processors.frame_processor import FrameDirection, FrameProcessor try: - from mem0 import MemoryClient # noqa: F401 + from mem0 import MemoryClient, Memory # noqa: F401 except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -49,7 +49,8 @@ class Mem0MemoryService(FrameProcessor): def __init__( self, *, - api_key: str, + api_key: str = None, + local_config: Dict[str, Any] = {}, user_id: str = None, agent_id: str = None, run_id: str = None, @@ -58,7 +59,10 @@ class Mem0MemoryService(FrameProcessor): # Important: Call the parent class __init__ first super().__init__() - self.memory_client = MemoryClient(api_key=api_key) + if local_config: + self.memory_client = Memory.from_config(local_config) + else: + self.memory_client = MemoryClient(api_key=api_key) # At least one of user_id, agent_id, or run_id must be provided if not any([user_id, agent_id, run_id]): raise ValueError("At least one of user_id, agent_id, or run_id must be provided") @@ -91,6 +95,9 @@ class Mem0MemoryService(FrameProcessor): for id in ["user_id", "agent_id", "run_id"]: if getattr(self, id): params[id] = getattr(self, id) + + 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) except Exception as e: @@ -107,20 +114,32 @@ class Mem0MemoryService(FrameProcessor): """ try: logger.debug(f"Retrieving memories for query: {query}") - 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 = {"AND": 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, - ) + if isinstance(self.memory_client, Memory): + params = { + "query": query, + "user_id": self.user_id, + "agent_id": self.agent_id, + "run_id": self.run_id, + "limit": self.search_limit + } + params = {k: v for k, v in params.items() if v is not None} + results = self.memory_client.search(**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 = {"AND": 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", + ) logger.debug(f"Retrieved {len(results)} memories from Mem0") return results @@ -147,7 +166,7 @@ class Mem0MemoryService(FrameProcessor): # Format memories as a message memory_text = self.system_prompt - for i, memory in enumerate(memories, 1): + for i, memory in enumerate(memories["results"], 1): memory_text += f"{i}. {memory.get('memory', '')}\n\n" # Add memories as a system message or user message based on configuration