Class: LittleGhost::Workflow

Inherits:
Object
  • Object
show all
Defined in:
lib/little_ghost/workflow.rb

Overview

Build agentic workflows with ordinary Ruby branching and local variables. A workflow composes several agents, consumes intermediate answers, and streams one final agent response.

A support workflow can route a difficult request through research before the responder writes the caller-visible answer:

class ResponseWorkflow < LittleGhost::Workflow
private

def perform
  route = invoke(RouterAgent).output
  return invoke(CustomerSupportAgent) unless route["research"]

  evidence = invoke(ResearchAgent).output
  invoke CustomerSupportAgent, input: <<~PROMPT
    #{input.text}

    Research:
    #{evidence}
  PROMPT
end
end

run = runtime.build_run(
{message: "Why is transfer 481 pending?"},
agent_class: CustomerSupportAgent,
entrypoint_class: ResponseWorkflow
).call
run.response # => "Transfer 481 is waiting for the receiving bank."

invoke returns a lazy Workflow::Invocation. Reading output consumes an intermediate invocation and returns RunResult#output; perform must return its final invocation without consuming it so those events reach the caller. Intermediate usage is added to the final result.

Every child inherits input, history, settings, cancellation, deadline, template paths, and trace parentage unless invoke overrides its input. JSON-like context is copied for each child, preventing one intermediate agent from mutating a sibling's state. Non-JSON-like workflow context raises ArgumentError.

A workflow instance streams once. Returning the wrong value, returning an already consumed invocation, or consuming an invocation twice raises ProtocolError. Built agents close in reverse order, and the first cleanup failure is re-raised after every invocation has been given a chance to close. Composition errors emit an invocation_error event and then re-raise.

Defined Under Namespace

Classes: Invocation

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(run:, runtime: run.runtime) ⇒ Workflow

:nodoc:



133
134
135
136
137
138
139
140
 
# File 'lib/little_ghost/workflow.rb', line 133

def initialize(run:, runtime: run.runtime) # :nodoc:
  @run = run
  @runtime = runtime
  @mutex = Mutex.new
  @closed = false
  @started = false
  @invocations = []
end

Instance Attribute Details

#run ⇒ Object (readonly)

Owning run and the runtime used to resolve workflow agents.



131
132
133
 
# File 'lib/little_ghost/workflow.rb', line 131

def run
  @run
end

#runtime ⇒ Object (readonly)

Owning run and the runtime used to resolve workflow agents.



131
132
133
 
# File 'lib/little_ghost/workflow.rb', line 131

def runtime
  @runtime
end

Instance Method Details

#close ⇒ Object

Closes all built agent invocations in reverse order.

The operation is idempotent, attempts every close, and raises the first cleanup failure.



222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/little_ghost/workflow.rb', line 222

def close
  invocations = @mutex.synchronize do
    return if @closed

    @closed = true
    @invocations.reverse
  end
  errors = []
  invocations.each do |invocation|
    invocation.close
  rescue => error
    errors << error
  end
  raise errors.first if errors.any?
end

#prompt_locals ⇒ Object

Additional prompt locals shared by agents invoked from the workflow. Subclasses may override this hook.



144
 
# File 'lib/little_ghost/workflow.rb', line 144

def prompt_locals = {}

#stream(input = nil, history: nil, context: nil, cancellation_token: Support::CancellationToken.new, deadline: nil, settings: nil, template_locals: nil, template_paths: nil, parent_operation_id: nil, checkpoint: nil) ⇒ Object

Streams the workflow once as StreamEvent objects.

perform must return a final, unconsumed Workflow::Invocation. The returned Enumerator is lazy, but calling stream reserves the single-use workflow instance even when enumeration has not started yet.

Raises:

  • (ArgumentError) 


151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/little_ghost/workflow.rb', line 151

def stream(
  input = nil,
  history: nil,
  context: nil,
  cancellation_token: Support::CancellationToken.new,
  deadline: nil,
  settings: nil,
  template_locals: nil,
  template_paths: nil,
  parent_operation_id: nil,
  checkpoint: nil
)
  raise ArgumentError, "input is required" if input.nil?

  @mutex.synchronize do
    raise Error, "workflow is already closed" if @closed
    raise Error, "workflow instances can only be streamed once" if @started

    @started = true
    @input = input.is_a?(Message) ? input : Message.new(role: :user, content: input)
    @history = normalize_history(history)
    @context = context || {}
    @cancellation_token = cancellation_token
    @deadline = deadline
    @settings = settings || {}
    @template_locals = template_locals || {}
    @template_paths = template_paths || []
    @parent_operation_id = parent_operation_id
    @checkpoint = checkpoint
    @intermediate_usage = Usage.new
  end

  Enumerator.new do |events|
    error_emitted = false
    observed_usage = nil
    ensure_open!
    final_invocation = perform
    unless final_invocation.is_a?(Invocation) && !final_invocation.consumed?
      raise ProtocolError, "#{self.class} must return its final invoke from perform"
    end

    final_invocation.each(checkpoint: @checkpoint) do |event|
      error_emitted = true if event.type == :invocation_error
      event = aggregate_usage(event)
      observed_usage = case event.type
      when :invocation_stop
        event.data.fetch(:result).usage
      when :invocation_error
        event.data[:usage] || observed_usage
      else
        observed_usage
      end
      events << event
    end
  rescue => error
    unless error_emitted
      events << StreamEvent.build(
        :invocation_error,
        error:,
        usage: observed_usage || workflow_usage,
        metadata: {}
      )
    end
    raise
  end
end