diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d55b97c3..065a5ced6 100644 --- a/CHANGELOG.md +++ b/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`. diff --git a/examples/foundational/07g-interruptible-openai.py b/examples/foundational/07g-interruptible-openai.py index f9cac5910..b66d0346a 100644 --- a/examples/foundational/07g-interruptible-openai.py +++ b/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.", }, ] diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index b5f5d9d54..4d3e1d6d7 100644 --- a/src/pipecat/services/openai.py +++ b/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, ): diff --git a/src/pipecat/services/openai_realtime_beta/__init__.py b/src/pipecat/services/openai_realtime_beta/__init__.py index 52b00f6c8..595105d7f 100644 --- a/src/pipecat/services/openai_realtime_beta/__init__.py +++ b/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 diff --git a/src/pipecat/services/openai_realtime_beta/context.py b/src/pipecat/services/openai_realtime_beta/context.py index 31639dc6b..c8381976f 100644 --- a/src/pipecat/services/openai_realtime_beta/context.py +++ b/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): - # the only thing we implement here is function calling. in all other cases, messages - # are added to the context when we receive openai realtime api events - if not self._function_call_result: - return + async def handle_function_call_result(self, frame: FunctionCallResultFrame): + await super().handle_function_call_result(frame) - properties: Optional[FunctionCallResultProperties] = None - - self.reset() - try: - run_llm = True - 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}") + # 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 + ) diff --git a/src/pipecat/services/openai_realtime_beta/events.py b/src/pipecat/services/openai_realtime_beta/events.py index 17ce0a6d4..f4133766b 100644 --- a/src/pipecat/services/openai_realtime_beta/events.py +++ b/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, diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 321c66826..a1f4bc731 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/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) - # errors are fatal, so exit the receive loop - return + if not await self._maybe_handle_evt_retrieve_conversation_item_error(evt): + await self._handle_evt_error(evt) + # 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,