Class: Phronomy::Agent::Orchestrator

Inherits:
Base
  • Object
show all
Defined in:
lib/phronomy/agent/orchestrator.rb

Overview

Base class for orchestrator agents that coordinate multiple subagents. Implements the Orchestrator-Subagent multi-agent coordination pattern (Anthropic blog, Pattern 2).

Extends Base with:

  • A +subagent+ class-level DSL for declarative subagent registration. Each declared subagent is automatically exposed as an LLM-callable tool.
  • +dispatch_parallel+ for programmatic parallel invocation of heterogeneous agents.
  • +fan_out+ for parallel invocation of the same agent across multiple inputs.

Examples:

Declarative DSL

class ResearchOrchestrator < Phronomy::Agent::Orchestrator
  model "gpt-4o"
  instructions "You coordinate research tasks."
  subagent :searcher,   SearchAgent
  subagent :summarizer, SummaryAgent
end

result = ResearchOrchestrator.new.invoke("Research the latest AI news.")

Programmatic parallel dispatch

class MyOrchestrator < Phronomy::Agent::Orchestrator
  model "gpt-4o"
  instructions "Dispatch tasks in parallel."

  def run(input)
    results = dispatch_parallel(
      { agent: SearchAgent,   input: "topic A" },
      { agent: AnalysisAgent, input: input }
    )
    results.map { |r| r[:output] }.join("\n")
  end
end

Fan-out (same agent, multiple inputs)

results = fan_out(agent: TranslationAgent, inputs: ["Hello", "World"])

See Also:

Instance Attribute Summary

Attributes inherited from Base

#before_completion

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#_add_handoff_tool, _before_completion, #_handoff_tools, _on_compact_callback, _on_compaction_trigger_callback, _on_trim_callback, #add_input_guardrail, #add_output_guardrail, before_completion, cache_instructions, context_overhead, #context_version_cache, context_window, instructions, #invoke, max_iterations, max_output_tokens, model, #on_approval_required, on_compact, on_compaction_trigger, on_trim, provider, #resume, retry_policy, static_knowledge, static_knowledge_sources, #stream, temperature, tool_aliases, tools

Methods included from Runnable

#batch, #invoke, #stream, #trace

Class Method Details

.registered_subagents ⇒ Hash{Symbol => Hash}

Returns the subagent registry for this specific class (not inherited).

Returns:

  • (Hash{Symbol => Hash}) 


83
84
85
 
# File 'lib/phronomy/agent/orchestrator.rb', line 83

def self.registered_subagents
  @registered_subagents ||= {}
end

.subagent(name, agent_class, on_error: :raise) ⇒ Object

Declares a named subagent and registers it as a tool accessible to the LLM during an +invoke+ call.

Each call appends a new tool to this class's tool list. The generated tool's function name is +dispatch_to_+. When the LLM calls the tool, a fresh instance of +agent_class+ is created and +invoke+ is called with the provided input string.

Parameters:

  • name (Symbol) —

    logical name that identifies the subagent

  • agent_class (Class) —

    subclass of Base

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

    +:raise+ (default) re-raises any exception from the subagent; +:skip+ returns +nil+ so the LLM can decide how to proceed



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/phronomy/agent/orchestrator.rb', line 58

def self.subagent(name, agent_class, on_error: :raise)
  tool_class = Class.new(Phronomy::Tool::Base) do
    tool_name "dispatch_to_#{name}"
    description "Dispatch work to the #{name} subagent (#{agent_class.name})"
    param :input, type: :string, desc: "The task or question for the subagent"

    define_method(:execute) do |input:|
      result = agent_class.new.invoke(input)
      result[:output]
    rescue
      raise if on_error == :raise
      nil
    end
  end

  # Append without clobbering previously registered tools or aliases.
  @tools = (@tools || []) + [tool_class]
  @tool_aliases ||= {}

  registered_subagents[name] = {agent_class: agent_class, on_error: on_error}
end

Instance Method Details

#dispatch_parallel(*tasks) ⇒ Array<Hash>

Dispatches multiple heterogeneous agent tasks in parallel using Ruby threads. Each task is a Hash describing one agent invocation.

Results are returned in the same order as the input +tasks+ array. If any thread raises an exception, the exception is re-raised in the calling thread after all threads have completed (via +Thread#value+).

Parameters:

  • tasks (Array<Hash>)
  • task (Hash) —

    a customizable set of options

Returns:

  • (Array<Hash>) —

    agent results in the same order as +tasks+



99
100
101
102
103
104
105
106
 
# File 'lib/phronomy/agent/orchestrator.rb', line 99

def dispatch_parallel(*tasks)
  threads = tasks.map do |task|
    Thread.new do
      task[:agent].new.invoke(task[:input], config: task.fetch(:config, {}))
    end
  end
  threads.map(&:value)
end

#fan_out(agent:, inputs:, config: {}) ⇒ Array<Hash>

Runs the same agent against multiple inputs in parallel (fan-out pattern).

Parameters:

  • agent (Class) —

    agent class to invoke for every input

  • inputs (Array<String>) —

    list of input strings

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

    forwarded to every +agent#invoke+ call

Returns:

  • (Array<Hash>) —

    results in the same order as +inputs+



114
115
116
 
# File 'lib/phronomy/agent/orchestrator.rb', line 114

def fan_out(agent:, inputs:, config: {})
  dispatch_parallel(*inputs.map { |input| {agent: agent, input: input, config: config} })
end