Class: Ask::Agent::StreamTransforms::Pipeline

Inherits:
Object
  • Object
show all
Defined in:
lib/ask/agent/stream_transforms/pipeline.rb

Overview

A composable chain of Base transforms applied to LLM stream chunks.

Transforms are applied in order — the first registered transform sees each chunk first. Each transform may yield zero, one, or many chunks downstream.

Examples:

pipeline = Pipeline.new
pipeline.use :thinking_separator
pipeline.use :text_buffer, min_size: 100

wrapped_block = pipeline.wrap { |chunk| handle_chunk(chunk) }
stream.each { |chunk| wrapped_block.call(chunk) }
pipeline.flush { |chunk| handle_chunk(chunk) }

Constant Summary collapse

KNOWN_TRANSFORMS = 
{
  thinking_separator: "Ask::Agent::StreamTransforms::ThinkingSeparator",
  text_buffer: "Ask::Agent::StreamTransforms::TextBuffer",
  extract_json: "Ask::Agent::StreamTransforms::ExtractJson"
}.freeze

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Pipeline

Returns a new instance of Pipeline.



27
28
29
 
# File 'lib/ask/agent/stream_transforms/pipeline.rb', line 27

def initialize
  @transforms = []
end

Instance Method Details

#configured? ⇒ Boolean

Returns whether any transform has been registered.

Returns:

  • (Boolean) —

    whether any transform has been registered



42
43
44
 
# File 'lib/ask/agent/stream_transforms/pipeline.rb', line 42

def configured?
  @transforms.any?
end

#each(&block) ⇒ Object

Iterate over the configured transform instances.



47
48
49
 
# File 'lib/ask/agent/stream_transforms/pipeline.rb', line 47

def each(&block)
  @transforms.each(&block)
end

#flush {|Ask::Chunk| ... } ⇒ Object

Flush any remaining buffered state from all transforms.

Call this once when the stream finishes to ensure buffering transforms (e.g. TextBuffer) emit their final content.

Yields:

  • (Ask::Chunk) —

    final chunks



76
77
78
79
80
 
# File 'lib/ask/agent/stream_transforms/pipeline.rb', line 76

def flush(&block)
  @transforms.each do |transform|
    transform.finish { |c| block.call(c) if block }
  end
end

#use(transform, **options) ⇒ Object

Register a transform in the chain.

Parameters:

  • transform (Symbol, Class) —

    a symbol from KNOWN_TRANSFORMS or a StreamTransforms::Base subclass

  • options (Hash) —

    keyword arguments passed to the transform's constructor



35
36
37
38
39
 
# File 'lib/ask/agent/stream_transforms/pipeline.rb', line 35

def use(transform, **options)
  klass = resolve(transform)
  @transforms << klass.new(**options)
  self
end

#wrap {|Ask::Chunk| ... } ⇒ Proc

Wrap a chunk-processing block with the transform chain.

The returned block accepts raw Chunks, runs them through each transform in sequence, and yields only to the original block for chunks that survive the chain.

Yields:

  • (Ask::Chunk) —

    transformed chunks

Returns:

  • (Proc) —

    a wrapped block that accepts raw chunks



59
60
61
62
63
64
65
66
67
68
 
# File 'lib/ask/agent/stream_transforms/pipeline.rb', line 59

def wrap(&block)
  return block unless configured?

  chain = block
  @transforms.reverse_each do |transform|
    current = chain
    chain = ->(chunk) { transform.call(chunk) { |c| current.call(c) } }
  end
  chain
end