fix: provide parsed outputs for resumed streams (#756)

* provide `parsed` to final response on resumed stream

* update example

* remove stream.close calls

* use internal method

---------

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

Author: 2 people

examples/responses/streaming_previous_response.rb

@@ -7,73 +7,150 @@
 
 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}")
+begin
+  puts "----- resuming stream from a previous response -----"
+
+  # Request 1: Create a new streaming response with background=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
 
-  # Simulate stopping after a few events
-  if events.length >= 5
-    puts "Terminating after #{events.length} events"
-    break
+  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(3)
+
+  # Request 2: Resume the stream using the captured response_id.
+  puts
+  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 "Collected #{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
 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
+begin
+  puts "\n----- resuming stream with structured outputs -----"
+
+  class Step < OpenAI::BaseModel
+    required :explanation, String
+    required :output, String
   end
 
-  break if resumed_events.length >= 10
-end
+  class MathResponse < OpenAI::BaseModel
+    required :steps, OpenAI::ArrayOf[Step]
+    required :final_answer, String
+  end
+
+  puts "Creating a background streaming response with structured output..."
+  stream = client.responses.stream(
+    input: "solve 8x + 31 = 2",
+    model: "gpt-4o-2024-08-06",
+    text: MathResponse,
+    background: true
+  )
 
-puts "\nCollected #{resumed_events.length} additional events"
+  events = []
+  response_id = ""
+
+  stream.each do |event|
+    events << event
+
+    case event
+    when OpenAI::Models::Responses::ResponseCreatedEvent
+      response_id = event.response.id if response_id.empty?
+    end
+
+    if events.length >= 5
+      break
+    end
+  end
+
+  puts "Waiting for the background response to complete...\n"
+  sleep(3)
+
+  puts
+  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,
+    # NOTE: You must pass the structured output format when resuming to access parsed
+    # outputs in the resumed stream.
+    text: MathResponse
+  )
+
+  resumed_stream.each do |event|
+    case event
+    when OpenAI::Streaming::ResponseTextDeltaEvent
+      print(event.delta)
+    when OpenAI::Streaming::ResponseTextDoneEvent
+      puts
+      puts("--- Parsed object from resumed stream ---")
+      pp(event.parsed)
+    when OpenAI::Models::Responses::ResponseCompletedEvent
+      puts("Response completed.")
+      break
+    end
+  end
 
-# 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}"
+  puts "\nFinal response parsed outputs:"
+  response = resumed_stream.get_final_response
+  response
+    .output
+    .flat_map { _1.content }
+    .each do |content|
+      pp(content.parsed)
+    end
 end

lib/openai/resources/responses.rb

@@ -170,20 +170,22 @@ def stream(params)
         end
         model, tool_models = get_structured_output_models(parsed)
 
+        unwrap = ->(raw) do
+          if raw[:type] == "response.completed" && raw[:response]
+            parse_structured_outputs!(raw[:response], model, tool_models)
+          end
+          raw
+        end
+
         if previous_response_id
-          retrieve_params = {}
-          retrieve_params[:include] = params[:include] if params[:include]
-          retrieve_params[:request_options] = params[:request_options] if params[:request_options]
+          retrieve_params = params.slice(:include, :request_options)
 
-          raw_stream = retrieve_streaming(previous_response_id, retrieve_params)
+          raw_stream = retrieve_streaming_internal(
+            previous_response_id,
+            params: retrieve_params,
+            unwrap: unwrap
+          )
         else
-          unwrap = ->(raw) do
-            if raw[:type] == "response.completed" && raw[:response]
-              parse_structured_outputs!(raw[:response], model, tool_models)
-            end
-            raw
-          end
-
           parsed[:stream] = true
 
           raw_stream = @client.request(
@@ -365,6 +367,21 @@ def retrieve_streaming(response_id, params = {})
         )
       end
 
+      private def retrieve_streaming_internal(response_id, params:, unwrap:)
+        parsed, options = OpenAI::Responses::ResponseRetrieveParams.dump_request(params)
+        parsed.store(:stream, true)
+        @client.request(
+          method: :get,
+          path: ["responses/%1$s", response_id],
+          query: parsed,
+          headers: {"accept" => "text/event-stream"},
+          stream: OpenAI::Internal::Stream,
+          model: OpenAI::Responses::ResponseStreamEvent,
+          options: options,
+          unwrap: unwrap
+        )
+      end
+
       # Deletes a model response with the given ID.
       #
       # @overload delete(response_id, request_options: {})

test/openai/resources/responses/streaming_test.rb

@@ -305,23 +305,23 @@ def test_resume_stream_with_response_id_and_starting_after
   end
 
   def test_resume_stream_with_structured_output
-    # Stub the GET request to retrieve the response (streaming).
-    stub_request(:get, "http://localhost/responses/msg_structured?stream=true")
+    stub_request(:get, "http://localhost/responses/msg_background_structured?stream=true")
       .to_return(
         status: 200,
         headers: {"Content-Type" => "text/event-stream"},
         body: resume_stream_structured_output_sse_response
       )
 
-    stream = @client.responses.stream(previous_response_id: "msg_structured", text: WeatherModel)
+    stream = @client.responses.stream(
+      previous_response_id: "msg_background_structured",
+      text: WeatherModel
+    )
     events = stream.to_a
 
     text_done = events.find { |e| e.type == :"response.output_text.done" }
-    assert_equal("{\"location\":\"San Francisco\",\"temperature\":72}", text_done.text)
-
-    # Verify the parsed content is available.
     assert_pattern do
       text_done => OpenAI::Streaming::ResponseTextDoneEvent[
+        text: "{\"location\":\"San Francisco\",\"temperature\":72}",
         parsed: WeatherModel[
           location: "San Francisco",
           temperature: 72
@@ -338,6 +338,28 @@ def test_resume_stream_with_structured_output
         }
       ]
     end
+
+    # Also verify the parsed field is available in the final response for resumed streams.
+    final_response = stream.get_final_response
+    assert_pattern do
+      final_response => OpenAI::Models::Responses::Response[
+        id: "msg_structured",
+        status: :completed,
+        output: [
+          {
+            content: [
+              {
+                text: "{\"location\":\"San Francisco\",\"temperature\":72}",
+                parsed: WeatherModel[
+                  location: "San Francisco",
+                  temperature: 72
+                ]
+              }
+            ]
+          }
+        ]
+      ]
+    end
   end
 
   def test_structured_output_parsed_in_final_response