From the blog

OpenAI Batch API in Elixir: Half-Price Bulk AI Jobs with Oban

By Liam Killingback ·

OpenAI Batch API in Elixir: Half-Price Bulk AI Jobs with Oban

Most AI features in a Phoenix app are interactive: a user asks, the model answers, LiveView streams it back. But a surprising amount of AI work is not interactive at all. Tagging ten thousand support tickets. Re-summarising every document after you change the prompt. Generating embeddings for a backlog of PDFs. Classifying last month’s uploads for a report nobody will read until Monday.

For that kind of work, calling the normal API in a loop is the expensive way to do it. You pay full price, you fight rate limits, and a crash halfway through leaves you guessing which rows were done. The OpenAI Batch API exists for exactly this case: you upload a file of requests, OpenAI runs them within 24 hours, and you pay roughly half the normal price.

This guide shows how to use the OpenAI Batch API in Elixir with Req and Oban: building the JSONL file, uploading it, creating the batch, polling it without blocking anything, and writing the results back to Postgres by custom_id. The code is plain Elixir you can drop into any Phoenix app.

When the Batch API is the right tool (and when it is not)

The trade is simple. You give up latency and get price and throughput in return.

What you get:

  • About 50% off the standard per-token price for the same model, according to OpenAI’s pricing page at the time of writing.
  • A separate rate limit pool. Batch requests do not eat into the per-minute limits your interactive features depend on, so a big backfill will not start throwing 429s at your chat users.
  • One file, one job. Up to 50,000 requests per batch (and an input file of up to 200 MB), tracked as a single object with request counts.

What you give up:

  • No latency promise beyond the window. The only completion window is 24h. Many batches finish much sooner, but you cannot rely on it.
  • No streaming and no conversation. Each line is a self-contained request.
  • More moving parts. Files, statuses, an output file and an error file, instead of one HTTP response.

A good rule: if a human is waiting on the answer, use the normal API (our post on streaming OpenAI responses in Phoenix LiveView covers that). If the answer is going into a table that someone will look at later, batch it.

How a batch works

The lifecycle has four steps:

  1. Write a JSONL file. One JSON object per line, each with a custom_id, an HTTP method, a url (the endpoint, such as /v1/chat/completions) and a body (exactly what you would have sent to that endpoint).
  2. Upload it to /v1/files with purpose=batch.
  3. Create the batch at /v1/batches, pointing at the uploaded file id.
  4. Poll /v1/batches/:id until the status is terminal, then download the output file (and the error file, if there is one).

The statuses you will see are validating, in_progress, finalizing and completed on the happy path, plus failed, expired, cancelling and cancelled. Treat completed, failed, expired and cancelled as terminal.

Two details matter more than they look:

  • Output order is not guaranteed. Results come back keyed by your custom_id, not by line number. Make custom_id something you can map straight back to a row, such as "doc:4821".
  • An expired batch still returns what finished. If the window runs out, completed requests are in the output file and billed, and the rest are reported as expired. Your code should handle partial results rather than assuming all or nothing.

The data model

You want a durable record of every batch you submit, so a deploy or a crash never loses track of one. A small table is enough:

defmodule MyApp.Repo.Migrations.CreateAiBatches do
  use Ecto.Migration

  def change do
    create table(:ai_batches) do
      add :purpose, :string, null: false
      add :openai_batch_id, :string
      add :input_file_id, :string
      add :status, :string, null: false, default: "pending"
      add :request_count, :integer, null: false, default: 0
      add :completed_count, :integer, null: false, default: 0
      add :failed_count, :integer, null: false, default: 0
      add :error, :text
      timestamps()
    end

    create unique_index(:ai_batches, [:openai_batch_id])

    alter table(:documents) do
      add :summary, :text
      add :summary_batch_id, references(:ai_batches)
    end
  end
end
defmodule MyApp.AI.Batch do
  use Ecto.Schema
  import Ecto.Changeset

  schema "ai_batches" do
    field :purpose, :string
    field :openai_batch_id, :string
    field :input_file_id, :string
    field :status, :string, default: "pending"
    field :request_count, :integer, default: 0
    field :completed_count, :integer, default: 0
    field :failed_count, :integer, default: 0
    field :error, :string
    timestamps()
  end

  def changeset(batch, attrs) do
    batch
    |> cast(attrs, [:purpose, :openai_batch_id, :input_file_id, :status,
                    :request_count, :completed_count, :failed_count, :error])
    |> validate_required([:purpose, :status])
  end
end

The example job here summarises documents, but the shape is the same for classification, extraction or embeddings.

A small Batch API client with Req

Keep the HTTP calls in one module. Req handles JSON for you, and its :form_multipart option covers the file upload.

defmodule MyApp.OpenAIBatch do
  @base "https://api.openai.com/v1"

  defp client do
    Req.new(
      base_url: @base,
      auth: {:bearer, Application.fetch_env!(:my_app, :openai_api_key)},
      receive_timeout: 60_000,
      retry: :transient
    )
  end

  @doc "Uploads a JSONL string and returns the file id."
  def upload_jsonl(jsonl) when is_binary(jsonl) do
    case Req.post(client(),
           url: "/files",
           form_multipart: [
             purpose: "batch",
             file: {jsonl, filename: "batch.jsonl", content_type: "application/jsonl"}
           ]
         ) do
      {:ok, %{status: 200, body: %{"id" => id}}} -> {:ok, id}
      {:ok, resp} -> {:error, {:upload_failed, resp.status, resp.body}}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc "Creates a batch against an uploaded file."
  def create_batch(file_id, endpoint, metadata \\ %{}) do
    body = %{
      input_file_id: file_id,
      endpoint: endpoint,
      completion_window: "24h",
      metadata: metadata
    }

    case Req.post(client(), url: "/batches", json: body) do
      {:ok, %{status: 200, body: batch}} -> {:ok, batch}
      {:ok, resp} -> {:error, {:create_failed, resp.status, resp.body}}
      {:error, reason} -> {:error, reason}
    end
  end

  def get_batch(batch_id) do
    case Req.get(client(), url: "/batches/#{batch_id}") do
      {:ok, %{status: 200, body: batch}} -> {:ok, batch}
      {:ok, resp} -> {:error, {:get_failed, resp.status, resp.body}}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc "Downloads a result file and decodes each JSONL line."
  def file_lines(file_id) do
    case Req.get(client(), url: "/files/#{file_id}/content", decode_body: false) do
      {:ok, %{status: 200, body: body}} ->
        lines =
          body
          |> String.split("\n", trim: true)
          |> Enum.map(&Jason.decode!/1)

        {:ok, lines}

      {:ok, resp} ->
        {:error, {:download_failed, resp.status}}

      {:error, reason} ->
        {:error, reason}
    end
  end

  def cancel(batch_id), do: Req.post(client(), url: "/batches/#{batch_id}/cancel")
end

Note decode_body: false on the download. The result file is JSONL, not one JSON document, so you want the raw text and decode it line by line.

For a very large result file you may prefer to stream it to disk with Req’s into: option and decode it lazily with File.stream!/1. For a few thousand summaries, holding it in memory is fine.

Building the JSONL file

Each line is a full request. The body is exactly what you would send to Chat Completions, which means everything you already know about prompts and structured output still applies.

defmodule MyApp.AI.SummaryBatch do
  import Ecto.Query
  alias MyApp.{Repo, Document}

  @model "gpt-4o-mini"

  def documents_to_summarise(limit) do
    from(d in Document,
      where: is_nil(d.summary) and is_nil(d.summary_batch_id),
      order_by: d.id,
      limit: ^limit
    )
    |> Repo.all()
  end

  def to_jsonl(documents) do
    documents
    |> Enum.map(&request_line/1)
    |> Enum.map_join("\n", &Jason.encode!/1)
  end

  defp request_line(doc) do
    %{
      custom_id: "doc:#{doc.id}",
      method: "POST",
      url: "/v1/chat/completions",
      body: %{
        model: @model,
        max_tokens: 300,
        messages: [
          %{role: "system", content: "Summarise the document in three plain sentences."},
          %{role: "user", content: String.slice(doc.text, 0, 20_000)}
        ]
      }
    }
  end
end

A few practical rules for the file:

  • Keep one model per file. Batches are processed per model, so mixing models in one file is asking for a validation failure.
  • Truncate inputs yourself. A request that exceeds the model’s context window fails on its own line, and you will only find out when the error file arrives. If your documents are long, chunk them first (our guide to AI document summarization in Elixir covers map-reduce summaries).
  • Make custom_id unique within the file. It is your only key for joining results back.

Submitting the batch with Oban

Submission is a job, not a controller action. It claims a set of rows, uploads the file, creates the batch, records it, and schedules the first poll.

defmodule MyApp.Workers.SubmitSummaryBatch do
  use Oban.Worker, queue: :ai_batch, max_attempts: 5

  import Ecto.Query
  alias MyApp.{Repo, Document, OpenAIBatch}
  alias MyApp.AI.{Batch, SummaryBatch}

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"limit" => limit}}) do
    case SummaryBatch.documents_to_summarise(limit) do
      [] ->
        :ok

      docs ->
        {:ok, batch} =
          %Batch{}
          |> Batch.changeset(%{purpose: "summaries", request_count: length(docs)})
          |> Repo.insert()

        ids = Enum.map(docs, & &1.id)

        # Claim the rows first so a second run cannot submit them again.
        from(d in Document, where: d.id in ^ids)
        |> Repo.update_all(set: [summary_batch_id: batch.id])

        with {:ok, file_id} <- OpenAIBatch.upload_jsonl(SummaryBatch.to_jsonl(docs)),
             {:ok, remote} <-
               OpenAIBatch.create_batch(file_id, "/v1/chat/completions", %{
                 "local_batch_id" => to_string(batch.id)
               }) do
          batch
          |> Batch.changeset(%{
            input_file_id: file_id,
            openai_batch_id: remote["id"],
            status: remote["status"]
          })
          |> Repo.update!()

          %{batch_id: batch.id}
          |> MyApp.Workers.PollSummaryBatch.new(schedule_in: 60)
          |> Oban.insert()

          :ok
        else
          {:error, reason} ->
            release(batch, ids, inspect(reason))
            {:error, reason}
        end
    end
  end

  defp release(batch, ids, error) do
    from(d in Document, where: d.id in ^ids)
    |> Repo.update_all(set: [summary_batch_id: nil])

    batch |> Batch.changeset(%{status: "submit_failed", error: error}) |> Repo.update!()
  end
end

Claiming the rows before the upload is the important part. Without it, a nightly cron and a manual “run now” button can both pick up the same documents and you pay for them twice. On failure the claim is released so the next run picks them up again.

Schedule it with Oban’s cron plugin, for example nightly:

config :my_app, Oban,
  queues: [default: 10, ai_batch: 2],
  plugins: [
    {Oban.Plugins.Cron,
     crontab: [{"0 2 * * *", MyApp.Workers.SubmitSummaryBatch, args: %{limit: 5_000}}]}
  ]

Polling without blocking: Oban snooze

The worst way to wait for a batch is a process that sleeps in a loop. The best way is a job that checks once and asks Oban to run it again later. Returning {:snooze, seconds} does exactly that, and it does not count as a failed attempt.

defmodule MyApp.Workers.PollSummaryBatch do
  use Oban.Worker,
    queue: :ai_batch,
    max_attempts: 10,
    unique: [period: 300, keys: [:batch_id]]

  alias MyApp.{Repo, OpenAIBatch}
  alias MyApp.AI.{Batch, SummaryResults}

  @terminal ~w(completed failed expired cancelled)

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"batch_id" => id}}) do
    batch = Repo.get!(Batch, id)

    with {:ok, remote} <- OpenAIBatch.get_batch(batch.openai_batch_id) do
      counts = remote["request_counts"] || %{}

      batch =
        batch
        |> Batch.changeset(%{
          status: remote["status"],
          completed_count: counts["completed"] || 0,
          failed_count: counts["failed"] || 0
        })
        |> Repo.update!()

      if remote["status"] in @terminal do
        SummaryResults.apply(batch, remote)
      else
        {:snooze, backoff(remote["status"])}
      end
    end
  end

  # Check often while it is validating, less often while it runs.
  defp backoff("validating"), do: 30
  defp backoff("finalizing"), do: 60
  defp backoff(_), do: 300
end

The unique option stops a second poller for the same batch from being inserted while one is already scheduled. Because every state change is written to ai_batches, a LiveView admin page can show progress by reading that table, or you can broadcast on Phoenix.PubSub after each update if you want it live.

Writing results back by custom_id

Each output line looks like this (trimmed):

{"id": "batch_req_...", "custom_id": "doc:4821",
 "response": {"status_code": 200, "body": {"choices": [{"message": {"content": "..."}}], "usage": {...}}},
 "error": null}

Parse the custom_id, check the status code, and update the row. Lines in the error file have the same custom_id with an error object instead.

defmodule MyApp.AI.SummaryResults do
  import Ecto.Query
  alias MyApp.{Repo, Document, OpenAIBatch}

  def apply(batch, remote) do
    with :ok <- apply_output(remote["output_file_id"]),
         :ok <- release_errors(remote["error_file_id"]) do
      release_unfinished(batch)
      :ok
    end
  end

  defp apply_output(nil), do: :ok

  defp apply_output(file_id) do
    with {:ok, lines} <- OpenAIBatch.file_lines(file_id) do
      Repo.transaction(fn ->
        Enum.each(lines, fn line ->
          with "doc:" <> id <- line["custom_id"],
               %{"status_code" => 200, "body" => body} <- line["response"],
               [%{"message" => %{"content" => text}} | _] <- body["choices"] do
            from(d in Document, where: d.id == ^String.to_integer(id))
            |> Repo.update_all(set: [summary: String.trim(text)])
          end
        end)
      end)

      :ok
    end
  end

  defp release_errors(nil), do: :ok

  defp release_errors(file_id) do
    with {:ok, lines} <- OpenAIBatch.file_lines(file_id) do
      ids = for %{"custom_id" => "doc:" <> id} <- lines, do: String.to_integer(id)
      from(d in Document, where: d.id in ^ids) |> Repo.update_all(set: [summary_batch_id: nil])
      :ok
    end
  end

  # Anything still unsummarised after a terminal status goes back in the queue.
  defp release_unfinished(batch) do
    from(d in Document, where: d.summary_batch_id == ^batch.id and is_nil(d.summary))
    |> Repo.update_all(set: [summary_batch_id: nil])
  end
end

Releasing failed and unfinished rows instead of marking them done is what makes the whole pipeline self-healing. An expired batch, a context-length error on one line, or a cancelled run all end the same way: the rows that did not get a summary are picked up by the next nightly submission. Writing the summaries is idempotent too, so if the poller crashes after applying results and runs again, it simply writes the same text a second time.

If a particular row fails repeatedly, add an attempt counter to the documents table and stop resubmitting after three tries. A prompt that cannot succeed will otherwise cost you a little every night forever.

Costs, usage and who pays for it

Every successful response body includes a usage object with prompt and completion tokens, the same as the synchronous API. Store it. It is how you check the batch discount against your actual bill, and it is how you find the one prompt template that quietly costs three times as much as the others.

If the batch work is done for a specific customer (say, an “analyse my whole archive” button on a paid plan), the token counts are also what you would bill or limit them on. That is a metering problem rather than an AI one, and Aurora Meter is the free MIT library we built for exactly that: counting usage per customer in Phoenix, gating it against plan limits and reporting it to Stripe.

Testing without calling OpenAI

Req makes this straightforward with Req.Test. Add a plug: option to the client in test, then stub each endpoint:

# in MyApp.OpenAIBatch.client/0, merge in Application.get_env(:my_app, :openai_req_options, [])
# config/test.exs
config :my_app, :openai_req_options, plug: {Req.Test, MyApp.OpenAIBatch}
test "applies completed summaries by custom_id" do
  batch = Repo.insert!(%MyApp.AI.Batch{purpose: "summaries", status: "completed"})
  doc = insert(:document, text: "Long text", summary_batch_id: batch.id)

  output =
    Jason.encode!(%{
      custom_id: "doc:#{doc.id}",
      response: %{status_code: 200, body: %{choices: [%{message: %{content: "Short."}}]}},
      error: nil
    })

  Req.Test.stub(MyApp.OpenAIBatch, fn conn ->
    Plug.Conn.send_resp(conn, 200, output)
  end)

  assert :ok = MyApp.AI.SummaryResults.apply(batch, %{"output_file_id" => "file_1"})
  assert Repo.reload!(doc).summary == "Short."
end

Test the unhappy paths too: an error-file line releases its row, and an expired batch with no output file releases everything.

A production checklist

  • Claim rows before submitting, release them on failure.
  • Persist the OpenAI batch id the moment you have it.
  • Poll with {:snooze, n}, never with Process.sleep/1.
  • Treat expired as partial success, not failure.
  • Join results on custom_id, never on line order.
  • Store token usage per request.
  • Cap retries per row so a broken prompt does not run nightly forever.
  • Keep interactive features on the synchronous API. Batch is for work nobody is waiting on.

Skip the plumbing

If your app is built around documents (upload, extract, summarise, search), phx_ai_document is our Phoenix starter for exactly that, with LiveView uploads, Oban pipelines and the OpenAI integration already wired. Adding a nightly Batch API backfill to it is the code in this post and not much more.

Summary

The OpenAI Batch API in Elixir comes down to four calls (upload, create, poll, download) and one discipline: treat every batch as a durable job whose rows you claim, track and release. Oban gives you scheduling, polling via snooze, and retries for free, and Postgres gives you the record of what was sent. In exchange for waiting up to a day, you pay about half price and keep your interactive rate limits clear for the users who are actually waiting. For more on handling the synchronous side, see our guide to OpenAI rate limits in Elixir.

If you are building more than one AI product on Phoenix, the Builder Pass gives you every PhxTemplates starter, current and future, for one price.

Back to blogs