Merge pull request #1414 from pipecat-ai/march-main

March OpenAI updates

CHANGELOG.md

@@ -128,8 +128,52 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   Gemini models. Added foundational example
   `14p-function-calling-gemini-vertex-ai.py`.
 
+- Added support in `OpenAIRealtimeBetaLLMService` for a slate of new features:
+
+  - The `'gpt-4o-transcribe'` input audio transcription model, along
+    with new `language` and `prompt` options specific to that model.
+  - The `input_audio_noise_reduction` session property.
+
+    ```python
+    session_properties = SessionProperties(
+      # ...
+      input_audio_noise_reduction=InputAudioNoiseReduction(
+        type="near_field" # also supported: "far_field"
+      )
+      # ...
+    )
+    ```
+
+  - The `'semantic_vad'` `turn_detection` session property value, a more
+    sophisticated model for detecting when the user has stopped speaking.
+  - `on_conversation_item_created` and `on_conversation_item_updated`
+    events to `OpenAIRealtimeBetaLLMService`.
+
+    ```python
+    @llm.event_handler("on_conversation_item_created")
+    async def on_conversation_item_created(llm, item_id, item):
+      # ...
+
+    @llm.event_handler("on_conversation_item_updated")
+    async def on_conversation_item_updated(llm, item_id, item):
+      # `item` may not always be available here
+      # ...
+    ```
+
+  - The `retrieve_conversation_item(item_id)` method for introspecting a
+    conversation item on the server.
+
+    ```python
+    item = await llm.retrieve_conversation_item(item_id)
+    ```
+
 ### Changed
 
+- Updated `OpenAISTTService` to use `gpt-4o-transcribe` as the default
+  transcription model.
+
+- Updated `OpenAITTSService` to use `gpt-4o-mini-tts` as the default TTS model.
+
 - Function calls are now executed in tasks. This means that the pipeline will
   not be blocked while the function call is being executed.
 
@@ -216,6 +260,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed an issue in `RimeTTSService` where the last line of text sent didn't
   result in an audio output being generated.
 
+- Fixed `OpenAIRealtimeBetaLLMService` by adding proper handling for:
+  - The `conversation.item.input_audio_transcription.delta` server message,
+    which was added server-side at some point and not handled client-side.
+  - Errors reported by the `response.done` server message.
+
 ### Other
 
 - Add foundational example `07w-interruptible-fal.py`, showing `FalSTTService`.

examples/foundational/07g-interruptible-openai.py

@@ -51,16 +51,20 @@ async def main():
         #     api_key="gsk_***",
         #     model="whisper-large-v3",
         # )
-        stt = OpenAISTTService(api_key=os.getenv("OPENAI_API_KEY"), model="whisper-1")
+        stt = OpenAISTTService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o-transcribe-latest",
+            prompt="Expect words related to dogs, such as breed names.",
+        )
 
-        tts = OpenAITTSService(api_key=os.getenv("OPENAI_API_KEY"), voice="alloy")
+        tts = OpenAITTSService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini-tts-latest")
 
         llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
 
         messages = [
             {
                 "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. 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.",
+                "content": "You are very knowledgable about dogs. 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.",
             },
         ]
 

src/pipecat/services/openai.py

@@ -409,13 +409,13 @@ class OpenAIImageGenService(ImageGenService):
 
 
 class OpenAISTTService(BaseWhisperSTTService):
-    """OpenAI Whisper speech-to-text service.
+    """OpenAI Speech-to-Text service that generates text from audio.
 
-    Uses OpenAI's Whisper API to convert audio to text. Requires an OpenAI API key
+    Uses OpenAI's transcription API to convert audio to text. Requires an OpenAI API key
     set via the api_key parameter or OPENAI_API_KEY environment variable.
 
     Args:
-        model: Whisper model to use. Defaults to "whisper-1".
+        model: Model to use — either gpt-4o or Whisper. Defaults to "gpt-4o-transcribe".
         api_key: OpenAI API key. Defaults to None.
         base_url: API base URL. Defaults to None.
         language: Language of the audio input. Defaults to English.
@@ -427,7 +427,7 @@ class OpenAISTTService(BaseWhisperSTTService):
     def __init__(
         self,
         *,
-        model: str = "whisper-1",
+        model: str = "gpt-4o-transcribe",
         api_key: Optional[str] = None,
         base_url: Optional[str] = None,
         language: Optional[Language] = Language.EN,
@@ -472,7 +472,7 @@ class OpenAITTSService(TTSService):
     Args:
         api_key: OpenAI API key. Defaults to None.
         voice: Voice ID to use. Defaults to "alloy".
-        model: TTS model to use. Defaults to "tts-1".
+        model: TTS model to use. Defaults to "gpt-4o-mini-tts".
         sample_rate: Output audio sample rate in Hz. Defaults to None.
         **kwargs: Additional keyword arguments passed to TTSService.
 
@@ -487,7 +487,7 @@ class OpenAITTSService(TTSService):
         *,
         api_key: Optional[str] = None,
         voice: str = "alloy",
-        model: str = "tts-1",
+        model: str = "gpt-4o-mini-tts",
         sample_rate: Optional[int] = None,
         **kwargs,
     ):

src/pipecat/services/openai_realtime_beta/__init__.py

@@ -1,3 +1,9 @@
 from .azure import AzureRealtimeBetaLLMService
-from .events import InputAudioTranscription, SessionProperties, TurnDetection
+from .events import (
+    InputAudioNoiseReduction,
+    InputAudioTranscription,
+    SemanticTurnDetection,
+    SessionProperties,
+    TurnDetection,
+)
 from .openai import OpenAIRealtimeBetaLLMService

src/pipecat/services/openai_realtime_beta/context.py

@@ -12,6 +12,7 @@ from loguru import logger
 
 from pipecat.frames.frames import (
     Frame,
+    FunctionCallResultFrame,
     FunctionCallResultProperties,
     LLMMessagesUpdateFrame,
     LLMSetToolsFrame,
@@ -174,67 +175,12 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator):
-    async def push_aggregation(self):
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
-        # the only thing we implement here is function calling. in all other cases, messages
+        await super().handle_function_call_result(frame)
-        # are added to the context when we receive openai realtime api events
-        if not self._function_call_result:
-            return
 
-        properties: Optional[FunctionCallResultProperties] = None
+        # The standard function callback code path pushes the FunctionCallResultFrame from the llm itself,
-
+        # so we didn't have a chance to add the result to the openai realtime api context. Let's push a
-        self.reset()
+        # special frame to do that.
-        try:
+        await self.push_frame(
-            run_llm = True
+            RealtimeFunctionCallResultFrame(result_frame=frame), FrameDirection.UPSTREAM
-            frame = self._function_call_result
+        )
-            properties = frame.properties
-            self._function_call_result = None
-            if frame.result:
-                # The "tool_call" message from the LLM that triggered the function call
-                self._context.add_message(
-                    {
-                        "role": "assistant",
-                        "tool_calls": [
-                            {
-                                "id": frame.tool_call_id,
-                                "function": {
-                                    "name": frame.function_name,
-                                    "arguments": json.dumps(frame.arguments),
-                                },
-                                "type": "function",
-                            }
-                        ],
-                    }
-                )
-                # The result of the function call. Need to add this both to our context here and to
-                # the openai realtime api context.
-                result_message = {
-                    "role": "tool",
-                    "content": json.dumps(frame.result),
-                    "tool_call_id": frame.tool_call_id,
-                }
-
-                self._context.add_message(result_message)
-                # The standard function callback code path pushes the FunctionCallResultFrame from the llm itself,
-                # so we didn't have a chance to add the result to the openai realtime api context. Let's push a
-                # special frame to do that.
-                await self.push_frame(
-                    RealtimeFunctionCallResultFrame(result_frame=frame), FrameDirection.UPSTREAM
-                )
-                if properties and properties.run_llm is not None:
-                    # If the tool call result has a run_llm property, use it
-                    run_llm = properties.run_llm
-                else:
-                    # Default behavior is to run the LLM if there are no function calls in progress
-                    run_llm = not bool(self._function_calls_in_progress)
-
-            if run_llm:
-                await self.push_context_frame(FrameDirection.UPSTREAM)
-
-            # Emit the on_context_updated callback once the function call result is added to the context
-            if properties and properties.on_context_updated is not None:
-                await properties.on_context_updated()
-
-            await self.push_context_frame()
-
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")

src/pipecat/services/openai_realtime_beta/events.py

@@ -14,10 +14,25 @@ from pydantic import BaseModel, Field
 #
 # session properties
 #
+InputAudioTranscriptionModel = Literal["whisper-1", "gpt-4o-transcribe"]
 
 
 class InputAudioTranscription(BaseModel):
-    model: Optional[str] = "whisper-1"
+    model: InputAudioTranscriptionModel
+    language: Optional[str]
+    prompt: Optional[str]
+
+    def __init__(
+        self,
+        model: Optional[InputAudioTranscriptionModel] = "whisper-1",
+        language: Optional[str] = None,
+        prompt: Optional[str] = None,
+    ):
+        super().__init__(model=model, language=language, prompt=prompt)
+        if self.model != "gpt-4o-transcribe" and (self.language or self.prompt):
+            raise ValueError(
+                "Fields 'language' and 'prompt' are only supported when model is 'gpt-4o-transcribe'"
+            )
 
 
 class TurnDetection(BaseModel):
@@ -27,6 +42,17 @@ class TurnDetection(BaseModel):
     silence_duration_ms: Optional[int] = 800
 
 
+class SemanticTurnDetection(BaseModel):
+    type: Optional[Literal["semantic_vad"]] = "semantic_vad"
+    eagerness: Optional[Literal["low", "medium", "high", "auto"]] = None
+    create_response: Optional[bool] = None
+    interrupt_response: Optional[bool] = None
+
+
+class InputAudioNoiseReduction(BaseModel):
+    type: Optional[Literal["near_field", "far_field"]]
+
+
 class SessionProperties(BaseModel):
     modalities: Optional[List[Literal["text", "audio"]]] = None
     instructions: Optional[str] = None
@@ -34,8 +60,11 @@ class SessionProperties(BaseModel):
     input_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None
     output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None
     input_audio_transcription: Optional[InputAudioTranscription] = None
+    input_audio_noise_reduction: Optional[InputAudioNoiseReduction] = None
     # set turn_detection to False to disable turn detection
-    turn_detection: Optional[Union[TurnDetection, bool]] = Field(default=None)
+    turn_detection: Optional[Union[TurnDetection, SemanticTurnDetection, bool]] = Field(
+        default=None
+    )
     tools: Optional[List[Dict]] = None
     tool_choice: Optional[Literal["auto", "none", "required"]] = None
     temperature: Optional[float] = None
@@ -93,6 +122,7 @@ class RealtimeError(BaseModel):
     code: Optional[str] = ""
     message: str
     param: Optional[str] = None
+    event_id: Optional[str] = None
 
 
 #
@@ -150,6 +180,11 @@ class ConversationItemDeleteEvent(ClientEvent):
     item_id: str
 
 
+class ConversationItemRetrieveEvent(ClientEvent):
+    type: Literal["conversation.item.retrieve"] = "conversation.item.retrieve"
+    item_id: str
+
+
 class ResponseCreateEvent(ClientEvent):
     type: Literal["response.create"] = "response.create"
     response: Optional[ResponseProperties] = None
@@ -193,6 +228,13 @@ class ConversationItemCreated(ServerEvent):
     item: ConversationItem
 
 
+class ConversationItemInputAudioTranscriptionDelta(ServerEvent):
+    type: Literal["conversation.item.input_audio_transcription.delta"]
+    item_id: str
+    content_index: int
+    delta: str
+
+
 class ConversationItemInputAudioTranscriptionCompleted(ServerEvent):
     type: Literal["conversation.item.input_audio_transcription.completed"]
     item_id: str
@@ -219,6 +261,11 @@ class ConversationItemDeleted(ServerEvent):
     item_id: str
 
 
+class ConversationItemRetrieved(ServerEvent):
+    type: Literal["conversation.item.retrieved"]
+    item: ConversationItem
+
+
 class ResponseCreated(ServerEvent):
     type: Literal["response.created"]
     response: "Response"
@@ -400,10 +447,12 @@ _server_event_types = {
     "input_audio_buffer.speech_started": InputAudioBufferSpeechStarted,
     "input_audio_buffer.speech_stopped": InputAudioBufferSpeechStopped,
     "conversation.item.created": ConversationItemCreated,
+    "conversation.item.input_audio_transcription.delta": ConversationItemInputAudioTranscriptionDelta,
     "conversation.item.input_audio_transcription.completed": ConversationItemInputAudioTranscriptionCompleted,
     "conversation.item.input_audio_transcription.failed": ConversationItemInputAudioTranscriptionFailed,
     "conversation.item.truncated": ConversationItemTruncated,
     "conversation.item.deleted": ConversationItemDeleted,
+    "conversation.item.retrieved": ConversationItemRetrieved,
     "response.created": ResponseCreated,
     "response.done": ResponseDone,
     "response.output_item.added": ResponseOutputItemAdded,

src/pipecat/services/openai_realtime_beta/openai.py

@@ -30,6 +30,7 @@ from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
     InputAudioRawFrame,
+    InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -115,12 +116,35 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         self._messages_added_manually = {}
         self._user_and_response_message_tuple = None
 
+        self._register_event_handler("on_conversation_item_created")
+        self._register_event_handler("on_conversation_item_updated")
+        self._retrieve_conversation_item_futures = {}
+
     def can_generate_metrics(self) -> bool:
         return True
 
     def set_audio_input_paused(self, paused: bool):
         self._audio_input_paused = paused
 
+    async def retrieve_conversation_item(self, item_id: str):
+        future = self.get_event_loop().create_future()
+        retrieval_in_flight = False
+        if not self._retrieve_conversation_item_futures.get(item_id):
+            self._retrieve_conversation_item_futures[item_id] = []
+        else:
+            retrieval_in_flight = True
+        self._retrieve_conversation_item_futures[item_id].append(future)
+        if not retrieval_in_flight:
+            await self.send_client_event(
+                # Set event_id to "rci_{item_id}" so that we can identify an
+                # error later if the retrieval fails. We don't need a UUID
+                # suffix to the event_id because we're ensuring only one
+                # in-flight retrieval per item_id. (Note: "rci" = "retrieve
+                # conversation item")
+                events.ConversationItemRetrieveEvent(item_id=item_id, event_id=f"rci_{item_id}")
+            )
+        return await future
+
     #
     # standard AIService frame handling
     #
@@ -354,8 +378,12 @@ class OpenAIRealtimeBetaLLMService(LLMService):
                 await self._handle_evt_audio_done(evt)
             elif evt.type == "conversation.item.created":
                 await self._handle_evt_conversation_item_created(evt)
+            elif evt.type == "conversation.item.input_audio_transcription.delta":
+                await self._handle_evt_input_audio_transcription_delta(evt)
             elif evt.type == "conversation.item.input_audio_transcription.completed":
                 await self.handle_evt_input_audio_transcription_completed(evt)
+            elif evt.type == "conversation.item.retrieved":
+                await self._handle_conversation_item_retrieved(evt)
             elif evt.type == "response.done":
                 await self._handle_evt_response_done(evt)
             elif evt.type == "input_audio_buffer.speech_started":
@@ -365,9 +393,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             elif evt.type == "response.audio_transcript.delta":
                 await self._handle_evt_audio_transcript_delta(evt)
             elif evt.type == "error":
-                await self._handle_evt_error(evt)
+                if not await self._maybe_handle_evt_retrieve_conversation_item_error(evt):
-                # errors are fatal, so exit the receive loop
+                    await self._handle_evt_error(evt)
-                return
+                    # errors are fatal, so exit the receive loop
+                    return
 
     async def _handle_evt_session_created(self, evt):
         # session.created is received right after connecting. Send a message
@@ -409,6 +438,8 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             # receive a BotStoppedSpeakingFrame from the output transport.
 
     async def _handle_evt_conversation_item_created(self, evt):
+        await self._call_event_handler("on_conversation_item_created", evt.item.id, evt.item)
+
         # This will get sent from the server every time a new "message" is added
         # to the server's conversation state, whether we create it via the API
         # or the server creates it from LLM output.
@@ -425,7 +456,16 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             self._current_assistant_response = evt.item
             await self.push_frame(LLMFullResponseStartFrame())
 
+    async def _handle_evt_input_audio_transcription_delta(self, evt):
+        if self._send_transcription_frames:
+            await self.push_frame(
+                # no way to get a language code?
+                InterimTranscriptionFrame(evt.delta, "", time_now_iso8601())
+            )
+
     async def handle_evt_input_audio_transcription_completed(self, evt):
+        await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
+
         if self._send_transcription_frames:
             await self.push_frame(
                 # no way to get a language code?
@@ -443,6 +483,12 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             # User message without preceding conversation.item.created. Bug?
             logger.warning(f"Transcript for unknown user message: {evt}")
 
+    async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved):
+        futures = self._retrieve_conversation_item_futures.pop(evt.item.id, None)
+        if futures:
+            for future in futures:
+                future.set_result(evt.item)
+
     async def _handle_evt_response_done(self, evt):
         # todo: figure out whether there's anything we need to do for "cancelled" events
         # usage metrics
@@ -455,7 +501,15 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         await self.stop_processing_metrics()
         await self.push_frame(LLMFullResponseEndFrame())
         self._current_assistant_response = None
+        # error handling
+        if evt.response.status == "failed":
+            await self.push_error(
+                ErrorFrame(error=evt.response.status_details["error"]["message"], fatal=True)
+            )
+            return
         # response content
+        for item in evt.response.output:
+            await self._call_event_handler("on_conversation_item_updated", item.id, item)
         pair = self._user_and_response_message_tuple
         if pair:
             user, assistant = pair
@@ -487,6 +541,22 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         await self.push_frame(StopInterruptionFrame())
         await self.push_frame(UserStoppedSpeakingFrame())
 
+    async def _maybe_handle_evt_retrieve_conversation_item_error(self, evt: events.ErrorEvent):
+        """If the given error event is an error retrieving a conversation item:
+        - set an exception on the future that retrieve_conversation_item() is waiting on
+        - return true
+        Otherwise:
+        - return false
+        """
+        if evt.error.code == "item_retrieve_invalid_item_id":
+            item_id = evt.error.event_id.split("_", 1)[1]  # event_id is of the form "rci_{item_id}"
+            futures = self._retrieve_conversation_item_futures.pop(item_id, None)
+            if futures:
+                for future in futures:
+                    future.set_exception(Exception(evt.error.message))
+            return True
+        return False
+
     async def _handle_evt_error(self, evt):
         # Errors are fatal to this connection. Send an ErrorFrame.
         await self.push_error(ErrorFrame(error=f"Error: {evt}", fatal=True))
@@ -509,7 +579,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             arguments = json.loads(item.arguments)
             if self.has_function(function_name):
                 run_llm = index == total_items - 1
-                if function_name in self._callbacks.keys():
+                if function_name in self._functions.keys():
                     await self.call_function(
                         context=self._context,
                         tool_call_id=tool_id,
@@ -517,7 +587,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
                         arguments=arguments,
                         run_llm=run_llm,
                     )
-                elif None in self._callbacks.keys():
+                elif None in self._functions.keys():
                     await self.call_function(
                         context=self._context,
                         tool_call_id=tool_id,