xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 16 Pull Requests 10 Actions 34.2k Packages Projects Releases 7 Wiki Activity

fix(viral-video): VLM lite失败自动降级pro + 10种IP人设注入prompt (#2139) #2140

Merged
auto-approve-bot merged 2 commits from fix/2139-vlm-persona-fixes into develop 2026-10-02 12:32:03 +08:00
Conversation 2 Commits 2 Files Changed 1
+164 -77
1 changed files with 164 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/worker_app/tasks/viral_video.py
+164 -77
View File
@@ -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")