diff --git a/examples/foundational/35-mem0.py b/examples/foundational/35-mem0.py new file mode 100644 index 000000000..2615479f7 --- /dev/null +++ b/examples/foundational/35-mem0.py @@ -0,0 +1,242 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Mem0 Personalized Voice Agent Example with Pipecat. + +This example demonstrates how to create a conversational AI assistant with memory capabilities +using Mem0 integration. It shows how to build an agent that remembers previous interactions +and personalizes responses based on conversation history. + +The example: + 1. Sets up a video/audio conversation between a user and an AI assistant + 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 + +Example usage (run from pipecat root directory): + $ pip install "pipecat-ai[daily,openai,elevenlabs,silero,mem0]" + $ python examples/foundational/35-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) + + Environment variables (set in .env or in your terminal using `export`): + DAILY_SAMPLE_ROOM_URL=daily_sample_room_url + DAILY_API_KEY=daily_api_key + OPENAI_API_KEY=openai_api_key + ELEVENLABS_API_KEY=elevenlabs_api_key + MEM0_API_KEY=mem0_api_key + +The bot runs as part of a pipeline that processes audio frames and manages the conversation flow. +""" + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor +from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.services.mem0 import Mem0MemoryService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +load_dotenv(override=True) + +try: + from mem0 import MemoryClient +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: MemoryClient, user_id: str, agent_id: str, run_id: str +) -> str: + """Fetch all memories for the user and create a personalized greeting. + + Returns: + 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 {} + + # Get all memories for this user + memories = memory_client.get_all(filters=filters, version="v2") + + if not memories or len(memories) == 0: + logger.debug(f"!!! No memories found for this user. {memories}") + 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. " + + # 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): + 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?" + + logger.debug(f"Created personalized greeting from {len(memories)} memories") + return greeting + + except Exception as e: + logger.error(f"Error retrieving initial memories from Mem0: {e}") + return "Hello! How can I help you today?" + + +async def main(): + """Main bot execution function. + + Sets up and runs the bot pipeline including: + - Daily video transport + - Speech-to-text and text-to-speech services + - Language model integration + - Mem0 memory service + - RTVI event handling + """ + # Note: You can pass the user_id as a parameter in API call + USER_ID = "pipecat-demo-user" + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + # Set up Daily transport with video/audio parameters + transport = DailyTransport( + room_url, + token, + "Chatbot", + DailyParams( + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + transcription_enabled=True, + ), + ) + + # Initialize text-to-speech service + tts = ElevenLabsTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id="pNInz6obpgDQGcFmaJgB", + ) + + # Initialize Mem0 memory service + 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 + params=Mem0MemoryService.InputParams( + search_limit=10, + search_threshold=0.3, + api_version="v2", + system_prompt="Based on previous conversations, I recall: \n\n", + add_as_system_message=True, + position=1, + ), + ) + + # Initialize LLM service + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini") + + messages = [ + { + "role": "system", + "content": """You are a personal assistant. You can remember things about the person you are talking to. + Some Guidelines: + - Make sure your responses are friendly yet short and concise. + - If the user asks you to remember something, make sure to remember it. + - Greet the user by their name if you know about it. + """, + }, + ] + + # Set up conversation context and management + # The context_aggregator will automatically collect conversation context + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + rtvi = RTVIProcessor(config=RTVIConfig(config=[])) + + pipeline = Pipeline( + [ + transport.input(), + rtvi, + context_aggregator.user(), + memory, + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + observers=[RTVIObserver(rtvi)], + ) + + @rtvi.event_handler("on_client_ready") + async def on_client_ready(rtvi): + await rtvi.set_bot_ready() + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + + # 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([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + print(f"Participant left: {participant}") + await task.cancel() + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index dddbf3678..6c198c236 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -85,6 +85,7 @@ ultravox = [ "transformers~=4.48.0", "vllm~=0.7.3" ] webrtc = [ "aiortc~=1.10.1", "opencv-python~=4.11.0.86" ] websocket = [ "websockets~=13.1", "fastapi~=0.115.6" ] whisper = [ "faster-whisper~=1.1.1" ] +mem0 = [ "mem0ai~=0.1.76" ] [tool.setuptools.packages.find] # All the following settings are optional: diff --git a/src/pipecat/services/mem0.py b/src/pipecat/services/mem0.py new file mode 100644 index 000000000..23d6fa8e1 --- /dev/null +++ b/src/pipecat/services/mem0.py @@ -0,0 +1,208 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Any, Dict, List + +from loguru import logger +from pydantic import BaseModel, Field + +from pipecat.frames.frames import ErrorFrame, Frame, LLMMessagesFrame +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +try: + from mem0 import 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}") + + +class Mem0MemoryService(FrameProcessor): + """A standalone memory service that integrates with Mem0. + + This service intercepts message frames in the pipeline, stores them in Mem0, + and enhances context with relevant memories before passing them downstream. + + Args: + api_key (str): The API key for accessing Mem0's API + user_id (str): The user ID to associate with memories in Mem0 + params (InputParams, optional): Configuration parameters for memory retrieval + """ + + class InputParams(BaseModel): + search_limit: int = Field(default=10, ge=1) + search_threshold: float = Field(default=0.1, ge=0.0, le=1.0) + api_version: str = Field(default="v2") + system_prompt: str = Field(default="Based on previous conversations, I recall: \n\n") + add_as_system_message: bool = Field(default=True) + position: int = Field(default=1) + + def __init__( + self, + *, + api_key: str, + user_id: str = None, + agent_id: str = None, + run_id: str = None, + params: InputParams = InputParams(), + ): + # Important: Call the parent class __init__ first + super().__init__() + + 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") + + self.user_id = user_id + self.agent_id = agent_id + self.run_id = run_id + self.search_limit = params.search_limit + self.search_threshold = params.search_threshold + self.api_version = params.api_version + self.system_prompt = params.system_prompt + self.add_as_system_message = params.add_as_system_message + self.position = params.position + 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]]): + """Store messages in Mem0. + + Args: + messages: List of message dictionaries to store + """ + try: + logger.debug(f"Storing {len(messages)} messages in Mem0") + params = { + "messages": messages, + "metadata": {"platform": "pipecat"}, + "output_format": "v1.1", + } + for id in ["user_id", "agent_id", "run_id"]: + if getattr(self, id): + params[id] = getattr(self, id) + # Note: You can run this in background to avoid blocking the conversation + 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]]: + """Retrieve relevant memories from Mem0. + + Args: + query: The query to search for relevant memories + + Returns: + List of relevant memory dictionaries + """ + 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, + ) + + logger.debug(f"Retrieved {len(results)} memories from Mem0") + return results + except Exception as e: + logger.error(f"Error retrieving memories from Mem0: {e}") + return [] + + def _enhance_context_with_memories(self, context: OpenAILLMContext, query: str): + """Enhance the LLM context with relevant memories. + + Args: + context: The OpenAILLMContext to enhance + query: The query to search for relevant memories + """ + # Skip if this is the same query we just processed + if self.last_query == query: + return + + self.last_query = query + + memories = self._retrieve_memories(query) + if not memories: + return + + # Format memories as a message + memory_text = self.system_prompt + for i, memory in enumerate(memories, 1): + 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}) + logger.debug(f"Enhanced context with {len(memories)} memories") + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming frames, intercept context frames for memory integration. + + Args: + frame: The incoming frame to process + direction: The direction of frame flow in the pipeline + """ + await super().process_frame(frame, direction) + + context = None + messages = None + + if isinstance(frame, OpenAILLMContextFrame): + context = frame.context + elif isinstance(frame, LLMMessagesFrame): + messages = frame.messages + context = OpenAILLMContext.from_messages(messages) + + if context: + try: + # Get the latest user message to use as a query for memory retrieval + context_messages = context.get_messages() + latest_user_message = None + + for message in reversed(context_messages): + if message.get("role") == "user" and isinstance(message.get("content"), str): + latest_user_message = message.get("content") + break + + 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) + + # If we received an LLMMessagesFrame, create a new one with the enhanced messages + if messages is not None: + await self.push_frame(LLMMessagesFrame(context.get_messages())) + else: + # Otherwise, pass the enhanced context frame downstream + await self.push_frame(frame) + except Exception as e: + logger.error(f"Error processing with Mem0: {str(e)}") + await self.push_frame(ErrorFrame(f"Error processing with Mem0: {str(e)}")) + await self.push_frame(frame) # Still pass the original frame through + else: + # For non-context frames, just pass them through + await self.push_frame(frame, direction)