v1.1: opt-in cross-request batching in serve.py (--batch-window-ms, off by default); weights unchanged

README.md

@@ -212,6 +212,21 @@ docker build -t blink-4b . && docker run --rm --gpus all -p 127.0.0.1:8000:8000
 - Requests run one at a time. Questions are batched; each batch takes one forward pass (large requests
   can take more than one). Serving the downloaded folder or Docker image enables Hugging Face offline
   mode before model loading (`hub_offline: true`). The server doesn't otherwise restrict network access.
+- Set `--batch-window-ms 5` to turn on cross-request batching with a 5 ms collection window, up to
+  `--max-batch-requests` requests at a time, which defaults to 16. The window defaults to 0, so requests still
+  run one at a time. `--max-queued-requests` lets up to 64 requests wait for a batch by default. Excess requests
+  get HTTP 503 with `Retry-After`, so clients should retry. On a 1,000-request Decision Index sample over HTTP,
+  throughput rose about 20% with 4 concurrent clients and 24% with 16. One client saw no gain. Offline runs on
+  long documents showed no meaningful gain. On the Decision Index sample, a set of long workflow documents, and
+  the public TypeSafe cases, with each document sent as a JSON object, batched answers passed the same
+  numerical-parity checks against an FP32 reference as one-at-a-time answers, covering argmax agreement and
+  probability differences. A few near-tied answers can still flip. `v1.1` changes code only, leaving `v1.0`
+  weights unchanged. Update the two code files in an existing `v1.0` download, then restart with the flag:
+
+  ```sh
+  hf download thegovind/blink-4b serve.py blink.py --revision v1.1 --local-dir blink-4b
+  python blink-4b/serve.py --model ./blink-4b --port 8000 --batch-window-ms 5
+  ```
 - blink-4b weights are 8.4 GB in bf16. Long prompts need more memory.
 
 </details>

blink.py

@@ -56,6 +56,29 @@ MAX_OPTIONS = 255
 MAX_QUESTIONS = 512
 MAX_INPUT_TOKENS = int(os.environ.get("BLINK_MAX_INPUT_TOKENS", "131072"))  # as evaluated
 TOKEN_BUDGET = int(os.environ.get("BLINK_TOKEN_BUDGET", "32768"))  # padded tokens per forward
+# Shared-prefix prefill (opt-in, BLINK_PREFIX_CACHE=1): the questions of one request share the system prompt and the
+# evidence, so that prefix is encoded once and each question runs only its own tail against a copy of the cached
+# prefix. Same tokens, same positions; only floating-point order differs. Measured 4.8-6.9x less model time on long
+# multi-question documents, but on ~18k-token documents blink-4b's bf16 answers drift slightly further from an FP32
+# reference than the plain path's do, so it is off by default (experiments/t5/PREREG.md, gate A').
+PREFIX_CACHE = os.environ.get("BLINK_PREFIX_CACHE", "0").lower() in ("1", "true", "on", "yes")
+
+
+def _prefix_setting(name: str, default: int) -> int:
+    """A shared-prefix setting, read only when BLINK_PREFIX_CACHE is on: a stray value can't break the default path."""
+    if not PREFIX_CACHE:
+        return default
+    value = int(os.environ.get(name, str(default)))
+    if value < 1:
+        raise ValueError(f"{name} must be at least 1")
+    return value
+
+
+PREFIX_MIN_TOKENS = _prefix_setting("BLINK_PREFIX_MIN_TOKENS", 256)  # below this the plain path is as fast
+PREFIX_KV_TOKENS = _prefix_setting("BLINK_PREFIX_KV_TOKENS", 131072)  # prefix copies x length held at once
+# Readout implementation: "rows" (release: per-question gather, upcast and GEMV) or "batched" (one gather, one FP32
+# matmul and one host transfer per forward; same arithmetic, different FP32 accumulation order). Opt-in under E1.
+READOUT = os.environ.get("BLINK_READOUT", "rows").strip().lower()
 GPU_DURATION = int(os.environ.get("BLINK_GPU_DURATION", "60"))
 
 SYSTEM = (
@@ -427,21 +450,42 @@ class TorchEngine:
     def logits(self, state, questions: dict, bias: dict | None = None) -> tuple[dict[str, list[float]], int]:
         del bias  # demo-only shaping; the trained model reads the evidence instead
         work = self.render(state, questions)
-        rows, self.last_model_ms = _forward(self.key, [w["ids"] for w in work], [w["cand"] for w in work])
+        seqs = [w["ids"] for w in work]
+        rows, self.last_model_ms = _forward(self.key, seqs, [w["cand"] for w in work])
+        self.last_prefill_tokens = _prefill_tokens(seqs, getattr(self, "prefix_cache", PREFIX_CACHE))
         return (
             {w["qkey"]: r for w, r in zip(work, rows)},
-            sum(len(w["ids"]) for w in work),
+            sum(len(s) for s in seqs),
         )
 
-
-def _batches(lengths: list[int], budget: int):
-    """Shortest first; a batch closes when (longest x count) would pass the padded-token budget.
+    def logits_many(self, requests: list) -> list[tuple[dict[str, list[float]], int]]:
+        """Several already-validated requests in one GPU call: every question of every request is packed into the
+        same padded forwards, then split back per request. Cross-request batching for a queueing server (E1 arm H1);
+        each request's rows come only from its own prompts."""
+        return self.logits_rendered([self.render(state, questions) for state, questions in requests])
+
+    def logits_rendered(self, works: list) -> list[tuple[dict[str, list[float]], int]]:
+        """logits_many for requests already rendered (one render() list per request), so a caller can render each
+        request on its own and keep one request's rendering error away from the others."""
+        seqs = [w["ids"] for work in works for w in work]
+        cands = [w["cand"] for work in works for w in work]
+        rows, self.last_model_ms = _forward(self.key, seqs, cands) if seqs else ([], 0.0)
+        out, k = [], 0
+        for work in works:
+            part = rows[k:k + len(work)]
+            k += len(work)
+            out.append(({w["qkey"]: r for w, r in zip(work, part)}, sum(len(w["ids"]) for w in work)))
+        return out
+
+
+def _batches(lengths: list[int], budget: int, max_rows: int | None = None):
+    """Shortest first; a batch closes when (longest x count) would pass the padded-token budget, or at max_rows.
     A sequence longer than the budget runs alone."""
     order = sorted(range(len(lengths)), key=lambda i: lengths[i])
     batch, longest = [], 0
     for i in order:
         grown = max(longest, lengths[i])
-        if batch and grown * (len(batch) + 1) > budget:
+        if batch and (grown * (len(batch) + 1) > budget or (max_rows and len(batch) >= max_rows)):
             yield batch
             batch, grown = [], lengths[i]
         batch.append(i)
@@ -450,31 +494,128 @@ def _batches(lengths: list[int], budget: int):
         yield batch
 
 
+def _shared_prefix_len(seqs: list[list[int]]) -> int:
+    """Length of the token prefix common to every sequence, leaving each at least one token of its own
+    (the answer position must be computed in the per-question pass)."""
+    if len(seqs) < 2:
+        return 0
+    lo, hi = min(seqs), max(seqs)  # the common prefix of all = the common prefix of the lexicographic extremes
+    n = min(len(lo), len(hi))
+    p = 0
+    while p < n and lo[p] == hi[p]:
+        p += 1
+    return min(p, min(len(s) for s in seqs) - 1)
+
+
+def _prefill_tokens(seqs: list[list[int]], prefix_cache: bool) -> int:
+    """Tokens the model actually computes for these prompts: the shared prefix once when it is reused."""
+    total = sum(len(s) for s in seqs)
+    shared = _shared_prefix_len(seqs) if prefix_cache else 0
+    return total - (len(seqs) - 1) * shared if shared >= PREFIX_MIN_TOKENS else total
+
+
+def _expand_cache(cache, repeats: int):
+    """A deep copy of a batch-1 cache repeated `repeats` times along the batch axis, for every tensor it holds:
+    full-attention keys/values and the linear-attention conv and recurrent states (the library's own
+    batch_repeat_interleave only covers keys/values)."""
+    import copy
+
+    import torch
+
+    cache = copy.deepcopy(cache)
+    if repeats == 1:
+        return cache
+
+    def rep(x):
+        return x.repeat_interleave(repeats, dim=0) if torch.is_tensor(x) and x.numel() else x
+
+    for layer in cache.layers:
+        for name in ("keys", "values"):
+            if torch.is_tensor(getattr(layer, name, None)):
+                setattr(layer, name, rep(getattr(layer, name)))
+        for name in ("conv_states", "recurrent_states"):
+            held = getattr(layer, name, None)
+            if isinstance(held, dict):
+                setattr(layer, name, {k: rep(v) for k, v in held.items()})
+            elif isinstance(held, (list, tuple)):
+                setattr(layer, name, type(held)(rep(v) for v in held))
+    return cache
+
+
+def _read(model, head, h, lengths, cands, rows, out, device):
+    import torch
+
+    if READOUT == "batched":
+        return _read_batched(head, h, lengths, cands, rows, out, device)
+    last = torch.tensor([n - 1 for n in lengths], device=device)
+    h = h[torch.arange(len(rows), device=device), last].float()
+    for r, i in enumerate(rows):
+        w = head[torch.tensor(cands[i], device=device)].float()
+        out[i] = (w @ h[r]).tolist()
+
+
+def _read_batched(head, h, lengths, cands, rows, out, device):
+    """The same FP32 readout as _read, for a whole forward at once: the batch's distinct label rows are gathered
+    and upcast once, one FP32 matmul scores every question against them, and one host transfer brings them back.
+    Only the order of FP32 accumulation differs from the per-question path (BLINK_READOUT=batched; opt-in)."""
+    import torch
+
+    last = torch.tensor([n - 1 for n in lengths], device=device)
+    hs = h[torch.arange(len(rows), device=device), last].float()
+    uniq = sorted({t for i in rows for t in cands[i]})
+    col = {t: j for j, t in enumerate(uniq)}
+    z = (hs @ head[torch.tensor(uniq, device=device)].float().T).tolist()
+    for r, i in enumerate(rows):
+        out[i] = [z[r][col[t]] for t in cands[i]]
+
+
+def _padded(seqs, rows, pad_id):
+    import torch
+
+    L = max(len(seqs[i]) for i in rows)
+    ids = torch.full((len(rows), L), pad_id, dtype=torch.long)
+    for r, i in enumerate(rows):
+        ids[r, : len(seqs[i])] = torch.tensor(seqs[i], dtype=torch.long)
+    return ids
+
+
+def _forward_shared(model, head, seqs, cands, prefix_len: int, pad_id: int, budget: int, out, device) -> None:
+    """Encode seqs[0][:prefix_len] once, then run every tail against an expanded copy of that cache."""
+    import torch
+
+    prefix = torch.tensor([seqs[0][:prefix_len]], dtype=torch.long, device=device)
+    cache = model.model(input_ids=prefix, use_cache=True).past_key_values
+    tails = [s[prefix_len:] for s in seqs]
+    rows_cap = max(1, PREFIX_KV_TOKENS // (prefix_len + max(len(t) for t in tails)))
+    for b in _batches([len(t) for t in tails], budget, max_rows=rows_cap):
+        ids = _padded(tails, b, pad_id).to(device)
+        h = model.model(input_ids=ids, past_key_values=_expand_cache(cache, len(b)), use_cache=True).last_hidden_state
+        _read(model, head, h, [len(tails[i]) for i in b], cands, b, out, device)
+
+
 @_gpu(GPU_DURATION)
 def _forward(key: int, seqs: list[list[int]], cands: list[list[int]]):
     """Right-padded maskless prefill; every layer is causal, so padding cannot reach the
-    last real position. Label rows of lm_head are applied in FP32."""
+    last real position. Label rows of lm_head are applied in FP32. With several questions over
+    a long shared prefix, the prefix is encoded once (see PREFIX_CACHE)."""
     import torch
 
     engine = _LIVE[key]
     model = engine.model
     device = next(model.parameters()).device
     head = model.lm_head.weight
+    budget = getattr(engine, "token_budget", TOKEN_BUDGET)
     out: list = [None] * len(seqs)
     t0 = time.perf_counter()
     with torch.no_grad():
-        for b in _batches([len(s) for s in seqs], getattr(engine, "token_budget", TOKEN_BUDGET)):
-            L = max(len(seqs[i]) for i in b)
-            ids = torch.full((len(b), L), engine.pad_id, dtype=torch.long)
-            for r, i in enumerate(b):
-                ids[r, : len(seqs[i])] = torch.tensor(seqs[i], dtype=torch.long)
-            ids = ids.to(device)
-            h = model.model(input_ids=ids, use_cache=False).last_hidden_state
-            last = torch.tensor([len(seqs[i]) - 1 for i in b], device=device)
-            h = h[torch.arange(len(b), device=device), last].float()
-            for r, i in enumerate(b):
-                w = head[torch.tensor(cands[i], device=device)].float()
-                out[i] = (w @ h[r]).tolist()
+        shared = _shared_prefix_len(seqs) if getattr(engine, "prefix_cache", PREFIX_CACHE) else 0
+        if shared > 0 and shared >= PREFIX_MIN_TOKENS:
+            _forward_shared(model, head, seqs, cands, shared, engine.pad_id, budget, out, device)
+        else:
+            for b in _batches([len(s) for s in seqs], budget):
+                ids = _padded(seqs, b, engine.pad_id).to(device)
+                h = model.model(input_ids=ids, use_cache=False).last_hidden_state
+                _read(model, head, h, [len(seqs[i]) for i in b], cands, b, out, device)
     return out, round((time.perf_counter() - t0) * 1000, 1)
 
 
@@ -620,4 +761,144 @@ def decide(state, questions: dict, temperature: float | None = None, bias: dict
         model_ms = getattr(getattr(eng, "live", eng), "last_model_ms", None)
         if model_ms is not None:
             meta["model_ms"] = model_ms  # forward passes only; excludes any wait for a device
+        prefill = getattr(getattr(eng, "live", eng), "last_prefill_tokens", None)
+        if prefill is not None:
+            meta["prefill_tokens"] = prefill  # tokens computed; input_tokens counts every question's full prompt
     return {"answers": answers, "meta": meta}
+
+
+def decide_many(requests: list, temperature: float | None = None, model: str | None = None) -> list:
+    """Several {state, questions} requests in one GPU call (E1 arm H1). Returns one item per request, in order: the
+    same {answers, meta} dict decide() returns, or the exception that request raised (a BlinkError when blink can't
+    answer it). Each request is validated and rendered on its own, so one request's failure never reaches another;
+    if the packed forward itself fails, every request is answered alone instead. Engines without logits_many
+    answer one request at a time."""
+    eng = engine(model)
+    T = float(temperature if temperature is not None else eng.temperature)
+    if T <= 0:
+        raise BlinkError("temperature must be positive")
+    live = getattr(eng, "live", eng)
+    packed = hasattr(live, "logits_rendered")
+    out: list = [None] * len(requests)
+    todo, works = [], []
+    for i, (state, questions) in enumerate(requests):
+        try:
+            validate(questions)
+            if packed:
+                works.append(live.render(state, questions))
+            todo.append(i)
+        except Exception as exc:  # noqa: BLE001 - this request's own error (422 for a BlinkError, else 500)
+            out[i] = exc
+    t0 = time.perf_counter()
+    got: dict = {}
+    if packed and todo:
+        try:
+            got = dict(zip(todo, live.logits_rendered(works)))
+        except Exception:  # noqa: BLE001 - e.g. out of memory on the packed batch: answer each request alone below
+            got = {}
+    n_packed = len(got)
+    for i in todo:
+        if i not in got:
+            try:
+                got[i] = (live if packed else eng).logits(*requests[i])
+            except Exception as exc:  # noqa: BLE001
+                got[i] = exc
+    latency_ms = round((time.perf_counter() - t0) * 1000, 1)
+    for i in todo:
+        res = got[i]
+        if isinstance(res, Exception):
+            out[i] = res
+            continue
+        try:
+            raw, n_tokens = res
+            questions = requests[i][1]
+            answers = {q: answer_for(spec, [k for k, _ in question_options(spec)], softmax(raw[q], T))
+                       for q, spec in questions.items()}
+            out[i] = {"answers": answers, "meta": {"model": getattr(eng, "model_id", MODEL_ID), "engine": eng.name,
+                                                   "temperature": T, "input_tokens": n_tokens, "generated_tokens": 0,
+                                                   "latency_ms": latency_ms, "batched_requests": n_packed or 1}}
+        except Exception as exc:  # noqa: BLE001
+            out[i] = exc
+    return out
+
+
+class BlinkBusy(RuntimeError):
+    """The batching queue is full: the server is over capacity (HTTP 503; retry shortly)."""
+
+
+class Batcher:
+    """Cross-request batching for a serving process (E1 arm H1). One worker thread owns the engine: it takes the
+    oldest waiting request, waits up to `window_s` for more (at most `max_requests` in all) and decides them together
+    with decide_many. At most `max_queued` requests wait; past that, submit() raises BlinkBusy at once. submit()
+    blocks the calling thread until its own result is ready and returns it, or raises that request's own error."""
+
+    MAX_WINDOW_S = 1.0
+    MAX_REQUESTS = 64  # the largest group E1 measured
+
+    def __init__(self, window_s: float, max_requests: int = 16, model: str | None = None, max_queued: int = 64):
+        import math
+        import queue
+        import threading
+
+        window_s, max_requests, max_queued = float(window_s), int(max_requests), int(max_queued)
+        if not (math.isfinite(window_s) and 0 < window_s <= self.MAX_WINDOW_S):
+            raise ValueError(f"batch window must be more than 0 and at most {self.MAX_WINDOW_S:g} s")
+        if not 1 <= max_requests <= self.MAX_REQUESTS:
+            raise ValueError(f"max_requests must be 1-{self.MAX_REQUESTS}")
+        if max_queued < 1:
+            raise ValueError("max_queued must be at least 1")
+        self.window_s, self.max_requests, self.max_queued, self.model = window_s, max_requests, max_queued, model
+        self.q = queue.Queue(maxsize=max_queued)
+        self.worker = threading.Thread(target=self._loop, name="blink-batcher", daemon=True)
+        self.worker.start()
+
+    def alive(self) -> bool:
+        return self.worker.is_alive()
+
+    def submit(self, state, questions: dict) -> dict:
+        import concurrent.futures
+        import queue
+
+        validate(questions)  # a malformed request fails at once and never takes a place in the queue
+        if not self.worker.is_alive():
+            raise RuntimeError("the batching worker has stopped")
+        fut = concurrent.futures.Future()
+        try:
+            self.q.put_nowait((state, questions, fut))
+        except queue.Full:
+            raise BlinkBusy(f"{self.max_queued} requests are already waiting; retry shortly") from None
+        while not fut.done():
+            concurrent.futures.wait([fut], timeout=1.0)
+            if not fut.done() and not self.worker.is_alive():
+                raise RuntimeError("the batching worker has stopped")
+        return fut.result()
+
+    def _loop(self) -> None:
+        import queue
+
+        while True:
+            batch = [self.q.get()]
+            results = None
+            try:
+                deadline = time.monotonic() + self.window_s
+                while len(batch) < self.max_requests:
+                    left = deadline - time.monotonic()
+                    if left <= 0:
+                        break
+                    try:
+                        batch.append(self.q.get(timeout=left))
+                    except queue.Empty:
+                        break
+                results = decide_many([(s, q) for s, q, _ in batch], model=self.model)
+            except Exception as exc:  # noqa: BLE001 - a failure outside any one request (e.g. the engine): each caller gets it
+                results = [exc] * len(batch)
+            finally:
+                if results is None or len(results) != len(batch):
+                    results = [None] * len(batch)
+                for (_, _, fut), res in zip(batch, results):
+                    if res is None:
+                        fut.set_exception(RuntimeError("the batching worker lost this request"))
+                    elif isinstance(res, BaseException):
+                        fut.set_exception(res)
+                    else:
+                        fut.set_result(res)

serve.py

@@ -12,6 +12,10 @@ questions per request) gets HTTP 422 with the reason; nothing is truncated. With
 the weights, every listed file is hashed before serving (weights_verified). Serving a local folder switches
 the Hugging Face libraries to offline mode before any of them loads (hub_offline reports the setting the
 libraries actually use); the server does not otherwise restrict the network.
+
+Opt-in cross-request batching (--batch-window-ms, default 0 = off): requests that arrive within the window are
+decided together, up to --max-batch-requests; at most --max-queued-requests wait, and past that a request gets
+HTTP 503 with Retry-After. Each request still gets its own answers or its own error.
 """
 
 from __future__ import annotations
@@ -67,12 +71,44 @@ def versions() -> dict:
     return out
 
 
+def window_ms(value: str) -> float:
+    """--batch-window-ms: 0 (off) or a finite window of at most 1000 ms."""
+    try:
+        ms = float(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"not a number: {value!r}") from None
+    if not 0 <= ms <= 1000:  # also rejects nan and inf
+        raise argparse.ArgumentTypeError("must be 0 (off) to 1000 ms")
+    return ms
+
+
+def int_range(lo: int, hi: int):
+    def parse(value: str) -> int:
+        try:
+            n = int(value)
+        except ValueError:
+            raise argparse.ArgumentTypeError(f"not an integer: {value!r}") from None
+        if not lo <= n <= hi:
+            raise argparse.ArgumentTypeError(f"must be {lo}-{hi}")
+        return n
+
+    return parse
+
+
 def main() -> None:
     ap = argparse.ArgumentParser(description="Serve blink over a Jev-compatible HTTP API.")
     ap.add_argument("--model", default=os.environ.get("BLINK_MODEL", "thegovind/blink-4b"))
     ap.add_argument("--revision", default=os.environ.get("BLINK_REVISION"))
     ap.add_argument("--host", default="127.0.0.1")
     ap.add_argument("--port", type=int, default=8000)
+    # a string default is parsed only when the flag is absent, so a stray BLINK_BATCH_WINDOW_MS can't block an explicit 0
+    ap.add_argument("--batch-window-ms", type=window_ms, default=os.environ.get("BLINK_BATCH_WINDOW_MS", "0"),
+                    help="opt-in cross-request batching: requests arriving within this window are decided in one "
+                         "GPU call (0 = one request at a time, the evaluated default; at most 1000)")
+    ap.add_argument("--max-batch-requests", type=int_range(1, 64), default=16,
+                    help="most requests decided together (1-64)")
+    ap.add_argument("--max-queued-requests", type=int_range(1, 1024), default=64,
+                    help="most requests waiting for a batch; past this a request gets HTTP 503 (1-1024)")
     a = ap.parse_args()
 
     local = os.path.isdir(a.model)
@@ -84,6 +120,8 @@ def main() -> None:
     sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
     import blink
 
+    busy = getattr(blink, "BlinkBusy", ())  # an older blink.py has no batching (and no BlinkBusy)
+
     if local:
         root = a.model
     else:
@@ -109,6 +147,17 @@ def main() -> None:
               "hub_offline": hub_offline, "warmup": {"repeat_identical": repeat_identical}, "kernels": kernels,
               "versions": ver}
     lock = threading.Lock()
+    batcher = (blink.Batcher(a.batch_window_ms / 1000.0, a.max_batch_requests, max_queued=a.max_queued_requests)
+               if a.batch_window_ms > 0 else None)
+    health["batching"] = ({"window_ms": a.batch_window_ms, "max_requests": a.max_batch_requests,
+                           "max_queued": a.max_queued_requests} if batcher else None)
+
+    def current_health() -> dict:
+        if batcher is None:
+            return health
+        alive = batcher.alive()
+        return {**health, "ok": health["ok"] and alive,
+                "batching": {**health["batching"], "worker_alive": alive, "queued": batcher.q.qsize()}}
 
     class Handler(BaseHTTPRequestHandler):
         protocol_version = "HTTP/1.1"
@@ -122,17 +171,19 @@ def main() -> None:
             # on the client's delayed ACK, a flat ~40 ms on every request of a kept-alive connection
             self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 
-        def _send(self, code: int, obj: dict) -> None:
+        def _send(self, code: int, obj: dict, headers: dict | None = None) -> None:
             body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
             self.send_response(code)
             self.send_header("Content-Type", "application/json")
             self.send_header("Content-Length", str(len(body)))
+            for name, value in (headers or {}).items():
+                self.send_header(name, value)
             self.end_headers()
             self.wfile.write(body)
 
         def do_GET(self):
             if self.path.rstrip("/") in ("/healthz", "/health"):
-                return self._send(200, health)
+                return self._send(200, current_health())
             return self._send(404, {"error": "not found"})
 
         def do_POST(self):
@@ -146,10 +197,15 @@ def main() -> None:
             if not isinstance(req, dict):
                 return self._send(400, {"error": "the body must be a JSON object"})
             try:
-                with lock:
-                    out = blink.decide(req.get("state"), req.get("questions"))
+                if batcher is not None:
+                    out = batcher.submit(req.get("state"), req.get("questions"))
+                else:
+                    with lock:
+                        out = blink.decide(req.get("state"), req.get("questions"))
             except blink.BlinkError as exc:
                 return self._send(422, {"error": str(exc)})
+            except busy as exc:
+                return self._send(503, {"error": str(exc)}, {"Retry-After": "1"})
             except Exception as exc:  # noqa: BLE001 - report, keep serving
                 return self._send(500, {"error": f"{type(exc).__name__}: {exc}"})
             return self._send(200, {