Class: LLM::Stream

Inherits:
Object
  • Object
show all
Defined in:
lib/llm/stream.rb,
lib/llm/stream/io.rb,
lib/llm/stream/queue.rb,
lib/llm/stream/disabled.rb

Overview

Note:

The on_* callbacks run inline with the streaming parser. They therefore block streaming progress and should generally return as quickly as possible.

The LLM::Stream class provides the callback interface for streamed model output in llm.rb.

A stream object can be an instance of LLM::Stream or a subclass that overrides the callbacks it needs. For basic streaming, llm.rb also accepts any object that implements #<<. #queue provides a small helper for collecting asynchronous tool work started from a callback.

The most common callback is #on_content, which also maps to #<<. Providers may also call #on_reasoning_content and #on_tool_call when that data is available. Runtime features such as context compaction may also emit lifecycle callbacks like #on_transform or #on_compaction.

Examples:

Subclass with callbacks

class MyStream < LLM::Stream
  def on_content(content)
    print content
  end

  def on_reasoning_content(content)
    warn content
  end
end

llm = LLM.deepseek(key: ENV["KEY"])
agent = LLM::Agent.new(llm, stream: MyStream.new)
agent.talk "Explain Ruby fibers."

See Also:

Direct Known Subclasses

Repl::Stream, Disabled, IO

Defined Under Namespace

Classes: Disabled, IO, Queue

Public callbacks collapse

Finders collapse

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.try(obj, extra: {}) ⇒ LLM::Stream

This method will try to convert its argument into an instance of LLM::Stream or a subclass of it.

Acceptable inputs include: LLM::Stream objects, IO objects who implement #<<, true, false, and nil. Anything else raises a TypeError.

Parameters:

Returns:

Raises:

  • (TypeError) 


57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/llm/stream.rb', line 57

def self.try(obj, extra: {})
  if LLM::Stream === obj
    obj.tap { _1.extra.merge!(extra) }
  elsif obj.respond_to?(:<<)
    LLM::Stream::IO.new(obj).tap { _1.extra.merge!(extra) }
  elsif obj == true
    LLM::Stream.new.tap { _1.extra.merge!(extra) }
  elsif obj.nil? || obj == false
    LLM::Stream::Disabled.new.tap { _1.extra.merge!(extra) }
  else
    raise TypeError, "invalid stream object"
  end
end

Instance Method Details

#__find__(name) ⇒ LLM::Function?

Resolves a streamed tool call against the current request tools first, then falls back to the global function registry.

Parameters:

  • name (String)

Returns:



193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/llm/stream.rb', line 193

def __find__(name)
  tools = extra[:tools] || ctx&.params&.dig(:tools) || []
  tool = tools.find do
    candidate = _1.respond_to?(:function) ? _1.function.name : _1.name
    candidate.to_s == name.to_s
  end
  if tool
    tool.respond_to?(:function) ? tool.function : tool
  else
    LLM::Function.find_by_name(name)
  end
end

#extra ⇒ Hash

Returns extra context associated with the current streamed request.

Returns:

  • (Hash) 


74
75
76
 
# File 'lib/llm/stream.rb', line 74

def extra
  @extra ||= LLM::Object.from({})
end

#ctx ⇒ LLM::Context?

Returns the current context, if one was attached to the stream.

Returns:



81
82
83
 
# File 'lib/llm/stream.rb', line 81

def ctx
  extra[:ctx]
end

#queue ⇒ LLM::Stream::Queue

Returns a lazily-initialized queue for tool results or spawned work.

Returns:



88
89
90
 
# File 'lib/llm/stream.rb', line 88

def queue
  @queue ||= Queue.new(self)
end

#wait ⇒ Array<LLM::Function::Return>

Waits for queued tool work to finish and returns function results. Any passed arguments are ignored because queued work is waited according to the actual task types already present in the queue.

Returns:



97
98
99
 
# File 'lib/llm/stream.rb', line 97

def wait(*)
  queue.wait
end

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


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

def enabled?
  true
end

#on_content(content) ⇒ nil Also known as: <<

Called when visible assistant output is streamed.

Parameters:

  • content (String) —

    A chunk of assistant-visible text.

Returns:

  • (nil) 


114
115
116
 
# File 'lib/llm/stream.rb', line 114

def on_content(content)
  nil
end

#on_reasoning_content(content) ⇒ nil

Called when reasoning output is streamed separately from visible content.

Parameters:

  • content (String) —

    A chunk of reasoning text.

Returns:

  • (nil) 


124
125
126
 
# File 'lib/llm/stream.rb', line 124

def on_reasoning_content(content)
  nil
end

#on_tool_call(tool) ⇒ nil

Called when a streamed tool call has been fully constructed. A stream implementation may start tool execution here by pushing @queue << tool.task(:thread) onto #queue.

Parameters:

Returns:

  • (nil) 


135
136
137
 
# File 'lib/llm/stream.rb', line 135

def on_tool_call(tool)
  nil
end

#on_tool_return(tool, result) ⇒ nil

Called when queued streamed tool work returns. This callback runs when #wait resolves work queued from #on_tool_call.

Parameters:

Returns:

  • (nil) 


148
149
150
 
# File 'lib/llm/stream.rb', line 148

def on_tool_return(tool, result)
  nil
end

#on_transform(transformer) ⇒ nil

Called before a context transformer rewrites a prompt.

Parameters:

Returns:

  • (nil) 


156
157
158
 
# File 'lib/llm/stream.rb', line 156

def on_transform(transformer)
  nil
end

#on_transform_finish(transformer) ⇒ nil

Called after a context transformer finishes rewriting a prompt.

Parameters:

Returns:

  • (nil) 


164
165
166
 
# File 'lib/llm/stream.rb', line 164

def on_transform_finish(transformer)
  nil
end

#on_compaction(compactor) ⇒ nil

Called before a context compaction starts.

Parameters:

Returns:

  • (nil) 


172
173
174
 
# File 'lib/llm/stream.rb', line 172

def on_compaction(compactor)
  nil
end

#on_compaction_finish(compactor) ⇒ nil

Called after a context compaction finishes.

Parameters:

Returns:

  • (nil) 


180
181
182
 
# File 'lib/llm/stream.rb', line 180

def on_compaction_finish(compactor)
  nil
end