Add mem0 as a service integration

examples/personalized-voice-agent/server/Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.10-bullseye
+
+RUN mkdir /app
+RUN mkdir /app/assets
+RUN mkdir /app/utils
+COPY requirements.txt /app/
+
+WORKDIR /app
+RUN pip3 install -r requirements.txt
+
+COPY *.py /app/
+COPY assets/* /app/assets/
+COPY .env /app/.env
+
+EXPOSE 7860
+
+CMD ["python3", "server.py"]

examples/personalized-voice-agent/server/README.md

@@ -0,0 +1,57 @@
+# Personalized Voice Agent Server
+
+A FastAPI server that manages bot instances and provides endpoints for both Daily Prebuilt and Pipecat client connections.
+
+## Environment Variables
+
+Copy `env.example` to `.env` and configure:
+
+```ini
+# Required API Keys
+DAILY_API_KEY=           # Your Daily API key
+MEM0_API_KEY=            # Your Mem0 API key
+OPENAI_API_KEY=          # Your OpenAI API key (required for OpenAI bot)
+ELEVENLABS_API_KEY=      # Your ElevenLabs API key
+
+# Optional Configuration
+DAILY_API_URL=           # Optional: Daily API URL (defaults to https://api.daily.co/v1)
+DAILY_SAMPLE_ROOM_URL=   # Optional: Fixed room URL for development
+HOST=                    # Optional: Host address (defaults to 0.0.0.0)
+FAST_API_PORT=           # Optional: Port number (defaults to 7860)
+```
+
+## Available Bots
+
+The server supports two bot implementations:
+
+1. **OpenAI Bot** (Default)
+
+   - Uses GPT-4 for conversation
+   - Requires OPENAI_API_KEY
+
+## Running the Server
+
+Set up and activate your virtual environment:
+
+```bash
+python3 -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+```
+
+Install dependencies:
+
+```bash
+pip install -r requirements.txt
+```
+
+If you want to use the local version of `pipecat` in this repo rather than the last published version, also run:
+
+```bash
+pip install --editable "../../../[daily,elevenlabs,openai,silero,mem0ai]"
+```
+
+Run the server:
+
+```bash
+python server.py
+```

examples/personalized-voice-agent/server/bot-openai.py

@@ -0,0 +1,243 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""OpenAI Bot Implementation.
+
+This module implements a chatbot using OpenAI's GPT-4 model for natural language
+processing. It includes:
+- Real-time audio/video interaction through Daily
+- Animated robot avatar
+- Text-to-speech using ElevenLabs
+- Support for both English and Spanish
+
+The bot runs as part of a pipeline that processes audio/video 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.services.mem0 import Mem0MemoryService
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
+from pipecat.services.elevenlabs import ElevenLabsTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+load_dotenv(override=True)
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+)
+
+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:
+            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 = "deshraj"
+    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,
+                camera_out_enabled=True,
+                camera_out_width=1024,
+                camera_out_height=576,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                transcription_enabled=True,
+                #
+                # Spanish
+                #
+                # transcription_settings=DailyTranscriptionSettings(
+                #     language="es",
+                #     tier="nova",
+                #     model="2-general"
+                # )
+            ),
+        )
+
+        # Initialize text-to-speech service
+        tts = ElevenLabsTTSService(
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            #
+            # English
+            #
+            voice_id="pNInz6obpgDQGcFmaJgB",
+            #
+            # Spanish
+            #
+            # model="eleven_multilingual_v2",
+            # voice_id="gD1IexrzCvsXPHUuT0s3",
+        )
+
+        # Initialize Mem0 memory service
+        memory = Mem0MemoryService(
+            api_key=os.getenv("MEM0_API_KEY"),
+            user_id=USER_ID,  # Unique identifier for the user
+            # agent_id="life_coach_bot",  # Optional identifier for the agent
+            # run_id="session_1", # 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")
+
+        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 events for Pipecat client UI
+        #
+        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())

examples/personalized-voice-agent/server/env.example

@@ -0,0 +1,6 @@
+DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
+DAILY_API_KEY=7df...
+OPENAI_API_KEY=sk-PL...
+GEMINI_API_KEY=AIza...
+ELEVENLABS_API_KEY=aeb...
+MEM0_API_KEY=m0-...

examples/personalized-voice-agent/server/requirements.txt

@@ -0,0 +1,5 @@
+python-dotenv
+fastapi[all]
+uvicorn
+pipecat-ai[daily,elevenlabs,openai,silero,google]
+mem0ai>=0.1.69

examples/personalized-voice-agent/server/runner.py

@@ -0,0 +1,56 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+import aiohttp
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    """Configure the Daily room and Daily REST helper."""
+    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    if not key:
+        raise Exception(
+            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
+        )
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token)

examples/personalized-voice-agent/server/server.py

@@ -0,0 +1,270 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""RTVI Bot Server Implementation.
+
+This FastAPI server manages RTVI bot instances and provides endpoints for both
+direct browser access and RTVI client connections. It handles:
+- Creating Daily rooms
+- Managing bot processes
+- Providing connection credentials
+- Monitoring bot status
+
+Requirements:
+- Daily API key (set in .env file)
+- Python 3.10+
+- FastAPI
+- Running bot implementation
+"""
+
+import argparse
+import os
+import subprocess
+from contextlib import asynccontextmanager
+from typing import Any, Dict
+
+import aiohttp
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, RedirectResponse
+from mem0 import MemoryClient
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
+
+# Load environment variables from .env file
+load_dotenv(override=True)
+
+# Maximum number of bot instances allowed per room
+MAX_BOTS_PER_ROOM = 1
+
+# Dictionary to track bot processes: {pid: (process, room_url)}
+bot_procs = {}
+
+# Store Daily API helpers
+daily_helpers = {}
+
+
+def cleanup():
+    """Cleanup function to terminate all bot processes.
+
+    Called during server shutdown.
+    """
+    for entry in bot_procs.values():
+        proc = entry[0]
+        proc.terminate()
+        proc.wait()
+
+
+def get_bot_file():
+    bot_implementation = os.getenv("BOT_IMPLEMENTATION", "openai").lower().strip()
+    # If blank or None, default to openai
+    if not bot_implementation:
+        bot_implementation = "openai"
+    if bot_implementation not in ["openai", "gemini"]:
+        raise ValueError(
+            f"Invalid BOT_IMPLEMENTATION: {bot_implementation}. Must be 'openai' or 'gemini'"
+        )
+    return f"bot-{bot_implementation}"
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """FastAPI lifespan manager that handles startup and shutdown tasks.
+
+    - Creates aiohttp session
+    - Initializes Daily API helper
+    - Cleans up resources on shutdown
+    """
+    aiohttp_session = aiohttp.ClientSession()
+    daily_helpers["rest"] = DailyRESTHelper(
+        daily_api_key=os.getenv("DAILY_API_KEY", ""),
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+    yield
+    await aiohttp_session.close()
+    cleanup()
+
+
+# Initialize FastAPI app with lifespan manager
+app = FastAPI(lifespan=lifespan)
+
+# Configure CORS to allow requests from any origin
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+async def create_room_and_token() -> tuple[str, str]:
+    """Helper function to create a Daily room and generate an access token.
+
+    Returns:
+        tuple[str, str]: A tuple containing (room_url, token)
+
+    Raises:
+        HTTPException: If room creation or token generation fails
+    """
+    room = await daily_helpers["rest"].create_room(DailyRoomParams())
+    if not room.url:
+        raise HTTPException(status_code=500, detail="Failed to create room")
+
+    token = await daily_helpers["rest"].get_token(room.url)
+    if not token:
+        raise HTTPException(status_code=500, detail=f"Failed to get token for room: {room.url}")
+
+    return room.url, token
+
+
+@app.get("/")
+async def start_agent(request: Request):
+    """Endpoint for direct browser access to the bot.
+
+    Creates a room, starts a bot instance, and redirects to the Daily room URL.
+
+    Returns:
+        RedirectResponse: Redirects to the Daily room URL
+
+    Raises:
+        HTTPException: If room creation, token generation, or bot startup fails
+    """
+    print("Creating room")
+    room_url, token = await create_room_and_token()
+    print(f"Room URL: {room_url}")
+
+    # Check if there is already an existing process running in this room
+    num_bots_in_room = sum(
+        1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None
+    )
+    if num_bots_in_room >= MAX_BOTS_PER_ROOM:
+        raise HTTPException(status_code=500, detail=f"Max bot limit reached for room: {room_url}")
+
+    # Spawn a new bot process
+    try:
+        bot_file = get_bot_file()
+        proc = subprocess.Popen(
+            [f"python3 -m {bot_file} -u {room_url} -t {token}"],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__)),
+        )
+        bot_procs[proc.pid] = (proc, room_url)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    return RedirectResponse(room_url)
+
+
+@app.get("/memories")
+async def get_memories(request: Request):
+    """Endpoint for getting memories from the bot.
+
+    Returns:
+        Dict[Any, Any]: Memories from the bot
+    """
+    memory_client = MemoryClient(api_key=os.getenv("MEM0_API_KEY"))
+    filters = {
+        "AND": [
+            {"user_id": "deshraj"},
+        ]
+    }
+    memories = memory_client.get_all(filters=filters, version="v2")
+
+    # Format memories for emission
+    formatted_memories = [
+        {
+            "id": str(memory.get("id", "")),
+            "content": memory.get("memory", ""),
+            "createdAt": memory.get("created_at", ""),
+            "categories": memory.get("categories", [])
+        } for memory in memories
+    ]
+
+    return formatted_memories
+
+
+@app.post("/connect")
+async def rtvi_connect(request: Request) -> Dict[Any, Any]:
+    """RTVI connect endpoint that creates a room and returns connection credentials.
+
+    This endpoint is called by RTVI clients to establish a connection.
+
+    Returns:
+        Dict[Any, Any]: Authentication bundle containing room_url and token
+
+    Raises:
+        HTTPException: If room creation, token generation, or bot startup fails
+    """
+    print("Creating room for RTVI connection")
+    room_url, token = await create_room_and_token()
+    print(f"Room URL: {room_url}")
+
+    # Start the bot process
+    try:
+        bot_file = get_bot_file()
+        proc = subprocess.Popen(
+            [f"python3 -m {bot_file} -u {room_url} -t {token}"],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__)),
+        )
+        bot_procs[proc.pid] = (proc, room_url)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    # Return the authentication bundle in format expected by DailyTransport
+    return {"room_url": room_url, "token": token}
+
+
+@app.get("/status/{pid}")
+def get_status(pid: int):
+    """Get the status of a specific bot process.
+
+    Args:
+        pid (int): Process ID of the bot
+
+    Returns:
+        JSONResponse: Status information for the bot
+
+    Raises:
+        HTTPException: If the specified bot process is not found
+    """
+    # Look up the subprocess
+    proc = bot_procs.get(pid)
+
+    # If the subprocess doesn't exist, return an error
+    if not proc:
+        raise HTTPException(status_code=404, detail=f"Bot with process id: {pid} not found")
+
+    # Check the status of the subprocess
+    status = "running" if proc[0].poll() is None else "finished"
+    return JSONResponse({"bot_id": pid, "status": status})
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    # Parse command line arguments for server configuration
+    default_host = os.getenv("HOST", "0.0.0.0")
+    default_port = int(os.getenv("FAST_API_PORT", "7860"))
+
+    parser = argparse.ArgumentParser(description="Daily Storyteller FastAPI server")
+    parser.add_argument("--host", type=str, default=default_host, help="Host address")
+    parser.add_argument("--port", type=int, default=default_port, help="Port number")
+    parser.add_argument("--reload", action="store_true", help="Reload code on change")
+
+    config = parser.parse_args()
+
+    # Start the FastAPI server
+    uvicorn.run(
+        "server:app",
+        host=config.host,
+        port=config.port,
+        reload=config.reload,
+    )