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 (#48735)
* feat(tencent): add Seedream image generation and editing through TokenHub * feat(tencent): price Seedream image generation and edits from TokenHub usage * test(tencent): cover Seedream image generation, edits, and pricing through TokenHub * fix(tencent): keep non-Seedream image models on the OpenAI-compatible route Only seedream- models use TokenHub's wand route, so Hy-Image and other tencent image models keep working as on main. The response now carries TokenHub's image size and format, Pro's flat fallback bills its default 2K size, and the new tests are typed. * fix(tencent): bill other Tencent image models as before and forward user and TokenHub params on edits * fix(tencent): keep dropped edit params out and parse form-string watermark and seed * fix(tencent): send extra_body edit fields to TokenHub as given * fix(tencent): keep the routed model on Seedream edits and bill a deployment's per-token price * fix(tencent): accept string drop params on edits, rebuild nested form fields, and honor a per-image deployment price An image edit with a non-list additional_drop_params answered 500 for every provider after this branch validated it as a list, so a non-list value now drops nothing again, as chat treats it. Seedream edits sent through the OpenAI SDK arrive as multipart form fields, so TokenHub-only booleans and integers are parsed and bracketed keys like optimize_prompt_options[mode] are rebuilt into objects. A deployment that sets only output_cost_per_image is billed that price, and the TokenHub URL uses the shared api_base helpers. --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
parent
0f786fa1c7
commit
8e8f6dc66e
21 changed files with 1204 additions and 102 deletions
|
|
@ -1,9 +1,9 @@
|
|||
import asyncio
|
||||
import contextvars
|
||||
import importlib
|
||||
from collections.abc import Coroutine, Mapping
|
||||
from collections.abc import Collection, Coroutine, Mapping
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, Final, Literal, cast, overload
|
||||
from typing import TYPE_CHECKING, Final, Literal, cast, get_type_hints, overload
|
||||
|
||||
from pydantic import TypeAdapter, ValidationError
|
||||
|
||||
|
|
@ -61,6 +61,7 @@ from litellm.types.utils import (
|
|||
ProviderRoutingHints,
|
||||
)
|
||||
from litellm.utils import (
|
||||
EXTRA_BODY_ROUTING_KEYS,
|
||||
ImageResponse,
|
||||
ProviderConfigManager,
|
||||
get_llm_provider,
|
||||
|
|
@ -466,7 +467,7 @@ def image_generation(
|
|||
litellm.LlmProviders.QWENCLOUD,
|
||||
litellm.LlmProviders.QWEN_AI_PLATFORM,
|
||||
litellm.LlmProviders.EDENAI,
|
||||
):
|
||||
) or (custom_llm_provider == litellm.LlmProviders.TENCENT and image_generation_config is not None):
|
||||
if image_generation_config is None:
|
||||
raise ValueError(f"image generation config is not supported for {custom_llm_provider}")
|
||||
|
||||
|
|
@ -807,6 +808,32 @@ def image_variation(
|
|||
return response
|
||||
|
||||
|
||||
_IMAGE_EDIT_OPENAI_PARAMS: Final = frozenset(get_type_hints(ImageEditOptionalRequestParams))
|
||||
_DROP_PARAM_KWARGS: Final = frozenset({"drop_params", "additional_drop_params"})
|
||||
_ADDITIONAL_DROP_PARAMS: Final = TypeAdapter[tuple[str, ...]](tuple[str, ...])
|
||||
|
||||
|
||||
def _listed_drop_params(additional_drop_params: object) -> tuple[str, ...]:
|
||||
try:
|
||||
return _ADDITIONAL_DROP_PARAMS.validate_python(additional_drop_params)
|
||||
except ValidationError:
|
||||
return ()
|
||||
|
||||
|
||||
def _tokenhub_passthrough_edit_params(
|
||||
non_default_params: Mapping[str, object],
|
||||
extra_body: Mapping[str, object] | None,
|
||||
mapped_params: Collection[str],
|
||||
additional_drop_params: Collection[str] | None,
|
||||
) -> Mapping[str, object]:
|
||||
dropped: Final = frozenset(additional_drop_params or ())
|
||||
unmapped_excluded: Final = _DROP_PARAM_KWARGS | _IMAGE_EDIT_OPENAI_PARAMS | frozenset(mapped_params) | dropped
|
||||
return {
|
||||
**{key: value for key, value in non_default_params.items() if key not in unmapped_excluded},
|
||||
**{key: value for key, value in (extra_body or {}).items() if key not in dropped | EXTRA_BODY_ROUTING_KEYS},
|
||||
}
|
||||
|
||||
|
||||
@client
|
||||
def image_edit(
|
||||
image: FileTypes | list[FileTypes] | None = None,
|
||||
|
|
@ -935,22 +962,24 @@ def image_edit(
|
|||
raise ValueError(f"image edit is not supported for {custom_llm_provider}")
|
||||
|
||||
local_vars.update(kwargs)
|
||||
provider_supported_params: Final[frozenset[str]] = frozenset(
|
||||
image_edit_provider_config.get_supported_openai_params(model)
|
||||
)
|
||||
# Get ImageEditOptionalRequestParams with only valid parameters
|
||||
image_edit_optional_params: Final[ImageEditOptionalRequestParams] = (
|
||||
_get_ImageEditRequestUtils().get_requested_image_edit_optional_param(
|
||||
local_vars,
|
||||
provider_supported_params=frozenset(
|
||||
image_edit_provider_config.get_supported_openai_params(model)
|
||||
).intersection(non_default_params),
|
||||
provider_supported_params=provider_supported_params.intersection(non_default_params),
|
||||
)
|
||||
)
|
||||
# Get optional parameters for the responses API
|
||||
additional_drop_params: Final = _listed_drop_params(kwargs.get("additional_drop_params"))
|
||||
image_edit_request_params: Final[dict] = _get_ImageEditRequestUtils().get_optional_params_image_edit(
|
||||
model=model,
|
||||
image_edit_provider_config=image_edit_provider_config,
|
||||
image_edit_optional_params=image_edit_optional_params,
|
||||
drop_params=kwargs.get("drop_params"),
|
||||
additional_drop_params=kwargs.get("additional_drop_params"),
|
||||
additional_drop_params=additional_drop_params,
|
||||
)
|
||||
|
||||
if image_edit_provider_config.use_multipart_form_data() and (
|
||||
|
|
@ -1039,7 +1068,19 @@ def image_edit(
|
|||
image=images,
|
||||
prompt=prompt,
|
||||
image_edit_provider_config=image_edit_provider_config,
|
||||
image_edit_optional_request_params=image_edit_request_params,
|
||||
image_edit_optional_request_params={
|
||||
**image_edit_request_params,
|
||||
**(
|
||||
_tokenhub_passthrough_edit_params(
|
||||
non_default_params,
|
||||
extra_body if isinstance(extra_body, dict) else None,
|
||||
provider_supported_params,
|
||||
additional_drop_params,
|
||||
)
|
||||
if custom_llm_provider == litellm.LlmProviders.TENCENT.value
|
||||
else {}
|
||||
),
|
||||
},
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
litellm_params=litellm_params,
|
||||
logging_obj=litellm_logging_obj,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -16,7 +19,7 @@ class ImageEditRequestUtils:
|
|||
image_edit_provider_config: BaseImageEditConfig,
|
||||
image_edit_optional_params: ImageEditOptionalRequestParams,
|
||||
drop_params: bool | None = None,
|
||||
additional_drop_params: list[str] | None = None,
|
||||
additional_drop_params: Collection[str] | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Get optional parameters for the image edit API.
|
||||
|
|
@ -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')}"
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from collections.abc import Mapping, Sequence
|
|||
from dataclasses import dataclass, field, replace
|
||||
from datetime import datetime, timezone, tzinfo
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Literal, TypedDict, cast
|
||||
from typing import Final, Literal, Protocol, TypedDict, cast
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
|
||||
from typing_extensions import ReadOnly
|
||||
|
|
@ -2085,6 +2085,41 @@ def calculate_image_response_web_search_cost(
|
|||
)
|
||||
|
||||
|
||||
class _ModelPricedImageCostCalculator(Protocol):
|
||||
def __call__(self, model: str, image_response: ImageResponse, model_info: ModelInfo | None = None) -> float: ...
|
||||
|
||||
|
||||
def _model_priced_image_cost_calculator(
|
||||
custom_llm_provider: str | None, model: str
|
||||
) -> _ModelPricedImageCostCalculator | None:
|
||||
match custom_llm_provider:
|
||||
case litellm.LlmProviders.RECRAFT.value:
|
||||
from litellm.llms.recraft.cost_calculator import cost_calculator as recraft_image_cost_calculator
|
||||
|
||||
return recraft_image_cost_calculator
|
||||
case litellm.LlmProviders.AIML.value:
|
||||
from litellm.llms.aiml.image_generation.cost_calculator import cost_calculator as aiml_image_cost_calculator
|
||||
|
||||
return aiml_image_cost_calculator
|
||||
case litellm.LlmProviders.COMETAPI.value:
|
||||
from litellm.llms.cometapi.image_generation.cost_calculator import (
|
||||
cost_calculator as cometapi_image_cost_calculator,
|
||||
)
|
||||
|
||||
return cometapi_image_cost_calculator
|
||||
case litellm.LlmProviders.RUNWAYML.value:
|
||||
from litellm.llms.runwayml.cost_calculator import cost_calculator as runwayml_image_cost_calculator
|
||||
|
||||
return runwayml_image_cost_calculator
|
||||
case litellm.LlmProviders.TENCENT.value:
|
||||
from litellm.llms.tencent.cost_calculator import image_cost_calculator as tencent_image_cost_calculator
|
||||
from litellm.llms.tencent.image_generation.transformation import is_tokenhub_image_model
|
||||
|
||||
return tencent_image_cost_calculator if is_tokenhub_image_model(model) else None
|
||||
case _:
|
||||
return None
|
||||
|
||||
|
||||
class CostCalculatorUtils:
|
||||
@staticmethod
|
||||
def call_type_has_image_response(call_type: str) -> bool:
|
||||
|
|
@ -2131,9 +2166,6 @@ class CostCalculatorUtils:
|
|||
from litellm.llms.gemini.image_generation.cost_calculator import (
|
||||
cost_calculator as gemini_image_cost_calculator,
|
||||
)
|
||||
from litellm.llms.recraft.cost_calculator import (
|
||||
cost_calculator as recraft_image_cost_calculator,
|
||||
)
|
||||
from litellm.llms.vertex_ai.image_generation.cost_calculator import (
|
||||
cost_calculator as vertex_ai_image_cost_calculator,
|
||||
)
|
||||
|
|
@ -2146,7 +2178,10 @@ class CostCalculatorUtils:
|
|||
)
|
||||
resolved_n: Final = n if n is not None else (len(completion_response.data) if completion_response.data else 0)
|
||||
pricing: Final = deployment_pricing(model_info)
|
||||
model_priced_calculator: Final = _model_priced_image_cost_calculator(custom_llm_provider, model)
|
||||
|
||||
if model_priced_calculator is not None:
|
||||
return model_priced_calculator(model=model, image_response=completion_response, model_info=pricing)
|
||||
if custom_llm_provider == litellm.LlmProviders.VERTEX_AI.value:
|
||||
if isinstance(completion_response, ImageResponse):
|
||||
return vertex_ai_image_cost_calculator(
|
||||
|
|
@ -2164,36 +2199,6 @@ class CostCalculatorUtils:
|
|||
optional_params=optional_params,
|
||||
)
|
||||
raise TypeError("completion_response must be of type ImageResponse for bedrock image cost calculation")
|
||||
elif custom_llm_provider == litellm.LlmProviders.RECRAFT.value:
|
||||
from litellm.llms.recraft.cost_calculator import (
|
||||
cost_calculator as recraft_image_cost_calculator,
|
||||
)
|
||||
|
||||
return recraft_image_cost_calculator(
|
||||
model=model,
|
||||
image_response=completion_response,
|
||||
model_info=pricing,
|
||||
)
|
||||
elif custom_llm_provider == litellm.LlmProviders.AIML.value:
|
||||
from litellm.llms.aiml.image_generation.cost_calculator import (
|
||||
cost_calculator as aiml_image_cost_calculator,
|
||||
)
|
||||
|
||||
return aiml_image_cost_calculator(
|
||||
model=model,
|
||||
image_response=completion_response,
|
||||
model_info=pricing,
|
||||
)
|
||||
elif custom_llm_provider == litellm.LlmProviders.COMETAPI.value:
|
||||
from litellm.llms.cometapi.image_generation.cost_calculator import (
|
||||
cost_calculator as cometapi_image_cost_calculator,
|
||||
)
|
||||
|
||||
return cometapi_image_cost_calculator(
|
||||
model=model,
|
||||
image_response=completion_response,
|
||||
model_info=pricing,
|
||||
)
|
||||
elif custom_llm_provider == litellm.LlmProviders.GEMINI.value:
|
||||
if call_type in (
|
||||
CallTypes.image_edit.value,
|
||||
|
|
@ -2237,16 +2242,6 @@ class CostCalculatorUtils:
|
|||
optional_params=optional_params,
|
||||
model_info=pricing,
|
||||
)
|
||||
elif custom_llm_provider == litellm.LlmProviders.RUNWAYML.value:
|
||||
from litellm.llms.runwayml.cost_calculator import (
|
||||
cost_calculator as runwayml_image_cost_calculator,
|
||||
)
|
||||
|
||||
return runwayml_image_cost_calculator(
|
||||
model=model,
|
||||
image_response=completion_response,
|
||||
model_info=pricing,
|
||||
)
|
||||
elif (
|
||||
custom_llm_provider == litellm.LlmProviders.OPENAI.value
|
||||
or custom_llm_provider == litellm.LlmProviders.AZURE.value
|
||||
|
|
|
|||
|
|
@ -4,13 +4,14 @@ from typing import Final
|
|||
|
||||
from httpx._types import RequestFiles
|
||||
|
||||
from litellm.images.utils import image_to_data_url
|
||||
from litellm.litellm_core_utils.url_utils import join_api_base
|
||||
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")
|
||||
|
|
@ -68,7 +69,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,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,49 @@
|
|||
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
|
||||
from litellm.types.utils import Usage
|
||||
from typing import Final
|
||||
|
||||
from litellm.litellm_core_utils.llm_cost_calc.utils import (
|
||||
calculate_image_response_cost_from_usage,
|
||||
flat_image_cost,
|
||||
generic_cost_per_token,
|
||||
resolve_image_model_info,
|
||||
)
|
||||
from litellm.types.utils import ImageResponse, ModelInfo, Usage
|
||||
|
||||
|
||||
def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
|
||||
return generic_cost_per_token(model=model, usage=usage, custom_llm_provider="tencent")
|
||||
|
||||
|
||||
def _deployment_image_token_pricing(model_info: ModelInfo | None) -> ModelInfo | None:
|
||||
if model_info is None or model_info.get("output_cost_per_image_token") is not None:
|
||||
return model_info
|
||||
output_cost_per_token: Final = model_info.get("output_cost_per_token")
|
||||
if output_cost_per_token is None:
|
||||
return model_info
|
||||
priced: Final[ModelInfo] = {**model_info, "output_cost_per_image_token": output_cost_per_token}
|
||||
return priced
|
||||
|
||||
|
||||
def _deployment_prices_per_image_only(model_info: ModelInfo | None) -> bool:
|
||||
return (
|
||||
model_info is not None
|
||||
and model_info.get("output_cost_per_image") is not None
|
||||
and model_info.get("output_cost_per_image_token") is None
|
||||
and model_info.get("output_cost_per_token") is None
|
||||
)
|
||||
|
||||
|
||||
def image_cost_calculator(model: str, image_response: ImageResponse, model_info: ModelInfo | None = None) -> float:
|
||||
resolved_model_info: Final = resolve_image_model_info(
|
||||
model=model, custom_llm_provider="tencent", model_info=_deployment_image_token_pricing(model_info)
|
||||
)
|
||||
if _deployment_prices_per_image_only(model_info):
|
||||
return flat_image_cost(resolved_model_info, image_response)
|
||||
token_based_cost: Final = calculate_image_response_cost_from_usage(
|
||||
model=model,
|
||||
image_response=image_response,
|
||||
custom_llm_provider="tencent",
|
||||
model_info=resolved_model_info,
|
||||
)
|
||||
if token_based_cost is not None:
|
||||
return token_based_cost
|
||||
return flat_image_cost(resolved_model_info, image_response)
|
||||
|
|
|
|||
0
litellm/llms/tencent/image_edit/__init__.py
Normal file
0
litellm/llms/tencent/image_edit/__init__.py
Normal file
147
litellm/llms/tencent/image_edit/transformation.py
Normal file
147
litellm/llms/tencent/image_edit/transformation.py
Normal file
|
|
@ -0,0 +1,147 @@
|
|||
from collections.abc import Mapping
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Final, TypeAlias
|
||||
|
||||
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, ...])
|
||||
_FORM_BOOLEANS: Final = MappingProxyType({"true": True, "false": False})
|
||||
_OPENAI_PARAMS: Final = frozenset(TOKENHUB_IMAGE_PARAMS)
|
||||
|
||||
_FormEntries: TypeAlias = tuple[tuple[tuple[str, ...], object], ...]
|
||||
|
||||
|
||||
def _form_scalar(value: object) -> object:
|
||||
match value:
|
||||
case str() if value.lower() in _FORM_BOOLEANS:
|
||||
return _FORM_BOOLEANS[value.lower()]
|
||||
case str() if value.removeprefix("-").isdigit():
|
||||
return int(value)
|
||||
case _:
|
||||
return value
|
||||
|
||||
|
||||
def _form_path(key: str) -> tuple[str, ...]:
|
||||
root, bracket, rest = key.partition("[")
|
||||
if not bracket or not rest.endswith("]"):
|
||||
return (key,)
|
||||
return (root, *rest.removesuffix("]").split("]["))
|
||||
|
||||
|
||||
def _nested_form_value(entries: _FormEntries) -> object:
|
||||
match entries:
|
||||
case (((), value),):
|
||||
return _form_scalar(value)
|
||||
case _:
|
||||
return _nested_form_fields(tuple(entry for entry in entries if entry[0]))
|
||||
|
||||
|
||||
def _nested_form_fields(entries: _FormEntries) -> Mapping[str, object]:
|
||||
roots: Final = tuple(dict.fromkeys(path[0] for path, _ in entries))
|
||||
return {
|
||||
root: _nested_form_value(tuple((path[1:], value) for path, value in entries if path[0] == root))
|
||||
for root in roots
|
||||
}
|
||||
|
||||
|
||||
def _tokenhub_form_params(params: Mapping[str, object]) -> Mapping[str, object]:
|
||||
tokenhub_entries: Final = tuple(
|
||||
(_form_path(key), value) for key, value in params.items() if key not in _OPENAI_PARAMS
|
||||
)
|
||||
return {
|
||||
**{key: value for key, value in params.items() if key in _OPENAI_PARAMS},
|
||||
**_nested_form_fields(tokenhub_entries),
|
||||
}
|
||||
|
||||
|
||||
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),
|
||||
**_tokenhub_form_params(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
200
litellm/llms/tencent/image_generation/transformation.py
Normal file
200
litellm/llms/tencent/image_generation/transformation.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
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.litellm_core_utils.url_utils import ensure_api_route, join_api_base
|
||||
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,
|
||||
LlmProviders,
|
||||
)
|
||||
|
||||
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_MODEL_PREFIX: Final = "seedream-"
|
||||
TOKENHUB_IMAGE_PARAMS: Final[tuple[OpenAIImageGenerationOptionalParams, ...]] = (
|
||||
"n",
|
||||
"size",
|
||||
"response_format",
|
||||
"output_format",
|
||||
"background",
|
||||
"user",
|
||||
)
|
||||
|
||||
_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
|
||||
size: str | None = None
|
||||
output_format: 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 is_tokenhub_image_model(model: str) -> bool:
|
||||
return model.removeprefix(f"{LlmProviders.TENCENT.value}/").startswith(TOKENHUB_IMAGE_MODEL_PREFIX)
|
||||
|
||||
|
||||
def tokenhub_image_url(api_base: str | None) -> str:
|
||||
base: Final = api_base or get_secret_str("TENCENT_API_BASE") or TOKENHUB_API_BASE
|
||||
return join_api_base(ensure_api_route(base, "v1"), 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
|
||||
first_image: Final = parsed.data[0] if parsed.data else _TokenHubImage()
|
||||
return ImageResponse(
|
||||
created=parsed.created,
|
||||
data=[ImageObject(url=image.url, b64_json=image.b64_json) for image in parsed.data],
|
||||
size=first_image.size,
|
||||
output_format=first_image.output_format,
|
||||
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
|
||||
|
|
@ -61986,6 +61986,39 @@
|
|||
"supports_reasoning": true,
|
||||
"supports_vision": false
|
||||
},
|
||||
"tencent/seedream-image-v5.0-flash": {
|
||||
"litellm_provider": "tencent",
|
||||
"mode": "image_generation",
|
||||
"output_cost_per_image": 0.018,
|
||||
"output_cost_per_image_token": 1.6e-06,
|
||||
"source": "https://cloud.tencent.com/document/product/1823/130055",
|
||||
"supported_endpoints": [
|
||||
"/v1/images/generations",
|
||||
"/v1/images/edits"
|
||||
]
|
||||
},
|
||||
"tencent/seedream-image-v5.0-lite": {
|
||||
"litellm_provider": "tencent",
|
||||
"mode": "image_generation",
|
||||
"output_cost_per_image": 0.035,
|
||||
"output_cost_per_image_token": 1.6e-06,
|
||||
"source": "https://www.tencentcloud.com/document/product/1300/78937",
|
||||
"supported_endpoints": [
|
||||
"/v1/images/generations",
|
||||
"/v1/images/edits"
|
||||
]
|
||||
},
|
||||
"tencent/seedream-image-v5.0-pro": {
|
||||
"litellm_provider": "tencent",
|
||||
"mode": "image_generation",
|
||||
"output_cost_per_image": 0.09,
|
||||
"output_cost_per_image_token": 1.6e-06,
|
||||
"source": "https://www.tencentcloud.com/document/product/1300/78937",
|
||||
"supported_endpoints": [
|
||||
"/v1/images/generations",
|
||||
"/v1/images/edits"
|
||||
]
|
||||
},
|
||||
"cognition/swe-1.6": {
|
||||
"input_cost_per_token": 5e-07,
|
||||
"output_cost_per_token": 2.5e-06,
|
||||
|
|
|
|||
|
|
@ -3793,8 +3793,8 @@
|
|||
"responses": true,
|
||||
"text_completion": true,
|
||||
"embeddings": false,
|
||||
"image_generations": false,
|
||||
"image_edits": false,
|
||||
"image_generations": true,
|
||||
"image_edits": true,
|
||||
"audio_transcriptions": false,
|
||||
"audio_speech": false,
|
||||
"rerank": false,
|
||||
|
|
|
|||
|
|
@ -10418,6 +10418,13 @@ 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,
|
||||
is_tokenhub_image_model,
|
||||
)
|
||||
|
||||
return TencentImageGenerationConfig() if is_tokenhub_image_model(model) else None
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -10606,6 +10613,15 @@ class ProviderConfigManager:
|
|||
)
|
||||
|
||||
return get_openrouter_image_edit_config(model)
|
||||
elif LlmProviders.TENCENT == provider:
|
||||
from litellm.llms.tencent.image_edit.transformation import (
|
||||
TencentImageEditConfig,
|
||||
)
|
||||
from litellm.llms.tencent.image_generation.transformation import (
|
||||
is_tokenhub_image_model,
|
||||
)
|
||||
|
||||
return TencentImageEditConfig() if is_tokenhub_image_model(model) else None
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -61986,6 +61986,39 @@
|
|||
"supports_reasoning": true,
|
||||
"supports_vision": false
|
||||
},
|
||||
"tencent/seedream-image-v5.0-flash": {
|
||||
"litellm_provider": "tencent",
|
||||
"mode": "image_generation",
|
||||
"output_cost_per_image": 0.018,
|
||||
"output_cost_per_image_token": 1.6e-06,
|
||||
"source": "https://cloud.tencent.com/document/product/1823/130055",
|
||||
"supported_endpoints": [
|
||||
"/v1/images/generations",
|
||||
"/v1/images/edits"
|
||||
]
|
||||
},
|
||||
"tencent/seedream-image-v5.0-lite": {
|
||||
"litellm_provider": "tencent",
|
||||
"mode": "image_generation",
|
||||
"output_cost_per_image": 0.035,
|
||||
"output_cost_per_image_token": 1.6e-06,
|
||||
"source": "https://www.tencentcloud.com/document/product/1300/78937",
|
||||
"supported_endpoints": [
|
||||
"/v1/images/generations",
|
||||
"/v1/images/edits"
|
||||
]
|
||||
},
|
||||
"tencent/seedream-image-v5.0-pro": {
|
||||
"litellm_provider": "tencent",
|
||||
"mode": "image_generation",
|
||||
"output_cost_per_image": 0.09,
|
||||
"output_cost_per_image_token": 1.6e-06,
|
||||
"source": "https://www.tencentcloud.com/document/product/1300/78937",
|
||||
"supported_endpoints": [
|
||||
"/v1/images/generations",
|
||||
"/v1/images/edits"
|
||||
]
|
||||
},
|
||||
"cognition/swe-1.6": {
|
||||
"input_cost_per_token": 5e-07,
|
||||
"output_cost_per_token": 2.5e-06,
|
||||
|
|
|
|||
|
|
@ -3793,8 +3793,8 @@
|
|||
"responses": true,
|
||||
"text_completion": true,
|
||||
"embeddings": false,
|
||||
"image_generations": false,
|
||||
"image_edits": false,
|
||||
"image_generations": true,
|
||||
"image_edits": true,
|
||||
"audio_transcriptions": false,
|
||||
"audio_speech": false,
|
||||
"rerank": false,
|
||||
|
|
|
|||
|
|
@ -357,6 +357,24 @@ def test_image_edit_prices_a_vertex_deployment_at_its_configured_location(
|
|||
assert cost_at("us-central1") == pytest.approx(0.044)
|
||||
|
||||
|
||||
def test_image_edit_accepts_a_string_additional_drop_params(respx_mock: respx.MockRouter) -> None:
|
||||
edits_route: Final = respx_mock.post("https://api.openai.com/v1/images/edits").mock(
|
||||
return_value=httpx.Response(status_code=200, json={"created": 1792551838, "data": [{"b64_json": "aW1n"}]})
|
||||
)
|
||||
|
||||
response: Final = litellm.image_edit(
|
||||
model="openai/gpt-image-2.5-sunburst",
|
||||
image=b"\x89PNG\r\n\x1a\nfakepng",
|
||||
prompt="make the circle blue",
|
||||
api_key="sk-test",
|
||||
quality="high",
|
||||
additional_drop_params="quality",
|
||||
)
|
||||
|
||||
assert edits_route.called
|
||||
assert response.data[0].b64_json == "aW1n"
|
||||
|
||||
|
||||
class TestCustomLogger(CustomLogger):
|
||||
__test__ = 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,232 @@
|
|||
import base64
|
||||
import io
|
||||
import json
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
PNG_BYTES: Final = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
|
||||
TOKENHUB_RESPONSE: Final[dict[str, object]] = {
|
||||
"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[str, object]) -> 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() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
png_data_url: Final = f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"
|
||||
|
||||
response: Final = 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() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
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() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
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()}"],
|
||||
}
|
||||
|
||||
|
||||
def test_edit_is_not_offered_for_tencent_models_outside_seedream() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
with pytest.raises(Exception, match="image edit is not supported for tencent"):
|
||||
litellm.image_edit(
|
||||
model="tencent/hy-image-v3.5-preview",
|
||||
image=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
)
|
||||
|
||||
assert captured == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("form_params", "expected_body_params"),
|
||||
[
|
||||
({"watermark": "false", "seed": "7"}, {"watermark": False, "seed": 7}),
|
||||
({"watermark": "True", "seed": "-1"}, {"watermark": True, "seed": -1}),
|
||||
({"watermark": "maybe"}, {"watermark": "maybe"}),
|
||||
(
|
||||
{"layer_decomposition": "true", "optimize_prompt_options[mode]": "fast"},
|
||||
{"layer_decomposition": True, "optimize_prompt_options": {"mode": "fast"}},
|
||||
),
|
||||
(
|
||||
{"sequential_image_generation_options[max_images]": "3", "user": "12345"},
|
||||
{"sequential_image_generation_options": {"max_images": 3}, "user": "12345"},
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_edit_parses_tokenhub_values_sent_as_form_strings(
|
||||
form_params: dict[str, object], expected_body_params: dict[str, object]
|
||||
) -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-flash",
|
||||
image=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
**form_params,
|
||||
)
|
||||
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-flash",
|
||||
"prompt": "turn the square into a paper lantern",
|
||||
"images": [f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"],
|
||||
**expected_body_params,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("params", "expected_body_params"),
|
||||
[
|
||||
({"background": "auto", "watermark": False}, {"watermark": False}),
|
||||
({"watermark": False, "seed": 7, "additional_drop_params": ["watermark"]}, {"seed": 7}),
|
||||
({"extra_body": {"watermark": False}, "additional_drop_params": ["watermark"]}, {}),
|
||||
],
|
||||
)
|
||||
def test_edit_keeps_dropped_params_out_of_the_tokenhub_body(
|
||||
params: dict[str, object], expected_body_params: dict[str, object]
|
||||
) -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-flash",
|
||||
image=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
**params,
|
||||
)
|
||||
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-flash",
|
||||
"prompt": "turn the square into a paper lantern",
|
||||
"images": [f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"],
|
||||
**expected_body_params,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("params", "expected_body_params"),
|
||||
[
|
||||
({"extra_body": {"response_format": "b64_json"}}, {"response_format": "b64_json"}),
|
||||
({"size": "1K", "extra_body": {"size": "2K"}}, {"size": "2K"}),
|
||||
({"extra_body": {"model": "seedream-image-v5.0-pro", "seed": 7}}, {"seed": 7}),
|
||||
({"extra_body": '{"seed": 7}'}, {}),
|
||||
],
|
||||
)
|
||||
def test_edit_sends_extra_body_fields_as_given(
|
||||
params: dict[str, object], expected_body_params: dict[str, object]
|
||||
) -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-flash",
|
||||
image=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
**params,
|
||||
)
|
||||
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-flash",
|
||||
"prompt": "turn the square into a paper lantern",
|
||||
"images": [f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"],
|
||||
**expected_body_params,
|
||||
}
|
||||
|
||||
|
||||
def test_edit_forwards_tokenhub_only_params_and_extra_body() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
litellm.image_edit(
|
||||
model="tencent/seedream-image-v5.0-flash",
|
||||
image=PNG_BYTES,
|
||||
prompt="turn the square into a paper lantern",
|
||||
api_key="sk-tokenhub",
|
||||
client=_client(captured),
|
||||
user="end-user-42",
|
||||
watermark=False,
|
||||
seed=7,
|
||||
extra_body={"sequential_image_generation": "disabled"},
|
||||
)
|
||||
|
||||
assert captured["body"] == {
|
||||
"model": "seedream-image-v5.0-flash",
|
||||
"prompt": "turn the square into a paper lantern",
|
||||
"images": [f"data:image/png;base64,{base64.b64encode(PNG_BYTES).decode()}"],
|
||||
"user": "end-user-42",
|
||||
"watermark": False,
|
||||
"seed": 7,
|
||||
"sequential_image_generation": "disabled",
|
||||
}
|
||||
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,215 @@
|
|||
import json
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import openai
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
TOKENHUB_RESPONSE: Final[dict[str, object]] = {
|
||||
"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[str, object], status_code: int = 200, body: dict[str, object] | 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() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
response: Final = 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.size == "1248x832"
|
||||
assert response.output_format == "jpeg"
|
||||
assert response.created == 1792551825
|
||||
assert response.usage.total_tokens == 28125
|
||||
assert response.usage.output_tokens == 28125
|
||||
|
||||
|
||||
def test_generation_returns_base64_images() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
body: Final[dict[str, object]] = {
|
||||
**TOKENHUB_RESPONSE,
|
||||
"data": [{"output_format": "png", "size": "1024x1024", "b64_json": "aW1n"}],
|
||||
}
|
||||
|
||||
response: Final = 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: str, expected_url: str) -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
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"}),
|
||||
({"user": "end-user-42"}, {"user": "end-user-42"}),
|
||||
],
|
||||
)
|
||||
def test_generation_sends_only_values_tokenhub_accepts(
|
||||
params: dict[str, object], expected_body_params: dict[str, object]
|
||||
) -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
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() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
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: dict[str, object], expected_message: str) -> None:
|
||||
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: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("TENCENT_API_KEY", raising=False)
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
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 == {}
|
||||
|
||||
|
||||
def test_generation_keeps_other_tencent_image_models_on_the_openai_compatible_route() -> None:
|
||||
captured: Final[dict[str, object]] = {}
|
||||
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
captured["url"] = str(request.url)
|
||||
captured["body"] = json.loads(request.content)
|
||||
return httpx.Response(200, json={"created": 1792551825, "data": [{"url": "https://cos.example/hy.png"}]})
|
||||
|
||||
response: Final = litellm.image_generation(
|
||||
model="tencent/hy-image-v3.5-preview",
|
||||
prompt="a red paper lantern",
|
||||
quality="high",
|
||||
api_key="sk-tokenhub",
|
||||
client=openai.OpenAI(
|
||||
api_key="sk-tokenhub",
|
||||
base_url="https://tokenhub-intl.tencentcloudmaas.com/v1",
|
||||
http_client=httpx.Client(transport=httpx.MockTransport(respond)),
|
||||
),
|
||||
)
|
||||
|
||||
assert captured["url"] == "https://tokenhub-intl.tencentcloudmaas.com/v1/images/generations"
|
||||
assert captured["body"] == {"model": "hy-image-v3.5-preview", "prompt": "a red paper lantern", "quality": "high"}
|
||||
assert response.data[0].url == "https://cos.example/hy.png"
|
||||
128
tests/unit/llms/tencent/test_tencent_cost_calculator.py
Normal file
128
tests/unit/llms/tencent/test_tencent_cost_calculator.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import openai
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
from litellm.types.utils import ImageObject, ImageResponse, ImageUsage, ImageUsageInputTokensDetails
|
||||
|
||||
SEEDREAM_MODELS: Final = (
|
||||
"tencent/seedream-image-v5.0-pro",
|
||||
"tencent/seedream-image-v5.0-flash",
|
||||
"tencent/seedream-image-v5.0-lite",
|
||||
)
|
||||
|
||||
|
||||
def _image_response(total_tokens: int, images: int = 1) -> ImageResponse:
|
||||
return ImageResponse(
|
||||
created=1792551825,
|
||||
data=[ImageObject(url=f"https://cos.example/{index}.jpg") for index in range(images)],
|
||||
usage=ImageUsage(
|
||||
input_tokens=0,
|
||||
input_tokens_details=ImageUsageInputTokensDetails(image_tokens=0, text_tokens=0),
|
||||
output_tokens=total_tokens,
|
||||
total_tokens=total_tokens,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("model", SEEDREAM_MODELS)
|
||||
@pytest.mark.parametrize("call_type", ["image_generation", "image_edit"])
|
||||
def test_seedream_bills_tokenhub_usage_tokens_for_a_tier_size(
|
||||
local_model_cost_map: None, model: str, call_type: str
|
||||
) -> None:
|
||||
token_rate: Final = litellm.model_cost[model]["output_cost_per_image_token"]
|
||||
|
||||
cost: Final = litellm.completion_cost(
|
||||
completion_response=_image_response(total_tokens=56250),
|
||||
model=model,
|
||||
call_type=call_type,
|
||||
size="2K",
|
||||
)
|
||||
|
||||
assert cost == pytest.approx(56250 * token_rate)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("model", SEEDREAM_MODELS)
|
||||
def test_seedream_falls_back_to_the_per_image_price_without_usage(local_model_cost_map: None, model: str) -> None:
|
||||
per_image_price: Final = litellm.model_cost[model]["output_cost_per_image"]
|
||||
|
||||
cost: Final = litellm.completion_cost(
|
||||
completion_response=_image_response(total_tokens=0, images=2),
|
||||
model=model,
|
||||
call_type="image_generation",
|
||||
size="2K",
|
||||
)
|
||||
|
||||
assert cost == pytest.approx(2 * per_image_price)
|
||||
|
||||
|
||||
async def test_a_deployment_priced_tencent_image_model_outside_seedream_bills_its_deployment_price() -> None:
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, json={"created": 1792551825, "data": [{"url": "https://cos.example/hy.png"}]})
|
||||
|
||||
router: Final = litellm.Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "hy-image",
|
||||
"litellm_params": {"model": "tencent/hy-image-v3.5-preview", "api_key": "sk-tokenhub"},
|
||||
"model_info": {"input_cost_per_image": 0.04},
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
response: Final = await router.aimage_generation(
|
||||
model="hy-image",
|
||||
prompt="a red paper lantern",
|
||||
client=openai.AsyncOpenAI(
|
||||
api_key="sk-tokenhub",
|
||||
base_url="https://tokenhub-intl.tencentcloudmaas.com/v1",
|
||||
http_client=httpx.AsyncClient(transport=httpx.MockTransport(respond)),
|
||||
),
|
||||
)
|
||||
|
||||
assert response._hidden_params["response_cost"] == pytest.approx(0.04)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("deployment_price", "expected_cost"),
|
||||
[
|
||||
({"output_cost_per_token": 0.000001}, 11250 * 0.000001),
|
||||
({"output_cost_per_image": 0.05}, 0.05),
|
||||
],
|
||||
)
|
||||
def test_a_seedream_deployment_priced_in_the_ui_bills_its_own_rate(
|
||||
deployment_price: dict[str, float], expected_cost: float
|
||||
) -> None:
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"created": 1792551825,
|
||||
"data": [{"output_format": "jpeg", "size": "1024x1024", "url": "https://cos.example/0.jpg"}],
|
||||
"tokenhub_usage": {"total_tokens": 11250},
|
||||
},
|
||||
)
|
||||
|
||||
router: Final = litellm.Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "seedream-flash",
|
||||
"litellm_params": {
|
||||
"model": "tencent/seedream-image-v5.0-flash",
|
||||
"api_key": "sk-tokenhub",
|
||||
**deployment_price,
|
||||
},
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
response: Final = router.image_generation(
|
||||
model="seedream-flash",
|
||||
prompt="a red paper lantern",
|
||||
client=HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond))),
|
||||
)
|
||||
|
||||
assert response._hidden_params["response_cost"] == pytest.approx(expected_cost)
|
||||
Loading…
Add table
Reference in a new issue