Class: LittleGhost::Graph

Inherits:
Assembly show all
Extended by:
Support::ClassAttributes
Defined in:
lib/little_ghost/graph.rb

Overview

Routes a request through named Assembly nodes and declared edges.

A graph is an Assembly for flows whose allowed paths should be visible in application code. A node may contain an Agent, Workflow, Swarm, or another Graph. Ordinary edges choose one next node; explicit forks and joins add bounded parallel work.

class SupportFlowGraph < LittleGhost::Graph
node :triage, TriageAgent
node :research, ResearchAgent
node :respond, CustomerSupportAgent

start :triage
edge :triage, :research
edge :research, :respond
finish :respond
end

SupportFlowGraph.validate!
run = SupportFlowGraph.ask("Why is my transfer pending?")

Conditions and input mappers receive immutable Graph::State. Nodes do not receive caller history or application context unless their declaration opts in with history: true or context: true. Validate the topology before execution; to_mermaid renders the same definition as a flowchart.

Defined Under Namespace

Classes: BranchResult, Edge, ErrorEdge, EventSink, Fork, Join, Node, State

Constant Summary

Constants inherited from Assembly

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

Instance Attribute Summary

Attributes inherited from Assembly

#run, #runtime, #sandbox, #workspace

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Support::ClassAttributes

class_attribute, included

Methods inherited from Assembly

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

Constructor Details

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

:nodoc:



410
411
412
413
414
415
416
 
# File 'lib/little_ghost/graph.rb', line 410

def initialize(run: nil, runtime: nil) # :nodoc:
  super(run:, runtime:, standalone: run.nil?)
  @graph_mutex = Mutex.new
  @graph_started = false
  @graph_children = []
  @graph_execution_count = 0
end

Class Method Details

.edge(from, to, input: nil, **options, &condition) ⇒ Object

Declares one exclusive route with an optional condition and input mapper.



112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/little_ghost/graph.rb', line 112

def edge(from, to, input: nil, **options, &condition)
  condition = extract_condition(options, condition)
  validate_callable!(input, "edge input mapper")
  declaration = Edge.new(
    from: normalize_node_name(from),
    to: normalize_node_name(to),
    condition:,
    input_mapper: input
  )
  self.graph_edges_value = (graph_edges_value + [declaration]).freeze
  declaration
end

.error_edge(from, to, on:, input: nil) ⇒ Object

Routes selected node errors after retries are exhausted.



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/little_ghost/graph.rb', line 126

def error_edge(from, to, on:, input: nil)
  errors = Array(on)
  unless errors.any? && errors.all? { |error| error.is_a?(Class) && error <= Exception }
    raise ArgumentError, "error edge on: must contain exception classes"
  end
  validate_callable!(input, "error edge input mapper")
  declaration = ErrorEdge.new(
    from: normalize_node_name(from),
    to: normalize_node_name(to),
    errors: errors.freeze,
    input_mapper: input
  )
  self.graph_error_edges_value = (graph_error_edges_value + [declaration]).freeze
  declaration
end

.finish(name = nil) ⇒ Object

Reads or assigns the terminal node.



171
172
173
174
175
 
# File 'lib/little_ghost/graph.rb', line 171

def finish(name = nil)
  return graph_finish_value if name.nil?

  self.graph_finish_value = normalize_node_name(name)
end

.fork(from, to:, max_concurrency: 8) ⇒ Object

Starts two or more independent branches with bounded concurrency.

Raises:

  • (ArgumentError) 


143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/little_ghost/graph.rb', line 143

def fork(from, to:, max_concurrency: 8)
  targets = Array(to).map { |name| normalize_node_name(name) }
  raise ArgumentError, "fork requires at least two targets" if targets.length < 2
  raise ArgumentError, "fork targets must be unique" unless targets.uniq.length == targets.length
  max_concurrency = Integer(max_concurrency)
  raise ArgumentError, "max_concurrency must be at least 1" if max_concurrency < 1

  declaration = Fork.new(from: normalize_node_name(from), to: targets.freeze, max_concurrency:)
  self.graph_forks_value = (graph_forks_value + [declaration]).freeze
  declaration
end

.graph_definition! ⇒ Object

:nodoc:

Raises:



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
 
# File 'lib/little_ghost/graph.rb', line 193

def graph_definition! # :nodoc:
  nodes = graph_nodes_value
  start_name = graph_start_value
  finish_name = graph_finish_value
  raise ConfigurationError, "graph must declare at least one node" if nodes.empty?
  raise ConfigurationError, "graph must declare a start node" unless start_name
  raise ConfigurationError, "graph must declare a finish node" unless finish_name
  validate_declared_node!(nodes, start_name, "start")
  validate_declared_node!(nodes, finish_name, "finish")
  nodes.each_value { |node| validate_step_policy!(node.policies) }

  graph_edges_value.each do |declaration|
    validate_declared_node!(nodes, declaration.from, "edge source")
    validate_declared_node!(nodes, declaration.to, "edge target")
    if declaration.from == finish_name
      raise ConfigurationError, "graph finish node #{finish_name.inspect} cannot have outgoing edges"
    end
  end
  graph_error_edges_value.each do |declaration|
    validate_declared_node!(nodes, declaration.from, "error edge source")
    validate_declared_node!(nodes, declaration.to, "error edge target")
  end
  graph_forks_value.each do |declaration|
    validate_declared_node!(nodes, declaration.from, "fork source")
    declaration.to.each { |target| validate_declared_node!(nodes, target, "fork target") }
    if graph_forks_value.count { |fork| fork.from == declaration.from } > 1
      raise ConfigurationError, "graph node #{declaration.from.inspect} has more than one fork"
    end
  end
  graph_joins_value.each do |declaration|
    declaration.from.each { |source| validate_declared_node!(nodes, source, "join source") }
    validate_declared_node!(nodes, declaration.to, "join target")
  end
  validate_parallel_structure!
  validate_success_routes!(nodes, finish_name)
  validate_reachability!(nodes, start_name, finish_name)
  [
    nodes, graph_edges_value, graph_error_edges_value,
    graph_forks_value, graph_joins_value, start_name, finish_name
  ]
end

.join(from, to:, input: nil) ⇒ Object

Joins the terminal results of one declared fork.

Raises:

  • (ArgumentError) 


156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/little_ghost/graph.rb', line 156

def join(from, to:, input: nil)
  sources = Array(from).map { |name| normalize_node_name(name) }
  raise ArgumentError, "join requires at least two sources" if sources.length < 2
  raise ArgumentError, "join sources must be unique" unless sources.uniq.length == sources.length
  validate_callable!(input, "join input mapper")
  declaration = Join.new(
    from: sources.freeze,
    to: normalize_node_name(to),
    input_mapper: input
  )
  self.graph_joins_value = (graph_joins_value + [declaration]).freeze
  declaration
end

.max_steps(value = nil) ⇒ Object

Reads or assigns the maximum node executions.

Raises:

  • (ArgumentError) 


178
179
180
181
182
183
184
185
 
# File 'lib/little_ghost/graph.rb', line 178

def max_steps(value = nil)
  return graph_max_steps_value if value.nil?

  value = Integer(value)
  raise ArgumentError, "max_steps must be at least 1" if value < 1

  self.graph_max_steps_value = value
end

.node(name, assembly, timeout: nil, retries: 0, retry_on: nil, retry_delay: 0, history: false, context: false) ⇒ Object

Declares an Assembly node and its optional execution policy.

Raises:



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/little_ghost/graph.rb', line 87

def node(name, assembly, timeout: nil, retries: 0, retry_on: nil, retry_delay: 0,
  history: false, context: false)
  name = normalize_node_name(name)
  raise ConfigurationError, "graph node #{name.inspect} is already declared" if graph_nodes_value.key?(name)
  unless [history, context].all? { |value| value == true || value == false }
    raise ArgumentError, "graph node history and context options must be true or false"
  end

  policies = {timeout:, retries:, retry_on:, retry_delay:}.freeze
  declaration = Node.new(
    name:, assembly:, policies:,
    inherit_history: history,
    inherit_context: context
  )
  self.graph_nodes_value = graph_nodes_value.merge(name => declaration).freeze
end

.start(name = nil) ⇒ Object

Reads or assigns the entry node.



105
106
107
108
109
 
# File 'lib/little_ghost/graph.rb', line 105

def start(name = nil)
  return graph_start_value if name.nil?

  self.graph_start_value = normalize_node_name(name)
end

.to_mermaid ⇒ Object

Renders the validated topology as Mermaid flowchart text.



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/little_ghost/graph.rb', line 236

def to_mermaid
  nodes, edges, error_edges, forks, joins, start_name, finish_name = graph_definition!
  lines = ["flowchart TD"]
  nodes.each_key { |name| lines << "  #{mermaid_id(name)}[#{name}]" }
  lines << "  START((start)) --> #{mermaid_id(start_name)}"
  lines << "  #{mermaid_id(finish_name)} --> FINISH((finish))"
  edges.each do |edge|
    label = edge.condition ? "condition" : nil
    lines << mermaid_edge(edge.from, edge.to, label:)
  end
  error_edges.each { |edge| lines << mermaid_edge(edge.from, edge.to, label: "error", dotted: true) }
  forks.each do |fork|
    fork.to.each { |target| lines << mermaid_edge(fork.from, target, label: "fork") }
  end
  joins.each do |join|
    join.from.each { |source| lines << mermaid_edge(source, join.to, label: "join") }
  end
  lines.join("\n")
end

.validate! ⇒ Object

Validates the current topology and returns this Graph class.



188
189
190
191
 
# File 'lib/little_ghost/graph.rb', line 188

def validate!
  graph_definition!
  self
end

Instance Method Details

#close ⇒ Object 



579
580
581
582
583
584
585
586
587
588
589
 
# File 'lib/little_ghost/graph.rb', line 579

def close
  children = @graph_mutex.synchronize { @graph_children.reverse }
  first_error = nil
  children.each do |child|
    child.close
  rescue => error
    first_error ||= error
  end
  super
  raise first_error if first_error
end

#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, **_options) ⇒ Object

Streams lifecycle events and the finish node's ordinary response events.

Raises:

  • (ArgumentError) 


419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
 
# File 'lib/little_ghost/graph.rb', line 419

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, **_options)
  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

  reserve_execution!
  original_input = input.is_a?(Message) ? input : Message.new(role: :user, content: input)
  original_history = normalize_history(history)
  original_context = frozen_state(context || {})
  settings ||= {}
  template_locals ||= {}
  template_paths ||= []

  usage = Usage.new
  error_emitted = false
  Enumerator.new do |events|
    definition = self.class.graph_definition!
    nodes, edges, error_edges, forks, _joins, current, finish = definition
    results = {}
    steps = []
    previous = nil
    incoming_edge = nil
    join_context = nil
    routed_error = nil
    previous_step_id = nil

    loop do
      count = next_graph_step!(cancellation_token, deadline)
      state = routing_state(
        input: original_input, history: original_history, context: original_context,
        step: count, current:, previous:, results:,
        predecessors: join_context&.fetch(:predecessors, []),
        branch_results: join_context&.fetch(:results, {}) || {},
        error: routed_error
      )
      node_input = if join_context
        join_input_for(state, join_context.fetch(:join))
      else
        node_input_for(state, incoming_edge)
      end
      terminal = current == finish
      begin
        execution = execute_graph_node(
          node: nodes.fetch(current), input: node_input, history: original_history,
          context: original_context, cancellation_token:, deadline:, settings:,
          template_locals:, template_paths:, parent_operation_id:,
          predecessor_ids: join_context&.fetch(:step_ids, []) || Array(previous_step_id),
          terminal:, events:
        )
      rescue => error
        route = select_error_edge(error_edges.select { |edge| edge.from == current }, error)
        raise unless route

        usage += step_error_usage(error)
        error.instance_variable_set(:@little_ghost_step_usage_accounted, true)
        failed = failed_step(
          error,
          nodes.fetch(current),
          current,
          predecessor_ids: Array(previous_step_id)
        )
        steps << failed
        previous_step_id = failed.id
        incoming_edge = Edge.new(from: current, to: route.to, condition: nil, input_mapper: route.input_mapper)
        events << transition_event(count, current, route.to, error: true)
        previous = current
        current = route.to
        join_context = nil
        routed_error = error
        next
      end

      results[current] = execution.result
      previous_step_id = execution.step.id
      usage += execution.step.usage
      steps.concat(execution.result.steps)
      if terminal
        final = copy_run_result(execution.result, usage:, steps: steps.freeze)
        execution.events.each do |event|
          event = StreamEvent.build(event.type, **event.data.merge(result: final)) if event.type == :invocation_stop
          error_emitted = true if event.type == :invocation_error
          events << event
        end
        break
      end

      fork = forks.find { |declaration| declaration.from == current }
      if fork
        join = self.class.send(:matching_join_for, fork)
        events << StreamEvent.build(
          :assembly_fork,
          assembly_id: self.class.assembly_id,
          assembly_kind: :graph,
          from: current,
          branches: fork.to
        )
        branch_outputs = run_graph_branches(
          fork:, join:, nodes:, edges:, error_edges:,
          original_input:, original_history:, original_context:,
          results:, cancellation_token:, deadline:, settings:,
          template_locals:, template_paths:, parent_operation_id:, events:
        )
        unless join.from.sort == branch_outputs.map(&:terminal).sort
          raise AssemblyRoutingError, "graph fork at #{current.inspect} did not reach its declared join"
        end

        branch_outputs.each do |branch|
          results.merge!(branch.results)
          steps.concat(branch.steps)
          usage += branch.usage
          branch.events.each { |event| events << event }
        end
        events << StreamEvent.build(
          :assembly_join,
          assembly_id: self.class.assembly_id,
          assembly_kind: :graph,
          from: join.from,
          to: join.to
        )
        previous = current
        current = join.to
        incoming_edge = nil
        join_context = {
          join:,
          predecessors: join.from,
          step_ids: branch_outputs.map { |branch| branch.steps.last.id },
          results: join.from.to_h { |name| [name, results.fetch(name)] }
        }
        previous_step_id = nil
        next
      end

      state = routing_state(
        input: original_input, history: original_history, context: original_context,
        step: count, current:, previous:, results:
      )
      selected = select_edge(edges.select { |edge| edge.from == current }, state)
      events << transition_event(count, current, selected.to)
      previous = current
      current = selected.to
      incoming_edge = selected
      join_context = nil
      routed_error = nil
    end
  rescue => error
    usage += unaccounted_step_error_usage(error)
    unless error_emitted
      events << StreamEvent.build(:invocation_error, error:, usage:, metadata: {})
    end
    raise
  end
end