434 lines
15 KiB
Python
434 lines
15 KiB
Python
"""
|
|
Tests for Anthropic-compatible image input via the /v1/messages endpoint.
|
|
|
|
python3 anthorpic_api/test/manual/vlm/test_anthropic_vision.py
|
|
"""
|
|
|
|
import json
|
|
import unittest
|
|
|
|
import pybase64
|
|
import requests
|
|
|
|
from sglang.srt.utils import kill_process_tree
|
|
from sglang.test.test_utils import (
|
|
DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST,
|
|
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
|
DEFAULT_URL_FOR_TEST,
|
|
CustomTestCase,
|
|
popen_launch_server,
|
|
)
|
|
|
|
IMAGE_MAN_IRONING_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/images/man_ironing_on_back_of_suv.png"
|
|
IMAGE_SGL_LOGO_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/images/sgl_logo.png"
|
|
|
|
|
|
def _fetch_image_base64(url: str) -> str:
|
|
"""Download an image and return its base64-encoded content."""
|
|
resp = requests.get(url, timeout=30)
|
|
resp.raise_for_status()
|
|
return pybase64.b64encode(resp.content).decode("utf-8")
|
|
|
|
|
|
class TestAnthropicVision(CustomTestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.model = DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST
|
|
cls.base_url = DEFAULT_URL_FOR_TEST
|
|
cls.api_key = "sk-123456"
|
|
cls.process = popen_launch_server(
|
|
cls.model,
|
|
cls.base_url,
|
|
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
|
api_key=cls.api_key,
|
|
other_args=[
|
|
"--trust-remote-code",
|
|
"--enable-multimodal",
|
|
"--cuda-graph-max-bs=4",
|
|
],
|
|
)
|
|
cls.messages_url = cls.base_url + "/v1/messages"
|
|
# Pre-fetch the image as base64 once for all tests
|
|
cls.image_base64 = _fetch_image_base64(IMAGE_MAN_IRONING_URL)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
kill_process_tree(cls.process.pid)
|
|
|
|
def _make_request(self, payload, stream=False):
|
|
"""Send a request to the /v1/messages endpoint."""
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
}
|
|
return requests.post(
|
|
self.messages_url,
|
|
headers=headers,
|
|
json=payload,
|
|
stream=stream,
|
|
)
|
|
|
|
def _parse_sse_events(self, response):
|
|
"""Parse SSE events from a streaming response."""
|
|
events = []
|
|
for line in response.iter_lines(decode_unicode=True):
|
|
if not line:
|
|
continue
|
|
if line.startswith("data: "):
|
|
data_str = line[6:].strip()
|
|
if data_str == "[DONE]":
|
|
continue
|
|
try:
|
|
events.append(json.loads(data_str))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return events
|
|
|
|
def _verify_ironing_image_content(self, text):
|
|
"""Verify the response text describes the man-ironing-on-SUV image."""
|
|
text_lower = text.lower()
|
|
self.assertTrue(
|
|
any(w in text_lower for w in ["man", "person", "driver", "someone"]),
|
|
f"Expected mention of a person, got: {text}",
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
w in text_lower
|
|
for w in ["cab", "taxi", "suv", "vehicle", "car", "trunk", "back"]
|
|
),
|
|
f"Expected mention of a vehicle, got: {text}",
|
|
)
|
|
self.assertTrue(
|
|
any(
|
|
w in text_lower
|
|
for w in ["iron", "hang", "cloth", "holding", "laundry", "shirt"]
|
|
),
|
|
f"Expected mention of ironing/clothes, got: {text}",
|
|
)
|
|
|
|
# ---- Base64 image tests ----
|
|
|
|
def test_single_image_base64(self):
|
|
"""Test sending a single base64 image in Anthropic format."""
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "Describe this image in a sentence.",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
resp = self._make_request(payload)
|
|
self.assertEqual(resp.status_code, 200, f"Response: {resp.text}")
|
|
|
|
body = resp.json()
|
|
self.assertEqual(body["type"], "message")
|
|
self.assertEqual(body["role"], "assistant")
|
|
self.assertTrue(len(body["content"]) > 0)
|
|
self.assertEqual(body["content"][0]["type"], "text")
|
|
text = body["content"][0]["text"]
|
|
self.assertIsInstance(text, str)
|
|
self.assertTrue(len(text) > 0, "Response text should not be empty")
|
|
|
|
# Verify response describes the image content
|
|
self._verify_ironing_image_content(text)
|
|
|
|
# Verify usage
|
|
self.assertIn("usage", body)
|
|
self.assertGreater(body["usage"]["input_tokens"], 0)
|
|
self.assertGreater(body["usage"]["output_tokens"], 0)
|
|
|
|
# Verify id format
|
|
self.assertTrue(
|
|
body["id"].startswith("msg_"),
|
|
f"ID should start with 'msg_', got: {body['id']}",
|
|
)
|
|
|
|
def test_single_image_url(self):
|
|
"""Test sending an image via URL (converted to data URI internally)."""
|
|
# Anthropic format uses source.type="base64", but we test the data URI path
|
|
# by pre-encoding the URL image as base64
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "What objects do you see in this image?",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
"temperature": 0,
|
|
}
|
|
resp = self._make_request(payload)
|
|
self.assertEqual(resp.status_code, 200, f"Response: {resp.text}")
|
|
|
|
body = resp.json()
|
|
self.assertEqual(body["type"], "message")
|
|
self.assertTrue(len(body["content"]) > 0)
|
|
text = body["content"][0]["text"]
|
|
self.assertIsInstance(text, str)
|
|
self.assertTrue(len(text) > 0)
|
|
|
|
# Verify response describes the image content
|
|
self._verify_ironing_image_content(text)
|
|
|
|
def test_image_with_text_blocks(self):
|
|
"""Test image combined with multiple text content blocks."""
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": "Look at this image carefully.",
|
|
},
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "Describe what you see in one sentence.",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
resp = self._make_request(payload)
|
|
self.assertEqual(resp.status_code, 200, f"Response: {resp.text}")
|
|
|
|
body = resp.json()
|
|
self.assertEqual(body["type"], "message")
|
|
self.assertTrue(len(body["content"]) > 0)
|
|
self.assertEqual(body["content"][0]["type"], "text")
|
|
text = body["content"][0]["text"]
|
|
self.assertTrue(len(text) > 0)
|
|
|
|
# Verify response describes the image content
|
|
self._verify_ironing_image_content(text)
|
|
|
|
# ---- Streaming with image ----
|
|
|
|
def test_image_stream(self):
|
|
"""Test streaming response with image input."""
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"stream": True,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "Describe this image briefly.",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
resp = self._make_request(payload, stream=True)
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn("text/event-stream", resp.headers.get("content-type", ""))
|
|
|
|
events = self._parse_sse_events(resp)
|
|
event_types = [e["type"] for e in events]
|
|
|
|
# Verify event sequence
|
|
self.assertIn("message_start", event_types)
|
|
self.assertIn("message_stop", event_types)
|
|
self.assertEqual(events[0]["type"], "message_start")
|
|
|
|
# Verify we got content
|
|
content_deltas = [e for e in events if e["type"] == "content_block_delta"]
|
|
self.assertTrue(len(content_deltas) > 0, "Expected content_block_delta events")
|
|
|
|
# Reconstruct text
|
|
full_text = "".join(
|
|
e["delta"]["text"]
|
|
for e in content_deltas
|
|
if e["delta"].get("type") == "text_delta"
|
|
)
|
|
self.assertTrue(len(full_text) > 0, "Streamed text should not be empty")
|
|
|
|
# Verify streamed response describes the image content
|
|
self._verify_ironing_image_content(full_text)
|
|
|
|
# Verify message_delta has stop_reason
|
|
message_deltas = [e for e in events if e["type"] == "message_delta"]
|
|
self.assertTrue(len(message_deltas) > 0)
|
|
self.assertIn("stop_reason", message_deltas[-1]["delta"])
|
|
|
|
# ---- Multi-image tests ----
|
|
|
|
def test_multi_image(self):
|
|
"""Test sending multiple images in a single message."""
|
|
logo_base64 = _fetch_image_base64(IMAGE_SGL_LOGO_URL)
|
|
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": logo_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "How many images do you see? Describe each briefly.",
|
|
},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
resp = self._make_request(payload)
|
|
self.assertEqual(resp.status_code, 200, f"Response: {resp.text}")
|
|
|
|
body = resp.json()
|
|
self.assertEqual(body["type"], "message")
|
|
self.assertTrue(len(body["content"]) > 0)
|
|
text = body["content"][0]["text"]
|
|
self.assertIsInstance(text, str)
|
|
self.assertTrue(len(text) > 0)
|
|
|
|
# ---- Multi-turn with image ----
|
|
|
|
def test_multi_turn_with_image(self):
|
|
"""Test multi-turn conversation with image context."""
|
|
# First turn: send image
|
|
payload = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "What is in this image?",
|
|
},
|
|
],
|
|
},
|
|
],
|
|
"temperature": 0,
|
|
}
|
|
resp1 = self._make_request(payload)
|
|
self.assertEqual(resp1.status_code, 200, f"Response: {resp1.text}")
|
|
body1 = resp1.json()
|
|
first_response_text = body1["content"][0]["text"]
|
|
|
|
# Verify first turn describes the image
|
|
self._verify_ironing_image_content(first_response_text)
|
|
|
|
# Second turn: ask follow-up without re-sending image
|
|
payload2 = {
|
|
"model": self.model,
|
|
"max_tokens": 128,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": self.image_base64,
|
|
},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": "What is in this image?",
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"role": "assistant",
|
|
"content": first_response_text,
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": "Can you describe the colors you see?",
|
|
},
|
|
],
|
|
"temperature": 0,
|
|
}
|
|
resp2 = self._make_request(payload2)
|
|
self.assertEqual(resp2.status_code, 200, f"Response: {resp2.text}")
|
|
|
|
body2 = resp2.json()
|
|
self.assertEqual(body2["type"], "message")
|
|
self.assertTrue(len(body2["content"]) > 0)
|
|
self.assertEqual(body2["content"][0]["type"], "text")
|
|
self.assertTrue(len(body2["content"][0]["text"]) > 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|