feat(vision): optional region zoom crop on vision_analyze
Add an optional `region: [x1, y1, x2, y2]` parameter to vision_analyze (pixel coordinates in the ORIGINAL image space). The crop is applied with Pillow BEFORE the downscale/embed-cap pipeline, so the cropped region gets the full resolution budget — a zoom for reading small text or UI details after a full shot. - New `_crop_image_region` helper: clamps out-of-bounds coordinates to the image, rejects zero-area/inverted/malformed regions with an error naming the actual image dimensions so the model can retry sensibly. - Wired into both the native fast path (`_vision_analyze_native`) and the legacy aux-LLM path (`vision_analyze_tool`). - Schema gains one static optional param (byte-stable thereafter); the description documents the intended flow: full shot first, then zoom. - No region supplied = behavior unchanged (regression-guarded). Tests: tests/tools/test_vision_region.py (11 tests — crop applied, clamping, zero-area rejection with dims, malformed input, pre-downscale full-budget zoom, schema shape, handler pass-through, no-region unchanged). Widened one narrow fake_native stub in test_vision_tools.py to be kwargs-tolerant. Ported from: QwenLM/qwen-code zoom-image.ts (Apache-2.0)
This commit is contained in:
Teknium
parent
fe66596df3
commit
e166159f26
|
|
@ -0,0 +1,197 @@
|
|||
"""Tests for the optional region crop parameter on vision_analyze.
|
||||
|
||||
``region: [x1, y1, x2, y2]`` (pixel coords in the ORIGINAL image space) crops
|
||||
the image BEFORE the downscale pipeline so the cropped area gets the full
|
||||
resolution budget — a "zoom" for detail work after a full shot.
|
||||
|
||||
Ported from: QwenLM/qwen-code zoom-image.ts (Apache-2.0).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import io
|
||||
|
||||
import pytest
|
||||
|
||||
try:
|
||||
from PIL import Image
|
||||
except ImportError: # pragma: no cover
|
||||
Image = None
|
||||
|
||||
pytestmark = pytest.mark.skipif(Image is None, reason="Pillow not installed")
|
||||
|
||||
|
||||
def _make_png(path, width=100, height=50):
|
||||
img = Image.new("RGB", (width, height), (200, 30, 30))
|
||||
img.save(path, format="PNG")
|
||||
return path
|
||||
|
||||
|
||||
def _decoded_size(data_url: str):
|
||||
"""Return (w, h) of the image inside a base64 data URL."""
|
||||
b64 = data_url.split(",", 1)[1]
|
||||
with Image.open(io.BytesIO(base64.b64decode(b64))) as img:
|
||||
return img.size
|
||||
|
||||
|
||||
# ─── _crop_image_region helper ───────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCropImageRegion:
|
||||
def test_crop_applied(self, tmp_path):
|
||||
from tools.vision_tools import _crop_image_region
|
||||
|
||||
src = _make_png(tmp_path / "src.png", 100, 50)
|
||||
cropped_path, mime, err = _crop_image_region(src, [10, 10, 60, 40])
|
||||
assert err is None
|
||||
assert cropped_path is not None and cropped_path.exists()
|
||||
with Image.open(cropped_path) as img:
|
||||
assert img.size == (50, 30)
|
||||
|
||||
def test_out_of_bounds_clamped_to_image(self, tmp_path):
|
||||
from tools.vision_tools import _crop_image_region
|
||||
|
||||
src = _make_png(tmp_path / "src.png", 100, 50)
|
||||
cropped_path, mime, err = _crop_image_region(src, [-10, -10, 200, 200])
|
||||
assert err is None
|
||||
with Image.open(cropped_path) as img:
|
||||
assert img.size == (100, 50)
|
||||
|
||||
def test_zero_area_rejected_with_actual_dims_in_error(self, tmp_path):
|
||||
from tools.vision_tools import _crop_image_region
|
||||
|
||||
src = _make_png(tmp_path / "src.png", 100, 50)
|
||||
cropped_path, mime, err = _crop_image_region(src, [200, 200, 300, 300])
|
||||
assert cropped_path is None
|
||||
assert err is not None
|
||||
# Error must name the actual image dimensions so the model can retry.
|
||||
assert "100" in err and "50" in err
|
||||
|
||||
def test_inverted_coords_rejected_with_dims(self, tmp_path):
|
||||
from tools.vision_tools import _crop_image_region
|
||||
|
||||
src = _make_png(tmp_path / "src.png", 100, 50)
|
||||
cropped_path, mime, err = _crop_image_region(src, [60, 40, 10, 10])
|
||||
assert cropped_path is None
|
||||
assert "100" in err and "50" in err
|
||||
|
||||
def test_malformed_region_rejected(self, tmp_path):
|
||||
from tools.vision_tools import _crop_image_region
|
||||
|
||||
src = _make_png(tmp_path / "src.png", 100, 50)
|
||||
for bad in ([1, 2, 3], "10,10,60,40", [1, 2, 3, "x"], None):
|
||||
cropped_path, mime, err = _crop_image_region(src, bad)
|
||||
assert cropped_path is None
|
||||
assert err is not None
|
||||
|
||||
|
||||
# ─── native fast path with region ────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestNativePathRegion:
|
||||
def test_region_crops_before_embed(self, tmp_path):
|
||||
from tools.vision_tools import _vision_analyze_native
|
||||
|
||||
src = _make_png(tmp_path / "img.png", 100, 50)
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
_vision_analyze_native(str(src), "zoom", region=[10, 10, 60, 40])
|
||||
)
|
||||
assert isinstance(result, dict) and result.get("_multimodal") is True
|
||||
url = next(
|
||||
p["image_url"]["url"]
|
||||
for p in result["content"]
|
||||
if p.get("type") == "image_url"
|
||||
)
|
||||
assert _decoded_size(url) == (50, 30)
|
||||
|
||||
def test_no_region_behavior_unchanged(self, tmp_path):
|
||||
from tools.vision_tools import _vision_analyze_native
|
||||
|
||||
src = _make_png(tmp_path / "img.png", 100, 50)
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
_vision_analyze_native(str(src), "full shot")
|
||||
)
|
||||
assert isinstance(result, dict) and result.get("_multimodal") is True
|
||||
url = next(
|
||||
p["image_url"]["url"]
|
||||
for p in result["content"]
|
||||
if p.get("type") == "image_url"
|
||||
)
|
||||
assert _decoded_size(url) == (100, 50)
|
||||
|
||||
def test_zero_area_region_returns_error_with_dims(self, tmp_path):
|
||||
import json
|
||||
|
||||
from tools.vision_tools import _vision_analyze_native
|
||||
|
||||
src = _make_png(tmp_path / "img.png", 100, 50)
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
_vision_analyze_native(str(src), "zoom", region=[500, 500, 600, 600])
|
||||
)
|
||||
assert isinstance(result, str)
|
||||
payload = json.loads(result)
|
||||
assert payload.get("success") is False
|
||||
msg = json.dumps(payload)
|
||||
assert "100" in msg and "50" in msg
|
||||
|
||||
def test_crop_applied_before_downscale_gets_full_budget(self, tmp_path):
|
||||
"""The crop happens BEFORE _resize_image_for_vision, so a small region
|
||||
of a huge image survives at native resolution instead of being
|
||||
downscaled with the rest."""
|
||||
from tools.vision_tools import _EMBED_MAX_DIMENSION, _vision_analyze_native
|
||||
|
||||
# Taller than the 7900px embed cap — full shot would be downscaled.
|
||||
big = tmp_path / "big.png"
|
||||
Image.new("RGB", (200, _EMBED_MAX_DIMENSION + 500), (0, 100, 0)).save(
|
||||
big, format="PNG"
|
||||
)
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
_vision_analyze_native(str(big), "zoom", region=[0, 0, 200, 300])
|
||||
)
|
||||
assert isinstance(result, dict) and result.get("_multimodal") is True
|
||||
url = next(
|
||||
p["image_url"]["url"]
|
||||
for p in result["content"]
|
||||
if p.get("type") == "image_url"
|
||||
)
|
||||
# Region kept at native resolution — no downscale applied to the crop.
|
||||
assert _decoded_size(url) == (200, 300)
|
||||
|
||||
|
||||
# ─── schema + handler wiring ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSchemaAndHandler:
|
||||
def test_schema_declares_optional_region(self):
|
||||
from tools.vision_tools import VISION_ANALYZE_SCHEMA
|
||||
|
||||
props = VISION_ANALYZE_SCHEMA["parameters"]["properties"]
|
||||
assert "region" in props
|
||||
assert props["region"]["type"] == "array"
|
||||
assert "region" not in VISION_ANALYZE_SCHEMA["parameters"]["required"]
|
||||
# Description must document original-image pixel space.
|
||||
assert "original" in props["region"]["description"].lower()
|
||||
|
||||
def test_handler_passes_region_to_native_path(self, tmp_path, monkeypatch):
|
||||
from tools import vision_tools
|
||||
from tools.vision_tools import _handle_vision_analyze
|
||||
|
||||
src = _make_png(tmp_path / "img.png", 100, 50)
|
||||
seen = {}
|
||||
|
||||
async def _fake_native(image_url, question, task_id=None, region=None):
|
||||
seen["region"] = region
|
||||
return {"_multimodal": True, "content": []}
|
||||
|
||||
monkeypatch.setattr(vision_tools, "_vision_analyze_native", _fake_native)
|
||||
monkeypatch.setattr(
|
||||
vision_tools, "_should_use_native_vision_fast_path", lambda: True
|
||||
)
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
_handle_vision_analyze(
|
||||
{"image_url": str(src), "question": "q", "region": [1, 2, 30, 40]}
|
||||
)
|
||||
)
|
||||
assert seen["region"] == [1, 2, 30, 40]
|
||||
|
|
@ -957,7 +957,7 @@ class TestVisionCpuBurstCap:
|
|||
enc_inflight -= 1
|
||||
return "data:image/jpeg;base64,AAAA"
|
||||
|
||||
async def fake_native(image_url, question, task_id=None):
|
||||
async def fake_native(image_url, question, task_id=None, **_kw):
|
||||
nonlocal calls_inflight, calls_peak
|
||||
calls_inflight += 1
|
||||
calls_peak = max(calls_peak, calls_inflight)
|
||||
|
|
|
|||
|
|
@ -678,6 +678,70 @@ def _image_exceeds_dimension(image_path: Path, max_dimension: int) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def _crop_image_region(
|
||||
image_path: Path,
|
||||
region: Any,
|
||||
) -> tuple[Optional[Path], Optional[str], Optional[str]]:
|
||||
"""Crop ``image_path`` to ``region`` = [x1, y1, x2, y2] (original-image pixels).
|
||||
|
||||
Applied BEFORE :func:`_resize_image_for_vision` so the cropped area gets
|
||||
the full downscale resolution budget — a "zoom" into a detail region.
|
||||
Coordinates are clamped to the image bounds; a region that clamps to zero
|
||||
area (or is inverted/malformed) is rejected with an error naming the
|
||||
actual image dimensions so the caller can retry with sensible values.
|
||||
|
||||
Ported from: QwenLM/qwen-code zoom-image.ts (Apache-2.0).
|
||||
|
||||
Returns:
|
||||
(cropped_temp_path, out_mime, None) on success — the caller owns
|
||||
cleanup of the temp file — or (None, None, error_message) on failure.
|
||||
"""
|
||||
try:
|
||||
from PIL import Image
|
||||
except ImportError:
|
||||
return None, None, (
|
||||
"region cropping requires Pillow (`pip install Pillow`); "
|
||||
"retry without the region parameter."
|
||||
)
|
||||
|
||||
if (
|
||||
not isinstance(region, (list, tuple))
|
||||
or len(region) != 4
|
||||
or not all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in region)
|
||||
):
|
||||
return None, None, (
|
||||
"Invalid region: expected [x1, y1, x2, y2] as four numbers "
|
||||
"(pixel coordinates in the original image)."
|
||||
)
|
||||
|
||||
try:
|
||||
with Image.open(image_path) as img:
|
||||
width, height = img.size
|
||||
x1, y1, x2, y2 = (int(v) for v in region)
|
||||
# Clamp to image bounds.
|
||||
cx1 = max(0, min(x1, width))
|
||||
cy1 = max(0, min(y1, height))
|
||||
cx2 = max(0, min(x2, width))
|
||||
cy2 = max(0, min(y2, height))
|
||||
if cx2 <= cx1 or cy2 <= cy1:
|
||||
return None, None, (
|
||||
f"Invalid region [{x1}, {y1}, {x2}, {y2}]: crops to zero "
|
||||
f"area after clamping to the image bounds. The image is "
|
||||
f"{width}x{height} px — pick x1<x2 and y1<y2 inside "
|
||||
f"[0, 0, {width}, {height}]."
|
||||
)
|
||||
cropped = img.crop((cx1, cy1, cx2, cy2))
|
||||
out_path = image_path.with_name(
|
||||
f"{image_path.stem}_region_{uuid.uuid4().hex[:8]}.png"
|
||||
)
|
||||
if cropped.mode not in ("RGB", "RGBA", "L", "LA", "P"):
|
||||
cropped = cropped.convert("RGB")
|
||||
cropped.save(out_path, format="PNG")
|
||||
return out_path, "image/png", None
|
||||
except Exception as exc:
|
||||
return None, None, f"Failed to crop region: {exc}"
|
||||
|
||||
|
||||
def _resize_image_for_vision(image_path: Path, mime_type: Optional[str] = None,
|
||||
max_base64_bytes: int = _RESIZE_TARGET_BYTES,
|
||||
max_dimension: Optional[int] = None) -> str:
|
||||
|
|
@ -1013,6 +1077,7 @@ async def _vision_analyze_native(
|
|||
image_url: str,
|
||||
question: str,
|
||||
task_id: Optional[str] = None,
|
||||
region: Optional[list] = None,
|
||||
) -> Any:
|
||||
"""Fast path for vision-capable main models.
|
||||
|
||||
|
|
@ -1082,6 +1147,24 @@ async def _vision_analyze_native(
|
|||
should_cleanup = True
|
||||
image_size_bytes = temp_image_path.stat().st_size
|
||||
|
||||
# Optional region zoom: crop BEFORE the downscale/embed-cap pipeline
|
||||
# so the cropped area gets the full resolution budget.
|
||||
if region is not None:
|
||||
cropped_path, cropped_mime, crop_err = await asyncio.to_thread(
|
||||
_crop_image_region, temp_image_path, region,
|
||||
)
|
||||
if crop_err or cropped_path is None:
|
||||
return tool_error(crop_err or "Region crop failed.", success=False)
|
||||
if should_cleanup and temp_image_path.exists():
|
||||
try:
|
||||
temp_image_path.unlink()
|
||||
except Exception:
|
||||
pass
|
||||
temp_image_path = cropped_path
|
||||
detected_mime_type = cropped_mime
|
||||
should_cleanup = True
|
||||
image_size_bytes = temp_image_path.stat().st_size
|
||||
|
||||
image_data_url = await _run_encode_on_cpu_executor(
|
||||
_image_to_base64_data_url,
|
||||
temp_image_path, mime_type=detected_mime_type,
|
||||
|
|
@ -1145,6 +1228,7 @@ async def vision_analyze_tool(
|
|||
user_prompt: str,
|
||||
model: str = None,
|
||||
task_id: Optional[str] = None,
|
||||
region: Optional[list] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Analyze an image from a URL or local file path using vision AI.
|
||||
|
|
@ -1250,6 +1334,23 @@ async def vision_analyze_tool(
|
|||
temp_image_path = normalized_path
|
||||
should_cleanup = True
|
||||
|
||||
# Optional region zoom: crop BEFORE the encode/downscale pipeline so
|
||||
# the cropped area gets the full resolution budget.
|
||||
if region is not None:
|
||||
cropped_path, cropped_mime, crop_err = await asyncio.to_thread(
|
||||
_crop_image_region, temp_image_path, region,
|
||||
)
|
||||
if crop_err or cropped_path is None:
|
||||
raise ValueError(crop_err or "Region crop failed.")
|
||||
if should_cleanup and temp_image_path.exists():
|
||||
try:
|
||||
temp_image_path.unlink()
|
||||
except Exception:
|
||||
pass
|
||||
temp_image_path = cropped_path
|
||||
detected_mime_type = cropped_mime
|
||||
should_cleanup = True
|
||||
|
||||
# Convert image to base64 — send at full resolution first.
|
||||
# If the provider rejects it as too large, we auto-resize and retry.
|
||||
# Offloaded to the bounded vision CPU executor so a fan-out of encodes
|
||||
|
|
@ -1544,6 +1645,20 @@ VISION_ANALYZE_SCHEMA = {
|
|||
"question": {
|
||||
"type": "string",
|
||||
"description": "Your specific question or request about the image. Optional context the model uses on the next turn after seeing the image."
|
||||
},
|
||||
"region": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
"minItems": 4,
|
||||
"maxItems": 4,
|
||||
"description": (
|
||||
"Optional [x1, y1, x2, y2] crop region in pixel coordinates "
|
||||
"of the ORIGINAL image, applied before any downscaling so "
|
||||
"the region keeps full resolution. Intended flow: load the "
|
||||
"full image first, then call again with a region to zoom "
|
||||
"into a detail (small text, UI element, fine print). "
|
||||
"Coordinates are clamped to the image bounds."
|
||||
)
|
||||
}
|
||||
},
|
||||
"required": ["image_url", "question"]
|
||||
|
|
@ -1554,6 +1669,7 @@ VISION_ANALYZE_SCHEMA = {
|
|||
async def _handle_vision_analyze(args: Dict[str, Any], **kw: Any) -> str:
|
||||
image_url = args.get("image_url", "")
|
||||
question = args.get("question", "")
|
||||
region = args.get("region")
|
||||
task_id = kw.get("task_id")
|
||||
|
||||
# The fan-out cap lives inside the encode/resize step (offloaded to the
|
||||
|
|
@ -1569,7 +1685,7 @@ async def _handle_vision_analyze(args: Dict[str, Any], **kw: Any) -> str:
|
|||
# information loss, no extra latency.
|
||||
if _should_use_native_vision_fast_path():
|
||||
logger.info("vision_analyze: native fast path")
|
||||
return await _vision_analyze_native(image_url, question, task_id=task_id)
|
||||
return await _vision_analyze_native(image_url, question, task_id=task_id, region=region)
|
||||
|
||||
# Legacy path: aux LLM describes the image and we return its text.
|
||||
full_prompt = (
|
||||
|
|
@ -1588,7 +1704,7 @@ async def _handle_vision_analyze(args: Dict[str, Any], **kw: Any) -> str:
|
|||
pass
|
||||
if not model:
|
||||
model = os.getenv("AUXILIARY_VISION_MODEL", "").strip() or None
|
||||
return await vision_analyze_tool(image_url, full_prompt, model, task_id=task_id)
|
||||
return await vision_analyze_tool(image_url, full_prompt, model, task_id=task_id, region=region)
|
||||
|
||||
|
||||
registry.register(
|
||||
|
|
|
|||
Loading…
Reference in New Issue