class LittleGhost::Workflow

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.

Attributes

run [R]

Owning run and the runtime used to resolve workflow agents.

runtime [R]

Owning run and the runtime used to resolve workflow agents.

Public Instance Methods

close ()

Source

# 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

Closes all built agent invocations in reverse order.

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

prompt_locals ()

Source

# File lib/little_ghost/workflow.rb, line 144
def prompt_locals = {}

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

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)

Source

# 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

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.

Private Instance Methods

invoke (agent_class_or_name, input: self.input, history: self.history, context: self.context)

Source

# File lib/little_ghost/workflow.rb, line 257
def invoke(agent_class_or_name, input: self.input, history: self.history, context: self.context)
  invocation = Invocation.new(
    input:,
    history:,
    context: isolated_state(context),
    build: lambda {
      agent = runtime.build_agent(agent_class_or_name, run:)
      begin
        [
          agent,
          {
            cancellation_token: @cancellation_token,
            deadline: @deadline,
            settings: @settings,
            template_locals: template_locals_for(agent),
            template_paths: @template_paths,
            parent_operation_id: @parent_operation_id
          }
        ]
      rescue
        agent.close
        raise
      end
    },
    on_usage: ->(usage) { record_intermediate_usage(usage) }
  )
  @mutex.synchronize do
    raise Error, "workflow is already closed" if @closed

    @invocations << invocation
  end
  invocation
end

Creates a lazy invocation for agent_class_or_name.

Intermediate calls may use output; the final call must be returned from perform without being consumed.

perform ()

Source

# File lib/little_ghost/workflow.rb, line 248
def perform
  raise NotImplementedError, "#{self.class} must implement #perform"
end

Implements the composition and returns its final unconsumed invocation. Subclasses must override this hook.