feat(python): add opt-in request batching to TypeSafeReranker (#4316)

Adds opt-in `batch_size=40`: 80 non-null candidates use **2 requests
instead of 80**. Default `batch_size=1` preserves the existing payload.
Each batched question sees only its document and the shared query;
instructions/criteria stay unchanged. Prompts referencing
`state.document` need adaptation. Concurrency still limits requests;
retries remain SDK-managed. Batch-local IDs and ordered results preserve
duplicate documents; invalid IDs/probabilities raise without fallback.

Validation: **124 automated tests passed**, plus live search
integrations in both modes. A **100-query live subset test of the
initial implementation** (6,851 candidate pairs; `jev-1.13.0`, SDK
0.7.1, concurrency 32) passed: unbatched/batched calls **6,851/199**,
median **877/351 ms**, p95 **9,446/598 ms**. Hybrid Hit@5/Hit@10 was
**83%/89% vs 84%/90%**; vector/FTS results and exact configuration are
in the [test
report](https://github.com/lancedb/lancedb/blob/typesafe-request-batching/python/benchmarks/typesafe_batching.md).
Ruff and MkDocs passed. The simplified batching flow also passed a fresh
four-query live check: 292 candidate pairs, with 292 unbatched versus 8
batched calls.

Builds on #4209; motivated by [research
#7](https://github.com/lancedb/research/pull/7).

python/python/lancedb/rerankers/typesafe.py

@@ -4,7 +4,8 @@
 
 from concurrent.futures import ThreadPoolExecutor
 from functools import cached_property
-from typing import Any, Dict, Mapping, Optional
+from itertools import chain
+from typing import Any, Dict, List, Mapping, Optional
 
 import pyarrow as pa
 
@@ -38,7 +39,11 @@ class TypeSafeReranker(Reranker):
     and can vary slightly between identical calls, so results with close scores
     may swap places when the same search is repeated.
 
-    One request is sent per result, up to ``max_concurrency`` at a time.
+    By default, one request is sent per non-null result. Set ``batch_size=40``
+    to score up to 40 documents per request, with at most ``max_concurrency``
+    requests in flight. Each question sees only its own document and the shared
+    query. Null documents keep a score of zero without an API request; empty
+    strings are scored normally. API and response-validation errors propagate.
 
     Parameters
     ----------
@@ -48,8 +53,10 @@ class TypeSafeReranker(Reranker):
         The name of the column holding the document text to score.
     instructions : str, optional
         The yes/no question asked about each query and document pair. The state
-        TypeSafe reads is ``{"query": <query>, "document": <column value>}``.
-        Defaults to a generic relevance question.
+        TypeSafe reads with ``batch_size=1`` is
+        ``{"query": <query>, "document": <column value>}``.
+        Defaults to a generic relevance question. See ``batch_size`` for the
+        different location of the document in batched requests.
     criteria : Mapping[str, str], optional
         What a "yes" and a "no" mean, as a mapping with the keys ``"true"`` and
         ``"false"``. Domain-specific criteria usually rank better than the
@@ -62,6 +69,19 @@ class TypeSafeReranker(Reranker):
         ``TYPESAFE_API_KEY`` environment variable.
     max_concurrency : int, default 8
         The maximum number of TypeSafe requests in flight for one rerank call.
+    batch_size : int, default 1
+        Positive integer limiting non-null documents per request. The default
+        preserves the original request format. With a value greater than one,
+        state is ``{"query": <query>}`` and each question's instructions are
+        ``{"question": <instructions>, "document": <column value>}``.
+        Instructions and criteria are preserved verbatim. Custom prompts that
+        explicitly reference the document in request state (for example,
+        ``state.document``) must be adapted before opting into batching.
+        The TypeSafe SDK handles retries in both modes.
+
+    Examples
+    --------
+    >>> reranker = TypeSafeReranker(batch_size=40, max_concurrency=8)
     """
 
     def __init__(
@@ -73,10 +93,17 @@ class TypeSafeReranker(Reranker):
         return_score: str = "relevance",
         api_key: Optional[str] = None,
         max_concurrency: int = 8,
+        batch_size: int = 1,
     ):
         super().__init__(return_score)
         if max_concurrency < 1:
             raise ValueError("max_concurrency must be at least 1")
+        if (
+            isinstance(batch_size, bool)
+            or not isinstance(batch_size, int)
+            or batch_size < 1
+        ):
+            raise ValueError("batch_size must be a positive integer")
         criteria = DEFAULT_CRITERIA if criteria is None else dict(criteria)
         unknown = set(criteria) - {"true", "false"}
         if unknown:
@@ -89,6 +116,7 @@ class TypeSafeReranker(Reranker):
         self.criteria = criteria
         self.api_key = api_key
         self.max_concurrency = max_concurrency
+        self.batch_size = batch_size
 
     def __str__(self):
         return f"TypeSafeReranker(model_name={self.model_name})"
@@ -105,27 +133,70 @@ class TypeSafeReranker(Reranker):
             question["criteria"] = self.criteria
         return question
 
-    def _score(self, query: str, document: Optional[str]) -> float:
-        if document is None:
-            return 0.0
+    def _score_batch(self, query: str, documents: List[str]) -> List[float]:
+        if self.batch_size == 1:
+            state = {"query": query, "document": documents[0]}
+            questions = {_QUESTION_ID: self._question}
+        else:
+            state = {"query": query}
+            # IDs only need to be unique within this request, even for duplicates.
+            questions = {
+                str(index): {
+                    **self._question,
+                    "instructions": {
+                        "question": self.instructions,
+                        "document": document,
+                    },
+                }
+                for index, document in enumerate(documents)
+            }
         response = self._client.system_one(
-            state={"query": query, "document": document},
-            questions={_QUESTION_ID: self._question},
+            state=state,
+            questions=questions,
             model=self.model_name,
         )
-        return response.answers[_QUESTION_ID].noul
+        expected = set(questions)
+        actual = set(response.answers)
+        if actual != expected:
+            raise ValueError(
+                "TypeSafe returned mismatched answer IDs: "
+                f"missing {sorted(expected - actual)}, "
+                f"unexpected {sorted(actual - expected)}"
+            )
+        scores = []
+        for question_id in questions:
+            score = getattr(response.answers[question_id], "noul", None)
+            if (
+                isinstance(score, bool)
+                or not isinstance(score, (int, float))
+                or not 0 <= score <= 1
+            ):
+                raise ValueError(
+                    "TypeSafe returned an invalid relevance probability "
+                    f"for answer {question_id!r}: {score!r}"
+                )
+            scores.append(float(score))
+        return scores
 
     def _rerank(self, result_set: pa.Table, query: str) -> pa.Table:
         result_set = self._handle_empty_results(result_set)
         if len(result_set) == 0:
             return result_set
         docs = result_set[self.column].to_pylist()
+        documents = [doc for doc in docs if doc is not None]
+        batches = [
+            documents[start : start + self.batch_size]
+            for start in range(0, len(documents), self.batch_size)
+        ]
         # Rerankers are also called synchronously from inside the async query
         # APIs, so the requests run on threads rather than on an event loop.
-        with ThreadPoolExecutor(
-            max_workers=min(self.max_concurrency, len(docs))
-        ) as pool:
-            scores = list(pool.map(lambda doc: self._score(query, doc), docs))
+        with ThreadPoolExecutor(max_workers=self.max_concurrency) as pool:
+            batch_scores = pool.map(
+                lambda batch: self._score_batch(query, batch), batches
+            )
+            # Both map and _score_batch preserve input order.
+            scored_documents = chain.from_iterable(batch_scores)
+            scores = [0.0 if doc is None else next(scored_documents) for doc in docs]
         result_set = result_set.append_column(
             "_relevance_score", pa.array(scores, type=pa.float32())
         )

python/python/tests/test_rerankers.py

@@ -536,28 +536,43 @@ class _FakeTypeSafeClient:
     def system_one(self, state, questions, model):
         self.requests.append((state, questions, model))
         query_words = set(state["query"].lower().split())
-        doc_words = set(state["document"].lower().split())
-        noul = len(query_words & doc_words) / len(query_words)
-        answers = {key: type("NoulAnswer", (), {"noul": noul})() for key in questions}
+        answers = {}
+        for key, question in questions.items():
+            document = (
+                state["document"]
+                if "document" in state
+                else question["instructions"]["document"]
+            )
+            doc_words = set(document.lower().split())
+            noul = len(query_words & doc_words) / len(query_words)
+            answers[key] = type("NoulAnswer", (), {"noul": noul})()
         return type("SystemOneResponse", (), {"answers": answers})()
 
 
-def test_typesafe_reranker_with_fake_client(tmp_path):
-    reranker = TypeSafeReranker(max_concurrency=4)
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize("return_score", ["relevance", "all"])
+def test_typesafe_reranker_with_fake_client(tmp_path, batch_size, return_score):
+    reranker = TypeSafeReranker(
+        max_concurrency=4, batch_size=batch_size, return_score=return_score
+    )
     reranker._client = _FakeTypeSafeClient()
     table, schema = get_test_table(tmp_path)
     _run_test_reranker(reranker, table, "single player experience", None, schema)
 
     state, questions, model = reranker._client.requests[0]
     assert model == "jev-latest"
-    assert set(state) == {"query", "document"}
-    assert questions == {
-        "relevance": {
-            "type": "noul",
-            "instructions": reranker.instructions,
-            "criteria": reranker.criteria,
+    if batch_size == 1:
+        assert set(state) == {"query", "document"}
+        assert questions == {
+            "relevance": {
+                "type": "noul",
+                "instructions": reranker.instructions,
+                "criteria": reranker.criteria,
+            }
         }
-    }
+    else:
+        assert set(state) == {"query"}
+        assert 1 <= len(questions) <= batch_size
 
 
 def test_typesafe_reranker_scores_each_row():

python/python/tests/test_typesafe_reranker.py

@@ -0,0 +1,338 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import json
+from threading import Barrier, Event, Lock
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import lancedb
+import pyarrow as pa
+import pytest
+from lancedb.index import FTS
+from lancedb.rerankers import TypeSafeReranker
+
+
+def response(questions, probability=0.5):
+    return SimpleNamespace(
+        answers={key: SimpleNamespace(noul=probability) for key in reversed(questions)}
+    )
+
+
+def vector_results(documents):
+    return pa.table(
+        {
+            "text": pa.array(documents, type=pa.string()),
+            "_rowid": pa.array(range(len(documents)), type=pa.uint64()),
+            "_distance": pa.array(range(len(documents)), type=pa.float32()),
+        }
+    )
+
+
+def fake_reranker(**kwargs):
+    reranker = TypeSafeReranker(**kwargs)
+    reranker._client = Mock()
+    reranker._client.system_one.side_effect = lambda **kw: response(kw["questions"])
+    return reranker
+
+
+@pytest.mark.parametrize("batch_size", [0, -1, 1.5, 40.0, True, False, "40", None])
+def test_invalid_batch_size(batch_size):
+    with pytest.raises(ValueError, match="batch_size must be a positive integer"):
+        TypeSafeReranker(batch_size=batch_size)
+
+
+@pytest.mark.parametrize("batch_size", [1, 7, 40])
+@pytest.mark.parametrize("count", [0, 1, 39, 40, 41, 80, 81])
+def test_request_counts_and_boundaries(batch_size, count):
+    reranker = fake_reranker(batch_size=batch_size)
+    docs = [f"candidate {i}" for i in range(count)]
+    result = reranker.rerank_vector("query", vector_results(docs))
+    calls = reranker._client.system_one.call_args_list
+    assert len(calls) == (count + batch_size - 1) // batch_size
+    assert sorted(len(call.kwargs["questions"]) for call in calls) == sorted(
+        min(batch_size, count - start) for start in range(0, count, batch_size)
+    )
+    assert result["_rowid"].to_pylist() == list(range(count))
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize(
+    "criteria", [{}, {"true": "  state.document fits\nexactly", "false": "No!"}]
+)
+def test_payload_isolation_and_custom_prompts(batch_size, criteria):
+    instructions = "Read state.document and state.query.\n Preserve this text. "
+    reranker = fake_reranker(
+        batch_size=batch_size,
+        instructions=instructions,
+        criteria=criteria,
+        model_name="jev-1.13.0",
+        column="body",
+    )
+    docs = [f"UNIQUE_CANDIDATE_{i}" for i in range(41)]
+    table = vector_results(docs).rename_columns(["body", "_rowid", "_distance"])
+    reranker.rerank_vector("shared query", table)
+    seen = []
+    for call in reranker._client.system_one.call_args_list:
+        state, questions = call.kwargs["state"], call.kwargs["questions"]
+        assert call.kwargs["model"] == "jev-1.13.0"
+        if batch_size == 1:
+            doc = state["document"]
+            assert state == {"query": "shared query", "document": doc}
+            assert list(questions) == ["relevance"]
+            expected_instructions = instructions
+        else:
+            assert state == {"query": "shared query"}
+        for question in questions.values():
+            if batch_size > 1:
+                doc = question["instructions"]["document"]
+                expected_instructions = {"question": instructions, "document": doc}
+            expected = {"type": "noul", "instructions": expected_instructions}
+            if criteria:
+                expected["criteria"] = criteria
+            # Exact equality rules out other candidates anywhere in this question.
+            assert question == expected
+            seen.append(doc)
+    assert sorted(seen) == sorted(docs)
+    assert reranker.instructions == instructions
+    assert reranker.criteria == criteria
+
+
+def test_default_payload():
+    reranker = fake_reranker()
+    reranker.rerank_vector("query", vector_results(["document"]))
+    reranker._client.system_one.assert_called_once_with(
+        state={"query": "query", "document": "document"},
+        questions={
+            "relevance": {
+                "type": "noul",
+                "instructions": reranker.instructions,
+                "criteria": reranker.criteria,
+            }
+        },
+        model="jev-latest",
+    )
+
+
+def test_response_mapping_with_duplicates_and_nulls():
+    reranker = fake_reranker(batch_size=2, max_concurrency=2, return_score="all")
+    second_batch_scored = Event()
+
+    def score(**kw):
+        questions = kw["questions"]
+        documents = [
+            question["instructions"]["document"] for question in questions.values()
+        ]
+        if documents == ["same", "same"]:
+            # Prepare the later batch first; result order must still match input.
+            assert second_batch_scored.wait(timeout=10)
+            scores = [0.0, 0.5]
+        else:
+            assert documents == ["different", "same"]
+            scores = [0.75, 1.0]
+        result = SimpleNamespace(
+            answers={
+                key: SimpleNamespace(noul=value)
+                for key, value in reversed(list(zip(questions, scores)))
+            }
+        )
+        second_batch_scored.set()
+        return result
+
+    reranker._client.system_one.side_effect = score
+    result = reranker.rerank_vector(
+        "query", vector_results(["same", None, "same", "different", "same"])
+    )
+    assert result["_rowid"].to_pylist() == [4, 3, 2, 0, 1]
+    assert result["_distance"].to_pylist() == [4, 3, 2, 0, 1]
+    assert result["_relevance_score"].to_pylist() == [1, 0.75, 0.5, 0, 0]
+    assert reranker._client.system_one.call_count == 2
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize("problem", ["missing", "unexpected", "replaced", "no_noul"])
+def test_invalid_answers(batch_size, problem):
+    reranker = fake_reranker(batch_size=batch_size)
+
+    def invalid(**kw):
+        result = response(kw["questions"])
+        key = next(iter(result.answers))
+        if problem in ("missing", "replaced"):
+            del result.answers[key]
+        if problem in ("unexpected", "replaced"):
+            result.answers["unknown"] = SimpleNamespace(noul=0.5)
+        if problem == "no_noul":
+            result.answers[key] = SimpleNamespace()
+        return result
+
+    reranker._client.system_one.side_effect = invalid
+    with pytest.raises(ValueError, match="invalid relevance|answer IDs"):
+        reranker.rerank_vector("query", vector_results(["one", "two"]))
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize(
+    "probability",
+    [None, True, False, "0.5", -0.01, 1.01, float("nan"), float("inf"), -float("inf")],
+)
+def test_invalid_probabilities(batch_size, probability):
+    reranker = fake_reranker(batch_size=batch_size)
+    reranker._client.system_one.side_effect = lambda **kw: response(
+        kw["questions"], probability
+    )
+    with pytest.raises(ValueError, match="invalid relevance probability"):
+        reranker.rerank_vector("query", vector_results(["one", "two"]))
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+def test_api_failure_propagates(batch_size):
+    reranker = fake_reranker(batch_size=batch_size)
+    error = RuntimeError("SDK exhausted retries")
+    reranker._client.system_one.side_effect = error
+    with pytest.raises(RuntimeError) as caught:
+        reranker.rerank_vector("query", vector_results(["one"]))
+    assert caught.value is error
+    assert reranker._client.system_one.call_count == 1
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+def test_concurrency_limits_requests(batch_size):
+    reranker = fake_reranker(batch_size=batch_size, max_concurrency=2)
+    barrier, lock = Barrier(2), Lock()
+    active = peak = 0
+
+    def blocking(**kw):
+        nonlocal active, peak
+        with lock:
+            active += 1
+            peak = max(peak, active)
+        barrier.wait(timeout=10)
+        with lock:
+            active -= 1
+        return response(kw["questions"])
+
+    reranker._client.system_one.side_effect = blocking
+    reranker.rerank_vector("query", vector_results(["doc"] * (6 * batch_size)))
+    assert peak == 2
+    assert active == 0
+    assert reranker._client.system_one.call_count == 6
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize("return_score", ["relevance", "all"])
+@pytest.mark.parametrize("method", ["vector", "fts", "hybrid"])
+@pytest.mark.parametrize("docs", [[], [None, None], [None, "", None]])
+def test_empty_and_null_inputs(batch_size, return_score, method, docs):
+    reranker = fake_reranker(batch_size=batch_size, return_score=return_score)
+    vector = vector_results(docs)
+    fts = vector.rename_columns(["text", "_rowid", "_score"])
+    args = (
+        (vector, fts)
+        if method == "hybrid"
+        else (vector if method == "vector" else fts,)
+    )
+    result = getattr(reranker, f"rerank_{method}")("query", *args)
+    assert len(result) == len(docs)
+    assert result["_relevance_score"].type == pa.float32()
+    assert sorted(result["_relevance_score"].to_pylist()) == sorted(
+        0 if doc is None else 0.5 for doc in docs
+    )
+    assert reranker._client.system_one.call_count == docs.count("")
+    if return_score == "relevance":
+        assert "_distance" not in result.column_names
+        assert "_score" not in result.column_names
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize("return_score", ["relevance", "all"])
+def test_hybrid_deduplicates_by_row_id_and_keeps_scores(batch_size, return_score):
+    reranker = fake_reranker(batch_size=batch_size, return_score=return_score)
+    vector = vector_results(["same", "same", "third"])
+    fts = vector.slice(1).rename_columns(["text", "_rowid", "_score"])
+    result = reranker.rerank_hybrid("query", vector.slice(0, 2), fts)
+    assert result["_rowid"].to_pylist() == [0, 1, 2]
+    assert (
+        sum(
+            len(call.kwargs["questions"])
+            for call in reranker._client.system_one.call_args_list
+        )
+        == 3
+    )
+    assert reranker._client.system_one.call_count == (3 if batch_size == 1 else 1)
+    if return_score == "all":
+        assert result["_distance"].to_pylist() == [0, 1, None]
+        assert result["_score"].to_pylist() == [None, 1, 2]
+    else:
+        assert result.column_names == ["text", "_rowid", "_relevance_score"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("batch_size", [1, 40])
+async def test_async_search_integrations(tmp_path, batch_size):
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.create_table(
+        "docs",
+        [
+            {"text": "cat naps", "vector": [1.0, 0.0]},
+            {"text": "cat plays", "vector": [0.0, 1.0]},
+        ],
+    )
+    await table.create_index("text", config=FTS())
+    reranker = fake_reranker(batch_size=batch_size)
+    queries = [
+        table.query().nearest_to([1.0, 0.0]).rerank(reranker, query_string="cat"),
+        table.query().nearest_to_text("cat").rerank(reranker),
+        table.query().nearest_to([1.0, 0.0]).nearest_to_text("cat").rerank(reranker),
+    ]
+    for query in queries:
+        reranker._client.system_one.reset_mock()
+        result = await query.with_row_id().to_arrow()
+        assert sorted(result["_rowid"].to_pylist()) == [0, 1]
+        assert result["_relevance_score"].to_pylist() == [0.5, 0.5]
+        assert "_distance" not in result.column_names
+        assert "_score" not in result.column_names
+        assert reranker._client.system_one.call_count == (2 if batch_size == 1 else 1)
+
+
+@pytest.mark.parametrize("batch_size", [1, 40])
+@pytest.mark.parametrize("retry_once", [False, True])
+def test_sdk_transport_and_retries(batch_size, retry_once):
+    sdk = pytest.importorskip("typesafe_sdk")
+    httpx = pytest.importorskip("httpx2")
+    payloads = []
+
+    def handle(request):
+        payload = json.loads(request.content)
+        payloads.append(payload)
+        if retry_once and len(payloads) == 1:
+            return httpx.Response(503, headers={"retry-after-ms": "0"})
+        return httpx.Response(
+            200,
+            json={
+                "model": "jev-1.13.0",
+                "usage": {},
+                "answers": {
+                    key: {"type": "noul", "noul": 0.5}
+                    for key in reversed(payload["questions"])
+                },
+            },
+        )
+
+    with sdk.TypeSafeClient(
+        api_key="test-key", transport=httpx.MockTransport(handle)
+    ) as client:
+        reranker = TypeSafeReranker(batch_size=batch_size, max_concurrency=1)
+        reranker._client = client
+        result = reranker.rerank_vector("query", vector_results(["document"] * 80))
+    assert len(payloads) == (80 if batch_size == 1 else 2) + int(retry_once)
+    assert result["_relevance_score"].to_pylist() == [0.5] * 80
+    if retry_once:
+        assert payloads[0] == payloads[1]
+    for payload in payloads:
+        assert len(payload["questions"]) == batch_size
+        if batch_size == 40:
+            assert payload["state"] == {"query": "query"}
+            assert all(
+                question["instructions"]["document"] == "document"
+                for question in payload["questions"].values()
+            )