Class: Phronomy::MultiAgent::TeamCoordinator

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/multi_agent/team_coordinator.rb

Overview

Implements the "Agent teams" coordination pattern (Anthropic blog, Pattern 3).

A coordinator LLM agent decomposes work into tasks and enqueues them dynamically via built-in tools. A fixed set of worker agents processes tasks sequentially — one task per worker per turn — carrying forward their conversation history across assignments to accumulate domain context over time.

Workers are selected in sequence (the worker with the fewest accumulated messages is chosen by default). Task dispatch is synchronous; there is no concurrent or parallel execution.

The coordinator is an Agent::Base subclass that has two built-in tools:

  • enqueue_task — adds a task description to the queue
  • finalize — signals that all tasks have been enqueued

Worker persistence is implemented by passing each worker's accumulated messages array back as a top-level messages: argument on every subsequent invoke call, so the LLM retains context across multiple task assignments.

Examples:

Basic usage

class MigrationTeam < Phronomy::MultiAgent::TeamCoordinator
  coordinator_model        "claude-3-5-sonnet-20241022"
  coordinator_instructions <<~INST
    Analyze the request and enqueue one migration task per service.
    Call enqueue_task for each service, then call finalize.
  INST

  pool size: 3, agent: MigrationAgent

  aggregate do |assignments|
    { reports: assignments.map { |a| { task: a[:task][:description], result: a[:result] } } }
  end
end

result = MigrationTeam.new.invoke("Migrate all services to Rails 8")

See Also:

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.aggregate {|Array<Hash>| ... } ⇒ Object

Defines how task assignments are merged into the final return value. The block receives an Array of assignment Hashes:

{ task: Hash, result: String|nil, worker: Integer, error: Exception|nil }

When omitted, the raw assignments array is returned.

Yields:

  • (Array<Hash>) —

    all completed (and skipped) task assignments



122
123
124
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 122

def aggregate(&block)
  @aggregator = block
end

.coordinator_instructions(value = nil) ⇒ Object

Sets the system instructions for the coordinator agent. The prompt should direct the LLM to call enqueue_task for each task and then call finalize when all tasks are enqueued.

Parameters:

  • value (String, nil) (defaults to: nil) 


75
76
77
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 75

def coordinator_instructions(value = nil)
  value ? @coordinator_instructions = value : @coordinator_instructions
end

.coordinator_model(value = nil) ⇒ Object

Sets the LLM model for the coordinator agent. Falls back to Phronomy.configuration.default_model when not set.

Parameters:

  • value (String, nil) (defaults to: nil) 


65
66
67
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 65

def coordinator_model(value = nil)
  value ? @coordinator_model = value : @coordinator_model
end

.coordinator_provider(value = nil) ⇒ Object

Sets the LLM provider for the coordinator agent. Required when using a custom BASE_URL (e.g. LM Studio, Ollama, vLLM) so that RubyLLM does not attempt to resolve an unknown model name. Pass the same value as LLMConfig::PROVIDER in your examples.

Parameters:

  • value (Symbol, nil) (defaults to: nil) 


86
87
88
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 86

def coordinator_provider(value = nil)
  value ? @coordinator_provider = value : @coordinator_provider
end

.pool(size:, agent:, on_error: :raise) ⇒ Object

Configures the set of workers.

Parameters:

  • size (Integer) —

    number of persistent worker instances (tasks are assigned sequentially)

  • agent (Class) —

    Agent::Base subclass used for all workers

  • on_error (Symbol) (defaults to: :raise) —

    :raise (default) propagates worker exceptions; :skip records the failure and continues with remaining tasks



97
98
99
100
101
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 97

def pool(size:, agent:, on_error: :raise)
  @pool_size = Integer(size)
  @worker_agent = agent
  @on_error = on_error
end

.schedule {|Array<WorkerState>| ... } ⇒ Object

Customises the worker selection algorithm. The block receives an Array of available WorkerState objects and must return the one to assign the next task to. Default: worker with the fewest accumulated messages (round-robin-like).

Yields:

  • (Array<WorkerState>) —

    available workers

Yield Returns:

  • (WorkerState) —

    the chosen worker



111
112
113
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 111

def schedule(&block)
  @scheduler = block
end

Instance Method Details

#invoke(team_input, config: {}) ⇒ Object

Runs the full team coordination: coordinator generates tasks, workers process them sequentially, and the aggregate block merges the results.

Parameters:

  • team_input (String, Hash) —

    the high-level objective given to the coordinator

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

    reserved for future use

Returns:

  • (Object) —

    the return value of the aggregate block, or the raw assignments Array

Raises:

  • (ArgumentError) —

    when pool :agent has not been configured



152
153
154
155
156
157
158
159
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 152

def invoke(team_input, config: {})
  raise ArgumentError, "pool :agent must be configured before invoking" unless self.class._worker_agent

  task_queue = []
  run_coordinator(team_input, task_queue)
  assignments = run_workers(task_queue)
  finalize_result(assignments)
end

#stream(team_input, config: {}) {|Hash| ... } ⇒ Object

Streaming version of invoke. Yields a Hash event for each completed or failed task assignment.

Yielded Hash keys:

:type   — +:task_completed+ or +:task_failed+
:worker — worker index (Integer)
:task   — the task Hash from the queue ({ id:, description:, metadata:, enqueued_at: })
:result — output string, or +nil+ on failure
:error  — Exception, or +nil+ on success

Parameters:

  • team_input (String, Hash)
  • config (Hash) (defaults to: {})

Yields:

  • (Hash) —

    one event per completed/failed task

Returns:

  • (Object) —

    same as invoke

Raises:

  • (ArgumentError) —

    when pool :agent has not been configured



177
178
179
180
181
182
183
184
185
186
 
# File 'lib/phronomy/multi_agent/team_coordinator.rb', line 177

def stream(team_input, config: {}, &block)
  return invoke(team_input, config: config) unless block

  raise ArgumentError, "pool :agent must be configured before invoking" unless self.class._worker_agent

  task_queue = []
  run_coordinator(team_input, task_queue)
  assignments = run_workers(task_queue, &block)
  finalize_result(assignments)
end