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
56
57
58
59
 
# 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
  @released = +''
  # Rewrites happen in two places: the per-chunk pass, which runs its own
  # Engine over the offline rails, and `finish`, which runs the caller's. A
  # result from either one only knows about its own pass, so a redaction
  # applied mid-stream was missing from the report at the end.
  @rewritten_by = []
end

Instance Attribute Details

#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) 


61
62
63
 
# File 'lib/vangrail/stream_guard.rb', line 61

def blocked?
  !@blocked.nil?
end

#content ⇒ Object

The prefix a rail has read. The unread tail stays in the buffer.



98
99
100
 
# File 'lib/vangrail/stream_guard.rb', line 98

def content
  @buffer[0, @checked].to_s
end

#finish ⇒ Object

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



82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/vangrail/stream_guard.rb', line 82

def finish
  return @blocked if blocked?

  result = engine.check_output(buffer, **context)
  result = merge_rewrites(result)
  @blocked = result if result.blocked?
  # Duplicated, because the buffer is appended to in place and a rewrite
  # hands back a string the rail may still own. A memoized rail returns the
  # same Result to the next caller with the same text, so appending to its
  # content puts one turn's tokens inside another turn's answer.
  @buffer = result.content_or(buffer).dup 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.



68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/vangrail/stream_guard.rb', line 68

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.



118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/vangrail/stream_guard.rb', line 118

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

  @released = current.dup
  current
end