apps/worker/worker_app/tasks/viral_video.py

@@ -850,16 +850,18 @@ def _analyze_single_image(
 
 
 def _step_image_analysis(job: ViralVideoJob) -> dict:
-    """步骤 1: 图片 VLM 分析 — 识别产品特征（v1.6/#2198 优化：lite/pro 并行竞速）。
-    #2188/#2194/#2198: (1) 所有图片 URL 先归一化（storage_key→公网URL+空值报400）
-           (2) 爆款视频强制 lite-first，不依赖 .env USE_LITE 开关
-           (3) max_tokens=1200，max_workers=min(2,n) 防方舟限流（竞速模式总并发=4）
-           (4) lite/pro 并行竞速：单张图同时发 lite(30s) 和 pro(75s)，
-               谁先返回 usable 结果就用谁。单图最坏 75s（pro慢），典型 40-50s，
-               3图2并发最坏约75s，比原串行 lite→pro 240s 改善70%+
-           (5) 整个阶段统一关闭底层 httpx 重试（外层 max_retries=0，finally 恢复），
-               子线程只读不改 client 属性避免竞态
-           (6) 每张图 VLM 调用结束打印 elapsed 耗时日志便于排查
+    """步骤 1: 图片分析。
+
+    V2（VISION_V2_ENABLED=true，10-05 新方案）:
+        - 每图并行 2 路：火山 MediaKit OCR（专用API）+ doubao-seed-2.1-lite 强约束 JSON
+          （火山云端无人体属性/商品检测/图像标签公开 HTTP API，用 lite JSON-only VLM 弥补），
+          目标单图 <3s；
+        - 外层 8 图全并发，目标 8 图 <15s；
+        - 置信度低/全失败时降级 doubao-seed-2.1-pro 完整 VLM 兜底（复用旧竞速逻辑）；
+        - 输出 dict 格式与旧 _normalize() 完全一致，下游信任链/t2i 零改动。
+
+    V1（默认，#2198/#2199 lite/pro 并行竞速）:
+        - 过渡版兜底，单图 lite(30s)/pro(75s) 竞速，外层 max_workers=2，典型 40-75s/图。
     """
     try:
         from packages.shared.ai_service import call_vision  # noqa: F401
@@ -871,17 +873,16 @@ def _step_image_analysis(job: ViralVideoJob) -> dict:
         logger.warning("[爆款视频] 任务无 images，跳过图片分析")
         return {"products": []}
 
-    # #2188 BUG1: URL 归一化 — storage_key→公网URL + 空值报400
+    # URL 归一化 — storage_key→公网URL + 空值报400
     normalized_urls: list[str] = []
     for idx, raw in enumerate(job.images):
         try:
             normalized_urls.append(_normalize_image_url(raw, idx))
         except ValueError as _ve:
-            # 空/非法URL：直接让任务失败，不默默走 fallback
             logger.error("[爆款视频] 图片 #%d URL 归一化失败: %s", idx, _ve)
-            raise  # 上层 celery 捕获后标记任务失败，避免"未识别·无法判断"误导
+            raise
 
-    # #2188/#2198 BUG2: 爆款视频强制 lite-first（不依赖 .env 开关），lite/pro 并行竞速
+    # 模型配置（V1/V2 共用）
     try:
         _s = get_shared_settings()
         lite_model = _s.doubao_vision_lite_model
@@ -889,13 +890,35 @@ def _step_image_analysis(job: ViralVideoJob) -> dict:
     except Exception:
         lite_model = "doubao-seed-2-1-lite-260915"
         pro_model = "doubao-seed-2-1-pro-260915"
+
+    # ========== V2 路径（VISION_V2_ENABLED=true）==========
+    import os as _os_v2
+    _v2_enabled = _os_v2.environ.get("VISION_V2_ENABLED", "false").lower() in ("1", "true", "yes", "on")
+    if _v2_enabled:
+        try:
+            from worker_app.tasks.vision import analyze_images_v2 as _aiv2
+        except ImportError:
+            try:
+                from tasks.vision import analyze_images_v2 as _aiv2  # type: ignore
+            except ImportError:
+                logger.warning("[vision.v2] 模块导入失败，回退 V1 路径")
+                _aiv2 = None  # type: ignore
+        if _aiv2 is not None:
+            from packages.shared.ai_client import get_doubao_client as _gdc_v2
+            _cli = _gdc_v2()
+            _orig_retries_v2 = _cli.max_retries
+            _cli.max_retries = 0
+            try:
+                results_v2 = _aiv2(normalized_urls, lite_model=lite_model, pro_model=pro_model)
+            finally:
+                _cli.max_retries = _orig_retries_v2
+            return {"products": list(results_v2)}
+        # 导入失败 → fallthrough 走 V1
+
+    # ========== V1 路径（默认，lite/pro 并行竞速）==========
     vision_model = lite_model
-    # #2198: lite 单次 30s 封顶（竞速快速路径，30s 还没出就等 pro），pro 75s（在 _analyze_single_image
-    # 的内部竞速池里设置），外层不感知。单图最坏 75s（仅 pro 成功），典型 40-50s（pro 正常返回）。
     vision_timeout = 30
 
-    # #2194/#2198: 整个并行图片分析阶段统一把共享 client 的 max_retries 置 0，
-    # 阶段结束 finally 恢复。子线程 _call 只读不改，避免竞态。
     from packages.shared.ai_client import get_doubao_client as _gdc_step
 
     _step_client = _gdc_step()
@@ -905,7 +928,7 @@ def _step_image_analysis(job: ViralVideoJob) -> dict:
     results: list[dict] = [None] * len(normalized_urls)  # type: ignore
     max_workers = min(2, max(1, len(normalized_urls)))  # 并发≤2 防方舟限流（竞速模式下总并发=4）
     logger.info(
-        "[爆款视频] 开始并行竞速图片分析 n=%d lite=%s(%ds) pro=%s(75s) img_workers=%d",
+        "[爆款视频] 开始并行竞速图片分析(V1) n=%d lite=%s(%ds) pro=%s(75s) img_workers=%d",
         len(normalized_urls),
         vision_model,
         vision_timeout,

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

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""V2 图片分析：专用 API 组合路径（OCR + lite JSON VLM 并行 + pro VLM 兜底）。
+
+灵应指令（10-05）：
+- 优先火山引擎视觉智能 API：OCR 是真实专用云端 API；
+- 人体属性/商品检测/图像标签：火山云端无公开 HTTP API（仅有移动端 SDK），
+  采用 doubao-seed-2.1-lite + 强约束 JSON-only prompt 作为"伪专用 API"，
+  目标 1-3s 返回结构化字段；
+- VLM（doubao-seed-2.1-pro）保留为终极兜底（置信度低/全失败时降级）；
+- 单图并行 2 路（OCR + lite JSON VLM），外层 8 图全并发，目标 8 图 <15s。
+
+输出 dict 格式与 viral_video._normalize() 完全一致，下游信任链/t2i 零改动。
+灰度开关：VISION_V2_ENABLED=true（默认 false，走旧 #2198/#2199 竞速逻辑）。
+"""
+from .fast_path import analyze_image_v2, analyze_images_v2  # noqa: F401

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

@@ -0,0 +1,297 @@
+# -*- coding: utf-8 -*-
+"""把 fast_json VLM 输出 + OCR 文本组装为与旧 _normalize() 完全一致的 dict。
+
+目标：下游（信任链t2i/intent_parsing/script_generation）零改动。
+必出字段：name, brand, category, appearance, packaging, text_on_package,
+         key_features, scene, mood, portrait_prompt, summary, _source
+"""
+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 = {
+    "儿童": "小女孩" if None else "儿童",
+    "青少年": "少女" if None else "少年",
+    "青年": "年轻",
+    "中年": "中年",
+    "老年": "老年",
+}
+# gender 后缀
+_GENDER_WORD = {"男": "男性", "女": "女性"}
+
+
+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 == "女":
+            return "小女孩"
+        if gender == "男":
+            return "小男孩"
+        return "儿童"
+    if age == "青少年":
+        if gender == "女":
+            return "少女"
+        if gender == "男":
+            return "少年"
+        return "青少年"
+    prefix = _AGE_PREFIX.get(age, "")
+    if gw:
+        return f"{prefix}{gw}" if prefix else gw
+    return f"{prefix}人物" if prefix else "人物"
+
+
+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 ""
+    lower_color = fj.get("lower_color") or ""
+    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
+        wear = f"{c}{upper}" if c else upper
+        if material and material not in wear:
+            wear = f"{material}{wear}"
+        if pattern and pattern not in wear and pattern != "纯色":
+            wear += f"，{pattern}图案"
+        return f"身穿{wear}"
+
+    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}" if up else "")
+    if lower:
+        lo = f"{lower_color}{lower}" if lower_color else lower
+        parts.append(f"下身{lo}" if lo else "")
+    return "，".join(p for p in parts if p)
+
+
+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 []
+        style = fj.get("style") or ""
+        scene = fj.get("scene") or ""
+        mood = fj.get("mood") or ""
+        pieces = []
+        if brand:
+            pieces.append(brand)
+        pieces.append(name)
+        if colors:
+            pieces.append("、".join(colors[:3]) + "配色")
+        if style:
+            pieces.append(style + "风格")
+        if mood:
+            pieces.append(mood + "氛围")
+        if scene and scene not in ("通用",):
+            pieces.append(scene + "场景")
+        pieces.append("产品特写")
+        prompt = "，".join(p for p in pieces if p)
+        return prompt if len(prompt) >= 10 else "产品展示图，特写镜头"
+
+    subject = _person_subject(fj)
+    wear = _build_wear_sentence(fj)
+
+    accessories = fj.get("accessories") or []
+    if isinstance(accessories, str):
+        accessories = [accessories]
+    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: 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: 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)
+    if acc_str:
+        pieces.append(acc_str.lstrip("，"))
+    if detail_parts:
+        pieces.append("，".join(detail_parts))
+    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(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(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(fj: dict[str, Any]) -> str:
+    cat = fj.get("category")
+    if cat:
+        return str(cat)
+    if fj.get("has_person"):
+        return "服饰"
+    return "非产品图"
+
+
+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:
+        if fj.get("has_person"):
+            return "人像穿搭整体造型"
+        return "无法判断"
+    return "、".join(parts)
+
+
+def _build_key_features(fj: dict[str, Any], ocr_texts: list[str]) -> list[str]:
+    feats: list[str] = []
+    for key in ("upper_wear", "lower_wear", "upper_color", "lower_color", "dress_color",
+                "material", "pattern", "style", "accessories"):
+        v = fj.get(key)
+        if not v:
+            continue
+        if isinstance(v, list):
+            feats.extend(str(x) for x in v if x)
+        elif isinstance(v, str) and v not in ("无法判断", "未知", "纯色"):
+            feats.append(v)
+    if ocr_texts:
+        feats.append(f"画面文字: {'/'.join(ocr_texts[:3])}")
+    # 去重
+    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:
+            seen.add(f)
+            out.append(f)
+    return out[:6] if out else ["无法判断"]
+
+
+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 []
+
+    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 ""
+    packaging = "无法判断"  # 包装细节专用API无，保留占位
+    text_on_package = ocr_texts[:8]
+    summary = _build_summary(fj, name, brand, category)
+
+    return {
+        "name": name,
+        "brand": brand,
+        "category": category,
+        "appearance": appearance,
+        "packaging": packaging,
+        "text_on_package": text_on_package,
+        "key_features": key_features,
+        "scene": scene,
+        "mood": mood,
+        "portrait_prompt": portrait_prompt,
+        "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

@@ -0,0 +1,242 @@
+# -*- coding: utf-8 -*-
+"""V2 快速路径：每图并行 OCR + lite JSON VLM，失败降级 pro VLM。
+
+单图并行 2 路（OCR + lite JSON VLM），目标 <3s。
+外层 8 图全并发，目标 8 图 <15s。
+终极兜底：复用旧 _analyze_single_image 完整 pro VLM 逻辑。
+"""
+from __future__ import annotations
+
+import logging
+import os
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Any
+
+from . import assembler, ocr_volc, vlm_fast_json
+
+logger = logging.getLogger(__name__)
+
+# ---------- 配置项（可通过环境变量覆盖） ----------
+VISION_V2_ENABLED = os.environ.get("VISION_V2_ENABLED", "false").lower() in ("1", "true", "yes", "on")
+# 单图 fast 路径总超时（包含 OCR + fast_json 并行）
+V2_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "10"))
+# 外层图片并发（默认 8，即全并行）
+V2_IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8"))
+# fast_json 单次超时
+V2_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "8"))
+# OCR 单次超时
+V2_OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "8"))
+# pro VLM 兜底超时（仅在 fast 路径完全失败时触发）
+V2_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45"))
+V2_LITE_TIMEOUT = float(os.environ.get("VISION_V2_LITE_TIMEOUT", "20"))
+
+
+def _is_result_usable(result: dict[str, Any]) -> bool:
+    """与 viral_video._is_vision_result_usable 对齐的可用判定。"""
+    pp = (result.get("portrait_prompt") or "").strip()
+    if pp and pp not in ("无人像", "无法判断", "未识别"):
+        return True
+    name = result.get("name") or ""
+    if name and name not in ("未识别", "无法判断", "未知"):
+        return True
+    summary = result.get("summary") or ""
+    if len(summary) >= 5 and summary not in ("无法判断", "未识别"):
+        return True
+    cat = result.get("category") or ""
+    if cat == "非产品图" and pp != "无人像":
+        return True
+    kf = result.get("key_features") or []
+    if kf and kf != ["无法判断"]:
+        # 只要有非默认特征且非空
+        return True
+    return False
+
+
+def _call_pro_fallback(img_url: str, idx: int, lite_model: str, pro_model: str) -> dict[str, Any] | None:
+    """fast 路径失败时，调用旧的 lite/pro 竞速 VLM。
+
+    复用 viral_video._analyze_single_image 的实现，避免重复代码。
+    """
+    try:
+        from worker_app.tasks.viral_video import _analyze_single_image
+    except ImportError:
+        try:
+            from tasks.viral_video import _analyze_single_image  # type: ignore
+        except ImportError:
+            logger.warning("[vision.v2] 无法 import _analyze_single_image，跳过 pro 兜底")
+            return None
+    try:
+        return _analyze_single_image(
+            idx,
+            img_url,
+            vision_model=lite_model,
+            timeout=int(V2_LITE_TIMEOUT),
+            pro_fallback_model=pro_model,
+        )
+    except Exception as e:
+        logger.warning("[vision.v2] 图片 #%d pro 兜底异常 err=%s", idx, e, exc_info=True)
+        return None
+
+
+def analyze_image_v2(
+    idx: int,
+    img_url: str,
+    *,
+    lite_model: str | None = None,
+    pro_model: str | None = None,
+) -> dict[str, Any]:
+    """单张图片 V2 分析：OCR + lite JSON VLM 并行，必要时降级 pro VLM。
+
+    返回的 dict 与 viral_video._normalize() 输出格式完全一致。
+    """
+    t0 = time.time()
+    # ---- 第 1 层：fast 路径并行 ----
+    fast_json_result: dict[str, Any] | None = None
+    ocr_result: list[str] = []
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        f_fj = pool.submit(
+            vlm_fast_json.call_fast_json,
+            img_url,
+            model=lite_model,
+            timeout=V2_FAST_JSON_TIMEOUT,
+        )
+        f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=V2_OCR_TIMEOUT)
+
+        # 等全部完成或超时
+        for fut in as_completed([f_fj, f_ocr], timeout=V2_FAST_TIMEOUT):
+            try:
+                res = fut.result(timeout=1)
+            except Exception as e:
+                logger.warning("[vision.v2] 图片 #%d fast 子任务异常: %s", idx, e)
+                continue
+            if fut is f_fj:
+                fast_json_result = res if isinstance(res, dict) else None
+            elif fut is f_ocr:
+                ocr_result = res if isinstance(res, list) else []
+
+    fast_elapsed = time.time() - t0
+
+    # ---- 组装 fast 结果 ----
+    assembled: dict[str, Any] | None = None
+    if fast_json_result:
+        assembled = assembler.assemble_result(idx, fast_json_result, ocr_result)
+        if _is_result_usable(assembled):
+            assembled["_fast_elapsed"] = round(fast_elapsed, 2)
+            logger.info(
+                "[vision.v2] 图片 #%d fast 路径命中 elapsed=%.2fs portrait_prompt=%s",
+                idx,
+                fast_elapsed,
+                (assembled.get("portrait_prompt") or "")[:40],
+            )
+            return assembled
+        logger.info(
+            "[vision.v2] 图片 #%d fast 结果不可用 portrait_prompt=%s，走 pro 兜底",
+            idx,
+            (assembled.get("portrait_prompt") or "")[:40],
+        )
+    else:
+        logger.info("[vision.v2] 图片 #%d fast_json 返回空 elapsed=%.2fs，走 pro 兜底", idx, fast_elapsed)
+
+    # ---- 第 2 层：pro VLM 兜底（复用旧竞速逻辑）----
+    pro_t0 = time.time()
+    use_lite = lite_model or vlm_fast_json.DEFAULT_LITE_MODEL
+    use_pro = pro_model or "doubao-seed-2-1-pro-260915"
+    pro_result = _call_pro_fallback(img_url, idx, use_lite, use_pro)
+    if pro_result and _is_result_usable(pro_result):
+        pro_result["_fallback_used"] = True
+        pro_result["_fast_elapsed"] = round(fast_elapsed, 2)
+        pro_result["_pro_elapsed"] = round(time.time() - pro_t0, 2)
+        logger.info(
+            "[vision.v2] 图片 #%d pro 兜底命中 total_elapsed=%.2fs",
+            idx,
+            time.time() - t0,
+        )
+        return pro_result
+
+    # ---- 第 3 层：兜底失败，返回 assembled 或标准 fallback ----
+    if assembled:
+        assembled["_source"] = "v2_fast_json_degraded"
+        logger.warning(
+            "[vision.v2] 图片 #%d pro 兜底也失败，返回降级 fast 结果 elapsed=%.2fs",
+            idx,
+            time.time() - t0,
+        )
+        return assembled
+
+    # 最后的最后：返回最小可用结构
+    logger.warning("[vision.v2] 图片 #%d 所有路径均失败 elapsed=%.2fs", idx, time.time() - t0)
+    return {
+        "name": "未识别",
+        "brand": "无法判断",
+        "category": "非产品图",
+        "appearance": "无法判断",
+        "packaging": "无法判断",
+        "text_on_package": ocr_result[:8],
+        "key_features": ["无法判断"],
+        "scene": "通用",
+        "mood": "",
+        "portrait_prompt": "无法判断",
+        "summary": "未识别",
+        "_source": "v2_all_failed",
+    }
+
+
+def analyze_images_v2(
+    img_urls: list[str],
+    *,
+    lite_model: str | None = None,
+    pro_model: str | None = None,
+    max_workers: int | None = None,
+) -> list[dict[str, Any]]:
+    """批量图片 V2 分析（外层全并行）。"""
+    if not img_urls:
+        return []
+    workers = max_workers if max_workers and max_workers > 0 else V2_IMG_WORKERS
+    workers = min(workers, len(img_urls), 16)  # 安全上限 16
+    results: list[dict[str, Any] | None] = [None] * len(img_urls)
+
+    logger.info(
+        "[vision.v2] 开始 V2 并行图片分析 n=%d workers=%d fast_timeout=%.0fs",
+        len(img_urls),
+        workers,
+        V2_FAST_TIMEOUT,
+    )
+    t0 = time.time()
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        future_to_idx = {
+            pool.submit(analyze_image_v2, idx, url, lite_model=lite_model, pro_model=pro_model): idx
+            for idx, url in enumerate(img_urls)
+        }
+        for fut in as_completed(future_to_idx):
+            idx = future_to_idx[fut]
+            try:
+                results[idx] = fut.result()
+            except Exception as e:
+                logger.warning("[vision.v2] 图片 #%d future 异常 err=%s", idx, e, exc_info=True)
+                results[idx] = {
+                    "name": "未识别",
+                    "brand": "无法判断",
+                    "category": "非产品图",
+                    "appearance": "无法判断",
+                    "packaging": "无法判断",
+                    "text_on_package": [],
+                    "key_features": ["无法判断"],
+                    "scene": "通用",
+                    "mood": "",
+                    "portrait_prompt": "无法判断",
+                    "summary": "未识别",
+                    "_source": "v2_future_exception",
+                }
+    elapsed = time.time() - t0
+    succ = sum(1 for r in results if r and _is_result_usable(r))
+    fb = sum(1 for r in results if r and r.get("_fallback_used"))
+    logger.info(
+        "[vision.v2] V2 图片分析完成 n=%d success=%d pro_fallback=%d elapsed=%.2fs",
+        len(img_urls),
+        succ,
+        fb,
+        elapsed,
+    )
+    return [r for r in results if r is not None]

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

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+"""火山引擎 AI MediaKit OCR（同步）调用封装。
+
+接口：POST {mediakit_base_url}/tools-sync/ocr
+鉴权：Bearer {mediakit_api_key}
+请求体：{"image_url": "<公网可访问URL>"}  （部分版本也支持 image_base64）
+响应：{"code":0,"data":{"texts":[{"text":"...","bbox":[x,y,w,h],...},...],...}}
+
+目标：识别商品包装/Logo/水印上的文字，作为 fast_json VLM 的补充。
+返回值：识别到的文本字符串列表（失败返回 []）。
+"""
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_TIMEOUT = 8  # OCR 秒级返回，8s 绰绰有余
+
+
+def call_ocr(img_url: str, *, timeout: int = DEFAULT_TIMEOUT) -> list[str]:
+    """调用 MediaKit 同步 OCR，返回去重后的纯文本列表。
+
+    不做重试（外层降级逻辑负责）。失败/未配置返回空列表，不抛异常。
+    """
+    t0 = time.time()
+    try:
+        import httpx
+
+        from packages.shared.mediakit_client import get_mediakit_client
+
+        client = get_mediakit_client()
+        if not client.is_available:
+            logger.info("[vision.v2] mediakit 未配置，跳过 OCR")
+            return []
+
+        url = f"{client.base_url}/tools-sync/ocr"
+        headers = {
+            "Authorization": f"Bearer {client.api_key}",
+            "Content-Type": "application/json",
+        }
+        payload: dict[str, Any] = {"image_url": img_url}
+        # 部分文档版本用 image_base64，但公网 URL 场景下 image_url 最简
+        resp = httpx.post(url, headers=headers, json=payload, timeout=timeout)
+        elapsed = time.time() - t0
+        if resp.status_code != 200:
+            logger.warning(
+                "[vision.v2] OCR HTTP %d elapsed=%.1fs body=%s",
+                resp.status_code,
+                elapsed,
+                resp.text[:200],
+            )
+            return []
+        data = resp.json()
+        # 兼容几种可能的响应结构
+        code = data.get("code", data.get("status", 0))
+        if code not in (0, "OK", "success", 200):
+            logger.warning("[vision.v2] OCR 业务错误 code=%s elapsed=%.1fs resp=%s", code, elapsed, str(data)[:200])
+            return []
+        texts = _extract_texts(data)
+        # 去重 + 过滤空
+        seen: set[str] = set()
+        out: list[str] = []
+        for t in texts:
+            t = (t or "").strip()
+            if t and t not in seen and len(t) <= 100:  # 过滤过长的误识别
+                seen.add(t)
+                out.append(t)
+        logger.info("[vision.v2] OCR 完成 elapsed=%.1fs n=%d texts=%s", elapsed, len(out), out[:5])
+        return out
+    except Exception as e:
+        elapsed = time.time() - t0
+        logger.warning("[vision.v2] OCR 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
+        return []
+
+
+def _extract_texts(data: dict) -> list[str]:
+    """从 OCR 响应中抽取文本，兼容多种结构。"""
+    out: list[str] = []
+    # 常见结构1: data.texts = [{"text": "..."}, ...]
+    d = data.get("data") or data
+    if isinstance(d, dict):
+        for key in ("texts", "lines", "words", "items", "result"):
+            items = d.get(key)
+            if isinstance(items, list):
+                for it in items:
+                    if isinstance(it, dict):
+                        txt = it.get("text") or it.get("content") or it.get("word")
+                        if txt:
+                            out.append(str(txt))
+                    elif isinstance(it, str):
+                        out.append(it)
+                break
+        # 结构2: data.text = "..."
+        if not out:
+            t = d.get("text")
+            if isinstance(t, str):
+                out.append(t)
+        # 结构3: data.ocr_text / data.content
+        if not out:
+            for key in ("ocr_text", "content", "raw_text"):
+                v = d.get(key)
+                if isinstance(v, str) and v.strip():
+                    out.append(v)
+                    break
+    return out

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

@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+"""doubao-seed-2.1-lite 强约束 JSON-only 调用。
+
+目标：替代"人体属性/商品检测/图像标签"三个火山不存在的专用云端 API。
+设计要点：
+- system prompt 极致精简，只给字段 schema 和强约束（禁止自然语言、禁止 markdown）
+- max_tokens=350（比旧 VLM 的 1200 小很多，降低延迟）
+- temperature=0.1（极低，稳定输出 JSON）
+- timeout=8s（够快，失败则由外层走 pro VLM 兜底）
+- 期望返回纯 JSON object（无 ```json 包裹、无解释文字）
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# 极简 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对象。"
+
+# 默认模型
+DEFAULT_LITE_MODEL = "doubao-seed-2-1-lite-260915"
+DEFAULT_TIMEOUT = 8
+DEFAULT_MAX_TOKENS = 350
+
+
+def _strip_code_fence(s: str) -> str:
+    """剥离 ```json ... ``` 包裹（即使要求纯 JSON，模型偶尔仍会包代码块）。"""
+    s = s.strip()
+    if s.startswith("```"):
+        lines = s.split("\n")
+        # 去掉首行 ```json
+        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 call_fast_json(
+    img_url: str,
+    *,
+    model: str | None = None,
+    timeout: int = DEFAULT_TIMEOUT,
+    max_tokens: int = DEFAULT_MAX_TOKENS,
+) -> dict[str, Any] | None:
+    """调用 lite VLM 返回结构化 dict；失败/非 JSON 返回 None。
+
+    注意：不做重试（外层竞速/降级逻辑负责），max_retries=0 由外层统一设置。
+    """
+    t0 = time.time()
+    try:
+        from packages.shared.ai_client import get_doubao_client
+
+        client = get_doubao_client()
+        if not client.is_available:
+            logger.warning("[vision.v2] doubao client 不可用，跳过 fast_json")
+            return None
+
+        use_model = model or DEFAULT_LITE_MODEL
+        raw = client.vision_completion(
+            messages=[
+                {"role": "system", "content": _FAST_SYSTEM},
+                {"role": "user", "content": _FAST_USER},
+            ],
+            images=[img_url],
+            temperature=0.1,
+            max_tokens=max_tokens,
+            timeout=timeout,
+            model=use_model,
+        )
+        elapsed = time.time() - t0
+        if raw is None:
+            logger.warning("[vision.v2] fast_json 返回 None elapsed=%.1fs model=%s", elapsed, use_model)
+            return None
+
+        text = _strip_code_fence(raw)
+        # 截到第一个 { 和最后一个 } 之间，容忍前后偶发文字
+        l = text.find("{")
+        r = text.rfind("}")
+        if l >= 0 and r > l:
+            text = text[l : r + 1]
+        try:
+            obj = json.loads(text)
+        except json.JSONDecodeError:
+            logger.warning(
+                "[vision.v2] fast_json JSON 解析失败 elapsed=%.1fs head=%s",
+                elapsed,
+                raw[:200],
+            )
+            return None
+        if not isinstance(obj, dict):
+            logger.warning("[vision.v2] fast_json 非 dict: %s", type(obj))
+            return None
+        logger.info(
+            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs has_person=%s has_product=%s category=%s",
+            use_model,
+            elapsed,
+            obj.get("has_person"),
+            obj.get("has_product"),
+            obj.get("category"),
+        )
+        return obj
+    except Exception as e:
+        elapsed = time.time() - t0
+        logger.warning("[vision.v2] fast_json 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
+        return None