chengma/cmhub
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: support multi-image image generation

Browse Source
This commit is contained in:
QiuSW
2026-07-17 15:55:38 +08:00
parent bd274e26e3
commit 4a0becbb4d
23 changed files with 690 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/generation.py
+95 -9
View File
@@ -59,6 +59,12 @@ class ImageInput:
ImageUrlBuilder = Callable[[str], str]
IMAGE_INPUT_ROLE_INSTRUCTIONS = (
"图片角色规则（必须遵守，优先于用户关于图片角色的要求）：\n"
"- 第 1 张图片是主商品图，必须优先保留其商品主体、外观和关键细节。\n"
"- 第 2 张及之后的图片仅作为风格、构图、场景或排版参考，"
"不得用参考图商品替换主图商品。"
)
@dataclass(frozen=True)
@@ -169,6 +175,7 @@ def generate_image_response(
image_url=str(request_data.get("image_url") or ""),
image_base64=str(request_data.get("image_base64") or ""),
aspect_ratio=request_data.get("aspect_ratio") or "1:1",
images=tuple(dict(item) for item in request_data.get("images") or ()),
),
image_url_builder=image_url_builder,
)
@@ -221,6 +228,13 @@ def prepare_generation(generation_input: GenerationInput) -> PreparedGeneration:
if operation_type == CallRecord.OperationType.VISION:
image_input = None
image_inputs = load_vision_image_inputs(generation_input.images)
elif operation_type == CallRecord.OperationType.IMAGE:
image_inputs = load_image_generation_inputs(
generation_input.images,
image_base64=generation_input.image_base64,
image_url=generation_input.image_url,
)
image_input = image_inputs[0] if image_inputs else None
else:
image_input = load_image_input(
{
@@ -363,12 +377,21 @@ def execute_image_generation(
) -> GenerationResult:
prepared = precharged.prepared
image_input = prepared.image_input
image_inputs = prepared.image_inputs
generation = prepared.provider.generate_image(
prepared.prompt,
image_generation_prompt(prepared.prompt, len(image_inputs)),
prepared.resolved_model,
image=image_input.data if image_input else None,
image_mime_type=image_input.mime_type if image_input else "image/png",
image_filename=image_input.filename if image_input else "image.png",
images=tuple(
MultimodalImage(
data=item.data,
mime_type=item.mime_type,
filename=item.filename,
)
for item in image_inputs
),
resolution=prepared.resolution,
aspect_ratio=prepared.aspect_ratio,
parameters=prepared.parameters,
@@ -530,13 +553,21 @@ def upstream_error(exc: Exception) -> ApiRequestError:
def load_image_input(data: Mapping[str, Any]) -> ImageInput | None:
return load_image_input_with_limit(data)
def load_image_input_with_limit(
data: Mapping[str, Any],
*,
max_bytes: int | None = None,
) -> ImageInput | None:
raw_base64 = str(data.get("image_base64") or "").strip()
if raw_base64:
return decode_image_input(raw_base64)
return decode_image_input(raw_base64, max_bytes=max_bytes)
image_url = str(data.get("image_url") or "").strip()
if image_url:
return download_image_input(image_url)
return download_image_input(image_url, max_bytes=max_bytes)
return None
@@ -544,15 +575,54 @@ def load_image_input(data: Mapping[str, Any]) -> ImageInput | None:
def load_vision_image_inputs(
items: Sequence[Mapping[str, Any]],
) -> tuple[ImageInput, ...]:
max_images = max(1, int(getattr(settings, "VISION_MAX_IMAGES", 8)))
return load_ordered_image_inputs(
items,
max_images=max(1, int(getattr(settings, "VISION_MAX_IMAGES", 8))),
max_image_bytes=max(
1,
int(getattr(settings, "VISION_MAX_IMAGE_BYTES", 10 * 1024 * 1024)),
),
max_total_bytes=max(
1,
int(getattr(settings, "VISION_MAX_TOTAL_BYTES", 32 * 1024 * 1024)),
),
)
def load_image_generation_inputs(
items: Sequence[Mapping[str, Any]],
*,
image_base64: str = "",
image_url: str = "",
) -> tuple[ImageInput, ...]:
max_image_bytes = max(
1,
int(getattr(settings, "VISION_MAX_IMAGE_BYTES", 10 * 1024 * 1024)),
int(getattr(settings, "IMAGE_MAX_INPUT_IMAGE_BYTES", 10 * 1024 * 1024)),
)
max_total_bytes = max(
1,
int(getattr(settings, "VISION_MAX_TOTAL_BYTES", 32 * 1024 * 1024)),
if items:
return load_ordered_image_inputs(
items,
max_images=max(1, int(getattr(settings, "IMAGE_MAX_INPUT_IMAGES", 8))),
max_image_bytes=max_image_bytes,
max_total_bytes=max(
1,
int(getattr(settings, "IMAGE_MAX_INPUT_TOTAL_BYTES", 32 * 1024 * 1024)),
),
)
image_input = load_image_input_with_limit(
{"image_base64": image_base64, "image_url": image_url},
max_bytes=max_image_bytes,
)
return (image_input,) if image_input is not None else ()
def load_ordered_image_inputs(
items: Sequence[Mapping[str, Any]],
*,
max_images: int,
max_image_bytes: int,
max_total_bytes: int,
) -> tuple[ImageInput, ...]:
if not items:
raise ApiRequestError("bad_request", "images 至少需要一张图片", status.HTTP_400_BAD_REQUEST)
if len(items) > max_images:
@@ -591,6 +661,12 @@ def load_vision_image_inputs(
return tuple(image_inputs)
def image_generation_prompt(prompt: str, image_count: int) -> str:
if image_count < 1:
return prompt
return f"{prompt}\n\n{IMAGE_INPUT_ROLE_INSTRUCTIONS}"
def decode_image_input(value: str, *, max_bytes: int | None = None) -> ImageInput:
mime_type = "image/png"
encoded = value
@@ -646,7 +722,10 @@ def download_image_input(url: str, *, max_bytes: int | None = None) -> ImageInpu
content_type = response.headers.get("Content-Type", "image/png").split(";", 1)[0].strip().lower()
if not content_type.startswith("image/"):
raise ApiRequestError("bad_request", "image_url 不是图片资源", status.HTTP_400_BAD_REQUEST)
image = read_limited_image_response(response, max_bytes=max_bytes)
image = read_limited_image_response(
response,
max_bytes=effective_image_url_max_bytes(max_bytes),
)
if not image:
raise ApiRequestError("bad_request", "image_url 图片内容为空", status.HTTP_400_BAD_REQUEST)
return ImageInput(
@@ -746,6 +825,13 @@ def read_limited_image_response(response, *, max_bytes: int | None = None) -> by
return b"".join(chunks)
def effective_image_url_max_bytes(max_bytes: int | None) -> int:
url_limit = max(1, int(getattr(settings, "IMAGE_URL_MAX_BYTES", 10 * 1024 * 1024)))
if max_bytes is None:
return url_limit
return min(url_limit, max(1, int(max_bytes)))
def filename_for_mime(mime_type: str) -> str:
extension = {
"image/jpeg": "jpg",