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_filter 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

Console::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.

The stream it answers with is ready to be told about a step: LLM::Step is prepended onto it here, once, rather than by every caller that means to save a conversation.

Parameters:

Returns:

Raises:

  • (TypeError) 


63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/llm/stream.rb', line 63

def self.try(obj, extra: {})
  if LLM::Stream === obj
    obj.tap { _1.extra.merge!(extra) }.tap { _1.singleton_class.prepend(LLM::Step) }
  elsif obj.respond_to?(:<<)
    LLM::Stream::IO.new(obj).tap { _1.extra.merge!(extra) }.tap { _1.singleton_class.prepend(LLM::Step) }
  elsif obj == true
    LLM::Stream.new.tap { _1.extra.merge!(extra) }.tap { _1.singleton_class.prepend(LLM::Step) }
  elsif obj.nil? || obj == false
    LLM::Stream::Disabled.new.tap { _1.extra.merge!(extra) }.tap { _1.singleton_class.prepend(LLM::Step) }
  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:



242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/llm/stream.rb', line 242

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) 


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

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

#ctx ⇒ LLM::Context?

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

Returns:



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

def ctx
  extra[:ctx]
end

#queue ⇒ LLM::Stream::Queue

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

Returns:



94
95
96
 
# File 'lib/llm/stream.rb', line 94

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:



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

def wait(*)
  queue.wait
end

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


109
110
111
 
# File 'lib/llm/stream.rb', line 109

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) 


120
121
122
 
# File 'lib/llm/stream.rb', line 120

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) 


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

def on_reasoning_content(content)
  nil
end

#on_tool_call(tool) ⇒ nil

This method is called when the model requests a tool call. A little known feature: it is possible to spawn a tool call during a stream through:

@queue << tool.task(:thread).tap(&:spawn)

Parameters:

Returns:

  • (nil) 


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

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) 


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

def on_tool_return(tool, result)
  nil
end

#on_step(ctx, res) ⇒ nil

Called when a request in a turn has completed.

This is the boundary between one request and the next: the response is in the context's message history, and a turn that asked for tools runs them after this while a turn that is finished ends here.

It is emitted once per successful request, so a request that is retried after a rate limit or a timeout calls this when it lands and not once per attempt. It is also emitted when the stream is disabled: a disabled stream has no output to report, but it is still told that a step ended.

The context is passed with the response so that a stream does not have to reach for it: LLM::Step, which saves the conversation the context holds, is written against this callback alone.

Parameters:

  • ctx (LLM::Context) —

    The context the request belongs to

  • res (LLM::Response) —

    The response for the request that completed

Returns:

  • (nil) 


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

def on_step(ctx, res)
  nil
end

#on_filter(filter) ⇒ nil

Called before a message filter rewrites a prompt.

Parameters:

Returns:

  • (nil) 


192
193
194
 
# File 'lib/llm/stream.rb', line 192

def on_filter(filter)
  nil
end

#on_filter_finish(filter) ⇒ nil

Called after a message filter finishes rewriting a prompt.

Parameters:

Returns:

  • (nil) 


200
201
202
 
# File 'lib/llm/stream.rb', line 200

def on_filter_finish(filter)
  nil
end

#on_compaction(compactor) ⇒ nil

Called before a context compaction starts.

Parameters:

Returns:

  • (nil) 


208
209
210
 
# File 'lib/llm/stream.rb', line 208

def on_compaction(compactor)
  nil
end

#on_compaction_finish(compactor) ⇒ nil

Called after a context compaction finishes.

Parameters:

Returns:

  • (nil) 


216
217
218
 
# File 'lib/llm/stream.rb', line 216

def on_compaction_finish(compactor)
  nil
end

#on_retry(error, attempt) ⇒ nil Also known as: on_rate_limit

Called when a request is retried because it was rate limited or timed out.

Parameters:

  • error (Exception) —

    The error that caused the retry

  • attempt (Integer) —

    The one-based retry attempt number

Returns:

  • (nil) 


228
229
230
 
# File 'lib/llm/stream.rb', line 228

def on_retry(error, attempt)
  nil
end