feat: responses streaming helpers (#721)

* add responses streaming helpers

* refactor tests

* fix helpers.md

* add param canonization step to stream

* update README streaming example

* support parsed responses in the final response

* misc: PR touch ups

---------

Co-authored-by: dogisgreat <[email protected]>

Author: 2 people

README.md

@@ -42,16 +42,14 @@ puts(chat_completion)
 
 We provide support for streaming responses using Server-Sent Events (SSE).
 
-**coming soon:** `openai.chat.completions.stream` will soon come with Python SDK-style higher-level streaming responses support.
-
 ```ruby
-stream = openai.chat.completions.stream_raw(
-  messages: [{role: "user", content: "Say this is a test"}],
+stream = openai.responses.stream(
+  input: "Write a haiku about OpenAI.",
   model: :"gpt-4.1"
 )
 
-stream.each do |completion|
-  puts(completion)
+stream.each do |event|
+  puts(event.type)
 end
 ```
 

examples/responses/streaming_basic.rb

@@ -0,0 +1,23 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+# typed: strict
+
+require_relative "../../lib/openai"
+
+client = OpenAI::Client.new
+
+stream = client.responses.stream(
+  input: "Write a haiku about OpenAI.",
+  model: "gpt-4o-2024-08-06"
+)
+
+stream.each do |event|
+  case event
+  when OpenAI::Streaming::ResponseTextDeltaEvent
+    print(event.delta)
+  when OpenAI::Streaming::ResponseTextDoneEvent
+    puts("\n--------------------------")
+  when OpenAI::Streaming::ResponseCompletedEvent
+    puts("Response completed! (response id: #{event.response.id})")
+  end
+end

examples/responses/streaming_previous_response.rb

@@ -0,0 +1,79 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require_relative "../../lib/openai"
+
+# This example demonstrates how to resume a streaming response.
+
+client = OpenAI::Client.new
+
+# Request 1: Create a new streaming response with store=true
+puts "Creating a new streaming response..."
+stream = client.responses.stream(
+  model: "o4-mini",
+  input: "Tell me a short story about a robot learning to paint.",
+  instructions: "You are a creative storyteller.",
+  background: true
+)
+
+events = []
+response_id = ""
+
+stream.each do |event|
+  events << event
+  puts "Event from initial stream: #{event.type} (seq: #{event.sequence_number})"
+  case event
+
+  when OpenAI::Models::Responses::ResponseCreatedEvent
+    response_id = event.response.id if response_id.empty?
+    puts("Captured response ID: #{response_id}")
+  end
+
+  # Simulate stopping after a few events
+  if events.length >= 5
+    puts "Terminating after #{events.length} events"
+    break
+  end
+end
+
+stream.close
+
+puts
+puts "Collected #{events.length} events"
+puts "Response ID: #{response_id}"
+puts "Last event sequence number: #{events.last.sequence_number}.\n"
+
+# Give the background response some time to process more events.
+puts "Waiting a moment for the background response to progress...\n"
+sleep(2)
+
+# Request 2: Resume the stream using the captured response_id.
+puts "Resuming stream from sequence #{events.last.sequence_number}..."
+
+resumed_stream = client.responses.stream(
+  previous_response_id: response_id,
+  starting_after: events.last.sequence_number
+)
+
+resumed_events = []
+resumed_stream.each do |event|
+  resumed_events << event
+  puts "Event from resumed stream: #{event.type} (seq: #{event.sequence_number})"
+  # Stop when we get the completed event or collect enough events.
+  if event.is_a?(OpenAI::Models::Responses::ResponseCompletedEvent)
+    puts "Response completed!"
+    break
+  end
+
+  break if resumed_events.length >= 10
+end
+
+puts "\nCollected #{resumed_events.length} additional events"
+
+# Show that we properly resumed from where we left off.
+if resumed_events.any?
+  first_resumed_event = resumed_events.first
+  last_initial_event = events.last
+  puts "First resumed event sequence: #{first_resumed_event.sequence_number}"
+  puts "Should be greater than last initial event: #{last_initial_event.sequence_number}"
+end

examples/responses/streaming_structured_outputs.rb

@@ -0,0 +1,46 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require_relative "../../lib/openai"
+
+# Defining structured output models.
+class Step < OpenAI::BaseModel
+  required :explanation, String
+  required :output, String
+end
+
+class MathResponse < OpenAI::BaseModel
+  required :steps, OpenAI::ArrayOf[Step]
+  required :final_answer, String
+end
+
+client = OpenAI::Client.new
+
+stream = client.responses.stream(
+  input: "solve 8x + 31 = 2",
+  model: "gpt-4o-2024-08-06",
+  text: MathResponse
+)
+
+stream.each do |event|
+  case event
+  when OpenAI::Streaming::ResponseTextDeltaEvent
+    print(event.delta)
+  when OpenAI::Streaming::ResponseTextDoneEvent
+    puts
+    puts("--- Parsed object ---")
+    pp(event.parsed)
+  end
+end
+
+response = stream.get_final_response
+
+puts
+puts("----- parsed outputs from final response -----")
+response
+  .output
+  .flat_map { _1.content }
+  .each do |content|
+    # parsed is an instance of `MathResponse`
+    pp(content.parsed)
+  end

examples/responses/streaming_text.rb

@@ -0,0 +1,21 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+# typed: strong
+
+require_relative "../../lib/openai"
+
+client = OpenAI::Client.new
+
+stream = client.responses.stream(
+  input: "Write a haiku about OpenAI.",
+  model: "gpt-4o-2024-08-06"
+)
+
+stream.text.each do |text|
+  print(text)
+end
+
+puts
+
+# Get all of the text that was streamed with .get_output_text
+puts "Character count: #{stream.get_output_text.length}"

examples/responses/streaming_tools.rb

@@ -0,0 +1,63 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+# typed: true
+
+require_relative "../../lib/openai"
+
+class DynamicValue < OpenAI::BaseModel
+  required :column_name, String
+end
+
+class Condition < OpenAI::BaseModel
+  required :column, String
+  required :operator, OpenAI::EnumOf[:eq, :gt, :lt, :le, :ge, :ne]
+  required :value, OpenAI::UnionOf[String, Integer, DynamicValue]
+end
+
+# you can assign `OpenAI::{...}` schema specifiers to a constant
+Columns = OpenAI::EnumOf[
+  :id,
+  :status,
+  :expected_delivery_date,
+  :delivered_at,
+  :shipped_at,
+  :ordered_at,
+  :canceled_at
+]
+
+class Query < OpenAI::BaseModel
+  required :table_name, OpenAI::EnumOf[:orders, :customers, :products]
+  required :columns, OpenAI::ArrayOf[Columns]
+  required :conditions, OpenAI::ArrayOf[Condition]
+  required :order_by, OpenAI::EnumOf[:asc, :desc]
+end
+
+client = OpenAI::Client.new
+
+stream = client.responses.stream(
+  model: "gpt-4o-2024-08-06",
+  input: "look up all my orders in november of last year that were fulfilled but not delivered on time",
+  tools: [Query]
+)
+
+stream.each do |event|
+  case event
+  when OpenAI::Streaming::ResponseFunctionCallArgumentsDeltaEvent
+    puts("delta: #{event.delta}")
+    puts("snapshot: #{event.snapshot}")
+  end
+end
+
+response = stream.get_final_response
+
+puts
+puts("----- parsed outputs from final response -----")
+response
+  .output
+  .each do |output|
+    case output
+    when OpenAI::Models::Responses::ResponseFunctionToolCall
+      # parsed is an instance of `Query`
+      pp(output.parsed)
+    end
+  end