LangExtract.Runner (LangExtract v0.7.0)

A caller-owned, supervised extraction runner with a shared request budget.

Place it in your supervision tree — no global names, no app env, no library-side singleton; multiple independent runners coexist:

# application.ex
{LangExtract.Runner,
 name: MyApp.Extractor,
 client: LangExtract.new(:claude, api_key: key),
 max_in_flight: 20,
 rpm: 2_000,
 chunk_retries: 3}

All extraction scheduled through the runner shares one budget: concurrent callers cannot jointly exceed :rpm or :max_in_flight, and a single 429 pauses every in-flight chunk until the server's retry-after deadline (see LangExtract.Runner.Limiter).

Requests made through the runner disable Req's transient retry — the runner owns the retry policy (see LangExtract.Runner.Request). Standalone LangExtract.run/4 and stream/4 keep Req's retry exactly as before.

Options

:client (required) — the LangExtract.Client to extract with
:name — registered name for the runner
:max_in_flight — concurrent request cap (default 10)
:rpm — requests-per-minute budget (default :infinity)
:chunk_retries — retry budget per chunk for 5xx/transport failures (default 3); 429 waits never consume it
:rate_limit_retries — cap on 429 retries per chunk (default 10); past it the chunk fails with the rate-limit error
:retry_backoff_ms — base backoff for consumed retries (default 200)
:buffer — bound on undelivered stream results (default: max_in_flight); a slow consumer halts admission at this bound
:drain_timeout — grace period in ms for in-flight requests to finish when the runner shuts down (default 5_000); chunks never started are reported as %ChunkError{reason: :drained}

Summary

Types

option()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

run(runner, source, template, opts \\ [])

Runs a full extraction through the runner's shared budget.

start_link(opts)

Starts the runner supervisor. See the module docs for options.

stream(runner, source, template, opts \\ [])

Streams per-chunk results through the runner's shared budget.

stream_corpus(runner, docs, template, opts \\ [])

Streams a whole corpus through the runner's shared budget.

Types

option()

@type option() ::
  {:client, LangExtract.Client.t()}
  | {:name, GenServer.name()}
  | {:max_in_flight, pos_integer()}
  | {:rpm, pos_integer() | :infinity}
  | {:chunk_retries, non_neg_integer()}
  | {:rate_limit_retries, non_neg_integer()}
  | {:retry_backoff_ms, pos_integer()}
  | {:buffer, pos_integer()}
  | {:drain_timeout, non_neg_integer()}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

run(runner, source, template, opts \\ [])

@spec run(Supervisor.supervisor(), String.t(), LangExtract.Template.t(), keyword()) ::
  {:ok,
   {[LangExtract.Alignment.Span.t()], [LangExtract.Pipeline.ChunkError.t()]}}

Runs a full extraction through the runner's shared budget.

Collects stream/4 and restores document order. Always returns {:ok, {spans, chunk_errors}}: in runner mode every failure is per-chunk (a crashed or timed-out chunk task lands in chunk_errors with reason {:task_exit, reason}), so there is no abandon-the-document error path.

start_link(opts)

@spec start_link([option()]) :: Supervisor.on_start()

Starts the runner supervisor. See the module docs for options.

stream(runner, source, template, opts \\ [])

@spec stream(Supervisor.supervisor(), String.t(), LangExtract.Template.t(), keyword()) ::
  Enumerable.t()

Streams per-chunk results through the runner's shared budget.

Same event shape as LangExtract.stream/4 — {:ok, %ChunkResult{}} and {:error, %ChunkError{}} in completion order — but chunk requests are scheduled through the runner's Limiter (rpm + in-flight budget, global 429 backoff) and retried per the runner's policy. Task-level failures stay per-chunk. Delivery is bounded: at most :buffer results are outstanding, so a slow consumer throttles admission.

Accepts run/4's chunking and alignment options plus :buffer (default: the runner's configured buffer).

stream_corpus(runner, docs, template, opts \\ [])

@spec stream_corpus(
  Supervisor.supervisor(),
  Enumerable.t(),
  LangExtract.Template.t(),
  keyword()
) ::
  Enumerable.t()

Streams a whole corpus through the runner's shared budget.

Takes an enumerable of {id, source} pairs and yields {id, event} in the same event shape as stream/4. Documents are processed in order, each with its own document telemetry span; chunk-level concurrency within a document follows the runner's budget and buffer. Lazy — a document's extraction starts only when the stream reaches it.