small changes to make 35-mem0.py like the other foundational single-file examples.

This reverts commit 397bae29f7.

CHANGELOG.md

@@ -5,6 +5,43 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- ElevenLabs TTS services now support a sample rate of 8000.
+
+### Fixed
+
+- Fixed an issue where the `RTVIObserver` would report two bot started and
+  stopped speaking events for each bot turn.
+
+- Fixed an issue in `UltravoxSTTService` that caused improper audio processing
+  and incorrect LLM frame output.
+
+### Other
+
+- Added `examples/foundational/07x-interruptible-local.py` to show how a local
+  transport can be used.
+
+## [0.0.60] - 2025-03-20
+
+### Added
+
+- Added `default_headers` parameter to `BaseOpenAILLMService` constructor.
+
+### Changed
+
+- Rollback to `deepgram-sdk` 3.8.0 since 3.10.1 was causing connections issues.
+
+- Changed the default `InputAudioTranscription` model to `gpt-4o-transcribe`
+  for `OpenAIRealtimeBetaLLMService`.
+
+### Other
+
+- Update the `19-openai-realtime-beta.py` and `19a-azure-realtime-beta.py`
+  examples to use the FunctionSchema format.
+
 ## [0.0.59] - 2025-03-20
 
 ### Added

dev-requirements.txt

@@ -3,10 +3,10 @@ coverage~=7.6.12
 grpcio-tools~=1.67.1
 pip-tools~=7.4.1
 pre-commit~=4.0.1
-pyright~=1.1.394
+pyright~=1.1.397
 pytest~=8.3.4
 pytest-asyncio~=0.25.3
-ruff~=0.9.7
+ruff~=0.11.1
 setuptools~=70.0.0
 setuptools_scm~=8.1.0
 python-dotenv~=1.0.1

examples/foundational/07u-interruptible-ultravox.py

@@ -14,6 +14,7 @@ from loguru import logger
 from runner import configure
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask

examples/foundational/07x-interruptible-local.py

@@ -0,0 +1,91 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+from dotenv import load_dotenv
+from loguru import logger
+
+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.cartesia import CartesiaTTSService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    transport = LocalAudioTransport(
+        LocalAudioTransportParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(),
+            vad_audio_passthrough=True,
+        )
+    )
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            report_only_initial_ttfb=True,
+        ),
+    )
+
+    messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+    await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    runner = PipelineRunner()
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/foundational/19-openai-realtime-beta.py

@@ -14,6 +14,8 @@ from dotenv import load_dotenv
 from loguru import logger
 from runner import configure
 
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.pipeline.pipeline import Pipeline
@@ -21,10 +23,11 @@ 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.openai_realtime_beta import (
+    InputAudioNoiseReduction,
     InputAudioTranscription,
     OpenAIRealtimeBetaLLMService,
+    SemanticTurnDetection,
     SessionProperties,
-    TurnDetection,
 )
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -46,28 +49,25 @@ async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context
     )
 
 
-tools = [
-    {
-        "type": "function",
-        "name": "get_current_weather",
-        "description": "Get the current weather",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "location": {
-                    "type": "string",
-                    "description": "The city and state, e.g. San Francisco, CA",
-                },
-                "format": {
-                    "type": "string",
-                    "enum": ["celsius", "fahrenheit"],
-                    "description": "The temperature unit to use. Infer this from the users location.",
-                },
-            },
-            "required": ["location", "format"],
+weather_function = FunctionSchema(
+    name="get_current_weather",
+    description="Get the current weather",
+    properties={
+        "location": {
+            "type": "string",
+            "description": "The city and state, e.g. San Francisco, CA",
         },
-    }
-]
+        "format": {
+            "type": "string",
+            "enum": ["celsius", "fahrenheit"],
+            "description": "The temperature unit to use. Infer this from the users location.",
+        },
+    },
+    required=["location", "format"],
+)
+
+# Create tools schema
+tools = ToolsSchema(standard_tools=[weather_function])
 
 
 async def main():
@@ -92,9 +92,10 @@ async def main():
             input_audio_transcription=InputAudioTranscription(),
             # Set openai TurnDetection parameters. Not setting this at all will turn it
             # on by default
-            turn_detection=TurnDetection(silence_duration_ms=1000),
+            turn_detection=SemanticTurnDetection(),
             # Or set to False to disable openai turn detection and use transport VAD
             # turn_detection=False,
+            input_audio_noise_reduction=InputAudioNoiseReduction(type="near_field"),
             # tools=tools,
             instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI.
 

examples/foundational/19a-azure-realtime-beta.py

@@ -10,11 +10,12 @@ import sys
 from datetime import datetime
 
 import aiohttp
-import websockets
 from dotenv import load_dotenv
 from loguru import logger
 from runner import configure
 
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.pipeline.pipeline import Pipeline
@@ -25,7 +26,6 @@ from pipecat.services.openai_realtime_beta import (
     AzureRealtimeBetaLLMService,
     InputAudioTranscription,
     SessionProperties,
-    TurnDetection,
 )
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -47,28 +47,26 @@ async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context
     )
 
 
-tools = [
-    {
-        "type": "function",
-        "name": "get_current_weather",
-        "description": "Get the current weather",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "location": {
-                    "type": "string",
-                    "description": "The city and state, e.g. San Francisco, CA",
-                },
-                "format": {
-                    "type": "string",
-                    "enum": ["celsius", "fahrenheit"],
-                    "description": "The temperature unit to use. Infer this from the users location.",
-                },
-            },
-            "required": ["location", "format"],
+# Define weather function using standardized schema
+weather_function = FunctionSchema(
+    name="get_current_weather",
+    description="Get the current weather",
+    properties={
+        "location": {
+            "type": "string",
+            "description": "The city and state, e.g. San Francisco, CA",
         },
-    }
-]
+        "format": {
+            "type": "string",
+            "enum": ["celsius", "fahrenheit"],
+            "description": "The temperature unit to use. Infer this from the users location.",
+        },
+    },
+    required=["location", "format"],
+)
+
+# Create tools schema
+tools = ToolsSchema(standard_tools=[weather_function])
 
 
 async def main():

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())

examples/telnyx-chatbot/env.example

@@ -1,3 +1,3 @@
 OPENAI_API_KEY=
 DEEPGRAM_API_KEY=
-ELEVENLABS_API_KEY=
+CARTESIA_API_KEY=

examples/telnyx-chatbot/requirements.txt

@@ -1,4 +1,4 @@
-pipecat-ai[openai,silero,deepgram,elevenlabs]
+pipecat-ai[openai,silero,deepgram,cartesia]
 fastapi
 uvicorn
 python-dotenv

pyproject.toml

@@ -49,7 +49,7 @@ neuphonic = [ "pyneuphonic~=1.5.13", "websockets~=13.1" ]
 cerebras = []
 deepseek = []
 daily = [ "daily-python~=0.15.0" ]
-deepgram = [ "deepgram-sdk~=3.10.1" ]
+deepgram = [ "deepgram-sdk~=3.8.0" ]
 elevenlabs = [ "websockets~=13.1" ]
 fal = [ "fal-client~=0.5.9" ]
 fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
@@ -84,6 +84,7 @@ together = []
 ultravox = [ "transformers~=4.48.0", "vllm~=0.7.3" ]
 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:

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -157,7 +157,7 @@ class OpenAILLMContext:
         self._tool_choice = tool_choice
 
     def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = NOT_GIVEN):
-        if tools != NOT_GIVEN and len(tools) == 0:
+        if tools != NOT_GIVEN and isinstance(tools, list) and len(tools) == 0:
             tools = NOT_GIVEN
         self._tools = tools
 

src/pipecat/processors/frameworks/rtvi.py

@@ -440,7 +440,9 @@ class RTVIObserver(BaseObserver):
 
         if isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)):
             await self._handle_interruptions(frame)
-        elif isinstance(frame, (BotStartedSpeakingFrame, BotStoppedSpeakingFrame)):
+        elif isinstance(frame, (BotStartedSpeakingFrame, BotStoppedSpeakingFrame)) and (
+            direction == FrameDirection.UPSTREAM
+        ):
             await self._handle_bot_speaking(frame)
         elif isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)):
             await self._handle_user_transcriptions(frame)

src/pipecat/services/azure.py

@@ -686,8 +686,11 @@ class AzureSTTService(STTService):
     ):
         super().__init__(sample_rate=sample_rate, **kwargs)
 
-        self._speech_config = SpeechConfig(subscription=api_key, region=region)
-        self._speech_config.speech_recognition_language = language
+        self._speech_config = SpeechConfig(
+            subscription=api_key,
+            region=region,
+            speech_recognition_language=language_to_azure_language(language),
+        )
 
         self._audio_stream = None
         self._speech_recognizer = None

src/pipecat/services/elevenlabs.py

@@ -102,6 +102,8 @@ def language_to_elevenlabs_language(language: Language) -> Optional[str]:
 
 def output_format_from_sample_rate(sample_rate: int) -> str:
     match sample_rate:
+        case 8000:
+            return "pcm_8000"
         case 16000:
             return "pcm_16000"
         case 22050:
@@ -113,7 +115,7 @@ def output_format_from_sample_rate(sample_rate: int) -> str:
     logger.warning(
         f"ElevenLabsTTSService: No output format available for {sample_rate} sample rate"
     )
-    return "pcm_16000"
+    return "pcm_24000"
 
 
 def build_elevenlabs_voice_settings(

src/pipecat/services/mem0.py

@@ -0,0 +1,199 @@
+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
+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)

src/pipecat/services/openai.py

@@ -112,6 +112,7 @@ class BaseOpenAILLMService(LLMService):
         base_url=None,
         organization=None,
         project=None,
+        default_headers: Mapping[str, str] | None = None,
         params: InputParams = InputParams(),
         **kwargs,
     ):
@@ -128,10 +129,23 @@ class BaseOpenAILLMService(LLMService):
         }
         self.set_model_name(model)
         self._client = self.create_client(
-            api_key=api_key, base_url=base_url, organization=organization, project=project, **kwargs
+            api_key=api_key,
+            base_url=base_url,
+            organization=organization,
+            project=project,
+            default_headers=default_headers,
+            **kwargs,
         )
 
-    def create_client(self, api_key=None, base_url=None, organization=None, project=None, **kwargs):
+    def create_client(
+        self,
+        api_key=None,
+        base_url=None,
+        organization=None,
+        project=None,
+        default_headers=None,
+        **kwargs,
+    ):
         return AsyncOpenAI(
             api_key=api_key,
             base_url=base_url,
@@ -142,6 +156,7 @@ class BaseOpenAILLMService(LLMService):
                     max_keepalive_connections=100, max_connections=1000, keepalive_expiry=None
                 )
             ),
+            default_headers=default_headers,
         )
 
     def can_generate_metrics(self) -> bool:

src/pipecat/services/openai_realtime_beta/events.py

@@ -14,17 +14,24 @@ from pydantic import BaseModel, Field
 #
 # session properties
 #
-InputAudioTranscriptionModel = Literal["whisper-1", "gpt-4o-transcribe"]
 
 
 class InputAudioTranscription(BaseModel):
-    model: InputAudioTranscriptionModel
+    """Configuration for audio transcription settings.
+
+    Attributes:
+        model: Transcription model to use (e.g., "gpt-4o-transcribe", "whisper-1").
+        language: Optional language code for transcription.
+        prompt: Optional transcription hint text.
+    """
+
+    model: str = "gpt-4o-transcribe"
     language: Optional[str]
     prompt: Optional[str]
 
     def __init__(
         self,
-        model: Optional[InputAudioTranscriptionModel] = "whisper-1",
+        model: Optional[str] = "gpt-4o-transcribe",
         language: Optional[str] = None,
         prompt: Optional[str] = None,
     ):

src/pipecat/services/ultravox.py

@@ -21,14 +21,15 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMTextFrame,
     StartFrame,
-    TranscriptionFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import AIService
-from pipecat.utils.time import time_now_iso8601
 
 try:
     from transformers import AutoTokenizer
@@ -339,6 +340,12 @@ class UltravoxSTTService(AIService):
             # Concatenate audio frames - all should be int16 now
             audio_data = np.concatenate(audio_arrays)
 
+            audio_int16 = audio_data  # Already in int16 format
+            # Save int16 audio
+
+            # Convert int16 to float32 and normalize for model input
+            audio_float32 = audio_int16.astype(np.float32) / 32768.0
+
             # Generate text using the model
             if self._model:
                 try:
@@ -349,11 +356,11 @@ class UltravoxSTTService(AIService):
                     await self.start_ttfb_metrics()
                     await self.start_processing_metrics()
 
-                    async for response in self._model.generate(
+                    async for response in self.model.generate(
                         messages=[{"role": "user", "content": "<|audio|>\n"}],
-                        temperature=self._temperature,
-                        max_tokens=self._max_tokens,
-                        audio=audio_data,
+                        temperature=self.temperature,
+                        max_tokens=self.max_tokens,
+                        audio=audio_float32,
                     ):
                         # Stop TTFB metrics after first response
                         await self.stop_ttfb_metrics()
@@ -369,18 +376,13 @@ class UltravoxSTTService(AIService):
                     await self.stop_processing_metrics()
 
                     logger.info(f"Generated text: {full_response}")
-
                     # Create a transcription frame with the generated text
-                    transcription = full_response.strip()
-                    if transcription:
-                        yield TranscriptionFrame(
-                            user_id="",
-                            text=transcription,
-                            timestamp=time_now_iso8601(),
-                        )
-                    else:
-                        logger.warning("Empty transcription result")
-                        yield ErrorFrame("Empty transcription result")
+                    yield LLMFullResponseStartFrame()
+
+                    text_frame = LLMTextFrame(text=full_response.strip())
+                    yield text_frame
+
+                    yield LLMFullResponseEndFrame()
 
                 except Exception as e:
                     logger.error(f"Error generating text from model: {e}")