Class: Vangrail::StreamGuard

Inherits:
Object
  • Object
show all
Defined in:
lib/vangrail/stream_guard.rb

Overview

Runs output rails while the answer is still arriving.

An output rail that only runs on the finished text is a rail that runs after the reader has read it. Streaming makes that worse, not better: the tokens are on screen as they arrive, so the best a caller can do at the end is withdraw text somebody has already seen, and a credential that appeared for four seconds has appeared.

So the deterministic rails run as the buffer grows, and they run often, because they cost microseconds and cannot fail. A block stops the stream at the chunk that crossed the line rather than at the end of the answer.

guard = Vangrail::StreamGuard.new(engine, user_input: question)
stream.each do |chunk|
verdict = guard.push(chunk)
break if verdict&.blocked?
emit(guard.take)
end
final = guard.finish

The model-backed rails do not run per chunk. They cost a round trip, and calling one every few tokens turns a two second answer into a minute. They run once at finish, which is where the old behaviour still applies: a block there is a retraction, and the caller has to say so.

What this buys is bounded rather than total: everything the deterministic rails can see is caught before display, and everything only a model can see is caught at the end as before. That is worth stating plainly, because a stream guard that implied otherwise would be the more dangerous thing.

Constant Summary collapse

DEFAULT_INTERVAL =

How much new text has to arrive before the rails look again. A rail that runs per token spends more time in regexps than the model spends generating; one that runs per paragraph lets a whole paragraph through.

40

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(engine, interval: DEFAULT_INTERVAL, **context) ⇒ StreamGuard

Returns a new instance of StreamGuard.



44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/vangrail/stream_guard.rb', line 44

def initialize(engine, interval: DEFAULT_INTERVAL, **context)
  @engine = engine
  @context = context
  @interval = interval
  @buffer = +''
  @emitted = 0
  @checked = 0
  @checks = 0
  @blocked = nil
  @modified = false
  @released = +''
end

Instance Attribute Details

#buffer ⇒ Object (readonly)

Returns the value of attribute buffer.



42
43
44
 
# File 'lib/vangrail/stream_guard.rb', line 42

def buffer
  @buffer
end

#checked ⇒ Object (readonly)

Returns the value of attribute checked.



42
43
44
 
# File 'lib/vangrail/stream_guard.rb', line 42

def checked
  @checked
end

#checks ⇒ Object (readonly)

Returns the value of attribute checks.



42
43
44
 
# File 'lib/vangrail/stream_guard.rb', line 42

def checks
  @checks
end

#context ⇒ Object (readonly)

Returns the value of attribute context.



42
43
44
 
# File 'lib/vangrail/stream_guard.rb', line 42

def context
  @context
end

#emitted ⇒ Object (readonly)

Returns the value of attribute emitted.



42
43
44
 
# File 'lib/vangrail/stream_guard.rb', line 42

def emitted
  @emitted
end

#engine ⇒ Object (readonly)

Returns the value of attribute engine.



42
43
44
 
# File 'lib/vangrail/stream_guard.rb', line 42

def engine
  @engine
end

Instance Method Details

#blocked? ⇒ Boolean

Returns:

  • (Boolean) 


57
58
59
 
# File 'lib/vangrail/stream_guard.rb', line 57

def blocked?
  !@blocked.nil?
end

#content ⇒ Object

What the caller should show, given everything decided so far.



89
90
91
 
# File 'lib/vangrail/stream_guard.rb', line 89

def content
  buffer
end

#finish ⇒ Object

Everything the deterministic rails could not decide. Runs the full rail set, model-backed ones included, over the finished answer.



78
79
80
81
82
83
84
85
86
 
# File 'lib/vangrail/stream_guard.rb', line 78

def finish
  return @blocked if blocked?

  result = engine.check_output(buffer, **context)
  @blocked = result if result.blocked?
  @buffer = result.content_or(buffer) if result.modified?
  @checked = buffer.length unless result.blocked?
  result
end

#push(chunk) ⇒ Object

Adds a chunk and returns a Result when something changed, or nil when there is nothing to say. A caller that ignores the return value gets the old end-of-stream behaviour and nothing worse.



64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/vangrail/stream_guard.rb', line 64

def push(chunk)
  return @blocked if blocked?

  text = chunk.to_s
  return nil if text.empty?

  @buffer << text
  return nil unless due?

  inspect_buffer
end

#take ⇒ Object

Text the caller has not been given yet, and that a rail has read.

The second half of that sentence is the point. Only the inspected prefix is handed out: the tail that has arrived since the last check is held back until a check covers it, or until finish. Releasing it early would put text on screen that no rail has seen, which is the failure this class exists to prevent, and it is easy to write by accident because the buffer is right there.

The cost is that up to interval characters lag behind the model. The alternative is a guard that streams the credential and redacts it afterwards.

After a rewrite that keeps the already-shown prefix, this returns only the new suffix. After one that changes what was already shown, it returns the whole checked buffer, because the prefix on screen is no longer true.



109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/vangrail/stream_guard.rb', line 109

def take
  current = content[0, @checked].to_s
  if @released.empty? || current.start_with?(@released)
    out = current[@released.length..] || ''
    @released = current.dup
    return out
  end

  @released = current.dup
  current
end