wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate with Mem0 OSS

Browse Source
This commit is contained in:
Dev-Khant
2025-04-28 17:15:53 +05:30
parent 029ef4f8c2
commit c0c41789ab
3 changed files with 101 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
examples/foundational/37-mem0.py
View File

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

2
pyproject.toml
View File

@@ -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 = []

55
src/pipecat/services/mem0/memory.py
View File

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