pyrit.prompt_target.RealtimeTarget

pyrit.prompt_target.RealtimeTarget#

class RealtimeTarget(*, voice: Literal['alloy', 'echo', 'shimmer'] | None = None, existing_convo: dict | None = None, **kwargs)[source]#

Bases: OpenAITarget

__init__(*, voice: Literal['alloy', 'echo', 'shimmer'] | None = None, existing_convo: dict | None = None, **kwargs) → None[source]#

RealtimeTarget class for Azure OpenAI Realtime API.

Read more at https://learn.microsoft.com/en-us/azure/ai-services/openai/realtime-audio-reference and https://platform.openai.com/docs/guides/realtime-websocket

Parameters:

  • model_name (str, Optional) – The name of the model. If no value is provided, the OPENAI_REALTIME_MODEL environment variable will be used.

  • endpoint (str, Optional) – The target URL for the OpenAI service. Defaults to the OPENAI_REALTIME_ENDPOINT environment variable.

  • api_key (str | Callable[[], str], Optional) – The API key for accessing the OpenAI service, or a callable that returns an access token. For Azure endpoints with Entra authentication, pass a token provider from pyrit.auth (e.g., get_azure_openai_auth(endpoint)). Defaults to the OPENAI_REALTIME_API_KEY environment variable.

  • headers (str, Optional) – Headers of the endpoint (JSON).

  • max_requests_per_minute (int, Optional) – Number of requests the target can handle per minute before hitting a rate limit. The number of requests sent to the target will be capped at the value provided.

  • voice (literal str, Optional) – The voice to use. Defaults to None. the only supported voices by the AzureOpenAI Realtime API are “alloy”, “echo”, and “shimmer”.

  • existing_convo (dict[str, websockets.WebSocketClientProtocol], Optional) – Existing conversations.

  • httpx_client_kwargs (dict, Optional) – Additional kwargs to be passed to the httpx.AsyncClient() constructor. For example, to specify a 3 minutes timeout: httpx_client_kwargs={“timeout”: 180}

Methods

__init__(*[, voice, existing_convo])

RealtimeTarget class for Azure OpenAI Realtime API.

cleanup_conversation(conversation_id)

Disconnects from the Realtime API for a specific conversation.

cleanup_target()

Disconnects from the Realtime API connections.

connect(conversation_id)

Connects to Realtime API using AsyncOpenAI client.

dispose_db_engine()

Dispose database engine to release database connections and resources.

get_identifier()

is_json_response_supported()

Indicates that this target supports JSON response format.

is_response_format_json(message_piece)

Checks if the response format is JSON and ensures the target supports it.

receive_events(conversation_id)

Continuously receive events from the OpenAI Realtime API connection.

save_audio(audio_bytes[, num_channels, ...])

Saves audio bytes to a WAV file.

send_audio_async(filename, conversation_id)

Send an audio message using OpenAI Realtime API client.

send_config(conversation_id)

Sends the session configuration using OpenAI client.

send_prompt_async(**kwargs)

Sends a normalized prompt async to the prompt target.

send_response_create(conversation_id)

Sends response.create using OpenAI client.

send_text_async(text, conversation_id)

Sends text prompt using OpenAI Realtime API client.

set_model_name(*, model_name)

Set the model name for this target.

set_system_prompt(*, system_prompt, ...[, ...])

Sets the system prompt for the prompt target.

Attributes

ADDITIONAL_REQUEST_HEADERS

model_name_environment_variable

endpoint_environment_variable

api_key_environment_variable

supported_converters

api_key_environment_variable: str#
async cleanup_conversation(conversation_id: str)[source]#

Disconnects from the Realtime API for a specific conversation.

Parameters:

conversation_id (str) – The conversation ID to disconnect from.

async cleanup_target()[source]#

Disconnects from the Realtime API connections.

async connect(conversation_id: str)[source]#

Connects to Realtime API using AsyncOpenAI client. Returns the realtime connection.

endpoint_environment_variable: str#
is_json_response_supported() → bool[source]#

Indicates that this target supports JSON response format.

model_name_environment_variable: str#
async receive_events(conversation_id: str) → RealtimeTargetResult[source]#

Continuously receive events from the OpenAI Realtime API connection.

Uses a robust “soft-finish” strategy to handle cases where response.done may not arrive. After receiving audio.done, waits for a grace period before soft-finishing if no response.done arrives.

Parameters:

conversation_id – conversation ID

Returns:

RealtimeTargetResult with audio data and transcripts

Raises:
async save_audio(audio_bytes: bytes, num_channels: int = 1, sample_width: int = 2, sample_rate: int = 16000, output_filename: str | None = None) → str[source]#

Saves audio bytes to a WAV file.

Parameters:

  • audio_bytes (bytes) – Audio bytes to save.

  • num_channels (int) – Number of audio channels. Defaults to 1 for the PCM16 format

  • sample_width (int) – Sample width in bytes. Defaults to 2 for the PCM16 format

  • sample_rate (int) – Sample rate in Hz. Defaults to 16000 Hz for the PCM16 format

  • output_filename (str) – Output filename. If None, a UUID filename will be used.

Returns:

The path to the saved audio file.

Return type:

str

async send_audio_async(filename: str, conversation_id: str) → Tuple[str, RealtimeTargetResult][source]#

Send an audio message using OpenAI Realtime API client.

Parameters:

  • filename (str) – The path to the audio file.

  • conversation_id (str) – Conversation ID

async send_config(conversation_id: str)[source]#

Sends the session configuration using OpenAI client.

Parameters:

conversation_id (str) – Conversation ID

async send_prompt_async(**kwargs)#

Sends a normalized prompt async to the prompt target.

Returns:

A list of message responses. Most targets return a single message,

but some (like response target with tool calls) may return multiple messages.

Return type:

list[Message]

async send_response_create(conversation_id: str)[source]#

Sends response.create using OpenAI client.

Parameters:

conversation_id (str) – Conversation ID

async send_text_async(text: str, conversation_id: str) → Tuple[str, RealtimeTargetResult][source]#

Sends text prompt using OpenAI Realtime API client.

Parameters:

  • text – prompt to send.

  • conversation_id – conversation ID

supported_converters: list#
Contents