Module: RubyLLM::Protocols::Mistral::Conversations::Streaming

Defined in:
lib/ruby_llm/protocols/mistral/conversations/streaming.rb

Overview

:nodoc:

Class Method Summary collapse

Class Method Details

.accumulate_conversation_entry(data) ⇒ Object 



42
43
44
45
46
47
 
# File 'lib/ruby_llm/protocols/mistral/conversations/streaming.rb', line 42

def accumulate_conversation_entry(data)
  type = data['type'].delete_suffix('.delta').delete_suffix('.started').delete_suffix('.done')
  entry = @conversation_output[data.fetch('output_index', 0)] ||= { 'type' => type, 'arguments' => +'' }
  entry.merge!(data.slice('id', 'model', 'name', 'tool_call_id', 'confirmation_status', 'function', 'info'))
  entry['arguments'] << data['arguments'].to_s
end

.build_chunk(data) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/ruby_llm/protocols/mistral/conversations/streaming.rb', line 20

def build_chunk(data)
  @conversation_output ||= {}
  @conversation_response ||= { 'object' => 'conversation.response' }
  case data['type']
  when 'conversation.response.done'
    @conversation_done = true
    @conversation_response['usage'] = data['usage']
    return final_conversation_chunk
  when 'conversation.response.error'
    raise Error, data['message'] || data.dig('error', 'message') || 'Mistral conversation failed'
  when 'message.output.delta'
    return conversation_text_chunk(data)
  when 'function.call.delta', 'tool.execution.started', 'tool.execution.delta', 'tool.execution.done'
    accumulate_conversation_entry(data)
  end
  Chunk.new(role: :assistant, content: nil)
end

.conversation_text_chunk(data) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/ruby_llm/protocols/mistral/conversations/streaming.rb', line 49

def conversation_text_chunk(data)
  entry = @conversation_output[data.fetch('output_index',
                                          0)] ||= { 'type' => 'message.output', 'content' => [] }
  entry.merge!(data.slice('id', 'model', 'role'))
  part = data['content'].is_a?(String) ? { 'type' => 'text', 'text' => data['content'] } : data['content']
  merge_conversation_part(entry['content'], data.fetch('content_index', 0), part)
  content = { text: +'', thinking: +'', attachments: [], citations: [] }
  parse_conversation_parts([part], content)
  Chunk.new(role: :assistant, content: content[:text], model: data['model'],
            thinking: Thinking.build(text: content[:thinking].empty? ? nil : content[:thinking]))
end

.final_conversation_chunk ⇒ Object 



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

def final_conversation_chunk
  message = parse_completion_body(streamed_conversation_response, raw: nil)
  Chunk.new(role: :assistant, content: nil, model: message.model, tokens: message.tokens,
            citations: message.citations, tool_calls: message.tool_calls,
            server_tool_calls: message.server_tool_calls, raw_content: message.raw_content,
            attachments: message.attachments, finish_reason: message.finish_reason)
end

.merge_conversation_part(parts, index, part) ⇒ Object 



61
62
63
64
65
66
67
68
69
70
 
# File 'lib/ruby_llm/protocols/mistral/conversations/streaming.rb', line 61

def merge_conversation_part(parts, index, part)
  existing = parts[index]
  if existing && part['type'] == 'text'
    existing['text'] << part['text'].to_s
  elsif existing && part['type'] == 'thinking'
    existing['thinking'].concat(Array(part['thinking']))
  else
    parts[index] = Support::Utils.deep_dup(part)
  end
end

.stream_response(payload, additional_headers = {}) ⇒ Object

Raises:



10
11
12
13
14
15
16
17
18
 
# File 'lib/ruby_llm/protocols/mistral/conversations/streaming.rb', line 10

def stream_response(payload, additional_headers = {})
  @conversation_output = {}
  @conversation_response = { 'object' => 'conversation.response' }
  @conversation_done = false
  response = stream_events(completion_url, payload, additional_headers) { |data| yield build_chunk(data) }
  raise Error.new('Mistral conversation stream ended before completion', response:) unless @conversation_done

  parse_completion_body(streamed_conversation_response, raw: response)
end

.streamed_conversation_response ⇒ Object 



38
39
40
 
# File 'lib/ruby_llm/protocols/mistral/conversations/streaming.rb', line 38

def streamed_conversation_response
  @conversation_response.merge('outputs' => @conversation_output.sort.map(&:last))
end