feat(vision): 新增DashScope(阿里云百炼/qwen) provider支持

- 新增_provider.py统一后端切换：VISION_V2_PROVIDER=ark|dashscope（默认dashscope对比测试）
- vlm_fast_json.py：双后端httpx直连，ark用thinking={type:disabled}、dashscope用enable_thinking=false，400降级逻辑按provider处理
- vlm_fallback.py：dashscope用qwen3.7-plus+精简JSON prompt（<20s），ark保留原有ai_client+XML/JSON双解析
- fast_path.py：超时默认值从provider读取（dashscope pro 20s、ark pro 45s）
- 通过环境变量DASHSCOPE_API_KEY配置key（敏感信息不入代码）
- 本地curl验证：qwen3.8-flash关thinking单图2.6s、3图并发7.3s、8图并发10.3s（7/8可用）

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

@@ -0,0 +1,112 @@
+# -*- coding: utf-8 -*-
+"""V2 VLM 多后端 provider 切换：ark（火山方舟/豆包，默认）/ dashscope（阿里云百炼/qwen）。
+
+通过环境变量 VISION_V2_PROVIDER 切换，默认 dashscope（对比测试用）。
+各后端的 fast（小模型 JSON-only）和 pro（大模型兜底）模型配置如下：
+- ark:     doubao-seed-2-1-lite-260915 / doubao-seed-2-1-pro-260915
+- dashscope: qwen3.8-flash / qwen3.7-plus
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# 默认 dashscope 做对比测试（灵应10-05指示）
+PROVIDER = os.environ.get("VISION_V2_PROVIDER", "dashscope").lower()
+
+# fast 模型（JSON-only，强约束输出）
+_FAST_MODEL_MAP = {
+    "ark": "doubao-seed-2-1-lite-260915",
+    "dashscope": "qwen3.8-flash",
+}
+# pro 模型（兜底）
+_PRO_MODEL_MAP = {
+    "ark": "doubao-seed-2-1-pro-260915",
+    "dashscope": "qwen3.7-plus",
+}
+# base URL
+_BASE_URL_MAP = {
+    "ark": "https://ark.cn-beijing.volces.com/api/v3",
+    "dashscope": "https://dashscope.aliyuncs.com/compatible-mode/v1",
+}
+# API key 环境变量
+_API_KEY_ENVS = {
+    "ark": "DOUBAO_API_KEY",
+    "dashscope": "DASHSCOPE_API_KEY",
+}
+# 关 thinking 参数（各后端不一样）
+_THINKING_PARAMS = {
+    # 豆包 Seed 2.x 用 thinking={type:"disabled"}，不要和 reasoning_effort 同时传
+    "ark": {"thinking": {"type": "disabled"}},
+    # 百炼 qwen3 用 enable_thinking:false（顶层字段，不是 OpenAI 标准）
+    "dashscope": {"enable_thinking": False},
+}
+# fast/pro 超时
+_FAST_TIMEOUT_MAP = {"ark": 8, "dashscope": 8}
+_PRO_TIMEOUT_MAP = {"ark": 45, "dashscope": 20}
+
+
+def get_provider() -> str:
+    return PROVIDER
+
+
+def fast_model() -> str:
+    return _FAST_MODEL_MAP.get(PROVIDER, _FAST_MODEL_MAP["ark"])
+
+
+def pro_model() -> str:
+    return _PRO_MODEL_MAP.get(PROVIDER, _PRO_MODEL_MAP["ark"])
+
+
+def base_url() -> str:
+    return _BASE_URL_MAP.get(PROVIDER, _BASE_URL_MAP["ark"])
+
+
+def api_key() -> str | None:
+    """返回当前 provider 的 API key；ark 从 shared_settings 读，dashscope 从环境变量读。"""
+    if PROVIDER == "ark":
+        try:
+            from packages.shared import get_shared_settings
+            return get_shared_settings().doubao_api_key
+        except Exception:
+            return os.environ.get("DOUBAO_API_KEY")
+    return os.environ.get(_API_KEY_ENVS.get(PROVIDER, "DASHSCOPE_API_KEY"))
+
+
+def thinking_params() -> dict[str, Any]:
+    return dict(_THINKING_PARAMS.get(PROVIDER, {}))
+
+
+def is_dashscope() -> bool:
+    return PROVIDER == "dashscope"
+
+
+def fast_timeout_default() -> int:
+    return _FAST_TIMEOUT_MAP.get(PROVIDER, 8)
+
+
+def pro_timeout_default() -> int:
+    return _PRO_TIMEOUT_MAP.get(PROVIDER, 45)
+
+
+def is_400_thinking_error(body_text: str) -> bool:
+    """400 响应是否是因为 thinking 参数不被支持（触发降级重试）。"""
+    b = body_text.lower()
+    if PROVIDER == "ark":
+        return "thinking" in b or "reasoning" in b
+    if PROVIDER == "dashscope":
+        return "enable_thinking" in b or "thinking" in b
+    return False
+
+
+def pop_thinking_param(payload: dict[str, Any]) -> None:
+    """从 payload 里移除 thinking 相关参数（降级重试用）。"""
+    if PROVIDER == "ark":
+        payload.pop("thinking", None)
+        payload.pop("reasoning_effort", None)
+    elif PROVIDER == "dashscope":
+        payload.pop("enable_thinking", None)
+        payload.pop("thinking", None)

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

@@ -1,7 +1,8 @@
 # -*- coding: utf-8 -*-
-"""V2 图片分析主路径：每图并行 OCR（火山专用API）+ lite JSON VLM，失败时单次 pro VLM 兜底。
+"""V2 图片分析主路径：每图并行 OCR（火山专用API，未配置时自动跳过）+ lite JSON VLM，失败时单次 pro VLM 兜底。
 
-设计原则（灵应10-05要求）：
+支持双后端（环境变量 VISION_V2_PROVIDER=ark|dashscope，默认 dashscope）。
+设计原则：
 - 主力路径简洁：单图2路并行，外层N图全并发
 - 兜底简单：单次 pro VLM 调用，无竞速/重试/复杂超时
 - 输出 dict 格式与旧版完全一致，下游零改动
@@ -15,16 +16,16 @@ import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any
 
-from . import assembler, ocr_volc, vlm_fallback, vlm_fast_json
+from . import _provider, assembler, ocr_volc, vlm_fallback, vlm_fast_json
 
 logger = logging.getLogger(__name__)
 
-# 可通过环境变量调参（有默认值，无需配置即可跑）
+# 可通过环境变量调参（默认值随 provider 变化）
 _IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8"))
 _FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "8"))
-_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "8"))
+_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", str(_provider.fast_timeout_default())))
 _OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "6"))
-_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45"))
+_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", str(_provider.pro_timeout_default())))
 
 _FALLBACK_RESULT = {
     "name": "未识别",
@@ -42,7 +43,6 @@ _FALLBACK_RESULT = {
 
 
 def _is_usable(r: dict[str, Any]) -> bool:
-    """结果可用判定：portrait_prompt 是核心，有效就算 usable。"""
     pp = (r.get("portrait_prompt") or "").strip()
     if pp and pp not in ("无人像", "无法判断", "未识别"):
         return True
@@ -53,10 +53,8 @@ def _is_usable(r: dict[str, Any]) -> bool:
 
 
 def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
-    """单张图片 V2 分析。"""
     t0 = time.time()
 
-    # 第1层：OCR + lite JSON VLM 并行
     fj_result: dict[str, Any] | None = None
     ocr_result: list[str] = []
     with ThreadPoolExecutor(max_workers=2) as pool:
@@ -74,7 +72,6 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
                 elif fut is f_ocr and isinstance(res, list):
                     ocr_result = res
         except TimeoutError:
-            # fast 整体超时，取消还没跑完的子任务，继续走 pro 兜底
             for f in (f_fj, f_ocr):
                 if not f.done():
                     f.cancel()
@@ -82,20 +79,17 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
 
     fast_elapsed = time.time() - t0
 
-    # 组装 fast 结果
     if fj_result:
         assembled = assembler.assemble_result(idx, fj_result, ocr_result)
         if _is_usable(assembled):
             assembled["_fast_elapsed"] = round(fast_elapsed, 2)
             logger.info(
-                "[vision.v2] 图片 #%d fast命中 elapsed=%.2fs pp=%s",
-                idx,
-                fast_elapsed,
+                "[vision.v2] 图片 #%d fast命中 provider=%s elapsed=%.2fs pp=%s",
+                idx, _provider.get_provider(), fast_elapsed,
                 (assembled.get("portrait_prompt") or "")[:40],
             )
             return assembled
 
-    # 第2层：pro VLM 单次兜底
     pro_t0 = time.time()
     pro_result = vlm_fallback.call_pro_vlm(img_url, idx, timeout=_PRO_TIMEOUT)
     if pro_result and _is_usable(pro_result):
@@ -107,8 +101,7 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
         logger.info("[vision.v2] 图片 #%d pro兜底命中 total=%.2fs", idx, time.time() - t0)
         return pro_result
 
-    # 最终：返回最小可用结果
-    logger.warning("[vision.v2] 图片 #%d 全路径失败 elapsed=%.2fs", idx, time.time() - t0)
+    logger.warning("[vision.v2] 图片 #%d 全路径失败 provider=%s elapsed=%.2fs", idx, _provider.get_provider(), time.time() - t0)
     out = dict(_FALLBACK_RESULT)
     out["_source"] = "v2_all_failed"
     out["text_on_package"] = ocr_result[:8]
@@ -117,13 +110,15 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
 
 
 def analyze_images_v2(img_urls: list[str]) -> list[dict[str, Any]]:
-    """批量图片 V2 分析，外层全并发。"""
     if not img_urls:
         return []
     workers = min(_IMG_WORKERS, len(img_urls), 16)
     results: list[dict[str, Any] | None] = [None] * len(img_urls)
 
-    logger.info("[vision.v2] 开始图片分析 n=%d workers=%d fast_timeout=%.0fs", len(img_urls), workers, _FAST_TIMEOUT)
+    logger.info(
+        "[vision.v2] 开始图片分析 n=%d workers=%d provider=%s fast_timeout=%.0fs pro_timeout=%.0fs",
+        len(img_urls), workers, _provider.get_provider(), _FAST_TIMEOUT, _PRO_TIMEOUT,
+    )
     t0 = time.time()
     with ThreadPoolExecutor(max_workers=workers) as pool:
         future_to_idx = {pool.submit(analyze_image_v2, idx, url): idx for idx, url in enumerate(img_urls)}

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

@@ -1,7 +1,10 @@
 # -*- coding: utf-8 -*-
-"""VLM 兜底：专用API路径失败时的最后一道防线，单次调用 doubao-seed-2.1-pro。
+"""VLM pro 兜底：fast 路径失败时的单次调用，支持双后端。
 
-设计原则：简单、直接、无竞速、无复杂超时逻辑。只在 fast_json 结果不可用时调用。
+- ark（火山/豆包）：通过 ai_client 走 doubao-seed-2-1-pro，保留 XML+JSON 双解析和 prompt_loader
+- dashscope（百炼/qwen）：直接 httpx 走 qwen3.7-plus，用精简 JSON-only prompt 提升速度
+
+设计原则：简单、直接、无竞速、无复杂重试。
 """
 
 from __future__ import annotations
@@ -12,12 +15,36 @@ import re
 import time
 from typing import Any
 
+from . import _provider
+
 logger = logging.getLogger(__name__)
 
-DEFAULT_PRO_MODEL = "doubao-seed-2-1-pro-260915"
-DEFAULT_TIMEOUT = 45
+DEFAULT_TIMEOUT = _provider.pro_timeout_default()
 DEFAULT_MAX_TOKENS = 800
 
+# DashScope 兜底用的精简 JSON-only prompt（比 prompt_loader 模板短很多，减少延迟）
+_DS_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": "产品名称",\n'
+    '  "brand": "品牌",\n'
+    '  "key_features": ["特征数组"]\n'
+    "}"
+)
+_DS_PRO_USER = "分析这张图片，返回符合schema的JSON。"
+
 
 def _strip_code_fence(s: str) -> str:
     s = s.strip()
@@ -42,7 +69,7 @@ def _xml_attr(tag: str, attr: str, xml: str) -> str:
 
 
 def _xml_to_product(raw: str, idx: int) -> dict[str, Any]:
-    """解析 VLM 输出的 XML 格式（简化版）。"""
+    """解析 ARK pro 返回的 XML 格式。"""
     scene = _xml_text("scene", raw) or "通用"
     mood = _xml_text("mood", raw) or ""
 
@@ -71,86 +98,151 @@ def _xml_to_product(raw: str, idx: int) -> dict[str, Any]:
 
     m = re.search(r"<product[^>]*>(.*?)</product>", raw, re.S)
     if m:
-        pbody = m.group(1)
-        name = _xml_attr("product", "name", raw) or _xml_text("name", pbody) or "未识别"
-        brand = _xml_attr("product", "brand", raw) or _xml_text("brand", pbody) or "无法判断"
-        category = _xml_attr("product", "category", raw) or _xml_text("category", pbody) or "无法判断"
-        appearance = _xml_attr("product", "appearance", raw) or _xml_text("appearance", pbody) or "无法判断"
-        packaging = _xml_attr("product", "packaging", raw) or _xml_text("packaging", pbody) or "无法判断"
-        feat = _xml_attr("product", "features", raw) or _xml_text("features", pbody) or ""
+        name = _xml_attr("product", "name", raw) or "未识别"
+        brand = _xml_attr("product", "brand", raw) or "无法判断"
+        category = _xml_attr("product", "category", raw) or "无法判断"
+        appearance = _xml_attr("product", "appearance", raw) or "无法判断"
+        packaging = _xml_attr("product", "packaging", raw) or "无法判断"
+        feat = _xml_attr("product", "features", raw) or ""
         feat_list = [x.strip() for x in re.split(r"[,，;；]", feat) if x.strip()] if feat else ["无法判断"]
-        top_text = _xml_attr("product", "text_on_package", raw) or _xml_text("text_on_package", pbody) or ""
+        top_text = _xml_attr("product", "text_on_package", raw) or ""
         text_list = [x.strip() for x in re.split(r"[,，;；]", top_text) if x.strip()] if top_text else []
-        summary = _xml_attr("product", "summary", raw) or _xml_text("summary", pbody) or f"{brand} {name}"
-        pp_attr = _xml_attr("product", "portrait_prompt", raw)
-        if pp_attr and pp_attr != "无人像":
-            portrait_prompt = pp_attr
+        summary = _xml_attr("product", "summary", raw) or f"{brand} {name}"
         return {
-            "name": name,
-            "brand": brand,
-            "category": category,
-            "appearance": appearance,
-            "packaging": packaging,
-            "text_on_package": text_list,
-            "key_features": feat_list,
-            "scene": scene,
-            "mood": mood,
-            "portrait_prompt": portrait_prompt,
-            "summary": summary,
-            "_source": "vlm_pro_xml",
+            "name": name, "brand": brand, "category": category,
+            "appearance": appearance, "packaging": packaging,
+            "text_on_package": text_list, "key_features": feat_list,
+            "scene": scene, "mood": mood, "portrait_prompt": portrait_prompt,
+            "summary": summary, "_source": "vlm_pro_xml",
         }
-
     if portrait_prompt != "无人像":
         return {
-            "name": "未识别",
-            "brand": "无法判断",
-            "category": "无法判断",
-            "appearance": "无法判断",
-            "packaging": "无法判断",
-            "text_on_package": [],
-            "key_features": ["无法判断"],
-            "scene": scene,
-            "mood": mood,
-            "portrait_prompt": portrait_prompt,
-            "summary": "未识别",
-            "_source": "vlm_pro_no_product",
+            "name": "未识别", "brand": "无法判断", "category": "无法判断",
+            "appearance": "无法判断", "packaging": "无法判断",
+            "text_on_package": [], "key_features": ["无法判断"],
+            "scene": scene, "mood": mood, "portrait_prompt": portrait_prompt,
+            "summary": "未识别", "_source": "vlm_pro_no_product",
         }
     return {
-        "name": "未识别",
-        "brand": "无法判断",
-        "category": "无法判断",
-        "appearance": "无法判断",
-        "packaging": "无法判断",
-        "text_on_package": [],
-        "key_features": ["无法判断"],
-        "scene": scene,
-        "mood": mood,
-        "portrait_prompt": "无人像",
-        "summary": "未识别",
-        "_source": "vlm_pro_no_tag",
+        "name": "未识别", "brand": "无法判断", "category": "无法判断",
+        "appearance": "无法判断", "packaging": "无法判断",
+        "text_on_package": [], "key_features": ["无法判断"],
+        "scene": scene, "mood": mood, "portrait_prompt": "无人像",
+        "summary": "未识别", "_source": "vlm_pro_no_tag",
     }
 
 
-def call_pro_vlm(
-    img_url: str,
-    idx: int,
-    *,
-    model: str | None = None,
-    timeout: int = DEFAULT_TIMEOUT,
-) -> dict[str, Any] | None:
-    """单次调用 pro VLM，解析后返回 product dict；失败返回 None。"""
+def _assemble_pp_from_json(obj: dict[str, Any]) -> str:
+    """从 DashScope pro 返回的 JSON 组装 portrait_prompt。"""
+    if not obj.get("has_person", False):
+        return "无人像"
+    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")
+    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 "无人像"
+
+
+def _call_dashscope_pro(img_url: str, idx: int, timeout: int) -> dict[str, Any] | None:
+    """DashScope qwen3.7-plus 兜底，直接 httpx 发精简 JSON prompt。"""
+    import httpx
+    t0 = time.time()
+    api_key = _provider.api_key()
+    if not api_key:
+        logger.warning("[vision.vlm] dashscope api_key 未配置")
+        return None
+    url = f"{_provider.base_url().rstrip('/')}/chat/completions"
+    payload: dict[str, Any] = {
+        "model": _provider.pro_model(),
+        "messages": [
+            {"role": "system", "content": _DS_PRO_SYSTEM},
+            {"role": "user", "content": [
+                {"type": "image_url", "image_url": {"url": img_url}},
+                {"type": "text", "text": _DS_PRO_USER},
+            ]},
+        ],
+        "temperature": 0.3,
+        "max_tokens": DEFAULT_MAX_TOKENS,
+        "stream": False,
+    }
+    payload.update(_provider.thinking_params())
+    try:
+        r = httpx.post(
+            url,
+            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+            json=payload,
+            timeout=timeout,
+        )
+        elapsed = time.time() - t0
+        if r.status_code != 200:
+            logger.warning("[vision.vlm] dashscope pro HTTP %d elapsed=%.1fs body=%s", r.status_code, elapsed, r.text[:200])
+            return None
+        data = r.json()
+        raw = (data.get("choices") or [{}])[0].get("message", {}).get("content")
+        if not raw:
+            logger.warning("[vision.vlm] dashscope pro 返回空 elapsed=%.1fs", elapsed)
+            return None
+        usage = data.get("usage") or {}
+        logger.info(
+            "[vision.vlm] dashscope pro 完成 idx=%d elapsed=%.1fs in=%d out=%d",
+            idx, elapsed, usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0),
+        )
+        text = _strip_code_fence(raw)
+        l, r_pos = text.find("{"), text.rfind("}")
+        if l < 0 or r_pos <= l:
+            logger.warning("[vision.vlm] dashscope pro 无JSON elapsed=%.1fs head=%s", elapsed, raw[:200])
+            return None
+        obj = json.loads(text[l:r_pos+1])
+        if not isinstance(obj, dict):
+            return None
+        scene = obj.get("scene") or "通用"
+        mood = obj.get("mood") or ""
+        pp = _assemble_pp_from_json(obj)
+        name = obj.get("product_name") or "未识别"
+        brand = obj.get("brand") or "无法判断"
+        category = obj.get("category") or ("非产品图" if obj.get("has_person") else "无法判断")
+        return {
+            "name": name, "brand": brand, "category": category,
+            "appearance": obj.get("outfit") or "无法判断",
+            "packaging": "无法判断",
+            "text_on_package": [],
+            "key_features": obj.get("key_features") or ["无法判断"],
+            "scene": scene, "mood": mood, "portrait_prompt": pp,
+            "summary": f"{brand} {name}" if name != "未识别" else "未识别",
+            "_source": "vlm_pro_dashscope_json",
+        }
+    except Exception as e:
+        logger.warning("[vision.vlm] dashscope pro 异常 idx=%d elapsed=%.1fs err=%s", idx, time.time()-t0, e)
+        return None
+
+
+def _call_ark_pro(img_url: str, idx: int, model: str | None, timeout: int) -> dict[str, Any] | None:
+    """ARK 豆包 pro 兜底，保留 ai_client + prompt_loader + XML/JSON 双解析。"""
     t0 = time.time()
     try:
         from packages.application.viral_video.prompt_loader import (
-            get_template,
-            render_system_prompt,
-            render_user_prompt,
+            get_template, render_system_prompt, render_user_prompt,
         )
         from packages.shared.ai_client import get_doubao_client
     except ImportError as e:
         logger.warning("[vision.vlm] 导入失败: %s", e)
         return None
-
     try:
         template = get_template("image_analysis")
         system = render_system_prompt(template)
@@ -158,12 +250,10 @@ def call_pro_vlm(
     except Exception as e:
         logger.warning("[vision.vlm] 模板加载失败: %s", e)
         return None
-
     client = get_doubao_client()
     if not client.is_available:
         return None
-
-    use_model = model or DEFAULT_PRO_MODEL
+    use_model = model or _provider.pro_model()
     _orig_retries = client.max_retries
     client.max_retries = 0
     try:
@@ -176,23 +266,21 @@ def call_pro_vlm(
             model=use_model,
         )
     except Exception as e:
-        logger.warning("[vision.vlm] 图片 #%d pro VLM 调用失败 elapsed=%.1fs err=%s", idx, time.time() - t0, e)
+        logger.warning("[vision.vlm] ark pro 调用失败 idx=%d elapsed=%.1fs err=%s", idx, time.time()-t0, e)
         client.max_retries = _orig_retries
         return None
     client.max_retries = _orig_retries
-
     elapsed = time.time() - t0
     if not raw:
-        logger.warning("[vision.vlm] 图片 #%d pro VLM 返回空 elapsed=%.1fs", idx, elapsed)
+        logger.warning("[vision.vlm] ark pro 返回空 elapsed=%.1fs", elapsed)
         return None
-
     text = _strip_code_fence(raw)
-    l, r = text.find("{"), text.rfind("}")
-    if l >= 0 and r > l:
+    l, r_pos = text.find("{"), text.rfind("}")
+    if l >= 0 and r_pos > l:
         try:
-            obj = json.loads(text[l : r + 1])
+            obj = json.loads(text[l:r_pos+1])
             if isinstance(obj, dict):
-                logger.info("[vision.vlm] 图片 #%d pro VLM JSON 完成 elapsed=%.1fs", idx, elapsed)
+                logger.info("[vision.vlm] ark pro JSON 完成 idx=%d elapsed=%.1fs", idx, elapsed)
                 return {
                     "name": obj.get("name") or "未识别",
                     "brand": obj.get("brand") or "无法判断",
@@ -209,18 +297,21 @@ def call_pro_vlm(
                 }
         except json.JSONDecodeError:
             pass
-
     try:
-        result = _xml_to_product(text, idx)
-        result["_fallback_used"] = True
-        result["_pro_elapsed"] = round(elapsed, 2)
-        logger.info(
-            "[vision.vlm] 图片 #%d pro VLM XML 完成 elapsed=%.2fs pp=%s",
-            idx,
-            elapsed,
-            (result.get("portrait_prompt") or "")[:40],
-        )
-        return result
+        return _xml_to_product(text, idx)
     except Exception as e:
-        logger.warning("[vision.vlm] 图片 #%d 解析失败 elapsed=%.1fs err=%s head=%s", idx, elapsed, e, raw[:200])
+        logger.warning("[vision.vlm] ark pro 解析失败 idx=%d elapsed=%.1fs err=%s head=%s", idx, elapsed, e, raw[:200])
         return None
+
+
+def call_pro_vlm(
+    img_url: str,
+    idx: int,
+    *,
+    model: str | None = None,
+    timeout: int = DEFAULT_TIMEOUT,
+) -> dict[str, Any] | None:
+    """单次调用 pro VLM，解析后返回 product dict；失败返回 None。"""
+    if _provider.is_dashscope():
+        return _call_dashscope_pro(img_url, idx, timeout)
+    return _call_ark_pro(img_url, idx, model, timeout)

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

@@ -1,12 +1,16 @@
 # -*- coding: utf-8 -*-
-"""doubao-seed-2.1-lite 强约束 JSON-only 调用。
+"""V2 快速路径 JSON-only VLM 调用。
+
+支持双后端切换（通过环境变量 VISION_V2_PROVIDER=ark|dashscope）：
+- ark（火山方舟/豆包）：doubao-seed-2-1-lite-260915，thinking={type:"disabled"}
+- dashscope（阿里云百炼/qwen，默认对比测试）：qwen3.8-flash，enable_thinking=false
 
-目标：替代"人体属性/商品检测/图像标签"三个火山不存在的专用云端 API。
 设计要点：
-- system prompt 极致精简，只给字段 schema 和强约束（禁止自然语言、禁止 markdown）
-- max_tokens=350（比旧 VLM 的 1200 小很多，降低延迟）
-- temperature=0.1（极低，稳定输出 JSON）
-- timeout=8s（够快，失败则由外层走 pro VLM 兜底）
+- 直接用 httpx 发最小 payload，不走 ai_client 包装
+- 关闭 thinking/推理链（reasoning_tokens 是延迟主因）
+- system prompt 极致精简，只给字段 schema 和强约束
+- max_tokens=350、temperature=0.1（稳定输出 JSON）
+- 单次调用不重试（失败由外层走 pro 兜底）
 - 期望返回纯 JSON object（无 ```json 包裹、无解释文字）
 """
 
@@ -17,6 +21,8 @@ import logging
 import time
 from typing import Any
 
+from . import _provider
+
 logger = logging.getLogger(__name__)
 
 # 极简 system prompt：只给字段定义 + 硬性输出要求
@@ -24,12 +30,12 @@ _FAST_SYSTEM = (
     "你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
     "不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
     "{\n"
-    '  "has_person": true/false,  // 图中是否有人\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_wear": "下装款式；穿连衣裙时填null",\n'
     '  "lower_color": "下装主色",\n'
     '  "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
     '  "accessories": ["眼镜"/"帽子"/"项链"/"耳环"/"背包"/"手表"等数组],\n'
@@ -38,7 +44,7 @@ _FAST_SYSTEM = (
     '  "pose": "姿势，如站立/坐姿/侧身/行走等",\n'
     '  "scene": "场景，如室内/街拍/户外/办公室/家居/海边/雪景/森林等",\n'
     '  "style": "风格，如休闲/商务/运动/复古/潮流/甜美/酷飒/优雅/街头/法式等",\n'
-    '  "has_product": true/false,  // 是否有明确商品展示\n'
+    '  "has_product": true/false,\n'
     '  "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
     '  "product_name": "产品名称，非产品图填null",\n'
     '  "brand": "品牌或文字标识，无则null",\n'
@@ -51,21 +57,16 @@ _FAST_SYSTEM = (
 
 _FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
 
-# 默认模型
-DEFAULT_LITE_MODEL = "doubao-seed-2-1-lite-260915"
-DEFAULT_TIMEOUT = 8
+DEFAULT_TIMEOUT = _provider.fast_timeout_default()
 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()
@@ -79,27 +80,18 @@ def call_fast_json(
     timeout: int = DEFAULT_TIMEOUT,
     max_tokens: int = DEFAULT_MAX_TOKENS,
 ) -> dict[str, Any] | None:
-    """调用 lite VLM 返回结构化 dict；失败/非 JSON 返回 None。
-
-    直接用 httpx 发最小 payload（关闭 thinking），不走 ai_client 包装：
-    - 关闭 thinking/推理链（reasoning_tokens 是延迟主因，单次要10-12s）
-    - 单次调用不重试（失败由外层走 pro 兜底）
-    - 温度=0.1 稳定输出 JSON
-    """
+    """调用 fast VLM 返回结构化 dict；失败/非 JSON 返回 None。"""
     t0 = time.time()
     import httpx
 
     try:
-        from packages.shared import get_shared_settings
-
-        settings = get_shared_settings()
-        api_key = settings.doubao_api_key
-        base_url = (settings.doubao_base_url or "https://ark.cn-beijing.volces.com/api/v3").rstrip("/")
+        api_key = _provider.api_key()
+        base_url = _provider.base_url().rstrip("/")
         if not api_key:
-            logger.warning("[vision.v2] doubao api_key 未配置，跳过 fast_json")
+            logger.warning("[vision.v2] provider=%s api_key 未配置，跳过 fast_json", _provider.get_provider())
             return None
 
-        use_model = model or DEFAULT_LITE_MODEL
+        use_model = model or _provider.fast_model()
         url = f"{base_url}/chat/completions"
         payload: dict[str, Any] = {
             "model": use_model,
@@ -117,9 +109,8 @@ def call_fast_json(
             "max_tokens": max_tokens,
             "stream": False,
         }
-        # 关键：关闭 thinking（reasoning_tokens 是延迟主因，单次要10-12s）
-        # 方舟/豆包 Seed 2.x 支持 thinking={type:"disabled"}，且不要和 reasoning_effort 同时传（两者互斥会400）
-        payload["thinking"] = {"type": "disabled"}
+        # 按 provider 设置关 thinking 参数
+        payload.update(_provider.thinking_params())
 
         resp = httpx.post(
             url,
@@ -128,12 +119,13 @@ def call_fast_json(
             timeout=timeout,
         )
         elapsed = time.time() - t0
-        # 400 说明模型不支持 thinking 参数（极少数旧模型），重试一次不带 thinking
         if resp.status_code == 400:
-            body_preview = resp.text[:300].lower()
-            logger.warning("[vision.v2] fast_json HTTP 400 elapsed=%.1fs body=%s", elapsed, resp.text[:200])
-            if "thinking" in body_preview or "reasoning" in body_preview:
-                payload.pop("thinking", None)
+            logger.warning(
+                "[vision.v2] fast_json HTTP 400 provider=%s elapsed=%.1fs body=%s",
+                _provider.get_provider(), elapsed, resp.text[:200],
+            )
+            if _provider.is_400_thinking_error(resp.text[:500]):
+                _provider.pop_thinking_param(payload)
                 resp = httpx.post(
                     url,
                     headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
@@ -147,7 +139,10 @@ def call_fast_json(
             else:
                 return None
         elif resp.status_code != 200:
-            logger.warning("[vision.v2] fast_json HTTP %d elapsed=%.1fs body=%s", resp.status_code, elapsed, resp.text[:200])
+            logger.warning(
+                "[vision.v2] fast_json HTTP %d provider=%s elapsed=%.1fs body=%s",
+                resp.status_code, _provider.get_provider(), elapsed, resp.text[:200],
+            )
             return None
         data = resp.json()
         raw = (data.get("choices") or [{}])[0].get("message", {}).get("content")
@@ -155,21 +150,16 @@ def call_fast_json(
             logger.warning("[vision.v2] fast_json 返回 None 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] fast_json 直连完成 model=%s elapsed=%.1fs in=%d out=%d reasoning=%d",
-            use_model,
-            elapsed,
-            usage.get("prompt_tokens", 0),
-            usage.get("completion_tokens", 0),
-            usage.get("reasoning_tokens", 0),
+            "[vision.v2] fast_json 完成 provider=%s model=%s elapsed=%.1fs in=%d out=%d reasoning=%d",
+            _provider.get_provider(), use_model, elapsed,
+            usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), reasoning_tokens,
         )
-        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:
@@ -177,25 +167,18 @@ def call_fast_json(
         try:
             obj = json.loads(text)
         except json.JSONDecodeError:
-            logger.warning(
-                "[vision.v2] fast_json JSON 解析失败 elapsed=%.1fs head=%s",
-                elapsed,
-                raw[:200],
-            )
+            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"),
+            "[vision.v2] fast_json 完成 provider=%s model=%s elapsed=%.1fs has_person=%s has_product=%s category=%s",
+            _provider.get_provider(), 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)
+        logger.warning("[vision.v2] fast_json 异常 provider=%s elapsed=%.1fs err=%s", _provider.get_provider(), elapsed, e, exc_info=True)
         return None