fix: 脚本超时优化 + 主题智能匹配

问题A - 超时优化:
1. DoubaoClient httpx timeout 改为显式 Timeout(connect=10, read=T, write=10, pool=5)
2. intent_parsing 单次超时 60s→20s
3. copy_review(reviewer) 添加 timeout=25s
4. 脚本 fast_timeout 150→90s, pro_timeout 150→60s
5. _step_intent_parsing 加 60s 总 deadline
6. _step_script_generation 加 180s 总 deadline
7. lite/fallback 模型与 primary 相同时跳过重复调用

问题B - 主题匹配:
1. 新增 _determine_theme() 根据图片类型+营销目的智能推断主题
2. _fallback_script 动态主题+对应口播文案
3. _empty_copy_result/_validate_and_normalize/_script_from_xml 去除硬编码好物分享
4. intent_parsing prompt 注入 marketing_purpose 和 image_category_hint
5. storyboard prompt 增加主题匹配指导

分支: fix/script-generation-timeout-and-theme

alembic/versions/103_v7_prompt_and_tokens_3000.py

@@ -0,0 +1,194 @@
+# -*- coding: utf-8 -*-
+"""image_analysis v7 prompt + max_tokens 3000 + max_retries 3
+
+Revision ID: 103_v7_prompt_and_tokens_3000
+Revises: 102_image_analysis_max_tokens_1500
+Create Date: 2026-10-07
+
+变更：
+1. 插入v7精简prompt（~1KB，v6 ~4.5KB，删除few-shot/冗长规则，减少输出token占用），设为active
+2. v6停用（is_active=False），保留历史
+3. image_analysis capability: max_tokens 1500→3000，max_retries 1→3
+
+ai_capability_configs 由应用 create_all 创建，全新 alembic-only 库可能不存在，
+故第3步做 to_regclass 守卫（同 102）。
+"""
+
+from sqlalchemy import text
+
+from alembic import op
+
+revision = "103_v7_prompt_and_tokens_3000"
+down_revision = "102_image_analysis_max_tokens_1500"
+branch_labels = None
+depends_on = None
+
+V7_SYSTEM = """# 角色
+你是一位专业的图片分析师，擅长准确识别图片中的场景、人物、物体、文字、氛围。
+
+# 任务
+对用户上传的图片逐张分析，描述你看到的内容，输出JSON格式。
+
+## 技能
+
+### 技能1：判断图片类型
+判断图片属于哪种类型，type字段填对应的英文值：
+- 商品图（product）：单个或多个商品、产品包装
+- 门店场景图（store）：店铺内部、门头招牌、货架陈列
+- 人物图（person）：人物形象、穿搭造型、肖像照片
+- 风景图（scene）：风景、动物、美食、街景
+- 其他（other）：以上都不是
+
+### 技能2：描述通用信息
+不管什么图都要描述：
+- type：图片类型，填product/store/person/scene/other其中一个
+- scene：一句话描述场景，例如"理疗养生店内部，摆着多张理疗床和产品货架"
+- mood：整体氛围，2-4个词，例如"整洁专业"、"热闹温馨"
+- colors：主要颜色，最多5个，写具体颜色名（亮红色/米白色/深蓝色，不写笼统的红色蓝色）
+- visible_text：图片里看到的文字，说明什么字、在什么位置，最多5条；没看到就空数组
+- lighting：光线情况，例如"明亮柔光"、"自然光"、"室内暖黄灯"
+- composition：怎么拍的，例如"居中特写"、"中景平视"、"俯拍"
+- has_person：有没有人，true或false
+
+### 技能3：描述门店场景
+如果是门店场景图（type="store"），还要描述：
+- store_type：什么类型的店，例如"养生馆"、"便利店"、"餐饮店"、"母婴店"
+- brand_signage：招牌上写了什么字、有什么品牌标识
+- visual_elements：看到哪些显眼的东西（招牌样式、灯光、货架、商品陈列、海报、收银台等），最多8个
+- product_categories：看到哪些品类的商品，例如"饮料零食"、"养生产品"
+- promotion_elements：有没有促销活动（打折海报、满减吊旗等），没有就空数组
+- atmosphere：店内什么氛围，例如"亲民生活化"、"老字号专业感"
+- cleanliness：店内干净程度，例如"干净整洁"、"货架整齐"
+- 看到顾客或店员要描述他们在做什么，has_person填true
+
+### 技能4：描述商品
+如果是商品图（type="product"），逐个商品描述：
+- product_name：商品名称，尽量具体，例如"OMO奥妙除菌除螨洗衣液"；看不出来填null
+- brand：什么牌子，看不出来填null
+- category：类目，从以下选一个：服饰鞋包/美妆/数码/食品/家居清洁/母婴/配饰/其他
+- package_type：什么包装，例如"瓶装"、"盒装"、"罐装"、"袋装"、"多瓶装"
+- package_color：包装主要颜色，写具体色（亮红色不写红色）
+- body_shape：瓶身或包装形状，例如"圆润胖瓶"、"竖款带把手瓶身"
+- label_design：标签设计，例如"红色标签印白色品牌logo"
+- key_text_on_package：包装上最显眼的文字（品牌名、功能词、卖点词），最多5个
+- product_features：包装特征，3-6个短语，包含颜色、瓶盖、形状、标签图案
+- key_selling_points：核心卖点，1-3个短语
+
+### 技能5：描述人物
+如果是人物图（type="person"），描述：
+- person_count：几个人
+- gender：性别（男/女/无法判断）
+- age_range：年龄段（儿童/青少年/青年/中年/老年/无法判断）
+- outfit_style：穿搭风格，例如"休闲日常"、"通勤商务"、"街头潮流"
+- upper_wear：上装（颜色+款式+材质），穿裙装不填
+- lower_wear：下装（颜色+款式+版型），穿裙装不填
+- dress_wear：裙装描述，穿上下装不填
+- outerwear：外套
+- shoes：鞋子
+- bag：包袋，没有填null
+- accessories：配饰（眼镜/帽子/项链/耳环/手表/手链/围巾/腰带等），没有填空数组
+- hairstyle：发型
+- makeup：妆容，男生或看不出填null
+- expression：表情，例如"微笑看镜头"、"冷酷无表情"
+- pose：姿势动作，例如"身直立正对镜头"、"单手撩发"
+- body_type：身材，例如"纤细苗条"、"高挑身材"、"丰满匀称"
+- portrait_prompt：80-150字详细描述人物形象（后面用来AI生成肖像图），要写清年龄段、穿搭完整细节、发型发色、妆容、表情、姿势、场景、光线、风格感觉，语言要有画面感
+
+### 技能6：描述风景
+如果是风景图（type="scene"），描述：
+- scene_type：什么场景，例如"自然风景"、"城市街景"、"动物"、"美食"
+- main_subject：画面主体是什么
+- key_elements：关键元素，最多8个
+- environment_objects：周围环境物体，最多8个
+- atmosphere：整体氛围，例如"秋日慵懒氛围感"、"清新自然氧气感"
+- 有人物就描述人物特征
+
+## 限制
+- 只输出JSON，不要任何解释文字，不要markdown代码块包裹，不要写"好的""以下是分析结果"这种废话
+- 颜色写具体色调（亮红色/米白色/深蓝色/翠绿色），不写笼统词汇
+- 瓶身、包装、招牌上的文字尽量识别出来（品牌名、功能词、卖点词）
+- 多个商品、多个人物分开描述，不要合并
+- 看不出来、不确定的字段填null或空数组，布尔值填true/false，绝对不要瞎编
+- 确保JSON格式合法，所有大括号、中括号、引号正确闭合
+- 数组字段控制数量：colors最多5个，visible_text最多5条，visual_elements最多8个，accessories最多10个"""
+V7_USER = "请分析这张图片，按系统消息的JSON结构输出。"
+
+
+def _capability_table_exists(bind) -> bool:
+    return bool(bind.execute(text("SELECT to_regclass('public.ai_capability_configs')")).scalar())
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    # 1. 停用旧的active image_analysis prompt（含v6）
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = FALSE "
+            "WHERE prompt_type = 'image_analysis' AND is_active = TRUE"
+        )
+    )
+    # 2. 幂等插入v7（存在则更新并重新激活）
+    existing = bind.execute(
+        text("SELECT id FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = 7")
+    ).fetchone()
+    if existing:
+        bind.execute(
+            text(
+                "UPDATE viral_video_prompt_templates SET is_active = TRUE, "
+                "system_prompt = :sys, user_prompt_template = :usr, "
+                "name = 'v7 精简结构化分析', updated_at = NOW() "
+                "WHERE prompt_type = 'image_analysis' AND version = 7"
+            ),
+            {"sys": V7_SYSTEM, "usr": V7_USER},
+        )
+    else:
+        bind.execute(
+            text(
+                "INSERT INTO viral_video_prompt_templates "
+                "(prompt_type, version, name, system_prompt, user_prompt_template, "
+                "is_active, created_at, updated_at) "
+                "VALUES ('image_analysis', 7, 'v7 精简结构化分析', "
+                ":sys, :usr, TRUE, NOW(), NOW())"
+            ),
+            {"sys": V7_SYSTEM, "usr": V7_USER},
+        )
+    # 3. capability max_tokens=3000、max_retries=3（表不存在则跳过）
+    if _capability_table_exists(bind):
+        bind.execute(
+            text(
+                "UPDATE ai_capability_configs SET max_tokens = 3000, "
+                "updated_at = NOW() "
+                "WHERE capability_key = 'image_analysis' AND "
+                "(max_tokens IS NULL OR max_tokens < 3000)"
+            )
+        )
+        bind.execute(
+            text(
+                "UPDATE ai_capability_configs SET max_retries = 3, updated_at = NOW() "
+                "WHERE capability_key = 'image_analysis' AND "
+                "(max_retries IS NULL OR max_retries < 3)"
+            )
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    # 删除v7
+    bind.execute(
+        text("DELETE FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = 7")
+    )
+    # 恢复v6为active
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = TRUE "
+            "WHERE prompt_type = 'image_analysis' AND version = 6"
+        )
+    )
+    # tokens/retries回退
+    if _capability_table_exists(bind):
+        bind.execute(
+            text(
+                "UPDATE ai_capability_configs SET max_tokens = 1500, max_retries = 1, "
+                "updated_at = NOW() WHERE capability_key = 'image_analysis'"
+            )
+        )

apps/worker/worker_app/tasks/viral_video.py

@@ -316,9 +316,9 @@ _DEFAULT_NEGATIVE_PROMPTS = [
 ]
 
 
-def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict:
+def _empty_copy_result(duration: int = 15, ratio: str = "9:16", theme: str = "") -> dict:
     return {
-        "overview": {"theme": "好物推荐", "total_duration": duration, "aspect_ratio": ratio},
+        "overview": {"theme": theme or "好物推荐", "total_duration": duration, "aspect_ratio": ratio},
         "scene_and_lighting": "简洁明亮的室内场景，柔和自然光，产品主体清晰",
         "shots": [],
         "hard_constraints": list(_DEFAULT_HARD_CONSTRAINTS),
@@ -471,11 +471,15 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
 
     template = get_template("intent_parsing")
     system = render_system_prompt(template)
+    marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
+    image_category_hint = _determine_theme(image_analysis, marketing_purpose)
     user = render_user_prompt(
         template,
         user_copy_text=job.user_copy_text or "（未提供，全由 AI 创作）",
         industry=job.industry or "未指定",
         image_analysis=products_summary or "- （无图片分析结果）",
+        marketing_purpose=marketing_purpose,
+        image_category_hint=image_category_hint,
     )
 
     def _parse(raw: str) -> dict:
@@ -504,16 +508,25 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
 
     _client_fast = ai_router.get_llm_client("intent_parsing", variant="primary")
     _client_pro = ai_router.get_llm_client("intent_parsing", variant="lite")
+    _intent_deadline = time.time() + 60
+    _seen_models: set[str] = set()
     for _client, _lbl in [(_client_fast, "fast"), (_client_pro, "pro-fallback")]:
         if not _client or not _client.is_available:
             continue
+        if _client.model in _seen_models:
+            logger.info("[爆款视频] 意图解析跳过重复模型 %s label=%s", _client.model, _lbl)
+            continue
+        _seen_models.add(_client.model)
+        if time.time() > _intent_deadline:
+            logger.warning("[爆款视频] 意图解析超过60s总预算，跳过 label=%s", _lbl)
+            break
         try:
             logger.info("[爆款视频] 意图解析 model=%s label=%s", _client.model, _lbl)
             raw = _client.chat_completion(
                 [{"role": "system", "content": system}, {"role": "user", "content": user}],
                 temperature=0.4,
                 max_tokens=1024,
-                timeout=60,
+                timeout=20,
             )
             if not raw:
                 continue
@@ -550,6 +563,70 @@ def _persona_style_hint(persona_id: str) -> str:
     return "【人设风格：未指定】亲切自然、像朋友分享好物"
 
 
+def _determine_theme(image_analysis: dict | None, marketing_purpose: str = "") -> str:
+    """根据图片分析结果和营销目的，智能推断默认主题。
+
+    门店类→门店探店/到店体验；商品图→好物分享/产品种草；
+    人物图→穿搭/人物故事；场景图→场景氛围/空间体验。
+    """
+    products = (image_analysis or {}).get("products", []) or []
+    type_counts: dict[str, int] = {}
+    for p in products:
+        if not isinstance(p, dict):
+            continue
+        cat = (p.get("category") or "").strip()
+        if any(
+            k in cat
+            for k in (
+                "门店",
+                "店铺",
+                "餐饮",
+                "美容",
+                "美发",
+                "养生",
+                "健身",
+                "酒店",
+                "咖啡",
+                "奶茶",
+                "餐厅",
+                "颈肩",
+                "调理",
+            )
+        ):
+            type_counts["store"] = type_counts.get("store", 0) + 1
+        elif any(k in cat for k in ("人物", "穿搭", "人像", "服装")):
+            type_counts["person"] = type_counts.get("person", 0) + 1
+        elif any(k in cat for k in ("场景", "空间", "环境", "非产品")):
+            type_counts["scene"] = type_counts.get("scene", 0) + 1
+        elif cat and cat not in ("无法判断", "非产品图", ""):
+            type_counts["product"] = type_counts.get("product", 0) + 1
+        src = p.get("_source") or ""
+        if "store" in src:
+            type_counts["store"] = type_counts.get("store", 0) + 1
+        elif "person" in src:
+            type_counts["person"] = type_counts.get("person", 0) + 1
+
+    dominant = max(type_counts, key=type_counts.get) if type_counts else "product"
+    mp = (marketing_purpose or "").strip()
+
+    if any(k in mp for k in ("获客", "引流", "到店")):
+        if dominant == "store":
+            return "门店探店·到店体验"
+        return "门店探店·到店体验"
+    if any(k in mp for k in ("品牌", "宣传")):
+        return "品牌故事·门店体验" if dominant == "store" else "品牌故事·产品展示"
+    if any(k in mp for k in ("种草", "推荐")):
+        return "穿搭分享·人物种草" if dominant == "person" else "好物分享·产品种草"
+
+    theme_map = {
+        "store": "门店探店·到店体验",
+        "person": "穿搭分享·人物故事",
+        "scene": "空间体验·场景氛围",
+        "product": "好物分享·产品种草",
+    }
+    return theme_map.get(dominant, "好物分享·产品种草")
+
+
 def _build_products_summary(image_analysis: dict) -> str:
     """把 VLM 返回的商品分析结果拼给文案/分镜生成 prompt 用。
     优先用 summary（自然段落）；没有时用结构化字段兜底拼一段。"""
@@ -666,14 +743,36 @@ def _fallback_script(job: ViralVideoJob) -> dict:
     """脚本生成失败时的兜底脚本（极简但可用）。"""
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
     ratio = getattr(job, "video_ratio", None) or "9:16"
-    base = _empty_copy_result(dur, ratio)
-    voiceover = job.user_copy_text or "你好，给大家分享一款我最近在用的好物，真的很不错，推荐你们也试试。"
+    _ia = getattr(job, "image_analysis", None) or {}
+    _mp = getattr(job, "marketing_purpose", "") or ""
+    default_theme = _determine_theme(_ia, _mp)
+    base = _empty_copy_result(dur, ratio, theme=default_theme)
+    _voiceover_map = {
+        "store": "带你探店！今天来到这家店，环境真的超棒，服务也很到位，推荐大家来体验一下。",
+        "person": "哈喽，今天给大家分享我的日常穿搭，简单舒适又好看，你们觉得怎么样？",
+        "scene": "带大家感受一下这个空间，氛围感拉满，真的很适合打卡体验。",
+        "product": "你好，给大家分享一款我最近在用的好物，真的很不错，推荐你们也试试。",
+    }
+    _products = (_ia or {}).get("products", []) or []
+    _dominant = "product"
+    for p in _products:
+        if not isinstance(p, dict):
+            continue
+        src = p.get("_source") or ""
+        cat = p.get("category") or ""
+        if "store" in src or any(k in cat for k in ("门店", "店铺", "餐饮", "美容", "颈肩", "调理")):
+            _dominant = "store"
+            break
+        elif "person" in src or any(k in cat for k in ("人物", "穿搭", "人像")):
+            _dominant = "person"
+            break
+    voiceover = job.user_copy_text or _voiceover_map.get(_dominant, _voiceover_map["product"])
     shots = [
         {
             "time_range": f"0-{dur}秒",
             "shot_type_angle_movement": "中景平视，缓慢推镜",
-            "scene_and_dialogue": "明亮室内，人物自然出镜，微笑着看向镜头。" + voiceover,
-            "action_details": "人物手持产品自然展示，表情亲切，动作流畅",
+            "scene_and_dialogue": voiceover,
+            "action_details": "自然展示，表情亲切，动作流畅",
             "audio_bgm": "轻快流行BGM",
             "transition": "结束",
             "reference_image_index": 0 if job.images else None,
@@ -683,7 +782,7 @@ def _fallback_script(job: ViralVideoJob) -> dict:
     base["voiceover_script"] = voiceover
     base["final_copy"] = voiceover
     base["suggested_copy"] = voiceover
-    base["title"] = "好物分享"
+    base["title"] = default_theme
     return base
 
 
@@ -701,12 +800,18 @@ def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict:
     ov = raw.get("overview")
     if isinstance(ov, dict):
         base["overview"] = {
-            "theme": str(ov.get("theme") or "好物分享"),
+            "theme": str(
+                ov.get("theme")
+                or _determine_theme(getattr(job, "image_analysis", None), getattr(job, "marketing_purpose", ""))
+            ),
             "total_duration": int(ov.get("total_duration") or dur),
             "aspect_ratio": str(ov.get("aspect_ratio") or ratio),
         }
     else:
-        base["overview"]["theme"] = str(raw.get("title") or "好物分享")
+        base["overview"]["theme"] = str(
+            raw.get("title")
+            or _determine_theme(getattr(job, "image_analysis", None), getattr(job, "marketing_purpose", ""))
+        )
 
     base["scene_and_lighting"] = str(raw.get("scene_and_lighting") or base["scene_and_lighting"])
 
@@ -793,7 +898,11 @@ def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None:
     base = _empty_copy_result(dur, ratio)
     if not raw:
         return None
-    base["overview"]["theme"] = xp.text_of(raw, "overview_theme") or xp.text_of(raw, "title") or "好物分享"
+    base["overview"]["theme"] = (
+        xp.text_of(raw, "overview_theme")
+        or xp.text_of(raw, "title")
+        or _determine_theme(getattr(job, "image_analysis", None), getattr(job, "marketing_purpose", ""))
+    )
     est = xp.attr_int(xp.text_of(raw, "estimated_duration"), 0)
     if est:
         base["overview"]["total_duration"] = est
@@ -883,8 +992,11 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
     system_tpl = system_tpl.replace("{global_constraints}", GLOBAL_CONSTRAINTS)
     system_tpl = system_tpl.replace("{negative_rules}", NEGATIVE_RULES)
 
+    marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
+    image_category_hint = _determine_theme(image_analysis, marketing_purpose)
     fusion_brief = (
         f"意图：{intent_str}\n关键信息：{key_msgs}\n调性：{tone}\n"
+        f"营销目的：{marketing_purpose}\n建议主题方向：{image_category_hint}\n"
         f"用户原文：{job.user_copy_text or '（未提供）'}\n创作模式：{fusion_level}"
     )
     user = render_user_prompt(
@@ -936,22 +1048,32 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
     # #2220: 直接用 ai_router 获取 client，不再手动提取 model_key
     _client_fast = ai_router.get_llm_client("storyboard", variant="primary")
     _client_pro = ai_router.get_llm_client("storyboard", variant="lite")
-    _script_fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "150"))
-    _script_pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "150"))
+    _script_fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "90"))
+    _script_pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "60"))
+    _script_deadline = time.time() + 180
     try:
-        # #2217: doubao-seed-2-1-pro生成长编导脚本高峰期>90s，上调到150s，支持ENV覆盖
+        # #2233: fast_timeout=90s, pro_timeout=60s，总deadline 180s
         normalized = _try_gen(_client_fast, 0.8, 2500, "fast-first", tmo=_script_fast_tmo)
         if normalized is not None:
             return normalized
+        if time.time() > _script_deadline:
+            logger.warning("[爆款视频] 编导脚本超过180s总预算，使用兜底脚本")
+            return _fallback_script(job)
         normalized = _try_gen(_client_fast, 0.6, 3200, "fast-retry", tmo=_script_fast_tmo)
         if normalized is not None:
             return normalized
-        # 第三次：用 lite/pro 模型兜底
+        # 第三次：用 lite/pro 模型兜底，跳过与 primary 相同的模型
         if _client_pro and _client_pro.is_available:
-            normalized = _try_gen(_client_pro, 0.7, 3500, "pro-fallback", tmo=_script_pro_tmo)
-            if normalized is not None:
-                return normalized
-        logger.warning("[爆款视频] 编导脚本三次都未生成合格结果，使用兜底脚本")
+            if _client_pro.model != _client_fast.model:
+                if time.time() <= _script_deadline:
+                    normalized = _try_gen(_client_pro, 0.7, 3500, "pro-fallback", tmo=_script_pro_tmo)
+                    if normalized is not None:
+                        return normalized
+                else:
+                    logger.warning("[爆款视频] 编导脚本超过180s总预算，跳过pro-fallback")
+            else:
+                logger.info("[爆款视频] pro-fallback模型与primary相同(%s)，跳过重复调用", _client_pro.model)
+        logger.warning("[爆款视频] 编导脚本均未生成合格结果，使用兜底脚本")
         return _fallback_script(job)
     except Exception as e:
         logger.warning("[爆款视频] 编导脚本生成异常: %s，使用兜底脚本", e, exc_info=True)

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

@@ -381,9 +381,55 @@ def assemble_result(idx: int, fast_json: dict | None, ocr_texts: list[str]) -> d
     ocr_texts = ocr_texts or []
 
     if _is_v4_schema(fj):
-        return _assemble_v4(idx, fj, ocr_texts)
+        result = _assemble_v4(idx, fj, ocr_texts)
     else:
-        return _assemble_old(idx, fj, ocr_texts)
+        result = _assemble_old(idx, fj, ocr_texts)
+    return _apply_partial_fallback(result, fj)
+
+
+def _apply_partial_fallback(result: dict[str, Any], fj: dict) -> dict[str, Any]:
+    """partial（截断修复）产物的字段兜底：用已有碎片填充空字段，
+    避免"无法判断"直接透传给下游。非partial产物原样返回。"""
+    if not fj.get("_partial"):
+        return result
+    desc = str(fj.get("description") or "").strip()
+    # 收集所有顶层标量碎片作为兜底素材
+    fragments: list[str] = []
+    for k in ("main_subject", "store_type", "scene_type", "description"):
+        v = fj.get(k)
+        if isinstance(v, str) and v.strip() and v != "无法判断":
+            fragments.append(v.strip())
+    for arr_k in ("environment_objects", "key_elements", "visual_elements"):
+        arr = fj.get(arr_k) or []
+        if isinstance(arr, list):
+            for item in arr[:3]:
+                if isinstance(item, str) and item.strip():
+                    fragments.append(item.strip())
+                elif isinstance(item, dict):
+                    tv = item.get("text") or item.get("name")
+                    if tv:
+                        fragments.append(str(tv))
+    frag_text = "；".join(fragments[:3])
+
+    if result.get("name") in ("未识别", "", None) and (desc or frag_text):
+        result["name"] = (desc or fragments[0])[:30]
+    if str(result.get("appearance", "")).startswith("无法判断"):
+        if desc:
+            result["appearance"] = desc[:200]
+        elif frag_text:
+            result["appearance"] = frag_text[:200]
+    if result.get("key_features") in (["无法判断"], []) and (desc or fragments):
+        kf = []
+        if desc:
+            kf.append(desc[:30])
+        for f in fragments[:3]:
+            if f not in kf:
+                kf.append(f[:40])
+        result["key_features"] = kf[:8]
+    if result.get("summary") in ("未识别", "", None) and (desc or fragments):
+        result["summary"] = (desc or fragments[0])[:40]
+    result["_partial"] = True
+    return result
 
 
 def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
@@ -441,6 +487,16 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
                 parts.append(f"人物:{pv2}")
         return parts
 
+    # partial截断保护：声明了product但products数组没来得及输出时，
+    # 按已返回的碎片字段改路由，避免直接掉到other丢信息
+    if fj.get("_partial") and vtype == "product" and not products:
+        if any(fj.get(k) for k in ("signage_details", "store_layout", "brand_signage", "store_type")):
+            vtype = "store"
+        elif any(fj.get(k) for k in ("key_elements", "main_subject", "scene_type", "spatial_layout")):
+            vtype = "scene"
+        else:
+            vtype = "other"
+
     # ── 人物类 ──
     if vtype == "person":
         # 取第一个人物信息（v5 schema人物信息在顶层）
@@ -601,15 +657,49 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
     # 门店类
     if vtype == "store":
         store_type = fj.get("store_type") or "店铺"
-        name = store_type
-        brand = fj.get("brand_signage") or "无法判断"
+        # brand 多级兜底：brand_signage → visible_text招牌文字 → text_on_package短词
+        brand_raw = fj.get("brand_signage")
+        if not brand_raw or brand_raw in ("无法判断", "", None):
+            brand = None
+            # 从visible_text找招牌文字（通常是位置含招牌/门头/背景的短词）
+            for vt in visible_text:
+                vt_str = vt.get("text") if isinstance(vt, dict) else str(vt)
+                if not vt_str or len(vt_str) < 2 or len(vt_str) > 12:
+                    continue
+                loc = (vt.get("location") or "") if isinstance(vt, dict) else ""
+                if any(k in loc for k in ("招牌", "门头", "背景", "招牌墙")):
+                    brand = vt_str
+                    break
+            # 从text_on_package找2-8字的短词（非描述性）
+            if not brand:
+                _desc_words = {"干净", "整洁", "温馨", "专业", "明亮", "舒适", "宽敞", "现代", "传统", "时尚"}
+                for t in text_on_package:
+                    if 2 <= len(t) <= 8 and t not in _desc_words and not any(c in t for c in "的了是在我"):
+                        brand = t
+                        break
+            if not brand:
+                brand = "无法判断"
+        else:
+            brand = brand_raw
+        # name兜底：store_type为空时用brand
+        name = store_type if store_type != "店铺" else (brand if brand != "无法判断" else store_type)
         category = "门店场景"
         # appearance: store_layout + furnishings + 陈设色调
         appearance_parts = []
         if fj.get("store_layout"):
             appearance_parts.append(str(fj["store_layout"]))
         furnishings = fj.get("furnishings") or []
-        if isinstance(furnishings, list) and furnishings:
+        if isinstance(furnishings, dict):
+            _furn_vals = []
+            for fk in ("materials", "furniture", "shelving", "seating"):
+                fv = furnishings.get(fk)
+                if isinstance(fv, list):
+                    _furn_vals.extend(str(x) for x in fv if x)
+                elif isinstance(fv, str) and fv:
+                    _furn_vals.append(fv)
+            if _furn_vals:
+                appearance_parts.append("陈设：" + "、".join(_furn_vals[:4]))
+        elif isinstance(furnishings, list) and furnishings:
             appearance_parts.append("陈设：" + "、".join(str(f) for f in furnishings[:4] if f))
         if fj.get("cleanliness"):
             appearance_parts.append(str(fj["cleanliness"]))

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

@@ -23,10 +23,10 @@ logger = logging.getLogger(__name__)
 
 # 超时（可通过环境变量覆盖）
 _IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8"))
-_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "15"))
-_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "15"))
+_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "20"))
+_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "20"))
 _OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "6"))
-_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "30"))
+_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45"))
 
 _FALLBACK_RESULT = {
     "name": "未识别",
@@ -58,27 +58,29 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
 
     fj_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, timeout=_FAST_JSON_TIMEOUT)
-        f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=_OCR_TIMEOUT)
-        try:
-            for fut in as_completed([f_fj, f_ocr], timeout=_FAST_TIMEOUT):
-                try:
-                    res = fut.result(timeout=1)
-                except Exception as e:
-                    logger.warning("[vision.v2] 图片 #%d 子任务异常: %s", idx, e)
-                    continue
-                if fut is f_fj and isinstance(res, dict):
-                    fj_result = res
-                elif fut is f_ocr and isinstance(res, list):
-                    ocr_result = res
-        except TimeoutError:
-            for f in (f_fj, f_ocr):
-                if not f.done():
-                    f.cancel()
-            logger.warning("[vision.v2] 图片 #%d fast路径超时(%.0fs)，走pro兜底", idx, _FAST_TIMEOUT)
-
-    fast_elapsed = time.time() - t0
+    fast_elapsed = 0.0
+    pool = ThreadPoolExecutor(max_workers=2)
+    f_fj = pool.submit(vlm_fast_json.call_fast_json, img_url, timeout=_FAST_JSON_TIMEOUT)
+    f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=_OCR_TIMEOUT)
+    try:
+        for fut in as_completed([f_fj, f_ocr], timeout=_FAST_TIMEOUT):
+            try:
+                res = fut.result(timeout=1)
+            except Exception as e:
+                logger.warning("[vision.v2] 图片 #%d 子任务异常: %s", idx, e)
+                continue
+            if fut is f_fj and isinstance(res, dict):
+                fj_result = res
+            elif fut is f_ocr and isinstance(res, list):
+                ocr_result = res
+    except TimeoutError:
+        for f in (f_fj, f_ocr):
+            if not f.done():
+                f.cancel()
+        logger.warning("[vision.v2] 图片 #%d fast路径超时(%.0fs)，走pro兜底", idx, _FAST_TIMEOUT)
+    finally:
+        fast_elapsed = time.time() - t0
+        pool.shutdown(wait=False)  # 不等待未完成的线程，避免计时膨胀
 
     if fj_result:
         assembled = assembler.assemble_result(idx, fj_result, ocr_result)

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

@@ -0,0 +1,138 @@
+# -*- coding: utf-8 -*-
+"""VLM 返回文本的稳健 JSON 提取工具。
+
+背景：复杂门店图 VLM 输出经常被 max_tokens 截断（finish_reason=length），
+json.loads 失败后整个结果被丢弃，导致"未识别"。本工具提供：
+1. markdown 代码块剥离（含只开不闭的截断场景）
+2. 最外层 { } 切片
+3. 非法控制字符清理
+4. 直接 json.loads
+5. 截断 JSON 括号/引号栈补全修复
+6. 尾部逐字符截断重试（去除最后一个不完整 token 后修复）
+
+成功返回 dict；截断修复产物带 _partial=True 标记；彻底失败返回 None。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+_CODE_FENCE_RE = re.compile(r"^```(?:json)?\s*\n?(.*?)\n?```\s*$", re.DOTALL)
+
+
+def _strip_code_fence(s: str) -> str:
+    s = s.strip()
+    m = _CODE_FENCE_RE.match(s)
+    if m:
+        return m.group(1).strip()
+    # 兼容开头 ```json 但结尾无 ```（截断场景）
+    if s.startswith("```"):
+        lines = s.split("\n")
+        if lines and lines[0].startswith("```"):
+            lines = lines[1:]
+        s = "\n".join(lines).strip()
+    return s
+
+
+def _repair_truncated_json(text: str) -> str:
+    """尝试补全被截断的JSON：维护 bracket/quote 栈，在末尾补闭合符。"""
+    stack: list[str] = []
+    in_string = False
+    escape = False
+    for ch in text:
+        if escape:
+            escape = False
+            continue
+        if ch == "\\" and in_string:
+            escape = True
+            continue
+        if ch == '"':
+            in_string = not in_string
+            continue
+        if in_string:
+            continue
+        if ch in "{[":
+            stack.append(ch)
+        elif ch == "}":
+            if stack and stack[-1] == "{":
+                stack.pop()
+        elif ch == "]":
+            if stack and stack[-1] == "[":
+                stack.pop()
+    repair = ""
+    if in_string:
+        repair += '"'
+    for opener in reversed(stack):
+        repair += "}" if opener == "{" else "]"
+    if repair:
+        logger.info(
+            "[json_utils] 截断JSON修复: 补全%d个闭合符 in_string=%s",
+            len(repair),
+            in_string,
+        )
+    return text + repair
+
+
+def _clean_invalid_chars(text: str) -> str:
+    """清理JSON中非法的控制字符（tab/newline 之外的 0x00-0x1f 段）。"""
+    return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text)
+
+
+def extract_json_object(text: str) -> dict | None:
+    """从VLM返回文本中稳健提取JSON对象。
+
+    返回 dict 或 None。成功的 dict 可能带 _partial=True 标记，
+    表示原始文本被截断、经括号补全后得到的产物。
+    """
+    if not text or not isinstance(text, str):
+        return None
+    # 1. 剥离 markdown
+    text = _strip_code_fence(text)
+    # 2. 找最外层 { }
+    lpos = text.find("{")
+    if lpos < 0:
+        return None
+    rpos = text.rfind("}")
+    if rpos > lpos:
+        text = text[lpos : rpos + 1]
+    else:
+        # 截断场景：无任何闭合 }，取到末尾交给修复器
+        text = text[lpos:]
+    # 3. 清理非法控制字符
+    text = _clean_invalid_chars(text)
+    # 4. 直接 loads
+    try:
+        obj = json.loads(text)
+        return obj if isinstance(obj, dict) else None
+    except json.JSONDecodeError:
+        pass
+    # 5. 尝试截断修复
+    repaired = _repair_truncated_json(text)
+    try:
+        obj = json.loads(repaired)
+        if isinstance(obj, dict):
+            obj["_partial"] = True
+            return obj
+    except json.JSONDecodeError:
+        pass
+    # 6. 尾部逐字符截断重试（去除最后一个不完整 token）
+    for _ in range(50):
+        last_comma = repaired.rfind(",")
+        last_brace = max(repaired.rfind("}"), repaired.rfind("]"))
+        cut = max(last_comma, last_brace)
+        if cut < 10:
+            break
+        repaired = repaired[: cut + 1]
+        repaired = _repair_truncated_json(repaired)
+        try:
+            obj = json.loads(repaired)
+            if isinstance(obj, dict):
+                obj["_partial"] = True
+                return obj
+        except json.JSONDecodeError:
+            continue
+    return None

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

@@ -13,7 +13,6 @@ fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
 
 from __future__ import annotations
 
-import json
 import logging
 import time
 from typing import Any
@@ -22,7 +21,7 @@ from . import _prompt, assembler
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_TIMEOUT = 30
+_DEFAULT_TIMEOUT = 45
 
 
 def call_pro_vlm(
@@ -38,7 +37,7 @@ def call_pro_vlm(
     try:
         from packages.shared.ai_router import ai_router
 
-        client = ai_router.get_vision_client("image_analysis", variant="primary")
+        client = ai_router.get_vision_client("image_analysis", variant="fallback")
         if not client or not client.is_available:
             logger.warning("[vision.v2] pro vision client 不可用，跳过")
             return None
@@ -68,30 +67,43 @@ def call_pro_vlm(
             "enable_thinking": False,
             "response_format": {"type": "json_object"},
         }
-        if max_tokens is not None:
-            call_kwargs["max_tokens"] = max_tokens
-        raw = client.vision_completion(**call_kwargs)
-        elapsed = time.time() - t0
-        if not raw:
-            logger.warning("[vision.v2] pro 返回空 elapsed=%.1fs", elapsed)
-            return None
+        # pro fallback：显式4000 tokens给复杂门店图留足空间
+        call_kwargs["max_tokens"] = max_tokens if max_tokens is not None else 4000
 
+        from .json_utils import extract_json_object
+
+        raw = None
+        obj = None
+        for _outer in range(2):
+            kw = dict(call_kwargs)
+            if _outer == 1:
+                kw.pop("response_format", None)
+                msgs2 = [dict(messages[0]), dict(messages[1])]
+                cont = [dict(c) for c in list(msgs2[1]["content"])]
+                cont[-1] = {"type": "text", "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。"}
+                msgs2[1] = {"role": "user", "content": cont}
+                kw["messages"] = msgs2
+            raw = client.vision_completion(**kw)
+            if not raw:
+                logger.warning("[vision.v2] pro 返回空 outer=%s", _outer)
+                continue
+            obj = extract_json_object(raw)
+            if obj is not None:
+                break
+            logger.warning("[vision.v2] pro 非JSON(100字) outer=%s: %s", _outer, raw[:100])
+
+        elapsed = time.time() - t0
+        if obj is None:
+            logger.warning("[vision.v2] pro 两次均未得到JSON elapsed=%.1fs", elapsed)
+            return None
+        if obj.get("_partial"):
+            logger.warning("[vision.v2] pro 返回截断JSON(partial) elapsed=%.1fs", elapsed)
         logger.info(
-            "[vision.v2] pro 完成 model=%s elapsed=%.1fs",
+            "[vision.v2] pro 完成 model=%s elapsed=%.1fs type=%s",
             client.model,
             elapsed,
+            obj.get("type"),
         )
-        s = _strip_code_fence(raw)
-        lpos, rr = s.find("{"), s.rfind("}")
-        if lpos >= 0 and rr > lpos:
-            s = s[lpos : rr + 1]
-        try:
-            obj = json.loads(s)
-        except json.JSONDecodeError:
-            logger.warning("[vision.v2] pro JSON 解析失败 head=%s", raw[:200])
-            return None
-        if not isinstance(obj, dict):
-            return None
 
         # 通过assembler统一组装，兼容v4嵌套schema和旧扁平schema
         result = assembler.assemble_result(idx, obj, [])
@@ -102,15 +114,3 @@ def call_pro_vlm(
         elapsed = time.time() - t0
         logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
         return None
-
-
-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

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

@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import json
 import logging
 import time
 from typing import Any
@@ -23,19 +22,7 @@ from . import _prompt
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_TIMEOUT = 15
-
-
-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
+_DEFAULT_TIMEOUT = 20
 
 
 def call_fast_json(
@@ -86,35 +73,48 @@ def call_fast_json(
         }
         if max_tokens is not None:
             call_kwargs["max_tokens"] = max_tokens
-        raw = client.vision_completion(**call_kwargs)
-        elapsed = time.time() - t0
-        if not raw:
-            logger.warning("[vision.v2] fast_json 返回空 elapsed=%.1fs", elapsed)
-            return None
 
+        # 双重防护：第1次正常调用；第2次去掉json_object强约束（部分模型在该约束下
+        # 反而幻觉），并加严格指令。解析全部走 json_utils，截断partial产物可用。
+        from .json_utils import extract_json_object
+
+        raw = None
+        obj = None
+        for _outer in range(2):
+            kw = dict(call_kwargs)
+            if _outer == 1:
+                kw.pop("response_format", None)
+                msgs2 = [dict(messages[0]), dict(messages[1])]
+                cont = list(msgs2[1]["content"])
+                cont = [dict(c) for c in cont]
+                cont[-1] = {"type": "text", "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。"}
+                msgs2[1] = {"role": "user", "content": cont}
+                kw["messages"] = msgs2
+            raw = client.vision_completion(**kw)
+            if not raw:
+                logger.warning("[vision.v2] fast_json 返回空 outer=%s", _outer)
+                continue
+            obj = extract_json_object(raw)
+            if obj is not None:
+                break
+            logger.warning(
+                "[vision.v2] fast_json 非JSON(100字) outer=%s: %s",
+                _outer,
+                raw[:100],
+            )
+
+        elapsed = time.time() - t0
+        if obj is None:
+            logger.warning("[vision.v2] fast_json 两次均未得到JSON elapsed=%.1fs", elapsed)
+            return None
+        if obj.get("_partial"):
+            logger.warning("[vision.v2] fast_json 返回截断JSON(partial) elapsed=%.1fs", elapsed)
         logger.info(
-            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs",
+            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs has_person=%s type=%s",
             client.model,
             elapsed,
-        )
-        text = _strip_code_fence(raw)
-        lpos, r = text.find("{"), text.rfind("}")
-        if lpos >= 0 and r > lpos:
-            text = text[lpos : 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 完成 elapsed=%.1fs has_person=%s has_product=%s category=%s",
-            elapsed,
             obj.get("has_person"),
-            obj.get("has_product"),
-            obj.get("category"),
+            obj.get("type"),
         )
         return obj
     except Exception as e:

packages/application/viral_video/prompts.py

@@ -110,10 +110,12 @@ _INTENT_SYSTEM = f"""你负责理解用户的营销意图。用户给的文案
 
 _INTENT_USER = """用户原始文案：{user_copy_text}
 所属行业：{industry}
+营销目的：{marketing_purpose}
 图片分析结果（供参考）：
 {image_analysis}
+图片类型推断：{image_category_hint}
 
-请理解用户意图，按标签格式输出。"""
+请理解用户意图，按标签格式输出。注意：theme和emotion_tone应与图片类型和营销目的匹配——门店类图片偏向"门店探店/到店体验"，商品图偏向"好物分享/产品种草"，人物图偏向"穿搭/人物故事"。"""
 
 _INTENT_EXAMPLE = """<intent_summary>一款厨房去油污神器，喷一喷油污就掉</intent_summary>
 <core_messages>
@@ -176,7 +178,7 @@ _FUSION_EXAMPLE = """<title>厨房重油污，别再用洗洁精硬擦了</title
 <segment duration_sec="4" image_index="0">39块钱625ml，厨房重油污的可以试一瓶</segment>
 </script_segments>
 <voiceover_script>这油污我真的忍很久了，用洗洁精擦半天都没用。后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净。39块钱625ml，厨房重油污的可以试一瓶。</voiceover_script>
-<overview_theme>厨房油污清洁好物分享</overview_theme>
+<overview_theme>厨房好物分享·产品种草</overview_theme>
 <scene_and_lighting>简洁明亮的厨房台面场景，自然光从窗户洒入，色调温暖柔和，突出产品白色瓶身与去油污对比效果。</scene_and_lighting>
 <word_count>58</word_count>
 <estimated_duration>13</estimated_duration>"""
@@ -215,6 +217,8 @@ _STORYBOARD_USER = """目标时长：{duration}秒
 图片分析结果：
 {image_analysis}
 
+重要：overview_theme 必须与图片实际内容和营销目的匹配。门店/餐饮/服务类图片用"门店探店·到店体验"；商品图用"好物分享·产品种草"；人物图用"穿搭分享·人物故事"；场景图用"空间体验·场景氛围"。不要对所有图片都使用"好物分享"。
+
 请按标签格式输出分镜。"""
 
 _STORYBOARD_EXAMPLE = """<clips>

packages/application/viral_video/reviewer.py

@@ -101,6 +101,7 @@ class Reviewer:
             ],
             temperature=0.2,
             max_tokens=1024,
+            timeout=25,
         )
         if not raw:
             return None
@@ -247,6 +248,7 @@ class Reviewer:
             ],
             temperature=0.5,
             max_tokens=2048,
+            timeout=25,
         )
         if not raw:
             return self._rule_fix(fusion, review)

packages/config/base.py

@@ -96,7 +96,7 @@ class SharedSettings(BaseSettings):
     doubao_fast_model: str = ""
     doubao_base_url: str = ""
     doubao_timeout: int = 45
-    doubao_max_retries: int = 1
+    doubao_max_retries: int = 3
     doubao_vision_model: str = ""
     doubao_vision_lite_model: str = ""
     doubao_vision_use_lite: bool = True

packages/shared/ai_client.py

@@ -198,6 +198,7 @@ class DoubaoClient:
         self.temperature: float | None = temperature
         self.extra_params: dict = extra_params or {}
         self.vision_model: str = settings.doubao_vision_model
+        self.last_finish_reason: str = ""
         self.vision_lite_model: str = settings.doubao_vision_lite_model
         self.fast_model: str = settings.doubao_fast_model
         self.embedding_model: str = settings.doubao_embedding_model
@@ -210,6 +211,13 @@ class DoubaoClient:
         # 最近一次图片生成的详细错误，供上层读取
         self.last_image_error: dict = {}
 
+    def _resolve_timeout(self, timeout) -> "httpx.Timeout":
+        """将整数超时转为 httpx.Timeout，区分 connect/read/write/pool，避免 read 卡到 TCP 120s 默认值."""
+        if isinstance(timeout, httpx.Timeout):
+            return timeout
+        t = int(timeout) if timeout else 60
+        return httpx.Timeout(connect=10, read=max(t, 10), write=10, pool=5)
+
     def embed_text(self, text: str, timeout: int | None = None) -> list[float] | None:
         """调用豆包文本 Embedding API，返回浮点向量；失败返回 None。"""
         if not self.is_available or not text or not text.strip():
@@ -300,7 +308,7 @@ class DoubaoClient:
         _t0 = time.time()
         for attempt in range(self.max_retries + 1):
             try:
-                _req_timeout = timeout if timeout is not None else self.timeout
+                _req_timeout = self._resolve_timeout(timeout if timeout is not None else self.timeout)
                 response = httpx.post(
                     url,
                     headers=headers,
@@ -311,9 +319,9 @@ class DoubaoClient:
                 data = response.json()
                 finish_reason = (data.get("choices") or [{}])[0].get("finish_reason", "")
                 if finish_reason == "length" and attempt < self.max_retries:
-                    # 输出被 max_tokens 截断：1.5x 扩容后重试（计入 max_retries，不额外增加）
+                    # 输出被 max_tokens 截断：2.0x 扩容后重试（计入 max_retries，不额外增加）
                     old_max = int(payload["max_tokens"])
-                    new_max = int(old_max * 1.5)
+                    new_max = int(old_max * 2)
                     payload["max_tokens"] = new_max
                     wait = 0.5 * (2**attempt)
                     logger.warning(
@@ -326,6 +334,7 @@ class DoubaoClient:
                     time.sleep(wait)
                     continue
                 content = data["choices"][0]["message"]["content"]
+                self.last_finish_reason = finish_reason
                 _elapsed = time.time() - _t0
                 logger.info(
                     "[doubao] chat_completion 完成 model=%s tokens_in=%d tokens_out=%d elapsed=%.1fs attempt=%d timeout=%d",
@@ -424,7 +433,7 @@ class DoubaoClient:
         if kwargs:
             payload.update(kwargs)
 
-        req_timeout = timeout or self.timeout
+        req_timeout = self._resolve_timeout(timeout or self.timeout)
         last_error: Optional[Exception] = None
         _t0 = time.time()
         for attempt in range(self.max_retries + 1):
@@ -439,9 +448,9 @@ class DoubaoClient:
                 data = response.json()
                 finish_reason = (data.get("choices") or [{}])[0].get("finish_reason", "")
                 if finish_reason == "length" and attempt < self.max_retries:
-                    # 视觉输出被 max_tokens 截断：1.5x 扩容后重试（计入 max_retries）
+                    # 视觉输出被 max_tokens 截断：2.0x 扩容后重试（计入 max_retries）
                     old_max = int(payload["max_tokens"])
-                    new_max = int(old_max * 1.5)
+                    new_max = int(old_max * 2)
                     payload["max_tokens"] = new_max
                     wait = 0.5 * (2**attempt)
                     logger.warning(
@@ -454,6 +463,7 @@ class DoubaoClient:
                     time.sleep(wait)
                     continue
                 content = data["choices"][0]["message"]["content"]
+                self.last_finish_reason = finish_reason
                 _elapsed = time.time() - _t0
                 logger.info(
                     "[doubao] vision_completion 完成 model=%s tokens_in=%d tokens_out=%d elapsed=%.1fs attempt=%d",

packages/shared/ai_router.py

@@ -62,7 +62,15 @@ class CapabilityConfig:
 class TTSClient:
     """TTS 客户端（简单配置持有者，实际调用由 CosyVoiceService 完成）"""
 
-    def __init__(self, provider: str, api_key: str, base_url: str, model: str, timeout: int = 60, extra_params: dict | None = None):
+    def __init__(
+        self,
+        provider: str,
+        api_key: str,
+        base_url: str,
+        model: str,
+        timeout: int = 60,
+        extra_params: dict | None = None,
+    ):
         self.provider = provider
         self.api_key = api_key
         self.base_url = base_url
@@ -78,7 +86,15 @@ class TTSClient:
 class ImageGenClient:
     """图片生成客户端（简单配置持有者）"""
 
-    def __init__(self, provider: str, api_key: str, base_url: str, model: str, timeout: int = 60, extra_params: dict | None = None):
+    def __init__(
+        self,
+        provider: str,
+        api_key: str,
+        base_url: str,
+        model: str,
+        timeout: int = 60,
+        extra_params: dict | None = None,
+    ):
         self.provider = provider
         self.api_key = api_key
         self.base_url = base_url
@@ -94,7 +110,15 @@ class ImageGenClient:
 class VideoGenClient:
     """视频生成客户端（简单配置持有者）"""
 
-    def __init__(self, provider: str, api_key: str, base_url: str, model: str, timeout: int = 600, extra_params: dict | None = None):
+    def __init__(
+        self,
+        provider: str,
+        api_key: str,
+        base_url: str,
+        model: str,
+        timeout: int = 600,
+        extra_params: dict | None = None,
+    ):
         self.provider = provider
         self.api_key = api_key
         self.base_url = base_url
@@ -253,13 +277,27 @@ class AIRouter:
             return config
 
     def _get_model_or_fallback(self, cap: CapabilityConfig, variant: str = "primary") -> ModelConfig | None:
-        """按 variant 选择模型，不存在则 fallback"""
-        if variant == "lite" and cap.lite_model:
-            return cap.lite_model
-        if cap.primary_model:
-            return cap.primary_model
-        if cap.fallback_model:
-            return cap.fallback_model
+        """按 variant 选择模型，不存在则降级。
+
+        - primary: primary → fallback
+        - lite: lite → primary
+        - fallback: fallback → primary（修复点：此前 fallback variant 被忽略，错误地使用了 primary 模型）
+        """
+        if variant == "fallback":
+            if cap.fallback_model:
+                return cap.fallback_model
+            if cap.primary_model:
+                return cap.primary_model
+        elif variant == "lite":
+            if cap.lite_model:
+                return cap.lite_model
+            if cap.primary_model:
+                return cap.primary_model
+        else:  # primary
+            if cap.primary_model:
+                return cap.primary_model
+            if cap.fallback_model:
+                return cap.fallback_model
         return None
 
     # ── 构建客户端 ─────────────────────────────────────────────────────────

tests/unit/test_api_settings.py

@@ -81,7 +81,7 @@ class TestSharedSettingsDefaults:
         s = SharedSettings()
         assert s.doubao_model == ""  # 零硬编码：默认值已清空
         assert s.doubao_timeout == 45  # #2180 默认提到45s
-        assert s.doubao_max_retries == 1
+        assert s.doubao_max_retries == 3
 
 
 class TestAPISettingsDefaults:

tests/unit/test_config_base.py

@@ -111,7 +111,7 @@ class TestSharedSettingsDefaults:
         """豆包默认配置"""
         s = self._make_settings()
         assert s.doubao_timeout == 45  # #2180 默认提到45s
-        assert s.doubao_max_retries == 1
+        assert s.doubao_max_retries == 3
         assert s.doubao_base_url == ""  # 零硬编码：默认值已清空
 
     def test_default_empty_api_keys(self):