Module: RubyLLM::Protocols::Responses::Streaming

Defined in:
lib/ruby_llm/protocols/responses/streaming.rb

Overview

Streaming methods of the OpenAI Responses API. Events are semantic: each SSE data frame carries a type describing what changed.

Constant Summary collapse

ERROR_STATUSES = 
{
  'server_error' => 500,
  'rate_limit_exceeded' => 429,
  'insufficient_quota' => 429
}.freeze

Class Method Summary collapse

Class Method Details

.build_annotation_chunk(data) ⇒ Object 



53
54
55
56
57
58
59
60
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 53

def build_annotation_chunk(data)
  position = citation_content_position(data)
  offset = (@citation_content_lengths || {}).sum do |key, length|
    (key <=> position).negative? ? length : 0
  end

  chunk citations: offset_citations(parse_annotations([data['annotation']], nil), offset, nil)
end

.build_chunk(data) ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 22

def build_chunk(data)
  case data['type']
  when 'response.output_text.delta', 'response.refusal.delta'
    build_text_chunk(data)
  when 'response.reasoning_summary_text.delta'
    chunk thinking: Thinking.build(text: data['delta'])
  when 'response.reasoning_summary_part.added'
    build_reasoning_summary_part_chunk(data)
  when 'response.output_text.annotation.added'
    build_annotation_chunk(data)
  when 'response.output_item.added'
    build_item_added_chunk(data)
  when 'response.function_call_arguments.delta'
    chunk tool_calls: { data['output_index'] => ToolCall.new(id: nil, name: nil, arguments: data['delta']) }
  when 'response.output_item.done'
    build_item_done_chunk(data)
  when 'response.completed', 'response.incomplete'
    build_final_chunk(data)
  when 'response.failed'
    raise Error, data.dig('response', 'error', 'message')
  else
    chunk
  end
end

.build_final_chunk(data) ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 88

def build_final_chunk(data)
  response = data['response'] || {}
  output = response['output'] || []
  server_tool_calls = parse_server_tool_items(output)

  chunk model: response['model'],
        tool_calls: parse_tool_approvals(output, finish_reason: parse_finish_reason(response)),
        finish_reason: parse_finish_reason(response),
        citations: parse_citations(response, output, nil),
        server_tool_calls: server_tool_calls,
        raw_content: server_tool_calls.any? ? output : nil,
        **parse_usage(response['usage'] || {})
end

.build_item_added_chunk(data) ⇒ Object 



72
73
74
75
76
77
78
79
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 72

def build_item_added_chunk(data)
  item = data['item']
  return chunk unless item['type'] == 'function_call'

  chunk tool_calls: {
    data['output_index'] => ToolCall.new(id: item['call_id'], name: item['name'], arguments: +'')
  }
end

.build_item_done_chunk(data) ⇒ Object 



81
82
83
84
85
86
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 81

def build_item_done_chunk(data)
  item = data['item']
  return chunk unless item['type'] == 'reasoning' && item['encrypted_content']

  chunk thinking: Thinking.build(text: nil, signature: item['encrypted_content'])
end

.build_reasoning_summary_part_chunk(data) ⇒ Object 



66
67
68
69
70
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 66

def build_reasoning_summary_part_chunk(data)
  return chunk unless data['summary_index'].positive?

  chunk thinking: Thinking.build(text: "\n\n")
end

.build_text_chunk(data) ⇒ Object 



47
48
49
50
51
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 47

def build_text_chunk(data)
  @citation_content_lengths ||= Hash.new(0)
  @citation_content_lengths[citation_content_position(data)] += data['delta'].to_s.length
  chunk content: data['delta']
end

.chunk(content: nil, **attributes) ⇒ Object 



111
112
113
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 111

def chunk(content: nil, **attributes)
  Chunk.new(role: :assistant, content: content, **attributes)
end

.citation_content_position(data) ⇒ Object 



62
63
64
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 62

def citation_content_position(data)
  [data.fetch('output_index', 0), data.fetch('content_index', 0)]
end

.parse_streaming_error(data) ⇒ Object

Responses reports a stream error as a flat event carrying a code, where Chat Completions nests type and message under an error object.



104
105
106
107
108
109
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 104

def parse_streaming_error(data)
  event = JSON.parse(data)
  return super unless event.is_a?(Hash) && event['type'] == 'error'

  [ERROR_STATUSES.fetch(event['code'], 400), event['message']]
end

.stream_response ⇒ Object 



17
18
19
20
 
# File 'lib/ruby_llm/protocols/responses/streaming.rb', line 17

def stream_response(...)
  @citation_content_lengths = Hash.new(0)
  super
end