Module: RubyLLM::Protocols::XAI::StreamingTranscription

Included in:
RubyLLM::Providers::XAI::ChatCompletions, RubyLLM::Providers::XAI::Responses
Defined in:
lib/ruby_llm/protocols/xai/streaming_transcription.rb

Overview

:nodoc: all

Constant Summary collapse

SAMPLE_RATES = 
[8000, 16_000, 22_050, 24_000, 44_100, 48_000].freeze
ENCODINGS = 
{ [1, 16] => 'pcm', [6, 8] => 'alaw', [7, 8] => 'mulaw' }.freeze

Instance Method Summary collapse

Instance Method Details

#build_streamed_transcription(segments, completed, model:, language:) ⇒ Object 



109
110
111
112
113
114
115
116
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 109

def build_streamed_transcription(segments, completed, model:, language:)
  RubyLLM::Transcription.new(
    text: segments.map { |segment| segment.fetch('text') }.join(' '), model:,
    language: language || segments.filter_map { |segment| segment['language'] }.last,
    duration: completed.filter_map { |event| event['duration'] }.max,
    segments:, words: segments.flat_map { |segment| segment['words'] || [] }
  )
end

#duplicate_transcription_segment?(previous, segment) ⇒ Boolean

Returns:

  • (Boolean) 


103
104
105
106
107
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 103

def duplicate_transcription_segment?(previous, segment)
  same_text = previous.values_at('text', 'channel', 'words') == segment.values_at('text', 'channel', 'words')
  same_time = previous.values_at('start', 'end') == segment.values_at('start', 'end')
  same_text && (segment['start'].nil? || same_time)
end

#parse_transcription_segment(event) ⇒ Object 



97
98
99
100
101
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 97

def parse_transcription_segment(event)
  { 'text' => event['text'], 'start' => event['start'],
    'end' => event['start'] && (event['start'] + event['duration'].to_f),
    'channel' => event['channel_index'], 'words' => event['words'], 'language' => event['language'] }.compact
end

#process_transcription_event(event, segments, completed, ready) ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 40

def process_transcription_event(event, segments, completed, ready, &)
  case event['type']
  when 'transcript.created'
    ready << true
  when 'transcript.partial'
    process_transcription_segment(event, segments, &)
  when 'transcript.done'
    process_transcription_segment(event, segments, &)
    completed << event unless completed.any? { |item| item['channel_index'] == event['channel_index'] }
  when 'error'
    raise Error, event['message'] || 'xAI transcription failed'
  end
end

#process_transcription_segment(event, segments) {|TranscriptionChunk.new(type: TranscriptionChunk::SEGMENT, delta:, segment:, raw: event)| ... } ⇒ Object

Yields:



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 80

def process_transcription_segment(event, segments)
  text = event['text'].to_s
  return if text.empty?

  unless event['is_final'] || event['type'] == 'transcript.done'
    yield TranscriptionChunk.new(type: TranscriptionChunk::PARTIAL, text:, raw: event)
    return
  end

  segment = parse_transcription_segment(event)
  return if segments.any? { |previous| duplicate_transcription_segment?(previous, segment) }

  delta = segments.empty? ? text : " #{text}"
  segments << segment
  yield TranscriptionChunk.new(type: TranscriptionChunk::SEGMENT, delta:, segment:, raw: event)
end

#receive_streamed_transcription(socket, audio, segments, completed, &block) ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 28

def receive_streamed_transcription(socket, audio, segments, completed, &block)
  ready = Queue.new
  write = lambda do |connection|
    ready.pop
    send_transcription_audio(connection, audio)
  end
  socket.each_message(write:) do |message|
    process_transcription_event(JSON.parse(message), segments, completed, ready, &block)
    socket.close if completed.size == audio.channels
  end
end

#send_transcription_audio(socket, audio) ⇒ Object 



69
70
71
72
73
74
75
76
77
78
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 69

def send_transcription_audio(socket, audio)
  offset = 0
  bytes = audio.sample_rate * audio.channels * audio.bits_per_sample / 8 / 10
  while offset < audio.data.bytesize
    socket.send_binary(audio.data.byteslice(offset, bytes))
    offset += bytes
    sleep 0.1
  end
  socket.send_text(JSON.generate(type: 'audio.done'))
end

#stream_transcription(payload, model:, &block) ⇒ Object 



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 10

def stream_transcription(payload, model:, &block)
  audio = RubyLLM::Transcription::WavAudio.new(payload.fetch(:file).io.read)
  url = streaming_transcription_url(payload, audio:)
  segments = []
  completed = []
  @usage_tracker.start
  Transport::WebsocketConnection.open(url, headers: @provider.headers, config: @config) do |socket|
    receive_streamed_transcription(socket, audio, segments, completed, &block)
  end
  raise Error, 'xAI transcription ended before its final transcript' unless completed.size == audio.channels

  result = build_streamed_transcription(segments, completed, model:, language: payload[:language])
  block.call(TranscriptionChunk.new(type: TranscriptionChunk::DONE, text: result.text, raw: completed.last))
  result
ensure
  payload[:file]&.io&.close
end

#streaming_transcription_url(payload, audio:) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/ruby_llm/protocols/xai/streaming_transcription.rb', line 54

def streaming_transcription_url(payload, audio:)
  encoding = ENCODINGS[[audio.encoding, audio.bits_per_sample]]
  unless encoding && SAMPLE_RATES.include?(audio.sample_rate) && (1..8).cover?(audio.channels)
    raise ArgumentError, 'xAI streaming requires 16-bit PCM or 8-bit G.711 WAV audio at a supported sample rate'
  end

  params = payload.except(:file).merge(encoding:, sample_rate: audio.sample_rate, interim_results: true)
  params[:channels] = audio.channels
  params[:multichannel] = audio.channels > 1
  pairs = params.flat_map { |key, value| Array(value).map { |item| [key, item] } }
  uri = URI.join("#{@provider.api_base.sub(%r{/+\z}, '')}/", "stt?#{URI.encode_www_form(pairs)}")
  uri.scheme = uri.scheme == 'https' ? 'wss' : 'ws'
  uri.to_s
end