style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

#2211的判定逻辑有bug：用'开头是否是默认XML标记'判断是否用户自定义，
但用户后台配的prompt就是以这句话开头的XML格式，会被误判为内置默认不生效。

修正为简单直接规则（不做字符串匹配判断）：
- DB有is_active=true的image_analysis记录（含种子默认XML和用户修改版）：
  system = DB.system_prompt + JSON_SCHEMA_APPEND（追加完整JSON字段schema，
  末尾硬约束'不要输出XML标签'，覆盖XML输出格式要求）
- DB无记录/连接异常/返回空：纯硬编码JSON schema prompt
- 直接调prompt_loader._load_from_db查DB（不用get_template避免被
  DEFAULT_TEMPLATES fallback混淆）
- 30s TTL缓存，fast追加22字段schema，pro追加13字段精简schema

apps/worker/worker_app/tasks/viral_video.py

@@ -421,14 +421,10 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
             render_system_prompt,
             render_user_prompt,
         )
-        from packages.shared.ai_client import get_doubao_client
+        from packages.shared.ai_service import call_llm
     except ImportError:
         return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""}
 
-    _llm_client = get_doubao_client()
-    if not _llm_client.is_available:
-        return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""}
-
     products_summary = ""
     products = (image_analysis or {}).get("products", []) or []
     for p in products:
@@ -483,13 +479,13 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
     for _m, _lbl in [(_fast, "fast"), (_pro, "pro-fallback")]:
         try:
             logger.info("[爆款视频] 意图解析 model=%s label=%s", _m, _lbl)
-            raw = _llm_client.chat_completion(
+            raw = call_llm(
                 [{"role": "system", "content": system}, {"role": "user", "content": user}],
                 temperature=0.4,
                 max_tokens=1024,
                 model=_m,
                 timeout=60,
-            )  # #2180/#2215: 直接用 client.chat_completion 传 messages list，不再走 call_llm 字符串包装
+            )  # #2180: 意图解析 LLM 实测需更长响应，原25s太紧
             if not raw:
                 continue
             parsed = _parse(raw)
@@ -829,14 +825,10 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
             GLOBAL_CONSTRAINTS,
             NEGATIVE_RULES,
         )
-        from packages.shared.ai_client import get_doubao_client
+        from packages.shared.ai_service import call_llm
     except ImportError:
         return _fallback_script(job)
 
-    _llm_client2 = get_doubao_client()
-    if not _llm_client2.is_available:
-        return _fallback_script(job)
-
     products_summary = _build_products_summary(image_analysis)
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
 
@@ -872,7 +864,7 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
 
     def _try_gen(model: str, temp: float, max_tok: int, label: str, tmo: int = 25):
         logger.info("[爆款视频] 编导脚本生成 model=%s label=%s timeout=%d", model, label, tmo)
-        raw = _llm_client2.chat_completion(
+        raw = call_llm(
             [{"role": "system", "content": system_tpl}, {"role": "user", "content": user}],
             temperature=temp,
             max_tokens=max_tok,

apps/worker/worker_app/tasks/vision/_prompt.py

@@ -3,9 +3,8 @@
 且 is_active=true），30s TTL 热加载；DB 无有效记录/异常时，fallback 到纯硬编码 JSON schema prompt。
 
 规则（简单直接，不做字符串匹配判断）：
-- DB 有 is_active=true 的 image_analysis 记录（含种子版本和用户修改后的版本）：
-  * system = DB.system_prompt（DB prompt 自带完整输出格式，不追加硬编码 schema，
-    避免 DB 写 XML、调用强制 json_object 造成的格式冲突）
+- DB 有 is_active=true 的 image_analysis 记录（含种子默认XML和用户修改后的版本）：
+  * system = DB.system_prompt + JSON_SCHEMA_APPEND（追加完整JSON字段schema，覆盖XML等其他输出格式要求）
   * user   = DB.user_prompt_template 渲染后使用；渲染后为空则用硬编码默认
 - DB 无记录/连接异常/返回空：system/user 全部用纯硬编码 JSON schema prompt
 """
@@ -74,8 +73,7 @@ _PRO_JSON_SCHEMA = (
 )
 DEFAULT_PRO_USER = "分析这张图片，返回符合schema的JSON。"
 
-# 保留旧 JSON schema 追加文本作为常量（DB prompt 完全控制输出格式后不再使用，
-# 保留以便排查历史行为）。
+# DB 配置存在时，追加在用户 system_prompt 末尾的JSON schema约束
 _FAST_JSON_APPEND = (
     "\n\n【输出格式要求】无论上文如何要求，最终你必须只返回一个合法的JSON对象，"
     "严格包含以下字段（字段值不确定时填null或空数组）：\n"
@@ -175,6 +173,7 @@ def _resolve(kind: str) -> tuple[str, str]:
 
     default_sys = _FAST_JSON_SCHEMA if kind == "fast" else _PRO_JSON_SCHEMA
     default_user = DEFAULT_FAST_USER if kind == "fast" else DEFAULT_PRO_USER
+    append = _FAST_JSON_APPEND if kind == "fast" else _PRO_JSON_APPEND
 
     sys_prompt = default_sys
     usr_prompt = default_user
@@ -183,7 +182,7 @@ def _resolve(kind: str) -> tuple[str, str]:
         if tpl is not None:
             db_sys = (getattr(tpl, "system_prompt", "") or "").strip()
             if db_sys:
-                sys_prompt = db_sys  # DB prompt自带完整输出格式，不追加硬编码schema避免冲突
+                sys_prompt = db_sys + append
                 usr_prompt = _render_user(tpl, default_user)
                 logger.info(
                     "[vision.v2] 使用DB image_analysis prompt (kind=%s version=%s sys_len=%d)",

apps/worker/worker_app/tasks/vision/assembler.py

@@ -1,8 +1,5 @@
 # -*- coding: utf-8 -*-
-"""把 fast_json VLM 输出 + OCR 文本组装为下游兼容的 product dict。
-
-v4 schema: DB prompt完全控制输出格式，可能是v4嵌套schema(type/products/people/store_info)
-或旧扁平schema(has_person/upper_wear/product_name/brand等)。assembler兼容两种格式。
+"""把 fast_json VLM 输出 + OCR 文本组装为与旧 _normalize() 完全一致的 dict。
 
 目标：下游（信任链t2i/intent_parsing/script_generation）零改动。
 必出字段：name, brand, category, appearance, packaging, text_on_package,
@@ -13,16 +10,29 @@ from __future__ import annotations
 
 from typing import Any
 
+# ---------- portrait_prompt 模板 ----------
+# 目标：60-100 字的人物穿搭描述，用于 Seedream 纯文生图。要求具体、风格化、视觉细节丰富。
+# 旧 VLM 输出格式参考："一位25岁左右的亚洲女性，身穿白色V领短袖T恤，黑色高腰阔腿裤，
+# 搭配银色项链，长发披肩，表情自信，街拍风格，阳光明媚的城市街头"
+
 
 def _join_parts(*parts: str | None) -> str:
     return "".join(p for p in parts if p)
 
 
-_AGE_PREFIX = {"青年": "年轻", "中年": "中年", "老年": "老年"}
+_AGE_PREFIX = {
+    "青年": "年轻",
+    "中年": "中年",
+    "老年": "老年",
+}
+# gender 后缀
 _GENDER_WORD = {"男": "男性", "女": "女性"}
 
 
-def _person_subject(gender: str, age: str) -> str:
+def _person_subject(fj: dict[str, Any]) -> str:
+    """人物主语：年轻女性 / 中年男性 / 少女 / 小男孩 / 人物 等。"""
+    gender = fj.get("gender") or ""
+    age = fj.get("age_range") or ""
     gw = _GENDER_WORD.get(gender, "")
     if age == "儿童":
         if gender == "女":
@@ -42,147 +52,8 @@ def _person_subject(gender: str, age: str) -> str:
     return f"{prefix}人物" if prefix else "人物"
 
 
-def _build_wear_from_v4(p: dict) -> str:
-    """v4 person schema: upper_wear/upper_color/lower_wear/lower_color/dress_color"""
-    upper = p.get("upper_wear") or ""
-    upper_color = p.get("upper_color") or ""
-    lower = p.get("lower_wear") or ""
-    lower_color = p.get("lower_color") or ""
-    dress_color = p.get("dress_color") or ""
-    is_dress = ("连衣裙" in upper) or ("裙" in upper and not lower)
-    if is_dress:
-        c = dress_color or upper_color
-        return f"身穿{c}{upper}" if c else f"身穿{upper}"
-    parts = []
-    if upper:
-        up = f"{upper_color}{upper}" if upper_color else upper
-        parts.append(f"上身{up}")
-    if lower:
-        lo = f"{lower_color}{lower}" if lower_color else lower
-        parts.append(f"下身{lo}")
-    return "，".join(parts)
-
-
-def _build_portrait_prompt_from_v4(p: dict) -> str:
-    """v4 person: 直接用portrait_prompt字段；没有就拼"""
-    direct = p.get("portrait_prompt")
-    if direct and len(direct) >= 10:
-        return direct
-    subject = _person_subject(p.get("gender", ""), p.get("age_range", ""))
-    wear = _build_wear_from_v4(p)
-    acc = p.get("accessories") or []
-    if isinstance(acc, str):
-        acc = [acc]
-    acc_str = "，佩戴" + "、".join(str(a) for a in acc if a) if acc else ""
-    hair = p.get("hairstyle") or ""
-    expr = p.get("expression") or ""
-    pose = p.get("pose") or ""
-    style = p.get("outfit_style") or p.get("style") or ""
-    scene = p.get("scene") or ""
-    mood = p.get("mood") or ""
-    details = []
-    if hair:
-        details.append(hair)
-    if expr and expr not in ("自然", "平静"):
-        details.append(f"神情{expr}")
-    if pose and pose not in ("站立",):
-        details.append(pose)
-    style_parts = []
-    if style:
-        style_parts.append(style)
-    if mood:
-        style_parts.append(mood)
-    if scene and scene not in ("通用",):
-        style_parts.append(scene)
-    pieces = [f"一位{subject}"]
-    if wear:
-        pieces.append(wear)
-    if acc_str:
-        pieces.append(acc_str.lstrip("，"))
-    if details:
-        pieces.append("，".join(details))
-    pieces.append(("".join(style_parts) + "风格") if style_parts else "人像写真")
-    full = "，".join(p for p in pieces if p)
-    if len(full) < 40:
-        full += "，自然光线下人像特写，画面清晰"
-    if len(full) > 120:
-        full = full[:120].rstrip("，") + "。"
-    return full
-
-
-def _build_product_prompt_from_v4(prod: dict, top: dict) -> str:
-    """v4 product: 拼商品视觉描述prompt（用于AI生图参考）"""
-    name = prod.get("product_name") or "商品"
-    brand = prod.get("brand") or ""
-    lead = f"{brand} {name}" if brand and brand not in name else name
-    pkg_color = prod.get("package_color") or ""
-    pkg_type = prod.get("package_type") or ""
-    cap = prod.get("cap_type") or ""
-    body = prod.get("body_shape") or ""
-    features = prod.get("product_features") or []
-    sell = prod.get("key_selling_points") or []
-    colors = top.get("colors") or []
-    style = top.get("style") or ""
-    scene = top.get("scene") or ""
-    mood = top.get("mood") or ""
-
-    parts = [lead]
-    desc = []
-    if pkg_color:
-        desc.append(pkg_color)
-    if pkg_type:
-        desc.append(pkg_type)
-    if cap and len(desc) < 3:
-        desc.append(f"配{cap}")
-    if body and len(desc) < 3:
-        desc.append(body)
-    if desc:
-        parts.append("，".join(desc))
-    if features:
-        core = [str(f) for f in features[:3] if f and len(str(f)) <= 25]
-        if core:
-            parts.append("；".join(core))
-    if sell:
-        s = [str(x) for x in sell[:2] if x]
-        if s:
-            parts.append("突出" + "、".join(s))
-    cnames = []
-    for cc in colors:
-        if isinstance(cc, dict) and cc.get("name"):
-            cnames.append(cc["name"])
-        elif isinstance(cc, str):
-            cnames.append(cc)
-    cnames = cnames[:3]
-    if cnames:
-        parts.append("、".join(cnames) + "主色")
-    if style:
-        parts.append(style)
-    if mood:
-        parts.append(mood)
-    if scene and not any(k in scene for k in ("白色背景", "纯色", "通用")):
-        parts.append(scene)
-    parts.append("产品特写，画面清晰")
-    prompt = "，".join(p for p in parts if p)
-    return prompt if len(prompt) >= 10 else "产品展示图，特写镜头"
-
-
-def _is_v4_schema(fj: dict) -> bool:
-    """判断是v4嵌套schema还是旧扁平schema"""
-    return (
-        isinstance(fj.get("products"), list)
-        or fj.get("type") in ("product", "store", "person", "other")
-        or isinstance(fj.get("people"), dict)
-    )
-
-
-# ---------- 旧扁平schema兼容（保留原逻辑） ----------
-
-
-def _person_subject_old(fj: dict) -> str:
-    return _person_subject(fj.get("gender", ""), fj.get("age_range", ""))
-
-
-def _build_wear_sentence_old(fj: dict) -> str:
+def _build_wear_sentence(fj: dict[str, Any]) -> str:
+    """穿搭段：上装+下装/连衣裙，带颜色+材质+图案。"""
     upper = fj.get("upper_wear") or ""
     upper_color = fj.get("upper_color") or ""
     lower = fj.get("lower_wear") or ""
@@ -190,6 +61,7 @@ def _build_wear_sentence_old(fj: dict) -> str:
     dress_color = fj.get("dress_color") or ""
     material = fj.get("material") or ""
     pattern = fj.get("pattern") or ""
+
     is_dress = ("连衣裙" in upper) or ("裙" in upper and not lower)
     if is_dress:
         c = dress_color or upper_color
@@ -199,22 +71,25 @@ def _build_wear_sentence_old(fj: dict) -> str:
         if pattern and pattern not in wear and pattern != "纯色":
             wear += f"，{pattern}图案"
         return f"身穿{wear}"
-    parts = []
+
+    parts: list[str] = []
     if upper:
         up = f"{upper_color}{upper}" if upper_color else upper
         if material and material not in up:
             up = f"{material}{up}"
         if pattern and pattern != "纯色" and pattern not in up:
             up += f"（{pattern}）"
-        parts.append(f"上身{up}")
+        parts.append(f"上身{up}" if up else "")
     if lower:
         lo = f"{lower_color}{lower}" if lower_color else lower
-        parts.append(f"下身{lo}")
+        parts.append(f"下身{lo}" if lo else "")
     return "，".join(p for p in parts if p)
 
 
-def _build_portrait_prompt_old(fj: dict) -> str:
+def _build_portrait_prompt(fj: dict[str, Any]) -> str:
+    """组装最终 portrait_prompt（目标 60-100 字，用于 Seedream 纯文生图）。"""
     if not fj.get("has_person"):
+        # 非人像：用商品+场景+mood 拼一段
         name = fj.get("product_name") or "商品"
         brand = fj.get("brand") or ""
         colors = fj.get("colors") or []
@@ -226,15 +101,7 @@ def _build_portrait_prompt_old(fj: dict) -> str:
             pieces.append(brand)
         pieces.append(name)
         if colors:
-            cnames = []
-            for c in colors:
-                if isinstance(c, dict):
-                    cnames.append(c.get("name", ""))
-                elif isinstance(c, str):
-                    cnames.append(c)
-            cnames = [c for c in cnames if c][:3]
-            if cnames:
-                pieces.append("、".join(cnames) + "配色")
+            pieces.append("、".join(colors[:3]) + "配色")
         if style:
             pieces.append(style + "风格")
         if mood:
@@ -244,32 +111,40 @@ def _build_portrait_prompt_old(fj: dict) -> str:
         pieces.append("产品特写")
         prompt = "，".join(p for p in pieces if p)
         return prompt if len(prompt) >= 10 else "产品展示图，特写镜头"
-    subject = _person_subject_old(fj)
-    wear = _build_wear_sentence_old(fj)
+
+    subject = _person_subject(fj)
+    wear = _build_wear_sentence(fj)
+
     accessories = fj.get("accessories") or []
     if isinstance(accessories, str):
         accessories = [accessories]
-    acc_str = "，佩戴" + "、".join(str(a) for a in accessories if a) if accessories else ""
+    acc_str = ""
+    if accessories:
+        acc_str = "，佩戴" + "、".join(str(a) for a in accessories if a)
+
     hairstyle = fj.get("hairstyle") or ""
     expression = fj.get("expression") or ""
     pose = fj.get("pose") or ""
     style = fj.get("style") or ""
     scene = fj.get("scene") or ""
     mood = fj.get("mood") or ""
-    detail_parts = []
+
+    detail_parts: list[str] = []
     if hairstyle:
         detail_parts.append(hairstyle)
     if expression and expression not in ("自然", "平静"):
         detail_parts.append(f"神情{expression}")
     if pose and pose not in ("站立",):
         detail_parts.append(pose)
-    style_parts = []
+
+    style_parts: list[str] = []
     if style:
         style_parts.append(style)
     if mood:
         style_parts.append(mood)
     if scene and scene not in ("通用",):
         style_parts.append(scene)
+
     pieces = [f"一位{subject}"]
     if wear:
         pieces.append(wear)
@@ -277,40 +152,53 @@ def _build_portrait_prompt_old(fj: dict) -> str:
         pieces.append(acc_str.lstrip("，"))
     if detail_parts:
         pieces.append("，".join(detail_parts))
-    pieces.append("".join(style_parts) + "风格" if style_parts else "人像写真")
+    if style_parts:
+        # 风格词之间不用逗号，用空格紧凑
+        pieces.append("".join(style_parts) + "风格")
+    else:
+        pieces.append("人像写真")
+
     full = "，".join(p for p in pieces if p)
+    # 过短补充镜头词
     if len(full) < 40:
         full += "，自然光线下人像特写，画面清晰"
+    # 过长截断
     if len(full) > 120:
         full = full[:120].rstrip("，") + "。"
     return full
 
 
-def _infer_name_old(fj: dict, ocr_texts: list[str]) -> str:
+# ---------- 商品字段 ----------
+
+
+def _infer_name(fj: dict[str, Any], ocr_texts: list[str]) -> str:
     pname = fj.get("product_name")
     if pname and pname != "未识别":
         return str(pname)
+    # 人物图 → name 用穿搭主件
     if fj.get("has_person"):
         up = fj.get("upper_wear") or ""
         if "连衣裙" in up:
             return up
         return up or "人物穿搭"
     if ocr_texts:
+        # 商品名可能是 OCR 最长的一行（品牌/产品名）
         return max(ocr_texts, key=len)
     return "未识别"
 
 
-def _infer_brand_old(fj: dict, ocr_texts: list[str]) -> str:
+def _infer_brand(fj: dict[str, Any], ocr_texts: list[str]) -> str:
     brand = fj.get("brand")
     if brand:
         return str(brand)
+    # OCR 里短的、纯字母/汉字短串可能是 brand
     for t in ocr_texts:
         if 1 < len(t) <= 12:
             return t
     return "无法判断"
 
 
-def _infer_category_old(fj: dict) -> str:
+def _infer_category(fj: dict[str, Any]) -> str:
     cat = fj.get("category")
     if cat:
         return str(cat)
@@ -319,19 +207,27 @@ def _infer_category_old(fj: dict) -> str:
     return "非产品图"
 
 
-def _build_appearance_old(fj: dict) -> str:
-    parts = []
-    for key in ("upper_color", "upper_wear", "material", "pattern"):
+def _build_appearance(fj: dict[str, Any]) -> str:
+    """外观描述：颜色+款式+材质+图案 拼成一段。"""
+    parts: list[str] = []
+    for key, _label in [
+        ("upper_color", "主色"),
+        ("upper_wear", "款式"),
+        ("material", "材质"),
+        ("pattern", "图案"),
+    ]:
         v = fj.get(key)
         if v and v not in ("无法判断", "未知", "纯色"):
             parts.append(str(v))
     if not parts:
-        return "人像穿搭整体造型" if fj.get("has_person") else "无法判断"
+        if fj.get("has_person"):
+            return "人像穿搭整体造型"
+        return "无法判断"
     return "、".join(parts)
 
 
-def _build_key_features_old(fj: dict, ocr_texts: list[str]) -> list[str]:
-    feats = []
+def _build_key_features(fj: dict[str, Any], ocr_texts: list[str]) -> list[str]:
+    feats: list[str] = []
     for key in (
         "upper_wear",
         "lower_wear",
@@ -352,7 +248,9 @@ def _build_key_features_old(fj: dict, ocr_texts: list[str]) -> list[str]:
             feats.append(v)
     if ocr_texts:
         feats.append(f"画面文字: {'/'.join(ocr_texts[:3])}")
-    out, seen = [], set()
+    # 去重
+    out: list[str] = []
+    seen: set[str] = set()
     for f in feats:
         f = f.strip()
         if f and f not in seen and len(f) <= 30:
@@ -361,275 +259,27 @@ def _build_key_features_old(fj: dict, ocr_texts: list[str]) -> list[str]:
     return out[:6] if out else ["无法判断"]
 
 
-def _flatten_colors(c) -> list[str]:
-    """colors可能是字符串数组或[{hex,name,coverage}]，统一返回名字数组"""
-    if not c:
-        return []
-    out = []
-    for item in c:
-        if isinstance(item, dict):
-            n = item.get("name")
-            if n:
-                out.append(n)
-        elif isinstance(item, str):
-            out.append(item)
-    return out
-
-
-def assemble_result(idx: int, fast_json: dict | None, ocr_texts: list[str]) -> dict[str, Any]:
+def assemble_result(
+    idx: int,
+    fast_json: dict[str, Any] | None,
+    ocr_texts: list[str],
+) -> dict[str, Any]:
+    """把 fast_json 结果 + OCR 文本组装成下游兼容的 product dict。"""
     fj = fast_json or {}
     ocr_texts = ocr_texts or []
 
-    if _is_v4_schema(fj):
-        return _assemble_v4(idx, fj, ocr_texts)
-    else:
-        return _assemble_old(idx, fj, ocr_texts)
-
-
-def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
-    """v4嵌套schema → 下游product dict"""
-    vtype = fj.get("type") or "other"
-    products = fj.get("products") or []
+    portrait_prompt = _build_portrait_prompt(fj)
+    name = _infer_name(fj, ocr_texts)
+    brand = _infer_brand(fj, ocr_texts)
+    category = _infer_category(fj)
+    appearance = _build_appearance(fj)
+    key_features = _build_key_features(fj, ocr_texts)
     scene = fj.get("scene") or "通用"
     mood = fj.get("mood") or ""
-    colors = fj.get("colors") or []
-    visible_text = fj.get("visible_text") or []
-    color_names = _flatten_colors(colors)
-
-    # 合并OCR文字和visible_text
-    pkg_texts = []
-    for vt in visible_text:
-        if isinstance(vt, dict):
-            t = vt.get("text")
-            if t:
-                pkg_texts.append(str(t))
-        elif isinstance(vt, str):
-            pkg_texts.append(vt)
-    pkg_texts.extend(ocr_texts[:5])
-    # 去重
-    seen_t = set()
-    text_on_package = []
-    for t in pkg_texts:
-        t = str(t).strip()
-        if t and t not in seen_t and len(t) <= 50:
-            seen_t.add(t)
-            text_on_package.append(t)
-    text_on_package = text_on_package[:8]
-
-    has_person = fj.get("has_person", False)
-
-    # 人物类
-    if vtype == "person" or has_person:
-        # 取第一个人物信息（v4 schema人物信息在顶层）
-        person_info = fj
-        # 兼容people嵌套
-        ppl = fj.get("people")
-        if isinstance(ppl, dict) and ppl.get("has_person"):
-            person_info = {**fj, **ppl}
-            has_person = True
-
-        portrait_prompt = _build_portrait_prompt_from_v4(person_info)
-        name = person_info.get("upper_wear") or "人物穿搭"
-        if "连衣裙" in name:
-            pass
-        else:
-            lower = person_info.get("lower_wear") or ""
-            if lower:
-                name = f"{name}+{lower}"
-        brand = "无法判断"
-        category = "服饰"
-        outfit_parts = []
-        for k in ("upper_wear", "lower_wear", "dress_color", "upper_color", "lower_color", "outfit_style"):
-            v = person_info.get(k)
-            if v and v not in ("null", None):
-                outfit_parts.append(str(v))
-        appearance = "、".join(outfit_parts) if outfit_parts else "人像穿搭整体造型"
-        # key_features: 穿搭特征+配饰
-        kf = []
-        for k in (
-            "upper_wear",
-            "lower_wear",
-            "upper_color",
-            "lower_color",
-            "hairstyle",
-            "expression",
-            "pose",
-            "outfit_style",
-        ):
-            v = person_info.get(k)
-            if v and v not in ("null", None, "无法判断"):
-                kf.append(str(v))
-        acc = person_info.get("accessories") or []
-        if isinstance(acc, list):
-            kf.extend(str(a) for a in acc if a)
-        if text_on_package:
-            kf.append(f"画面文字: {'/'.join(text_on_package[:3])}")
-        kf = kf[:6] or ["无法判断"]
-        summary = (person_info.get("outfit_style") or "") + (person_info.get("upper_wear") or "穿搭")
-        if not summary or summary == "穿搭":
-            summary = "人物穿搭"
-        return {
-            "name": name[:30],
-            "brand": brand,
-            "category": category,
-            "appearance": appearance,
-            "packaging": "人物形象无包装",
-            "text_on_package": text_on_package,
-            "key_features": kf,
-            "scene": scene,
-            "mood": mood,
-            "portrait_prompt": portrait_prompt,
-            "summary": summary[:40],
-            "_source": "v2_fast_json_v4",
-        }
-
-    # 商品类
-    if vtype == "product" and products:
-        # 主商品（第一个position=main或第一个）
-        main = products[0]
-        for p in products:
-            if p.get("position") == "main":
-                main = p
-                break
-        name = main.get("product_name") or "未识别"
-        brand = main.get("brand") or "无法判断"
-        category = main.get("category") or "非产品图"
-        # appearance: 包装外观
-        app_parts = []
-        for k in ("package_color", "package_type", "cap_type", "body_shape", "label_design"):
-            v = main.get(k)
-            if v and v not in ("null", None):
-                app_parts.append(str(v))
-        appearance = "；".join(app_parts) if app_parts else "无法判断"
-        # packaging: 包装信息（直接用package_type+package_color）
-        pkg_parts = []
-        if main.get("package_type"):
-            pkg_parts.append(str(main["package_type"]))
-        if main.get("package_color"):
-            pkg_parts.append(str(main["package_color"]))
-        if main.get("cap_type"):
-            pkg_parts.append(f"配{main['cap_type']}")
-        packaging = "，".join(pkg_parts) if pkg_parts else "无法判断"
-        # key_features: product_features字段
-        feats = main.get("product_features") or []
-        if not isinstance(feats, list):
-            feats = [str(feats)]
-        kf = [str(f) for f in feats if f and len(str(f)) <= 40][:6]
-        # 补充卖点
-        sell = main.get("key_selling_points") or []
-        if isinstance(sell, list):
-            for s in sell[:2]:
-                if s and len(str(s)) <= 30 and str(s) not in kf:
-                    kf.append(f"卖点:{s}")
-        if text_on_package:
-            kf.append(f"文字: {'/'.join(text_on_package[:3])}")
-        kf = kf[:6] or ["无法判断"]
-        portrait_prompt = _build_product_prompt_from_v4(main, fj)
-        if brand != "无法判断" and brand not in name:
-            summary = f"{brand} {name}"
-        else:
-            summary = name
-        return {
-            "name": str(name)[:50],
-            "brand": str(brand)[:30],
-            "category": str(category)[:20],
-            "appearance": appearance[:200],
-            "packaging": packaging[:100],
-            "text_on_package": text_on_package,
-            "key_features": kf,
-            "scene": scene,
-            "mood": mood,
-            "portrait_prompt": portrait_prompt[:200],
-            "summary": str(summary)[:60],
-            "_source": "v2_fast_json_v4",
-        }
-
-    # 门店类或其他
-    if vtype == "store":
-        store_type = fj.get("store_type") or "店铺"
-        name = store_type
-        brand = fj.get("brand_signage") or "无法判断"
-        category = "门店场景"
-        visual = fj.get("visual_elements") or []
-        if isinstance(visual, str):
-            visual = [visual]
-        atmosphere = fj.get("atmosphere") or mood
-        appearance_parts = []
-        if fj.get("store_layout"):
-            appearance_parts.append(str(fj["store_layout"]))
-        if visual:
-            appearance_parts.append("、".join(str(v) for v in visual[:3]))
-        if fj.get("cleanliness"):
-            appearance_parts.append(str(fj["cleanliness"]))
-        appearance = "；".join(appearance_parts) if appearance_parts else "门店环境"
-        kf = []
-        if isinstance(visual, list):
-            kf.extend(str(v) for v in visual if v and len(str(v)) <= 30)
-        prods_vis = fj.get("product_categories_visible") or []
-        if isinstance(prods_vis, list):
-            kf.extend(str(c) for c in prods_vis[:3] if c)
-        promo = fj.get("promotion_elements") or []
-        if isinstance(promo, list) and promo:
-            kf.append("促销活动:" + "、".join(str(p) for p in promo[:2]))
-        if text_on_package:
-            kf.append(f"文字: {'/'.join(text_on_package[:3])}")
-        kf = kf[:6] or ["门店场景"]
-        portrait_prompt = f"{brand if brand!='无法判断' else ''}{store_type}，{atmosphere}，{scene}场景，{('、'.join(color_names[:3])+'配色，') if color_names else ''}产品陈列丰富，门店实拍"
-        portrait_prompt = portrait_prompt.strip("，")
-        summary = f"{store_type}场景"
-        return {
-            "name": name[:30],
-            "brand": str(brand)[:30],
-            "category": category,
-            "appearance": appearance[:200],
-            "packaging": "门店场景无包装",
-            "text_on_package": text_on_package,
-            "key_features": kf,
-            "scene": scene,
-            "mood": atmosphere or mood,
-            "portrait_prompt": portrait_prompt[:200],
-            "summary": summary[:40],
-            "_source": "v2_fast_json_v4",
-        }
-
-    # other 兜底
-    desc = fj.get("description") or "未识别"
-    return {
-        "name": desc[:30],
-        "brand": "无法判断",
-        "category": "非产品图",
-        "appearance": desc[:200],
-        "packaging": "无法判断",
-        "text_on_package": text_on_package,
-        "key_features": [desc[:30]] if desc != "未识别" else ["无法判断"],
-        "scene": scene,
-        "mood": mood,
-        "portrait_prompt": f"{scene}，{mood}氛围，{desc}"[:200],
-        "summary": desc[:40],
-        "_source": "v2_fast_json_v4_other",
-    }
-
-
-def _assemble_old(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
-    """旧扁平schema（兼容存量prompt或pro兜底输出）"""
-    portrait_prompt = _build_portrait_prompt_old(fj)
-    name = _infer_name_old(fj, ocr_texts)
-    brand = _infer_brand_old(fj, ocr_texts)
-    category = _infer_category_old(fj)
-    appearance = _build_appearance_old(fj)
-    key_features = _build_key_features_old(fj, ocr_texts)
-    scene = fj.get("scene") or "通用"
-    mood = fj.get("mood") or ""
-    packaging = "无法判断"
+    packaging = "无法判断"  # 包装细节专用API无，保留占位
     text_on_package = ocr_texts[:8]
-    if fj.get("has_person"):
-        up = fj.get("upper_wear") or "穿搭"
-        style = fj.get("style") or ""
-        summary = f"{style}{up}" if style and style not in up else up
-    elif brand != "无法判断" and name != brand:
-        summary = f"{brand} {name}"
-    else:
-        summary = name
+    summary = _build_summary(fj, name, brand, category)
+
     return {
         "name": name,
         "brand": brand,
@@ -644,3 +294,14 @@ def _assemble_old(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
         "summary": summary,
         "_source": "v2_fast_json",
     }
+
+
+def _build_summary(fj: dict, name: str, brand: str, category: str) -> str:
+    if fj.get("has_person"):
+        up = fj.get("upper_wear") or "穿搭"
+        style = fj.get("style") or ""
+        base = f"{style}{up}" if style and style not in up else up
+        return base
+    if brand != "无法判断" and name != brand:
+        return f"{brand} {name}"
+    return name

apps/worker/worker_app/tasks/vision/fast_path.py

@@ -23,10 +23,10 @@ logger = logging.getLogger(__name__)
 
 # 超时（可通过环境变量覆盖）
 _IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8"))
-_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "15"))
-_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "15"))
+_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "12"))
+_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "12"))
 _OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "6"))
-_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "30"))
+_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "25"))
 
 _FALLBACK_RESULT = {
     "name": "未识别",

apps/worker/worker_app/tasks/vision/vlm_fallback.py

@@ -8,7 +8,7 @@ fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
 - system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
 - timeout=25s
 - API Key 从环境变量 DASHSCOPE_API_KEY 读取
-- 返回 dict 统一走 assembler.assemble_result 组装，与 fast 路径输出格式完全一致
+- 返回 dict 字段与旧 _normalize() 兼容，下游零改动
 """
 
 from __future__ import annotations
@@ -19,13 +19,13 @@ import os
 import time
 from typing import Any
 
-from . import _prompt, assembler
+from . import _prompt
 
 logger = logging.getLogger(__name__)
 
 _BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
 _PRO_MODEL = "qwen3.7-plus"
-_DEFAULT_TIMEOUT = 30
+_DEFAULT_TIMEOUT = 25
 _DEFAULT_MAX_TOKENS = 800
 
 
@@ -33,6 +33,84 @@ def _api_key() -> str | None:
     return os.environ.get("DASHSCOPE_API_KEY")
 
 
+def _assemble_pp(obj: dict[str, Any]) -> str:
+    """从 JSON 字段组装 portrait_prompt（60-100字人物穿搭描述，给 Seedream t2i 用）。"""
+    if not obj.get("has_person"):
+        name = obj.get("product_name") or "商品"
+        brand = obj.get("brand") or ""
+        kf = obj.get("key_features") or []
+        scene = obj.get("scene") or ""
+        mood = obj.get("mood") or ""
+        outfit = obj.get("outfit") or ""
+        if outfit:
+            return outfit
+        pieces = []
+        if brand:
+            pieces.append(brand)
+        pieces.append(str(name))
+        if isinstance(kf, list):
+            pieces.extend(str(x) for x in kf[:2] if x)
+        if mood:
+            pieces.append(str(mood) + "氛围")
+        if scene:
+            pieces.append(str(scene) + "场景")
+        pieces.append("产品特写")
+        p = "，".join(x for x in pieces if x)
+        return p if len(p) >= 10 else "产品展示图，特写镜头"
+
+    parts: list[str] = []
+    gender = obj.get("gender") or ""
+    age = obj.get("age_range") or ""
+    subj = ""
+    if age == "儿童":
+        subj = "小女孩" if gender == "女" else ("小男孩" if gender == "男" else "儿童")
+    elif age == "青少年":
+        subj = "少女" if gender == "女" else ("少年" if gender == "男" else "青少年")
+    else:
+        prefix_map = {"青年": "年轻", "中年": "中年", "老年": "老年"}
+        gw = {"男": "男性", "女": "女性"}.get(gender, "")
+        prefix = prefix_map.get(age, "")
+        subj = (prefix + gw) if (prefix or gw) else "人物"
+    parts.append(f"一位{subj}")
+
+    outfit = obj.get("outfit") or ""
+    if outfit:
+        parts.append(f"身着{outfit}")
+
+    hair = obj.get("hair") or ""
+    if hair:
+        parts.append(str(hair))
+
+    pose = obj.get("pose") or ""
+    expr = obj.get("expression") or ""
+    det = []
+    if expr and expr not in ("自然", "平静"):
+        det.append(f"神情{expr}")
+    if pose and pose not in ("站立",):
+        det.append(str(pose))
+    if det:
+        parts.append("，".join(det))
+
+    style_parts = []
+    mood = obj.get("mood") or ""
+    scene = obj.get("scene") or ""
+    if mood:
+        style_parts.append(str(mood))
+    if scene and scene != "通用":
+        style_parts.append(str(scene))
+    if style_parts:
+        parts.append("".join(style_parts) + "风格")
+    else:
+        parts.append("人像写真")
+
+    full = "，".join(p for p in parts if p)
+    if len(full) < 40:
+        full += "，自然光线下人像特写，画面清晰"
+    if len(full) > 120:
+        full = full[:120].rstrip("，") + "。"
+    return full
+
+
 def call_pro_vlm(
     img_url: str,
     idx: int,
@@ -104,9 +182,9 @@ def call_pro_vlm(
             if lines and lines[-1].strip().startswith("```"):
                 lines = lines[:-1]
             s = "\n".join(lines).strip()
-        lpos, rr = s.find("{"), s.rfind("}")
-        if lpos >= 0 and rr > lpos:
-            s = s[lpos : rr + 1]
+        l, rr = s.find("{"), s.rfind("}")
+        if l >= 0 and rr > l:
+            s = s[l : rr + 1]
         try:
             obj = json.loads(s)
         except json.JSONDecodeError:
@@ -115,11 +193,32 @@ def call_pro_vlm(
         if not isinstance(obj, dict):
             return None
 
-        # 通过assembler统一组装，兼容v4嵌套schema和旧扁平schema
-        result = assembler.assemble_result(idx, obj, [])
-        result["_source"] = "vlm_pro"
-        result["_fallback_used"] = True
-        return result
+        pp = _assemble_pp(obj)
+        kf = obj.get("key_features")
+        if not isinstance(kf, list):
+            kf = [str(kf)] if kf else ["无法判断"]
+        else:
+            kf = [str(x) for x in kf if x] or ["无法判断"]
+
+        name = obj.get("product_name") or "未识别"
+        if obj.get("has_person") and (not name or name == "未识别"):
+            name = obj.get("outfit") or "人物穿搭"
+        brand = obj.get("brand") or "无法判断"
+        category = obj.get("category") or ("服饰" if obj.get("has_person") else "非产品图")
+        return {
+            "name": str(name),
+            "brand": str(brand),
+            "category": str(category),
+            "appearance": str(obj.get("outfit") or "无法判断"),
+            "packaging": "无法判断",
+            "text_on_package": [],
+            "key_features": kf[:6],
+            "scene": str(obj.get("scene") or "通用"),
+            "mood": str(obj.get("mood") or ""),
+            "portrait_prompt": pp,
+            "summary": str(name),
+            "_source": "vlm_pro",
+        }
     except Exception as e:
         elapsed = time.time() - t0
         logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)

apps/worker/worker_app/tasks/vision/vlm_fast_json.py

@@ -27,7 +27,7 @@ logger = logging.getLogger(__name__)
 # DashScope OpenAI 兼容 endpoint
 _BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
 _FAST_MODEL = "qwen3.8-flash"
-_DEFAULT_TIMEOUT = 15
+_DEFAULT_TIMEOUT = 12
 _DEFAULT_MAX_TOKENS = 350
 
 

tests/unit/test_vision_v4_prompt.py

@@ -1,291 +0,0 @@
-# -*- coding: utf-8 -*-
-"""vision v4 prompt / assembler 单元测试：
-
-- assembler 正确识别 v4 嵌套 schema 与旧扁平 schema
-- v4 product/person/store/other 四类输出组装出下游必出字段
-- 旧扁平 schema 行为不变
-- _prompt._resolve：DB 有 active prompt 时原样使用（不追加硬编码 schema）；
-  DB 无记录时回落到硬编码 JSON schema
-"""
-
-from __future__ import annotations
-
-import types
-from typing import Any
-
-import pytest
-from worker_app.tasks.vision import _prompt, assembler
-
-REQUIRED_KEYS = {
-    "name",
-    "brand",
-    "category",
-    "appearance",
-    "packaging",
-    "text_on_package",
-    "key_features",
-    "scene",
-    "mood",
-    "portrait_prompt",
-    "summary",
-    "_source",
-}
-
-
-# ---------- schema 识别 ----------
-
-
-def test_is_v4_schema_products_list() -> None:
-    assert assembler._is_v4_schema({"type": "product", "products": []})
-
-
-def test_is_v4_schema_type_only() -> None:
-    assert assembler._is_v4_schema({"type": "person"})
-
-
-def test_is_v4_schema_people_dict() -> None:
-    assert assembler._is_v4_schema({"people": {"has_person": True}})
-
-
-def test_is_not_v4_schema_flat() -> None:
-    assert not assembler._is_v4_schema({"has_person": True, "upper_wear": "T恤"})
-
-
-# ---------- v4 product ----------
-
-V4_PRODUCT: dict[str, Any] = {
-    "type": "product",
-    "scene": "白色背景产品图",
-    "mood": "清新专业",
-    "style": "商业产品摄影",
-    "colors": [{"hex": "#E60012", "name": "亮红色", "coverage": 0.6}],
-    "visible_text": [{"text": "OMO奥妙除菌除螨", "position": "瓶身正面"}],
-    "products": [
-        {
-            "product_name": "OMO奥妙除菌除螨洗衣液",
-            "brand": "OMO奥妙",
-            "category": "洗护",
-            "package_type": "瓶装",
-            "package_color": "亮红色瓶身",
-            "cap_type": "透明翻盖式按压瓶口",
-            "body_shape": "带侧面握持把手的竖款瓶身",
-            "label_design": "瓶身印十字盾牌图案",
-            "product_features": ["亮红色瓶装", "按压式瓶口", "十字盾牌标签"],
-            "key_selling_points": ["天然除菌除螨"],
-            "position": "main",
-        }
-    ],
-    "has_person": False,
-}
-
-
-def test_assemble_v4_product_fields() -> None:
-    r = assembler.assemble_result(0, V4_PRODUCT, ["OMO奥妙"])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["name"] == "OMO奥妙除菌除螨洗衣液"
-    assert r["brand"] == "OMO奥妙"
-    assert r["category"] == "洗护"
-    assert "瓶装" in r["packaging"]
-    assert isinstance(r["key_features"], list) and r["key_features"]
-    assert any("除菌" in str(t) for t in r["text_on_package"])
-    assert len(r["portrait_prompt"]) >= 10
-    assert r["_source"] == "v2_fast_json_v4"
-
-
-def test_assemble_v4_product_multi_selects_main() -> None:
-    fj = {
-        "type": "product",
-        "products": [
-            {"product_name": "次要商品", "brand": "B"},
-            {"product_name": "主商品", "brand": "A", "position": "main"},
-        ],
-    }
-    r = assembler.assemble_result(1, fj, [])
-    assert r["name"] == "主商品"
-
-
-# ---------- v4 person ----------
-
-V4_PERSON: dict[str, Any] = {
-    "type": "person",
-    "scene": "户外街拍",
-    "mood": "自信",
-    "style": "街拍",
-    "colors": [],
-    "visible_text": [],
-    "has_person": True,
-    "gender": "女",
-    "age_range": "青年",
-    "upper_wear": "白色V领短袖T恤",
-    "upper_color": "白色",
-    "lower_wear": "黑色高腰阔腿裤",
-    "lower_color": "黑色",
-    "dress_color": None,
-    "accessories": ["银色项链"],
-    "hairstyle": "黑色长直发",
-    "expression": "自信",
-    "pose": "侧身站立",
-    "outfit_style": "休闲日常",
-    "portrait_prompt": (
-        "一位年轻女性，身穿白色V领短袖T恤、黑色高腰阔腿裤，佩戴银色项链，"
-        "黑色长直发，神情自信，侧身站立，休闲日常风格，城市街拍场景"
-    ),
-    "products": [],
-}
-
-
-def test_assemble_v4_person() -> None:
-    r = assembler.assemble_result(0, V4_PERSON, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["category"] == "服饰"
-    assert "T恤" in r["name"]
-    assert "阔腿裤" in r["name"]
-    assert "年轻女性" in r["portrait_prompt"]
-    assert "项链" in r["portrait_prompt"]
-    assert isinstance(r["key_features"], list) and len(r["key_features"]) <= 6
-
-
-def test_assemble_v4_person_people_nested() -> None:
-    fj = {"type": "person", "people": {**V4_PERSON, "has_person": True}}
-    r = assembler.assemble_result(0, fj, [])
-    assert r["category"] == "服饰"
-    assert "年轻女性" in r["portrait_prompt"]
-
-
-# ---------- v4 store ----------
-
-
-def test_assemble_v4_store() -> None:
-    fj = {
-        "type": "store",
-        "scene": "便利店内部",
-        "mood": "日常便民",
-        "style": "门店实拍",
-        "store_type": "社区便利店",
-        "store_layout": "纵深货架布局",
-        "brand_signage": "全家FamilyMart",
-        "visual_elements": ["红白主色调", "促销海报"],
-        "product_categories_visible": ["饮料", "零食"],
-        "promotion_elements": ["第二件半价海报"],
-        "atmosphere": "亲民生活化",
-        "has_person": False,
-    }
-    r = assembler.assemble_result(0, fj, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["name"] == "社区便利店"
-    assert r["brand"] == "全家FamilyMart"
-    assert r["category"] == "门店场景"
-    assert any("饮料" in str(f) for f in r["key_features"])
-    assert "门店实拍" in r["portrait_prompt"]
-
-
-# ---------- v4 other ----------
-
-
-def test_assemble_v4_other() -> None:
-    fj = {"type": "other", "description": "海边日落风景", "scene": "海边", "mood": "宁静"}
-    r = assembler.assemble_result(0, fj, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["name"] == "海边日落风景"
-    assert r["category"] == "非产品图"
-
-
-# ---------- 旧扁平 schema 兼容 ----------
-
-
-def test_assemble_old_flat_person() -> None:
-    fj = {
-        "has_person": True,
-        "gender": "男",
-        "age_range": "中年",
-        "upper_wear": "西装",
-        "upper_color": "深灰色",
-        "lower_wear": "西裤",
-        "lower_color": "黑色",
-        "accessories": ["手表"],
-        "hairstyle": "短发",
-        "expression": "严肃",
-        "scene": "办公室",
-        "style": "商务",
-        "mood": "专业",
-    }
-    r = assembler.assemble_result(0, fj, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert "中年男性" in r["portrait_prompt"]
-    assert r["_source"] == "v2_fast_json"
-
-
-def test_assemble_old_flat_product() -> None:
-    fj = {
-        "has_person": False,
-        "product_name": "口红",
-        "brand": "Dior",
-        "category": "美妆",
-        "colors": ["红色"],
-        "scene": "通用",
-        "style": "商业",
-        "mood": "高级",
-    }
-    r = assembler.assemble_result(0, fj, ["Dior"])
-    assert r["name"] == "口红"
-    assert r["brand"] == "Dior"
-    assert r["text_on_package"] == ["Dior"]
-
-
-def test_assemble_none_input() -> None:
-    r = assembler.assemble_result(0, None, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-
-
-# ---------- _prompt 解析 ----------
-
-
-@pytest.fixture(autouse=True)
-def _clear_prompt_cache() -> Any:
-    _prompt.invalidate_cache()
-    yield
-    _prompt.invalidate_cache()
-
-
-def _fake_tpl(system_prompt: str = "v4 system prompt 只返回JSON") -> Any:
-    return types.SimpleNamespace(
-        system_prompt=system_prompt,
-        user_prompt_template="分析 {image_count} 张图",
-        version=4,
-    )
-
-
-def test_resolve_uses_db_prompt_without_append(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(_prompt, "_load_db_template", lambda: _fake_tpl("DB_V4_PROMPT_XYZ"))
-    sys_prompt, user_prompt = _prompt.resolve_fast_prompt()
-    assert sys_prompt == "DB_V4_PROMPT_XYZ"
-    assert "DB_V4_PROMPT_XYZ" not in _prompt._FAST_JSON_APPEND  # sanity: 旧append是另一段文本
-    assert "分析 1 张图" in user_prompt
-
-
-def test_resolve_pro_uses_db_prompt_without_append(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(_prompt, "_load_db_template", lambda: _fake_tpl("DB_V4_PRO_PROMPT"))
-    sys_prompt, _ = _prompt.resolve_pro_prompt()
-    assert sys_prompt == "DB_V4_PRO_PROMPT"
-    assert "【输出格式要求】" not in sys_prompt
-
-
-def test_resolve_falls_back_when_no_db(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(_prompt, "_load_db_template", lambda: None)
-    sys_prompt, user_prompt = _prompt.resolve_fast_prompt()
-    assert sys_prompt == _prompt._FAST_JSON_SCHEMA
-    assert user_prompt == _prompt.DEFAULT_FAST_USER
-
-
-def test_resolve_caches(monkeypatch: pytest.MonkeyPatch) -> None:
-    calls = {"n": 0}
-
-    def _load() -> Any:
-        calls["n"] += 1
-        return _fake_tpl("CACHED_PROMPT")
-
-    monkeypatch.setattr(_prompt, "_load_db_template", _load)
-    s1, _ = _prompt.resolve_fast_prompt()
-    s2, _ = _prompt.resolve_fast_prompt()
-    assert s1 == s2 == "CACHED_PROMPT"
-    assert calls["n"] == 1