Add image input support for OpenAI Realtime

This commit is contained in:
Mark Backman
2026-01-06 14:11:46 -05:00
parent 8f83ba5878
commit 7ae9eebc34
3 changed files with 91 additions and 2 deletions

6
changelog/3360.added.md Normal file
View File

@@ -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

View File

@@ -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):

View File

@@ -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",