Class: RCrewAI::Flow

Inherits:
Object
  • Object
show all
Defined in:
lib/rcrewai/flow.rb,
lib/rcrewai/flow/state.rb,
lib/rcrewai/flow/state_store.rb

Overview

Event-driven workflow engine — CrewAI's second pillar, in Ruby.

Subclass Flow and declare methods with the class-level DSL:

class GuideFlow < RCrewAI::Flow
start :pick_topic
def pick_topic = state.topic = 'ruby'

listen :pick_topic
def research(prev) = "researched #{prev}"

router :research
def route(prev) = prev.include?('ruby') ? :publish : :revise

listen :publish
def publish = state.done = true
end

GuideFlow.new.kickoff

Triggers combine with and_/or_. State is a schemaless object with a UUID and can be persisted/restored via a state store.

Defined Under Namespace

Classes: FileStateStore, MemoryStateStore, State, Trigger

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(state_store: nil, feedback_handler: nil) ⇒ Flow

Returns a new instance of Flow.



109
110
111
112
113
 
# File 'lib/rcrewai/flow.rb', line 109

def initialize(state_store: nil, feedback_handler: nil)
  @state = State.new
  @state_store = state_store
  @feedback_handler = feedback_handler
end

Instance Attribute Details

#state ⇒ Object (readonly)

--- Instance ------------------------------------------------------------



107
108
109
 
# File 'lib/rcrewai/flow.rb', line 107

def state
  @state
end

Class Method Details

.and_(*names) ⇒ Object 



72
73
74
 
# File 'lib/rcrewai/flow.rb', line 72

def and_(*names)
  Trigger.new(:and, names.map(&:to_sym))
end

.inherited(subclass) ⇒ Object

Merge inherited declarations so subclasses of a Flow subclass compose.



90
91
92
93
94
95
 
# File 'lib/rcrewai/flow.rb', line 90

def inherited(subclass)
  super
  subclass.instance_variable_set(:@start_methods, start_methods.dup)
  subclass.instance_variable_set(:@listeners, listeners.dup)
  subclass.instance_variable_set(:@routers, routers.dup)
end

.listen(trigger) ⇒ Object 



60
61
62
 
# File 'lib/rcrewai/flow.rb', line 60

def listen(trigger)
  @pending = [:listen, normalize_trigger(trigger)]
end

.listeners ⇒ Object

name => Trigger. Populated when a listen/router declaration is bound to the next defined method.



48
49
50
 
# File 'lib/rcrewai/flow.rb', line 48

def listeners
  @listeners ||= {}
end

.method_added(method_name) ⇒ Object

Binds a pending listen/router declaration to the method just defined.



77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/rcrewai/flow.rb', line 77

def method_added(method_name)
  super
  return unless @pending

  kind, trigger = @pending
  @pending = nil
  case kind
  when :listen then listeners[method_name.to_sym] = trigger
  when :router then routers[method_name.to_sym] = trigger
  end
end

.or_(*names) ⇒ Object 



68
69
70
 
# File 'lib/rcrewai/flow.rb', line 68

def or_(*names)
  Trigger.new(:or, names.map(&:to_sym))
end

.router(trigger) ⇒ Object 



64
65
66
 
# File 'lib/rcrewai/flow.rb', line 64

def router(trigger)
  @pending = [:router, normalize_trigger(trigger)]
end

.routers ⇒ Object 



52
53
54
 
# File 'lib/rcrewai/flow.rb', line 52

def routers
  @routers ||= {}
end

.start(method_name) ⇒ Object 



56
57
58
 
# File 'lib/rcrewai/flow.rb', line 56

def start(method_name)
  start_methods << method_name.to_sym
end

.start_methods ⇒ Object 



42
43
44
 
# File 'lib/rcrewai/flow.rb', line 42

def start_methods
  @start_methods ||= []
end

Instance Method Details

#human_feedback(prompt) ⇒ Object

A pause point for human feedback. Calls the configured feedback_handler with the prompt and returns its response; without a handler, prompts on the console. Mirrors CrewAI's @human_feedback.



118
119
120
121
122
123
124
 
# File 'lib/rcrewai/flow.rb', line 118

def human_feedback(prompt)
  return @feedback_handler.call(prompt) if @feedback_handler

  require_relative 'human_input'
  response = HumanInput.new.request_input(prompt)
  response.is_a?(Hash) ? response[:input] : response
end

#kickoff(inputs: {}) ⇒ Object

Runs the flow to completion. Optional inputs seed the state.



127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/rcrewai/flow.rb', line 127

def kickoff(inputs: {})
  inputs.each { |k, v| @state[k] = v }

  @completed = []       # method names that have finished
  @outputs = {}         # method name => return value
  @router_labels = []   # labels emitted by routers, act as pseudo-triggers

  self.class.start_methods.each { |m| run_method(m) }
  drain_listeners

  persist
  @state
end

#restore(id) ⇒ Object

Restores state previously persisted under id.

Raises:



142
143
144
145
146
147
148
149
150
 
# File 'lib/rcrewai/flow.rb', line 142

def restore(id)
  raise FlowError, 'no state store configured' unless @state_store

  hash = @state_store.load(id)
  raise FlowError, "no persisted state for id #{id}" unless hash

  @state = State.new(symbolize(hash))
  @state
end