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. Edges declare which node may run next.

class SupportFlowGraph < LittleGhost::Graph
node :triage, TriageAgent
node :ledger, LedgerResearchAgent
node :policy, PolicyResearchAgent
node :respond, CustomerSupportAgent

start :triage
edge :triage, :ledger
edge :triage, :policy
edge :ledger, :respond
edge :policy, :respond
finish :respond
end

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

Call a named Graph with ask for its final Run, or the streaming entrypoint for routing and final-response events.

Multiple unconditional edges from one source run in parallel and converge at their first unambiguous common successor. Array endpoints declare an explicit fan-out or wait-for-all fan-in. Parallel groups cannot nest. Conditions and input mappers receive a read-only 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. Conditions and mappers are application callbacks and can inspect copied input, history, context, and completed results. 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

#agent_stream_path, #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, #bind_agent_stream_path, #build_run, #call, definition, description, #entrypoint_name, #interject, #prompt_locals, #start_execution, stream_ask, #stream_ask, to_builder, validate_step_policy!

Constructor Details

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

:nodoc:



578
579
580
581
582
583
584
 
# File 'lib/little_ghost/graph.rb', line 578

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, max_concurrency: nil, **options, &condition) ⇒ Object

Declares one route or one bounded parallel edge group.

input receives Graph::State and returns the value passed to the target node or nodes. A scalar source and Array target fan out; an Array source and scalar target wait for every listed predecessor. At most one conditional scalar or grouped route may match from the current node; one unconditional route may act as the fallback. Multiple unconditional scalar edges with the same source infer one fan-out when no conditional route is present. Supply a condition with if: or a block.

max_concurrency overrides Graph.max_concurrency for a scalar-to-Array fan-out. The original request and complete source output cross to every branch unless an input mapper replaces them. Array-to-Array edges, conditional fan-in edges, and max_concurrency on other edge shapes raise ArgumentError.



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

def edge(from, to, input: nil, max_concurrency: nil, **options, &condition)
  condition = extract_condition(options, condition)
  validate_callable!(input, "edge input mapper")
  from = normalize_endpoint(from, "edge source")
  to = normalize_endpoint(to, "edge target")
  if from.is_a?(Array) && to.is_a?(Array)
    raise ArgumentError, "graph edges cannot use arrays for both source and target"
  end
  if from.is_a?(Array) && condition
    raise ArgumentError, "fan-in edges cannot be conditional"
  end
  unless max_concurrency.nil?
    raise ArgumentError, "max_concurrency is only valid for a fan-out edge" unless to.is_a?(Array)

    max_concurrency = normalize_max_concurrency(max_concurrency)
  end
  declaration = Edge.new(
    from:,
    to:,
    condition:,
    input_mapper: input,
    max_concurrency:
  )
  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.

on lists the exception classes this route accepts. An input mapper may turn Graph::State, including state.error, into recovery input.



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
 
# File 'lib/little_ghost/graph.rb', line 195

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.



212
213
214
215
216
 
# File 'lib/little_ghost/graph.rb', line 212

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

  self.graph_finish_value = normalize_node_name(name)
end

.graph_definition! ⇒ Object

:nodoc:

Raises:



247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
 
# File 'lib/little_ghost/graph.rb', line 247

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|
    Array(declaration.from).each { |name| validate_declared_node!(nodes, name, "edge source") }
    Array(declaration.to).each { |name| validate_declared_node!(nodes, name, "edge target") }
    if Array(declaration.from).include?(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
  edges, forks, joins = compile_edges
  validate_routes!(edges, forks)
  validate_parallel_structure!(edges, forks, joins)
  validate_success_routes!(nodes, finish_name, edges, forks, joins)
  validate_reachability!(nodes, start_name, finish_name, edges, forks, joins)
  [
    nodes, edges.freeze, graph_error_edges_value,
    forks.freeze, joins.freeze, start_name, finish_name
  ]
end

.max_concurrency(value = nil) ⇒ Object

Reads or assigns the concurrency bound for parallel groups.

The default is 8. A scalar-to-Array edge may override it for one group.



231
232
233
234
235
 
# File 'lib/little_ghost/graph.rb', line 231

def max_concurrency(value = nil)
  return graph_max_concurrency_value if value.nil?

  self.graph_max_concurrency_value = normalize_max_concurrency(value)
end

.max_steps(value = nil) ⇒ Object

Reads or assigns the maximum node executions.

Raises:

  • (ArgumentError) 


219
220
221
222
223
224
225
226
 
# File 'lib/little_ghost/graph.rb', line 219

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, input: nil) ⇒ Object

Declares an Assembly node and its optional execution policy.

An input mapper receives Graph::State and replaces the default input whenever the selected edge or edge group does not declare its own mapper. history and context opt this node into the corresponding caller data; both default to false.

Raises:



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

def node(name, assembly, timeout: nil, retries: 0, retry_on: nil, retry_delay: 0,
  history: false, context: false, input: nil)
  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
  validate_callable!(input, "node input mapper")

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

.start(name = nil) ⇒ Object

Reads or assigns the entry node.



143
144
145
146
147
 
# File 'lib/little_ghost/graph.rb', line 143

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.



281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
 
# File 'lib/little_ghost/graph.rb', line 281

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))"
  fan_in_pairs = joins.flat_map { |join| join.from.map { |source| [source, join.to] } }.to_set
  edges.each do |edge|
    next if fan_in_pairs.include?([edge.from, edge.to])

    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 topology and returns this Graph class.

Raises ConfigurationError for undeclared or unreachable nodes, ambiguous convergence, competing routes at an inferred branch boundary, and overlapping or nested parallel groups.



242
243
244
245
 
# File 'lib/little_ghost/graph.rb', line 242

def validate!
  graph_definition!
  self
end

Instance Method Details

#close ⇒ Object 



754
755
756
757
758
759
760
761
762
763
764
 
# File 'lib/little_ghost/graph.rb', line 754

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) 


587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
 
# File 'lib/little_ghost/graph.rb', line 587

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, []),
        incoming_results: join_context&.fetch(:results, {}) || {},
        error: routed_error
      )
      node_input = if join_context
        join_input_for(state, join_context.fetch(:join), nodes.fetch(current))
      else
        node_input_for(state, incoming_edge, nodes.fetch(current))
      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, max_concurrency: nil
        )
        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

      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 } + forks.select { |fork| fork.from == current },
        state
      )
      if selected.is_a?(Fork)
        fork = selected
        join = joins.find { |candidate| candidate.fork == 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:, source_step_id: previous_step_id,
          cancellation_token:, deadline:, settings:,
          template_locals:, template_paths:, parent_operation_id:, events:
        )
        unless join.from.sort == branch_outputs.map(&:terminal).sort
          raise AssemblyRoutingError, "graph fan-out at #{current.inspect} did not reach its inferred fan-in"
        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 = nil
        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

      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