395 lines
18 KiB
Python
395 lines
18 KiB
Python
"""出图方案规划器：DeepSeek 根据采集的商品信息生成套图方案。
|
||
|
||
方案每项 = 一类图（标题 + 说明 + 生图提示 + 张数 + 可选 SKU 绑定），
|
||
生成时按方案逐张出图；参考图可按 variant_name 精确绑定到对应 SKU 图。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
|
||
import httpx
|
||
|
||
from config import get_settings
|
||
|
||
log = logging.getLogger("suite.planner")
|
||
|
||
# 规划器可选用的图类型（与 prompt.py 的 builder 对应）
|
||
ALLOWED_KINDS = [
|
||
"white_bg", "key_features", "selling_pt", "material",
|
||
"lifestyle", "multi_scene", "ecommerce_detail",
|
||
"size_chart", "sku_collection", "custom",
|
||
]
|
||
|
||
# 方案总张数安全上限（仅防模型跑飞，不干预正常规划；生图按张计费）：
|
||
# 默认 AI 自主决定张数（prompt 建议 8 张左右），上限 15；
|
||
# 「生图要求」明确写了「N 张」时按用户指定的来，上限放宽到 24。
|
||
PLAN_SAFETY_MAX = 15
|
||
PLAN_REQUIREMENTS_MAX = 24
|
||
MAX_VARIANT_MAIN_IMAGES = 2
|
||
|
||
# 规划器的图类归并。不同模型会把“细节图”写成材质图、卖点图或详情图，
|
||
# 因此均视作同一类覆盖，而不是强迫模型只使用 material。
|
||
SCENE_KINDS = {"lifestyle", "multi_scene"}
|
||
DETAIL_KINDS = {"material", "key_features", "selling_pt", "ecommerce_detail"}
|
||
_DIMENSION_PATTERN = re.compile(
|
||
r"尺寸|长宽高|直径|厚度|(?:长|宽|高)\s*[:：×x*]?\s*\d+|\d+(?:\.\d+)?\s*(?:厘米|毫米|cm|mm|inch|in)\b",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _total_budget(requirements: str | None) -> int:
|
||
"""从生图要求提取「N 张」作为总张数预算；未指定时用默认安全上限。"""
|
||
if requirements:
|
||
m = re.search(r"(\d+)\s*张", requirements)
|
||
if m:
|
||
return min(max(int(m.group(1)), 1), PLAN_REQUIREMENTS_MAX)
|
||
return PLAN_SAFETY_MAX
|
||
|
||
SYSTEM_PROMPT = """你是一名资深电商视觉策划。根据商品信息规划一套电商套图的出图方案。
|
||
|
||
## 输出硬性约束（违反即失败）
|
||
1. 输出必须是**单行紧凑 JSON**：无换行、无缩进、无空格填充、无注释、无 markdown 围栏。
|
||
2. 顶层只有 summary 和 items 两个字段；每个 item 严格只有 kind/title/detail/prompt_hint/count/variant_name 六个字段，不得增删。
|
||
3. 文本长度上限（中文字符/英文单词数）：summary ≤ 25 字；title ≤ 8 字；detail ≤ 20 字；prompt_hint ≤ 15 个英文词。超限必须删减，不得省略号截断。
|
||
4. count 默认 1，仅当该类图确需多个变体时才 >1，最大 3。总张数由你根据商品复杂度与信息量自主决定，不设硬性数量，但建议 8 张左右；若「生图要求」明确指定了张数，必须严格按指定张数输出。
|
||
5. variant_name 只能从「SKU规格」列表原样照抄；没有绑定就输出 null。
|
||
|
||
## 规划规则
|
||
1. 套图必须均衡覆盖：白底主图、场景图、细节图；参数含可靠的长宽高/尺寸时还必须有尺寸标注图。按这个顺序先预留张数，再安排其他类型。
|
||
2. 白底主图（kind=white_bg）通常只出 1 张；有多个视觉差异明显的 SKU 时最多选 2 个代表款各出 1 张，variant_name 填对应规格名。**禁止因为 SKU 多而为每个 SKU 都规划主图**。仅当「生图要求」明确要求每个/全部 SKU 分别出图时才可例外。
|
||
3. 场景图（kind=lifestyle）：至少 1 张，按核心使用场景出 1-2 张，每张聚焦一个场景。
|
||
4. 细节图（kind=material / key_features / selling_pt / ecommerce_detail）：至少 1 张，展示关键细节、材质或结构；通常 1-2 张，每张聚焦一个卖点。
|
||
5. 尺寸标注图（kind=size_chart）：参数含长宽高/尺寸时必须出 1 张；无可靠尺寸参数时不要虚构尺寸。
|
||
6. SKU 合集图（kind=sku_collection）：仅在 SKU >1 且确有必要展示差异时出 1 张，优先级低于上述四类。
|
||
7. kind 枚举：white_bg / key_features / selling_pt / material / lifestyle / multi_scene / ecommerce_detail / size_chart / sku_collection / custom。
|
||
8. title 用中文短语（如「主图·粉色」「浴室壁挂」）；detail 中文说明这张图展示什么；prompt_hint 用英文描述构图要点。
|
||
|
||
## 输出示例（紧凑单行）
|
||
{"summary":"三色收纳盒全套图","items":[{"kind":"white_bg","title":"主图·粉色","detail":"粉色SKU白底主视觉","prompt_hint":"front view on white background","count":1,"variant_name":"粉色"}]}"""
|
||
|
||
|
||
def _system_prompt_with_requirements(requirements: str | None) -> str:
|
||
"""把生图要求作为最高优先级约束注入 system prompt（置于规划规则之前）。
|
||
|
||
不仅声明优先级，还明确要求把要求落地到每个方案项的 prompt_hint，
|
||
避免模型只把要求当作背景信息而不影响输出。
|
||
"""
|
||
if not (requirements and requirements.strip()):
|
||
return SYSTEM_PROMPT
|
||
marker = "\n## 输出硬性约束"
|
||
idx = SYSTEM_PROMPT.find(marker)
|
||
if idx < 0:
|
||
return SYSTEM_PROMPT
|
||
req = requirements.strip()
|
||
block = (
|
||
"\n## 生图要求（最高优先级，硬性约束，覆盖下方所有规划规则与约束）\n"
|
||
+ req
|
||
+ "\n\n"
|
||
+ "规划方案时，必须把上述生图要求落地到每一项：\n"
|
||
+ "1. 每个方案项的 prompt_hint 必须融入上述要求的关键约束（如要求纯黑背景，则每个 prompt_hint 都要写明 black background）；\n"
|
||
+ "2. title / detail 措辞不得与上述要求矛盾；\n"
|
||
+ "3. 任何规划规则与上述要求冲突时，一律以本生图要求为准。\n"
|
||
)
|
||
return SYSTEM_PROMPT[:idx] + block + SYSTEM_PROMPT[idx:]
|
||
|
||
|
||
def _normalize_items(raw_items: list, sku_variants: list[str]) -> list[dict]:
|
||
"""清洗模型输出：kind 白名单、count 钳制、variant 必须真实存在。"""
|
||
items: list[dict] = []
|
||
for it in raw_items:
|
||
if not isinstance(it, dict):
|
||
continue
|
||
kind = str(it.get("kind") or "custom")
|
||
if kind not in ALLOWED_KINDS:
|
||
kind = "custom"
|
||
# title 缺失不再丢弃条目：从 detail / kind 兜底合成，避免整份方案被判空
|
||
title = str(it.get("title") or "").strip()[:20] or str(it.get("detail") or "").strip()[:12] or kind
|
||
try:
|
||
count = max(0, min(3, int(it.get("count", 1))))
|
||
except (TypeError, ValueError):
|
||
count = 1
|
||
variant = str(it.get("variant_name") or "").strip() or None
|
||
if variant and variant not in sku_variants:
|
||
variant = None # 幻觉规格：丢弃绑定，回退主图
|
||
items.append({
|
||
"kind": kind,
|
||
"title": title,
|
||
"detail": str(it.get("detail") or "").strip()[:80],
|
||
"prompt_hint": str(it.get("prompt_hint") or "").strip()[:300],
|
||
"count": count,
|
||
"variant_name": variant,
|
||
})
|
||
return items
|
||
|
||
|
||
def _has_dimensions(product_info: dict) -> bool:
|
||
"""仅在已采集到尺寸信息时规划尺寸图，避免模型编造数值。"""
|
||
for param in product_info.get("params") or []:
|
||
if not isinstance(param, dict):
|
||
continue
|
||
text = f"{param.get('key') or ''} {param.get('value') or ''}"
|
||
if _DIMENSION_PATTERN.search(text):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _requires_every_sku(requirements: str | None) -> bool:
|
||
"""用户显式要求逐 SKU 出图时，保留其优先级，不施加代表款上限。"""
|
||
if not requirements:
|
||
return False
|
||
return bool(re.search(
|
||
r"(?:每个|全部|所有|逐个|分别).{0,10}(?:sku|SKU|规格|款式|颜色|变体)",
|
||
requirements,
|
||
))
|
||
|
||
|
||
def _coverage_group(kind: str) -> str:
|
||
if kind == "white_bg":
|
||
return "main"
|
||
if kind in SCENE_KINDS:
|
||
return "scene"
|
||
if kind in DETAIL_KINDS:
|
||
return "detail"
|
||
if kind == "size_chart":
|
||
return "size"
|
||
return "other"
|
||
|
||
|
||
def _total(items: list[dict]) -> int:
|
||
return sum(item["count"] for item in items)
|
||
|
||
|
||
def _group_total(items: list[dict], group: str) -> int:
|
||
return sum(item["count"] for item in items if _coverage_group(item["kind"]) == group)
|
||
|
||
|
||
def _remove_one(items: list[dict], predicate) -> bool:
|
||
"""从后往前删除一个生成槽位，优先保留模型较早规划的构图。"""
|
||
for index in range(len(items) - 1, -1, -1):
|
||
if not predicate(items[index]):
|
||
continue
|
||
if items[index]["count"] > 1:
|
||
items[index]["count"] -= 1
|
||
else:
|
||
items.pop(index)
|
||
return True
|
||
return False
|
||
|
||
|
||
def _make_room(items: list[dict], budget: int, minimums: dict[str, int]) -> bool:
|
||
"""为必需图类腾出一张，先去除非核心图或某类的冗余张数。"""
|
||
candidates: list[tuple[int, int]] = []
|
||
for index, item in enumerate(items):
|
||
group = _coverage_group(item["kind"])
|
||
if _group_total(items, group) <= minimums.get(group, 0):
|
||
continue
|
||
# other 最低优先；同类冗余时从后面的项开始删。
|
||
priority = 0 if group == "other" else 1
|
||
candidates.append((priority, -index))
|
||
if not candidates:
|
||
return False
|
||
_, neg_index = min(candidates)
|
||
index = -neg_index
|
||
item = items[index]
|
||
if item["count"] > 1:
|
||
item["count"] -= 1
|
||
else:
|
||
items.pop(index)
|
||
return True
|
||
|
||
|
||
def _balanced_items(
|
||
items: list[dict],
|
||
sku_variants: list[str],
|
||
product_info: dict,
|
||
requirements: str | None,
|
||
budget: int,
|
||
) -> list[dict]:
|
||
"""将 LLM 方案约束为信息均衡的商品套图。
|
||
|
||
提示词能改善规划倾向，但不能保证每次输出。这里把 SKU 主图总量限制为
|
||
代表款数量，并在预算允许时补齐核心图类，避免 SKU 多时挤掉场景/细节/尺寸图。
|
||
"""
|
||
balanced = [dict(item) for item in items if item["count"] > 0]
|
||
|
||
if not _requires_every_sku(requirements):
|
||
max_mains = 1 if len(sku_variants) <= 1 else MAX_VARIANT_MAIN_IMAGES
|
||
while sum(item["count"] for item in balanced if item["kind"] == "white_bg") > max_mains:
|
||
_remove_one(balanced, lambda item: item["kind"] == "white_bg")
|
||
|
||
required: list[tuple[str, dict]] = [
|
||
("main", {
|
||
"kind": "white_bg", "title": "白底主图", "detail": "商品白底主视觉",
|
||
"prompt_hint": "front view on pure white background", "count": 1,
|
||
"variant_name": sku_variants[0] if sku_variants else None,
|
||
}),
|
||
("scene", {
|
||
"kind": "lifestyle", "title": "使用场景", "detail": "展示商品实际使用场景",
|
||
"prompt_hint": "product in a natural everyday use setting", "count": 1,
|
||
"variant_name": None,
|
||
}),
|
||
("detail", {
|
||
"kind": "material", "title": "细节展示", "detail": "突出材质与结构细节",
|
||
"prompt_hint": "close-up of authentic material and construction details", "count": 1,
|
||
"variant_name": None,
|
||
}),
|
||
]
|
||
if _has_dimensions(product_info):
|
||
required.append(("size", {
|
||
"kind": "size_chart", "title": "尺寸标注", "detail": "展示已采集的商品尺寸",
|
||
"prompt_hint": "clear size chart using known product dimensions", "count": 1,
|
||
"variant_name": None,
|
||
}))
|
||
|
||
# 用户明确要求较小张数时，优先保留前面的核心图类；不突破其指定预算。
|
||
required = required[:budget]
|
||
minimums = {group: 1 for group, _ in required}
|
||
for group, fallback in required:
|
||
if _group_total(balanced, group) > 0:
|
||
continue
|
||
while _total(balanced) >= budget and _make_room(balanced, budget, minimums):
|
||
pass
|
||
if _total(balanced) < budget:
|
||
balanced.append(fallback)
|
||
|
||
while _total(balanced) > budget and _make_room(balanced, budget, minimums):
|
||
pass
|
||
return balanced
|
||
|
||
|
||
def _repair_truncated(s: str) -> dict | None:
|
||
"""截断修复：从最后一个完整的 '}' 处截断，剥尾逗号后按括号配平补全闭合。
|
||
|
||
适用于「items 数组中途被 max_tokens 截断」的场景——截断点在完整对象边界，
|
||
此前的字符串必然已闭合，简单计数配平即可。
|
||
"""
|
||
for cut in (m.end() for m in reversed(list(re.finditer(r'\}', s)))):
|
||
cand = s[:cut].rstrip().rstrip(',')
|
||
opens: list[str] = []
|
||
for ch in cand:
|
||
if ch in '{[':
|
||
opens.append(ch)
|
||
elif ch == '}' and opens and opens[-1] == '{':
|
||
opens.pop()
|
||
elif ch == ']' and opens and opens[-1] == '[':
|
||
opens.pop()
|
||
suffix = ''.join('}' if o == '{' else ']' for o in reversed(opens))
|
||
try:
|
||
data = json.loads(cand + suffix)
|
||
if isinstance(data, dict):
|
||
return data
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _extract_json(text: str) -> dict:
|
||
"""从模型输出提取 JSON：剥离思考块/markdown 围栏，截断时尝试修复。"""
|
||
s = (text or '').strip()
|
||
# 剥离思考块（思考型模型会把推理过程放进 <think>）
|
||
s = re.sub(r'<think>.*?</think>', '', s, flags=re.S).strip()
|
||
# 剥离 markdown 代码围栏
|
||
m = re.search(r'```(?:json)?\s*(.*?)```', s, flags=re.S)
|
||
if m:
|
||
s = m.group(1).strip()
|
||
try:
|
||
return json.loads(s)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
start = s.find('{')
|
||
if start >= 0:
|
||
repaired = _repair_truncated(s[start:])
|
||
if repaired is not None:
|
||
log.warning("规划器输出疑似被截断，已自动截断修复（可能丢失末尾部分方案项）")
|
||
return repaired
|
||
raise ValueError("模型输出无法解析为 JSON")
|
||
|
||
|
||
async def generate_plan(
|
||
product_info: dict,
|
||
sku_variants: list[str],
|
||
image_stats: dict,
|
||
platform: str,
|
||
requirements: str | None = None,
|
||
) -> dict:
|
||
"""调用 DeepSeek 生成方案。返回 {summary, items}。
|
||
|
||
requirements：生图要求，最高优先级注入 system prompt，规划方案必须遵循。
|
||
"""
|
||
s = get_settings()
|
||
if not s.deepseek_api_key:
|
||
raise RuntimeError("未配置 DEEPSEEK_API_KEY（.env）")
|
||
|
||
user_payload: dict = {
|
||
"商品信息": product_info, # {title, desc, params:[{key,value}], sellingPoints, price}
|
||
"SKU规格": sku_variants, # 带图的 SKU 规格名（variant_name 只能从中选）
|
||
"图片统计": image_stats, # {main: n, sku: n, detail: n}
|
||
"目标平台": platform, # ozon/wb/cn（决定图内文案语言）
|
||
}
|
||
# 生图要求同时在 user 侧强调（与 system prompt 双重约束），确保模型真正遵循
|
||
if requirements and requirements.strip():
|
||
user_payload["生图要求（最高优先级，必须体现在每个方案项中）"] = requirements.strip()
|
||
user_content = json.dumps(user_payload, ensure_ascii=False)
|
||
|
||
last_error: RuntimeError | ValueError | None = None
|
||
for attempt in range(2): # 解析失败/空方案自动重试一次
|
||
async with httpx.AsyncClient(timeout=90, verify=False) as client:
|
||
resp = await client.post(
|
||
f"{s.deepseek_base_url.rstrip('/')}/chat/completions",
|
||
headers={"Authorization": f"Bearer {s.deepseek_api_key}", "Content-Type": "application/json"},
|
||
json={
|
||
"model": s.deepseek_model,
|
||
"messages": [
|
||
{"role": "system", "content": _system_prompt_with_requirements(requirements)},
|
||
{"role": "user", "content": user_content},
|
||
],
|
||
"response_format": {"type": "json_object"},
|
||
"temperature": 0.3,
|
||
"max_tokens": 8000,
|
||
# deepseek-v4 是思维链模型：不禁用 thinking 会把 max_tokens 烧在推理上，
|
||
# 正文为空 → 只能拿 reasoning_content 兜底解析 → 频发「解析失败/未返回有效方案项」
|
||
"thinking": {"type": "disabled"},
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
body = resp.json()
|
||
message = body["choices"][0]["message"]
|
||
finish_reason = body["choices"][0].get("finish_reason", "")
|
||
usage = body.get("usage") or {}
|
||
log.info(
|
||
"规划器 token 用量: prompt=%s completion=%s finish=%s",
|
||
usage.get("prompt_tokens", "?"), usage.get("completion_tokens", "?"), finish_reason,
|
||
)
|
||
content = message.get("content") or ""
|
||
# 思考型输出：content 为空时从 reasoning_content 里捞
|
||
if not content.strip() and message.get("reasoning_content"):
|
||
content = message["reasoning_content"]
|
||
|
||
try:
|
||
data = _extract_json(content)
|
||
except ValueError as exc:
|
||
last_error = exc
|
||
log.error(
|
||
"规划器第 %d 次输出解析失败 finish_reason=%s content[:200]=%s",
|
||
attempt + 1, finish_reason, content[:200],
|
||
)
|
||
continue
|
||
|
||
items = _normalize_items(data.get("items") or [], sku_variants)
|
||
if items:
|
||
# 先压缩多 SKU 的重复主图、补齐核心视觉类别，再执行总量保护。
|
||
# 生图要求指定「N 张」时按 N 放宽上限，并始终不突破该预算。
|
||
budget = _total_budget(requirements)
|
||
items = _balanced_items(items, sku_variants, product_info, requirements, budget)
|
||
return {"summary": str(data.get("summary") or "").strip()[:100], "items": items}
|
||
|
||
last_error = RuntimeError("规划器未返回有效方案项")
|
||
log.warning(
|
||
"规划器第 %d 次返回无有效方案项 finish_reason=%s content[:200]=%s",
|
||
attempt + 1, finish_reason, content[:200],
|
||
)
|
||
|
||
raise last_error or RuntimeError("规划器未返回有效方案项")
|