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.

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) 


39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/llm/stream.rb', line 39

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

#ctx ⇒ LLM::Context?

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

Returns:



63
64
65
 
# File 'lib/llm/stream.rb', line 63

def ctx
  extra[:ctx]
end

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


85
86
87
 
# File 'lib/llm/stream.rb', line 85

def enabled?
  true
end

#extra ⇒ Hash

Returns extra context associated with the current streamed request.

Returns:

  • (Hash) 


56
57
58
 
# File 'lib/llm/stream.rb', line 56

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

#on_compaction(ctx, compactor) ⇒ nil

Called before a context compaction starts.

Parameters:

Returns:

  • (nil) 


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

def on_compaction(ctx, compactor)
  nil
end

#on_compaction_finish(ctx, 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(ctx, 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) 


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

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) 


106
107
108
 
# File 'lib/llm/stream.rb', line 106

def on_reasoning_content(content)
  nil
end

#on_tool_call(tool, error) ⇒ nil

Note:

A stream implementation may start tool execution here, for example by pushing ctx.spawn(tool, :thread), ctx.spawn(tool, :fiber), or ctx.spawn(tool, :task) onto #queue. Mixed strategies can also be selected per tool, such as tool.mcp? ? ctx.spawn(tool, :task) : ctx.spawn(tool, :ractor). When a streamed tool cannot be resolved, error is passed as an Function::Return. It can be sent back to the model, allowing the tool-call path to recover and the session to continue. Streamed tool resolution now prefers the current request tools, so LLM.function, MCP tools, bound tool instances, and normal LLM::Tool classes can all resolve through the same request-local path. The current :ractor mode is for class-based tools and does not support MCP tools.

Called when a streamed tool call has been fully constructed.

Parameters:

Returns:

  • (nil) 


130
131
132
 
# File 'lib/llm/stream.rb', line 130

def on_tool_call(tool, error)
  nil
end

#on_tool_return(tool, result) ⇒ nil

Note:

This callback runs when #wait resolves work that was queued from #on_tool_call, such as values returned by ctx.spawn(tool, :thread), ctx.spawn(tool, :fiber), or ctx.spawn(tool, :task).

Called when queued streamed tool work returns.

Parameters:

Returns:

  • (nil) 


144
145
146
 
# File 'lib/llm/stream.rb', line 144

def on_tool_return(tool, result)
  nil
end

#on_transform(ctx, transformer) ⇒ nil

Called before a context transformer rewrites a prompt.

Parameters:

Returns:

  • (nil) 


153
154
155
 
# File 'lib/llm/stream.rb', line 153

def on_transform(ctx, transformer)
  nil
end

#on_transform_finish(ctx, transformer) ⇒ nil

Called after a context transformer finishes rewriting a prompt.

Parameters:

Returns:

  • (nil) 


162
163
164
 
# File 'lib/llm/stream.rb', line 162

def on_transform_finish(ctx, transformer)
  nil
end

#queue ⇒ LLM::Stream::Queue

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

Returns:



70
71
72
 
# File 'lib/llm/stream.rb', line 70

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:



79
80
81
 
# File 'lib/llm/stream.rb', line 79

def wait(*)
  queue.wait
end