# ============================================================ # Mage-VL 第七版 # 实时洗衣机衣物分析 # # 功能： # 1. 4 秒窗口实时分析 # 2. StreamMind Gate 数值稳定修复 # 3. BF16 优先 / T4 自动 FP16 + FP32 Gate fallback # 4. 累计衣物状态 # 5. 动态洗涤程序 # 6. 动态漂洗次数 # 7. 视频结束后的最终综合洗涤方案 # # 显存优化： # - Codec max_pixels = 48000 # - target_canvas = 20 # - max_new_tokens = 48 # - 不保存 GPU 上的历史 inputs # - 每段结束主动清理 CUDA cache # ============================================================ import os import gc import time import subprocess import importlib import warnings import json import math from pathlib import Path # ============================================================ # 0. 环境 # ============================================================ os.environ.setdefault( "PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True,max_split_size_mb:128" ) os.environ.setdefault( "ONLINE_CODEC_CACHE_DIR", "/tmp/mage_vl_codec_cache" ) os.environ.setdefault( "STREAMMIND_CLIP_CACHE", "/tmp/mage_vl_clips" ) os.makedirs( os.environ["ONLINE_CODEC_CACHE_DIR"], exist_ok=True ) os.makedirs( os.environ["STREAMMIND_CLIP_CACHE"], exist_ok=True ) # ============================================================ # 1. Import # ============================================================ import torch from transformers import ( AutoModelForCausalLM, AutoProcessor, ) # ============================================================ # 2. 配置 # ============================================================ MODEL_ID = "microsoft/Mage-VL" VIDEO_FILE = "/content/test.mp4" # ---------- 实时窗口 ---------- SEGMENT_SEC = 4.0 MAX_SEGMENTS = None # 测试时可以改成： # MAX_SEGMENTS = 3 # ---------- Gate ---------- GATE_THRESHOLD = 0.50 # ---------- LLM ---------- MAX_NEW_TOKENS = 48 # ---------- 最终总结 ---------- FINAL_MAX_NEW_TOKENS = 192 # ---------- Codec 显存 ---------- # 第六版 64000 # 第七版降低到 48000 CODEC_MAX_PIXELS = 48_000 # 第六版 24 # 第七版降低到 20 CODEC_TARGET_CANVAS = 20 # ---------- 实时模拟 ---------- SIMULATE_REALTIME = True # ============================================================ # 3. CUDA # ============================================================ if not torch.cuda.is_available(): raise RuntimeError("未检测到 CUDA GPU") DEVICE = "cuda" GPU_NAME = torch.cuda.get_device_name(0) CUDA_VERSION = torch.version.cuda # ============================================================ # 4. 自动选择 dtype # ============================================================ BF16_SUPPORTED = False try: BF16_SUPPORTED = torch.cuda.is_bf16_supported() except Exception: BF16_SUPPORTED = False if BF16_SUPPORTED: MODEL_DTYPE = torch.bfloat16 print("[dtype] GPU 支持 BF16") print("[dtype] Mage-VL = BF16") print("[dtype] StreamMind Gate = BF16") else: # T4 等旧 GPU： # 模型仍然使用 FP16 节省显存 # Gate 单独尝试 FP32 MODEL_DTYPE = torch.float16 print("[dtype] GPU 不支持 BF16") print("[dtype] Mage-VL = FP16") print("[dtype] StreamMind Gate = FP32 fallback") print() print("=" * 60) print("Mage-VL 第七版") print("=" * 60) print(f"[GPU] {GPU_NAME}") print(f"[CUDA] {CUDA_VERSION}") print(f"[BF16] {BF16_SUPPORTED}") print(f"[MODEL DTYPE] {MODEL_DTYPE}") print(f"[CODEC] max_pixels={CODEC_MAX_PIXELS}") print(f"[CODEC] target_canvas={CODEC_TARGET_CANVAS}") print("=" * 60) # ============================================================ # 5. Prompt # ============================================================ PROMPT_TEMPLATE = """ 你是洗衣机的实时视觉分析助手。 一个人正在不断把衣服放进洗衣机。 你的任务不是简单描述画面，而是： 【识别当前新增衣物】 + 【维护累计衣物】 + 【根据累计衣物动态选择最佳洗涤程序】。 -------------------------------------------------- 已有累计状态 -------------------------------------------------- {history} -------------------------------------------------- 当前视频片段 -------------------------------------------------- 请只重点判断当前这个视频片段中新出现或正在加入洗衣机的衣物。 请分析： 1. 当前新增衣物 - 类型： 牛仔裤、T恤、衬衫、毛衣、外套、 运动服、裙子、裤子、床品、大件衣物等 - 颜色 - 厚薄 - 重量感 - 可能材质： 棉、牛仔、涤纶、尼龙、羊毛、混纺等 如果无法确定材质，必须说“可能是”，不要假装确定。 -------------------------------------------------- 2. 累计衣物变化 -------------------------------------------------- 结合已有累计状态。 不要重复计算已经出现过的同一件衣服。 判断当前加入后，最终洗衣桶中的主要衣物组成。 -------------------------------------------------- 3. 动态洗涤程序 -------------------------------------------------- 根据目前累计衣物重新判断。 不要机械沿用上一段程序。 特别考虑： - 牛仔 - 棉质 - 化纤 - 混纺 - 毛衣 - 针织 - 外套 - 厚重衣物 - 大件衣物 - 不同材质混洗 - 深浅颜色混合 如果新增衣物导致程序应该改变，请明确指出。 例如： 牛仔 → 混合 混合 → 大物/混合 漂洗 2 次 → 3 次 -------------------------------------------------- 4. 当前推荐 -------------------------------------------------- 程序： 水温： 漂洗： 脱水： -------------------------------------------------- 5. 判断原则 -------------------------------------------------- 最终程序优先保护最脆弱或最限制程序选择的衣物。 如果存在明显不适合混洗的衣物，需要提醒。 例如： 羊毛/针织 → 不建议与普通衣物混洗 深浅色混合 → 注意掉色 厚外套 + 普通衣物 → 优先考虑厚重衣物的洗涤要求 -------------------------------------------------- 输出要求 -------------------------------------------------- 简洁。 只输出： 【新增衣物】 【累计变化】 【当前程序】 【理由】 不要长篇解释。 """.strip() # ============================================================ # 6. 最终总结 Prompt # ============================================================ FINAL_PROMPT = """ 你是洗衣机最终洗涤决策模块。 下面是视频实时分析过程中得到的所有衣物状态： {history} 请综合整个视频。 不要只参考最后一个片段。 -------------------------------------------------- 任务 -------------------------------------------------- 1. 汇总最终实际放入洗衣机的衣物 2. 删除重复描述 3. 判断主要衣物类型 4. 判断主要可能材质 5. 找出最影响洗涤程序选择的衣物 6. 判断是否存在不适合混洗的衣物 7. 给出最终最佳洗涤程序 -------------------------------------------------- 洗涤程序选择原则 -------------------------------------------------- 如果只有牛仔类： 优先考虑牛仔程序。 如果普通棉质、T恤、衬衫等混合： 优先考虑混合程序。 如果出现较厚外套、大件或明显重量较大的衣物： 考虑大物/混合或更适合厚重衣物的程序。 如果存在羊毛、针织等特殊衣物： 优先保护特殊衣物，并明确提示不建议混洗。 漂洗次数： 普通衣物一般较少。 衣物较多、较厚、洗涤剂残留风险较高时增加漂洗。 脱水： 根据最终衣物厚薄和特殊材质判断。 不要为了保持前面的建议而保持原程序。 如果最终衣物组成发生变化，必须重新选择。 -------------------------------------------------- 输出格式 -------------------------------------------------- 【最终衣物】 列出主要衣物。 【主要材质】 列出主要可能材质。 【最终洗涤程序】 程序： 水温： 漂洗： 脱水： 【为什么】 用2～4句话说明最终选择。 【注意事项】 只列真正重要的问题。 """.strip() # ============================================================ # 7. 加载 Processor # ============================================================ print() print("[init] 加载 Processor...") processor = AutoProcessor.from_pretrained( MODEL_ID, trust_remote_code=True, ) # ============================================================ # 8. 加载 Mage-VL # ============================================================ print("[init] 加载 Mage-VL...") model = AutoModelForCausalLM.from_pretrained( MODEL_ID, trust_remote_code=True, torch_dtype=MODEL_DTYPE, attn_implementation="sdpa", low_cpu_mem_usage=True, ).to(DEVICE).eval() print("[init] 模型加载完成") # ============================================================ # 9. Codec # ============================================================ _REMOTE_PKG = type(processor).__module__.rsplit(".", 1)[0] codec_mod = importlib.import_module( _REMOTE_PKG + ".codec_video_processing_mage_vl" ) CodecConfig = codec_mod.CodecConfig def codec_config(): override = { "engine": "hevc", "target_canvas": CODEC_TARGET_CANVAS, "patch": 16, } kwargs = dict( processor._codec_config_defaults ) dcvc = dict( kwargs.get("dcvc") or {} ) kwargs.update(override) if dcvc: kwargs["dcvc"] = dcvc kwargs["max_pixels"] = CODEC_MAX_PIXELS return CodecConfig(**kwargs), override # ============================================================ # 10. 加载 Gate # ============================================================ print("[gate] 加载 Cognition Gate...") try: gate = model.model._load_streammind_gate() print("[gate] Cognition Gate 加载成功") try: gate_dtype = next( gate.parameters() ).dtype print( f"[gate] 当前 dtype = {gate_dtype}" ) except Exception: pass except Exception as e: gate = None print( f"[gate] 加载失败: {type(e).__name__}: {e}" ) # ============================================================ # 11. T4 / FP16 下： # 尝试把 Gate 单独提升到 FP32 # ============================================================ if ( gate is not None and not BF16_SUPPORTED ): try: gate = gate.float() model.model.streammind_gate = gate print( "[gate] FP32 fallback 已启用" ) except Exception as e: print( "[gate] FP32 fallback 失败:", e ) # ============================================================ # 12. 工具 # ============================================================ def duration(path): try: r = subprocess.run( [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path), ], capture_output=True, text=True, check=True, timeout=60, ) return float( r.stdout.strip() ) except Exception: return 0.0 # ============================================================ # 13. Prompt # ============================================================ def make_prompt(history_text=""): text = PROMPT_TEMPLATE.format( history=( history_text if history_text else "暂无，这是第一个视频片段。" ) ) messages = [ { "role": "user", "content": [ { "type": "video" }, { "type": "text", "text": text, }, ], } ] return processor.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, ) # ============================================================ # 14. 视频切片 # ============================================================ def extract_clip( src, start, length, out, ): tmp = out.with_name( ".tmp_" + out.name ) if tmp.exists(): try: tmp.unlink() except Exception: pass subprocess.run( [ "ffmpeg", "-y", "-loglevel", "error", "-ss", f"{start:.3f}", "-i", str(src), "-t", f"{length:.3f}", "-c:v", "libx264", "-preset", "ultrafast", "-crf", "26", "-an", "-pix_fmt", "yuv420p", str(tmp), ], check=True, capture_output=True, timeout=180, ) tmp.replace(out) return out # ============================================================ # 15. CPU → GPU # ============================================================ def to_cuda(inputs): result = {} for k, v in inputs.items(): if not hasattr(v, "to"): continue v = v.to(DEVICE) if k == "pixel_values": v = v.to(MODEL_DTYPE) result[k] = v return result # ============================================================ # 16. 清理 GPU # ============================================================ def cleanup(*objs): for x in objs: try: del x except Exception: pass gc.collect() try: torch.cuda.empty_cache() torch.cuda.ipc_collect() except Exception: pass # ============================================================ # 17. GPU 显存 # ============================================================ def gpu_mem(): allocated = ( torch.cuda.memory_allocated() / 1024**3 ) reserved = ( torch.cuda.memory_reserved() / 1024**3 ) peak = ( torch.cuda.max_memory_allocated() / 1024**3 ) return ( f"GPU={allocated:.2f}GB " f"/ reserved={reserved:.2f}GB " f"/ peak={peak:.2f}GB" ) # ============================================================ # 18. Gate # ============================================================ @torch.inference_mode() def run_gate(inputs): feat = {} for k in ( "pixel_values", "image_grid_thw", "patch_positions", ): if k not in inputs: continue v = inputs[k].to(DEVICE) if k == "pixel_values": v = v.to(MODEL_DTYPE) feat[k] = v logits = None try: # ---------------------------------------------------- # BF16 GPU # ---------------------------------------------------- if BF16_SUPPORTED: logits = ( model.model .streammind_gate_forward_segments( [feat] ) ) # ---------------------------------------------------- # T4 / FP16 fallback # # 不直接使用官方 gate forward， # 而是： # # visual token → FP32 → Gate FP32 # # 尽量避免 FP16 Gate overflow # ---------------------------------------------------- else: visual_tokens = ( model.model ._streammind_vision_tokens( feat["pixel_values"], feat["image_grid_thw"], patch_positions=feat.get( "patch_positions" ), ) ) if not torch.isfinite( visual_tokens ).all(): print( "[gate] visual tokens " "出现 NaN/Inf" ) return 0.0 visual_tokens = ( visual_tokens.float() ) logits = gate( torch.cat( [visual_tokens], dim=1, ) ) # ---------------------------------------------------- # 官方 Gate 输出： # # [B,T,2] # # 0 = silence # 1 = response # ---------------------------------------------------- if not isinstance( logits, torch.Tensor ): raise RuntimeError( "Gate 输出不是 Tensor" ) if logits.ndim != 3: raise RuntimeError( "Unexpected gate logits shape: " + str(tuple(logits.shape)) ) gate_logits = logits[0] # Codec 时间长度 length = int( inputs[ "image_grid_thw" ][:, 0] .sum() .item() ) if length <= 0: raise RuntimeError( f"Invalid gate length={length}" ) pos = min( length - 1, gate_logits.shape[0] - 1, ) last_logits = ( gate_logits[pos] .float() ) # ---------------------------------------------------- # 关键： # 在 softmax 之前检查 # ---------------------------------------------------- if not torch.isfinite( last_logits ).all(): print( "[gate] WARNING:" " logits =", last_logits .detach() .cpu() .tolist(), ) return 0.0 # ---------------------------------------------------- # 用 log_softmax 更稳定 # ---------------------------------------------------- log_prob = torch.log_softmax( last_logits, dim=-1, ) prob = torch.exp( log_prob[1] ) if not torch.isfinite(prob): print( "[gate] WARNING:" " probability NaN/Inf" ) return 0.0 return float( prob.item() ) except Exception as e: print( "[gate] ERROR:", type(e).__name__, e, ) return 0.0 finally: cleanup( feat, logits, ) # ============================================================ # 19. Codec + Gate # ============================================================ @torch.inference_mode() def analyze_clip( clip, prompt, codec_override, ): with warnings.catch_warnings(): warnings.simplefilter( "ignore" ) inputs = processor( text=[prompt], videos=[ str(clip) ], video_backend="codec", max_pixels=CODEC_MAX_PIXELS, codec_config=codec_override, return_tensors="pt", padding=False, ) prob = run_gate( inputs ) return ( inputs, prob, ) # ============================================================ # 20. LLM # ============================================================ @torch.inference_mode() def generate( inputs, max_new_tokens=MAX_NEW_TOKENS, ): out = model.generate( **inputs, max_new_tokens=max_new_tokens, do_sample=False, use_cache=True, ) n = ( inputs[ "input_ids" ].shape[1] ) text = processor.tokenizer.decode( out[0, n:], skip_special_tokens=True, ).strip() del out return text # ============================================================ # 21. 最终总结 # ============================================================ @torch.inference_mode() def final_summary(history): if not history: return ( "没有获得有效的衣物分析结果。" ) history_text = "\n\n".join( history ) prompt = FINAL_PROMPT.format( history=history_text ) inputs = processor.tokenizer( prompt, return_tensors="pt", truncation=True, max_length=4096, ) inputs = { k: v.to(DEVICE) for k, v in inputs.items() } try: out = model.generate( **inputs, max_new_tokens=FINAL_MAX_NEW_TOKENS, do_sample=False, use_cache=True, ) n = ( inputs[ "input_ids" ].shape[1] ) result = ( processor.tokenizer.decode( out[0, n:], skip_special_tokens=True, ) .strip() ) return result finally: cleanup( inputs ) try: del out except Exception: pass # ============================================================ # 22. 实时视频分析 # ============================================================ def stream_video_analysis( video_path, segment_sec=SEGMENT_SEC, gate_threshold=GATE_THRESHOLD, max_segments=MAX_SEGMENTS, max_new_tokens=MAX_NEW_TOKENS, simulate_realtime=SIMULATE_REALTIME, ): path = Path( video_path ) if not path.exists(): raise FileNotFoundError( video_path ) total = duration( path ) if total <= 0: raise RuntimeError( "无法获取视频时长" ) clip_dir = Path( os.environ[ "STREAMMIND_CLIP_CACHE" ] ) clip_dir.mkdir( parents=True, exist_ok=True, ) # -------------------------------------------------------- # 两个历史： # # results： # 给用户看的完整输出 # # history： # 给模型看的累计状态 # -------------------------------------------------------- results = [] history = [] start = 0.0 index = 0 yield ( f"开始实时分析 | " f"时长={total:.1f}s | " f"窗口={segment_sec:.1f}s | " f"Gate={gate_threshold:.2f} | " f"Codec={CODEC_MAX_PIXELS}px" ) # -------------------------------------------------------- # Codec # -------------------------------------------------------- _, codec_override = codec_config() # ======================================================== # 主循环 # ======================================================== while start < total - 0.05: if ( max_segments is not None and index >= max_segments ): break end = min( total, start + segment_sec, ) length = ( end - start ) index += 1 t0 = time.time() clip = clip_dir / ( f"{path.stem}_" f"{index:06d}_" f"{start:.2f}_" f"{end:.2f}.mp4" ) inputs = None cuda_inputs = None said = None try: # ------------------------------------------------ # 1. 当前累计历史 # ------------------------------------------------ history_text = "\n".join( history[-6:] ) prompt = make_prompt( history_text ) # ------------------------------------------------ # 2. 精确切片 # ------------------------------------------------ extract_clip( path, start, length, clip, ) # ------------------------------------------------ # 3. Codec + Gate # ------------------------------------------------ inputs, prob = analyze_clip( clip, prompt, codec_override, ) head = ( f"**{start:.1f}–" f"{end:.1f}s** " f"Gate=`{prob:.2f}`" ) # ------------------------------------------------ # 4. Gate # # 注意： # # 本项目不让 Gate 决定“是否识别衣服”。 # # Gate 只是告诉我们： # 这个片段是否具有明显的事件变化。 # # 为避免漏掉新衣物： # 每个片段仍然进入 Mage-VL。 # ------------------------------------------------ if prob < gate_threshold: elapsed = time.time() - t0 speed = length / max(elapsed, 0.001) line = ( f"{head} → 🔇 跳过" f"（Gate < {gate_threshold:.2f}）" f" | Total={elapsed:.2f}s" ) results.append(line) yield "\n\n".join(results) # -------------------------------------------- # 关键： # # 不进入 Mage-VL # 不 generate # 不加入 history # # 直接进入下一个 4 秒片段 # -------------------------------------------- continue # ------------------------------------------------ # Gate 通过 # ------------------------------------------------ yield ( "\n\n".join(results) + ( "\n\n" if results else "" ) + f"{head} → 🧠 分析衣物中..." ) gt = time.time() # ------------------------------------------------ # GPU # ------------------------------------------------ cuda_inputs = to_cuda( inputs ) # ------------------------------------------------ # Mage-VL # ------------------------------------------------ said = generate( cuda_inputs, max_new_tokens=max_new_tokens, ) gen_time = ( time.time() - gt ) elapsed = ( time.time() - t0 ) speed = ( length / max(elapsed, 0.001) ) # ------------------------------------------------ # 保存结果 # ------------------------------------------------ line = ( f"{head} → 🗣️ " f"{said}\n" f" Generate=" f"{gen_time:.2f}s | " f"Total=" f"{elapsed:.2f}s | " f"Realtime=" f"{speed:.2f}×" ) results.append( line ) history.append( f"{start:.1f}-" f"{end:.1f}s：" f"{said}" ) yield "\n\n".join( results ) # if prob < gate_threshold: # gate_state = ( # "Gate较低" # ) # else: # gate_state = ( # "Gate触发" # ) # # ------------------------------------------------ # # 5. 进入 Mage-VL # # ------------------------------------------------ # yield ( # "\n\n".join( # results # ) # + ( # "\n\n" # if results # else "" # ) # + f"{head} " # + f"→ 🧠 {gate_state}，" # + "分析衣物中..." # ) # gt = time.time() # # ------------------------------------------------ # # GPU # # ------------------------------------------------ # cuda_inputs = to_cuda( # inputs # ) # # ------------------------------------------------ # # Generate # # ------------------------------------------------ # said = generate( # cuda_inputs, # max_new_tokens=max_new_tokens, # ) # gen_time = ( # time.time() # - gt # ) # elapsed = ( # time.time() # - t0 # ) # speed = ( # length # / max(elapsed, 0.001) # ) # # ------------------------------------------------ # # 6. 当前结果 # # ------------------------------------------------ # line = ( # f"{head} → 🗣️ " # f"{said}\n" # f" Generate=" # f"{gen_time:.2f}s | " # f"Total=" # f"{elapsed:.2f}s | " # f"Realtime=" # f"{speed:.2f}×" # ) # results.append( # line # ) # # ------------------------------------------------ # # 7. 只把模型结果加入累计历史 # # ------------------------------------------------ # history.append( # f"{start:.1f}-" # f"{end:.1f}s：" # f"{said}" # ) # # ------------------------------------------------ # # 8. 实时输出 # # ------------------------------------------------ # yield "\n\n".join( # results # ) # ------------------------------------------------ # 9. 模拟真实播放速度 # ------------------------------------------------ elapsed = ( time.time() - t0 ) if ( simulate_realtime and elapsed < length ): time.sleep( length - elapsed ) except Exception as e: line = ( f"**{start:.1f}–" f"{end:.1f}s** → " f"❌ " f"{type(e).__name__}: " f"{e}" ) results.append( line ) yield "\n\n".join( results ) finally: # ------------------------------------------------ # 清理 GPU # ------------------------------------------------ cleanup( cuda_inputs, inputs, said, ) # ------------------------------------------------ # 删除临时视频 # ------------------------------------------------ if clip.exists(): try: clip.unlink() except Exception: pass start = end # ------------------------------------------------ # 显存统计 # ------------------------------------------------ print( f"[segment {index}] " f"{start:.1f}/" f"{total:.1f}s | " f"{gpu_mem()}", flush=True, ) # ------------------------------------------------ # 重置 peak # ------------------------------------------------ try: torch.cuda.reset_peak_memory_stats() except Exception: pass # ======================================================== # 视频结束 # ======================================================== yield ( "\n\n".join( results ) + "\n\n" + "========================================\n" + "🔄 视频播放完成，正在综合全部衣物...\n" + "========================================" ) # ======================================================== # 最终综合 # ======================================================== try: final = final_summary( history ) final_output = ( "\n\n".join( results ) + "\n\n" + "========================================\n" + "🧺 最终综合洗涤方案\n" + "========================================\n\n" + final + "\n\n" + f"✅ 完成，共 {index} 个片段" ) yield final_output except Exception as e: yield ( "\n\n".join( results ) + "\n\n" + "========================================\n" + "🧺 最终综合洗涤方案\n" + "========================================\n\n" + f"❌ 综合分析失败：" + f"{type(e).__name__}: {e}" ) # ============================================================ # 23. Main # ============================================================ if __name__ == "__main__": for output in stream_video_analysis( VIDEO_FILE, segment_sec=SEGMENT_SEC, gate_threshold=GATE_THRESHOLD, max_segments=MAX_SEGMENTS, max_new_tokens=MAX_NEW_TOKENS, simulate_realtime=SIMULATE_REALTIME, ): # 清屏 print( "\033[H\033[J", end="" ) print( "=== Mage-VL 第七版：" "洗衣机实时衣物分析 ===\n" ) print(output) print( "\n" + "-" * 60 )