From b31205e965119056d7267edf123f1079f059727f Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 2 Oct 2026 12:32:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(viral-video):=20VLM=20lite=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E8=87=AA=E5=8A=A8=E9=99=8D=E7=BA=A7pro=20+=2010?= =?UTF-8?q?=E7=A7=8DIP=E4=BA=BA=E8=AE=BE=E6=B3=A8=E5=85=A5prompt=20(#2139)?= =?UTF-8?q?=20(#2140)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: xiaoxia Co-committed-by: xiaoxia --- apps/worker/worker_app/tasks/viral_video.py | 241 +++++++++++++------- 1 file changed, 164 insertions(+), 77 deletions(-) diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index 51e2c3f48..a5d2fb6da 100644 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -154,49 +154,42 @@ def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict: # ── 流水线各步骤 ──────────────────────────────────────────────────────── -_IMAGE_ANALYSIS_SYSTEM_PROMPT = """你是资深电商商品视觉分析师，擅长从商品图片中提取结构化商品信息。严格遵守以下规则： - -1. 只基于图片中真实可见的内容进行分析，严禁编造图片中不存在的品牌、文字、规格、卖点或功效。 -2. 看不清、包装上没有、无法判断的字段统一填「无法判断」。 -3. 包装上的文字只 OCR 你能清晰看到的，模糊的不要瞎猜。 -4. 输出必须是严格 JSON（不要 Markdown 代码块，不要额外解释文字）。 - -字段说明： +_IMAGE_ANALYSIS_SYSTEM_PROMPT = """你是电商商品视觉分析师，从商品图片中提取关键商品信息。严格规则： +1. 只说图片里真实可见的内容，看不清/没有的填「无法判断」，不要瞎猜。 +2. 输出必须是严格 JSON（不要 Markdown 代码块，不要额外解释文字）。 +3. 字段说明： { - "name": "商品全名（从包装 OCR 读出品牌+产品名+规格/容量/型号，如『李字无烟檀香型蚊香 30单盘装』）", - "brand": "品牌名（从 Logo/包装文字读出，如『李字』『奥妙(OMO)』；看不清填『无法判断』）", - "category": "商品品类（如『家用清洁/洗衣液』『日化驱蚊/盘式蚊香』『护肤/面霜』；非产品图填『非产品图』）", - "spec": "规格/型号/容量/净含量（从包装文字 OCR 读出，如『30单盘装』『3kg』『500ml』；无法判断填『无法判断』）", - "appearance": "外观与视觉特征（80-150字自然语言描述整体形状、颜色搭配、材质质感、尺寸感知，让没看到图的人能想象长什么样）", - "packaging": "包装设计细节（80-150字自然语言描述标签颜色分区、图案元素、图形标识、封口/塑封状态、瓶盖/泵头/罐型等）", - "text_on_package": ["包装上能清晰 OCR 读出的文字逐条列出，按主次：品牌名、产品名、卖点文案、功效描述、规格参数等，看不清的不列"], - "colors": ["主体颜色 2-4 个"], - "material_or_texture": "材质/质地（如玻璃瓶装/塑料软管/哑光质感/金属外壳/纸塑复合袋等；无法判断填『无法判断』）", + "name": "商品全名（品牌+产品名+规格，如『大公鸡头管家 多功能油污净 625ml』，从包装 OCR 读出）", + "brand": "品牌名（从 Logo/包装文字读出，看不清填『无法判断』）", + "category": "商品品类（如『家用清洁/油污清洁剂』『日化/洗衣液』；非产品图填『非产品图』）", + "appearance": "外观特征（50-100字：瓶身形状、颜色、瓶盖、标签颜色、尺寸感）", + "packaging": "包装细节（50-100字：标签分区、图案元素、瓶盖/泵头样式、塑封状态）", + "text_on_package": ["包装上清晰可见的文字列表（品牌、产品名、卖点、规格等，看不清的不列）"], "key_features": [ - "3-6 条图片中确实能看到的外观特征/视觉卖点（如『按压式泵头设计』『瓶身有金色装饰线条』『罐身带塑封痕迹』等），不要写功效词（除非包装上明确印了）" + "3-5 条图片中能看到的外观/视觉特征（如『红色瓶盖白色瓶身』『鸡头图案 Logo』等）" ], - "visual_style": "整体视觉风格（如简约高端/粉嫩少女/国潮/科技感/家庭温馨/生活方式实拍等）", - "scene": "图片展示场景（如白底棚拍/浴室场景/户外街拍/桌面静物/手持实拍等；纯白底填『白底产品图』）", - "suitable_scenes": ["基于商品类型推断的 2-4 个适用/使用场景，如『家庭日常清洁』『卧室夜间驱蚊』"], - "target_audience": "从商品定位/包装风格推断的目标人群（如『家庭主妇/宝妈』『年轻租房群体』『男性商务人士』；不确定填『无法判断』）", - "selling_points": ["3-6 条可用于短视频营销的卖点（结合视觉+包装文字推断，尽量贴近电商话术）"], - "summary": "识别描述汇总：把上述各维度整合成一段 150-250 字的流畅中文自然段落，像电商详情页的商品介绍，口语化、有画面感，前端会直接展示这段文字" + "scene": "图片场景（如白底棚拍/浴室实拍/桌面静物/手持实拍等）", + "summary": "100-180字中文导购描述，连贯自然段落，像电商详情页介绍，前端直接展示，必须提到品牌/品名/核心外观特征，不能写『无法判断』" }""" -_IMAGE_ANALYSIS_USER_PROMPT = """请分析这张商品图片，按约定 JSON 字段完整输出。重点： -1. name/brand/spec/text_on_package 必须从图片包装上 OCR 读取，不要凭空编造； -2. appearance/packaging 两个字段要具体细致，80-150 字自然语言； -3. summary 字段务必连贯成一段 150-250 字的中文导购描述，方便前端直接展示； -4. 非产品图时 category 填『非产品图』，name 填实际看到的内容，其余字段按需填『无法判断』； -5. 所有无法判断的字段统一填「无法判断」。""" +_IMAGE_ANALYSIS_USER_PROMPT = """请分析这张商品图片，输出严格 JSON。重点： +1. name/brand/text_on_package 从图片包装 OCR 读取，不编造； +2. appearance/packaging 各写 50-100 字，要具体； +3. summary 必须是 100-180 字连贯中文段落，说清商品是什么、长什么样、适合谁用，不要写「无法判断」； +4. 非产品图时 category 填「非产品图」，name 填实际看到的内容； +5. 看不清的字段填「无法判断」。""" def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict: d = { "name": "未识别", "category": "无法判断", + "appearance": "无法判断", + "packaging": "无法判断", + "text_on_package": [], "key_features": [], "scene": "通用", + "summary": "", "_source": reason, } if extra: @@ -204,57 +197,112 @@ def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict: return d +def _is_vision_result_usable(result: dict) -> bool: + """判断 VLM 返回是否有效：name/summary 不能为未识别/无法判断/空，summary 要够长。""" + if not isinstance(result, dict): + return False + name = (result.get("name") or "").strip() + if not name or name in ("未识别", "无法判断", "未知"): + return False + summary = (result.get("summary") or "").strip() + if len(summary) < 30 or summary in ("无法判断", "未识别"): + return False + category = (result.get("category") or "").strip() + if category == "非产品图": + return True + feats = result.get("key_features") or [] + if not isinstance(feats, list) or len(feats) == 0: + return False + return True + + def _analyze_single_image( idx: int, img_url: str, vision_model: str, timeout: int, + *, + pro_fallback_model: str | None = None, ) -> dict: - """单张图片 VLM 分析（线程池并行调用）。失败/None 返回 fallback dict。""" + """单张图片 VLM 分析（线程池并行调用）。 + - lite 失败/结果不可用 时自动用 pro 模型降级重试 1 次。 + - 失败/None/不可用最终返回含默认字段的 dict（不会让用户看到「未识别·无法判断」裸结果）。 + """ from packages.shared.ai_service import call_vision if not img_url or not isinstance(img_url, str): return _vision_fallback(idx, "invalid_url") - try: - result = call_vision( - image_url=img_url, - prompt=_IMAGE_ANALYSIS_USER_PROMPT, - model=vision_model, - max_tokens=1024, - temperature=0.2, - timeout=timeout, - system_prompt=_IMAGE_ANALYSIS_SYSTEM_PROMPT, + + def _call(model: str, tmo: int): + try: + return call_vision( + image_url=img_url, + prompt=_IMAGE_ANALYSIS_USER_PROMPT, + model=model, + max_tokens=800, + temperature=0.1, + timeout=tmo, + system_prompt=_IMAGE_ANALYSIS_SYSTEM_PROMPT, + ) + except Exception as e: + logger.warning("[爆款视频] 图片 #%d call_vision(%s) 异常 err=%s", idx, model, e) + return None + + def _normalize(raw, source: str) -> dict: + if raw is None: + return _vision_fallback(idx, f"{source}_none") + if isinstance(raw, str): + logger.warning("[爆款视频] 图片 #%d VLM(%s) 返回非 JSON: %s", idx, source, raw[:200]) + return _vision_fallback(idx, f"{source}_text", {"_raw": raw[:500]}) + if not isinstance(raw, dict): + return _vision_fallback(idx, f"{source}_badtype") + raw.setdefault("_source", source) + raw.setdefault("name", "未识别") + raw.setdefault("brand", "无法判断") + raw.setdefault("category", "无法判断") + raw.setdefault("appearance", "无法判断") + raw.setdefault("packaging", "无法判断") + raw.setdefault("text_on_package", []) + raw.setdefault("key_features", []) + raw.setdefault("scene", "通用") + raw.setdefault("summary", "") + if not isinstance(raw.get("text_on_package"), list): + raw["text_on_package"] = [] + if not isinstance(raw.get("key_features"), list): + raw["key_features"] = [] + return raw + + # 第一次：传入模型（通常是 lite） + first_raw = _call(vision_model, timeout) + tag1 = vision_model.split("/")[-1] if "/" in vision_model else vision_model + first_result = _normalize(first_raw, tag1) + if _is_vision_result_usable(first_result): + return first_result + + logger.warning( + "[爆款视频] 图片 #%d VLM(%s) 结果不可用 name=%r summary_len=%d，尝试 pro 降级", + idx, + vision_model, + first_result.get("name"), + len(first_result.get("summary") or ""), + ) + + # 第二次：pro 降级重试 + if pro_fallback_model and pro_fallback_model != vision_model: + pro_raw = _call(pro_fallback_model, max(60, timeout)) + pro_result = _normalize(pro_raw, "pro_fallback") + if _is_vision_result_usable(pro_result): + pro_result["_fallback_used"] = True + return pro_result + logger.warning( + "[爆款视频] 图片 #%d pro 降级仍不可用 name=%r summary_len=%d", + idx, + pro_result.get("name"), + len(pro_result.get("summary") or ""), ) - if result is None: - logger.warning("[爆款视频] 图片 #%d call_vision 返回 None", idx) - return _vision_fallback(idx, "vision_none") - if isinstance(result, str): - logger.warning("[爆款视频] 图片 #%d VLM 返回非 JSON: %s", idx, result[:200]) - return _vision_fallback(idx, "vision_text", {"_raw": result[:500]}) - if isinstance(result, dict): - result.setdefault("_source", "vision") - result.setdefault("name", "未识别") - result.setdefault("brand", "无法判断") - result.setdefault("category", "无法判断") - result.setdefault("spec", "无法判断") - result.setdefault("appearance", "无法判断") - result.setdefault("packaging", "无法判断") - result.setdefault("text_on_package", []) - result.setdefault("colors", []) - result.setdefault("material_or_texture", "无法判断") - result.setdefault("key_features", []) - result.setdefault("visual_style", "通用") - result.setdefault("scene", "白底产品图") - result.setdefault("suitable_scenes", []) - result.setdefault("target_audience", "无法判断") - result.setdefault("selling_points", []) - result.setdefault("summary", "") - return result - logger.warning("[爆款视频] 图片 #%d VLM 返回意外类型 %s", idx, type(result)) - return _vision_fallback(idx, "vision_unexpected_type") - except Exception as e: - logger.warning("[爆款视频] 图片 #%d 分析失败 err=%s", idx, e, exc_info=True) - return _vision_fallback(idx, "vision_exception", {"_error": str(e)[:200]}) + return pro_result + + return first_result def _step_image_analysis(job: ViralVideoJob) -> dict: @@ -269,27 +317,37 @@ def _step_image_analysis(job: ViralVideoJob) -> dict: logger.warning("[爆款视频] 任务无 images，跳过图片分析") return {"products": []} - # 选择视觉模型：lite 速度优先（doubao_vision_use_lite=True 默认），pro 备用 + # 选择视觉模型：lite 速度优先（默认），pro 作为降级备用 try: _s = get_shared_settings() - vision_model = _s.doubao_vision_lite_model if _s.doubao_vision_use_lite else _s.doubao_vision_model - vision_timeout = 30 if _s.doubao_vision_use_lite else 60 + if _s.doubao_vision_use_lite: + vision_model = _s.doubao_vision_lite_model + pro_model = _s.doubao_vision_model + vision_timeout = 25 + else: + vision_model = _s.doubao_vision_model + pro_model = None # 已经是 pro，不再降级 + vision_timeout = 60 except Exception: vision_model = "doubao-1-5-vision-lite-250915" - vision_timeout = 30 + pro_model = "doubao-1-5-vision-pro-250915" + vision_timeout = 25 results: list[dict] = [None] * len(job.images) # type: ignore max_workers = min(4, max(1, len(job.images))) logger.info( - "[爆款视频] 开始并行图片分析 n=%d model=%s timeout=%d workers=%d", + "[爆款视频] 开始并行图片分析 n=%d model=%s pro_fallback=%s timeout=%d workers=%d", len(job.images), vision_model, + pro_model, vision_timeout, max_workers, ) with ThreadPoolExecutor(max_workers=max_workers) as pool: future_to_idx = { - pool.submit(_analyze_single_image, idx, url, vision_model, vision_timeout): idx + pool.submit( + _analyze_single_image, idx, url, vision_model, vision_timeout, pro_fallback_model=pro_model + ): idx for idx, url in enumerate(job.images) } for fut in as_completed(future_to_idx): @@ -390,6 +448,32 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: # ── 编导分镜脚本生成（核心，v1.6 新 prompt） ────────────────────────────── +# 人设 IP 类型 → 文案/出镜风格指导（前端下拉 10 个 IP 类型） +_PERSONA_STYLE_GUIDE = { + "通用个人IP": "亲切自然、像朋友分享好物，第一人称口语化，不端着", + "老板型IP": "沉稳大气、有行业格局感，适度使用『我做了XX年』『我一直坚持』等老板视角，语气自信不夸张", + "专家型IP": "专业权威、讲原理和数据支撑，用词严谨，少用网梗，像行业专家做科普", + "顾问型IP": "贴心周到、给建议给方案，多用『建议你』『可以试试』『我帮你梳理』", + "创始人IP": "真诚有温度、讲品牌故事和创业初心，带点情怀和个人观点，不端老板架子", + "创业者IP": "真实接地气、讲踩坑经验和创业心路，带点自嘲和韧劲，像身边的创业者朋友", + "从业者经验派": "内行视角、讲行业内幕/实操经验/踩坑教训，多用『干了X年我发现』『内行都知道』", + "避坑顾问型": "直接点出痛点和雷区，先讲『别买XX』『很多人踩过的坑』再给正确选择，节奏感强", + "知识科普型": "清晰讲原理、讲知识点，条理分明、信息密度高，像做一期小科普", + "测评种草型": "真实测评感、讲使用体验和优缺点对比，带『亲测』『我用了XX天』『实测下来』真实感词汇", +} + + +def _persona_style_hint(persona_id: str) -> str: + """根据 persona_id 查文案风格指导；未命中/空值返回通用提示。""" + pid = (persona_id or "").strip() + if pid in _PERSONA_STYLE_GUIDE: + return f"【人设风格：{pid}】{_PERSONA_STYLE_GUIDE[pid]}" + if pid: + # 前端传了自由值，照直提示，不阻塞 + return f"【人设风格：{pid}】按该人设的口吻、话术习惯组织口播和出镜动作" + return "【人设风格：未指定】亲切自然、像朋友分享好物" + + _SCRIPT_GENERATION_PROMPT = """你是资深短视频导演，为 Seedance 2.5（单次生成最多{duration}秒）写编导分镜脚本。脚本将整体作为 prompt 一次性传给视频模型，必须让模型在连贯镜头流中清楚每段时间拍什么、画面如何、人物说什么。 ## 产品 @@ -404,6 +488,7 @@ _SCRIPT_GENERATION_PROMPT = """你是资深短视频导演，为 Seedance 2.5（ - 时长：{duration}秒 / 画幅：{ratio} / 产品图：{n_images}张（第1张通常是主图/首帧） - 风格参考：{style_hint} - 爆款结构（必须严格遵循节奏/段落顺序）：{viral_structure_block} +- 人设/出镜口吻（必须贯穿全部对白和动作描写）：{persona_hint} ## 输出格式（必须输出严格 JSON，不要 Markdown，不要解释，字段一个都不能少） @@ -710,6 +795,7 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di else: viral_structure_block = "未指定（自由编排，但仍需有钩子开头+产品展示+行动号召的基本节奏）" + persona_hint = _persona_style_hint(getattr(job, "persona_id", "")) prompt = _SCRIPT_GENERATION_PROMPT.format( products_summary=products_summary, intent=intent_str, @@ -723,6 +809,7 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di style_hint=style_hint, approx_chars=approx_chars, viral_structure_block=viral_structure_block, + persona_hint=persona_hint, ) _s = get_shared_settings() @@ -777,7 +864,7 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str): from apps.worker.services.tts_service_factory import get_tts_service tts_service = get_tts_service() - voice_id = (getattr(job, "voice_id", "") or job.persona_id or "").strip() + voice_id = (getattr(job, "voice_id", "") or "").strip() text = (voiceover_script or "").strip() if not text: logger.warning("[爆款视频] voiceover_script 为空，跳过 TTS")