Normandy.Coordination.ParallelOrchestrator (normandy v2.2.0)

View Source

Runs agents concurrently on one input.

{:ok, results} =
  ParallelOrchestrator.execute(
    [%{id: "pros", agent: pros_agent}, %{id: "cons", agent: cons_agent}],
    %{chat_message: "Should we adopt Elixir?"},
    max_concurrency: 2
  )

results["pros"].chat_message

agents holds %Normandy.Agents.BaseAgentConfig{} structs (keyed "agent_0", "agent_1", ... in list order) or specs %{id: id, agent: config} with an optional :transform (fn output -> term end, applied to the agent's output) and an optional :input, which replaces input for that agent:

ParallelOrchestrator.execute(
  [%{id: "a", agent: agent, input: query_a}, %{id: "b", agent: agent, input: query_b}],
  nil
)

Results

{:ok, %{id => output}} when every agent succeeded; otherwise {:error, %Normandy.OrchestrationError{results: %{id => output}, errors: %{id => exception}}}, with every agent in one of the two maps. A failure never stops the other agents or exits the caller: an exception, an exit, a throw, a timeout and a crash of a process the agent linked to are all that agent's error (execute_stream/2 still links its tasks to the caller). See Normandy.OrchestrationError for what the exceptions are.

Telemetry

An agent that runs past :timeout is killed mid-run, so its [:normandy, :agent, :run] span never closes on its own. execute/3 and execute_stream/2 emit a synthetic [:normandy, :agent, :run, :exception] with reason: :killed for it (execute_stream/2 when the timed-out element is consumed) unless its run had already returned and the timeout hit its :transform or :on_agent_complete callback, possibly without a matching :start; the llm_call/tool spans open inside it stay open. See "Killed runs" in Normandy.Telemetry.

Summary

Functions

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

Runs agents concurrently, each on input or its spec's :input; see "Results".

execute_stream(agent_specs, opts \\ [])

Runs agent specs and yields each result as its agent finishes. Each element is one of

Functions

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

@spec execute([Normandy.Coordination.Orchestration.agent()], term(), keyword()) ::
  Normandy.Coordination.Orchestration.result()

Runs agents concurrently, each on input or its spec's :input; see "Results".

Options

  • :max_concurrency (pos_integer/0) - How many agents run at once. The default value is 10.

  • :timeout (timeout/0) - Each agent's limit in ms, or :infinity. An agent past it is stopped and fails with %Normandy.TurnError{type: :task_down, reason: :timeout}. The default value is 300000.

  • :on_agent_complete (function of arity 2) - fn agent_id, output -> any end, called in the agent's task after it succeeds (after :transform). A raise in it is that agent's error.

execute_stream(agent_specs, opts \\ [])

@spec execute_stream(
  [Normandy.Coordination.Orchestration.spec()],
  keyword()
) :: Enumerable.t()

Runs agent specs and yields each result as its agent finishes. Each element is one of:

  • {:ok, {:ok, {agent_id, output}}} - the agent finished
  • {:ok, {:error, {agent_id, exception}}} - the agent failed (exceptions as in Normandy.OrchestrationError)
  • {:exit, {agent_id, :timeout}} - the agent ran past :timeout and was stopped

Each spec runs on its own :input; a spec without one, or a bare agent config, runs with nil as its input.

Unlike execute/3, the tasks are linked to the process that consumes the stream: a crash in a process an agent linked to (which kills its task) exits the consumer instead of becoming that agent's error. Consume it from a process that may die with it, or use execute/3.

Emits no [:normandy, :workflow, :run] span: the stream is lazy, so it has no defined start and end (a current limitation).

Options

  • :max_concurrency (pos_integer/0) - How many agents run at once. The default value is 10.

  • :timeout (timeout/0) - Each agent's limit in ms, or :infinity. An agent past it is stopped and fails with %Normandy.TurnError{type: :task_down, reason: :timeout}. The default value is 300000.

Example

stream = ParallelOrchestrator.execute_stream(specs, timeout: 60_000)

stream
|> Stream.each(fn
  {:ok, {:ok, {agent_id, output}}} -> IO.puts("Agent #{agent_id}: #{inspect(output)}")
  {:ok, {:error, {agent_id, e}}} -> IO.puts("Agent #{agent_id} failed: #{Exception.message(e)}")
  {:exit, {agent_id, :timeout}} -> IO.puts("Agent #{agent_id} timed out")
end)
|> Stream.run()