laantungir/routstr-core
1
0
Fork 0
mirror of https://github.com/Routstr/routstr-core.git synced 2026-09-14 02:35:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(payment): reserve patch-based pricing for detail="original" images

Browse Source 
Images sent with detail="original" are billed by 32x32px patches at
their original resolution (ceil(patches * 1.2) tokens on the
original-capable gpt-5.4/5.5/5.6 families), and the API accepts up to
30,000 patches per image. The tiled estimator capped every non-"low"
image at 765 tokens, so a 2048x2048 original-detail image reserved 765
tokens while billing ~4,900, and file_id references with original
detail reserved nothing.

- Add _calculate_original_image_tokens: patch count from the decoded
  dimensions, bounded by the 30,000-patch rejection limit and billed
  at the documented 1.2x multiplier (exact integer math).
- Use the 36,000-token worst case (30,000 patches * 1.2) for file_id
  references whose dimensions cannot be fetched.
- Route input_image content parts inside messages through
  _estimate_input_image_tokens so their sibling detail and file_id are
  honored on the chat path too, and normalize null detail to auto.
This commit is contained in:
redshift
2026-09-12 12:37:39 +02:00
committed by 9qeklajc
parent efe9c5599d
commit c1c3244840
2 changed files with 219 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routstr/payment/helpers.py
+104
View File
@@ -431,15 +431,46 @@ async def _fetch_image_from_url(url: str) -> bytes | None:
return None
# Patch-based image pricing (OpenAI ``detail: "original"``): the image is
# covered with 32x32px patches and billed as ceil(patches * multiplier)
# tokens, with no 512px-tile downscaling. The API rejects images above
# 30,000 patches, so at the 1.2x multiplier documented for the
# original-capable model families (gpt-5.4/5.5/5.6) the worst case a
# single image can bill is 36,000 tokens.
_IMAGE_PATCH_PX = 32
_MAX_IMAGE_PATCHES = 30_000
_MAX_ORIGINAL_IMAGE_TOKENS = (_MAX_IMAGE_PATCHES * 6 + 4) // 5 # 36,000
def _calculate_original_image_tokens(width: int, height: int) -> int:
"""Estimate tokens for an image billed at ``detail: "original"``.
Patch-based models cover the image with 32x32px patches and bill
``ceil(patches * 1.2)`` tokens. The estimate is bounded by the
30,000-patch rejection limit, which is more conservative than the
per-model resizing patch budgets (e.g. 10,000 patches on gpt-5.4/5.5)
so it never under-reserves.
"""
patches = ((width + _IMAGE_PATCH_PX - 1) // _IMAGE_PATCH_PX) * (
(height + _IMAGE_PATCH_PX - 1) // _IMAGE_PATCH_PX
)
bounded = min(patches, _MAX_IMAGE_PATCHES)
return (bounded * 6 + 4) // 5 # ceil(bounded * 1.2) in exact integer math
def _calculate_image_tokens(width: int, height: int, detail: str = "auto") -> int:
"""Calculate image tokens based on OpenAI's vision pricing.
For low detail: 85 tokens
For high detail/auto: 85 base tokens + 170 tokens per 512px tile
For original detail: patch-based pricing at the original resolution
"""
if detail == "low":
return 85
if detail == "original":
return _calculate_original_image_tokens(width, height)
if width > 2048 or height > 2048:
aspect_ratio = width / height
if width > height:
@@ -495,6 +526,16 @@ async def estimate_image_tokens_in_messages(messages: list) -> int:
if content_type not in ("image_url", "input_image"):
continue
# Responses-style ``input_image`` parts carry their detail and
# file_id as siblings of the image reference; route them through
# the input_image estimator so original detail / file_id are
# honored on the chat path too.
if content_type == "input_image":
total_image_tokens += await _estimate_input_image_tokens(
content_item
)
continue
image_url_data = content_item.get("image_url")
if not image_url_data:
continue
@@ -562,6 +603,69 @@ async def estimate_image_tokens_in_messages(messages: list) -> int:
return total_image_tokens
async def _estimate_input_image_tokens(item: dict) -> int:
"""Estimate tokens for a Responses API ``input_image`` item.
Honors the item-level ``detail``. The dimensions of ``file_id``
references can't be fetched here, so they get conservative
estimates: the max-size tile math for high/auto and the 30,000-patch
worst case (36,000 tokens) for original, so we never under-reserve.
"""
detail = item.get("detail") or "auto"
if image_url := item.get("image_url"):
if isinstance(image_url, dict):
image_url = image_url.get("url", "")
if isinstance(image_url, str) and image_url.startswith("data:image/"):
try:
_, base64_data = image_url.split(",", 1)
image_bytes = base64.b64decode(base64_data)
width, height = _get_image_dimensions(image_bytes)
return _calculate_image_tokens(width, height, detail)
except Exception as e:
logger.warning(
"Failed to process base64 image", extra={"error": str(e)}
)
return 85
# Remote URLs and file_id both have unfetchable dimensions here; fall
# through to the conservative estimates below.
if item.get("file_id") or item.get("image_url"):
if detail == "original":
return _MAX_ORIGINAL_IMAGE_TOKENS
# We can't fetch an uploaded file's dimensions here; assume the
# largest vision image so we don't under-reserve.
return _calculate_image_tokens(2048, 2048, detail)
return 0
async def estimate_image_tokens_from_input(input_data: Any) -> int:
"""Estimate total tokens for images embedded in a Responses API ``input``.
Recognizes ``input_image`` items at the top level of the input list and
inside ``message`` content parts.
"""
if not isinstance(input_data, list):
return 0
total_image_tokens = 0
for item in input_data:
if not isinstance(item, dict):
continue
if item.get("type") == "input_image":
total_image_tokens += await _estimate_input_image_tokens(item)
continue
content = item.get("content")
if not isinstance(content, list):
continue
for part in content:
if isinstance(part, dict) and part.get("type") == "input_image":
total_image_tokens += await _estimate_input_image_tokens(part)
return total_image_tokens
def create_error_response(
error_type: str,
message: str,
tests/unit/test_payment_helpers.py
+115
View File
@@ -288,3 +288,118 @@ async def test_discount_cannot_be_dodged_by_hiding_prompt_in_tools() -> None:
# Same prompt weight → at least the same reservation, never the floor.
assert cost >= cost_messages, where
assert cost > 1000, where
async def test_estimate_image_tokens_from_input_detail_and_file_id() -> None:
import base64
from io import BytesIO
from PIL import Image
from routstr.payment.helpers import estimate_image_tokens_from_input
# file_id: dimensions can't be fetched, so use a conservative max-size
# estimate (4 tiles for auto/high) and honor the detail sibling for low.
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "file_id": "file-1"}]
) == 85 + (170 * 4)
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "file_id": "file-1", "detail": "low"}]
) == 85
# image_url honors the sibling detail instead of always defaulting to auto.
image = Image.new("RGB", (512, 512), "red")
buffer = BytesIO()
image.save(buffer, format="JPEG")
data_url = "data:image/jpeg;base64," + base64.b64encode(buffer.getvalue()).decode()
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "image_url": data_url, "detail": "low"}]
) == 85
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "image_url": data_url, "detail": "high"}]
) == 85 + 170 # 512x512 = 1 tile
def test_calculate_image_tokens_original_detail() -> None:
from routstr.payment.helpers import _calculate_image_tokens
# Patch-based pricing: ceil(patches * 1.2) tokens at 32x32px patches.
assert _calculate_image_tokens(640, 640, "original") == 480 # 400 patches
assert _calculate_image_tokens(2048, 2048, "original") == 4_916 # 4,096 patches
# The same image on the tiled high-detail path caps at 765 tokens.
assert _calculate_image_tokens(2048, 2048, "high") == 765
# Above the 30,000-patch rejection limit the estimate is capped at
# 36,000 tokens (30,000 patches * 1.2).
assert _calculate_image_tokens(10_000, 10_000, "original") == 36_000
async def test_estimate_image_tokens_from_input_original_detail() -> None:
import base64
from io import BytesIO
from PIL import Image
from routstr.payment.helpers import estimate_image_tokens_from_input
image = Image.new("RGB", (2048, 2048), "red")
buffer = BytesIO()
image.save(buffer, format="JPEG")
data_url = "data:image/jpeg;base64," + base64.b64encode(buffer.getvalue()).decode()
# image_url: billed at the decoded original resolution (4,096 patches),
# not the 765-token tile cap.
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "image_url": data_url, "detail": "original"}]
) == 4_916
# file_id: dimensions unknown, so use the 30,000-patch worst case.
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "file_id": "file-1", "detail": "original"}]
) == 36_000
# Explicit null detail behaves like the auto default (tiled math).
assert await estimate_image_tokens_from_input(
[{"type": "input_image", "file_id": "file-1", "detail": None}]
) == 85 + (170 * 4)
async def test_estimate_image_tokens_in_messages_original_detail() -> None:
"""Chat Completions also accepts original detail via the nested dict."""
import base64
from io import BytesIO
from PIL import Image
from routstr.payment.helpers import estimate_image_tokens_in_messages
image = Image.new("RGB", (640, 640), "blue")
buffer = BytesIO()
image.save(buffer, format="JPEG")
data_url = "data:image/jpeg;base64," + base64.b64encode(buffer.getvalue()).decode()
messages = [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": data_url, "detail": "original"},
}
],
}
]
# 640x640 -> 20x20 = 400 patches -> ceil(400 * 1.2) = 480 tokens.
assert await estimate_image_tokens_in_messages(messages) == 480
# input_image parts inside messages honor their sibling detail and
# file_id through the Responses estimator as well.
messages = [
{
"role": "user",
"content": [
{"type": "input_image", "file_id": "file-1", "detail": "original"}
],
}
]
assert await estimate_image_tokens_in_messages(messages) == 36_000