diff --git a/changelog/3360.added.md b/changelog/3360.added.md new file mode 100644 index 000000000..5f5d92941 --- /dev/null +++ b/changelog/3360.added.md @@ -0,0 +1,6 @@ +- Added image support to `OpenAIRealtimeLLMService` via `InputImageRawFrame`: + - New `start_image_paused` parameter to control initial image input state + - New `image_detail` parameter to set image processing quality ("auto", "low", or "high") + - `set_image_input_paused()` method to pause/resume image input at runtime + - `set_image_detail()` method to adjust image quality dynamically + - Automatic rate limiting (1 image per second) to prevent API overload diff --git a/src/pipecat/services/openai/realtime/events.py b/src/pipecat/services/openai/realtime/events.py index 09d1e8ae4..c0ef5b890 100644 --- a/src/pipecat/services/openai/realtime/events.py +++ b/src/pipecat/services/openai/realtime/events.py @@ -217,16 +217,22 @@ class ItemContent(BaseModel): """Content within a conversation item. Parameters: - type: Content type (text, audio, input_text, input_audio, output_text, or output_audio). + type: Content type (text, audio, input_text, input_audio, input_image, output_text, or output_audio). text: Text content for text-based items. audio: Base64-encoded audio data for audio items. transcript: Transcribed text for audio items. + image_url: Base64-encoded image data as a data URI for input_image items. + detail: Detail level for image processing ("auto", "low", or "high"). """ - type: Literal["text", "audio", "input_text", "input_audio", "output_text", "output_audio"] + type: Literal[ + "text", "audio", "input_text", "input_audio", "input_image", "output_text", "output_audio" + ] text: Optional[str] = None audio: Optional[str] = None # base64-encoded audio transcript: Optional[str] = None + image_url: Optional[str] = None # base64-encoded image as data URI + detail: Optional[Literal["auto", "low", "high"]] = None class ConversationItem(BaseModel): diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 5cb50e16d..21c8e58e8 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -7,12 +7,14 @@ """OpenAI Realtime LLM service implementation with WebSocket support.""" import base64 +import io import json import time from dataclasses import dataclass from typing import Optional from loguru import logger +from PIL import Image from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.adapters.services.open_ai_realtime_adapter import ( @@ -25,6 +27,7 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, + InputImageRawFrame, InterimTranscriptionFrame, InterruptionFrame, LLMContextFrame, @@ -106,6 +109,8 @@ class OpenAIRealtimeLLMService(LLMService): base_url: str = "wss://api.openai.com/v1/realtime", session_properties: Optional[events.SessionProperties] = None, start_audio_paused: bool = False, + start_image_paused: bool = False, + image_detail: str = "auto", send_transcription_frames: Optional[bool] = None, **kwargs, ): @@ -122,6 +127,10 @@ class OpenAIRealtimeLLMService(LLMService): These are session-level settings that can be updated during the session (except for voice and model). If None, uses default SessionProperties. start_audio_paused: Whether to start with audio input paused. Defaults to False. + start_image_paused: Whether to start with image input paused. Defaults to False. + image_detail: Detail level for image processing. Can be "auto", "low", or "high". + "auto" lets the model decide, "low" is faster and uses fewer tokens, + "high" provides more detail. Defaults to "auto". send_transcription_frames: Whether to emit transcription frames. .. deprecated:: 0.0.92 @@ -156,6 +165,9 @@ class OpenAIRealtimeLLMService(LLMService): session_properties or events.SessionProperties() ) self._audio_input_paused = start_audio_paused + self._image_input_paused = start_image_paused + self._image_detail = image_detail + self._last_image_sent_time = 0 self._websocket = None self._receive_task = None self._context: LLMContext = None @@ -193,6 +205,25 @@ class OpenAIRealtimeLLMService(LLMService): """ self._audio_input_paused = paused + def set_image_input_paused(self, paused: bool): + """Set whether image input is paused. + + Args: + paused: True to pause image input, False to resume. + """ + self._image_input_paused = paused + + def set_image_detail(self, detail: str): + """Set the detail level for image processing. + + Args: + detail: Detail level - "auto", "low", or "high". + """ + if detail not in ["auto", "low", "high"]: + logger.warning(f"Invalid image detail '{detail}', must be 'auto', 'low', or 'high'") + return + self._image_detail = detail + def _is_modality_enabled(self, modality: str) -> bool: """Check if a specific modality is enabled, "text" or "audio".""" modalities = self._session_properties.output_modalities or ["audio", "text"] @@ -379,6 +410,9 @@ class OpenAIRealtimeLLMService(LLMService): elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) + elif isinstance(frame, InputImageRawFrame): + if not self._image_input_paused: + await self._send_user_image(frame) elif isinstance(frame, InterruptionFrame): await self._handle_interruption() elif isinstance(frame, UserStartedSpeakingFrame): @@ -847,6 +881,49 @@ class OpenAIRealtimeLLMService(LLMService): payload = base64.b64encode(frame.audio).decode("utf-8") await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload)) + async def _send_user_image(self, frame: InputImageRawFrame): + """Send user image frame to OpenAI Realtime API. + + Args: + frame: The input image frame to send. + """ + if self._image_input_paused or self._disconnecting or not self._websocket: + return + + now = time.time() + if now - self._last_image_sent_time < 1: + return # Ignore if less than 1 second has passed + + self._last_image_sent_time = now # Update last sent time + logger.debug(f"Sending image frame to OpenAI Realtime: {frame}") + + # Convert image to JPEG format and encode as base64 + buffer = io.BytesIO() + Image.frombytes(frame.format, frame.size, frame.image).save(buffer, format="JPEG") + image_data = base64.b64encode(buffer.getvalue()).decode("utf-8") + + # Create data URI for the image + image_url = f"data:image/jpeg;base64,{image_data}" + + # Create a conversation item with the image + item = events.ConversationItem( + type="message", + role="user", + content=[ + events.ItemContent( + type="input_image", + image_url=image_url, + detail=self._image_detail, + ) + ], + ) + + # Send the conversation item + try: + await self.send_client_event(events.ConversationItemCreateEvent(item=item)) + except Exception as e: + await self.push_error(error_msg=f"Send error: {e}") + async def _send_tool_result(self, tool_call_id: str, result: str): item = events.ConversationItem( type="function_call_output",