Class: LittleGhost::Workflow

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

Overview

Coordinates Assembly participants with ordinary Ruby control flow.

A workflow is an Assembly whose perform method controls ordering, branching, parallel work, and local variables. Each participant may be an Agent or another coordinated Assembly. The workflow consumes intermediate answers and streams one final participant response.

A support workflow can guarantee that research happens before the responder writes the caller-visible answer:

class ResponseWorkflow < LittleGhost::Workflow
private

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

    Research:
    #{evidence}
  PROMPT
end
end

run = ResponseWorkflow.ask("Why is transfer 481 pending?")
run.response
# One possible response: Transfer 481 is waiting for the receiving bank.

Call a named Workflow with ask for its final Run, or the streaming entrypoint for live events.

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.

A child receives the Workflow input unless invoke supplies another one. It also inherits history, settings, cancellation, deadline, template paths, and the parent tracing relationship. 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 one twice raises ProtocolError. A composition error fails the owning top-level Run. Each child Assembly closes after its attempt, and a cleanup failure raises from that attempt.

Defined Under Namespace

Classes: Invocation

Constant Summary

Constants inherited from Assembly

Assembly::MAX_STEP_EVENTS, Assembly::MAX_STEP_EVENT_BYTES, Assembly::MAX_STEP_OUTPUT_BYTES

Instance Attribute Summary collapse

Attributes inherited from Assembly

#sandbox, #workspace

Instance Method Summary collapse

Methods inherited from Assembly

#as_tool, ask, #ask, assembly_id, assembly_kind, #build_run, #call, definition, description, #entrypoint_name, #interject, #start_execution, stream_ask, #stream_ask, to_builder, validate_step_policy!

Methods included from Support::ClassAttributes

#class_attribute, included

Constructor Details

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

:nodoc:



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

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

Instance Attribute Details

#run ⇒ Object (readonly)

Run that owns this run-scoped Workflow.



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

def run
  @run
end

#runtime ⇒ Object (readonly)

Runtime used to resolve child Assemblies.



133
134
135
 
# File 'lib/little_ghost/workflow.rb', line 133

def runtime
  @runtime
end

Instance Method Details

#close ⇒ Object

Closes all declared invocations in reverse order.

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



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/little_ghost/workflow.rb', line 241

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.



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

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) 


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
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/little_ghost/workflow.rb', line 152

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?

  if standalone?
    return build_run(entrypoint_payload(input, {
      history:,
      context:,
      settings:,
      template_paths:,
      deadline_at: deadline,
      cancellation_token:
    }.compact)).each
  end

  @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
    @workflow_steps = []
    @workflow_events = nil
  end

  Enumerator.new do |events|
    error_emitted = false
    observed_usage = nil
    @workflow_events = events
    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
      when :assembly_step_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
  ensure
    @workflow_events = nil
  end
end