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

根因：#2200重写V2时把 prompt_loader 调用链删了，system/user prompt
全部硬编码在 vlm_fast_json.py / vlm_fallback.py，用户在后台管理端
配置 viral_video_prompt_templates 表 prompt_type='image_analysis' 的
自定义提示词完全没被读取。

修复：
- 新增 vision/_prompt.py，30秒TTL热加载
- 规则简单直接（不做开头字符串匹配）：
  * DB有is_active=true的image_analysis记录（含用户修改后的版本）
    → system = DB.system_prompt + 追加JSON schema字段约束（覆盖XML等其他输出格式要求）
    → user = DB.user_prompt_template渲染后使用（image_count=1/industry=通用）
  * DB无记录/连接异常/返回空 → 纯硬编码JSON schema prompt
- 复用 prompt_loader._load_from_db 直查DB（不用get_template避免fallback到DEFAULT_TEMPLATES混淆判断）
- fast路径(qwen3.8-flash)追加22字段JSON schema，pro路径(qwen3.7-plus)追加13字段精简schema
- 末尾硬约束："不要输出任何其他文字、解释、XML标签或markdown"
- DB异常静默fallback，不阻断流程

本地冒烟：DB不可达时正确fallback到纯硬编码JSON。

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

@@ -0,0 +1,208 @@
+# -*- coding: utf-8 -*-
+"""V2 prompt 解析：优先读后台 viral_video_prompt_templates 表（prompt_type='image_analysis'
+且 is_active=true），30s TTL 热加载；DB 无有效记录/异常时，fallback 到纯硬编码 JSON schema prompt。
+
+规则（简单直接，不做字符串匹配判断）：
+- DB 有 is_active=true 的 image_analysis 记录（包括种子默认XML记录和用户自定义修改后的版本）：
+  * system = DB.system_prompt + JSON_SCHEMA_APPEND（覆盖XML输出要求，强制JSON）
+  * user   = DB.user_prompt_template 渲染后使用；渲染后为空则用硬编码默认
+- DB 无记录/连接异常/返回空：system/user 全部用纯硬编码 JSON schema prompt
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ---- 纯硬编码 JSON schema（DB 无有效配置时全量使用） ----
+
+_FAST_JSON_SCHEMA = (
+    "你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
+    "不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
+    "{\n"
+    '  "has_person": true/false,\n'
+    '  "gender": "男"/"女"/null,\n'
+    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
+    '  "upper_wear": "上装款式，如T恤/衬衫/卫衣/毛衣/西装/夹克/连衣裙/吊带/背心/外套等",\n'
+    '  "upper_color": "上装主色",\n'
+    '  "lower_wear": "下装款式；穿连衣裙时填null",\n'
+    '  "lower_color": "下装主色",\n'
+    '  "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
+    '  "accessories": ["眼镜"/"帽子"/"项链"/"耳环"/"背包"/"手表"等数组],\n'
+    '  "hairstyle": "发型，如短发/长发/马尾/卷发/丸子头/光头等",\n'
+    '  "expression": "表情，如微笑/严肃/酷/开心等",\n'
+    '  "pose": "姿势，如站立/坐姿/侧身/行走等",\n'
+    '  "scene": "场景，如室内/街拍/户外/办公室/家居/海边/雪景/森林等",\n'
+    '  "style": "风格，如休闲/商务/运动/复古/潮流/甜美/酷飒/优雅/街头/法式等",\n'
+    '  "has_product": true/false,\n'
+    '  "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
+    '  "product_name": "产品名称，非产品图填null",\n'
+    '  "brand": "品牌或文字标识，无则null",\n'
+    '  "material": "材质，如棉质/牛仔/皮革/真丝/针织/涤纶等",\n'
+    '  "pattern": "图案，如纯色/条纹/波点/格子/印花/碎花/Logo等",\n'
+    '  "colors": ["主色数组"],\n'
+    '  "mood": "整体氛围/情绪，如清新/活力/高级/温暖/冷峻/甜美/复古等"\n'
+    "}\n\n"
+    "你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释、XML标签或markdown。"
+)
+DEFAULT_FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
+
+_PRO_JSON_SCHEMA = (
+    "你是图片分析专家。严格按下方 JSON schema 返回一个对象，不要解释、不要markdown、不要代码块、不要XML标签。\n"
+    "{\n"
+    '  "has_person": true/false,\n'
+    '  "gender": "男"/"女"/null,\n'
+    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
+    '  "outfit": "整体穿着描述（含颜色款式）",\n'
+    '  "hair": "发型发色",\n'
+    '  "pose": "姿势",\n'
+    '  "expression": "表情",\n'
+    '  "scene": "场景",\n'
+    '  "mood": "氛围",\n'
+    '  "has_product": true/false,\n'
+    '  "category": "类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
+    '  "product_name": "产品名，非产品图填null",\n'
+    '  "brand": "品牌，无则null",\n'
+    '  "key_features": ["核心特征数组，3-6个短语"]\n'
+    "}\n\n"
+    "你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释、XML标签或markdown。"
+)
+DEFAULT_PRO_USER = "分析这张图片，返回符合schema的JSON。"
+
+# DB 配置存在时，追加在用户 system_prompt 末尾的JSON schema约束（强制覆盖XML等其他输出格式）
+_FAST_JSON_APPEND = (
+    "\n\n【输出格式要求】无论上文如何要求，最终你必须只返回一个合法的JSON对象，"
+    "严格包含以下字段（字段值不确定时填null或空数组）：\n"
+    "{\n"
+    '  "has_person": true/false,\n'
+    '  "gender": "男"/"女"/null,\n'
+    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
+    '  "upper_wear": "上装款式字符串",\n'
+    '  "upper_color": "上装主色",\n'
+    '  "lower_wear": "下装款式（穿连衣裙时填null）",\n'
+    '  "lower_color": "下装主色",\n'
+    '  "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
+    '  "accessories": ["配饰数组"],\n'
+    '  "hairstyle": "发型",\n'
+    '  "expression": "表情",\n'
+    '  "pose": "姿势",\n'
+    '  "scene": "场景",\n'
+    '  "style": "风格",\n'
+    '  "has_product": true/false,\n'
+    '  "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
+    '  "product_name": "产品名称，非产品图填null",\n'
+    '  "brand": "品牌或文字标识，无则null",\n'
+    '  "material": "材质",\n'
+    '  "pattern": "图案",\n'
+    '  "colors": ["主色数组"],\n'
+    '  "mood": "整体氛围"\n'
+    "}\n"
+    "不要输出任何其他文字、解释、XML标签或markdown。"
+)
+
+_PRO_JSON_APPEND = (
+    "\n\n【输出格式要求】无论上文如何要求，最终你必须只返回一个合法的JSON对象，"
+    "严格包含以下字段（字段值不确定时填null或空数组）：\n"
+    "{\n"
+    '  "has_person": true/false,\n'
+    '  "gender": "男"/"女"/null,\n'
+    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
+    '  "outfit": "整体穿着描述（含颜色款式）",\n'
+    '  "hair": "发型发色",\n'
+    '  "pose": "姿势",\n'
+    '  "expression": "表情",\n'
+    '  "scene": "场景",\n'
+    '  "mood": "氛围",\n'
+    '  "has_product": true/false,\n'
+    '  "category": "类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
+    '  "product_name": "产品名，非产品图填null",\n'
+    '  "brand": "品牌，无则null",\n'
+    '  "key_features": ["核心特征3-6个短语"]\n'
+    "}\n"
+    "不要输出任何其他文字、解释、XML标签或markdown。"
+)
+
+_cache_lock = threading.Lock()
+_cache: dict[str, tuple[float, Any]] = {}
+_CACHE_TTL = 30.0
+
+
+def _load_db_template() -> Any | None:
+    """直接查DB viral_video_prompt_templates 中 is_active=true 的 image_analysis 记录；
+    DB不可达/无记录/异常返回None。
+    复用 prompt_loader._load_from_db（它返回None表示DB无记录或异常，不会fallback到DEFAULT_TEMPLATES），
+    这样能准确区分"DB有配置（含用户修改后的版本）"和"DB无配置/异常"。"""
+    try:
+        # 直接调用 prompt_loader 内部的 _load_from_db，它只查DB不做fallback
+        from packages.application.viral_video.prompt_loader import _load_from_db
+
+        return _load_from_db("image_analysis")
+    except Exception as e:
+        logger.warning("[vision.v2] 查询DB prompt配置失败: %s", e)
+        return None
+
+
+def _render_user(tpl: Any | None, default_user: str) -> str:
+    if not tpl:
+        return default_user
+    tpl_str = getattr(tpl, "user_prompt_template", "") or ""
+    if not tpl_str.strip():
+        return default_user
+    rendered = tpl_str.replace("{image_count}", "1").replace("{industry}", "通用").replace("{image_urls}", "").strip()
+    return rendered or default_user
+
+
+def resolve_fast_prompt() -> tuple[str, str]:
+    return _resolve("fast")
+
+
+def resolve_pro_prompt() -> tuple[str, str]:
+    return _resolve("pro")
+
+
+def _resolve(kind: str) -> tuple[str, str]:
+    now = time.time()
+    cache_key = f"prompt_{kind}"
+    with _cache_lock:
+        hit = _cache.get(cache_key)
+        if hit and now - hit[0] < _CACHE_TTL:
+            return hit[1]
+
+    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
+    try:
+        tpl = _load_db_template()
+        if tpl is not None:
+            db_sys = (getattr(tpl, "system_prompt", "") or "").strip()
+            if db_sys:
+                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)",
+                    kind,
+                    getattr(tpl, "version", "?"),
+                    len(db_sys),
+                )
+            else:
+                logger.debug("[vision.v2] DB image_analysis system_prompt为空，使用默认JSON (kind=%s)", kind)
+        else:
+            logger.debug("[vision.v2] DB无image_analysis记录/不可达，使用默认JSON prompt (kind=%s)", kind)
+    except Exception as e:
+        logger.warning("[vision.v2] 解析DB prompt异常，使用默认: %s", e)
+
+    with _cache_lock:
+        _cache[cache_key] = (now, (sys_prompt, usr_prompt))
+    return sys_prompt, usr_prompt
+
+
+def invalidate_cache() -> None:
+    with _cache_lock:
+        _cache.clear()

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

@@ -1,8 +1,14 @@
 # -*- coding: utf-8 -*-
-"""V2 pro 兜底：qwen3.7-plus（阿里云百炼/DashScope）单次调用。
+"""V2 兜底路径：qwen3.7-plus（阿里云百炼/DashScope）单图调用。
 
-fast_json 结果不可用时单次调用，无竞速、无重试、无复杂超时逻辑。
-直接 httpx 发精简 JSON-only prompt（比旧版 prompt_loader XML 模板短很多，降低延迟）。
+fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
+设计要点：
+- 直接 httpx 直连 DashScope，不走 ai_client
+- enable_thinking=false + response_format=json_object
+- system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
+- timeout=25s
+- API Key 从环境变量 DASHSCOPE_API_KEY 读取
+- 返回 dict 字段与旧 _normalize() 兼容，下游零改动
 """
 
 from __future__ import annotations
@@ -13,6 +19,8 @@ import os
 import time
 from typing import Any
 
+from . import _prompt
+
 logger = logging.getLogger(__name__)
 
 _BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
@@ -20,65 +28,87 @@ _PRO_MODEL = "qwen3.7-plus"
 _DEFAULT_TIMEOUT = 25
 _DEFAULT_MAX_TOKENS = 800
 
-_PRO_SYSTEM = (
-    "你是图片分析助手。仔细观察图片，严格按JSON schema返回一个对象，不要任何解释、"
-    "不要markdown、不要代码块、不要前后缀文字。字段值不确定时填null或空数组。\n"
-    "{\n"
-    '  "has_person": true/false,\n'
-    '  "gender": "男"/"女"/null,\n'
-    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
-    '  "outfit": "人物穿搭描述，60字以内（例：白色T恤+牛仔裤）",\n'
-    '  "hair": "发型",\n'
-    '  "pose": "姿态",\n'
-    '  "expression": "表情",\n'
-    '  "scene": "场景",\n'
-    '  "mood": "氛围",\n'
-    '  "has_product": true/false,\n'
-    '  "category": "服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
-    '  "product_name": "产品名称，非产品图填null",\n'
-    '  "brand": "品牌或文字标识，无则null",\n'
-    '  "key_features": ["特征数组"]\n'
-    "}"
-)
-_PRO_USER = "分析这张图片，返回符合schema的JSON。"
 
-
-def _strip_code_fence(s: str) -> str:
-    s = s.strip()
-    if s.startswith("```"):
-        lines = s.split("\n")
-        if lines and lines[0].startswith("```"):
-            lines = lines[1:]
-        if lines and lines[-1].strip().startswith("```"):
-            lines = lines[:-1]
-        s = "\n".join(lines).strip()
-    return s
+def _api_key() -> str | None:
+    return os.environ.get("DASHSCOPE_API_KEY")
 
 
 def _assemble_pp(obj: dict[str, Any]) -> str:
-    if not obj.get("has_person", False):
-        return "无人像"
+    """从 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")
-    age = obj.get("age_range")
-    if gender:
-        parts.append(gender + ("性" if not gender.endswith("性") else ""))
-    if age:
-        parts.append(age)
-    parts.append("人物")
-    hair = obj.get("hair")
-    if hair:
-        parts.append(hair)
-    outfit = obj.get("outfit")
+    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}")
-    pose = obj.get("pose")
-    if pose:
-        parts.append(f"姿态{pose}")
-    expr = obj.get("expression")
-    if expr:
-        parts.append(f"表情{expr}")
-    return "，".join(parts) if parts else "无人像"
+
+    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(
@@ -87,25 +117,25 @@ def call_pro_vlm(
     *,
     timeout: int = _DEFAULT_TIMEOUT,
 ) -> dict[str, Any] | None:
-    """单次调用 qwen3.7-plus，解析后返回 product dict；失败返回 None。"""
     t0 = time.time()
     import httpx
 
-    api_key = os.environ.get("DASHSCOPE_API_KEY")
+    api_key = _api_key()
     if not api_key:
-        logger.warning("[vision.v2] DASHSCOPE_API_KEY 未配置，跳过 pro 兜底")
+        logger.warning("[vision.v2] pro DASHSCOPE_API_KEY 未配置，跳过")
         return None
 
-    url = f"{_BASE_URL}/chat/completions"
+    system_prompt, user_prompt = _prompt.resolve_pro_prompt()
+
     payload: dict[str, Any] = {
         "model": _PRO_MODEL,
         "messages": [
-            {"role": "system", "content": _PRO_SYSTEM},
+            {"role": "system", "content": system_prompt},
             {
                 "role": "user",
                 "content": [
                     {"type": "image_url", "image_url": {"url": img_url}},
-                    {"type": "text", "text": _PRO_USER},
+                    {"type": "text", "text": user_prompt},
                 ],
             },
         ],
@@ -117,7 +147,7 @@ def call_pro_vlm(
     }
     try:
         r = httpx.post(
-            url,
+            f"{_BASE_URL}/chat/completions",
             headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
             json=payload,
             timeout=timeout,
@@ -132,44 +162,64 @@ def call_pro_vlm(
             logger.warning("[vision.v2] pro 返回空 elapsed=%.1fs", elapsed)
             return None
         usage = data.get("usage") or {}
+        reasoning_tokens = usage.get("reasoning_tokens", 0)
+        ctd = usage.get("completion_tokens_details") or {}
+        if not reasoning_tokens:
+            reasoning_tokens = ctd.get("reasoning_tokens", 0)
         logger.info(
-            "[vision.v2] pro 完成 idx=%d model=%s elapsed=%.1fs in=%d out=%d",
-            idx,
+            "[vision.v2] pro 完成 model=%s elapsed=%.1fs in=%d out=%d reasoning=%d",
             _PRO_MODEL,
             elapsed,
             usage.get("prompt_tokens", 0),
             usage.get("completion_tokens", 0),
+            reasoning_tokens,
         )
-        text = _strip_code_fence(raw)
-        lpos, r_pos = text.find("{"), text.rfind("}")
-        if lpos < 0 or r_pos <= lpos:
-            logger.warning("[vision.v2] pro 无JSON elapsed=%.1fs head=%s", elapsed, raw[:200])
+        s = raw.strip()
+        if s.startswith("```"):
+            lines = s.split("\n")
+            if lines and lines[0].startswith("```"):
+                lines = lines[1:]
+            if lines and lines[-1].strip().startswith("```"):
+                lines = lines[:-1]
+            s = "\n".join(lines).strip()
+        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:
+            logger.warning("[vision.v2] pro JSON 解析失败 head=%s", raw[:200])
             return None
-        obj = json.loads(text[lpos : r_pos + 1])
         if not isinstance(obj, dict):
             return None
-        scene = obj.get("scene") or "通用"
-        mood = obj.get("mood") or ""
+
         pp = _assemble_pp(obj)
-        has_person = obj.get("has_person", False)
-        has_product = obj.get("has_product", False)
+        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 has_person and not has_product else "无法判断")
+        category = obj.get("category") or ("服饰" if obj.get("has_person") else "非产品图")
         return {
-            "name": name,
-            "brand": brand,
-            "category": category,
-            "appearance": obj.get("outfit") or "无法判断",
+            "name": str(name),
+            "brand": str(brand),
+            "category": str(category),
+            "appearance": str(obj.get("outfit") or "无法判断"),
             "packaging": "无法判断",
             "text_on_package": [],
-            "key_features": obj.get("key_features") or ["无法判断"],
-            "scene": scene,
-            "mood": mood,
+            "key_features": kf[:6],
+            "scene": str(obj.get("scene") or "通用"),
+            "mood": str(obj.get("mood") or ""),
             "portrait_prompt": pp,
-            "summary": f"{brand} {name}" if name != "未识别" else "未识别",
+            "summary": str(name),
             "_source": "vlm_pro",
         }
     except Exception as e:
-        logger.warning("[vision.v2] pro 异常 idx=%d elapsed=%.1fs err=%s", idx, time.time() - t0, e, exc_info=True)
+        elapsed = time.time() - t0
+        logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
         return None

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

@@ -5,7 +5,8 @@
 设计要点：
 - 直接用 httpx 发最小 payload 到 DashScope OpenAI 兼容 endpoint，不走 ai_client 包装
 - enable_thinking=false 关闭推理链（reasoning 是延迟主因）
-- system prompt 极致精简，只给字段 schema 和强约束（禁止自然语言、禁止 markdown）
+- response_format=json_object 强约束JSON输出
+- system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
 - max_tokens=350、temperature=0.1（稳定输出 JSON）
 - timeout=12s（失败由外层走 pro 兜底）
 - API Key 从环境变量 DASHSCOPE_API_KEY 读取
@@ -19,6 +20,8 @@ import os
 import time
 from typing import Any
 
+from . import _prompt
+
 logger = logging.getLogger(__name__)
 
 # DashScope OpenAI 兼容 endpoint
@@ -27,38 +30,6 @@ _FAST_MODEL = "qwen3.8-flash"
 _DEFAULT_TIMEOUT = 12
 _DEFAULT_MAX_TOKENS = 350
 
-# 极简 system prompt：只给字段定义 + 硬性输出要求
-_FAST_SYSTEM = (
-    "你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
-    "不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
-    "{\n"
-    '  "has_person": true/false,\n'
-    '  "gender": "男"/"女"/null,\n'
-    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
-    '  "upper_wear": "上装款式，如T恤/衬衫/卫衣/毛衣/西装/夹克/连衣裙/吊带/背心/外套等",\n'
-    '  "upper_color": "上装主色",\n'
-    '  "lower_wear": "下装款式；穿连衣裙时填null",\n'
-    '  "lower_color": "下装主色",\n'
-    '  "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
-    '  "accessories": ["眼镜"/"帽子"/"项链"/"耳环"/"背包"/"手表"等数组],\n'
-    '  "hairstyle": "发型，如短发/长发/马尾/卷发/丸子头/光头等",\n'
-    '  "expression": "表情，如微笑/严肃/酷/开心等",\n'
-    '  "pose": "姿势，如站立/坐姿/侧身/行走等",\n'
-    '  "scene": "场景，如室内/街拍/户外/办公室/家居/海边/雪景/森林等",\n'
-    '  "style": "风格，如休闲/商务/运动/复古/潮流/甜美/酷飒/优雅/街头/法式等",\n'
-    '  "has_product": true/false,\n'
-    '  "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
-    '  "product_name": "产品名称，非产品图填null",\n'
-    '  "brand": "品牌或文字标识，无则null",\n'
-    '  "material": "材质，如棉质/牛仔/皮革/真丝/针织/涤纶等",\n'
-    '  "pattern": "图案，如纯色/条纹/波点/格子/印花/碎花/Logo等",\n'
-    '  "colors": ["主色数组"],\n'
-    '  "mood": "整体氛围/情绪，如清新/活力/高级/温暖/冷峻/甜美/复古等"\n'
-    "}"
-)
-
-_FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
-
 
 def _api_key() -> str | None:
     return os.environ.get("DASHSCOPE_API_KEY")
@@ -91,16 +62,18 @@ def call_fast_json(
         logger.warning("[vision.v2] DASHSCOPE_API_KEY 未配置，跳过 fast_json")
         return None
 
+    system_prompt, user_prompt = _prompt.resolve_fast_prompt()
+
     url = f"{_BASE_URL}/chat/completions"
     payload: dict[str, Any] = {
         "model": _FAST_MODEL,
         "messages": [
-            {"role": "system", "content": _FAST_SYSTEM},
+            {"role": "system", "content": system_prompt},
             {
                 "role": "user",
                 "content": [
                     {"type": "image_url", "image_url": {"url": img_url}},
-                    {"type": "text", "text": _FAST_USER},
+                    {"type": "text", "text": user_prompt},
                 ],
             },
         ],
@@ -119,7 +92,6 @@ def call_fast_json(
         )
         elapsed = time.time() - t0
         if resp.status_code == 400 and "enable_thinking" in resp.text[:300].lower():
-            # 极少数 endpoint 版本不识别 enable_thinking，重试一次不带
             logger.warning("[vision.v2] fast_json HTTP 400 thinking 参数不兼容，重试 elapsed=%.1fs", elapsed)
             payload.pop("enable_thinking", None)
             resp = httpx.post(