Normandy.Coordination.HierarchicalCoordinator (normandy v2.2.0)

Coordinates hierarchical multi-agent systems with manager-worker patterns.

The Hierarchical Coordinator implements a manager-worker pattern where a manager agent delegates tasks to worker agents and aggregates results.

execute/3 runs the manager twice, both times from the config the coordinator holds: the aggregation call is a fresh conversation that does not include the planning call; it gets the workers' results, each under its worker id.

Errors

{:error, {:manager_failed, exception}} when the planning call fails and {:error, {:aggregation_failed, exception}} when the aggregation call does. The exception is Normandy's own (Normandy.LLMError, Normandy.Schema.ValidationError, ...), the exception a bug raised (logged with its stacktrace, arities only), or a :task_down Normandy.TurnError for an exit or a throw. No stacktrace is returned. A failed worker does not fail the call: the manager is told about it.

Example

coordinator = HierarchicalCoordinator.new(
  manager: manager_agent,
  workers: [
    %{id: "worker_1", agent: worker1},
    %{id: "worker_2", agent: worker2}
  ],
  delegation_strategy: :round_robin
)

{:ok, result} = HierarchicalCoordinator.execute(coordinator, "Analyze this dataset")

Summary

Types

delegation_strategy()

t()

Functions

add_worker(coordinator, worker_id, agent)

Adds a worker to the coordinator.

execute(coordinator, input, opts \\ [])

Executes the hierarchical coordination.

execute_with_tasks(coordinator, tasks, opts \\ [])

Executes with explicit task delegation.

new(opts)

Creates a new hierarchical coordinator.

remove_worker(coordinator, worker_id)

Removes a worker from the coordinator.

Types

delegation_strategy()

@type delegation_strategy() :: :round_robin | :broadcast | :conditional

t()

@type t() :: %Normandy.Coordination.HierarchicalCoordinator{
  delegation_strategy: delegation_strategy(),
  manager: struct(),
  workers: [%{id: String.t(), agent: struct()}]
}

Functions

add_worker(coordinator, worker_id, agent)

@spec add_worker(t(), String.t(), struct()) :: t()

Adds a worker to the coordinator.

Example

coordinator = HierarchicalCoordinator.add_worker(
  coordinator,
  "new_worker",
  new_agent
)

execute(coordinator, input, opts \\ [])

@spec execute(t(), term(), keyword()) ::
  {:ok, term()}
  | {:error, {:manager_failed | :aggregation_failed, Exception.t()}}

Executes the hierarchical coordination.

The manager agent analyzes the input and decides how to delegate work to worker agents.

Options

:max_concurrency (pos_integer/0) - Workers run at once. The default value is 5.
:manager_prompt (String.t/0) - A prompt put before the input the manager gets.

Example

{:ok, result} = HierarchicalCoordinator.execute(
  coordinator,
  input,
  manager_prompt: "Break down this task for worker agents"
)

execute_with_tasks(coordinator, tasks, opts \\ [])

@spec execute_with_tasks(t(), [%{worker_id: String.t(), task: term()}], keyword()) ::
  {:ok, term()} | {:error, {:aggregation_failed, Exception.t()}}

Executes with explicit task delegation.

Bypasses manager's delegation planning and directly assigns tasks to workers.

Options

:max_concurrency (pos_integer/0) - Workers run at once. The default value is 5.

Example

tasks = [
  %{worker_id: "worker_1", task: "Research topic A"},
  %{worker_id: "worker_2", task: "Research topic B"}
]

{:ok, result} = HierarchicalCoordinator.execute_with_tasks(
  coordinator,
  tasks
)

new(opts)

@spec new(keyword()) :: t()

Creates a new hierarchical coordinator.

Options

:manager (term/0) - Required. The manager agent.
:workers (a list of %{id: String.t(), agent: agent} maps) - Required. The workers, each %{id: String.t(), agent: agent}.
:delegation_strategy - How the manager's plan is delegated. The default value is :round_robin.

Example

coordinator = HierarchicalCoordinator.new(
  manager: manager_agent,
  workers: [
    %{id: "worker_1", agent: worker1},
    %{id: "worker_2", agent: worker2}
  ],
  delegation_strategy: :broadcast
)

remove_worker(coordinator, worker_id)

@spec remove_worker(t(), String.t()) :: t()

Removes a worker from the coordinator.

Example

coordinator = HierarchicalCoordinator.remove_worker(coordinator, "worker_1")