Livekit - Mem0

Documentation Index

Fetch the complete documentation index at: /llms.txt

Use this file to discover all available pages before exploring further.

This guide demonstrates how to create a memory-enabled voice assistant using LiveKit, Deepgram, OpenAI, and Mem0, focusing on creating an intelligent, context-aware travel planning agent.

Prerequisites

Before you begin, make sure you have:

Installed Livekit Agents SDK with voice dependencies of silero and deepgram:

pip install livekit livekit-agents \
livekit-plugins-silero \
livekit-plugins-deepgram \
livekit-plugins-openai \
livekit-plugins-turn-detector \
livekit-plugins-noise-cancellation

Installed Mem0 SDK:

    pip install mem0ai
    ```

3. Set up your API keys in a `.env` file:

LIVEKIT_URL=your_livekit_url
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret
DEEPGRAM_API_KEY=your_deepgram_api_key
MEM0_API_KEY=your_mem0_api_key
OPENAI_API_KEY=your_openai_api_key
```

Note: Make sure to have a Livekit and Deepgram account. You can find these variables LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET from the LiveKit Cloud Console. For more information, refer to the LiveKit Documentation. For DEEPGRAM_API_KEY, you can get it from the Deepgram Console. Refer to the Deepgram Documentation for more details.

Code Breakdown

Let’s break down the key components of this implementation using LiveKit Agents:

1. Setting Up Dependencies and Environment

    import os
    import logging
    from pathlib import Path
    from dotenv import load_dotenv

from mem0 import AsyncMemoryClient

from livekit.agents import (
        JobContext,
        WorkerOptions,
        cli,
        ChatContext,
        ChatMessage,
        RoomInputOptions,
        Agent,
        AgentSession,
    )
    from livekit.plugins import openai, silero, deepgram, noise_cancellation
    from livekit.plugins.turn_detector.english import EnglishModel

# Load environment variables
    load_dotenv()
    ```

### 2. Mem0 Client and Agent Definition

# User ID for RAG data in Mem0
RAG_USER_ID = "livekit-mem0"
mem0_client = AsyncMemoryClient()

class MemoryEnabledAgent(Agent): """ An agent that can answer questions using RAG (Retrieval Augmented Generation) with Mem0. """ def init(self) -> None: super().init( instructions=""" You are a helpful voice assistant. You are a travel guide named George and will help the user to plan a travel trip of their dreams. You should help the user plan for various adventures like work retreats, family vacations or solo backpacking trips. You should be careful to not suggest anything that would be dangerous, illegal or inappropriate. You can remember past interactions and use them to inform your answers. Use semantic memory retrieval to provide contextually relevant responses. """, ) self._seen_results = set() # Track previously seen result IDs logger.info(f"Mem0 Agent initialized. Using user_id: {RAG_USER_ID}")

async def on_enter(self): self.session.generate_reply( instructions="Briefly greet the user and offer your assistance." )

async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage) -> None: # Persist the user message in Mem0 try: logger.info(f"Adding user message to Mem0: {new_message.text_content}") add_result = await mem0_client.add( [{"role": "user", "content": new_message.text_content}], user_id=RAG_USER_ID ) logger.info(f"Mem0 add result (user): {add_result}") except Exception as e: logger.warning(f"Failed to store user message in Mem0: {e}")

RAG: Retrieve relevant context from Mem0 and inject as assistant message

        try:
            logger.info("About to await mem0_client.search for RAG context")
            search_results = await mem0_client.search(
                new_message.text_content,
                filters={"user_id": RAG_USER_ID},
            )
            logger.info(f"mem0_client.search returned: {search_results}")
            if search_results and search_results.get('results', []):
                context_parts = []
                for result in search_results.get('results', []):
                    paragraph = result.get("memory") or result.get("text")
                    if paragraph:
                        source = "mem0 Memories"
                        if "from [" in paragraph:
                            source = paragraph.split("from [")[1].split("]")[0]
                            paragraph = paragraph.split("]")[1].strip()
                        context_parts.append(f"Source: {source}\nContent: {paragraph}\n")
                if context_parts:
                    full_context = "\n\n".join(context_parts)
                    logger.info(f"Injecting RAG context: {full_context}")
                    turn_ctx.add_message(role="assistant", content=full_context)
                    await self.update_chat_ctx(turn_ctx)
        except Exception as e:
            logger.warning(f"Failed to inject RAG context from Mem0: {e}")

await super().on_user_turn_completed(turn_ctx, new_message) ```

3. Entrypoint and Session Setup

    async def entrypoint(ctx: JobContext):
        """Main entrypoint for the agent."""
        await ctx.connect()

session = AgentSession(
            stt=deepgram.STT(),
            llm=openai.LLM(model="gpt-5-mini"),
            tts=openai.TTS(voice="ash",),
            turn_detection=EnglishModel(),
            vad=silero.VAD.load(),
        )

await session.start(
            agent=MemoryEnabledAgent(),
            room=ctx.room,
            room_input_options=RoomInputOptions(
                noise_cancellation=noise_cancellation.BVC(),
            ),
        )

# Initial greeting
        await session.generate_reply(
            instructions="Greet the user warmly as George the travel guide and ask how you can help them plan their next adventure.",
            allow_interruptions=True
        )

# Run the application
    if __name__ == "__main__":
        cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
    ```

## Key Features of This Implementation

1. **Semantic Memory Retrieval**: Uses Mem0 to store and retrieve contextually relevant memories
2. **Voice Interaction**: Leverages LiveKit for voice communication with proper turn detection
3. **Intelligent Context Management**: Augments conversations with past interactions
4. **Travel Planning Specialization**: Focused on creating a helpful travel guide assistant
5. **Function Tools**: Modern tool definition for enhanced capabilities

## Running the Example

To run this example:

1. Install all required dependencies
2. Set up your `.env` file with the necessary API keys
3. Ensure your microphone and audio setup are configured
4. Run the script with Python 3.11 or newer and with the following command:

python mem0-livekit-voice-agent.py start
```

or to start your agent in console mode to run inside your terminal:

    python mem0-livekit-voice-agent.py console
    ```

5. After the script starts, you can interact with the voice agent using [LiveKit’s Agent Platform](https://agents-playground.livekit.io/) and connect to the agent to start conversations.

## Best Practices for Voice Agents with Memory

1. **Context Preservation**: Store enough context with each memory for effective retrieval
2. **Privacy Considerations**: Implement secure memory management
3. **Relevant Memory Filtering**: Use semantic search to retrieve only the most relevant memories
4. **Error Handling**: Implement robust error handling for memory operations

## Debugging Function Tools

- To run the script in debug mode simply start the assistant with `dev` mode:

python mem0-livekit-voice-agent.py dev
```

When working with memory-enabled voice agents, use Python’s logging module for effective debugging:

    import logging

# Set up logging
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger("memory_voice_agent")
    ```

- Check the logs for any issues with API keys, connectivity, or memory operations.
- Ensure your `.env` file is correctly configured and loaded.