Class: OpenAI::Helpers::Agents::SessionStream

Inherits:
Object
  • Object
show all
Includes:
Enumerable, Enumerable[OpenAI::Models::Beta::agent_session_event]
Defined in:
lib/openai/helpers/agents/session_stream.rb,
sig/openai/helpers/agents.rbs

Overview

Stream one turn on an idle session. The caller must be its only input writer until iteration ends, because input submission does not return a turn ID. Use a block or ensure #close when abandoning external iteration. Closing the connection does not cancel the backend turn.

Instance Method Summary collapse

Constructor Details

#initialize(sessions:, session_id:, input:, tool_handlers: {}, idempotency_key: nil, request_options: {}) ⇒ SessionStream

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 SessionStream.

Raises:

  • (ArgumentError) 


16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/openai/helpers/agents/session_stream.rb', line 16

def initialize(sessions:, session_id:, input:, tool_handlers: {}, idempotency_key: nil, request_options: {})
  messages = input.is_a?(String) ? [{role: :user, content: [{type: :input_text, text: input}]}] : input.to_a
  raise ArgumentError, "input must not be empty" if input == "" || messages.empty?

  @sessions = sessions
  @session_id = session_id
  @handlers = tool_handlers.to_h.dup
  @options = request_options.to_h.dup
  headers = {"OpenAI-Beta" => "agents=v1"}.merge(@options[:extra_headers].to_h)
  options_key = @options.delete(:idempotency_key)
  input_key = idempotency_key || options_key || SecureRandom.uuid
  headers.delete_if do |key, value|
    match = key.to_s.casecmp?("idempotency-key")
    input_key = value if match
    match
  end

  @options[:extra_headers] = headers
  @recent_events = {}
  @handled_calls = {}
  @closed = false
  session = @sessions.retrieve(@session_id, request_options: @options)
  unless session.status == :idle
    raise(
      ArgumentError,
      "sessions.stream requires an idle session; use sessions.events.stream_streaming to follow an active session"
    )
  end

  @raw_stream = @sessions.events.stream_streaming(@session_id, request_options: @options)
  begin
    @sessions.events.create(
      @session_id,
      events: [{type: :"agent.session.input.message", input: messages}],
      idempotency_key: input_key,
      request_options: @options
    )
    @iterator = iterator
    submitted = true
  ensure
    close unless submitted
  end
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Close the event connection without cancelling the backend turn.



89
90
91
92
93
94
 
# File 'lib/openai/helpers/agents/session_stream.rb', line 89

def close
  return if @closed

  @closed = true
  @raw_stream&.close
end

#each {|event| ... } ⇒ Enumerator, self

Yields the original typed events. Registered handlers run sequentially after their function-call event is yielded; unknown tools remain caller-managed.

Yield Parameters:

Returns:

  • (Enumerator, self) 


64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/openai/helpers/agents/session_stream.rb', line 64

def each
  return enum_for(:each) unless block_given?
  return self if @closed
  begin
    loop do
      break if @closed
      yield @iterator.next
    end

  ensure
    close
  end

  self
end

#until_done ⇒ self

Consume remaining events and registered tool calls.

Returns:

  • (self) 


82
83
84
85
 
# File 'lib/openai/helpers/agents/session_stream.rb', line 82

def until_done
  each { |_event| nil }
  self
end