This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new bcbb825e199 Stop rejecting valid non-ASCII Anthropic batches as too
large (#74662)
bcbb825e199 is described below
commit bcbb825e19967f86ab850d13fe0020f3bb1013bb
Author: Henry Chen <[email protected]>
AuthorDate: Wed Oct 14 20:15:57 2026 +0800
Stop rejecting valid non-ASCII Anthropic batches as too large (#74662)
* Cover the Anthropic batch payload-size estimate for non-ASCII prompts
The adapter takes the batch inline, so the bytes that count are the ones
httpx encodes with ensure_ascii=False as UTF-8. Sizing the payload with
len(json.dumps(...)) counts \uXXXX escapes instead, inflating CJK input
2x and non-BMP input 3x, so batches the API would accept are rejected
locally and the reported byte count in the error is wrong. The openai
adapter already measures encoded bytes, so the two disagree today.
The expectations are marked strict xfail so the inconsistency is
recorded without changing behaviour; the fix has to drop the markers.
* Stop rejecting valid non-ASCII Anthropic batches as too large
The adapter hands the requests to the SDK inline, so the bytes that count
against Anthropic's 256 MB payload limit are the ones httpx puts on the
wire: ensure_ascii=False, then UTF-8. Sizing the payload with json.dumps
defaults measured the \uXXXX escape instead, charging a CJK character six
bytes rather than three and a non-BMP character twelve rather than four.
A batch the API would have accepted was refused locally, and the byte
count in the error was up to three times the real one, so the suggested
remedy was sized against a number that did not exist.
The openai adapter measures the ASCII-escaped form on purpose, because it
builds the uploaded JSONL file that way itself; only the inline path was
wrong.
* Measure the Anthropic batch payload the way the SDK serializes it
The size the adapter quotes back is the number a user sizes their
.expand() split against, so it should be the number that reaches
Anthropic. Dropping the ASCII escaping got most of the way there, but
the json.dumps defaults still spend two bytes of whitespace per
separator, leaving the estimate around 8% above the body the SDK
actually sends -- a gap independent of the characters involved, so the
non-ASCII tests could not see it either.
They could not see it for a second reason: the expectation repeated the
expression under test, which only ever confirms the code agrees with
itself. Deriving it from the SDK's encoder instead makes the comparison
independent, and leaves a tripwire if a future SDK serializes a request
differently.
The serializer is the SDK's own, not httpx's: it dumps the body and
hands httpx the finished bytes, so the docstrings now point at the
encoder that decides the size.
* Charge the Anthropic batch limit for the body's framing
The limit this pre-check enforces is charged against the body Anthropic
receives, and the requests arrive inside one {"requests": [...]} array
rather than on their own. Summing the requests alone leaves out that
framing and the comma joining each pair -- around 100 KB at the
100,000-request maximum -- so a batch in that window passes here and is
refused by the API, which is the one outcome the pre-check exists to
prevent. openai.py is already charged its JSONL newline the same way.
Nothing caught this earlier because the default separators over-counted
by more than the framing omits, so the net estimate stayed above the real
body until the previous commit made each request exact.
* Cover the running total in the Anthropic payload-size test
One request cannot tell a running total from an overwrite: adding a
single size to zero reads the same as replacing it, so a loop that
dropped the accumulation would still pass while the limit it enforces
only ever saw the last request. The comma between requests is charged
from the second one on, which with one request means that term never ran
at all -- deleting it left the test green. Two prompts of unequal length
exercise both, the way the OpenAI size test already does.
---
.../airflow/providers/common/ai/batch/anthropic.py | 24 +++++++--
.../tests/unit/common/ai/batch/test_anthropic.py | 57 +++++++++++++++++++++-
2 files changed, 77 insertions(+), 4 deletions(-)
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/batch/anthropic.py
b/providers/common/ai/src/airflow/providers/common/ai/batch/anthropic.py
index d928026dddb..5cd53b5735d 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/batch/anthropic.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/batch/anthropic.py
@@ -73,6 +73,21 @@ def _batch_status(batch: Any) -> BatchStatus:
return _STATUS_MAP.get(batch.processing_status, "in_progress")
+def _utf8_size(payload: Any) -> int:
+ r"""
+ Measure what one inline ``batches.create`` request costs on the wire, in
bytes.
+
+ The SDK serializes the batch itself and hands httpx a finished body, so
its own
+ encoder settles the size (``anthropic/_utils/_json.py``):
``ensure_ascii=False``,
+ ``separators=(",", ":")``, then UTF-8. The ``json.dumps`` defaults would
charge a
+ CJK character the six bytes of its ``\uXXXX`` escape rather than the three
it
+ occupies, plus a byte of whitespace per separator. The ``{"requests":
[...]}``
+ framing the requests sit in belongs to the batch, not to one request, so
the
+ caller adds it.
+ """
+ return len(json.dumps(payload, ensure_ascii=False, separators=(",",
":")).encode())
+
+
def _build_client(api_key: str | None, base_url: str | None) -> Any:
try:
from anthropic import Anthropic
@@ -215,7 +230,7 @@ class AnthropicBatchAdapter(BatchAdapter):
)
directive = self.build_output_directive(output_spec)
- directive_bytes = len(json.dumps(directive))
+ directive_bytes = _utf8_size(directive)
built = self._iter_requests(
requests,
model=model,
@@ -225,9 +240,12 @@ class AnthropicBatchAdapter(BatchAdapter):
max_tokens=max_tokens,
request_params=request_params,
)
- total_bytes = 0
+ # The requests reach Anthropic as one ``{"requests": [...]}`` body, so
the limit is
+ # charged the framing and the comma joining each pair as well, the way
``openai.py``
+ # is charged a newline per JSONL line.
+ total_bytes = len('{"requests":[]}')
for index, item in enumerate(built):
- total_bytes += len(json.dumps(item))
+ total_bytes += _utf8_size(item) + (1 if index else 0)
if total_bytes > self.max_payload_bytes:
raise LLMBatchLimitExceededError(
f"Anthropic batch accepts a payload of at most
{self.max_payload_bytes} bytes; "
diff --git a/providers/common/ai/tests/unit/common/ai/batch/test_anthropic.py
b/providers/common/ai/tests/unit/common/ai/batch/test_anthropic.py
index 9470e2ec596..d01f411474a 100644
--- a/providers/common/ai/tests/unit/common/ai/batch/test_anthropic.py
+++ b/providers/common/ai/tests/unit/common/ai/batch/test_anthropic.py
@@ -19,7 +19,7 @@ from __future__ import annotations
from typing import Any
import pytest
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
from airflow.providers.common.ai.batch.anthropic import AnthropicBatchAdapter
from airflow.providers.common.ai.batch.base import RawResultItem
@@ -28,6 +28,7 @@ from airflow.providers.common.ai.exceptions import
LLMBatchLimitExceededError, L
anthropic = pytest.importorskip("anthropic")
+from anthropic._utils._json import openapi_dumps # noqa: E402
from anthropic.types import Message, TextBlock, ToolUseBlock # noqa: E402
from anthropic.types.messages import MessageBatch,
MessageBatchIndividualResponse # noqa: E402
@@ -266,6 +267,60 @@ class TestSubmitRequestShape:
)
+def _inline_payload_bytes(client) -> int:
+ """
+ Return the bytes the SDK's own encoder produces for the whole
``batches.create`` body.
+
+ Deriving the expectation from the encoder rather than repeating
``_utf8_size``'s
+ expression keeps it independent of the code under test, and turns these
tests into
+ a tripwire for an SDK that changes how it serializes a batch.
+ """
+ built = client.messages.batches.created[0]["requests"]
+ return len(openapi_dumps({"requests": built}))
+
+
+class TestPayloadSizeEstimate:
+ @pytest.mark.parametrize(
+ "prompts",
+ [
+ pytest.param(["abc", "abcdef"], id="ascii"),
+ pytest.param(["中文字", "中文字串測試"], id="cjk"),
+ pytest.param(["😀😀😀", "😀😀😀😀😀"], id="non-bmp"),
+ ],
+ )
+ def test_estimate_matches_the_inline_payload_bytes(self, adapter, client,
str_spec, prompts):
+ # Two requests of unequal length: with one, the running total reads
the same as an
+ # overwrite, and the comma is never charged at all -- it starts at the
second request.
+ requests = [{"prompt": prompt} for prompt in prompts]
+ _submit(adapter, requests, str_spec)
+ on_the_wire = _inline_payload_bytes(client)
+
+ adapter.max_payload_bytes = on_the_wire
+ adapter.validate_requests(
+ requests, model="claude-sonnet-5", idempotency_key=KEY,
output_spec=str_spec
+ )
+
+ adapter.max_payload_bytes = on_the_wire - 1
+ with pytest.raises(LLMBatchLimitExceededError):
+ adapter.validate_requests(
+ requests, model="claude-sonnet-5", idempotency_key=KEY,
output_spec=str_spec
+ )
+
+ def test_reported_directive_size_is_in_utf8_bytes(self, adapter):
+ class NonAsciiSchema(BaseModel):
+ field: str = Field(description="中" * 100)
+
+ spec = build_output_spec(NonAsciiSchema)
+ directive = adapter.build_output_directive(spec)
+ expected = len(openapi_dumps(directive))
+
+ adapter.max_payload_bytes = 0
+ with pytest.raises(LLMBatchLimitExceededError, match=rf"adds
~{expected} bytes"):
+ adapter.validate_requests(
+ [{"prompt": "a"}], model="claude-sonnet-5",
idempotency_key=KEY, output_spec=spec
+ )
+
+
class TestFindOrphanedBatch:
def test_always_returns_none(self, adapter):
assert adapter.find_orphaned_batch("any-key", "any-fp",
"1970-01-01T00:00:00+00:00") is None
