From 0a58b89919eb00b413e9417be18e5470bdae11fe Mon Sep 17 00:00:00 2001 From: hh <2587203630@qq.com> Date: Tue, 14 Jul 2026 15:18:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BC=98=E5=8C=96=E6=A8=A1=E7=89=B9?= =?UTF-8?q?=E4=B8=8A=E8=BA=AB=E5=9B=BE=E6=8F=90=E7=A4=BA=E8=AF=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/backend/.env.example | 7 + core/backend/airshelf/settings/base.py | 10 + .../ai/management/commands/run_tryon_ab.py | 271 +++++++ core/backend/apps/ai/services.py | 229 +++++- core/backend/apps/ai/test_tryon_ab.py | 180 +++++ core/backend/apps/ai/test_tryon_prompt.py | 333 +++++++++ core/backend/apps/ai/tests.py | 299 +++++++- core/backend/apps/ai/tryon_prompt.py | 659 ++++++++++++++++++ 8 files changed, 1973 insertions(+), 15 deletions(-) create mode 100644 core/backend/apps/ai/management/commands/run_tryon_ab.py create mode 100644 core/backend/apps/ai/test_tryon_ab.py create mode 100644 core/backend/apps/ai/test_tryon_prompt.py create mode 100644 core/backend/apps/ai/tryon_prompt.py diff --git a/core/backend/.env.example b/core/backend/.env.example index 4be2a95..cd12f60 100644 --- a/core/backend/.env.example +++ b/core/backend/.env.example @@ -30,3 +30,10 @@ YUNQI_BASE_URL=https://www.yunqiai.chat/v1 # 模特库三视图真实生成与计费开关：API 与 Celery worker 同步部署后才能开启。 # true 允许预留积分并投递生成任务；false 时提交接口直接拒绝，不产生任务或扣费。 MODEL_TRIVIEW_GENERATION_ENABLED=true + +# 模特上身图结构化提示词 V2.2： +# true = 所有团队的商品上身图使用 V2.2 提示词；只切提示词，不切换用户选择的 Seedream/GPT， +# 不改变 1/2/4 张、图片比例、积分价格、退款或资产落库。false = 仅白名单团队使用； +# false + 空白名单 = 全部使用旧提示词。修改后 API 与 Worker 必须使用同一组值并同步重启。 +MODEL_TRYON_PROMPT_V2_ENABLED=false +MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS= diff --git a/core/backend/airshelf/settings/base.py b/core/backend/airshelf/settings/base.py index 272b2a3..0d8968d 100644 --- a/core/backend/airshelf/settings/base.py +++ b/core/backend/airshelf/settings/base.py @@ -239,6 +239,16 @@ FREE_VIDEO_MAX_CONCURRENT = int(env("FREE_VIDEO_MAX_CONCURRENT", "3")) # 关闭时提交接口直接拒绝；开启后允许预留积分并把生成任务投递给 worker。 MODEL_TRIVIEW_GENERATION_ENABLED = env_bool("MODEL_TRIVIEW_GENERATION_ENABLED", False) +# 模特上身图结构化提示词 V2.2 全量开关： +# - true：所有团队的 mode=model 商品上身图使用 V2.2 提示词；不替用户切换 Seedream/GPT，不改变数量、比例或扣费； +# - false：仅 CANARY_TEAM_IDS 命中的团队使用 V2.2；白名单也为空时全部沿用旧提示词； +# 修改后必须同步重启 API 与 Celery Worker；紧急回退设为 false 并清空白名单。 +MODEL_TRYON_PROMPT_V2_ENABLED = env_bool("MODEL_TRYON_PROMPT_V2_ENABLED", False) +# Step 5.2 单团队试点：全局开关关闭时，仅命中该白名单的团队使用 V2；空值与无效值均不会误开全量。 +MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS = frozenset( + item.lower() for item in env_list("MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS") +) + # 火山引擎人像素材库审核(真人资产绿/红盾)· AK/SK 暂借 AirDrama 已邀测账号,张业昌待换成 AirShelf 自有 ASSETS_API = { "access_key": env("ASSETS_API_ACCESS_KEY", ""), diff --git a/core/backend/apps/ai/management/commands/run_tryon_ab.py b/core/backend/apps/ai/management/commands/run_tryon_ab.py new file mode 100644 index 0000000..3c3a37c --- /dev/null +++ b/core/backend/apps/ai/management/commands/run_tryon_ab.py @@ -0,0 +1,271 @@ +"""模特上身图 V2 同输入 A/B：默认只预估，显式 --execute 才创建计费任务。""" + +from __future__ import annotations + +import json +import uuid +from decimal import Decimal, InvalidOperation +from pathlib import Path +from time import monotonic + +import requests +from django.core.management.base import BaseCommand, CommandError + + +class Command(BaseCommand): + help = "对同一商品、模特、提示词和比例提交 Seedream/GPT 模特上身图 A/B；默认 dry-run" + + def add_arguments(self, parser): + parser.add_argument("--product", default="", help="商品 UUID") + parser.add_argument("--model-asset", default="", help="模特肖像 Asset UUID") + parser.add_argument("--model-entity", default="", help="可选：模特库 Model UUID，仅作溯源") + parser.add_argument("--prompt", default="", help="两组共同使用的用户提示词") + parser.add_argument("--ratio", default="3:4", help="两组共同使用的比例，默认 3:4") + parser.add_argument("--count", type=int, default=1, choices=(1, 2, 4), help="每个模型生成张数") + parser.add_argument( + "--variants", + nargs="+", + default=("volcano", "gpt-image"), + help="模型路由键，默认 volcano gpt-image", + ) + parser.add_argument("--max-points", default="", help="硬预算上限；--execute 时必填") + parser.add_argument("--experiment", default="", help="可选实验 UUID；默认自动生成") + parser.add_argument("--execute", action="store_true", help="真正预扣积分并提交任务") + parser.add_argument("--report", default="", help="只读查看指定实验 UUID 的任务与结果") + parser.add_argument("--export-dir", default="", help="配合 --report 匿名导出 A/B 图片与独立答案文件") + + def handle(self, *args, **opts): + if opts["report"]: + self._report(opts["report"], opts["export_dir"]) + return + + from apps.ai.services import resolve_image_model + from apps.assets.models import Asset + from apps.billing.models import CreditAccount + from apps.billing.pricing import quote_flat + from apps.products.models import Product + + if not opts["product"] or not opts["model_asset"] or not opts["prompt"].strip(): + raise CommandError("dry-run/execute 必须提供 --product、--model-asset 和 --prompt") + + product = Product.objects.select_related("team", "created_by", "team__owner").filter( + id=opts["product"], purged_at__isnull=True + ).first() + if product is None: + raise CommandError("商品不存在或已清理") + model_asset = Asset.objects.filter( + id=opts["model_asset"], + team=product.team, + is_deleted=False, + purged_at__isnull=True, + ).first() + if model_asset is None: + raise CommandError("模特肖像不存在、已删除或不属于商品团队") + from apps.ai.services import _asset_preview_url, _product_reference_urls + + product_reference_count = len(_product_reference_urls(product, limit=3)) + if product_reference_count < 1: + raise CommandError("商品没有可用的真实参考图，不能执行同输入上身图 A/B") + if not _asset_preview_url(model_asset): + raise CommandError("模特肖像没有可用图片文件") + user = product.created_by or product.team.owner + if user is None: + raise CommandError("商品团队没有可用于计费的提交用户") + + variants: list[tuple[str, object, Decimal]] = [] + seen: set[str] = set() + for key in opts["variants"]: + key = str(key).strip() + if not key or key in seen: + continue + seen.add(key) + model = resolve_image_model(key) + if model is None: + raise CommandError(f"找不到图像模型路由：{key}") + quote = quote_flat(model, team=product.team) + variants.append((key, model, quote.points)) + if len(variants) < 2: + raise CommandError("A/B 至少需要两个不同模型路由") + + total_points = sum((points * opts["count"] for _, _, points in variants), Decimal("0")) + account, _ = CreditAccount.objects.get_or_create(team=product.team) + available = account.balance - account.reserved_balance + experiment_id = self._experiment_id(opts["experiment"]) + + self.stdout.write(f"experiment={experiment_id}") + self.stdout.write( + f"product={product.id} {product.title} · model_asset={model_asset.id} · " + f"product_refs={product_reference_count} · ratio={opts['ratio']} · count/model={opts['count']}" + ) + for key, model, points in variants: + self.stdout.write( + f" {key}: {model.provider.name}:{model.name} · {points} 积分/张 · " + f"小计 {points * opts['count']}" + ) + self.stdout.write(f"总预算={total_points} 积分 · 当前可用={available} 积分") + + if not opts["execute"]: + self.stdout.write(self.style.WARNING("DRY-RUN：未创建任务、未预扣积分；加 --execute --max-points N 才会提交")) + return + + max_points = self._max_points(opts["max_points"]) + if total_points > max_points: + raise CommandError(f"预计 {total_points} 积分，超过硬预算 {max_points}，已拒绝提交") + if available < total_points: + raise CommandError(f"当前可用 {available} 积分，低于预计 {total_points}，已拒绝提交") + + from apps.ai.services import enqueue_standalone_images + + submitted = [] + for key, model, points in variants: + tasks = enqueue_standalone_images( + team=product.team, + user=user, + prompt=opts["prompt"].strip(), + mode="model", + count=opts["count"], + product_id=str(product.id), + model_id=str(model_asset.id), + model_entity_id=str(opts["model_entity"] or "") or None, + ratio=str(opts["ratio"] or "").strip() or None, + image_model=key, + tryon_prompt_v2_override=True, + tryon_ab={"experiment_id": experiment_id, "variant": key}, + dispatch=False, + ) + submitted.append((key, model, points, tasks)) + + for key, model, points, tasks in submitted: + batch_id = (tasks[0].request_payload or {}).get("batch_id") if tasks else "" + ids = ",".join(str(task.id) for task in tasks) + self.stdout.write( + self.style.SUCCESS( + f"RESERVED {key} · model={model.provider.name}:{model.name} · " + f"batch={batch_id} · tasks={ids} · reserved={points * len(tasks)}" + ) + ) + + # 不投递在线 Worker：它可能仍是未部署 V2 的旧版本。由当前命令进程串行执行，确保代码版本一致。 + from apps.ai.services import run_standalone_image_task + + for key, model, _points, tasks in submitted: + for task in tasks: + started = monotonic() + run_standalone_image_task(task_id=str(task.id)) + elapsed = round(monotonic() - started, 3) + task.refresh_from_db() + payload = dict(task.request_payload or {}) + ab = dict(payload.get("tryon_ab") or {}) + ab["elapsed_seconds"] = elapsed + payload["tryon_ab"] = ab + task.request_payload = payload + task.save(update_fields=["request_payload", "updated_at"]) + assets = ",".join(str(asset.id) for asset in task.generated_assets.all()) or "-" + self.stdout.write( + f"FINISHED {key} index={payload.get('index')} status={task.status} " + f"elapsed={elapsed}s actual={task.actual_cost} asset={assets} " + f"error={task.error_code or '-'}" + ) + + def _max_points(self, raw: str) -> Decimal: + if not str(raw or "").strip(): + raise CommandError("--execute 必须同时提供正数 --max-points") + try: + value = Decimal(str(raw)) + except InvalidOperation as exc: + raise CommandError("--max-points 必须是数字") from exc + if value <= 0: + raise CommandError("--max-points 必须大于 0") + return value + + def _experiment_id(self, raw: str) -> str: + if not str(raw or "").strip(): + return str(uuid.uuid4()) + try: + return str(uuid.UUID(str(raw))) + except ValueError as exc: + raise CommandError("--experiment 必须是合法 UUID") from exc + + def _report(self, experiment_id: str, export_dir: str = "") -> None: + from apps.ai.models import AITask + + experiment_id = self._experiment_id(experiment_id) + tasks = list( + AITask.objects.filter(request_payload__tryon_ab__experiment_id=experiment_id) + .select_related("model_config", "model_config__provider") + .prefetch_related("generated_assets", "generated_assets__files") + .order_by("created_at") + ) + if not tasks: + raise CommandError("找不到该 A/B 实验") + self.stdout.write(f"experiment={experiment_id} · tasks={len(tasks)}") + for task in tasks: + payload = task.request_payload or {} + ab = payload.get("tryon_ab") or {} + trace = payload.get("tryon_prompt") or {} + assets = ",".join(str(asset.id) for asset in task.generated_assets.all()) or "-" + self.stdout.write( + f" {ab.get('variant')} index={payload.get('index')} status={task.status} " + f"model={task.model_config.provider.name}:{task.model_config.name} " + f"estimated={task.estimated_cost} actual={task.actual_cost} " + f"prompt_v2={trace.get('applied')} elapsed={ab.get('elapsed_seconds')}s " + f"asset={assets} error={task.error_code or '-'}" + ) + if export_dir: + self._blind_export(experiment_id, tasks, export_dir) + + def _blind_export(self, experiment_id: str, tasks: list, export_dir: str) -> None: + from apps.ai.services import _asset_preview_url + + output = Path(export_dir).expanduser().resolve() + output.mkdir(parents=True, exist_ok=True) + variants = sorted({str((task.request_payload.get("tryon_ab") or {}).get("variant") or "") for task in tasks}) + variants = [variant for variant in variants if variant] + if len(variants) < 2: + raise CommandError("匿名导出至少需要两个有效 A/B 变体") + # 实验 UUID 决定匿名标签顺序；报告正文不输出映射，评分完成后再单独读取 answer_key.json。 + if uuid.UUID(experiment_id).int % 2: + variants.reverse() + labels = {variant: chr(ord("A") + index) for index, variant in enumerate(variants)} + manifest = {"experiment_id": experiment_id, "samples": []} + answer_key = {"experiment_id": experiment_id, "labels": {label: variant for variant, label in labels.items()}} + + for task in tasks: + payload = task.request_payload or {} + variant = str((payload.get("tryon_ab") or {}).get("variant") or "") + label = labels.get(variant) + asset = task.generated_assets.first() + if not label or asset is None: + continue + url = _asset_preview_url(asset) + if not url: + raise CommandError(f"任务 {task.id} 的成图没有可下载地址") + primary = asset.files.filter(is_primary=True).first() or asset.files.first() + content_type = str(getattr(primary, "content_type", "") or "") + suffix = ".jpg" if "jpeg" in content_type else (".webp" if "webp" in content_type else ".png") + shot_index = int(payload.get("index") or 0) + filename = f"{label}_{shot_index + 1}{suffix}" + response = requests.get(url, timeout=60) + response.raise_for_status() + (output / filename).write_bytes(response.content) + manifest["samples"].append( + { + "label": label, + "shot_index": shot_index, + "filename": filename, + "task_id": str(task.id), + "asset_id": str(asset.id), + } + ) + + (output / "blind_manifest.json").write_text( + json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (output / "answer_key.json").write_text( + json.dumps(answer_key, ensure_ascii=False, indent=2), encoding="utf-8" + ) + self.stdout.write( + self.style.SUCCESS( + f"BLIND_EXPORT samples={len(manifest['samples'])} dir={output} · 评分前不要读取 answer_key.json" + ) + ) diff --git a/core/backend/apps/ai/services.py b/core/backend/apps/ai/services.py index 04d2423..dc6cf09 100644 --- a/core/backend/apps/ai/services.py +++ b/core/backend/apps/ai/services.py @@ -779,14 +779,29 @@ def _ratio_to_image_size(ratio: str) -> str: 这里只能取最接近的「竖/横/方」近似;要拿到精确像素比例需在出图后做主体保护裁切/补边(见优化文档 §7.1 normalize_output_image,属待部署的后处理项)。修复点:4:5 原会 fallback 成 1024x1024(方图)→ 比例错误, 现归到竖图 1024x1536。""" - return { + known = { "1:1": "1024x1024", "3:4": "1024x1536", # 近似竖图(网关无精确 3:4) "4:5": "1024x1536", # 近似竖图(原 fallback 成方图是比例 bug) "9:16": "1024x1536", # 近似竖图(网关无精确 9:16) "4:3": "1536x1024", "16:9": "1536x864", # 真 16:9(原来误用 1536x1024 = 3:2) - }.get((ratio or "").strip(), "1024x1024") + } + normalized = (ratio or "").strip() + if normalized in known: + return known[normalized] + # 手动宽高比不能被静默当成方图。GPT 网关不接受任意精确尺寸时，至少保持横/竖方向； + # 原始比例仍写进有效提示词，供应商获得的是它支持的最近画布。 + try: + width, height = (float(part.strip()) for part in normalized.split(":", 1)) + if width > 0 and height > 0: + if width < height: + return "1024x1536" + if width > height: + return "1536x1024" + except (TypeError, ValueError): + pass + return "1024x1024" def _ratio_to_volcano_size(ratio: str) -> str: @@ -990,6 +1005,93 @@ def build_model_tryon_prompt_refs(product, has_model: bool, base_prompt: str = " return " ".join(lines) +def _model_tryon_prompt_v2_rollout_source(*, team_id, payload: dict) -> str | None: + """返回 V2.2 启用来源；未命中时失败关闭并继续使用旧提示词。""" + + if payload.get("tryon_prompt_v2_override") is True: + return "internal_ab" + if getattr(settings, "MODEL_TRYON_PROMPT_V2_ENABLED", False): + return "global" + + raw_allowlist = getattr(settings, "MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS", ()) or () + if isinstance(raw_allowlist, str): + raw_allowlist = raw_allowlist.split(",") + allowed_team_ids = { + str(value).strip().lower() + for value in raw_allowlist + if str(value).strip() + } + team_key = str(team_id or "").strip().lower() + return "canary" if team_key and team_key in allowed_team_ids else None + + +def _build_model_tryon_prompt_v2( + *, + product, + payload: dict, + has_model: bool, + index: int, + n_product: int, + rollout_source: str, +): + """使用任务创建时的分类快照构建单张有效提示词，并返回可追溯信息。""" + + from apps.ai.tryon_prompt import ( + ClassificationResult, + ProductContext, + build_tryon_prompt_plan, + classify_product_details, + default_ratio_for_kind, + ) + + batch_count = int(payload.get("tryon_batch_count") or 0) + if batch_count not in (1, 2, 4): + raise ValueError("unsupported_tryon_batch_count") + if index < 0 or index >= batch_count: + raise ValueError("tryon_index_out_of_range") + + context = ProductContext.create( + title=getattr(product, "title", "") or "商品", + category=getattr(product, "category", "") or "", + description=getattr(product, "description", "") or "", + selling_points=tuple(product.selling_points.values_list("title", flat=True)[:8]), + ) + classification = ClassificationResult.from_payload(payload.get("tryon_classification")) + if classification is None: + classification = classify_product_details(context, str(payload.get("prompt") or "")) + + requested_ratio = str(payload.get("ratio") or "").strip() + resolved_ratio = requested_ratio or default_ratio_for_kind(classification.kind) + plan = build_tryon_prompt_plan( + context=context, + user_prompt=str(payload.get("prompt") or ""), + count=batch_count, + ratio=resolved_ratio, + product_reference_count=n_product, + has_model_portrait=has_model, + has_model_triview=False, + classification=classification, + ) + prompt = plan.prompts[index] + trace = { + "version": "v2.2", + "applied": True, + "rollout_source": rollout_source, + "effective_prompt": prompt, + "shot_index": index, + "batch_count": batch_count, + "requested_ratio": requested_ratio or None, + "resolved_ratio": resolved_ratio, + "ratio_source": "user" if requested_ratio else "default", + "reference_roles": { + "product_numbers": list(plan.references.product_numbers), + "model_portrait_number": plan.references.model_portrait_number, + "model_triview_number": plan.references.model_triview_number, + }, + } + return prompt, trace, resolved_ratio + + # 平台套图同批多张要「同款商品、不同版式」,否则 N 张文案/构图雷同(PMC#24)。 # 锁死商品一致性,只让排版/构图/视角/配色基调按张变化。 # 注:仅纯文生图回落路径(无参考图)仍用这个简表;refs 版改用下面的 slot 体系。 @@ -2649,7 +2751,7 @@ def _reap_stale_standalone_image_tasks(*, team) -> None: continue -def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", count: int = 1, product_id: str | None = None, reference_product: bool = False, model_id: str | None = None, model_entity_id: str | None = None, ratio: str | None = None, image_model: str | None = None, conversation=None, reference_image_ids: list[str] | None = None, platform_id: str | None = None, batch_id: str | None = None) -> list[AITask]: +def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", count: int = 1, product_id: str | None = None, reference_product: bool = False, model_id: str | None = None, model_entity_id: str | None = None, ratio: str | None = None, image_model: str | None = None, conversation=None, reference_image_ids: list[str] | None = None, platform_id: str | None = None, batch_id: str | None = None, tryon_prompt_v2_override: bool = False, tryon_ab: dict | None = None, dispatch: bool = True) -> list[AITask]: """独立生图(图片创作 / 模特上身图 / 平台套图)改为**异步**:本函数在 Web 请求里只做「建任务 + 预留额度」这种秒级的活,真正 ~30s 的 ARK 出图交给 Celery worker(generate_standalone_image_task)。 @@ -2666,10 +2768,10 @@ def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", c task_type = _STANDALONE_TASK_TYPE.get(mode, AITask.Type.PRODUCT_IMAGE) count = max(1, min(int(count or 1), 12)) ref_ids = [str(r) for r in (reference_image_ids or []) if r] - # 带参考图的图片创作必须走 image_edit(gpt-image)才真的还原上传素材——火山 Seedream 的图生图只松散 - # 借色调、不锁主体(与三视图/模特上身图同理,那两条流程也强制 image_edit)。当前模型不支持 image_edit - # 就自动切到 gpt-image-2,否则用户选了火山却传了参考图 → 生成完全不参考素材(用户实测的根因)。 - if ref_ids and not hasattr(build_provider(model_config), "image_edit"): + # 普通图片创作带用户上传参考图时，当前模型不支持 image_edit 就沿用既有保护切到 gpt-image。 + # 模特上身图例外：商品图和模特图由专用 Worker 作为多参考图传入 Seedream/GPT，必须尊重用户自选模型； + # 即使旧客户端额外带了 reference_image_ids，也不得在任务创建阶段静默覆盖 image_model。 + if ref_ids and mode != "model" and not hasattr(build_provider(model_config), "image_edit"): alt = resolve_image_model("gpt-image") if alt is not None: model_config = alt @@ -2691,10 +2793,49 @@ def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", c platform_name = _PLATFORM_NAMES.get(platform_key, "") from apps.billing.pricing import quote_flat + # Step 2.1:只为“模特上身图 + 商品”记录一次确定性分类快照。这里不读取图片、不调用模型； + # Step 3 的 Worker 根据 MODEL_TRYON_PROMPT_V2_ENABLED 决定使用 V2 或旧提示词。 + tryon_classification: dict[str, str | None] | None = None + if mode == "model" and product_id: + from apps.ai.tryon_prompt import ClassificationResult, ProductContext, ProductKind, classify_product_details + from apps.products.models import Product + + product_for_classification = Product.objects.filter(id=product_id, team=team).only( + "title", "category", "description" + ).first() + if product_for_classification is None: + classification = ClassificationResult( + ProductKind.UNKNOWN, + "fallback", + None, + "product_not_found", + ) + else: + classification = classify_product_details( + ProductContext.create( + title=product_for_classification.title, + category=product_for_classification.category, + description=product_for_classification.description, + ), + prompt, + ) + tryon_classification = classification.as_payload() + tasks: list[AITask] = [] for index in range(count): quote = quote_flat(model_config, team=team) request_payload = {"model": model_config.name, "endpoint": model_config.endpoint, "prompt": prompt, "mode": mode, "index": index, "product_id": str(product_id) if product_id else None, "reference_product": bool(reference_product), "model_id": str(model_id) if model_id else None, "model_entity_id": str(model_entity_id) if model_entity_id else None, "batch_id": batch_id, "ratio": str(ratio) if ratio else None, "reference_image_ids": ref_ids, "platform_id": platform_key or None, "platform_name": platform_name or None} + if tryon_classification is not None: + request_payload["tryon_classification"] = dict(tryon_classification) + request_payload["tryon_batch_count"] = count + # 仅供内部 A/B 工具使用；GenerateImageView 不接收这两个参数，用户请求无法绕过全局开关。 + if tryon_prompt_v2_override: + request_payload["tryon_prompt_v2_override"] = True + if tryon_ab: + request_payload["tryon_ab"] = { + "experiment_id": str(tryon_ab.get("experiment_id") or ""), + "variant": str(tryon_ab.get("variant") or ""), + } # 只在重跑/补图时落键(不落 False):workbench 用 KeyTextTransform 抽文本,"false" 字符串也是真值,会误判 if is_append: request_payload["batch_append"] = True @@ -2719,8 +2860,9 @@ def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", c task.save(update_fields=["status", "updated_at"]) tasks.append(task) # 额度都预留成功后再统一派发,避免"派发了任务但后面某张预留失败"的半成品状态 - for task in tasks: - generate_standalone_image_task.delay(str(task.id)) + if dispatch: + for task in tasks: + generate_standalone_image_task.delay(str(task.id)) return tasks @@ -2732,7 +2874,7 @@ def run_standalone_image_task(*, task_id: str) -> None: return team = task.team user = task.created_by - payload = task.request_payload or {} + payload = dict(task.request_payload or {}) prompt = str(payload.get("prompt") or "") mode = str(payload.get("mode") or "image") index = int(payload.get("index") or 0) @@ -2755,7 +2897,7 @@ def run_standalone_image_task(*, task_id: str) -> None: if product_id: from apps.products.models import Product - product = Product.objects.filter(id=product_id).first() + product = Product.objects.filter(id=product_id, team=team).first() can_edit = hasattr(provider, "image_edit") model_url = "" if payload.get("model_id"): @@ -2777,13 +2919,49 @@ def run_standalone_image_task(*, task_id: str) -> None: # 参考图收集与 provider 无关:先把「该用哪些参考图 + 哪条提示词」定下来,再按模型能力选调用方式。 edit_images: list[str] = [] edit_prompt = "" + output_ratio = str(payload.get("ratio") or "") + tryon_prompt_trace: dict | None = None + tryon_rollout_source = ( + _model_tryon_prompt_v2_rollout_source(team_id=team.id, payload=payload) + if mode == "model" + else None + ) if mode == "model" and product_urls: # 模特上身图:参考图1~N=商品真实图(多角度),参考图N+1=模特(模特图可缺则让模型自取真人模特) edit_images = product_urls + ([model_url] if model_url else []) - edit_prompt = build_model_tryon_prompt_refs( + legacy_prompt = build_model_tryon_prompt_refs( product, has_model=bool(model_url), base_prompt=prompt, index=index, n_product=len(product_urls), ) + edit_prompt = legacy_prompt + if tryon_rollout_source is not None: + try: + edit_prompt, tryon_prompt_trace, output_ratio = _build_model_tryon_prompt_v2( + product=product, + payload=payload, + has_model=bool(model_url), + index=index, + n_product=len(product_urls), + rollout_source=tryon_rollout_source, + ) + except Exception as exc: # noqa: BLE001 — 规划器异常时安全回落旧提示词，不让已预扣任务悬空 + tryon_prompt_trace = { + "version": "v2.2", + "applied": False, + "rollout_source": tryon_rollout_source, + "effective_prompt": legacy_prompt, + "shot_index": index, + "batch_count": payload.get("tryon_batch_count"), + "requested_ratio": output_ratio or None, + "resolved_ratio": output_ratio or None, + "ratio_source": "user" if output_ratio else None, + "fallback_reason": str(exc) if isinstance(exc, ValueError) else type(exc).__name__, + "reference_roles": { + "product_numbers": list(range(1, len(product_urls) + 1)), + "model_portrait_number": len(product_urls) + 1 if model_url else None, + "model_triview_number": None, + }, + } elif mode == "cover" and product_urls: # 平台套图:参考图1~N=商品真实图(多角度,_product_reference_urls 已优先真实上传图/排除 AI 图), # 有模特则参考图N+1=模特(锁人脸/身形)。platform_id 注入平台版式块(优化版)。 @@ -2804,19 +2982,42 @@ def run_standalone_image_task(*, task_id: str) -> None: # 否则图生图只会松散借个色调、不真的还原上传的素材(=用户反馈的「没参考我的图」)。 edit_images = ref_urls edit_prompt = build_free_reference_prompt(prompt, n_refs=len(ref_urls), index=index) + elif mode == "model" and product_id and tryon_rollout_source is not None: + # 没有真实商品参考图时维持原纯文生图行为，但明确留下未应用原因，不能伪装成 V2 已生效。 + tryon_prompt_trace = { + "version": "v2.2", + "applied": False, + "rollout_source": tryon_rollout_source, + "effective_prompt": prompt, + "shot_index": index, + "batch_count": payload.get("tryon_batch_count"), + "requested_ratio": output_ratio or None, + "resolved_ratio": output_ratio or None, + "ratio_source": "user" if output_ratio else None, + "fallback_reason": "no_product_reference", + "reference_roles": { + "product_numbers": [], + "model_portrait_number": None, + "model_triview_number": None, + }, + } use_edit = bool(edit_images) try: + if tryon_prompt_trace is not None: + payload["tryon_prompt"] = tryon_prompt_trace + task.request_payload = payload + task.save(update_fields=["request_payload", "updated_at"]) if use_edit and can_edit: # gpt-image 等支持 image_edit(多图参考编辑接口) if payload.get("reference_product"): size = "1536x1024" # 三视图固定横向 else: - size = _ratio_to_image_size(str(payload.get("ratio") or "")) # 模特图按选中比例 + size = _ratio_to_image_size(output_ratio) # 模特图按用户比例或 V2 默认比例 response = provider.image_edit(model=model_config.name, prompt=edit_prompt, images=edit_images, size=size) elif use_edit: # 火山 Seedream 无 image_edit:走 image_generation 带 image=参考图(图生图多参考); # 尺寸按选中比例换算成火山可接受的 ~2K 尺寸(三视图固定横向) - vsize = "2304x1728" if payload.get("reference_product") else _ratio_to_volcano_size(str(payload.get("ratio") or "")) + vsize = "2304x1728" if payload.get("reference_product") else _ratio_to_volcano_size(output_ratio) response = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=edit_prompt, image=edit_images, size=vsize) else: # 纯文生图同批多张要不同构图,否则雷同(PMC#24);index>0 追加换版式指令。 diff --git a/core/backend/apps/ai/test_tryon_ab.py b/core/backend/apps/ai/test_tryon_ab.py new file mode 100644 index 0000000..da81d83 --- /dev/null +++ b/core/backend/apps/ai/test_tryon_ab.py @@ -0,0 +1,180 @@ +"""Step 4 模特上身图 A/B 命令：默认零消费、预算硬门槛与实验留痕。""" + +from io import BytesIO, StringIO +from unittest.mock import MagicMock, patch + +from django.core.management import call_command +from django.core.management.base import CommandError +from django.test import TestCase + +from apps.accounts.models import Team, User +from apps.ai.models import AITask, ModelConfig, ModelProvider +from apps.assets.models import Asset, AssetFile +from apps.billing.models import CreditAccount +from apps.products.models import Product, ProductImage + + +class TryonABCommandTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="tryon-ab", password="pass") + self.team = Team.objects.create(name="Tryon AB", owner=self.user) + self.account = CreditAccount.objects.create(team=self.team, balance="1000.0000") + volcano, _ = ModelProvider.objects.get_or_create( + name="volcengine", + defaults={"display_name": "火山", "status": ModelProvider.Status.ACTIVE}, + ) + gpt_provider, _ = ModelProvider.objects.get_or_create( + name="tryon-ab-gpt", + defaults={"display_name": "A/B GPT", "status": ModelProvider.Status.ACTIVE}, + ) + ModelConfig.objects.get_or_create( + provider=volcano, + name="doubao-seedream-5-0-test", + capability=ModelConfig.Capability.IMAGE, + defaults={"display_name": "Seedream test", "unit_price": "10.0000"}, + ) + ModelConfig.objects.get_or_create( + provider=gpt_provider, + name="gpt-image-2", + capability=ModelConfig.Capability.IMAGE, + defaults={"display_name": "GPT Image test", "unit_price": "20.0000"}, + ) + self.product = Product.objects.create( + team=self.team, + created_by=self.user, + title="女装裤子", + category="服饰内衣", + ) + product_asset = Asset.objects.create( + team=self.team, + created_by=self.user, + name="裤装实拍", + asset_type=Asset.Type.IMAGE, + source=Asset.Source.UPLOAD, + category=Asset.Category.PRODUCT_IMAGE, + ) + AssetFile.objects.create( + asset=product_asset, + object_key="pants.png", + bucket="b", + content_type="image/png", + preview_url="http://x/pants.png", + is_primary=True, + ) + ProductImage.objects.create(product=self.product, asset=product_asset, is_primary=True) + self.model_asset = Asset.objects.create( + team=self.team, + created_by=self.user, + name="模特肖像", + asset_type=Asset.Type.IMAGE, + source=Asset.Source.AI_GENERATED, + category=Asset.Category.MODEL_PORTRAIT, + ) + AssetFile.objects.create( + asset=self.model_asset, + object_key="model.png", + bucket="b", + content_type="image/png", + preview_url="http://x/model.png", + is_primary=True, + ) + + def _base_options(self): + return { + "product": str(self.product.id), + "model_asset": str(self.model_asset.id), + "prompt": "全身站姿，双腿完整露出，浅灰纯色背景", + "ratio": "3:4", + "count": 1, + } + + def test_default_is_dry_run_and_creates_no_task_or_reservation(self): + out = StringIO() + call_command("run_tryon_ab", stdout=out, **self._base_options()) + text = out.getvalue() + self.assertIn("volcano", text) + self.assertIn("gpt-image", text) + self.assertIn("DRY-RUN", text) + self.assertEqual(AITask.objects.count(), 0) + self.account.refresh_from_db() + self.assertEqual(float(self.account.reserved_balance), 0.0) + + def test_execute_requires_explicit_budget_and_rejects_low_cap_before_submit(self): + with self.assertRaisesMessage(CommandError, "--max-points"): + call_command("run_tryon_ab", execute=True, **self._base_options()) + self.assertEqual(AITask.objects.count(), 0) + + with self.assertRaisesMessage(CommandError, "超过硬预算"): + call_command("run_tryon_ab", execute=True, max_points="1", **self._base_options()) + self.assertEqual(AITask.objects.count(), 0) + + def test_missing_real_product_reference_is_rejected(self): + ProductImage.objects.filter(product=self.product).delete() + self.product.cover_asset = None + self.product.save(update_fields=["cover_asset"]) + with self.assertRaisesMessage(CommandError, "没有可用的真实参考图"): + call_command("run_tryon_ab", **self._base_options()) + + def test_execute_runs_both_provider_shapes_inline_and_report_is_traceable(self): + seedream = MagicMock() + del seedream.image_edit # Seedream 真实能力形态：多参考图走 image_generation(image=[...]) + seedream.image_generation.return_value = {"data": [{"url": "http://x/seedream-out.png"}]} + seedream.extract_first_media_url.return_value = "http://x/seedream-out.png" + gpt = MagicMock() + gpt.image_edit.return_value = {"data": [{"url": "http://x/gpt-out.png"}]} + gpt.extract_first_media_url.return_value = "http://x/gpt-out.png" + + def provider_for(model_config): + return seedream if model_config.provider.name in {"volcengine", "volcano", "ark"} else gpt + + stored = MagicMock(object_key="ab.png", bucket="b", content_type="image/png", size_bytes=3) + experiment_id = "b83d65d5-cc98-43a6-a7cc-17c167366f0a" + out = StringIO() + with ( + patch("apps.ai.services.get_image_provider", side_effect=provider_for), + patch("apps.ai.services.VolcanoArkProvider.media_to_bytes", return_value=(BytesIO(b"img"), "image/png")), + patch("apps.ai.services.TosStorage") as storage, + ): + storage.return_value.upload_fileobj.return_value = stored + call_command( + "run_tryon_ab", + execute=True, + max_points="100", + experiment=experiment_id, + stdout=out, + **self._base_options(), + ) + + tasks = list(AITask.objects.order_by("created_at")) + self.assertEqual(len(tasks), 2) + self.assertTrue(all(task.status == AITask.Status.SUCCEEDED for task in tasks)) + self.assertTrue(all(task.request_payload["tryon_prompt"]["applied"] for task in tasks)) + self.assertTrue( + all( + task.request_payload["tryon_prompt"]["rollout_source"] == "internal_ab" + for task in tasks + ) + ) + self.assertTrue(all(task.request_payload["tryon_ab"]["experiment_id"] == experiment_id for task in tasks)) + self.assertTrue(all(task.request_payload["tryon_ab"]["elapsed_seconds"] >= 0 for task in tasks)) + self.assertEqual( + tasks[0].request_payload["tryon_prompt"]["effective_prompt"], + tasks[1].request_payload["tryon_prompt"]["effective_prompt"], + ) + self.assertEqual(tasks[0].generated_assets.count(), 1) + self.assertEqual(tasks[1].generated_assets.count(), 1) + seedream.image_generation.assert_called_once() + self.assertEqual(len(seedream.image_generation.call_args.kwargs["image"]), 2) + gpt.image_edit.assert_called_once() + self.assertEqual(len(gpt.image_edit.call_args.kwargs["images"]), 2) + self.account.refresh_from_db() + self.assertEqual(float(self.account.balance), 970.0) + self.assertEqual(float(self.account.reserved_balance), 0.0) + + report = StringIO() + call_command("run_tryon_ab", report=experiment_id, stdout=report) + report_text = report.getvalue() + self.assertIn("tasks=2", report_text) + self.assertIn("volcano", report_text) + self.assertIn("gpt-image", report_text) + self.assertIn("prompt_v2=True", report_text) diff --git a/core/backend/apps/ai/test_tryon_prompt.py b/core/backend/apps/ai/test_tryon_prompt.py new file mode 100644 index 0000000..221680d --- /dev/null +++ b/core/backend/apps/ai/test_tryon_prompt.py @@ -0,0 +1,333 @@ +"""Step 2 / 2.1 模特上身图提示词规划器与分类留痕测试。""" + +import json +from unittest.mock import patch + +from django.test import SimpleTestCase, TestCase + +from apps.ai.tryon_prompt import ( + ClassificationResult, + ProductContext, + ProductKind, + ReferenceRoles, + build_tryon_prompt_plan, + classify_product, + classify_product_details, + default_ratio_for_kind, +) + + +class ProductClassificationTests(SimpleTestCase): + def test_cross_category_classification(self): + cases = ( + ("中蓝色高腰宽腿牛仔裤", "服饰内衣", ProductKind.LOWER), + ("莫代尔无痕内衣", "服饰内衣", ProductKind.UNDERWEAR), + ("法式收腰连衣裙", "女装", ProductKind.FULL_BODY), + ("复古跑步鞋", "鞋靴", ProductKind.FOOTWEAR), + ("智能运动手表", "数码配件", ProductKind.WRIST), + ("头戴式主动降噪耳机", "数码", ProductKind.HEAD_AUDIO), + ("真无线蓝牙耳机", "数码", ProductKind.EAR_AUDIO), + ("通勤斜挎包", "箱包", ProductKind.BAG), + ("便携无线小风扇", "家电", ProductKind.HANDHELD), + ("桌面收纳盒", "家居", ProductKind.NONWEARABLE), + ) + for title, category, expected in cases: + with self.subTest(title=title): + context = ProductContext.create(title=title, category=category) + self.assertEqual(classify_product(context), expected) + + def test_rule_outside_products_use_safe_compensation(self): + for title in ("羊绒围巾", "中筒袜子", "真皮腰带", "保暖手套"): + with self.subTest(title=title): + result = classify_product_details(ProductContext.create(title=title)) + self.assertEqual(result.kind, ProductKind.WEARABLE_GENERIC) + self.assertEqual(result.source, "title_description") + self.assertEqual(result.fallback_reason, "no_specific_product_rule") + + def test_broad_category_does_not_force_underwear(self): + result = classify_product_details( + ProductContext.create(title="春季神秘新品", category="服饰内衣") + ) + self.assertEqual(result.kind, ProductKind.WEARABLE_GENERIC) + self.assertEqual(result.source, "category") + self.assertEqual(result.fallback_reason, "broad_category_only") + + def test_user_relation_overrides_only_unclassified_product(self): + unknown_context = ProductContext.create(title="多功能新品", category="其他") + handheld = classify_product_details(unknown_context, "让模特手持商品进行正面展示") + self.assertEqual(handheld.kind, ProductKind.HANDHELD) + self.assertEqual(handheld.source, "user_prompt") + self.assertEqual(handheld.matched_rule, "relation:手持") + + known_watch = classify_product_details( + ProductContext.create(title="智能运动手表"), + "模特手持商品", + ) + self.assertEqual(known_watch.kind, ProductKind.WRIST) + self.assertEqual(known_watch.source, "title_description") + + def test_unknown_is_explicit_and_payload_is_json_safe(self): + result = classify_product_details(ProductContext.create(title="X7 多功能新品")) + self.assertEqual(result.kind, ProductKind.UNKNOWN) + self.assertEqual(result.source, "fallback") + self.assertEqual(result.fallback_reason, "no_specific_rule_matched") + payload = result.as_payload() + self.assertEqual(payload["kind"], "unknown") + self.assertEqual(payload["rule_version"], "v2.1") + self.assertEqual(json.loads(json.dumps(payload, ensure_ascii=False)), payload) + + def test_classification_payload_never_contains_enum_object(self): + payload = ClassificationResult(ProductKind.LOWER, "title_description", "lower:裤").as_payload() + self.assertIsInstance(payload["kind"], str) + self.assertFalse(any(isinstance(value, ProductKind) for value in payload.values())) + + def test_classification_payload_round_trip_and_invalid_fallback(self): + original = ClassificationResult(ProductKind.BAG, "title_description", "bag:斜挎包") + restored = ClassificationResult.from_payload(original.as_payload()) + self.assertEqual(restored, original) + self.assertIsNone(ClassificationResult.from_payload({"kind": "not-real", "source": "fallback"})) + self.assertIsNone(ClassificationResult.from_payload({"kind": "bag"})) + + def test_default_ratio_only_fills_missing_user_choice(self): + self.assertEqual(default_ratio_for_kind(ProductKind.LOWER), "3:4") + self.assertEqual(default_ratio_for_kind(ProductKind.BAG), "3:4") + self.assertEqual(default_ratio_for_kind(ProductKind.UNKNOWN), "1:1") + + +class ReferenceRoleTests(SimpleTestCase): + def test_single_product_then_model_portrait(self): + roles = ReferenceRoles.create(product_count=1, has_model_portrait=True) + self.assertEqual(roles.product_numbers, (1,)) + self.assertEqual(roles.model_portrait_number, 2) + self.assertIsNone(roles.model_triview_number) + self.assertIn("参考图1是同一商品", roles.instruction()) + self.assertIn("参考图2（模特肖像）", roles.instruction()) + + def test_three_products_portrait_and_triview(self): + roles = ReferenceRoles.create( + product_count=3, + has_model_portrait=True, + has_model_triview=True, + ) + self.assertEqual(roles.product_numbers, (1, 2, 3)) + self.assertEqual(roles.model_portrait_number, 4) + self.assertEqual(roles.model_triview_number, 5) + self.assertIn("参考图1-3", roles.instruction()) + + +class TryonPromptPlanTests(SimpleTestCase): + def _plan(self, *, title="女装裤子", count=4, ratio="1:1", prompt="电商模特上身图", **kwargs): + return build_tryon_prompt_plan( + context=ProductContext.create(title=title, category=kwargs.pop("category", "服饰内衣")), + user_prompt=prompt, + count=count, + ratio=ratio, + product_reference_count=kwargs.pop("product_reference_count", 1), + **kwargs, + ) + + def test_count_supports_one_two_and_four(self): + for count in (1, 2, 4): + with self.subTest(count=count): + plan = self._plan(count=count) + self.assertEqual(len(plan.shots), count) + self.assertEqual(len(plan.prompts), count) + with self.assertRaisesMessage(ValueError, "只支持 1、2、4"): + self._plan(count=3) + + def test_preset_and_custom_ratio_are_preserved(self): + for ratio in ("1:1", "3:4", "9:16", "7:10"): + with self.subTest(ratio=ratio): + plan = self._plan(count=1, ratio=ratio) + self.assertEqual(plan.ratio, ratio) + self.assertIn(f"严格使用 {ratio} 比例", plan.prompts[0]) + + def test_lower_garment_shots_never_use_conflicting_templates(self): + plan = self._plan(count=4) + combined = "\n".join(plan.shots) + for forbidden in ("半身", "手持", "领口", "袖型", "生活场景"): + self.assertNotIn(forbidden, combined) + self.assertTrue(all("全身站姿" in shot for shot in plan.shots)) + self.assertTrue(all("从头到脚完整入镜" in shot for shot in plan.shots)) + + def test_user_hard_requirement_precedes_default_shot(self): + hard_requirement = "全身站姿，双腿完整露出，浅灰纯色背景" + prompt = self._plan(count=1, prompt=hard_requirement).prompts[0] + self.assertLess(prompt.index("用户硬性要求"), prompt.index("本张镜头")) + self.assertIn(hard_requirement, prompt) + self.assertIn("用户原文若使用了", prompt) + + def test_v22_locks_non_target_regions_and_prioritizes_product_structure(self): + prompt = self._plan( + count=1, + ratio="3:4", + prompt="全身站姿，展示目标裤装", + ).prompts[0] + self.assertIn("区域编辑边界", prompt) + self.assertIn("只替换或修改下半身的目标下装区域", prompt) + self.assertIn("非目标服饰", prompt) + self.assertIn("无法辨认时不得猜写相似文字、乱码或新增文案", prompt) + self.assertIn("商品关键结构优先", prompt) + self.assertIn("独特轮廓、部件数量与位置关系", prompt) + self.assertIn("减少姿态和镜头变化", prompt) + self.assertIn("冲突处理优先级", prompt) + + def test_v22_full_body_shot_reserves_visible_canvas_margin(self): + prompt = self._plan( + title="宽松针织上衣", + count=1, + ratio="9:16", + prompt="真人写实，全身站姿，从头到脚完整展示", + ).prompts[0] + self.assertIn("构图安全边界", prompt) + self.assertIn("约画布短边的 3%-6%", prompt) + self.assertIn("不得裁切、触边或贴边", prompt) + + def test_explicit_full_body_requirement_applies_to_every_output(self): + plan = self._plan( + title="宽松针织上衣", + count=4, + prompt="真人写实，全身站姿，从头到脚完整展示", + ) + self.assertEqual(plan.kind, ProductKind.UPPER) + self.assertTrue(all("全身站姿" in shot for shot in plan.shots)) + self.assertTrue(all("从头到脚完整入镜" in shot for shot in plan.shots)) + + def test_wrist_product_is_worn_not_held(self): + plan = self._plan(title="智能运动手表", category="数码配件", count=4) + self.assertEqual(plan.kind, ProductKind.WRIST) + for prompt in plan.prompts: + self.assertIn("正确佩戴在模特手腕上", prompt) + self.assertIn("禁止手持商品", prompt) + + def test_headphones_are_worn_on_head_and_ears(self): + plan = self._plan(title="头戴式主动降噪耳机", category="数码", count=2) + self.assertEqual(plan.kind, ProductKind.HEAD_AUDIO) + for prompt in plan.prompts: + self.assertIn("正确佩戴在头部和双耳位置", prompt) + self.assertIn("禁止手持耳机", prompt) + self.assertIn("禁止", prompt) + + def test_handheld_product_uses_real_handheld_relation(self): + plan = self._plan(title="便携无线小风扇", category="家电", count=1) + self.assertEqual(plan.kind, ProductKind.HANDHELD) + self.assertIn("自然手持或操作", plan.prompts[0]) + self.assertIn("禁止把商品穿到身上", plan.prompts[0]) + + def test_unknown_uses_one_neutral_relation_not_a_forced_guess(self): + plan = self._plan(title="X7 多功能新品", category="其他", count=1) + self.assertEqual(plan.kind, ProductKind.UNKNOWN) + self.assertEqual(plan.classification.fallback_reason, "no_specific_rule_matched") + self.assertIn("选择且只选择一种正确的人物关系", plan.prompts[0]) + self.assertIn("不得同时混用", plan.prompts[0]) + + def test_dress_and_footwear_require_complete_visibility(self): + dress = self._plan(title="法式收腰连衣裙", category="女装", count=1) + shoes = self._plan(title="复古跑步鞋", category="鞋靴", count=1) + self.assertIn("从肩部到服装下摆完整展示", dress.prompts[0]) + self.assertIn("双脚和两只鞋必须完整清晰", shoes.prompts[0]) + + def test_selling_points_are_non_authoritative_hints(self): + context = ProductContext.create( + title="女装裤子", + category="服饰内衣", + selling_points=("冰感凉快", "显高显瘦"), + ) + plan = build_tryon_prompt_plan( + context=context, + user_prompt="电商模特上身图", + count=1, + ratio="3:4", + product_reference_count=1, + ) + prompt = plan.prompts[0] + self.assertIn("冰感凉快、显高显瘦", prompt) + self.assertIn("不是商品外观证据", prompt) + self.assertIn("不得据此臆造", prompt) + + def test_planner_has_no_generation_or_billing_side_effect_contract(self): + plan = self._plan(count=1) + self.assertEqual(len(plan.prompts), 1) + self.assertEqual(plan.references.model_portrait_number, 2) + + +class TryonClassificationPersistenceTests(TestCase): + def setUp(self): + from apps.accounts.models import Team, User + from apps.billing.models import CreditAccount + from apps.products.models import Product + + self.user = User.objects.create_user(username="tryon-classification", password="pass") + self.team = Team.objects.create(name="Tryon classification", owner=self.user) + CreditAccount.objects.create(team=self.team, balance="100.0000") + self.product = Product.objects.create( + team=self.team, + created_by=self.user, + title="春季神秘新品", + category="服饰内衣", + ) + + def test_model_batch_persists_same_namespaced_classification(self): + from apps.ai.services import enqueue_standalone_images + + with patch("apps.ai.tasks.generate_standalone_image_task.delay") as delay: + tasks = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身展示，自然光，电商主图", + mode="model", + count=2, + product_id=str(self.product.id), + ratio="3:4", + ) + + self.assertEqual(len(tasks), 2) + expected = { + "kind": "wearable_generic", + "source": "category", + "matched_rule": "broad_wearable:服饰内衣", + "fallback_reason": "broad_category_only", + "rule_version": "v2.1", + } + self.assertTrue(all(task.request_payload["tryon_classification"] == expected for task in tasks)) + self.assertTrue(all(task.request_payload["tryon_batch_count"] == 2 for task in tasks)) + self.assertEqual(len({task.request_payload["batch_id"] for task in tasks}), 1) + self.assertTrue(all(task.request_payload["prompt"] == "模特上身展示，自然光，电商主图" for task in tasks)) + self.assertEqual(delay.call_count, 2) + + def test_non_tryon_task_does_not_gain_classification_field(self): + from apps.ai.services import enqueue_standalone_images + + with patch("apps.ai.tasks.generate_standalone_image_task.delay") as delay: + task = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="自由创作一张图片", + mode="image", + count=1, + )[0] + self.assertNotIn("tryon_classification", task.request_payload) + + def test_internal_ab_override_and_experiment_metadata_are_namespaced(self): + from apps.ai.services import enqueue_standalone_images + + experiment_id = "52afde93-cd54-4b91-b1ea-1bc9d36b929e" + with patch("apps.ai.tasks.generate_standalone_image_task.delay") as delay: + task = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="全身站姿", + mode="model", + count=1, + product_id=str(self.product.id), + ratio="3:4", + tryon_prompt_v2_override=True, + tryon_ab={"experiment_id": experiment_id, "variant": "gpt-image"}, + dispatch=False, + )[0] + delay.assert_not_called() + self.assertIs(task.request_payload["tryon_prompt_v2_override"], True) + self.assertEqual( + task.request_payload["tryon_ab"], + {"experiment_id": experiment_id, "variant": "gpt-image"}, + ) diff --git a/core/backend/apps/ai/tests.py b/core/backend/apps/ai/tests.py index 1ed3f09..7383078 100644 --- a/core/backend/apps/ai/tests.py +++ b/core/backend/apps/ai/tests.py @@ -173,7 +173,7 @@ class NormalizeDraftTests(SimpleTestCase): from io import BytesIO from unittest.mock import patch -from django.test import TestCase +from django.test import TestCase, override_settings from rest_framework.test import APIClient @@ -392,6 +392,212 @@ class StandaloneImageReferenceTests(TestCase): self.assertEqual(prov.image_edit.call_args.kwargs["images"], ["http://x/cover.png", "http://x/model.png"]) prov.image_generation.assert_not_called() + @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) + def test_model_tryon_v2_uses_structured_prompt_and_persists_trace(self): + prov = self._patch_provider() + self.product.title = "中蓝色高腰宽腿牛仔裤" + self.product.category = "服饰内衣" + self.product.save(update_fields=["title", "category"]) + self.product.selling_points.create(title="显高显瘦", sort_order=0) + model_asset = Asset.objects.create( + team=self.team, created_by=self.user, name="模特", asset_type=Asset.Type.IMAGE, + source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, + ) + AssetFile.objects.create( + asset=model_asset, object_key="m-v2.png", bucket="b", content_type="image/png", + preview_url="http://x/model-v2.png", is_primary=True, + ) + user_prompt = "参考图1的女生穿上图2的牛仔裤，全身站姿，双腿完整露出，浅灰纯色背景" + + submitted = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt=user_prompt, + mode="model", + count=2, + product_id=str(self.product.id), + model_id=str(model_asset.id), + ratio="7:10", + ) + + tasks = list(AITask.objects.filter(id__in=[task.id for task in submitted]).order_by("request_payload__index")) + self.assertEqual(len(prov.image_edit.call_args_list), 2) + used_prompts = [call.kwargs["prompt"] for call in prov.image_edit.call_args_list] + used_sizes = [call.kwargs["size"] for call in prov.image_edit.call_args_list] + self.assertTrue(all(user_prompt in prompt for prompt in used_prompts)) + self.assertTrue(all("参考图角色（最高优先级）" in prompt for prompt in used_prompts)) + self.assertTrue(all("严格使用 7:10 比例" in prompt for prompt in used_prompts)) + self.assertTrue(all("半身近景" not in prompt and "手持或局部" not in prompt for prompt in used_prompts)) + self.assertNotEqual(used_prompts[0], used_prompts[1]) + self.assertEqual(used_sizes, ["1024x1536", "1024x1536"]) + + for index, task in enumerate(tasks): + trace = task.request_payload["tryon_prompt"] + self.assertTrue(trace["applied"]) + self.assertEqual(trace["version"], "v2.2") + self.assertEqual(trace["rollout_source"], "global") + self.assertEqual(trace["effective_prompt"], used_prompts[index]) + self.assertEqual(trace["shot_index"], index) + self.assertEqual(trace["batch_count"], 2) + self.assertEqual(trace["requested_ratio"], "7:10") + self.assertEqual(trace["resolved_ratio"], "7:10") + self.assertEqual(trace["reference_roles"]["product_numbers"], [1]) + self.assertEqual(trace["reference_roles"]["model_portrait_number"], 2) + self.assertIsNone(trace["reference_roles"]["model_triview_number"]) + self.assertEqual(task.request_payload["tryon_classification"]["kind"], "lower") + + @override_settings( + MODEL_TRYON_PROMPT_V2_ENABLED=False, + MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS=frozenset(), + ) + def test_model_tryon_flag_off_keeps_legacy_prompt(self): + prov = self._patch_provider() + self.product.title = "女装裤子" + self.product.save(update_fields=["title"]) + model_asset = Asset.objects.create( + team=self.team, created_by=self.user, name="模特", asset_type=Asset.Type.IMAGE, + source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, + ) + AssetFile.objects.create( + asset=model_asset, object_key="m-legacy.png", bucket="b", content_type="image/png", + preview_url="http://x/model-legacy.png", is_primary=True, + ) + submitted = enqueue_standalone_images( + team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, + product_id=str(self.product.id), model_id=str(model_asset.id), ratio="3:4", + ) + used_prompt = prov.image_edit.call_args.kwargs["prompt"] + self.assertIn("半身近景", used_prompt) + task = AITask.objects.get(id=submitted[0].id) + self.assertNotIn("tryon_prompt", task.request_payload) + + @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=False) + def test_model_tryon_canary_team_uses_v22_and_records_source(self): + prov = self._patch_provider() + self.product.title = "女装裤子" + self.product.save(update_fields=["title"]) + with self.settings(MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS={str(self.team.id).upper()}): + submitted = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="全身站姿，双腿完整露出", + mode="model", + count=1, + product_id=str(self.product.id), + ratio="3:4", + ) + + used_prompt = prov.image_edit.call_args.kwargs["prompt"] + self.assertIn("商品关键结构优先", used_prompt) + task = AITask.objects.get(id=submitted[0].id) + trace = task.request_payload["tryon_prompt"] + self.assertTrue(trace["applied"]) + self.assertEqual(trace["version"], "v2.2") + self.assertEqual(trace["rollout_source"], "canary") + + @override_settings( + MODEL_TRYON_PROMPT_V2_ENABLED=False, + MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS={"not-a-team-uuid"}, + ) + def test_model_tryon_non_canary_team_fails_closed_to_legacy_prompt(self): + prov = self._patch_provider() + self.product.title = "女装裤子" + self.product.save(update_fields=["title"]) + submitted = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身图", + mode="model", + count=1, + product_id=str(self.product.id), + ratio="3:4", + ) + + used_prompt = prov.image_edit.call_args.kwargs["prompt"] + self.assertIn("半身近景", used_prompt) + task = AITask.objects.get(id=submitted[0].id) + self.assertNotIn("tryon_prompt", task.request_payload) + + @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=False) + def test_model_tryon_canary_without_product_reference_records_safe_fallback(self): + prov = self._patch_provider() + self.product.cover_asset = None + self.product.save(update_fields=["cover_asset"]) + with self.settings(MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS={str(self.team.id)}): + submitted = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身图", + mode="model", + count=1, + product_id=str(self.product.id), + ratio="3:4", + ) + + prov.image_generation.assert_called_once() + task = AITask.objects.get(id=submitted[0].id) + trace = task.request_payload["tryon_prompt"] + self.assertFalse(trace["applied"]) + self.assertEqual(trace["rollout_source"], "canary") + self.assertEqual(trace["fallback_reason"], "no_product_reference") + + @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) + def test_model_tryon_v22_uses_category_default_ratio_when_user_omits_ratio(self): + prov = self._patch_provider() + self.product.title = "女装裤子" + self.product.save(update_fields=["title"]) + submitted = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身图", + mode="model", + count=1, + product_id=str(self.product.id), + ) + + self.assertEqual(prov.image_edit.call_args.kwargs["size"], "1024x1536") + task = AITask.objects.get(id=submitted[0].id) + trace = task.request_payload["tryon_prompt"] + self.assertEqual(trace["requested_ratio"], None) + self.assertEqual(trace["resolved_ratio"], "3:4") + self.assertEqual(trace["ratio_source"], "default") + self.assertEqual(trace["rollout_source"], "global") + + @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) + def test_model_tryon_old_queued_task_without_v2_metadata_falls_back_safely(self): + from apps.ai.services import run_standalone_image_task + + prov = self._patch_provider() + self.product.title = "女装裤子" + self.product.save(update_fields=["title"]) + with patch("apps.ai.tasks.generate_standalone_image_task.delay"): + task = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身图", + mode="model", + count=1, + product_id=str(self.product.id), + ratio="3:4", + )[0] + + legacy_payload = dict(task.request_payload) + legacy_payload.pop("tryon_batch_count", None) + legacy_payload.pop("tryon_classification", None) + task.request_payload = legacy_payload + task.save(update_fields=["request_payload", "updated_at"]) + + run_standalone_image_task(task_id=str(task.id)) + + task.refresh_from_db() + used_prompt = prov.image_edit.call_args.kwargs["prompt"] + self.assertIn("半身近景", used_prompt) + self.assertEqual(task.status, AITask.Status.SUCCEEDED) + trace = task.request_payload["tryon_prompt"] + self.assertFalse(trace["applied"]) + self.assertEqual(trace["rollout_source"], "global") + self.assertEqual(trace["fallback_reason"], "unsupported_tryon_batch_count") + def test_cover_mode_falls_back_to_t2i_without_main_image(self): prov = self._patch_provider() self.product.cover_asset = None @@ -443,6 +649,66 @@ class StandaloneImageReferenceTests(TestCase): # 任务最终落在支持 image_edit 的模型(gpt-image),而不是用户选的火山 Seedream self.assertIn("gpt-image", tasks[0].model_config.name) + def test_model_tryon_keeps_each_user_selected_model_even_with_extra_refs(self): + """模特上身图由专用 Worker 向两类模型传商品/人物参考图，不能沿用自由创作的自动换模逻辑。""" + vp, _ = ModelProvider.objects.get_or_create( + name="volcengine", + defaults={"display_name": "火山"}, + ) + seedream = ModelConfig.objects.create( + provider=vp, + name="seedream-v22-user-choice", + display_name="Seedream V2.2 user choice", + capability=ModelConfig.Capability.IMAGE, + ) + gpt = ModelConfig.objects.filter( + capability=ModelConfig.Capability.IMAGE, + name__icontains="gpt-image", + ).first() + self.assertIsNotNone(gpt) + extra_ref = Asset.objects.create( + team=self.team, + created_by=self.user, + name="额外参考", + asset_type=Asset.Type.IMAGE, + source=Asset.Source.UPLOAD, + category=Asset.Category.UPLOAD, + ) + AssetFile.objects.create( + asset=extra_ref, + object_key="extra.png", + bucket="b", + content_type="image/png", + preview_url="http://x/extra.png", + is_primary=True, + ) + + seedream_task = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身图", + mode="model", + count=1, + product_id=str(self.product.id), + image_model=f"{vp.name}:{seedream.name}", + reference_image_ids=[str(extra_ref.id)], + dispatch=False, + )[0] + gpt_task = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="模特上身图", + mode="model", + count=1, + product_id=str(self.product.id), + image_model=f"{gpt.provider.name}:{gpt.name}", + reference_image_ids=[str(extra_ref.id)], + dispatch=False, + )[0] + + self.assertEqual(seedream_task.model_config_id, seedream.id) + self.assertEqual(gpt_task.model_config_id, gpt.id) + class ImageConversationTests(TestCase): """图片创作「对话」实体:CRUD + 团队隔离 + 软删 + 生图自动归属对话 + 任务回填。""" @@ -766,6 +1032,37 @@ class StandaloneCategoryTests(TestCase): self.assertEqual(a.category, Asset.Category.MODEL_PORTRAIT) self.assertFalse(a.in_library) + @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) + def test_model_tryon_v2_provider_failure_releases_reserved_credit(self): + from apps.ai.services import run_standalone_image_task + + self.product.title = "女装裤子" + self.product.save(update_fields=["title"]) + prov = self._patch_provider() + prov.image_edit.side_effect = ValueError("生成失败") + with patch("apps.ai.tasks.generate_standalone_image_task.delay"): + task = enqueue_standalone_images( + team=self.team, + user=self.user, + prompt="全身站姿，双腿完整露出", + mode="model", + count=1, + product_id=str(self.product.id), + ratio="3:4", + )[0] + + run_standalone_image_task(task_id=str(task.id)) + task.refresh_from_db() + account = CreditAccount.objects.get(team=self.team) + self.assertEqual(task.status, AITask.Status.FAILED) + self.assertEqual(float(task.actual_cost), 0.0) + self.assertEqual(float(account.balance), 100.0) + self.assertEqual(float(account.reserved_balance), 0.0) + self.assertTrue(task.request_payload["tryon_prompt"]["applied"]) + self.assertTrue( + CreditLedger.objects.filter(task=task, ledger_type=CreditLedger.Type.RELEASE).exists() + ) + class TriviewModelDecouplingTests(TestCase): """项目角色三视图只归项目，不自动创建或更新团队模特。""" diff --git a/core/backend/apps/ai/tryon_prompt.py b/core/backend/apps/ai/tryon_prompt.py new file mode 100644 index 0000000..c19ee4c --- /dev/null +++ b/core/backend/apps/ai/tryon_prompt.py @@ -0,0 +1,659 @@ +"""模特上身图结构化提示词规划器（Step 3 受控接入）。 + +本模块只做确定性的文本规划： +- 不读取图片、不调用任何模型； +- 不创建任务、不预占或扣除积分； +- 明确参考图角色，并按商品的真实展示方式规划 1 / 2 / 4 张图片； +- 生产 Worker 仅在 MODEL_TRYON_PROMPT_V2_ENABLED 开启时使用本规划器。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Iterable + + +class ProductKind(str, Enum): + LOWER = "lower" + UPPER = "upper" + FULL_BODY = "full_body" + UNDERWEAR = "underwear" + FOOTWEAR = "footwear" + HEADWEAR = "headwear" + EYEWEAR = "eyewear" + WRIST = "wrist" + HEAD_AUDIO = "head_audio" + EAR_AUDIO = "ear_audio" + JEWELRY = "jewelry" + BAG = "bag" + WEARABLE_GENERIC = "wearable_generic" + HANDHELD = "handheld" + NONWEARABLE = "nonwearable" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class ProductContext: + title: str + category: str = "" + selling_points: tuple[str, ...] = () + description: str = "" + + @classmethod + def create( + cls, + *, + title: str, + category: str = "", + selling_points: Iterable[str] | None = None, + description: str = "", + ) -> "ProductContext": + return cls( + title=(title or "商品").strip(), + category=(category or "").strip(), + selling_points=tuple( + point.strip() for point in (selling_points or ()) if point and point.strip() + ), + description=(description or "").strip(), + ) + + +@dataclass(frozen=True) +class ReferenceRoles: + """API 实际上传顺序对应的参考图角色。""" + + product_numbers: tuple[int, ...] + model_portrait_number: int | None + model_triview_number: int | None + + @classmethod + def create( + cls, + *, + product_count: int, + has_model_portrait: bool = True, + has_model_triview: bool = False, + ) -> "ReferenceRoles": + if product_count < 1: + raise ValueError("product_count 必须至少为 1") + + product_numbers = tuple(range(1, product_count + 1)) + next_number = product_count + 1 + portrait_number = next_number if has_model_portrait else None + if has_model_portrait: + next_number += 1 + triview_number = next_number if has_model_triview else None + return cls(product_numbers, portrait_number, triview_number) + + @property + def product_label(self) -> str: + if len(self.product_numbers) == 1: + return f"参考图{self.product_numbers[0]}" + start, end = self.product_numbers[0], self.product_numbers[-1] + return f"参考图{start}-{end}" + + @property + def model_label(self) -> str: + labels: list[str] = [] + if self.model_portrait_number is not None: + labels.append(f"参考图{self.model_portrait_number}（模特肖像）") + if self.model_triview_number is not None: + labels.append(f"参考图{self.model_triview_number}（模特三视图）") + return "、".join(labels) if labels else "无指定模特参考图" + + def instruction(self) -> str: + return ( + f"参考图角色（最高优先级）：{self.product_label}是同一商品的真实商品图，" + f"{self.model_label}。商品图只决定商品外观，人物图只决定人物身份与体型。" + "用户原文若使用了与实际上传顺序不同的‘图1/图2’编号，不按数字机械执行：" + "人物、女生、模特相关要求统一对应人物参考图，商品、衣物、配件相关要求统一对应商品参考图。" + ) + + +@dataclass(frozen=True) +class DisplayPolicy: + placement: str + visibility: str + shots: tuple[str, str, str, str] + negative: str + + +@dataclass(frozen=True) +class ClassificationResult: + kind: ProductKind + source: str + matched_rule: str | None = None + fallback_reason: str | None = None + rule_version: str = "v2.1" + + def as_payload(self) -> dict[str, str | None]: + """返回只包含 JSON 原生值的任务快照。""" + + return { + "kind": self.kind.value, + "source": self.source, + "matched_rule": self.matched_rule, + "fallback_reason": self.fallback_reason, + "rule_version": self.rule_version, + } + + @classmethod + def from_payload(cls, payload: object) -> "ClassificationResult | None": + if not isinstance(payload, dict): + return None + try: + kind = ProductKind(str(payload.get("kind") or "")) + except ValueError: + return None + source = str(payload.get("source") or "").strip() + if not source: + return None + return cls( + kind=kind, + source=source, + matched_rule=(str(payload["matched_rule"]) if payload.get("matched_rule") is not None else None), + fallback_reason=( + str(payload["fallback_reason"]) if payload.get("fallback_reason") is not None else None + ), + rule_version=str(payload.get("rule_version") or "v2.1"), + ) + + +@dataclass(frozen=True) +class TryonPromptPlan: + kind: ProductKind + classification: ClassificationResult + ratio: str + references: ReferenceRoles + shots: tuple[str, ...] + prompts: tuple[str, ...] + + +_FULL_BODY_TERMS = ("全身", "从头到脚", "头到脚", "双腿完整", "脚部完整", "完整站姿") + + +_POLICIES: dict[ProductKind, DisplayPolicy] = { + ProductKind.LOWER: DisplayPolicy( + placement="商品必须真实、自然地穿在模特下半身，位置和人体结构正确，不得手持、悬挂或摆在身旁", + visibility="完整展示腰线、门襟、口袋、臀腿版型、裤腿或裙摆、长度和垂坠感；双腿及下装末端不得出画", + shots=( + "正面全身站姿，从头到脚完整入镜，双腿自然分开，清楚展示下装正面版型", + "三分之四角度全身站姿，从头到脚完整入镜，展示腰臀过渡、腿部轮廓与垂坠感", + "侧面全身站姿，从头到脚完整入镜，展示腰线、侧缝、裤长或裙长", + "背面全身站姿，从头到脚完整入镜，展示后腰、后袋、臀腿和下装末端", + ), + negative="禁止裁掉双腿、脚部、腰头或下装末端；禁止把下装改成手持商品", + ), + ProductKind.UPPER: DisplayPolicy( + placement="商品必须真实、自然地穿在模特上半身，正确替换对应上衣，不覆盖无关服装区域", + visibility="清楚展示领口、肩线、袖型、门襟、下摆、图案和面料垂感", + shots=( + "正面站姿，腰部以上完整入镜，双臂自然放松，展示上衣正面结构", + "三分之四角度站姿，腰部以上完整入镜，展示肩线、袖型和衣身轮廓", + "侧面站姿，腰部以上完整入镜，展示侧缝、衣长和面料垂感", + "背面站姿，腰部以上完整入镜，展示后领、后背和下摆", + ), + negative="禁止把上衣拿在手中或叠放展示；禁止遮挡领口、袖口和下摆", + ), + ProductKind.FULL_BODY: DisplayPolicy( + placement="商品必须作为一件完整服装真实穿在模特身上，肩部、腰部与下摆位置符合人体结构", + visibility="从肩部到服装下摆完整展示，连体结构、腰线、长度和整体廓形不得被裁切", + shots=( + "正面全身站姿，从头到脚完整入镜，展示服装完整正面廓形", + "三分之四角度全身站姿，从头到脚完整入镜，展示腰线和整体垂坠感", + "侧面全身站姿，从头到脚完整入镜，展示侧面轮廓与服装长度", + "背面全身站姿，从头到脚完整入镜，展示完整后背结构与下摆", + ), + negative="禁止截断服装主体或下摆；禁止拆成上下两件或擅自改变连体结构", + ), + ProductKind.UNDERWEAR: DisplayPolicy( + placement="商品必须作为贴身服饰正确穿着，肩带、罩杯、腰头等结构与身体位置自然贴合", + visibility="在合规、克制的电商展示中清楚呈现商品轮廓、边缘、面料和支撑结构", + shots=( + "正面自然站姿，商品完整入镜，姿态克制，展示正面结构和贴合度", + "三分之四角度自然站姿，商品完整入镜，展示侧翼、肩带或腰线", + "侧面自然站姿，商品完整入镜，展示厚度、包覆和贴合轮廓", + "背面自然站姿，商品完整入镜，展示背带、搭扣或后腰结构", + ), + negative="禁止色情化姿势、夸张身体曲线或透明化处理；禁止遮挡商品关键结构", + ), + ProductKind.FOOTWEAR: DisplayPolicy( + placement="商品必须正确穿在模特双脚上，左右脚成对一致，鞋底与地面接触自然", + visibility="双脚和两只鞋必须完整清晰，展示鞋面、鞋头、鞋跟、鞋底厚度和上脚比例", + shots=( + "正面全身站姿，从头到脚完整入镜，两只鞋完整可见", + "三分之四角度全身站姿，从头到脚完整入镜，展示鞋面与鞋侧", + "侧面全身站姿，从头到脚完整入镜，展示鞋底、鞋跟和侧面轮廓", + "低机位双脚展示，两只鞋完整入镜，展示上脚细节且保持真实人体比例", + ), + negative="禁止裁掉脚部、只生成一只鞋、左右鞋款不一致或把鞋拿在手中", + ), + ProductKind.HEADWEAR: DisplayPolicy( + placement="商品必须正确佩戴在模特头部，尺寸、方向和受力关系自然", + visibility="完整展示帽檐、帽冠、侧面结构及与发型的自然接触关系", + shots=( + "正面胸部以上人像，头饰完整入镜，展示正面形态", + "三分之四角度胸部以上人像，头饰完整入镜，展示帽檐和帽冠", + "侧面胸部以上人像，头饰完整入镜，展示侧面结构", + "背面胸部以上人像，头饰完整入镜，展示后部调节或收口结构", + ), + negative="禁止把头饰拿在手中、悬浮在头顶或遮挡整个面部", + ), + ProductKind.EYEWEAR: DisplayPolicy( + placement="商品必须正确佩戴在模特眼部和鼻梁位置，镜腿自然贴合双耳", + visibility="完整展示镜框、镜片、鼻托、镜腿及上脸比例，左右结构保持对称", + shots=( + "正面肩部以上人像，眼镜完整清晰，展示正面框型", + "三分之四角度肩部以上人像，展示镜框厚度、鼻托和镜腿", + "侧面肩部以上人像，展示镜腿与耳部的正确贴合", + "正面肩部以上近景，眼镜完整入镜，突出材质细节且不过度裁切", + ), + negative="禁止把眼镜拿在手中、戴在头顶、生成重影镜框或不对称镜腿", + ), + ProductKind.WRIST: DisplayPolicy( + placement="商品必须正确佩戴在模特手腕上，表盘朝向、表带闭合和尺寸比例真实", + visibility="清楚展示表盘、表带、扣具及佩戴关系，同时保留自然手臂和人物身份", + shots=( + "正面自然站姿，佩戴手表的手腕抬至胸前，商品完整清晰", + "三分之四角度人像，佩戴手表的手腕自然转向镜头，展示表盘和表带", + "侧面人像，手腕自然弯曲，展示表壳厚度和扣具位置", + "手腕局部特写，商品仍正确佩戴，表盘、表带和皮肤接触关系完整", + ), + negative="禁止手持商品、把手表放在掌心、生成多个表盘或错误佩戴位置", + ), + ProductKind.HEAD_AUDIO: DisplayPolicy( + placement="商品必须按真实使用方式正确佩戴在头部和双耳位置，头梁、耳罩方向及尺寸比例自然", + visibility="完整展示头梁、左右耳罩、连接结构及与发型和耳部的佩戴关系", + shots=( + "正面胸部以上人像，耳机正确佩戴，左右耳罩和头梁完整可见", + "三分之四角度胸部以上人像，展示耳罩、头梁和佩戴贴合度", + "侧面胸部以上人像，展示单侧耳罩厚度与头梁弧度，另一侧结构保持合理", + "正面胸部以上近景，耳机正确佩戴并完整入镜，突出材质和结构细节", + ), + negative="禁止手持耳机、挂在颈部、把头戴式耳机改成入耳式或遗漏一侧耳罩", + ), + ProductKind.EAR_AUDIO: DisplayPolicy( + placement="商品必须按真实使用方式正确佩戴在耳内或耳部，左右耳对应、方向和尺寸比例自然", + visibility="清楚展示耳塞主体、柄部或耳挂结构，以及商品与耳部的真实佩戴关系", + shots=( + "正面肩部以上人像，入耳式耳机正确佩戴，左右结构合理可见", + "三分之四角度肩部以上人像，靠近镜头一侧耳机完整清晰", + "侧面肩部以上人像，展示耳机在耳内或耳部的正确方向和贴合关系", + "耳部局部特写，商品保持正确佩戴，展示结构、材质和真实尺寸", + ), + negative="禁止手持耳机、把入耳式耳机改成头戴式、生成头梁或夸大耳机尺寸", + ), + ProductKind.JEWELRY: DisplayPolicy( + placement="商品必须佩戴在其对应身体位置，方向、数量、尺寸和受力关系自然", + visibility="商品完整清晰，展示造型、材质、连接件以及与人物的真实比例", + shots=( + "正面自然人像，饰品正确佩戴并完整清晰", + "三分之四角度人像，展示饰品造型、厚度和佩戴关系", + "侧面自然人像，展示饰品侧面结构与连接位置", + "佩戴位置局部特写，饰品保持完整且不过度美化或放大", + ), + negative="禁止手持饰品、改变佩戴位置、凭空增加数量或夸大商品尺寸", + ), + ProductKind.BAG: DisplayPolicy( + placement="根据商品真实包型选择唯一正确的携带方式：手提、单肩、斜挎、双肩或腰间固定；不得同时混用多种方式", + visibility="包身、肩带或提手、开合结构和五金必须完整清晰，尺寸及与人体的比例真实", + shots=( + "正面自然站姿，商品按真实包型正确携带，包身完整朝向镜头", + "三分之四角度自然站姿，展示包身厚度、肩带或提手与人物的关系", + "侧面自然站姿，展示包型轮廓、容量厚度和正确携带位置", + "商品局部展示，包身与主要肩带或提手保持完整，突出材质、五金和开合结构", + ), + negative="禁止改变包型、增加多余肩带或提手、错误背法、悬浮或复制多个包", + ), + ProductKind.WEARABLE_GENERIC: DisplayPolicy( + placement="商品必须按照真实用途正确穿戴在模特身上，不得手持、悬浮或摆在身旁", + visibility="商品主体、边缘、连接结构和实际穿戴关系必须完整清晰", + shots=( + "正面自然站姿，商品正确穿戴并完整入镜", + "三分之四角度自然站姿，展示商品轮廓与穿戴关系", + "侧面自然站姿，展示商品厚度、长度和贴合方式", + "背面自然站姿，展示商品后部结构与固定方式", + ), + negative="禁止把穿戴商品改成手持展示；禁止错误穿戴位置或改变用途", + ), + ProductKind.HANDHELD: DisplayPolicy( + placement="商品应由模特按真实用途自然手持或操作，握持位置不得遮挡核心结构", + visibility="商品主体完整清晰，展示外观、材质、控制区域及与手部的真实比例", + shots=( + "正面自然人像，模特单手自然拿稳商品，商品主体完整朝向镜头", + "三分之四角度人像，模特按真实用途操作商品，展示侧面结构", + "侧面自然人像，展示握持方式、商品厚度和使用关系", + "手部与商品局部特写，商品主体保持完整，清楚展示材质和操作细节", + ), + negative="禁止把商品穿到身上、遮挡商品主体、生成多余手指或改变商品用途", + ), + ProductKind.NONWEARABLE: DisplayPolicy( + placement="根据商品真实用途安排模特与商品的关系；不得把非穿戴商品强行穿在身上", + visibility="商品主体必须完整清晰，尺寸、结构和与人物的比例符合真实使用逻辑", + shots=( + "正面电商人像，商品完整清晰地处于真实使用位置", + "三分之四角度电商人像，展示商品侧面结构和真实使用关系", + "侧面电商人像，展示商品厚度、尺寸和操作方式", + "商品与操作部位局部特写，商品主体保持完整并突出关键结构", + ), + negative="禁止把商品强行穿戴、悬浮、复制成多个或改变真实用途", + ), + ProductKind.UNKNOWN: DisplayPolicy( + placement="先依据商品参考图和商品文字判断商品的真实用途，再选择且只选择一种正确的人物关系完成展示；不得同时混用穿着、佩戴、手持等互斥方式", + visibility="商品主体和关键结构必须完整清晰，尺寸、方向及与人物的关系符合参考图所体现的真实用途", + shots=( + "正面自然电商人像，商品按判断出的唯一真实用途展示并完整入镜", + "三分之四角度自然电商人像，展示商品结构和唯一正确的人物关系", + "侧面自然电商人像，展示商品厚度、尺寸和真实使用位置", + "商品及其真实使用位置局部展示，商品主体保持完整并突出关键结构", + ), + negative="禁止同时执行互斥的穿着、佩戴或手持关系；禁止在无法确认时擅自替换模特服装或改变商品用途", + ), +} + + +_KEYWORDS: tuple[tuple[ProductKind, tuple[str, ...]], ...] = ( + (ProductKind.FULL_BODY, ("连衣裙", "连体裤", "连体衣", "连身裙", "套装")), + (ProductKind.LOWER, ("牛仔裤", "裤", "半身裙", "短裙", "长裙", "下装")), + (ProductKind.UNDERWEAR, ("内衣", "文胸", "胸罩", "bra", "塑身衣", "内裤")), + (ProductKind.FOOTWEAR, ("鞋", "靴", "凉拖", "拖鞋")), + (ProductKind.HEAD_AUDIO, ("头戴式", "头戴耳机", "耳麦")), + (ProductKind.EAR_AUDIO, ("入耳式", "耳塞", "耳挂式", "真无线耳机", "蓝牙耳机", "耳机", "airpods")), + (ProductKind.WRIST, ("手表", "腕表", "手环")), + (ProductKind.EYEWEAR, ("眼镜", "墨镜", "镜框")), + (ProductKind.HEADWEAR, ("帽", "头巾", "发箍")), + (ProductKind.JEWELRY, ("项链", "耳环", "耳钉", "戒指", "手链", "脚链", "胸针")), + (ProductKind.BAG, ("双肩包", "背包", "斜挎包", "单肩包", "腰包", "手提包", "手拿包", "包袋")), + (ProductKind.UPPER, ("上衣", "t恤", "衬衫", "卫衣", "外套", "夹克", "毛衣", "背心", "吊带")), + (ProductKind.HANDHELD, ("风扇", "水壶", "水杯", "手机", "相机", "雨伞")), + (ProductKind.NONWEARABLE, ("收纳盒", "置物架", "台灯", "花瓶", "摆件", "桌椅", "餐具")), +) + +_GENERIC_WEARABLE_HINTS = ( + "围巾", "披肩", "领带", "领结", "袜", "腰带", "皮带", "手套", "护膝", "护腕", "服饰", "服装", "穿戴", "衣帽", "鞋服", +) +_BROAD_WEARABLE_CATEGORIES = ("服饰内衣", "服饰", "服装", "女装", "男装", "配饰") +_USER_RELATION_RULES: tuple[tuple[ProductKind, tuple[str, ...]], ...] = ( + (ProductKind.HANDHELD, ("手持", "拿着", "拿在手", "握住")), + (ProductKind.WEARABLE_GENERIC, ("穿上", "穿着", "换上", "佩戴", "戴上", "戴着")), + (ProductKind.NONWEARABLE, ("正在使用", "实际使用", "操作商品")), +) + + +def _match_rule(blob: str) -> tuple[ProductKind, str] | None: + for kind, keywords in _KEYWORDS: + for keyword in keywords: + if keyword in blob: + return kind, keyword + return None + + +def _match_user_relation(user_prompt: str) -> tuple[ProductKind, str] | None: + lowered = (user_prompt or "").lower() + for kind, keywords in _USER_RELATION_RULES: + for keyword in keywords: + if keyword in lowered: + return kind, keyword + return None + + +def classify_product_details(context: ProductContext, user_prompt: str = "") -> ClassificationResult: + """返回可追溯的分类结果；无法可靠判断时显式落入 UNKNOWN。""" + + title_blob = " ".join((context.title, context.description)).lower() + matched = _match_rule(title_blob) + if matched: + kind, keyword = matched + return ClassificationResult(kind, "title_description", f"{kind.value}:{keyword}") + + # 用户明确关系只覆盖规则表没有识别出的商品，不推翻裤装、手表等已确定的真实商品类型。 + user_relation = _match_user_relation(user_prompt) + if user_relation: + kind, keyword = user_relation + return ClassificationResult( + kind, + "user_prompt", + f"relation:{keyword}", + "no_specific_product_rule", + ) + + for keyword in _GENERIC_WEARABLE_HINTS: + if keyword in title_blob: + return ClassificationResult( + ProductKind.WEARABLE_GENERIC, + "title_description", + f"wearable_generic:{keyword}", + "no_specific_product_rule", + ) + + category = context.category.lower().strip() + # “服饰内衣”是平台宽泛大类，不能仅凭其中的“内衣”二字判成贴身内衣。 + for broad_category in _BROAD_WEARABLE_CATEGORIES: + if category == broad_category or broad_category in category: + return ClassificationResult( + ProductKind.WEARABLE_GENERIC, + "category", + f"broad_wearable:{broad_category}", + "broad_category_only", + ) + + matched = _match_rule(category) + if matched: + kind, keyword = matched + return ClassificationResult(kind, "category", f"{kind.value}:{keyword}") + + return ClassificationResult( + ProductKind.UNKNOWN, + "fallback", + None, + "no_specific_rule_matched", + ) + + +def classify_product(context: ProductContext, user_prompt: str = "") -> ProductKind: + """按展示方式分类，而不是简单区分“服饰 / 非服饰”。""" + + return classify_product_details(context, user_prompt).kind + + +def _requires_full_body(user_prompt: str) -> bool: + lowered = (user_prompt or "").lower() + return any(term in lowered for term in _FULL_BODY_TERMS) + + +def _full_body_shots(kind: ProductKind) -> tuple[str, str, str, str]: + subject = { + ProductKind.FOOTWEAR: "商品与两只鞋", + ProductKind.LOWER: "下装、双腿与脚部", + ProductKind.FULL_BODY: "整件服装与下摆", + }.get(kind, "人物与商品") + return ( + f"正面全身站姿，从头到脚完整入镜，{subject}完整清晰", + f"三分之四角度全身站姿，从头到脚完整入镜，{subject}完整清晰", + f"侧面全身站姿，从头到脚完整入镜，{subject}完整清晰", + f"背面全身站姿，从头到脚完整入镜，{subject}完整清晰", + ) + + +def plan_shots(kind: ProductKind, count: int, user_prompt: str = "") -> tuple[str, ...]: + if count not in (1, 2, 4): + raise ValueError("模特上身图生成数量只支持 1、2、4") + shots = _full_body_shots(kind) if _requires_full_body(user_prompt) else _POLICIES[kind].shots + return shots[:count] + + +def default_ratio_for_kind(kind: ProductKind) -> str: + """只在用户没有传比例时使用；显式比例始终优先。""" + + if kind in { + ProductKind.LOWER, + ProductKind.UPPER, + ProductKind.FULL_BODY, + ProductKind.UNDERWEAR, + ProductKind.FOOTWEAR, + ProductKind.HEADWEAR, + ProductKind.EYEWEAR, + ProductKind.WRIST, + ProductKind.HEAD_AUDIO, + ProductKind.EAR_AUDIO, + ProductKind.JEWELRY, + ProductKind.BAG, + ProductKind.WEARABLE_GENERIC, + }: + return "3:4" + return "1:1" + + +def _selling_points_instruction(context: ProductContext) -> str: + if not context.selling_points: + return "" + points = "、".join(context.selling_points) + return ( + f"商品文字卖点（仅用于理解展示意图）：{points}。" + "文字卖点不是商品外观证据，不得据此臆造参考图中看不到的颜色、结构、材质或功能。" + ) + + +_TARGET_AREAS: dict[ProductKind, str] = { + ProductKind.LOWER: "下半身的目标下装区域", + ProductKind.UPPER: "上半身的目标上装区域", + ProductKind.FULL_BODY: "目标连体服装正确覆盖的区域", + ProductKind.UNDERWEAR: "目标贴身服饰正确覆盖的区域", + ProductKind.FOOTWEAR: "双脚的目标鞋履区域", + ProductKind.HEADWEAR: "头部的目标头饰区域", + ProductKind.EYEWEAR: "眼部和鼻梁的目标眼镜区域", + ProductKind.WRIST: "手腕的目标商品区域", + ProductKind.HEAD_AUDIO: "头部和双耳的目标耳机区域", + ProductKind.EAR_AUDIO: "耳部的目标耳机区域", + ProductKind.JEWELRY: "商品真实佩戴位置对应的目标区域", + ProductKind.BAG: "商品真实携带方式所需的目标区域", + ProductKind.WEARABLE_GENERIC: "商品真实穿戴位置对应的目标区域", + ProductKind.HANDHELD: "商品真实握持或操作所需的目标区域", + ProductKind.NONWEARABLE: "商品真实使用关系所需的目标区域", + ProductKind.UNKNOWN: "依据商品真实用途确定的唯一目标区域", +} + + +def _editing_boundary_instruction(kind: ProductKind, references: ReferenceRoles) -> str: + target_area = _TARGET_AREAS[kind] + if references.model_portrait_number is None: + return ( + f"区域编辑边界：只修改或生成{target_area}；除目标商品正常穿戴或使用所必需的区域外，" + "不得连带重做非目标服饰、发型、妆容或配饰，不重新设计整套穿搭。" + ) + return ( + f"区域编辑边界：只替换或修改{target_area}。参考图{references.model_portrait_number}中的非目标服饰、" + "发型、妆容和已有配饰保持不变，不重新设计整套穿搭。非目标服饰上的文字或图案清楚可辨时，" + "保持原内容、拼写、位置和颜色；无法辨认时不得猜写相似文字、乱码或新增文案。" + ) + + +def _safe_margin_instruction(shot: str) -> str: + if "全身" in shot or "从头到脚" in shot: + return ( + "构图安全边界：头顶、双脚、裤脚、裙摆或商品末端必须完整处于画布内，" + "与画布上下左右边缘保留明显安全留白（约画布短边的 3%-6%）；不得裁切、触边或贴边。" + ) + return ( + "构图安全边界：商品主体及所有必须展示的边缘完整处于画布内并保留明显安全留白，" + "不得裁切、触边或贴边。" + ) + + +def _build_prompt( + *, + context: ProductContext, + kind: ProductKind, + policy: DisplayPolicy, + references: ReferenceRoles, + user_prompt: str, + ratio: str, + shot: str, +) -> str: + hard_requirement = (user_prompt or "").strip() or "生成真实、可用的电商模特商品展示图。" + if references.model_portrait_number is not None: + identity = ( + f"人物身份锁定：严格保持参考图{references.model_portrait_number}中人物的脸型、五官、发型、" + "肤色、年龄感、身材比例和气质，不换人、不混合其他人物特征。" + ) + else: + identity = "人物要求：使用同一位真人写实模特，人体比例自然，批次内人物身份保持一致。" + + sections = [ + references.instruction(), + f"用户硬性要求（高于默认镜头与默认场景）：{hard_requirement}", + identity, + ( + f"商品保真：{references.product_label}是“{context.title}”外观的唯一事实依据。" + "严格保持商品颜色、版型、结构、材质纹理、长度、口袋、拉链、缝线、图案、文字与 Logo；" + "多张商品图需综合理解为同一件商品，不重新设计、不改款、不生成相似款。" + ), + _editing_boundary_instruction(kind, references), + ( + "商品关键结构优先：商品参考图中可见的独特轮廓、部件数量与位置关系、开合结构、口袋、" + "拉链、缝线和五金，优先于姿态变化、修身美化与风格化；同一批每张必须保持为同一款商品。" + "若商品结构与复杂姿态或镜头变化冲突，减少姿态和镜头变化，绝不牺牲商品关键结构。" + ), + _selling_points_instruction(context), + f"穿戴或使用关系：{policy.placement}。", + f"商品可见性：{policy.visibility}。", + f"本张镜头：{shot}。", + _safe_margin_instruction(shot), + f"输出画幅：严格使用 {ratio} 比例；比例只影响构图，不得因此裁掉必须完整展示的商品部位。", + ( + "场景与光线：若用户硬性要求指定了背景、场景或光线，严格沿用用户要求；" + "否则使用浅灰或米白纯色背景、柔和均匀棚拍光、干净克制的电商布光。" + ), + "画质：真人写实，高分辨率，肤色和面料质感自然，商品为视觉重点，人物与商品比例真实协调。", + ( + "冲突处理优先级：用户硬性展示要求最高（其中图号按参考图角色语义解析），随后依次为人物身份、" + "商品身份与关键结构、正确穿戴或使用关系、必须展示区域；姿态、镜头变化、背景装饰和美化不得覆盖上述硬约束。" + ), + ( + f"禁止项：{policy.negative}；禁止换人、改变商品设计、错误 Logo 或乱码文字、多余商品、" + "多余肢体、畸形手脚、身体扭曲、低清模糊、过度磨皮、夸张滤镜、水印和边框。" + ), + ] + return "\n".join(section for section in sections if section) + + +def build_tryon_prompt_plan( + *, + context: ProductContext, + user_prompt: str, + count: int, + ratio: str, + product_reference_count: int, + has_model_portrait: bool = True, + has_model_triview: bool = False, + classification: ClassificationResult | None = None, +) -> TryonPromptPlan: + """生成一批确定性提示词；只规划，不调用图片模型。""" + + normalized_ratio = (ratio or "").strip() + if not normalized_ratio or ":" not in normalized_ratio: + raise ValueError("ratio 必须是有效的宽高比字符串，例如 3:4 或 7:10") + + classification = classification or classify_product_details(context, user_prompt) + kind = classification.kind + references = ReferenceRoles.create( + product_count=product_reference_count, + has_model_portrait=has_model_portrait, + has_model_triview=has_model_triview, + ) + shots = plan_shots(kind, count, user_prompt) + policy = _POLICIES[kind] + prompts = tuple( + _build_prompt( + context=context, + kind=kind, + policy=policy, + references=references, + user_prompt=user_prompt, + ratio=normalized_ratio, + shot=shot, + ) + for shot in shots + ) + return TryonPromptPlan(kind, classification, normalized_ratio, references, shots, prompts)