Class: OpenAI::Responses::SessionAccumulator Private

Inherits:
Object
  • Object
show all
Defined in:
lib/openai/helpers/responses_websocket/session.rb

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(limit) ⇒ SessionAccumulator

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of SessionAccumulator.



60
61
62
63
64
65
66
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 60

def initialize(limit)
  @limit = limit
  @bytes = 0
  @items = {}
  @state = OpenAI::Helpers::Streaming::ResponseStreamState.new(text_format: nil, starting_after: 0)
  @decoder = OpenAI::Helpers::Streaming::ResponseEventDecoder.new(model: OpenAI::Responses::ResponseStreamEvent)
end

Instance Attribute Details

#error ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



58
59
60
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 58

def error
  @error
end

#final_response ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



58
59
60
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 58

def final_response
  @final_response
end

Instance Method Details

#add(event, bytes) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/openai/helpers/responses_websocket/session.rb', line 68

def add(event, bytes)
  type = event.type.to_s
  if type == "response.created"
    @bytes = 0
    @items.clear
    @final_response = nil
    @error = nil
    @state = OpenAI::Helpers::Streaming::ResponseStreamState.new(text_format: nil, starting_after: 0)
  end

  if type == "error"
    @error = RequestError.new(event)
    clear
    return
  end

  return if @error
  unless %w[response.created response.output_item.done response.completed response.failed response.incomplete]
      .include?(type)
    return
  end

  @bytes += bytes
  raise BufferError, "Response collection exceeded max_response_bytes." if @bytes > @limit

  if type == "response.output_item.done"
    # Retain wire data independently of the mutable event returned by receive.
    completed_item = OpenAI::Internal::Type::BaseModel.recursively_to_h(event.item, convert: false)
    @items[event.output_index] = JSON.parse(JSON.generate(completed_item), symbolize_names: true)
  elsif %w[response.completed response.failed response.incomplete].include?(type)
    data = OpenAI::Internal::Type::BaseModel.recursively_to_h(event, convert: false)
    data = JSON.parse(JSON.generate(data), symbolize_names: true)
    response = data.fetch(:response)
    if response[:output].nil?
      response = response.merge(output: @items.sort.map { |_index, item| item.to_h })
      data = data.merge(response: response)
    end

    decoded = @decoder.decode(data)
    if type == "response.completed"
      @state.handle_event(decoded)
      @final_response = @state.completed_response
    else
      @final_response = decoded.response
    end

    clear
  else
    data = OpenAI::Internal::Type::BaseModel.recursively_to_h(event, convert: false)
    @state.handle_event(@decoder.decode(data))
  end

rescue StandardError => error
  @error = error.is_a?(SessionError) ? error : SessionError.new("Could not collect the Responses final response.")
  clear
end