mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
feat(tencent): add Seedream image generation and editing through TokenHub
This commit is contained in:
parent
11b5d1a817
commit
b4299f6423
14 changed files with 617 additions and 44 deletions
|
|
@ -403,6 +403,7 @@ def image_generation(
|
|||
litellm.LlmProviders.QWENCLOUD,
|
||||
litellm.LlmProviders.QWEN_AI_PLATFORM,
|
||||
litellm.LlmProviders.EDENAI,
|
||||
litellm.LlmProviders.TENCENT,
|
||||
):
|
||||
if image_generation_config is None:
|
||||
raise ValueError(f"image generation config is not supported for {custom_llm_provider}")
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
import base64
|
||||
import os
|
||||
from collections.abc import Collection, Mapping
|
||||
from io import BufferedReader, BytesIO
|
||||
from typing import Any, Final, cast, get_type_hints
|
||||
from pathlib import Path
|
||||
from typing import Any, Final, Protocol, cast, get_type_hints, runtime_checkable
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.token_counter import get_image_type
|
||||
|
|
@ -145,3 +148,36 @@ class ImageEditRequestUtils:
|
|||
except Exception:
|
||||
# If anything goes wrong, default to PNG
|
||||
return FILE_MIME_TYPES[FileType.PNG]
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class _SeekableBinaryReader(Protocol):
|
||||
def tell(self) -> int: ...
|
||||
|
||||
def seek(self, offset: int) -> int: ...
|
||||
|
||||
def read(self) -> bytes: ...
|
||||
|
||||
|
||||
def _read_image_bytes(image: object) -> bytes:
|
||||
if isinstance(image, bytes):
|
||||
return image
|
||||
if isinstance(image, tuple):
|
||||
return _read_image_bytes(image[1])
|
||||
if isinstance(image, os.PathLike):
|
||||
return Path(image).read_bytes()
|
||||
if isinstance(image, _SeekableBinaryReader):
|
||||
position: Final = image.tell()
|
||||
image.seek(0)
|
||||
data: Final = image.read()
|
||||
image.seek(position)
|
||||
return data
|
||||
raise ValueError(f"Unsupported image type for image edit: {type(image).__name__}")
|
||||
|
||||
|
||||
def image_to_data_url(image: object) -> str:
|
||||
if isinstance(image, str):
|
||||
return image
|
||||
image_bytes: Final = _read_image_bytes(image)
|
||||
mime_type: Final = ImageEditRequestUtils.get_image_content_type(image_bytes)
|
||||
return f"data:{mime_type};base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
|
|
|
|||
|
|
@ -4,12 +4,13 @@ from typing import Final
|
|||
|
||||
from httpx._types import RequestFiles
|
||||
|
||||
from litellm.images.utils import image_to_data_url
|
||||
from litellm.secret_managers.main import get_secret_str
|
||||
from litellm.types.images.main import ImageEditOptionalRequestParams
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.utils import FileTypes
|
||||
|
||||
from .transformation import DEFAULT_BASE_URL, FalAIImageEditConfig, to_data_url
|
||||
from .transformation import DEFAULT_BASE_URL, FalAIImageEditConfig
|
||||
|
||||
FLUX_LORA_DEPTH_ENDPOINT: Final[str] = "fal-ai/flux-lora-depth"
|
||||
SUPPORTED_OPENAI_PARAMS: Final[tuple[str, ...]] = ("n", "size")
|
||||
|
|
@ -67,7 +68,7 @@ class FalAIFluxLoraDepthEditConfig(FalAIImageEditConfig):
|
|||
)
|
||||
request_body: Final[dict[str, object]] = { # mutable-ok: base class contract returns a dict
|
||||
"prompt": prompt,
|
||||
"image_url": to_data_url(next(iter(images))),
|
||||
"image_url": image_to_data_url(next(iter(images))),
|
||||
**provider_params,
|
||||
}
|
||||
return request_body, ()
|
||||
|
|
|
|||
|
|
@ -1,15 +1,12 @@
|
|||
import base64
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Final, Protocol, runtime_checkable
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
import httpx
|
||||
from httpx._types import RequestFiles
|
||||
from pydantic import ConfigDict, TypeAdapter
|
||||
|
||||
from litellm.images.utils import ImageEditRequestUtils
|
||||
from litellm.images.utils import image_to_data_url
|
||||
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
|
||||
from litellm.llms.fal_ai.image_generation.gpt_image_2_transformation import (
|
||||
map_gpt_image_quality,
|
||||
|
|
@ -39,39 +36,6 @@ PARAM_TRANSLATION: Final[Mapping[str, str]] = MappingProxyType(
|
|||
)
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class _SeekableBinaryReader(Protocol):
|
||||
def tell(self) -> int: ...
|
||||
|
||||
def seek(self, offset: int) -> int: ...
|
||||
|
||||
def read(self) -> bytes: ...
|
||||
|
||||
|
||||
def _read_image_bytes(image: object) -> bytes:
|
||||
if isinstance(image, bytes):
|
||||
return image
|
||||
if isinstance(image, tuple):
|
||||
return _read_image_bytes(image[1])
|
||||
if isinstance(image, os.PathLike):
|
||||
return Path(image).read_bytes()
|
||||
if isinstance(image, _SeekableBinaryReader):
|
||||
position: Final = image.tell()
|
||||
image.seek(0)
|
||||
data: Final = image.read()
|
||||
image.seek(position)
|
||||
return data
|
||||
raise ValueError(f"Unsupported image type for Fal AI image edit: {type(image).__name__}")
|
||||
|
||||
|
||||
def to_data_url(image: object) -> str:
|
||||
if isinstance(image, str):
|
||||
return image
|
||||
image_bytes: Final = _read_image_bytes(image)
|
||||
mime_type: Final = ImageEditRequestUtils.get_image_content_type(image_bytes)
|
||||
return f"data:{mime_type};base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
|
||||
|
||||
def _first(value: object) -> object:
|
||||
return value[0] if isinstance(value, list) and value else value
|
||||
|
||||
|
|
@ -146,14 +110,14 @@ class FalAIImageEditConfig(BaseImageEditConfig):
|
|||
raise ValueError("Fal AI image edit requires at least one input image")
|
||||
mask: Final = _first(image_edit_optional_request_params.get("mask"))
|
||||
mask_field: Final[Mapping[str, str]] = (
|
||||
MappingProxyType({"mask_url": to_data_url(mask)}) if mask is not None else MappingProxyType({})
|
||||
MappingProxyType({"mask_url": image_to_data_url(mask)}) if mask is not None else MappingProxyType({})
|
||||
)
|
||||
provider_params: Final[Mapping[str, object]] = MappingProxyType(
|
||||
{key: value for key, value in image_edit_optional_request_params.items() if key != "mask"}
|
||||
)
|
||||
request_body: Final[dict[str, object]] = { # mutable-ok: base class contract returns a dict
|
||||
"prompt": prompt,
|
||||
"image_urls": tuple(to_data_url(img) for img in images),
|
||||
"image_urls": tuple(image_to_data_url(img) for img in images),
|
||||
**mask_field,
|
||||
**provider_params,
|
||||
}
|
||||
|
|
|
|||
0
litellm/llms/tencent/image_edit/__init__.py
Normal file
0
litellm/llms/tencent/image_edit/__init__.py
Normal file
99
litellm/llms/tencent/image_edit/transformation.py
Normal file
99
litellm/llms/tencent/image_edit/transformation.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
from collections.abc import Mapping
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
import httpx
|
||||
from httpx._types import RequestFiles
|
||||
from pydantic import TypeAdapter
|
||||
from typing_extensions import override
|
||||
|
||||
from litellm.images.utils import image_to_data_url
|
||||
from litellm.llms.base_llm.chat.transformation import BaseLLMException
|
||||
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
|
||||
from litellm.llms.tencent.image_generation.transformation import (
|
||||
TOKENHUB_IMAGE_PARAMS,
|
||||
tokenhub_image_headers,
|
||||
tokenhub_image_params,
|
||||
tokenhub_image_response,
|
||||
tokenhub_image_url,
|
||||
)
|
||||
from litellm.types.images.main import ImageEditOptionalRequestParams
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.utils import ImageResponse
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
|
||||
_IMAGES: Final = TypeAdapter(tuple[object, ...])
|
||||
|
||||
|
||||
class TencentImageEditConfig(BaseImageEditConfig):
|
||||
@override
|
||||
def get_supported_openai_params(self, model: str) -> list[str]: # mutable-ok: mirrors override contract
|
||||
return list(TOKENHUB_IMAGE_PARAMS)
|
||||
|
||||
@override
|
||||
def map_openai_params(
|
||||
self,
|
||||
image_edit_optional_params: ImageEditOptionalRequestParams,
|
||||
model: str,
|
||||
drop_params: bool,
|
||||
) -> dict[str, object]: # mutable-ok: mirrors override contract
|
||||
return dict(tokenhub_image_params(image_edit_optional_params, model, drop_params))
|
||||
|
||||
@override
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: Mapping[str, str],
|
||||
model: str,
|
||||
api_key: str | None = None,
|
||||
litellm_params: Mapping[str, object] | None = None,
|
||||
api_base: str | None = None,
|
||||
) -> dict[str, str]: # mutable-ok: mirrors override contract
|
||||
return dict(tokenhub_image_headers(headers, api_key, model))
|
||||
|
||||
@override
|
||||
def get_complete_url(
|
||||
self,
|
||||
model: str,
|
||||
api_base: str | None,
|
||||
litellm_params: Mapping[str, object],
|
||||
) -> str:
|
||||
return tokenhub_image_url(api_base)
|
||||
|
||||
@override
|
||||
def use_multipart_form_data(self) -> bool:
|
||||
return False
|
||||
|
||||
@override
|
||||
def transform_image_edit_request(
|
||||
self,
|
||||
model: str,
|
||||
prompt: str | None,
|
||||
image: object,
|
||||
image_edit_optional_request_params: Mapping[str, object],
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: Mapping[str, str],
|
||||
) -> tuple[dict[str, object], RequestFiles]: # mutable-ok: mirrors override contract
|
||||
images: Final = _IMAGES.validate_python(image) if isinstance(image, list) else (image,)
|
||||
return {
|
||||
"model": model,
|
||||
"prompt": prompt,
|
||||
"images": tuple(image_to_data_url(img) for img in images if img is not None),
|
||||
**image_edit_optional_request_params,
|
||||
}, ()
|
||||
|
||||
@override
|
||||
def transform_image_edit_response(
|
||||
self,
|
||||
model: str,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: "LiteLLMLoggingObj",
|
||||
) -> ImageResponse:
|
||||
image_response: Final = tokenhub_image_response(raw_response)
|
||||
if image_response is None:
|
||||
raise BaseLLMException(
|
||||
status_code=raw_response.status_code,
|
||||
message=raw_response.text,
|
||||
headers=raw_response.headers,
|
||||
)
|
||||
return image_response
|
||||
0
litellm/llms/tencent/image_generation/__init__.py
Normal file
0
litellm/llms/tencent/image_generation/__init__.py
Normal file
188
litellm/llms/tencent/image_generation/transformation.py
Normal file
188
litellm/llms/tencent/image_generation/transformation.py
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
from collections.abc import Mapping, Sequence
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
import httpx
|
||||
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
|
||||
from typing_extensions import override
|
||||
|
||||
from litellm.exceptions import AuthenticationError, UnsupportedParamsError
|
||||
from litellm.llms.base_llm.chat.transformation import BaseLLMException
|
||||
from litellm.llms.base_llm.image_generation.transformation import (
|
||||
BaseImageGenerationConfig,
|
||||
)
|
||||
from litellm.secret_managers.main import get_secret_str
|
||||
from litellm.types.llms.openai import (
|
||||
AllMessageValues,
|
||||
OpenAIImageGenerationOptionalParams,
|
||||
)
|
||||
from litellm.types.utils import (
|
||||
ImageObject,
|
||||
ImageResponse,
|
||||
ImageUsage,
|
||||
ImageUsageInputTokensDetails,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
|
||||
TOKENHUB_API_BASE: Final = "https://tokenhub-intl.tencentcloudmaas.com/v1"
|
||||
TOKENHUB_IMAGE_PATH: Final = "/wand/si-image/generation"
|
||||
TOKENHUB_IMAGE_PARAMS: Final[tuple[OpenAIImageGenerationOptionalParams, ...]] = (
|
||||
"n",
|
||||
"size",
|
||||
"response_format",
|
||||
"output_format",
|
||||
"background",
|
||||
)
|
||||
|
||||
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
|
||||
|
||||
|
||||
class _TokenHubImage(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
url: str | None = None
|
||||
b64_json: str | None = None
|
||||
|
||||
|
||||
class _TokenHubUsage(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
total_tokens: int
|
||||
|
||||
|
||||
class _TokenHubImageResponse(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, hide_input_in_errors=True)
|
||||
|
||||
created: int
|
||||
data: tuple[_TokenHubImage, ...]
|
||||
tokenhub_usage: _TokenHubUsage | None = None
|
||||
|
||||
|
||||
def tokenhub_image_url(api_base: str | None) -> str:
|
||||
base: Final = (api_base or get_secret_str("TENCENT_API_BASE") or TOKENHUB_API_BASE).rstrip("/")
|
||||
versioned_base: Final = base if base.endswith("/v1") else f"{base}/v1"
|
||||
return f"{versioned_base}{TOKENHUB_IMAGE_PATH}"
|
||||
|
||||
|
||||
def tokenhub_image_headers(headers: Mapping[str, str], api_key: str | None, model: str) -> Mapping[str, str]:
|
||||
resolved_api_key: Final = api_key or get_secret_str("TENCENT_API_KEY")
|
||||
if not resolved_api_key:
|
||||
raise AuthenticationError(
|
||||
message="TENCENT_API_KEY is not set. Set it in the environment or pass api_key.",
|
||||
llm_provider="tencent",
|
||||
model=model,
|
||||
)
|
||||
return {**headers, "Authorization": f"Bearer {resolved_api_key}", "Content-Type": "application/json"}
|
||||
|
||||
|
||||
def tokenhub_image_params(params: Mapping[str, object], model: str, drop_params: bool) -> Mapping[str, object]:
|
||||
n: Final = params.get("n")
|
||||
if n is not None and n != 1 and not drop_params:
|
||||
raise UnsupportedParamsError(
|
||||
status_code=400,
|
||||
message=f"Tencent TokenHub returns one image per request, so n={n} is not supported for {model}. "
|
||||
+ "To send n=1 instead, set `litellm.drop_params = True`.",
|
||||
)
|
||||
return {
|
||||
key: value
|
||||
for key, value in params.items()
|
||||
if key != "n" and value is not None and not (key == "background" and value == "auto")
|
||||
}
|
||||
|
||||
|
||||
def tokenhub_image_response(raw_response: httpx.Response) -> ImageResponse | None:
|
||||
try:
|
||||
parsed: Final = _TokenHubImageResponse.model_validate_json(raw_response.content)
|
||||
except ValidationError:
|
||||
return None
|
||||
total_tokens: Final = parsed.tokenhub_usage.total_tokens if parsed.tokenhub_usage is not None else 0
|
||||
return ImageResponse(
|
||||
created=parsed.created,
|
||||
data=[ImageObject(url=image.url, b64_json=image.b64_json) for image in parsed.data],
|
||||
usage=ImageUsage(
|
||||
input_tokens=0,
|
||||
input_tokens_details=ImageUsageInputTokensDetails(image_tokens=0, text_tokens=0),
|
||||
output_tokens=total_tokens,
|
||||
total_tokens=total_tokens,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class TencentImageGenerationConfig(BaseImageGenerationConfig):
|
||||
@override
|
||||
def get_supported_openai_params(
|
||||
self, model: str
|
||||
) -> list[OpenAIImageGenerationOptionalParams]: # mutable-ok: mirrors override contract
|
||||
return list(TOKENHUB_IMAGE_PARAMS)
|
||||
|
||||
@override
|
||||
def map_openai_params(
|
||||
self,
|
||||
non_default_params: Mapping[str, object],
|
||||
optional_params: Mapping[str, object],
|
||||
model: str,
|
||||
drop_params: bool,
|
||||
) -> dict[str, object]: # mutable-ok: mirrors override contract
|
||||
return {**optional_params, **tokenhub_image_params(non_default_params, model, drop_params)}
|
||||
|
||||
@override
|
||||
def get_complete_url(
|
||||
self,
|
||||
api_base: str | None,
|
||||
api_key: str | None,
|
||||
model: str,
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
stream: bool | None = None,
|
||||
) -> str:
|
||||
return tokenhub_image_url(api_base)
|
||||
|
||||
@override
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: Mapping[str, str],
|
||||
model: str,
|
||||
messages: Sequence[AllMessageValues],
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
api_key: str | None = None,
|
||||
api_base: str | None = None,
|
||||
) -> dict[str, str]: # mutable-ok: mirrors override contract
|
||||
return dict(tokenhub_image_headers(headers, api_key, model))
|
||||
|
||||
@override
|
||||
def transform_image_generation_request(
|
||||
self,
|
||||
model: str,
|
||||
prompt: str,
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
headers: Mapping[str, str],
|
||||
) -> dict[str, object]: # mutable-ok: mirrors override contract
|
||||
extra_body: Final = _JSON_OBJECT.validate_python(optional_params.get("extra_body") or {})
|
||||
params: Final = {key: value for key, value in optional_params.items() if key != "extra_body"}
|
||||
return {"model": model, "prompt": prompt, **params, **extra_body}
|
||||
|
||||
@override
|
||||
def transform_image_generation_response(
|
||||
self,
|
||||
model: str,
|
||||
raw_response: httpx.Response,
|
||||
model_response: ImageResponse,
|
||||
logging_obj: "LiteLLMLoggingObj",
|
||||
request_data: Mapping[str, object],
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
encoding: object,
|
||||
api_key: str | None = None,
|
||||
json_mode: bool | None = None,
|
||||
) -> ImageResponse:
|
||||
image_response: Final = tokenhub_image_response(raw_response)
|
||||
if image_response is None:
|
||||
raise BaseLLMException(
|
||||
status_code=raw_response.status_code,
|
||||
message=raw_response.text,
|
||||
headers=raw_response.headers,
|
||||
)
|
||||
return image_response
|
||||
|
|
@ -10414,6 +10414,12 @@ class ProviderConfigManager:
|
|||
return get_modelscope_image_generation_config(model)
|
||||
elif LlmProviders.EDENAI == provider:
|
||||
return litellm.EdenAIImageGenerationConfig()
|
||||
elif LlmProviders.TENCENT == provider:
|
||||
from litellm.llms.tencent.image_generation.transformation import (
|
||||
TencentImageGenerationConfig,
|
||||
)
|
||||
|
||||
return TencentImageGenerationConfig()
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -10602,6 +10608,12 @@ class ProviderConfigManager:
|
|||
)
|
||||
|
||||
return get_openrouter_image_edit_config(model)
|
||||
elif LlmProviders.TENCENT == provider:
|
||||
from litellm.llms.tencent.image_edit.transformation import (
|
||||
TencentImageEditConfig,
|
||||
)
|
||||
|
||||
return TencentImageEditConfig()
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -2555,7 +2555,8 @@
|
|||
"messages": true,
|
||||
"responses": true,
|
||||
"embeddings": false,
|
||||
"image_generations": false,
|
||||
"image_generations": true,
|
||||
"image_edits": true,
|
||||
"audio_transcriptions": false,
|
||||
"audio_speech": false,
|
||||
"moderations": false,
|
||||
|
|
|
|||
0
tests/unit/llms/tencent/image_edit/__init__.py
Normal file
0
tests/unit/llms/tencent/image_edit/__init__.py
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
import base64
|
||||
import io
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
|
||||
TOKENHUB_RESPONSE = {
|
||||
"created": 1792551838,
|
||||
"data": [{"output_format": "jpeg", "size": "2048x2048", "url": "https://cos.example/edited.jpg"}],
|
||||
"model": "doubao-seedream-5-0-lite",
|
||||
"tokenhub_usage": {"total_tokens": 21875},
|
||||
"request_id": "4f795d48-b227-4c35-af2c-1a0ba288b6b6",
|
||||
}
|
||||
|
||||
|
||||
def _client(captured: dict) -> HTTPHandler:
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
captured["url"] = str(request.url)
|
||||
captured["content_type"] = request.headers.get("content-type")
|
||||
captured["authorization"] = request.headers.get("authorization")
|
||||
captured["body"] = json.loads(request.content)
|
||||
return httpx.Response(200, json=TOKENHUB_RESPONSE)
|
||||
|
||||
return HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond)))
|
||||
|
||||
|
||||
def test_edit_sends_uploaded_images_as_data_urls_in_a_json_body():
|
||||
captured = {}
|
||||
png_data_url = f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"
|
||||
|
||||
response = litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-lite",
|
||||
image=[PNG_BYTES, io.BytesIO(PNG_BYTES), "https://ref.example/style.png"],
|
||||
prompt="turn the square into a paper lantern",
|
||||
size="2K",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
)
|
||||
|
||||
assert captured["url"] == "https://tokenhub-intl.tencentcloudmaas.com/v1/wand/si-image/generation"
|
||||
assert captured["content_type"] == "application/json"
|
||||
assert captured["authorization"] == "Bearer sk-tokenhub"
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-lite",
|
||||
"prompt": "turn the square into a paper lantern",
|
||||
"images": [png_data_url, png_data_url, "https://ref.example/style.png"],
|
||||
"size": "2K",
|
||||
}
|
||||
assert response.data[0].url == "https://cos.example/edited.jpg"
|
||||
assert response.usage.total_tokens == 21875
|
||||
|
||||
|
||||
def test_edit_rejects_a_mask_without_drop_params():
|
||||
captured = {}
|
||||
|
||||
with pytest.raises(litellm.UnsupportedParamsError):
|
||||
litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
image=PNG_BYTES,
|
||||
mask=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
)
|
||||
|
||||
assert captured == {}
|
||||
|
||||
|
||||
def test_edit_drops_a_mask_under_drop_params():
|
||||
captured = {}
|
||||
|
||||
litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
image=PNG_BYTES,
|
||||
mask=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
drop_params=True,
|
||||
)
|
||||
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-pro",
|
||||
"prompt": "turn the square into a paper lantern",
|
||||
"images": [f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"],
|
||||
}
|
||||
0
tests/unit/llms/tencent/image_generation/__init__.py
Normal file
0
tests/unit/llms/tencent/image_generation/__init__.py
Normal file
|
|
@ -0,0 +1,180 @@
|
|||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
TOKENHUB_RESPONSE = {
|
||||
"created": 1792551825,
|
||||
"data": [{"output_format": "jpeg", "size": "1248x832", "url": "https://cos.example/lantern.jpg"}],
|
||||
"model": "doubao-seedream-5-0-pro-260628",
|
||||
"tokenhub_usage": {"total_tokens": 28125},
|
||||
"request_id": "c57b0374-b5fe-47db-b4b2-751315f920aa",
|
||||
}
|
||||
|
||||
|
||||
def _client(captured: dict, status_code: int = 200, body: dict | None = None) -> HTTPHandler:
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
captured["url"] = str(request.url)
|
||||
captured["authorization"] = request.headers.get("authorization")
|
||||
captured["body"] = json.loads(request.content)
|
||||
return httpx.Response(status_code, json=TOKENHUB_RESPONSE if body is None else body)
|
||||
|
||||
return HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond)))
|
||||
|
||||
|
||||
def test_generation_posts_openai_and_tokenhub_params_to_the_wand_route():
|
||||
captured = {}
|
||||
|
||||
response = litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
prompt="a red paper lantern",
|
||||
size="2K",
|
||||
response_format="url",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
images=["https://ref.example/lantern.png"],
|
||||
watermark=False,
|
||||
)
|
||||
|
||||
assert captured["url"] == "https://tokenhub-intl.tencentcloudmaas.com/v1/wand/si-image/generation"
|
||||
assert captured["authorization"] == "Bearer sk-tokenhub"
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-pro",
|
||||
"prompt": "a red paper lantern",
|
||||
"size": "2K",
|
||||
"response_format": "url",
|
||||
"images": ["https://ref.example/lantern.png"],
|
||||
"watermark": False,
|
||||
}
|
||||
assert response.data[0].url == "https://cos.example/lantern.jpg"
|
||||
assert response.created == 1792551825
|
||||
assert response.usage.total_tokens == 28125
|
||||
assert response.usage.output_tokens == 28125
|
||||
|
||||
|
||||
def test_generation_returns_base64_images():
|
||||
captured = {}
|
||||
body = {**TOKENHUB_RESPONSE, "data": [{"output_format": "png", "size": "1024x1024", "b64_json": "aW1n"}]}
|
||||
|
||||
response = litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-flash",
|
||||
prompt="a red paper lantern",
|
||||
response_format="b64_json",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured, body=body),
|
||||
)
|
||||
|
||||
assert response.data[0].b64_json == "aW1n"
|
||||
assert response.data[0].url is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("api_base", "expected_url"),
|
||||
[
|
||||
("https://tokenhub.example", "https://tokenhub.example/v1/wand/si-image/generation"),
|
||||
("https://tokenhub.example/v1", "https://tokenhub.example/v1/wand/si-image/generation"),
|
||||
("https://tokenhub.example/v1/", "https://tokenhub.example/v1/wand/si-image/generation"),
|
||||
],
|
||||
)
|
||||
def test_generation_resolves_a_configured_api_base_to_the_wand_route(api_base, expected_url):
|
||||
captured = {}
|
||||
|
||||
litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-lite",
|
||||
prompt="a red paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
api_base=api_base,
|
||||
client=_client(captured),
|
||||
)
|
||||
|
||||
assert captured["url"] == expected_url
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("params", "expected_body_params"),
|
||||
[
|
||||
({"n": 1}, {}),
|
||||
({"n": 3, "drop_params": True}, {}),
|
||||
({"background": "auto"}, {}),
|
||||
({"background": "transparent", "output_format": "png"}, {"background": "transparent", "output_format": "png"}),
|
||||
],
|
||||
)
|
||||
def test_generation_sends_only_values_tokenhub_accepts(params, expected_body_params):
|
||||
captured = {}
|
||||
|
||||
litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
prompt="a red paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
**params,
|
||||
)
|
||||
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-pro",
|
||||
"prompt": "a red paper lantern",
|
||||
**expected_body_params,
|
||||
}
|
||||
|
||||
|
||||
def test_generation_rejects_more_than_one_image_without_drop_params():
|
||||
captured = {}
|
||||
|
||||
with pytest.raises(litellm.UnsupportedParamsError) as exc_info:
|
||||
litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
prompt="a red paper lantern",
|
||||
n=2,
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 400
|
||||
assert captured == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("error", "expected_message"),
|
||||
[
|
||||
(
|
||||
{
|
||||
"error": {
|
||||
"message": "The model or service ID 'seedream-image-v5.0-pro' does not exist.",
|
||||
"code": "400004",
|
||||
"type": "gateway_error",
|
||||
}
|
||||
},
|
||||
"does not exist",
|
||||
),
|
||||
({"Type": 2, "Code": 1001, "Msg": "Invalid param", "Desc": ""}, "Invalid param"),
|
||||
],
|
||||
)
|
||||
def test_generation_surfaces_the_tokenhub_error_message(error, expected_message):
|
||||
with pytest.raises(litellm.BadRequestError) as exc_info:
|
||||
litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
prompt="a red paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client({}, status_code=400, body=error),
|
||||
)
|
||||
|
||||
assert expected_message in str(exc_info.value)
|
||||
|
||||
|
||||
def test_generation_without_a_tokenhub_key_is_an_authentication_error(monkeypatch):
|
||||
monkeypatch.delenv("TENCENT_API_KEY", raising=False)
|
||||
captured = {}
|
||||
|
||||
with pytest.raises(litellm.AuthenticationError) as exc_info:
|
||||
litellm.image_generation(
|
||||
model="tencent/seedream-image-v5.0-pro",
|
||||
prompt="a red paper lantern",
|
||||
client=_client(captured),
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
assert "TENCENT_API_KEY" in str(exc_info.value)
|
||||
assert captured == {}
|
||||
Loading…
Add table
Reference in a new issue