Imp.Clients.ReqLLMBatch (Imp v0.12.0)

Copy Markdown View Source

Provider-neutral, bounded, resumable execution for durable LM batches.

Requests have stable caller-supplied IDs and JSON-safe payloads. A dispatcher receives each normalized request plus attempt metadata and must return one of:

  • {:ok, output}
  • {:transient, reason} - the request did not run and may be sent again
  • {:terminal, reason}
  • {:malformed, reason}
  • {:ambiguous, reason} - the request may have run; it is not sent again

A request that may already have run is never re-dispatched: sending it again could run and bill it twice. Only a :transient outcome is retried, up to :max_attempts. A dispatcher that raises, throws or exits, or that is still running at :timeout, may have sent its request first, so its outcome is :ambiguous.

The batch owns the wait before a retry as well as the retry. When the failure carries the provider's retry-after header (seconds or an HTTP date), the request is not sent again before then; otherwise the wait grows exponentially with the attempt, with jitter. A request that is waiting holds no dispatch slot: other requests are sent meanwhile, and the batch sleeps only when every remaining request is waiting. A process that traps exits and is stopped by its parent during that sleep exits at once with the parent's reason; a dispatch wave that is running is still waited for, for up to :timeout. The parent is the first of the process's $ancestors, which a process started by Task, GenServer or another OTP behaviour has; a process started with bare spawn/1 or spawn_link/1 has none, and its sleep cannot be interrupted. Backoff is capped at :max_retry_wait. When the provider asks for longer than :max_retry_wait, or the wait would end after the Imp.Deadline bound to the calling process, the request is not retried in this run: it stays :transient_failure with the attempts it has used, the summary is not complete?, and resume/3 retries it. The time a request may be sent again is kept in the checkpoint as a UTC time, and resume/3 waits for it under the same rules.

The checkpoint is an auditable JSON document. Every dispatch intent and outcome is appended to its event history before execution continues. Writes use a synced temporary file followed by an atomic rename. On resume, a request that was dispatched but has no committed outcome becomes :ambiguous for the same reason.

A checkpoint written by Imp 0.5.0 (schema version 1) retried timeouts and dispatcher crashes as transient failures, and its stored reason does not always say which a failure was. On resume, its :transient_failure requests become :ambiguous and are not sent again; the checkpoint is rewritten at schema version 2.

Summary

Functions

Builds a dispatcher backed by an existing ReqLLM client.

Resumes a batch from its checkpoint.

Starts a new durable batch.

Types

context()

@type context() :: %{request_id: String.t(), attempt: pos_integer()}

dispatcher()

@type dispatcher() :: (request(), context() -> outcome())

outcome()

@type outcome() ::
  {:ok, term()}
  | {:transient, term()}
  | {:terminal, term()}
  | {:malformed, term()}
  | {:ambiguous, term()}

request()

@type request() :: %{id: String.t(), payload: term()}

summary()

@type summary() :: %{
  checkpoint: Path.t(),
  complete?: boolean(),
  counts: %{required(atom()) => non_neg_integer()},
  requests: [map()]
}

Functions

req_llm_dispatcher(client, call_opts \\ [])

@spec req_llm_dispatcher(Imp.Clients.ReqLLM.t(), keyword()) :: dispatcher()

Builds a dispatcher backed by an existing ReqLLM client.

Request payloads may be a message list or %{"messages" => messages}. The adapter is provider-neutral because provider and model selection remain in the client ("openai:...", "anthropic:...", or "gemini:...").

The batch owns retries: every call is made with max_retries: 0, so ReqLLM sends each attempt once and the error describes the only send.

A failed call's Imp.LMError decides its outcome:

  • :transient - the request never reached the provider (the connection was refused, or no pooled connection was free), or the provider answered that it did not process it and to try later (408, 425, 429, 503, 529).
  • :terminal - the provider rejected it with any other 4xx status, or the call failed with no status and no transport reason.
  • :ambiguous - it may have run: a 5xx status other than 503 or 529 (500 included), any other status, or a timeout or closed connection with no response.

resume(checkpoint, dispatcher, opts \\ [])

@spec resume(Path.t(), dispatcher(), keyword()) :: {:ok, summary()} | {:error, term()}

Resumes a batch from its checkpoint.

Persisted requests, attempts, and retry policy are authoritative. Resume accepts only runtime options: :num_threads, :timeout, :validate_output, :max_retry_wait and :clock. A request waiting for a retry, or whose retry was stopped, is not sent before the not_before time its checkpoint holds: the batch waits for it when that is within :max_retry_wait and the deadline, and stops its retry again, unsent, when it is not.

run(requests, dispatcher, opts)

@spec run([map()], dispatcher(), keyword()) :: {:ok, summary()} | {:error, term()}

Starts a new durable batch.

Required options:

  • :checkpoint - destination for the atomic JSON checkpoint

Runtime options:

  • :num_threads - requests dispatched at once (default 4)
  • :max_attempts - sends per request, retries included (default 3)
  • :timeout - milliseconds one dispatch may run (default 30_000)
  • :validate_output - an optional arity-one callback returning :ok or {:error, reason}
  • :max_retry_wait - the longest wait, in milliseconds, before a retry (default 60_000)
  • :clock - the time source and sleep the retry wait uses, a map of :now (monotonic milliseconds), :utc_now (a DateTime) and :sleep (milliseconds) functions; it exists so tests need not wait