Normandy.Batch.Processor (normandy v2.2.0)

Batch processing utilities for handling multiple agent requests concurrently.

Provides efficient batch processing with configurable concurrency, a delay between chunks, and result aggregation for AI agent operations.

Features

Concurrent request processing with Task.async_stream
Configurable concurrency limits
Ordered and unordered result collection
Error handling and partial failure support
Progress tracking callbacks

Example

# Basic batch processing
inputs = [
  %{chat_message: "Hello"},
  %{chat_message: "How are you?"},
  %{chat_message: "Goodbye"}
]

{:ok, results} = Normandy.Batch.Processor.process_batch(
  agent,
  inputs,
  max_concurrency: 5
)

# With progress callback
{:ok, results} = Normandy.Batch.Processor.process_batch(
  agent,
  inputs,
  on_progress: fn completed, total ->
    IO.puts("Progress: #{completed}/#{total}")
  end
)

Telemetry

An item that runs past :timeout is killed mid-run, so its [:normandy, :agent, :run] span never closes on its own. The processor emits a synthetic [:normandy, :agent, :run, :exception] with reason: :killed for it, possibly without a matching :start; the llm_call/tool spans open inside it stay open. See "Killed runs" in Normandy.Telemetry.

Summary

Types

batch_option()

batch_options()

batch_result()

error_callback()

item_result()

progress_callback()

Functions

process_batch(agent, inputs, opts \\ [])

Process a batch of inputs through an agent concurrently.

process_batch_chunked(agent, inputs, opts \\ [])

Process a batch in chunks with a delay between chunks.

process_batch_with_stats(agent, inputs, opts \\ [])

Process a batch and return detailed statistics.

Types

batch_option()

@type batch_option() ::
  {:max_concurrency, pos_integer()}
  | {:ordered, boolean()}
  | {:timeout, timeout()}
  | {:on_progress, progress_callback()}
  | {:on_error, error_callback()}
  | {:chunk_size, pos_integer()}
  | {:chunk_delay, non_neg_integer()}

batch_options()

@type batch_options() :: [batch_option()]

batch_result()

@type batch_result() :: %{
  success: [term()],
  errors: [{term(), term()}],
  total: non_neg_integer(),
  success_count: non_neg_integer(),
  error_count: non_neg_integer()
}

error_callback()

@type error_callback() :: (term(), term() -> any())

item_result()

@type item_result() :: {:ok, term()} | {:error, term()}

progress_callback()

@type progress_callback() :: (pos_integer(), pos_integer() -> any())

Functions

process_batch(agent, inputs, opts \\ [])

@spec process_batch(agent :: struct(), inputs :: [term()], batch_options()) ::
  {:ok, [item_result()] | batch_result()}

Process a batch of inputs through an agent concurrently.

Options

:max_concurrency (pos_integer/0) - Items processed at once. The default value is 10.
:timeout (timeout/0) - Timeout per item, in ms. An item that runs past it is stopped and fails with a :task_down Normandy.TurnError; the other items carry on. The default value is 300000.
:on_progress (function of arity 2) - Called in the caller after each item completes: (completed, total -> any).
:on_error (function of arity 2) - Called in the caller for each failed item: (input, error -> any).
:ordered (boolean/0) - true: one result per input, in input order. false: a batch_result map. The default value is true.

Returns

With ordered: true: {:ok, [{:ok, response} | {:error, error}]}, one per input, in input order.
With ordered: false: {:ok, batch_result}. errors pairs each error with its input, {input, error}; both lists are in completion order.

An error is what the other coordinators return, never a stacktrace: Normandy's own exception (Normandy.LLMError, ...), the exception a bug raised (logged with its stacktrace, arities only), or a :task_down Normandy.TurnError for an exit, a throw or an item stopped at :timeout.

Examples

# Simple batch
{:ok, [{:ok, first} | _]} = Processor.process_batch(agent, inputs)

# With configuration
{:ok, results} = Processor.process_batch(
  agent,
  inputs,
  max_concurrency: 5,
  on_progress: fn completed, total ->
    IO.puts("Progress: #{completed}/#{total}")
  end
)

process_batch_chunked(agent, inputs, opts \\ [])

@spec process_batch_chunked(agent :: struct(), inputs :: [term()], batch_options()) ::
  {:ok, [item_result()] | batch_result()}

Process a batch in chunks with a delay between chunks.

Useful for very large batches: :chunk_delay is a fixed pause between chunks, not a rate limiter.

Options

:max_concurrency (pos_integer/0) - Items processed at once. The default value is 10.
:timeout (timeout/0) - Timeout per item, in ms. An item that runs past it is stopped and fails with a :task_down Normandy.TurnError; the other items carry on. The default value is 300000.
:on_progress (function of arity 2) - Called in the caller after each item completes: (completed, total -> any).
:on_error (function of arity 2) - Called in the caller for each failed item: (input, error -> any).
:ordered (boolean/0) - true: one result per input, in input order. false: a batch_result map. The default value is true.
:chunk_size (pos_integer/0) - Items per chunk. The default value is 100.
:chunk_delay (non_neg_integer/0) - Milliseconds to wait between chunks. The default value is 0.

Returns the same shape as process_batch/3 for the same :ordered option, covering all chunks.

Example

{:ok, results} = Processor.process_batch_chunked(
  agent,
  large_input_list,
  chunk_size: 50,
  chunk_delay: 1000,  # 1 second between chunks
  max_concurrency: 5
)

process_batch_with_stats(agent, inputs, opts \\ [])

@spec process_batch_with_stats(agent :: struct(), inputs :: [term()], batch_options()) ::
  {:ok, batch_result()}

Process a batch and return detailed statistics.

Returns a map with separate success/error lists and counts.

Options

:max_concurrency (pos_integer/0) - Items processed at once. The default value is 10.
:timeout (timeout/0) - Timeout per item, in ms. An item that runs past it is stopped and fails with a :task_down Normandy.TurnError; the other items carry on. The default value is 300000.
:on_progress (function of arity 2) - Called in the caller after each item completes: (completed, total -> any).
:on_error (function of arity 2) - Called in the caller for each failed item: (input, error -> any).

Example

result = Processor.process_batch_with_stats(agent, inputs)
#=> %{
  success: [result1, result2],
  errors: [{input3, error}],
  total: 3,
  success_count: 2,
  error_count: 1
}