Normandy.Coordination.SequentialOrchestrator (normandy v0.2.0)

Orchestrates sequential execution of multiple agents.

The Sequential Orchestrator runs agents one after another, passing the output of one agent as input to the next. This creates a pipeline where each agent builds upon the work of previous agents.

Example

# Define agents
agents = [
  %{id: "research", agent: research_agent, transform: &extract_data/1},
  %{id: "analyze", agent: analysis_agent, transform: &format_analysis/1},
  %{id: "write", agent: writing_agent}
]

# Execute pipeline
{:ok, result} = SequentialOrchestrator.execute(
  agents,
  initial_input,
  shared_context: context
)

Summary

Types

agent_spec()

execution_result()

Functions

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

Executes agents sequentially in a pipeline.

execute_with_messages(agent_specs, initial_message, opts \\ [])

Executes agents with message-based communication.

Types

agent_spec()

@type agent_spec() :: %{
  :id => String.t(),
  :agent => struct(),
  optional(:transform) => (term() -> term())
}

execution_result()

@type execution_result() :: %{
  success: boolean(),
  results: [term()],
  errors: [term()],
  context: Normandy.Coordination.SharedContext.t()
}

Functions

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

@spec execute([agent_spec()] | [struct()], term(), keyword()) ::
  {:ok, execution_result()} | {:ok, term()} | {:error, term()}

Executes agents sequentially in a pipeline.

Can be called in two ways:

With agent_specs and options: execute(agent_specs, input, opts)
With agents list and input: execute(agents, input) - returns final result directly

Options

:shared_context - SharedContext to use (default: new context)
:on_agent_complete - Callback after each agent: (agent_id, result, context -> any)
:on_error - Error handling: :stop (default) or :continue
:store_intermediate - Store each result in shared context (default: true)

Example

{:ok, result} = SequentialOrchestrator.execute(
  agents,
  initial_input,
  shared_context: context,
  on_agent_complete: fn agent_id, _result, _ctx ->
    IO.puts("Agent #{agent_id} completed")
  end
)

execute_with_messages(agent_specs, initial_message, opts \\ [])

@spec execute_with_messages(
  [agent_spec()],
  Normandy.Coordination.AgentMessage.t(),
  keyword()
) ::
  {:ok, [Normandy.Coordination.AgentMessage.t()]}

Executes agents with message-based communication.

Each agent receives an AgentMessage and returns an AgentMessage response.

Example

messages = SequentialOrchestrator.execute_with_messages(
  agents,
  initial_message,
  shared_context: context
)