Class: Ask::Stream

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/ask/stream.rb

Overview

Streaming response from an LLM provider. Wraps an enumerable of Chunks and provides accumulation into a single string or message.

stream = Ask::Stream.new { |chunk| puts chunk.content }
stream.each { |chunk| ... }
stream.accumulated_text  # => "full response text"

Instance Method Summary collapse

Constructor Details

#initialize(&chunk_handler) ⇒ Stream

Returns a new instance of Stream.

Parameters:

  • chunk_handler (Proc, nil) —

    optional callback invoked for each chunk as it arrives



65
66
67
68
69
70
 
# File 'lib/ask/stream.rb', line 65

def initialize(&chunk_handler)
  @chunks = []
  @chunk_handler = chunk_handler
  @accumulated = +""
  @finished = false
end

Instance Method Details

#accumulated_text ⇒ String Also known as: to_s

Returns full accumulated text from all chunks.

Returns:

  • (String) —

    full accumulated text from all chunks



103
104
105
 
# File 'lib/ask/stream.rb', line 103

def accumulated_text
  @accumulated.dup
end

#accumulated_usage ⇒ Hash

Returns accumulated usage across all chunks, merged by key.

Returns:

  • (Hash) —

    accumulated usage across all chunks, merged by key



109
110
111
112
113
114
115
116
 
# File 'lib/ask/stream.rb', line 109

def accumulated_usage
  @chunks
    .map(&:usage)
    .compact
    .reduce({}) do |acc, u|
      acc.merge(u) { |_key, old, new| old + new }
    end
end

#add(chunk) ⇒ Chunk

Add a chunk to the stream.

Parameters:

  • chunk (Chunk, String) —

    chunk to add (strings are wrapped in Chunk)

Returns:

  • (Chunk) —

    the added chunk



85
86
87
88
89
90
91
92
 
# File 'lib/ask/stream.rb', line 85

def add(chunk)
  chunk = Chunk.new(content: chunk) if chunk.is_a?(String)

  @chunks << chunk
  @accumulated << chunk.content.to_s if chunk.content
  @chunk_handler&.call(chunk)
  chunk
end

#chunks ⇒ Array<Chunk>

Returns a copy of all accumulated chunks.

Returns:

  • (Array<Chunk>) —

    a copy of all accumulated chunks



119
 
# File 'lib/ask/stream.rb', line 119

def chunks = @chunks.dup

#each {|Chunk| ... } ⇒ Enumerator

Iterate over all accumulated chunks.

Yields:

Returns:

  • (Enumerator) —

    if no block given



75
76
77
78
79
80
 
# File 'lib/ask/stream.rb', line 75

def each(&block)
  return enum_for(:each) unless block

  @chunks.each(&block)
  self
end

#finish! ⇒ Object

Mark the stream as finished.



95
96
97
 
# File 'lib/ask/stream.rb', line 95

def finish!
  @finished = true
end

#finished? ⇒ Boolean

Returns true if the stream has been finished.

Returns:

  • (Boolean) —

    true if the stream has been finished



100
 
# File 'lib/ask/stream.rb', line 100

def finished? = @finished

#inspect ⇒ String

Returns human-readable representation.

Returns:

  • (String) —

    human-readable representation



125
126
127
128
 
# File 'lib/ask/stream.rb', line 125

def inspect
  finished = @finished ? "finished" : "streaming"
  "#<Ask::Stream #{finished} chunks=#{@chunks.length}>"
end

#length ⇒ Integer

Returns number of chunks accumulated.

Returns:

  • (Integer) —

    number of chunks accumulated



122
 
# File 'lib/ask/stream.rb', line 122

def length = @chunks.length