feat(tencent): add Seedream image generation, edits, and pricing through TokenHub
Some checks failed
LiteLLM Rust / rust-lint (push) Has been cancelled
LiteLLM Rust / rust-test (push) Has been cancelled
LiteLLM Rust / rust-wheel (push) Has been cancelled

This commit is contained in:
mateo-berri 2026-10-21 07:12:55 -07:00
commit a163ccf65e
14 changed files with 472 additions and 90 deletions

View file

@ -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}")

View file

@ -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')}"

View file

@ -6,7 +6,7 @@ from collections.abc import Mapping, Sequence
from dataclasses import dataclass
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
@ -2054,6 +2054,38 @@ 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) -> _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
return tencent_image_cost_calculator
case _:
return None
class CostCalculatorUtils:
@staticmethod
def call_type_has_image_response(call_type: str) -> bool:
@ -2100,9 +2132,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,
)
@ -2115,7 +2144,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)
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(
@ -2133,36 +2165,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,
@ -2206,16 +2208,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

View file

@ -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, ()

View file

@ -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,
}

View file

@ -1,6 +1,28 @@
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 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=model_info
)
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)

View 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

View 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

View file

@ -62929,6 +62929,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://www.tencentcloud.com/document/product/1300/78937",
"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.045,
"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,

View file

@ -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

View file

@ -62929,6 +62929,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://www.tencentcloud.com/document/product/1300/78937",
"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.045,
"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,

View file

@ -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,