Class: Legion::Extensions::Llm::StreamAccumulator

Inherits:
Object
  • Object
show all
Includes:
Logging::Helper
Defined in:
lib/legion/extensions/llm/stream_accumulator.rb

Overview

Assembles Canonical streaming chunks into a complete Canonical::Response.

The provider's build_chunk yields Canonical::Chunk objects (text_delta / thinking_delta / tool_call_delta / usage). This accumulator owns:

- the stateful think-tag split (cross-chunk boundary buffering — the
only streaming-specific code; the segment logic is shared with
Responses::ThinkingExtractor, 10 U1),
- the untagged-preamble heuristic,
- tool-call fragment correlation: the provider's authoritative wire
INDEX first, recency only as the fallback for providers that emit
no index (the wire index-first/recency-fallback law).

Fragments are assembled into a complete JSON string before the ONE strict arguments parser runs (10 U2).

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(request_id: nil, conversation_id: nil, exchange_id: nil) ⇒ StreamAccumulator

Returns a new instance of StreamAccumulator.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/legion/extensions/llm/stream_accumulator.rb', line 26

def initialize(request_id: nil, conversation_id: nil, exchange_id: nil)
  @request_id = request_id
  @conversation_id = conversation_id
  @exchange_id = exchange_id
  @content = +''
  @thinking_text = +''
  @thinking_signature = nil
  @tool_calls = {}
  @stop_reason = nil
  @usage = {}
  @inside_think_tag = false
  @pending_think_tag = +''
  @active_think_close_tag = nil
  @untagged_preamble_pending = true
  @untagged_preamble_buffer = +''
  @latest_tool_call_id = nil
  @index_to_id = {}
end

Instance Attribute Details

#model_id ⇒ Object (readonly)

Returns the value of attribute model_id.



24
25
26
 
# File 'lib/legion/extensions/llm/stream_accumulator.rb', line 24

def model_id
  @model_id
end

#stop_reason ⇒ Object (readonly)

Returns the value of attribute stop_reason.



24
25
26
 
# File 'lib/legion/extensions/llm/stream_accumulator.rb', line 24

def stop_reason
  @stop_reason
end

Instance Method Details

#add(chunk) ⇒ Object

Consume one Canonical::Chunk; returns the Array of Canonical::Chunk objects to emit to the caller (empty when nothing is emitted).



47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/legion/extensions/llm/stream_accumulator.rb', line 47

def add(chunk)
  log.debug { chunk.inspect } if Legion::Extensions::Llm.config.log_stream_debug
  @model_id ||= chunk.metadata[:model] if chunk.metadata.is_a?(::Hash)
  @stop_reason = chunk.stop_reason if chunk.stop_reason
  track_usage(chunk.usage) if chunk.usage

  case chunk.type
  when :text_delta then add_text_delta(chunk)
  when :thinking_delta then add_thinking_delta(chunk)
  when :tool_call_delta then add_tool_call_delta(chunk)
  when :usage then [chunk]
  else []
  end
end

#flush_pending_chunk ⇒ Object

Flush any text still held by the untagged-preamble heuristic so short responses still stream at least one delta (the superset flush — 10 #19: folds the buffer into content AND emits the held text).



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/legion/extensions/llm/stream_accumulator.rb', line 65

def flush_pending_chunk
  return [] if @untagged_preamble_buffer.empty?

  content, thinking = Responses::ThinkingExtractor.extract_untagged_preamble(@untagged_preamble_buffer)
  emitted = []
  if thinking
    @content << content
    @thinking_text << thinking
    emitted << thinking_delta_for(thinking)
    emitted << text_delta_for(content) unless content.empty?
  else
    @content << @untagged_preamble_buffer
    emitted << text_delta_for(@untagged_preamble_buffer)
  end
  @untagged_preamble_buffer = +''
  @untagged_preamble_pending = false
  emitted
end

#to_response(model: nil) ⇒ Object

The accumulated Canonical::Response (05 O5). model falls back to the Selection-derived model when the provider wire reported none.



86
87
88
89
90
91
92
93
94
95
 
# File 'lib/legion/extensions/llm/stream_accumulator.rb', line 86

def to_response(model: nil)
  Canonical::Response.build(
    text: @content.empty? ? nil : @content,
    thinking: accumulated_thinking,
    tool_calls: accumulated_tool_calls,
    usage: accumulated_usage,
    stop_reason: @stop_reason,
    model: @model_id || model
  )
end