feat: integrate mem0 in pipecat pipeline

- Extend Mem0MemoryService to ImprovedMem0MemoryService class
- Add non-blocking async message storage to prevent pipeline
  delays
- Add incremental message tracking to reduce redundant API
  calls
- Extract configuration constants and helper methods for
  maintainability
- Improve user identification with email → name → session_id
  fallback
- Add comprehensive error handling without breaking
  conversation flow

Author: cmd-err

.env.example

@@ -108,3 +108,7 @@ LANGFUSE_PUBLIC_KEY=''
 LANGFUSE_BASEURL=https://us.cloud.langfuse.com
 AUTOMATIC_LANGFUSE_PROMPT_NAME=AUTOMATIC_VOICE_LANGFUSE_PROMPT
 AUTOMATIC_LANGFUSE_SYSTEM_PROMPT_LABEL=automatic_system_langfuse_prompt
+
+# Mem0 Configuration
+MEM0_API_KEY=""
+MEM0_ENABLED=false 

app/agents/voice/automatic/__init__.py

@@ -20,6 +20,7 @@
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
 from pipecat.services.google.rtvi import GoogleRTVIObserver
+from app.services.mem0.memory import ImprovedMem0MemoryService
 
 from app.core import config
 from app.agents.voice.automatic.utils.session_context import create_session_context, set_current_session_id
@@ -263,10 +264,23 @@ async def on_function_calls_started(service, function_calls):
         ptt_vad_filter = PTTVADFilter("PTTVADFilter")
         pipeline_components.append(ptt_vad_filter)  # Filter VAD frames after STT
 
-    # Add remaining components
     pipeline_components.extend([
         rtvi,
-        context_aggregator.user(),
+        context_aggregator.user()
+    ])
+    
+    if config.MEM0_ENABLED:
+        logger.info("Initializing Mem0 memory service")
+        memory_params = ImprovedMem0MemoryService.InputParams()
+        memory = ImprovedMem0MemoryService(
+            api_key=config.MEM0_API_KEY,
+            user_id=args.user_email or args.user_name or args.session_id,  # Fallback: email → name → session_id
+            params= memory_params,
+        )
+        pipeline_components.append(memory)
+
+    # Add remaining components
+    pipeline_components.extend([
         llm,
         tool_call_processor,
         tts,

app/core/config.py

@@ -66,6 +66,14 @@ def get_required_env(var_name: str) -> str:
 VAD_START_SECS = float(os.environ.get("VAD_START_SECS", 0.30))
 VAD_STOP_SECS = float(os.environ.get("VAD_STOP_SECS", 1.00))
 
+# Mem0 Configuration
+MEM0_API_KEY = os.getenv("MEM0_API_KEY", "")
+MEM0_ENABLED = os.getenv("MEM0_ENABLED", "false").lower() == "true"
+MEM0_MAX_FAILURES = int(os.getenv("MEM0_MAX_FAILURES", "3"))
+MEM0_RETRY_INTERVAL = int(os.getenv("MEM0_RETRY_INTERVAL", "300"))
+MEM0_SESSION_TIMEOUT = int(os.getenv("MEM0_SESSION_TIMEOUT", "3600"))
+MEM0_MIN_MESSAGE_LENGTH = int(os.getenv("MEM0_MIN_MESSAGE_LENGTH", "10"))
+
 # Tracing
 ENABLE_TRACING = os.environ.get("ENABLE_TRACING", "false").lower() == "true"
 OPEN_OBSERVE_BASE_URL = os.environ.get("OPEN_OBSERVE_BASE_URL", "https://periscope.breeze.in")

app/services/mem0/__init__.py

@@ -0,0 +1,7 @@
+"""
+Improved Mem0 integration for Pipecat with non-blocking storage.
+"""
+
+from .memory import ImprovedMem0MemoryService
+
+__all__ = ["ImprovedMem0MemoryService"]