From ed20fbad49f12517d45ab2af99b2095bb666076d Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 5 Oct 2026 17:06:05 +0800 Subject: [PATCH] =?UTF-8?q?feat(vision):=20V2=20=E5=BF=AB=E9=80=9F?= =?UTF-8?q?=E5=9B=BE=E7=89=87=E5=88=86=E6=9E=90=E8=B7=AF=E5=BE=84=20?= =?UTF-8?q?=E2=80=94=20OCR+lite=20JSON=20VLM=E5=B9=B6=E8=A1=8C=EF=BC=8C?= =?UTF-8?q?=E5=8D=95=E5=9B=BE<3s=E7=9B=AE=E6=A0=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 灵应直接指令(10-05)：图片分析专用API组合方案直接干，不要等方案确认。 API现实说明： - 火山引擎云端真实可用的视觉专用HTTP API：OCR（MediaKit tools-sync/ocr，Bearer鉴权） - 人体属性/商品检测/图像标签：火山云端无公开HTTP API，仅有移动端SDK（智能美化特效，年费6-60万） - 务实方案：OCR专用API + doubao-seed-2.1-lite强约束JSON-only prompt（替代3类缺失的专用API）， pro VLM保留为终极兜底 架构： - 新模块 apps/worker/worker_app/tasks/vision/： - vlm_fast_json.py：lite VLM极简JSON schema prompt，max_tokens=350，temp=0.1，timeout=8s - ocr_volc.py：MediaKit同步OCR封装，返回文本列表 - assembler.py：字段映射+portrait_prompt模板拼接，输出格式与旧_normalize完全一致 - fast_path.py：analyze_image_v2/analyze_images_v2，单图2路并行(OCR+lite JSON)， 外层8图全并发，置信度低/失败自动降级旧lite/pro竞速VLM - _step_image_analysis增加VISION_V2_ENABLED环境变量开关： - true→走V2快速路径 - false(默认)→走V1 #2198/#2199 lite/pro竞速路径（过渡期兜底） - 下游信任链/t2i零改动：输出dict字段(name/brand/category/appearance/key_features/ scene/mood/portrait_prompt/summary/_source)与旧格式完全兼容 性能目标： - 单图fast路径目标<3s（OCR+lite JSON并行取最慢） - 8图全并发<15s（较当前V1的~115s/3图提升10倍+） - 兜底路径仍复用现有V1竞速，最坏情况不劣化 --- apps/worker/worker_app/tasks/viral_video.py | 61 ++-- .../worker_app/tasks/vision/__init__.py | 15 + .../worker_app/tasks/vision/assembler.py | 297 ++++++++++++++++++ .../worker_app/tasks/vision/fast_path.py | 242 ++++++++++++++ .../worker_app/tasks/vision/ocr_volc.py | 108 +++++++ .../worker_app/tasks/vision/vlm_fast_json.py | 141 +++++++++ 6 files changed, 845 insertions(+), 19 deletions(-) create mode 100644 apps/worker/worker_app/tasks/vision/__init__.py create mode 100644 apps/worker/worker_app/tasks/vision/assembler.py create mode 100644 apps/worker/worker_app/tasks/vision/fast_path.py create mode 100644 apps/worker/worker_app/tasks/vision/ocr_volc.py create mode 100644 apps/worker/worker_app/tasks/vision/vlm_fast_json.py diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index f261717ba..1cac84445 100644 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -850,16 +850,18 @@ def _analyze_single_image( def _step_image_analysis(job: ViralVideoJob) -> dict: - """步骤 1: 图片 VLM 分析 — 识别产品特征（v1.6/#2198 优化：lite/pro 并行竞速）。 - #2188/#2194/#2198: (1) 所有图片 URL 先归一化（storage_key→公网URL+空值报400） - (2) 爆款视频强制 lite-first，不依赖 .env USE_LITE 开关 - (3) max_tokens=1200，max_workers=min(2,n) 防方舟限流（竞速模式总并发=4） - (4) lite/pro 并行竞速：单张图同时发 lite(30s) 和 pro(75s)， - 谁先返回 usable 结果就用谁。单图最坏 75s（pro慢），典型 40-50s， - 3图2并发最坏约75s，比原串行 lite→pro 240s 改善70%+ - (5) 整个阶段统一关闭底层 httpx 重试（外层 max_retries=0，finally 恢复）， - 子线程只读不改 client 属性避免竞态 - (6) 每张图 VLM 调用结束打印 elapsed 耗时日志便于排查 + """步骤 1: 图片分析。 + + V2（VISION_V2_ENABLED=true，10-05 新方案）: + - 每图并行 2 路：火山 MediaKit OCR（专用API）+ doubao-seed-2.1-lite 强约束 JSON + （火山云端无人体属性/商品检测/图像标签公开 HTTP API，用 lite JSON-only VLM 弥补）， + 目标单图 <3s； + - 外层 8 图全并发，目标 8 图 <15s； + - 置信度低/全失败时降级 doubao-seed-2.1-pro 完整 VLM 兜底（复用旧竞速逻辑）； + - 输出 dict 格式与旧 _normalize() 完全一致，下游信任链/t2i 零改动。 + + V1（默认，#2198/#2199 lite/pro 并行竞速）: + - 过渡版兜底，单图 lite(30s)/pro(75s) 竞速，外层 max_workers=2，典型 40-75s/图。 """ try: from packages.shared.ai_service import call_vision # noqa: F401 @@ -871,17 +873,16 @@ def _step_image_analysis(job: ViralVideoJob) -> dict: logger.warning("[爆款视频] 任务无 images，跳过图片分析") return {"products": []} - # #2188 BUG1: URL 归一化 — storage_key→公网URL + 空值报400 + # URL 归一化 — storage_key→公网URL + 空值报400 normalized_urls: list[str] = [] for idx, raw in enumerate(job.images): try: normalized_urls.append(_normalize_image_url(raw, idx)) except ValueError as _ve: - # 空/非法URL：直接让任务失败，不默默走 fallback logger.error("[爆款视频] 图片 #%d URL 归一化失败: %s", idx, _ve) - raise # 上层 celery 捕获后标记任务失败，避免"未识别·无法判断"误导 + raise - # #2188/#2198 BUG2: 爆款视频强制 lite-first（不依赖 .env 开关），lite/pro 并行竞速 + # 模型配置（V1/V2 共用） try: _s = get_shared_settings() lite_model = _s.doubao_vision_lite_model @@ -889,13 +890,35 @@ def _step_image_analysis(job: ViralVideoJob) -> dict: except Exception: lite_model = "doubao-seed-2-1-lite-260915" pro_model = "doubao-seed-2-1-pro-260915" + + # ========== V2 路径（VISION_V2_ENABLED=true）========== + import os as _os_v2 + _v2_enabled = _os_v2.environ.get("VISION_V2_ENABLED", "false").lower() in ("1", "true", "yes", "on") + if _v2_enabled: + try: + from worker_app.tasks.vision import analyze_images_v2 as _aiv2 + except ImportError: + try: + from tasks.vision import analyze_images_v2 as _aiv2 # type: ignore + except ImportError: + logger.warning("[vision.v2] 模块导入失败，回退 V1 路径") + _aiv2 = None # type: ignore + if _aiv2 is not None: + from packages.shared.ai_client import get_doubao_client as _gdc_v2 + _cli = _gdc_v2() + _orig_retries_v2 = _cli.max_retries + _cli.max_retries = 0 + try: + results_v2 = _aiv2(normalized_urls, lite_model=lite_model, pro_model=pro_model) + finally: + _cli.max_retries = _orig_retries_v2 + return {"products": list(results_v2)} + # 导入失败 → fallthrough 走 V1 + + # ========== V1 路径（默认，lite/pro 并行竞速）========== vision_model = lite_model - # #2198: lite 单次 30s 封顶（竞速快速路径，30s 还没出就等 pro），pro 75s（在 _analyze_single_image - # 的内部竞速池里设置），外层不感知。单图最坏 75s（仅 pro 成功），典型 40-50s（pro 正常返回）。 vision_timeout = 30 - # #2194/#2198: 整个并行图片分析阶段统一把共享 client 的 max_retries 置 0， - # 阶段结束 finally 恢复。子线程 _call 只读不改，避免竞态。 from packages.shared.ai_client import get_doubao_client as _gdc_step _step_client = _gdc_step() @@ -905,7 +928,7 @@ def _step_image_analysis(job: ViralVideoJob) -> dict: results: list[dict] = [None] * len(normalized_urls) # type: ignore max_workers = min(2, max(1, len(normalized_urls))) # 并发≤2 防方舟限流（竞速模式下总并发=4） logger.info( - "[爆款视频] 开始并行竞速图片分析 n=%d lite=%s(%ds) pro=%s(75s) img_workers=%d", + "[爆款视频] 开始并行竞速图片分析(V1) n=%d lite=%s(%ds) pro=%s(75s) img_workers=%d", len(normalized_urls), vision_model, vision_timeout, diff --git a/apps/worker/worker_app/tasks/vision/__init__.py b/apps/worker/worker_app/tasks/vision/__init__.py new file mode 100644 index 000000000..2ea2eccab --- /dev/null +++ b/apps/worker/worker_app/tasks/vision/__init__.py @@ -0,0 +1,15 @@ +# -*- coding: utf-8 -*- +"""V2 图片分析：专用 API 组合路径（OCR + lite JSON VLM 并行 + pro VLM 兜底）。 + +灵应指令（10-05）： +- 优先火山引擎视觉智能 API：OCR 是真实专用云端 API； +- 人体属性/商品检测/图像标签：火山云端无公开 HTTP API（仅有移动端 SDK）， + 采用 doubao-seed-2.1-lite + 强约束 JSON-only prompt 作为"伪专用 API"， + 目标 1-3s 返回结构化字段； +- VLM（doubao-seed-2.1-pro）保留为终极兜底（置信度低/全失败时降级）； +- 单图并行 2 路（OCR + lite JSON VLM），外层 8 图全并发，目标 8 图 <15s。 + +输出 dict 格式与 viral_video._normalize() 完全一致，下游信任链/t2i 零改动。 +灰度开关：VISION_V2_ENABLED=true（默认 false，走旧 #2198/#2199 竞速逻辑）。 +""" +from .fast_path import analyze_image_v2, analyze_images_v2 # noqa: F401 diff --git a/apps/worker/worker_app/tasks/vision/assembler.py b/apps/worker/worker_app/tasks/vision/assembler.py new file mode 100644 index 000000000..bbed48800 --- /dev/null +++ b/apps/worker/worker_app/tasks/vision/assembler.py @@ -0,0 +1,297 @@ +# -*- coding: utf-8 -*- +"""把 fast_json VLM 输出 + OCR 文本组装为与旧 _normalize() 完全一致的 dict。 + +目标：下游（信任链t2i/intent_parsing/script_generation）零改动。 +必出字段：name, brand, category, appearance, packaging, text_on_package, + key_features, scene, mood, portrait_prompt, summary, _source +""" +from __future__ import annotations + +from typing import Any + +# ---------- portrait_prompt 模板 ---------- +# 目标：60-100 字的人物穿搭描述，用于 Seedream 纯文生图。要求具体、风格化、视觉细节丰富。 +# 旧 VLM 输出格式参考："一位25岁左右的亚洲女性，身穿白色V领短袖T恤，黑色高腰阔腿裤， +# 搭配银色项链，长发披肩，表情自信，街拍风格，阳光明媚的城市街头" + + +def _join_parts(*parts: str | None) -> str: + return "".join(p for p in parts if p) + + +_AGE_PREFIX = { + "儿童": "小女孩" if None else "儿童", + "青少年": "少女" if None else "少年", + "青年": "年轻", + "中年": "中年", + "老年": "老年", +} +# gender 后缀 +_GENDER_WORD = {"男": "男性", "女": "女性"} + + +def _person_subject(fj: dict[str, Any]) -> str: + """人物主语：年轻女性 / 中年男性 / 少女 / 小男孩 / 人物 等。""" + gender = fj.get("gender") or "" + age = fj.get("age_range") or "" + gw = _GENDER_WORD.get(gender, "") + if age == "儿童": + if gender == "女": + return "小女孩" + if gender == "男": + return "小男孩" + return "儿童" + if age == "青少年": + if gender == "女": + return "少女" + if gender == "男": + return "少年" + return "青少年" + prefix = _AGE_PREFIX.get(age, "") + if gw: + return f"{prefix}{gw}" if prefix else gw + return f"{prefix}人物" if prefix else "人物" + + +def _build_wear_sentence(fj: dict[str, Any]) -> str: + """穿搭段：上装+下装/连衣裙，带颜色+材质+图案。""" + upper = fj.get("upper_wear") or "" + upper_color = fj.get("upper_color") or "" + lower = fj.get("lower_wear") or "" + lower_color = fj.get("lower_color") or "" + dress_color = fj.get("dress_color") or "" + material = fj.get("material") or "" + pattern = fj.get("pattern") or "" + + is_dress = ("连衣裙" in upper) or ("裙" in upper and not lower) + if is_dress: + c = dress_color or upper_color + wear = f"{c}{upper}" if c else upper + if material and material not in wear: + wear = f"{material}{wear}" + if pattern and pattern not in wear and pattern != "纯色": + wear += f"，{pattern}图案" + return f"身穿{wear}" + + parts: list[str] = [] + if upper: + up = f"{upper_color}{upper}" if upper_color else upper + if material and material not in up: + up = f"{material}{up}" + if pattern and pattern != "纯色" and pattern not in up: + up += f"（{pattern}）" + parts.append(f"上身{up}" if up else "") + if lower: + lo = f"{lower_color}{lower}" if lower_color else lower + parts.append(f"下身{lo}" if lo else "") + return "，".join(p for p in parts if p) + + +def _build_portrait_prompt(fj: dict[str, Any]) -> str: + """组装最终 portrait_prompt（目标 60-100 字，用于 Seedream 纯文生图）。""" + if not fj.get("has_person"): + # 非人像：用商品+场景+mood 拼一段 + name = fj.get("product_name") or "商品" + brand = fj.get("brand") or "" + colors = fj.get("colors") or [] + style = fj.get("style") or "" + scene = fj.get("scene") or "" + mood = fj.get("mood") or "" + pieces = [] + if brand: + pieces.append(brand) + pieces.append(name) + if colors: + pieces.append("、".join(colors[:3]) + "配色") + if style: + pieces.append(style + "风格") + if mood: + pieces.append(mood + "氛围") + if scene and scene not in ("通用",): + pieces.append(scene + "场景") + pieces.append("产品特写") + prompt = "，".join(p for p in pieces if p) + return prompt if len(prompt) >= 10 else "产品展示图，特写镜头" + + subject = _person_subject(fj) + wear = _build_wear_sentence(fj) + + accessories = fj.get("accessories") or [] + if isinstance(accessories, str): + accessories = [accessories] + acc_str = "" + if accessories: + acc_str = "，佩戴" + "、".join(str(a) for a in accessories if a) + + hairstyle = fj.get("hairstyle") or "" + expression = fj.get("expression") or "" + pose = fj.get("pose") or "" + style = fj.get("style") or "" + scene = fj.get("scene") or "" + mood = fj.get("mood") or "" + + detail_parts: list[str] = [] + if hairstyle: + detail_parts.append(hairstyle) + if expression and expression not in ("自然", "平静"): + detail_parts.append(f"神情{expression}") + if pose and pose not in ("站立",): + detail_parts.append(pose) + + style_parts: list[str] = [] + if style: + style_parts.append(style) + if mood: + style_parts.append(mood) + if scene and scene not in ("通用",): + style_parts.append(scene) + + pieces = [f"一位{subject}"] + if wear: + pieces.append(wear) + if acc_str: + pieces.append(acc_str.lstrip("，")) + if detail_parts: + pieces.append("，".join(detail_parts)) + if style_parts: + pieces.append("，".join(style_parts) + "风格") + else: + pieces.append("人像写真") + + full = "，".join(p for p in pieces if p) + # 过短补充镜头词 + if len(full) < 40: + full += "，自然光线下人像特写，画面清晰" + # 过长截断 + if len(full) > 120: + full = full[:120].rstrip("，") + "。" + return full + + +# ---------- 商品字段 ---------- + +def _infer_name(fj: dict[str, Any], ocr_texts: list[str]) -> str: + pname = fj.get("product_name") + if pname and pname != "未识别": + return str(pname) + # 人物图 → name 用穿搭主件 + if fj.get("has_person"): + up = fj.get("upper_wear") or "" + if "连衣裙" in up: + return up + return up or "人物穿搭" + if ocr_texts: + # 商品名可能是 OCR 最长的一行（品牌/产品名） + return max(ocr_texts, key=len) + return "未识别" + + +def _infer_brand(fj: dict[str, Any], ocr_texts: list[str]) -> str: + brand = fj.get("brand") + if brand: + return str(brand) + # OCR 里短的、纯字母/汉字短串可能是 brand + for t in ocr_texts: + if 1 < len(t) <= 12: + return t + return "无法判断" + + +def _infer_category(fj: dict[str, Any]) -> str: + cat = fj.get("category") + if cat: + return str(cat) + if fj.get("has_person"): + return "服饰" + return "非产品图" + + +def _build_appearance(fj: dict[str, Any]) -> str: + """外观描述：颜色+款式+材质+图案 拼成一段。""" + parts: list[str] = [] + for key, label in [ + ("upper_color", "主色"), + ("upper_wear", "款式"), + ("material", "材质"), + ("pattern", "图案"), + ]: + v = fj.get(key) + if v and v not in ("无法判断", "未知", "纯色"): + parts.append(str(v)) + if not parts: + if fj.get("has_person"): + return "人像穿搭整体造型" + return "无法判断" + return "、".join(parts) + + +def _build_key_features(fj: dict[str, Any], ocr_texts: list[str]) -> list[str]: + feats: list[str] = [] + for key in ("upper_wear", "lower_wear", "upper_color", "lower_color", "dress_color", + "material", "pattern", "style", "accessories"): + v = fj.get(key) + if not v: + continue + if isinstance(v, list): + feats.extend(str(x) for x in v if x) + elif isinstance(v, str) and v not in ("无法判断", "未知", "纯色"): + feats.append(v) + if ocr_texts: + feats.append(f"画面文字: {'/'.join(ocr_texts[:3])}") + # 去重 + out: list[str] = [] + seen: set[str] = set() + for f in feats: + f = f.strip() + if f and f not in seen and len(f) <= 30: + seen.add(f) + out.append(f) + return out[:6] if out else ["无法判断"] + + +def assemble_result( + idx: int, + fast_json: dict[str, Any] | None, + ocr_texts: list[str], +) -> dict[str, Any]: + """把 fast_json 结果 + OCR 文本组装成下游兼容的 product dict。""" + fj = fast_json or {} + ocr_texts = ocr_texts or [] + + portrait_prompt = _build_portrait_prompt(fj) + name = _infer_name(fj, ocr_texts) + brand = _infer_brand(fj, ocr_texts) + category = _infer_category(fj) + appearance = _build_appearance(fj) + key_features = _build_key_features(fj, ocr_texts) + scene = fj.get("scene") or "通用" + mood = fj.get("mood") or "" + packaging = "无法判断" # 包装细节专用API无，保留占位 + text_on_package = ocr_texts[:8] + summary = _build_summary(fj, name, brand, category) + + return { + "name": name, + "brand": brand, + "category": category, + "appearance": appearance, + "packaging": packaging, + "text_on_package": text_on_package, + "key_features": key_features, + "scene": scene, + "mood": mood, + "portrait_prompt": portrait_prompt, + "summary": summary, + "_source": "v2_fast_json", + } + + +def _build_summary(fj: dict, name: str, brand: str, category: str) -> str: + if fj.get("has_person"): + up = fj.get("upper_wear") or "穿搭" + style = fj.get("style") or "" + base = f"{style}{up}" if style and style not in up else up + return base + if brand != "无法判断" and name != brand: + return f"{brand} {name}" + return name diff --git a/apps/worker/worker_app/tasks/vision/fast_path.py b/apps/worker/worker_app/tasks/vision/fast_path.py new file mode 100644 index 000000000..5bd3f3d2c --- /dev/null +++ b/apps/worker/worker_app/tasks/vision/fast_path.py @@ -0,0 +1,242 @@ +# -*- coding: utf-8 -*- +"""V2 快速路径：每图并行 OCR + lite JSON VLM，失败降级 pro VLM。 + +单图并行 2 路（OCR + lite JSON VLM），目标 <3s。 +外层 8 图全并发，目标 8 图 <15s。 +终极兜底：复用旧 _analyze_single_image 完整 pro VLM 逻辑。 +""" +from __future__ import annotations + +import logging +import os +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any + +from . import assembler, ocr_volc, vlm_fast_json + +logger = logging.getLogger(__name__) + +# ---------- 配置项（可通过环境变量覆盖） ---------- +VISION_V2_ENABLED = os.environ.get("VISION_V2_ENABLED", "false").lower() in ("1", "true", "yes", "on") +# 单图 fast 路径总超时（包含 OCR + fast_json 并行） +V2_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "10")) +# 外层图片并发（默认 8，即全并行） +V2_IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8")) +# fast_json 单次超时 +V2_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "8")) +# OCR 单次超时 +V2_OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "8")) +# pro VLM 兜底超时（仅在 fast 路径完全失败时触发） +V2_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45")) +V2_LITE_TIMEOUT = float(os.environ.get("VISION_V2_LITE_TIMEOUT", "20")) + + +def _is_result_usable(result: dict[str, Any]) -> bool: + """与 viral_video._is_vision_result_usable 对齐的可用判定。""" + pp = (result.get("portrait_prompt") or "").strip() + if pp and pp not in ("无人像", "无法判断", "未识别"): + return True + name = result.get("name") or "" + if name and name not in ("未识别", "无法判断", "未知"): + return True + summary = result.get("summary") or "" + if len(summary) >= 5 and summary not in ("无法判断", "未识别"): + return True + cat = result.get("category") or "" + if cat == "非产品图" and pp != "无人像": + return True + kf = result.get("key_features") or [] + if kf and kf != ["无法判断"]: + # 只要有非默认特征且非空 + return True + return False + + +def _call_pro_fallback(img_url: str, idx: int, lite_model: str, pro_model: str) -> dict[str, Any] | None: + """fast 路径失败时，调用旧的 lite/pro 竞速 VLM。 + + 复用 viral_video._analyze_single_image 的实现，避免重复代码。 + """ + try: + from worker_app.tasks.viral_video import _analyze_single_image + except ImportError: + try: + from tasks.viral_video import _analyze_single_image # type: ignore + except ImportError: + logger.warning("[vision.v2] 无法 import _analyze_single_image，跳过 pro 兜底") + return None + try: + return _analyze_single_image( + idx, + img_url, + vision_model=lite_model, + timeout=int(V2_LITE_TIMEOUT), + pro_fallback_model=pro_model, + ) + except Exception as e: + logger.warning("[vision.v2] 图片 #%d pro 兜底异常 err=%s", idx, e, exc_info=True) + return None + + +def analyze_image_v2( + idx: int, + img_url: str, + *, + lite_model: str | None = None, + pro_model: str | None = None, +) -> dict[str, Any]: + """单张图片 V2 分析：OCR + lite JSON VLM 并行，必要时降级 pro VLM。 + + 返回的 dict 与 viral_video._normalize() 输出格式完全一致。 + """ + t0 = time.time() + # ---- 第 1 层：fast 路径并行 ---- + fast_json_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, + model=lite_model, + timeout=V2_FAST_JSON_TIMEOUT, + ) + f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=V2_OCR_TIMEOUT) + + # 等全部完成或超时 + for fut in as_completed([f_fj, f_ocr], timeout=V2_FAST_TIMEOUT): + try: + res = fut.result(timeout=1) + except Exception as e: + logger.warning("[vision.v2] 图片 #%d fast 子任务异常: %s", idx, e) + continue + if fut is f_fj: + fast_json_result = res if isinstance(res, dict) else None + elif fut is f_ocr: + ocr_result = res if isinstance(res, list) else [] + + fast_elapsed = time.time() - t0 + + # ---- 组装 fast 结果 ---- + assembled: dict[str, Any] | None = None + if fast_json_result: + assembled = assembler.assemble_result(idx, fast_json_result, ocr_result) + if _is_result_usable(assembled): + assembled["_fast_elapsed"] = round(fast_elapsed, 2) + logger.info( + "[vision.v2] 图片 #%d fast 路径命中 elapsed=%.2fs portrait_prompt=%s", + idx, + fast_elapsed, + (assembled.get("portrait_prompt") or "")[:40], + ) + return assembled + logger.info( + "[vision.v2] 图片 #%d fast 结果不可用 portrait_prompt=%s，走 pro 兜底", + idx, + (assembled.get("portrait_prompt") or "")[:40], + ) + else: + logger.info("[vision.v2] 图片 #%d fast_json 返回空 elapsed=%.2fs，走 pro 兜底", idx, fast_elapsed) + + # ---- 第 2 层：pro VLM 兜底（复用旧竞速逻辑）---- + pro_t0 = time.time() + use_lite = lite_model or vlm_fast_json.DEFAULT_LITE_MODEL + use_pro = pro_model or "doubao-seed-2-1-pro-260915" + pro_result = _call_pro_fallback(img_url, idx, use_lite, use_pro) + if pro_result and _is_result_usable(pro_result): + pro_result["_fallback_used"] = True + pro_result["_fast_elapsed"] = round(fast_elapsed, 2) + pro_result["_pro_elapsed"] = round(time.time() - pro_t0, 2) + logger.info( + "[vision.v2] 图片 #%d pro 兜底命中 total_elapsed=%.2fs", + idx, + time.time() - t0, + ) + return pro_result + + # ---- 第 3 层：兜底失败，返回 assembled 或标准 fallback ---- + if assembled: + assembled["_source"] = "v2_fast_json_degraded" + logger.warning( + "[vision.v2] 图片 #%d pro 兜底也失败，返回降级 fast 结果 elapsed=%.2fs", + idx, + time.time() - t0, + ) + return assembled + + # 最后的最后：返回最小可用结构 + logger.warning("[vision.v2] 图片 #%d 所有路径均失败 elapsed=%.2fs", idx, time.time() - t0) + return { + "name": "未识别", + "brand": "无法判断", + "category": "非产品图", + "appearance": "无法判断", + "packaging": "无法判断", + "text_on_package": ocr_result[:8], + "key_features": ["无法判断"], + "scene": "通用", + "mood": "", + "portrait_prompt": "无法判断", + "summary": "未识别", + "_source": "v2_all_failed", + } + + +def analyze_images_v2( + img_urls: list[str], + *, + lite_model: str | None = None, + pro_model: str | None = None, + max_workers: int | None = None, +) -> list[dict[str, Any]]: + """批量图片 V2 分析（外层全并行）。""" + if not img_urls: + return [] + workers = max_workers if max_workers and max_workers > 0 else V2_IMG_WORKERS + workers = min(workers, len(img_urls), 16) # 安全上限 16 + results: list[dict[str, Any] | None] = [None] * len(img_urls) + + logger.info( + "[vision.v2] 开始 V2 并行图片分析 n=%d workers=%d fast_timeout=%.0fs", + len(img_urls), + workers, + V2_FAST_TIMEOUT, + ) + t0 = time.time() + with ThreadPoolExecutor(max_workers=workers) as pool: + future_to_idx = { + pool.submit(analyze_image_v2, idx, url, lite_model=lite_model, pro_model=pro_model): idx + for idx, url in enumerate(img_urls) + } + for fut in as_completed(future_to_idx): + idx = future_to_idx[fut] + try: + results[idx] = fut.result() + except Exception as e: + logger.warning("[vision.v2] 图片 #%d future 异常 err=%s", idx, e, exc_info=True) + results[idx] = { + "name": "未识别", + "brand": "无法判断", + "category": "非产品图", + "appearance": "无法判断", + "packaging": "无法判断", + "text_on_package": [], + "key_features": ["无法判断"], + "scene": "通用", + "mood": "", + "portrait_prompt": "无法判断", + "summary": "未识别", + "_source": "v2_future_exception", + } + elapsed = time.time() - t0 + succ = sum(1 for r in results if r and _is_result_usable(r)) + fb = sum(1 for r in results if r and r.get("_fallback_used")) + logger.info( + "[vision.v2] V2 图片分析完成 n=%d success=%d pro_fallback=%d elapsed=%.2fs", + len(img_urls), + succ, + fb, + elapsed, + ) + return [r for r in results if r is not None] diff --git a/apps/worker/worker_app/tasks/vision/ocr_volc.py b/apps/worker/worker_app/tasks/vision/ocr_volc.py new file mode 100644 index 000000000..e01acdb76 --- /dev/null +++ b/apps/worker/worker_app/tasks/vision/ocr_volc.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +"""火山引擎 AI MediaKit OCR（同步）调用封装。 + +接口：POST {mediakit_base_url}/tools-sync/ocr +鉴权：Bearer {mediakit_api_key} +请求体：{"image_url": "<公网可访问URL>"} （部分版本也支持 image_base64） +响应：{"code":0,"data":{"texts":[{"text":"...","bbox":[x,y,w,h],...},...],...}} + +目标：识别商品包装/Logo/水印上的文字，作为 fast_json VLM 的补充。 +返回值：识别到的文本字符串列表（失败返回 []）。 +""" +from __future__ import annotations + +import logging +import time +from typing import Any + +logger = logging.getLogger(__name__) + +DEFAULT_TIMEOUT = 8 # OCR 秒级返回，8s 绰绰有余 + + +def call_ocr(img_url: str, *, timeout: int = DEFAULT_TIMEOUT) -> list[str]: + """调用 MediaKit 同步 OCR，返回去重后的纯文本列表。 + + 不做重试（外层降级逻辑负责）。失败/未配置返回空列表，不抛异常。 + """ + t0 = time.time() + try: + import httpx + + from packages.shared.mediakit_client import get_mediakit_client + + client = get_mediakit_client() + if not client.is_available: + logger.info("[vision.v2] mediakit 未配置，跳过 OCR") + return [] + + url = f"{client.base_url}/tools-sync/ocr" + headers = { + "Authorization": f"Bearer {client.api_key}", + "Content-Type": "application/json", + } + payload: dict[str, Any] = {"image_url": img_url} + # 部分文档版本用 image_base64，但公网 URL 场景下 image_url 最简 + resp = httpx.post(url, headers=headers, json=payload, timeout=timeout) + elapsed = time.time() - t0 + if resp.status_code != 200: + logger.warning( + "[vision.v2] OCR HTTP %d elapsed=%.1fs body=%s", + resp.status_code, + elapsed, + resp.text[:200], + ) + return [] + data = resp.json() + # 兼容几种可能的响应结构 + code = data.get("code", data.get("status", 0)) + if code not in (0, "OK", "success", 200): + logger.warning("[vision.v2] OCR 业务错误 code=%s elapsed=%.1fs resp=%s", code, elapsed, str(data)[:200]) + return [] + texts = _extract_texts(data) + # 去重 + 过滤空 + seen: set[str] = set() + out: list[str] = [] + for t in texts: + t = (t or "").strip() + if t and t not in seen and len(t) <= 100: # 过滤过长的误识别 + seen.add(t) + out.append(t) + logger.info("[vision.v2] OCR 完成 elapsed=%.1fs n=%d texts=%s", elapsed, len(out), out[:5]) + return out + except Exception as e: + elapsed = time.time() - t0 + logger.warning("[vision.v2] OCR 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True) + return [] + + +def _extract_texts(data: dict) -> list[str]: + """从 OCR 响应中抽取文本，兼容多种结构。""" + out: list[str] = [] + # 常见结构1: data.texts = [{"text": "..."}, ...] + d = data.get("data") or data + if isinstance(d, dict): + for key in ("texts", "lines", "words", "items", "result"): + items = d.get(key) + if isinstance(items, list): + for it in items: + if isinstance(it, dict): + txt = it.get("text") or it.get("content") or it.get("word") + if txt: + out.append(str(txt)) + elif isinstance(it, str): + out.append(it) + break + # 结构2: data.text = "..." + if not out: + t = d.get("text") + if isinstance(t, str): + out.append(t) + # 结构3: data.ocr_text / data.content + if not out: + for key in ("ocr_text", "content", "raw_text"): + v = d.get(key) + if isinstance(v, str) and v.strip(): + out.append(v) + break + return out diff --git a/apps/worker/worker_app/tasks/vision/vlm_fast_json.py b/apps/worker/worker_app/tasks/vision/vlm_fast_json.py new file mode 100644 index 000000000..78bfc1fb3 --- /dev/null +++ b/apps/worker/worker_app/tasks/vision/vlm_fast_json.py @@ -0,0 +1,141 @@ +# -*- coding: utf-8 -*- +"""doubao-seed-2.1-lite 强约束 JSON-only 调用。 + +目标：替代"人体属性/商品检测/图像标签"三个火山不存在的专用云端 API。 +设计要点： +- system prompt 极致精简，只给字段 schema 和强约束（禁止自然语言、禁止 markdown） +- max_tokens=350（比旧 VLM 的 1200 小很多，降低延迟） +- temperature=0.1（极低，稳定输出 JSON） +- timeout=8s（够快，失败则由外层走 pro VLM 兜底） +- 期望返回纯 JSON object（无 ```json 包裹、无解释文字） +""" +from __future__ import annotations + +import json +import logging +import time +from typing import Any + +logger = logging.getLogger(__name__) + +# 极简 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对象。" + +# 默认模型 +DEFAULT_LITE_MODEL = "doubao-seed-2-1-lite-260915" +DEFAULT_TIMEOUT = 8 +DEFAULT_MAX_TOKENS = 350 + + +def _strip_code_fence(s: str) -> str: + """剥离 ```json ... ``` 包裹（即使要求纯 JSON，模型偶尔仍会包代码块）。""" + s = s.strip() + if s.startswith("```"): + lines = s.split("\n") + # 去掉首行 ```json + 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 call_fast_json( + img_url: str, + *, + model: str | None = None, + timeout: int = DEFAULT_TIMEOUT, + max_tokens: int = DEFAULT_MAX_TOKENS, +) -> dict[str, Any] | None: + """调用 lite VLM 返回结构化 dict；失败/非 JSON 返回 None。 + + 注意：不做重试（外层竞速/降级逻辑负责），max_retries=0 由外层统一设置。 + """ + t0 = time.time() + try: + from packages.shared.ai_client import get_doubao_client + + client = get_doubao_client() + if not client.is_available: + logger.warning("[vision.v2] doubao client 不可用，跳过 fast_json") + return None + + use_model = model or DEFAULT_LITE_MODEL + 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, + ) + elapsed = time.time() - t0 + if raw is None: + logger.warning("[vision.v2] fast_json 返回 None elapsed=%.1fs model=%s", elapsed, use_model) + return None + + text = _strip_code_fence(raw) + # 截到第一个 { 和最后一个 } 之间，容忍前后偶发文字 + l = text.find("{") + r = text.rfind("}") + if l >= 0 and r > l: + text = text[l : 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 完成 model=%s elapsed=%.1fs has_person=%s has_product=%s category=%s", + use_model, + elapsed, + obj.get("has_person"), + obj.get("has_product"), + obj.get("category"), + ) + return obj + except Exception as e: + elapsed = time.time() - t0 + logger.warning("[vision.v2] fast_json 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True) + return None -- 2.54.0