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

fix(vision-v2): 恢复后台prompt配置读取，用户自定义提示词生效 #2211

Merged
xiaoxia merged 1 commits from fix/vision-v2-db-prompt into develop 2026-10-05 22:30:45 +08:00
Conversation 2 Commits 1 Files Changed 3
+302 -141
3 changed files with 302 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/worker_app/tasks/vision/_prompt.py
+153
View File
@@ -0,0 +1,153 @@
# -*- coding: utf-8 -*-
"""V2 prompt 解析：优先读后台 viral_video_prompt_templates 配置，30s TTL 热加载；
DB 不可用/读到默认XML模板时，fallback 到硬编码 JSON schema prompt。
"""
from __future__ import annotations
import logging
import threading
import time
from typing import Any
logger = logging.getLogger(__name__)
# ---- 默认硬编码 prompt（DB 不可用或读到默认XML模板时使用） ----
DEFAULT_FAST_SYSTEM = (
"你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
"不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
"{\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_color": "下装主色",\n'
' "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
' "accessories": ["眼镜"/"帽子"/"项链"/"耳环"/"背包"/"手表"等数组],\n'
' "hairstyle": "发型，如短发/长发/马尾/卷发/丸子头/光头等",\n'
' "expression": "表情，如微笑/严肃/酷/开心等",\n'
' "pose": "姿势，如站立/坐姿/侧身/行走等",\n'
' "scene": "场景，如室内/街拍/户外/办公室/家居/海边/雪景/森林等",\n'
' "style": "风格，如休闲/商务/运动/复古/潮流/甜美/酷飒/优雅/街头/法式等",\n'
' "has_product": true/false,\n'
' "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
' "product_name": "产品名称，非产品图填null",\n'
' "brand": "品牌或文字标识，无则null",\n'
' "material": "材质，如棉质/牛仔/皮革/真丝/针织/涤纶等",\n'
' "pattern": "图案，如纯色/条纹/波点/格子/印花/碎花/Logo等",\n'
' "colors": ["主色数组"],\n'
' "mood": "整体氛围/情绪，如清新/活力/高级/温暖/冷峻/甜美/复古等"\n'
"}"
)
DEFAULT_FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
DEFAULT_PRO_SYSTEM = (
"你是图片分析专家。严格按下方 JSON schema 返回一个对象，不要解释、不要markdown、不要代码块。\n"
"{\n"
' "has_person": true/false,\n'
' "gender": "男"/"女"/null,\n'
' "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
' "outfit": "整体穿着描述（含颜色款式）",\n'
' "hair": "发型发色",\n'
' "pose": "姿势",\n'
' "expression": "表情",\n'
' "scene": "场景",\n'
' "mood": "氛围",\n'
' "has_product": true/false,\n'
' "category": "类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
' "product_name": "产品名，非产品图填null",\n'
' "brand": "品牌，无则null",\n'
' "key_features": ["核心特征数组，3-6个短语"]\n'
"}"
)
DEFAULT_PRO_USER = "分析这张图片，返回符合schema的JSON。"
# 用户自定义 prompt 末尾追加的硬约束
_JSON_TAIL_FAST = "\n\n你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释或markdown。"
_JSON_TAIL_PRO = "\n\n你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释或markdown。"
# 默认模板特征头（用于识别是否是内置XML模板）
_DEFAULT_XML_MARKER = "你是电商商品视觉分析师"
_cache_lock = threading.Lock()
_cache: dict[str, tuple[float, Any]] = {}
_CACHE_TTL = 30.0
def _load_template() -> Any | None:
"""从DB读image_analysis模板，失败返回None。"""
try:
from packages.application.viral_video.prompt_loader import get_template
return get_template("image_analysis")
except Exception as e:
logger.warning("[vision.v2] 读取后台prompt配置失败: %s", e)
return None
def _is_custom(template: Any) -> bool:
"""判断读到的模板是不是用户自定义的（不是内置默认XML长prompt）。"""
if not template:
return False
sp = getattr(template, "system_prompt", "") or ""
# 默认模板开头是"你是电商商品视觉分析师..."，XML格式，不适合qwen+JSON
if _DEFAULT_XML_MARKER in sp[:30]:
return False
# 其他有内容的system_prompt视为用户自定义
return bool(sp.strip())
def resolve_fast_prompt() -> tuple[str, str]:
"""返回 (system_prompt, user_prompt) 给 qwen3.8-flash fast 路径。"""
return _resolve("fast")
def resolve_pro_prompt() -> tuple[str, str]:
"""返回 (system_prompt, user_prompt) 给 qwen3.7-plus fallback 路径。"""
return _resolve("pro")
def _resolve(kind: str) -> tuple[str, str]:
now = time.time()
cache_key = f"prompt_{kind}"
with _cache_lock:
hit = _cache.get(cache_key)
if hit and now - hit[0] < _CACHE_TTL:
return hit[1]
default_sys = DEFAULT_FAST_SYSTEM if kind == "fast" else DEFAULT_PRO_SYSTEM
default_user = DEFAULT_FAST_USER if kind == "fast" else DEFAULT_PRO_USER
tail = _JSON_TAIL_FAST if kind == "fast" else _JSON_TAIL_PRO
sys_prompt = default_sys
usr_prompt = default_user
try:
tpl = _load_template()
if _is_custom(tpl):
custom_sys = (getattr(tpl, "system_prompt", "") or "").strip()
custom_user_tpl = getattr(tpl, "user_prompt_template", "") or ""
if custom_sys:
sys_prompt = custom_sys + tail
if custom_user_tpl:
# V2 是单图调用，简单替换几个常用占位符；缺键原样保留
usr_prompt = (custom_user_tpl
.replace("{image_count}", "1")
.replace("{industry}", "通用")
.replace("{image_urls}", "")
.strip())
if not usr_prompt:
usr_prompt = default_user
logger.info("[vision.v2] 使用后台自定义prompt (kind=%s version=%s)",
kind, getattr(tpl, "version", "?"))
except Exception as e:
logger.warning("[vision.v2] 解析后台prompt失败，使用默认: %s", e)
with _cache_lock:
_cache[cache_key] = (now, (sys_prompt, usr_prompt))
return sys_prompt, usr_prompt
def invalidate_cache() -> None:
with _cache_lock:
_cache.clear()
apps/worker/worker_app/tasks/vision/vlm_fallback.py
+137 -93
View File
@@ -1,8 +1,14 @@
# -*- coding: utf-8 -*-
"""V2 pro 兜底：qwen3.7-plus（阿里云百炼/DashScope）单次调用。
"""V2 兜底路径：qwen3.7-plus（阿里云百炼/DashScope）单图调用。
fast_json 结果不可用时单次调用，无竞速、无重试、无复杂超时逻辑。
直接 httpx 发精简 JSON-only prompt（比旧版 prompt_loader XML 模板短很多，降低延迟）。
fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
设计要点：
- 直接 httpx 直连 DashScope，不走 ai_client
- enable_thinking=false + response_format=json_object
- system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
- timeout=25s
- API Key 从环境变量 DASHSCOPE_API_KEY 读取
- 返回 dict 字段与旧 _normalize() 兼容，下游零改动
"""
from __future__ import annotations
@@ -13,6 +19,8 @@ import os
import time
from typing import Any
from . import _prompt
logger = logging.getLogger(__name__)
_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
@@ -20,65 +28,87 @@ _PRO_MODEL = "qwen3.7-plus"
_DEFAULT_TIMEOUT = 25
_DEFAULT_MAX_TOKENS = 800
_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": "产品名称，非产品图填null",\n'
' "brand": "品牌或文字标识，无则null",\n'
' "key_features": ["特征数组"]\n'
"}"
)
_PRO_USER = "分析这张图片，返回符合schema的JSON。"
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
def _api_key() -> str | None:
return os.environ.get("DASHSCOPE_API_KEY")
def _assemble_pp(obj: dict[str, Any]) -> str:
if not obj.get("has_person", False):
return "无人像"
"""从 JSON 字段组装 portrait_prompt（60-100字人物穿搭描述，给 Seedream t2i 用）。"""
if not obj.get("has_person"):
name = obj.get("product_name") or "商品"
brand = obj.get("brand") or ""
kf = obj.get("key_features") or []
scene = obj.get("scene") or ""
mood = obj.get("mood") or ""
outfit = obj.get("outfit") or ""
if outfit:
return outfit
pieces = []
if brand:
pieces.append(brand)
pieces.append(str(name))
if isinstance(kf, list):
pieces.extend(str(x) for x in kf[:2] if x)
if mood:
pieces.append(str(mood) + "氛围")
if scene:
pieces.append(str(scene) + "场景")
pieces.append("产品特写")
p = "，".join(x for x in pieces if x)
return p if len(p) >= 10 else "产品展示图，特写镜头"
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")
gender = obj.get("gender") or ""
age = obj.get("age_range") or ""
subj = ""
if age == "儿童":
subj = "小女孩" if gender == "女" else ("小男孩" if gender == "男" else "儿童")
elif age == "青少年":
subj = "少女" if gender == "女" else ("少年" if gender == "男" else "青少年")
else:
prefix_map = {"青年": "年轻", "中年": "中年", "老年": "老年"}
gw = {"男": "男性", "女": "女性"}.get(gender, "")
prefix = prefix_map.get(age, "")
subj = (prefix + gw) if (prefix or gw) else "人物"
parts.append(f"一位{subj}")
outfit = obj.get("outfit") or ""
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 "无人像"
hair = obj.get("hair") or ""
if hair:
parts.append(str(hair))
pose = obj.get("pose") or ""
expr = obj.get("expression") or ""
det = []
if expr and expr not in ("自然", "平静"):
det.append(f"神情{expr}")
if pose and pose not in ("站立",):
det.append(str(pose))
if det:
parts.append("，".join(det))
style_parts = []
mood = obj.get("mood") or ""
scene = obj.get("scene") or ""
if mood:
style_parts.append(str(mood))
if scene and scene != "通用":
style_parts.append(str(scene))
if style_parts:
parts.append("".join(style_parts) + "风格")
else:
parts.append("人像写真")
full = "，".join(p for p in parts if p)
if len(full) < 40:
full += "，自然光线下人像特写，画面清晰"
if len(full) > 120:
full = full[:120].rstrip("，") + "。"
return full
def call_pro_vlm(
@@ -87,27 +117,24 @@ def call_pro_vlm(
*,
timeout: int = _DEFAULT_TIMEOUT,
) -> dict[str, Any] | None:
"""单次调用 qwen3.7-plus，解析后返回 product dict；失败返回 None。"""
t0 = time.time()
import httpx
api_key = os.environ.get("DASHSCOPE_API_KEY")
api_key = _api_key()
if not api_key:
logger.warning("[vision.v2] DASHSCOPE_API_KEY 未配置，跳过 pro 兜底")
logger.warning("[vision.v2] pro DASHSCOPE_API_KEY 未配置，跳过")
return None
url = f"{_BASE_URL}/chat/completions"
system_prompt, user_prompt = _prompt.resolve_pro_prompt()
payload: dict[str, Any] = {
"model": _PRO_MODEL,
"messages": [
{"role": "system", "content": _PRO_SYSTEM},
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": img_url}},
{"type": "text", "text": _PRO_USER},
],
},
{"role": "system", "content": system_prompt},
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": img_url}},
{"type": "text", "text": user_prompt},
]},
],
"temperature": 0.3,
"max_tokens": _DEFAULT_MAX_TOKENS,
@@ -117,7 +144,7 @@ def call_pro_vlm(
}
try:
r = httpx.post(
url,
f"{_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=payload,
timeout=timeout,
@@ -132,44 +159,61 @@ def call_pro_vlm(
logger.warning("[vision.v2] pro 返回空 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] pro 完成 idx=%d model=%s elapsed=%.1fs in=%d out=%d",
idx,
_PRO_MODEL,
elapsed,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0),
"[vision.v2] pro 完成 model=%s elapsed=%.1fs in=%d out=%d reasoning=%d",
_PRO_MODEL, elapsed,
usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), reasoning_tokens,
)
text = _strip_code_fence(raw)
lpos, r_pos = text.find("{"), text.rfind("}")
if lpos < 0 or r_pos <= lpos:
logger.warning("[vision.v2] pro 无JSON elapsed=%.1fs head=%s", elapsed, raw[:200])
s = raw.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()
l, rr = s.find("{"), s.rfind("}")
if l >= 0 and rr > l:
s = s[l : rr + 1]
try:
obj = json.loads(s)
except json.JSONDecodeError:
logger.warning("[vision.v2] pro JSON 解析失败 head=%s", raw[:200])
return None
obj = json.loads(text[lpos : r_pos + 1])
if not isinstance(obj, dict):
return None
scene = obj.get("scene") or "通用"
mood = obj.get("mood") or ""
pp = _assemble_pp(obj)
has_person = obj.get("has_person", False)
has_product = obj.get("has_product", False)
kf = obj.get("key_features")
if not isinstance(kf, list):
kf = [str(kf)] if kf else ["无法判断"]
else:
kf = [str(x) for x in kf if x] or ["无法判断"]
name = obj.get("product_name") or "未识别"
if obj.get("has_person") and (not name or name == "未识别"):
name = obj.get("outfit") or "人物穿搭"
brand = obj.get("brand") or "无法判断"
category = obj.get("category") or ("非产品图" if has_person and not has_product else "无法判断")
category = obj.get("category") or ("服饰" if obj.get("has_person") else "非产品图")
return {
"name": name,
"brand": brand,
"category": category,
"appearance": obj.get("outfit") or "无法判断",
"name": str(name),
"brand": str(brand),
"category": str(category),
"appearance": str(obj.get("outfit") or "无法判断"),
"packaging": "无法判断",
"text_on_package": [],
"key_features": obj.get("key_features") or ["无法判断"],
"scene": scene,
"mood": mood,
"key_features": kf[:6],
"scene": str(obj.get("scene") or "通用"),
"mood": str(obj.get("mood") or ""),
"portrait_prompt": pp,
"summary": f"{brand} {name}" if name != "未识别" else "未识别",
"summary": str(name),
"_source": "vlm_pro",
}
except Exception as e:
logger.warning("[vision.v2] pro 异常 idx=%d elapsed=%.1fs err=%s", idx, time.time() - t0, e, exc_info=True)
elapsed = time.time() - t0
logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
return None
apps/worker/worker_app/tasks/vision/vlm_fast_json.py
+12 -48
View File
@@ -5,7 +5,8 @@
设计要点：
- 直接用 httpx 发最小 payload 到 DashScope OpenAI 兼容 endpoint，不走 ai_client 包装
- enable_thinking=false 关闭推理链（reasoning 是延迟主因）
- system prompt 极致精简，只给字段 schema 和强约束（禁止自然语言、禁止 markdown）
- response_format=json_object 强约束JSON输出
- system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
- max_tokens=350、temperature=0.1（稳定输出 JSON）
- timeout=12s（失败由外层走 pro 兜底）
- API Key 从环境变量 DASHSCOPE_API_KEY 读取
@@ -19,6 +20,8 @@ import os
import time
from typing import Any
from . import _prompt
logger = logging.getLogger(__name__)
# DashScope OpenAI 兼容 endpoint
@@ -27,38 +30,6 @@ _FAST_MODEL = "qwen3.8-flash"
_DEFAULT_TIMEOUT = 12
_DEFAULT_MAX_TOKENS = 350
# 极简 system prompt：只给字段定义 + 硬性输出要求
_FAST_SYSTEM = (
"你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
"不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
"{\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_color": "下装主色",\n'
' "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
' "accessories": ["眼镜"/"帽子"/"项链"/"耳环"/"背包"/"手表"等数组],\n'
' "hairstyle": "发型，如短发/长发/马尾/卷发/丸子头/光头等",\n'
' "expression": "表情，如微笑/严肃/酷/开心等",\n'
' "pose": "姿势，如站立/坐姿/侧身/行走等",\n'
' "scene": "场景，如室内/街拍/户外/办公室/家居/海边/雪景/森林等",\n'
' "style": "风格，如休闲/商务/运动/复古/潮流/甜美/酷飒/优雅/街头/法式等",\n'
' "has_product": true/false,\n'
' "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
' "product_name": "产品名称，非产品图填null",\n'
' "brand": "品牌或文字标识，无则null",\n'
' "material": "材质，如棉质/牛仔/皮革/真丝/针织/涤纶等",\n'
' "pattern": "图案，如纯色/条纹/波点/格子/印花/碎花/Logo等",\n'
' "colors": ["主色数组"],\n'
' "mood": "整体氛围/情绪，如清新/活力/高级/温暖/冷峻/甜美/复古等"\n'
"}"
)
_FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
def _api_key() -> str | None:
return os.environ.get("DASHSCOPE_API_KEY")
@@ -91,16 +62,18 @@ def call_fast_json(
logger.warning("[vision.v2] DASHSCOPE_API_KEY 未配置，跳过 fast_json")
return None
system_prompt, user_prompt = _prompt.resolve_fast_prompt()
url = f"{_BASE_URL}/chat/completions"
payload: dict[str, Any] = {
"model": _FAST_MODEL,
"messages": [
{"role": "system", "content": _FAST_SYSTEM},
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": img_url}},
{"type": "text", "text": _FAST_USER},
{"type": "text", "text": user_prompt},
],
},
],
@@ -119,7 +92,6 @@ def call_fast_json(
)
elapsed = time.time() - t0
if resp.status_code == 400 and "enable_thinking" in resp.text[:300].lower():
# 极少数 endpoint 版本不识别 enable_thinking，重试一次不带
logger.warning("[vision.v2] fast_json HTTP 400 thinking 参数不兼容，重试 elapsed=%.1fs", elapsed)
payload.pop("enable_thinking", None)
resp = httpx.post(
@@ -130,9 +102,7 @@ def call_fast_json(
)
elapsed = time.time() - t0
if 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 elapsed=%.1fs body=%s", resp.status_code, elapsed, resp.text[:200])
return None
data = resp.json()
raw = (data.get("choices") or [{}])[0].get("message", {}).get("content")
@@ -146,11 +116,8 @@ def call_fast_json(
reasoning_tokens = ctd.get("reasoning_tokens", 0)
logger.info(
"[vision.v2] fast_json 完成 model=%s elapsed=%.1fs in=%d out=%d reasoning=%d",
_FAST_MODEL,
elapsed,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0),
reasoning_tokens,
_FAST_MODEL, elapsed,
usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), reasoning_tokens,
)
text = _strip_code_fence(raw)
lpos, r = text.find("{"), text.rfind("}")
@@ -166,10 +133,7 @@ def call_fast_json(
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"),
elapsed, obj.get("has_person"), obj.get("has_product"), obj.get("category"),
)
return obj
except Exception as e: