Class: RobotLab::Network

Inherits:
Object
  • Object
show all
Includes:
Runnable, Utils
Defined in:
lib/robot_lab/network.rb

Overview

Orchestrates multiple robots in a pipeline workflow

Network is a thin wrapper around SimpleFlow::Pipeline that provides a clean DSL for defining robot workflows with sequential, parallel, and conditional execution.

Shared Memory

Networks provide a shared reactive memory that all robots can read and write. Robots can subscribe to memory keys and be notified when values change, or use blocking reads to wait for values from other robots.

Broadcast Messages

Networks support a broadcast channel for network-wide announcements. Use broadcast to send messages to all robots, and on_broadcast to register handlers for incoming broadcasts.

Examples:

Sequential execution

network = RobotLab.create_network(name: "pipeline") do
  task :analyst, analyst_robot, depends_on: :none
  task :writer, writer_robot, depends_on: [:analyst]
end

With per-task context

network = RobotLab.create_network(name: "support") do
  task :classifier, classifier_robot, depends_on: :none
  task :billing, billing_robot,
       context: { department: "billing" },
       tools: [RefundTool],
       depends_on: :optional
end

Parallel execution with shared memory

network = RobotLab.create_network(name: "analysis") do
  task :fetch, fetcher_robot, depends_on: :none
  task :sentiment, sentiment_robot, depends_on: [:fetch]
  task :entities, entity_robot, depends_on: [:fetch]
  task :summarize, summary_robot, depends_on: [:sentiment, :entities]
end

# In sentiment_robot:
memory.set(:sentiment, analyze_sentiment(text))

# In summarize_robot:
results = memory.get(:sentiment, :entities, wait: 60)

Broadcasting

network.on_broadcast do |message|
  puts "Received: #{message[:event]}"
end

network.broadcast(event: :pause, reason: "rate limit")

Constant Summary collapse

BROADCAST_KEY =

Reserved key for broadcast messages in memory

:_network_broadcast

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Runnable

#chief, #robot_count, #single?

Constructor Details

#initialize(name:, concurrency: :auto, memory: nil, config: nil, parallel_mode: :async) { ... } ⇒ Network

Creates a new Network instance.

Examples:

network = Network.new(name: "support") do
  task :classifier, classifier, depends_on: :none
  task :billing, billing_robot, context: { dept: "billing" }, depends_on: :optional
end

Parameters:

  • name (String) —

    unique identifier for the network

  • concurrency (Symbol) (defaults to: :auto) —

    concurrency model (:auto, :threads, :async)

  • memory (Memory, nil) (defaults to: nil) —

    optional pre-configured memory instance

Yields:

  • Block for defining pipeline tasks



90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/robot_lab/network.rb', line 90

def initialize(name:, concurrency: :auto, memory: nil, config: nil, parallel_mode: :async, &)
  @name = name.to_s
  @robots = {}
  @tasks = {}
  @pipeline = SimpleFlow::Pipeline.new(concurrency: concurrency)
  @memory = memory || Memory.new(network_name: @name)
  @config = config || RunConfig.new
  @parallel_mode = parallel_mode
  @hooks = HookRegistry.new
  @broadcast_handlers = []
  @bus_poller = BusPoller.new.start

  instance_eval(&) if block_given?
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



75
76
77
 
# File 'lib/robot_lab/network.rb', line 75

def config
  @config
end

#hooks ⇒ Object (readonly)

Returns the value of attribute hooks.



75
76
77
 
# File 'lib/robot_lab/network.rb', line 75

def hooks
  @hooks
end

#memory ⇒ Memory (readonly)

Returns shared memory for all robots in the network.

Returns:

  • (Memory) —

    shared memory for all robots in the network



75
 
# File 'lib/robot_lab/network.rb', line 75

attr_reader :name, :pipeline, :robots, :memory, :config, :parallel_mode, :hooks

#name ⇒ String (readonly)

Returns unique identifier for the network.

Returns:

  • (String) —

    unique identifier for the network



75
76
77
 
# File 'lib/robot_lab/network.rb', line 75

def name
  @name
end

#parallel_mode ⇒ Object (readonly)

Returns the value of attribute parallel_mode.



75
76
77
 
# File 'lib/robot_lab/network.rb', line 75

def parallel_mode
  @parallel_mode
end

#pipeline ⇒ SimpleFlow::Pipeline (readonly)

Returns the underlying pipeline.

Returns:

  • (SimpleFlow::Pipeline) —

    the underlying pipeline



75
 
# File 'lib/robot_lab/network.rb', line 75

attr_reader :name, :pipeline, :robots, :memory, :config, :parallel_mode, :hooks

#robots ⇒ Hash<String, Robot> (readonly)

Returns robots in this network, keyed by name.

Returns:

  • (Hash<String, Robot>) —

    robots in this network, keyed by name



75
 
# File 'lib/robot_lab/network.rb', line 75

attr_reader :name, :pipeline, :robots, :memory, :config, :parallel_mode, :hooks

Instance Method Details

#add_robot(robot) ⇒ self

Add a robot to the network without adding it as a task

Useful for dynamically adding robots that will be referenced later.

Parameters:

  • robot (Robot) —

    the robot instance to add

Returns:

  • (self)

Raises:

  • (ArgumentError) —

    if a robot with the same name already exists



331
332
333
334
335
336
337
338
 
# File 'lib/robot_lab/network.rb', line 331

def add_robot(robot)
  if @robots.key?(robot.name)
    raise ArgumentError, "Robot '#{robot.name}' already exists in network '#{@name}'"
  end

  @robots[robot.name] = robot
  self
end

#available_robots ⇒ Array<Robot>

Get all robots in the network

Returns:



319
320
321
 
# File 'lib/robot_lab/network.rb', line 319

def available_robots
  @robots.values
end

#broadcast(payload) ⇒ self

Broadcast a message to all robots in the network.

This sends a network-wide message that all robots subscribed via on_broadcast will receive asynchronously.

Examples:

Pause all robots

network.broadcast(event: :pause, reason: "rate limit hit")

Signal completion

network.broadcast(event: :phase_complete, phase: "analysis")

Parameters:

  • payload (Hash) —

    the message payload

Returns:

  • (self) 


246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/robot_lab/network.rb', line 246

def broadcast(payload)
  message = {
    payload: payload,
    network: @name,
    timestamp: Time.now
  }

  # Notify handlers asynchronously
  @broadcast_handlers.each do |handler|
    dispatch_async { handler.call(message) }
  end

  # Also set in memory so robots can subscribe via memory.subscribe
  @memory.set(BROADCAST_KEY, message)

  self
end

#crew ⇒ Array<Robot>

Runnable protocol: the network's robots as an Array (in pipeline order), and the network? predicate. (robots itself stays a name=>robot Hash.)

Returns:



219
220
221
 
# File 'lib/robot_lab/network.rb', line 219

def crew
  robots.values
end

#execution_plan ⇒ String?

Get the execution plan

Returns:

  • (String, nil) 


381
382
383
 
# File 'lib/robot_lab/network.rb', line 381

def execution_plan
  @pipeline.execution_plan
end

#network? ⇒ Boolean

Returns always true for a Network.

Returns:

  • (Boolean) —

    always true for a Network



224
225
226
 
# File 'lib/robot_lab/network.rb', line 224

def network?
  true
end

#on(handler_class, context: nil) ⇒ Object 



228
229
230
 
# File 'lib/robot_lab/network.rb', line 228

def on(handler_class, context: nil)
  @hooks.on(handler_class, context: context)
end

#on_broadcast {|Hash| ... } ⇒ self

Register a handler for broadcast messages.

The handler is called asynchronously whenever broadcast is called.

Examples:

network.on_broadcast do |message|
  case message[:payload][:event]
  when :pause
    pause_current_work
  when :resume
    resume_work
  end
end

Yields:

  • (Hash) —

    the broadcast message with :payload, :network, :timestamp

Returns:

  • (self)

Raises:

  • (ArgumentError) 


281
282
283
284
285
286
 
# File 'lib/robot_lab/network.rb', line 281

def on_broadcast(&block)
  raise ArgumentError, "Block required for on_broadcast" unless block_given?

  @broadcast_handlers << block
  self
end

#parallel(name = nil, depends_on: :none) { ... } ⇒ self

Define a parallel execution block

Examples:

Named parallel group

parallel :fetch_data, depends_on: :validate do
  task :fetch_orders, orders_robot
  task :fetch_products, products_robot
end
task :process, processor, depends_on: :fetch_data

Parameters:

  • name (Symbol, nil) (defaults to: nil) —

    optional name for the parallel group

  • depends_on (Symbol, Array) (defaults to: :none) —

    dependencies for this group

Yields:

  • Block containing task definitions

Returns:

  • (self) 


165
166
167
168
 
# File 'lib/robot_lab/network.rb', line 165

def parallel(name = nil, depends_on: :none, &)
  @pipeline.parallel(name, depends_on: depends_on, &)
  self
end

#remove_robot(name) ⇒ Robot?

Remove a dynamically-added robot from the network by name. Returns the removed robot, or nil if no robot by that name was present.

Only removes the robot from the crew (@robots); it does not rewrite the pipeline, so callers should not remove a robot that is a pipeline task.

Parameters:

  • name (String, Symbol) —

    the robot's name

Returns:



349
350
351
 
# File 'lib/robot_lab/network.rb', line 349

def remove_robot(name)
  @robots.delete(name.to_s)
end

#reset_memory ⇒ self

Reset the shared memory.

Clears all values in the network's shared memory. This is useful between runs if you want to start with a fresh memory state.

Returns:

  • (self) 


295
296
297
298
 
# File 'lib/robot_lab/network.rb', line 295

def reset_memory
  @memory.reset
  self
end

#robot(name) ⇒ Robot? Also known as: []

Get a robot by name

Parameters:

  • name (String, Symbol)

Returns:



305
306
307
 
# File 'lib/robot_lab/network.rb', line 305

def robot(name)
  @robots[name.to_s]
end

#run(message = nil, **run_context) ⇒ SimpleFlow::Result

Run the network with the given context

All robots share the network's memory during execution. The memory is passed to each robot and can be used for inter-robot communication.

Examples:

result = network.run(message: "I need help with billing", user_id: 123)
result.value  # => RobotResult from last robot
result.context[:classifier]  # => RobotResult from classifier

Parameters:

  • run_context (Hash) —

    context passed to all robots (message:, user_id:, etc.)

Returns:

  • (SimpleFlow::Result) —

    final pipeline result



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
 
# File 'lib/robot_lab/network.rb', line 183

def run(message = nil, **run_context)
  # Runnable protocol: accept a positional message like Robot#run does, so
  # callers can `run(msg, ...)` uniformly. `run(message: msg)` still works.
  run_context[:message] = message unless message.nil?

  # Include shared memory in run params so robots can access it
  run_context[:network_memory] = @memory
  run_context[:network] = self

  # Pass network's config so robots can inherit it
  run_context[:network_config] = @config unless @config.empty?

  context = NetworkRunHookContext.new(
    network: self,
    context: run_context,
    memory: @memory,
    config: @config
  )

  RobotLab::Hooks.run(:network_run, context, registries: [RobotLab.hooks, @hooks]) do
    if @parallel_mode == :ractor
      run_with_ractor_scheduler(context.context)
    else
      initial_result = SimpleFlow::Result.new(
        context.context,
        context: { run_params: context.context }
      )
      @pipeline.call_parallel(initial_result, max_concurrent: @config.max_concurrent_robots)
    end
  end
end

#task(name, robot, context: {}, mcp: :none, tools: :none, memory: nil, config: nil, depends_on: :none, poller_group: :default) ⇒ self

Add a robot as a pipeline task with optional per-task configuration

Examples:

Entry point task

task :classifier, classifier_robot, depends_on: :none

Task with context and tools

task :billing, billing_robot,
     context: { department: "billing", escalation: 2 },
     tools: [RefundTool, InvoiceTool],
     depends_on: :optional

Task with dependencies

task :writer, writer_robot, depends_on: [:analyst]

Parameters:

  • name (Symbol) —

    task name

  • robot (Robot) —

    the robot instance

  • context (Hash) (defaults to: {}) —

    task-specific context (deep-merged with run params)

  • mcp (Symbol, Array) (defaults to: :none) —

    MCP server config (:none, :inherit, or array)

  • tools (Symbol, Array) (defaults to: :none) —

    tools config (:none, :inherit, or array)

  • memory (Memory, Hash, nil) (defaults to: nil) —

    task-specific memory

  • depends_on (Symbol, Array<Symbol>) (defaults to: :none) —

    dependencies (:none, :optional, or task names)

Returns:

  • (self) 


128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/robot_lab/network.rb', line 128

def task(name, robot, context: {}, mcp: :none, tools: :none, memory: nil, config: nil, depends_on: :none,
         poller_group: :default)
  task_wrapper = Task.new(
    name: name,
    robot: robot,
    context: context,
    mcp: mcp,
    tools: tools,
    memory: memory,
    config: config,
    network: self
  )

  # Register the group and assign the shared poller to the robot
  @bus_poller.add_group(poller_group)
  robot.assign_bus_poller(@bus_poller, group: poller_group) if robot.respond_to?(:assign_bus_poller, true)

  @robots[name.to_s] = robot
  @tasks[name.to_s] = task_wrapper
  @pipeline.step(name, task_wrapper, depends_on: depends_on)
  self
end

#to_dot ⇒ String?

Export pipeline to DOT format (Graphviz)

Returns:

  • (String, nil) 


373
374
375
 
# File 'lib/robot_lab/network.rb', line 373

def to_dot
  @pipeline.visualize_dot
end

#to_h ⇒ Hash

Converts the network to a hash representation

Returns:

  • (Hash) 


389
390
391
392
393
394
395
396
397
 
# File 'lib/robot_lab/network.rb', line 389

def to_h
  {
    name: name,
    robots: @robots.keys,
    tasks: @tasks.keys,
    optional_tasks: @pipeline.optional_steps.to_a,
    config: (@config.empty? ? nil : @config.to_json_hash)
  }.compact
end

#to_mermaid ⇒ String?

Export pipeline to Mermaid format

Returns:

  • (String, nil) 


365
366
367
 
# File 'lib/robot_lab/network.rb', line 365

def to_mermaid
  @pipeline.visualize_mermaid
end

#visualize ⇒ String?

Visualize the pipeline as ASCII

Returns:

  • (String, nil) 


357
358
359
 
# File 'lib/robot_lab/network.rb', line 357

def visualize
  @pipeline.visualize_ascii
end