test(vision): update wiring tests for V2 batch analyze_images_v2 path

根因：上版同时传 thinking={type:disabled} + reasoning_effort=low，方舟返回400
Invalid combination of reasoning_effort and thinking type。降级重试分支
把两个参数都pop掉，模型回到默认thinking开启→响应9-12s超时，fast全败。

修复：
1. 只传 thinking={"type":"disabled"}，去掉互斥的 reasoning_effort
2. 400降级只pop thinking，保留精简payload重试（不pop多个）
3. 先单独验证 lite 单次HTTP 200且reasoning_tokens=0再跑E2E，省一轮部署

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

@@ -21,9 +21,9 @@ 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", "6"))
-_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "5"))
-_OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "5"))
+_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "8"))
+_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "8"))
+_OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "6"))
 _PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45"))
 
 _FALLBACK_RESULT = {

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

@@ -187,10 +187,10 @@ def call_pro_vlm(
         return None
 
     text = _strip_code_fence(raw)
-    l, r = text.find("{"), text.rfind("}")
-    if l >= 0 and r > l:
+    lb, rb = text.find("{"), text.rfind("}")
+    if lb >= 0 and rb > lb:
         try:
-            obj = json.loads(text[l : r + 1])
+            obj = json.loads(text[lb : rb + 1])
             if isinstance(obj, dict):
                 logger.info("[vision.vlm] 图片 #%d pro VLM JSON 完成 elapsed=%.1fs", idx, elapsed)
                 return {

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

@@ -81,35 +81,90 @@ def call_fast_json(
 ) -> dict[str, Any] | None:
     """调用 lite VLM 返回结构化 dict；失败/非 JSON 返回 None。
 
-    注意：不做重试（外层竞速/降级逻辑负责），max_retries=0 由外层统一设置。
+    直接用 httpx 发最小 payload（关闭 thinking），不走 ai_client 包装：
+    - 关闭 thinking/推理链（reasoning_tokens 是延迟主因，单次要10-12s）
+    - 单次调用不重试（失败由外层走 pro 兜底）
+    - 温度=0.1 稳定输出 JSON
     """
     t0 = time.time()
-    try:
-        from packages.shared.ai_client import get_doubao_client
+    import httpx
 
-        client = get_doubao_client()
-        if not client.is_available:
-            logger.warning("[vision.v2] doubao client 不可用，跳过 fast_json")
+    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("/")
+        if not api_key:
+            logger.warning("[vision.v2] doubao api_key 未配置，跳过 fast_json")
             return None
 
         use_model = model or DEFAULT_LITE_MODEL
-        # 强制不重试：lite 是快速路径，失败直接走外层 pro 兜底
-        _orig_retries = client.max_retries
-        client.max_retries = 0
-        try:
-            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,
+        url = f"{base_url}/chat/completions"
+        payload: dict[str, Any] = {
+            "model": use_model,
+            "messages": [
+                {"role": "system", "content": _FAST_SYSTEM},
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "image_url", "image_url": {"url": img_url}},
+                        {"type": "text", "text": _FAST_USER},
+                    ],
+                },
+            ],
+            "temperature": 0.1,
+            "max_tokens": max_tokens,
+            "stream": False,
+        }
+        # 关键：关闭 thinking（reasoning_tokens 是延迟主因，单次要10-12s）
+        # 方舟/豆包 Seed 2.x 支持 thinking={type:"disabled"}，且不要和 reasoning_effort 同时传（两者互斥会400）
+        payload["thinking"] = {"type": "disabled"}
+
+        resp = httpx.post(
+            url,
+            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+            json=payload,
+            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)
+                resp = httpx.post(
+                    url,
+                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                    json=payload,
+                    timeout=timeout,
+                )
+                elapsed = time.time() - t0
+                if resp.status_code != 200:
+                    logger.warning("[vision.v2] fast_json 降级重试 HTTP %d elapsed=%.1fs", resp.status_code, elapsed)
+                    return None
+            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]
             )
-        finally:
-            client.max_retries = _orig_retries
+            return None
+        data = resp.json()
+        raw = (data.get("choices") or [{}])[0].get("message", {}).get("content")
+        if raw is None:
+            logger.warning("[vision.v2] fast_json 返回 None elapsed=%.1fs", elapsed)
+            return None
+        usage = data.get("usage") or {}
+        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),
+        )
         elapsed = time.time() - t0
         if raw is None:
             logger.warning("[vision.v2] fast_json 返回 None elapsed=%.1fs model=%s", elapsed, use_model)
@@ -117,10 +172,10 @@ def call_fast_json(
 
         text = _strip_code_fence(raw)
         # 截到第一个 { 和最后一个 } 之间，容忍前后偶发文字
-        l = text.find("{")
-        r = text.rfind("}")
-        if l >= 0 and r > l:
-            text = text[l : r + 1]
+        lb = text.find("{")
+        rb = text.rfind("}")
+        if lb >= 0 and rb > lb:
+            text = text[lb : rb + 1]
         try:
             obj = json.loads(text)
         except json.JSONDecodeError:

tests/unit/test_viral_video_wiring.py

@@ -97,21 +97,28 @@ def invalidate_loader_cache():
 
 
 class TestImageAnalysisWiring:
-    def test_uses_loader_template_and_xml_parse(self, job):
+    def test_v2_batch_analysis_returns_products(self, job):
+        """V2 路径：_step_image_analysis 批量调用 analyze_images_v2，返回 products。"""
         from apps.worker.worker_app.tasks import viral_video as vv
 
-        with patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML) as mock_v:
-            result = vv._analyze_single_image(0, "https://img/1.jpg", "vlm-lite", 15)
+        fake_product = {
+            "name": "lipstick",
+            "brand": "品牌X",
+            "category": "唇部彩妆",
+            "key_features": ["显白", "持久"],
+            "text_on_package": ["品牌X", "211"],
+            "_source": "v2_fast_json",
+        }
+        with patch("worker_app.tasks.vision.analyze_images_v2", return_value=[fake_product]) as mock_aiv2:
+            result = vv._step_image_analysis(job)
 
-        mock_v.assert_called_once()
-        # 验证调用时传入了 system_prompt（说明走了 loader 渲染的模板）
-        call_kwargs = mock_v.call_args.kwargs
-        assert "system_prompt" in call_kwargs and call_kwargs["system_prompt"]
-        # 结果包含从 XML 解析出的产品信息
-        assert result["name"] == "lipstick"
-        assert result["brand"] == "品牌X"
-        assert "显白" in result["key_features"]
-        assert result["text_on_package"] == ["品牌X", "211"]
+        mock_aiv2.assert_called_once()
+        products = result["products"]
+        assert len(products) == 1
+        assert products[0]["name"] == "lipstick"
+        assert products[0]["brand"] == "品牌X"
+        assert "显白" in products[0]["key_features"]
+        assert products[0]["text_on_package"] == ["品牌X", "211"]
 
 
 # ── 2) 意图解析走模板 ───────────────────────────────────────────────
@@ -252,18 +259,22 @@ class TestEndToEndLoaderUsed:
             called_types.append(prompt_type)
             return real_get(prompt_type, **kwargs)
 
+        # V2 图片分析不再走 prompt_loader（固定 lite JSON prompt），用 mock 产品代替
+        img_res = {
+            "name": "lipstick",
+            "brand": "品牌X",
+            "key_features": ["显白", "持久"],
+            "text_on_package": ["品牌X"],
+        }
         with (
             patch.object(pl, "get_template", side_effect=spy_get),
-            patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML),
             patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML),
         ):
-            # 1) image
-            img_res = vv._analyze_single_image(0, "https://img/1.jpg", "vlm", 15)
-            # 2) intent
+            # intent
             intent_res = vv._step_intent_parsing(job, {"products": [img_res]})
 
-        # 前两步分别调用了 image_analysis 和 intent_parsing
-        assert "image_analysis" in called_types
+        # V2 图片分析走固定 prompt（不经 loader）；intent 仍走 loader
+        assert "image_analysis" not in called_types
         assert "intent_parsing" in called_types
 
         # script 和 review 单独验证（需要不同的 LLM 返回）