Class: LLM::Stream

Inherits:
Object
  • Object
show all
Defined in:
lib/llm/stream.rb,
lib/llm/stream/queue.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, and #tool_not_found returns an in-band tool error when a streamed tool cannot be resolved.

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_compaction.

Defined Under Namespace

Classes: Queue

Public callbacks collapse

Error handlers collapse

Instance Method Summary collapse

Instance Method Details

#ctx ⇒ LLM::Context?

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

Returns:



36
37
38
 
# File 'lib/llm/stream.rb', line 36

def ctx
  extra[:ctx]
end

#extra ⇒ Hash

Returns extra context associated with the current streamed request.

Returns:

  • (Hash) 


29
30
31
 
# File 'lib/llm/stream.rb', line 29

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

#on_compaction(ctx, compactor) ⇒ nil

Called before a context compaction starts.

Parameters:

Returns:

  • (nil) 


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

def on_compaction(ctx, compactor)
  nil
end

#on_compaction_finish(ctx, compactor) ⇒ nil

Called after a context compaction finishes.

Parameters:

Returns:

  • (nil) 


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

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) 


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

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) 


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

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. Tool resolution depends on Function.registry, which includes LLM::Tool subclasses, including MCP tools, but not functions defined with LLM.function. 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) 


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

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) 


111
112
113
 
# File 'lib/llm/stream.rb', line 111

def on_tool_return(tool, result)
  nil
end

#queue ⇒ LLM::Stream::Queue

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

Returns:



43
44
45
 
# File 'lib/llm/stream.rb', line 43

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

#tool_not_found(tool) ⇒ LLM::Function::Return

Note:

This is mainly useful as a fallback from #on_tool_call. It should be uncommon in normal use, since streamed tool callbacks only run for tools already defined in the context.

Returns a function return describing a streamed tool that could not be resolved.

Parameters:

Returns:



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

def tool_not_found(tool)
  LLM::Function::Return.new(tool.id, tool.name, {
    error: true, type: LLM::NoSuchToolError.name, message: "tool not found"
  })
end

#wait(strategy) ⇒ Array<LLM::Function::Return>

Waits for queued tool work to finish and returns function results.

Parameters:

  • strategy (Symbol) —

    The concurrency strategy to use

Returns:



52
53
54
 
# File 'lib/llm/stream.rb', line 52

def wait(strategy)
  queue.wait(strategy)
end