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:



196
197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/llm/stream.rb', line 196

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

#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

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


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

def enabled?
  true
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

#on_compaction(compactor) ⇒ nil

Called before a context compaction starts.

Parameters:

Returns:

  • (nil) 


175
176
177
 
# File 'lib/llm/stream.rb', line 175

def on_compaction(compactor)
  nil
end

#on_compaction_finish(compactor) ⇒ nil

Called after a context compaction finishes.

Parameters:

Returns:

  • (nil) 


183
184
185
 
# File 'lib/llm/stream.rb', line 183

def on_compaction_finish(compactor)
  nil
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).tap(&:spawn) onto #queue — spawn returns nil, so tap keeps the task while starting it. The task carries the context's guard, so it is checked before the tool runs — a blocked call yields its guard return without executing.

Parameters:

Returns:

  • (nil) 


138
139
140
 
# File 'lib/llm/stream.rb', line 138

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) 


151
152
153
 
# File 'lib/llm/stream.rb', line 151

def on_tool_return(tool, result)
  nil
end

#on_transform(transformer) ⇒ nil

Called before a context transformer rewrites a prompt.

Parameters:

Returns:

  • (nil) 


159
160
161
 
# File 'lib/llm/stream.rb', line 159

def on_transform(transformer)
  nil
end

#on_transform_finish(transformer) ⇒ nil

Called after a context transformer finishes rewriting a prompt.

Parameters:

Returns:

  • (nil) 


167
168
169
 
# File 'lib/llm/stream.rb', line 167

def on_transform_finish(transformer)
  nil
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