utils

Helpers for model client for integrating models and parsing the output.

Functions

extract_complete_text_from_response_stream(event)	Extract complete text from response completion events.
extract_text_from_response_stream(event)	Extract text content from OpenAI Response API streaming events.
format_content_for_response_api(text[, images])	Format text and optional images into content array for responses.create API.
is_response_complete(event)	Check if a streaming event indicates the response is complete.
parse_embedding_response(api_response)	Parse embedding model output from the API response to EmbedderOutput.
process_images_for_response_api(images[, ...])	Process and validate images for OpenAI's responses.create API.

parse_embedding_response(api_response) → EmbedderOutput

Parse embedding model output from the API response to EmbedderOutput.

Follows the OpenAI API response pattern.

process_images_for_response_api(images: str | Dict | List[str | Dict], encode_local_images: bool = True) → List[Dict[str, Any]]

Process and validate images for OpenAI’s responses.create API.

This function handles various image input formats and converts them to the expected format for the responses.create API.

Parameters:

• images – Can be: - A single image URL (str) - A single local file path (str) - A pre-formatted image dict with type=’input_image’ - A list containing any combination of the above

• encode_local_images – Whether to encode local image files to base64

Returns:

• type: “input_image”

• image_url: Either a URL or base64-encoded data URI

Return type:

List of formatted image dicts ready for the API, each containing

Raises:

• ValueError – If image dict format is invalid

• FileNotFoundError – If local image file doesn’t exist

Examples

>>> # Single URL
>>> process_images_for_response_api("https://example.com/image.jpg")
[{"type": "input_image", "image_url": "https://example.com/image.jpg"}]

>>> # Local file
>>> process_images_for_response_api("/path/to/image.jpg")
[{"type": "input_image", "image_url": "data:image/jpeg;base64,..."}]

>>> # Multiple mixed sources
>>> process_images_for_response_api([
...     "https://example.com/img.jpg",
...     "/local/img.png",
...     {"type": "input_image", "image_url": "..."}
... ])
[...]

format_content_for_response_api(text: str, images: str | Dict | List[str | Dict] | None = None) → List[Dict[str, Any]]

Format text and optional images into content array for responses.create API.

Parameters:

• text – The text prompt/question

• images – Optional images in various formats (see process_images_for_response_api)

Returns:

List of content items formatted for the API

Examples

>>> # Text only
>>> format_content_for_response_api("What is this?")
[{"type": "input_text", "text": "What is this?"}]

>>> # Text with image
>>> format_content_for_response_api(
...     "What's in this image?",
...     "https://example.com/img.jpg"
... )
[
    {"type": "input_text", "text": "What's in this image?"},
    {"type": "input_image", "image_url": "https://example.com/img.jpg"}
]

extract_text_from_response_stream(event) → str | None

Extract text content from OpenAI Response API streaming events.

The Response API generates various event types during streaming: - ResponseCreatedEvent: Initial event when response starts - ResponseInProgressEvent: Status updates - ResponseOutputItemAddedEvent: When new output items are added - ResponseContentPartAddedEvent: When content parts are added - ResponseTextDeltaEvent: Contains actual text chunks - ResponseTextDoneEvent: When text generation completes - ResponseDoneEvent: Final event when response completes

This function extracts text only from ResponseTextDeltaEvent types.

Parameters:

event – A streaming event from OpenAI’s Response API

Returns:

The text delta if this is a text delta event, None otherwise

Return type:

str

Examples

>>> # In a streaming response handler
>>> async for event in stream:
...     text = extract_text_from_response_stream(event)
...     if text:
...         print(text, end="", flush=True)

extract_complete_text_from_response_stream(event) → str | None

Extract complete text from response completion events.

Some events contain the complete text rather than deltas: - ResponseTextDoneEvent: Contains the complete text when done - Response objects with output_text property

Parameters:

event – A streaming event from OpenAI’s Response API

Returns:

The complete text if available, None otherwise

Return type:

str

is_response_complete(event) → bool

Check if a streaming event indicates the response is complete.

Parameters:

event – A streaming event from OpenAI’s Response API

Returns:

True if this event indicates completion, False otherwise

Return type:

bool