Batch Mode

Providers sell the same model twice. A live call answers in seconds at list price; the same request submitted to a Message Batches API answers within 24 hours at roughly half the token cost. For a nightly enrichment run, an overnight backfill, or any step whose result nobody is waiting on, that is a 50% discount for latency you weren't spending anyway.

Baton makes that a one-line change to a step:

use Baton.LLMStep, mode: :batch

Everything else about the step is unchanged — same request/1, same handle_response/3, same decoding and error taxonomy. Downstream steps, completion, retries, and cost stats behave exactly as they do for a live step. See the LLM step basics first if you haven't written one yet.

A complete example

Here is an ordinary live step:

defmodule MyApp.Steps.Classify do
  use Baton.LLMStep

  @impl true
  def output_schema do
    %{
      "type" => "object",
      "properties" => %{"category" => %{"type" => "string"}},
      "required" => ["category"]
    }
  end

  @impl true
  def request(%Oban.Job{} = job) do
    {:ok, %{"text" => text}} = Baton.Results.get_result(job, :extract)

    messages = [%{role: "user", content: "Classify this document:\n\n#{text}"}]
    {:ok, messages, model: "claude-sonnet-4-20250514", max_tokens: 1024}
  end

  @impl true
  def handle_response(%{"category" => category}, _context, _job) do
    {:ok, %{"category" => category}}
  end
end

And here it is on the batch transport. The diff is the use line:

defmodule MyApp.Steps.Classify do
  use Baton.LLMStep, mode: :batch

  # ... output_schema/0, request/1, and handle_response/3 unchanged ...
end

It goes into a workflow like any other step, and its dependents don't know the difference:

Baton.new(workflow_name: "nightly-enrichment")
|> Baton.add(:extract,  MyApp.Steps.Extract.new(%{doc_id: id}))
|> Baton.add(:classify, MyApp.Steps.Classify.new(%{}), deps: [:extract])
|> Baton.add(:store,    MyApp.Steps.Store.new(%{}),    deps: [:classify])
|> Baton.insert!()

:store waits for :classify exactly as it would for a live step — it just waits hours instead of seconds, then reads the same result map.

What the engine does

A batch step runs as several short Oban attempts rather than one long one:

1. Submit. request/1 builds the messages, the engine gates on the rate limiter, submits a one-request batch, records the batch id on the step's node, and snoozes.
2. Poll. Each later attempt reads the batch id and asks the provider whether the batch has ended. If not, it snoozes again — no dependency reads, no prompt rendering, nothing sent.
3. Ingest. Once the batch ends, the engine collects the result and runs it through the same pipeline a live response takes: decode/1 → handle_response/3 → attach usage.

The client contract

Batch mode needs three more functions from your :llm_client than the live path does. They're optional callbacks on Baton.LLMClient:

defmodule MyApp.LLM.Client do
  @behaviour Baton.LLMClient

  @impl true
  def complete(messages, opts) do
    # ... the live path you already have ...
  end

  @impl true
  def submit_batch(requests, opts) do
    # requests: [%{custom_id: "job-4171", messages: [...], opts: [...]}]
    # Return the provider's batch id.
    {:ok, "msgbatch_01ABC..."}
  end

  @impl true
  def poll_batch(batch_id, _opts) do
    # {:ok, :pending} keeps the step snoozing; {:ok, :ended} moves it on.
    # "Ended" means the provider is done, not that the request succeeded.
    {:ok, :ended}
  end

  @impl true
  def batch_results(batch_id, _opts) do
    {:ok,
     [
       %{
         custom_id: "job-4171",
         type: :succeeded,       # or :errored | :canceled | :expired
         response: %{text: "...", model: "...", stop_reason: "end_turn", usage: %{...}},
         error: nil
       }
     ]}
  end
end

submit_batch/2 takes a list even though Baton submits one request per step today — provider APIs are list-shaped, and a future coalescer that packs many steps into one batch will reuse the callback unchanged.

Baton matches its result out of the list by custom_id, which it sets to "job-<oban job id>": stable across attempts and short enough for any provider's limit.

If your client doesn't implement these, a batch step cancels on its first attempt with {:batch_unsupported, MyApp.LLM.Client} rather than retrying against a gap no retry can close.

That check is module-wide, so it can't see a failure that depends on the call: a model routed to a provider whose batch API you haven't implemented, a missing credential, an endpoint you can't speak. Those have no HTTP status to report and no retry that fixes them, so say so directly rather than inventing a status you never received:

{:error, {:cancel, {:batch_unsupported_provider, "openai"}}}

The taxonomy honours the verdict and cancels on the first attempt. Anything else keeps its normal retry budget.

Example: an OpenAI batch client

Nothing in Baton is provider-specific — it never speaks HTTP, and the three callbacks are all it knows. Anthropic's batch API takes requests inline; OpenAI's is file-based (upload a JSONL, create a batch from the file, download an output file), which is more work but fits the same contract. Here is the whole of it, using Req for HTTP:

defmodule MyApp.LLM.OpenAIBatch do
  @behaviour Baton.LLMClient

  @base "https://api.openai.com/v1"
  @endpoint "/v1/chat/completions"

  # Statuses that mean "still working". `cancelling` belongs here: it has not
  # settled yet, and polling again is how we find out that it did.
  @pending ~w(validating in_progress finalizing cancelling)

  @impl true
  def complete(messages, opts), do: MyApp.LLM.OpenAI.complete(messages, opts)

  @impl true
  def submit_batch(requests, _opts) do
    jsonl =
      Enum.map_join(requests, "\n", fn request ->
        Jason.encode!(%{
          custom_id: request.custom_id,
          method: "POST",
          url: @endpoint,
          body: %{
            model: request.opts[:model],
            max_tokens: request.opts[:max_tokens],
            messages: request.messages
          }
        })
      end)

    with {:ok, %{"id" => file_id}} <- upload(jsonl),
         {:ok, %{"id" => batch_id}} <- create_batch(file_id) do
      {:ok, batch_id}
    end
  end

  @impl true
  def poll_batch(batch_id, _opts) do
    case get_batch(batch_id) do
      {:ok, %{"status" => status}} when status in @pending -> {:ok, :pending}
      {:ok, %{"status" => _settled}} -> {:ok, :ended}
      {:error, _} = error -> error
    end
  end

  @impl true
  def batch_results(batch_id, _opts) do
    with {:ok, batch} <- get_batch(batch_id) do
      # Successes and failures come back in two separate files; Baton wants one
      # list, so read both. Either may be absent.
      results =
        [batch["output_file_id"], batch["error_file_id"]]
        |> Enum.reject(&is_nil/1)
        |> Enum.flat_map(&download_lines/1)
        |> Enum.map(&to_result/1)

      {:ok, results}
    end
  end

  # ── HTTP ──────────────────────────────────────────────────────────────────

  defp upload(jsonl) do
    Req.post("#{@base}/files",
      auth: {:bearer, api_key()},
      form_multipart: [purpose: "batch", file: {jsonl, filename: "batch.jsonl"}]
    )
    |> unwrap()
  end

  defp create_batch(file_id) do
    Req.post("#{@base}/batches",
      auth: {:bearer, api_key()},
      json: %{input_file_id: file_id, endpoint: @endpoint, completion_window: "24h"}
    )
    |> unwrap()
  end

  defp get_batch(batch_id) do
    Req.get("#{@base}/batches/#{batch_id}", auth: {:bearer, api_key()}) |> unwrap()
  end

  defp download_lines(file_id) do
    case Req.get("#{@base}/files/#{file_id}/content", auth: {:bearer, api_key()}) do
      {:ok, %{status: 200, body: body}} ->
        body |> String.split("\n", trim: true) |> Enum.map(&Jason.decode!/1)

      _ ->
        []
    end
  end

  # Failures are reported as `%{status: n}` so Baton's error taxonomy can read
  # them: a 429 while submitting becomes a snooze rather than a burned attempt.
  defp unwrap({:ok, %{status: 200, body: body}}), do: {:ok, body}
  defp unwrap({:ok, %{status: status, body: body}}), do: {:error, %{status: status, body: body}}
  defp unwrap({:error, reason}), do: {:error, reason}

  defp api_key, do: Application.fetch_env!(:my_app, :openai_api_key)

  # ── Shape mapping ─────────────────────────────────────────────────────────

  defp to_result(%{"custom_id" => id, "response" => %{"status_code" => 200, "body" => body}}) do
    %{custom_id: id, type: :succeeded, response: normalize(body), error: nil}
  end

  # Hand the failure back as `%{status: n}` — the shape `classify_error/1`
  # reads. That is what makes a 400 cancel the step (resubmitting the same
  # request would fail identically) while a 429 only snoozes it.
  defp to_result(%{"custom_id" => id, "response" => %{"status_code" => status} = response}) do
    %{
      custom_id: id,
      type: :errored,
      response: nil,
      error: %{status: status, body: response["body"]}
    }
  end

  defp to_result(%{"custom_id" => id} = line) do
    %{custom_id: id, type: :errored, response: nil, error: line["error"]}
  end

  defp normalize(%{"choices" => [choice | _], "model" => model, "usage" => usage}) do
    cached = get_in(usage, ["prompt_tokens_details", "cached_tokens"]) || 0

    %{
      text: get_in(choice, ["message", "content"]),
      model: model,
      stop_reason: stop_reason(choice["finish_reason"]),
      usage: %{
        # OpenAI's prompt_tokens *includes* cached tokens; the convention Baton
        # records and reconciles against is Anthropic's, where input_tokens
        # excludes them. Subtract, or every cache read is counted twice.
        input_tokens: usage["prompt_tokens"] - cached,
        output_tokens: usage["completion_tokens"],
        cache_read_input_tokens: cached
      }
    }
  end

  # "length" is OpenAI's truncation signal. It has to become "max_tokens" or
  # the check that makes a truncated reply retryable never fires.
  defp stop_reason("length"), do: "max_tokens"
  defp stop_reason(other), do: other
end

Three things in there are easy to get wrong and worth calling out:

• finish_reason: "length" must become stop_reason: "max_tokens". Baton matches that exact string to turn a truncated reply into a retryable error. Skip it and a half-finished response gets decoded as if it were complete.
• prompt_tokens double-counts cache reads relative to the convention Baton's stats and rate reconciliation use. Subtract cached_tokens out of it.
• A failed batch (input validation) produces no output file. Mapping it to {:ok, :ended}, as above, means Baton finds no result for its custom_id and fails the step with :batch_result_missing — correct, if terse. Return {:error, ...} from poll_batch/2 instead if you would rather the step retry, but note a retry resubmits the same request, which will fail the same way.

The first two apply equally to a live OpenAI client, so if you already have a working complete/2 you likely have them solved and can share the mapping.

Tuning

use Baton.LLMStep, mode: :batch, poll_interval: 600

The poll interval trades wasted wake-ups against how long a finished batch sits unnoticed; five minutes is negligible against a multi-hour turnaround. The deadline is a backstop above the provider's own 24-hour expiry — reaching it means something is wrong with the handle, not the batch.

In a portable flow

use Baton.LLMStep, mode: :batch fixes the transport when the module compiles, which is right for a step module that exists to do one thing. A portable flow node can't work that way: every llm node in every definition runs through the single Baton.Flow.Workers.LLM. So a flow node names its transport in config instead, and the worker reads it on each attempt:

%Baton.Flow.NodeSpec{
  id: "classify",
  type: "llm",
  config: %{
    "model" => "claude-sonnet-4-20250514",
    "user_prompt" => %{"body" => "..."},
    "transport" => "batch",
    "poll_interval" => 600
  }
}

"transport" is "live" (the default) or "batch"; "poll_interval" and "batch_deadline" are the same seconds-valued options as above, and anything you leave out falls back to the engine's defaults. Nothing else about the node changes — prompts, bindings, schemas, adapters, and the results downstream nodes read are all identical.

Baton.Flow.Validator checks this before a definition compiles, because every way of getting it wrong fails silently otherwise — an unrecognized transport would just run live, and you'd find out from the bill. It rejects:

That last one is a judgment call worth explaining. The sequential gate chains expanded nodes so each waits for the one before it. Batched, that is N waits of up to 24 hours apiece — and the gate's whole purpose, priming a prompt cache whose TTL is measured in minutes, cannot survive a gap that long. There is no configuration where the pair does what its author intended, so it's an error rather than a footgun. Use gate: "parallel".

One definition, either transport

Hardcoding "transport" => "batch" in a definition fixes its latency class for every run. Often that isn't what you want: the same analysis should answer an analyst in minutes when they click run, and cost half as much overnight when a scheduler runs it across a thousand inputs. Compile with a per-run default instead:

# Interactive: no override, every node defaults live.
Compiler.compile(definition, input: input)

# Bulk overnight: same definition, batched, with deadline headroom for
# whatever queueing sits between the step and the provider.
Compiler.compile(definition, input: input, transport: "batch", batch_deadline: 93_600)

The :transport option fills the default for every llm node that doesn't declare one; a node's own explicit "transport" always wins, which is how a flow pins a cheap synthesis step live even in a batch run. :poll_interval and :batch_deadline fill tuning defaults the same way, only on nodes that end up batched — tuning belongs to the run, not the definition, because only the caller knows what sits between submission and the provider.

The override merges before validation, so it cannot smuggle batch past the topology rules: compiling a sequential fan-out with transport: "batch" fails with :sequential_fan_out right there, while the same definition compiles live untouched. And both the flow snapshot and each job's args carry the merged config — what ran is what is recorded.

Cost tracking

Batch usage flows into workflow_step_stats like any other call, with one addition: the usage map carries service_tier: "batch" so your pricing module can apply the discount. Nothing downstream can recover that on its own — a stored cost looks identical either way — so branch on it:

defmodule MyApp.Pricing do
  @behaviour Baton.Pricing

  @impl true
  def cost(%{service_tier: "batch"} = usage) do
    usage |> standard_cost() |> Decimal.mult(Decimal.new("0.5"))
  end

  def cost(usage), do: standard_cost(usage)
end

Live calls carry no :service_tier key at all.

latency_ms for a batch step is the end-to-end turnaround — from submission to result, not the duration of any one HTTP call — which is the number worth comparing against a live twin when you're deciding what else to move over.

Watching a step that waits for hours

A batch step broadcasts an awaiting event on submit and on every poll, carrying the batch id and the poll interval it parked itself for:

def handle_info({:workflow_step_updated, %{state: "awaiting"} = payload}, socket) do
  # payload.detail == %{batch_id: "msgbatch_01ABC...", seconds: 300}
  {:noreply, mark_waiting(socket, payload)}
end

This is distinct from snoozed, which means a step is waiting for a resource of its own — its detail.reason says which ("deps", "rate_budget", "provider_limit", "batch_slot", or "step"). The batch wait itself is never double-reported: an awaiting step does not also broadcast snoozed. If you pattern-match on state anywhere, keep a catch-all clause — new states get added as the engine grows ways to wait.

Rate-limiting submissions

Provider batch traffic draws from a separate pool from live traffic, so Baton.RateLimiter's acquire/3 (which reserves input/output token budget) doesn't apply. What can still be exceeded is the rate of submissions. Implement the optional acquire_batch/1 if you need to pace them:

@impl true
def acquire_batch(opts) do
  if slots_available?(), do: {:ok, opts}, else: {:snooze, 120}
end

Polling is deliberately ungated — it's cheap, and asking a limiter about it would only add latency to a step that is already waiting.

When not to use it

• Anything a user is waiting on. Hours is hours.
• Chained batch steps. Two in a row can take 2× the latency; the DAG doesn't stop you, but prefer batch mode for tails or where the downstream work is cheap and local.
• Steps that already coalesce. If you have a pipeline that batches many items into one provider request itself, it's already getting the discount at better packing than one-request-per-step.

Requirements

Batch mode needs schema v7 for the workflow_nodes.checkpoint column, which is where a step keeps its batch id between attempts:

defmodule MyApp.Repo.Migrations.UpgradeBatonV7 do
  use Ecto.Migration
  def up,   do: Baton.Migration.up()
  def down, do: Baton.Migration.down(version: 6)
end

The column is only written by steps that use it; nothing else changes.