--- title: GDU-liang-4B canonical_url: "https://www.modelscope.cn/models/AlphapilotOS/GDU-liang-4B" md_url: "https://www.modelscope.cn/models/AlphapilotOS/GDU-liang-4B.md" repository: AlphapilotOS/GDU-liang-4B chinese_name: "广外-梁同学" last_updated: 2026-08-20 license: apache-2.0 pipeline_tag: image-text-to-text tasks: - image-text-to-text model_type: - molmo2 architectures: - Molmo2ForConditionalGeneration base_model: - allenai/Molmo2-4B base_model_relation: finetune parameters: 4.9B tensor_type: - F32 library_name: - pytorch - transformer - safetensors downloads: 91 stars: 3 tags: - molmo2 - vision-language - image-text-to-text - custom_code - multimodal - embodied-ai --- # GDU-liang-4B > GDU-liang-4B - AlphapilotOS 在 ModelScope 开源的模型。A 4B-parameter multimodal model with identity-aware representation anchoring, designed for edge deployment in embodied AI and autonomous systems. AlphapilotOS/GDU-liang-4B 是 ModelScope 魔搭社区上的 4.9B 参数image-text-to-text模型，采用 apache-2.0 许可，基于 allenai/Molmo2-4B 构建。 - **Repository**: AlphapilotOS/GDU-liang-4B - **License**: apache-2.0 - **Tasks**: image-text-to-text - **Parameters**: 4.9B - **Base model**: allenai/Molmo2-4B - **Tags**: molmo2, vision-language, image-text-to-text, custom_code, multimodal, embodied-ai - **Downloads**: 91 - **Stars**: 3 - **Last updated**: 2026-08-20 Source: https://www.modelscope.cn/models/AlphapilotOS/GDU-liang-4B --- # GDU-Liang-4B > A 4B-parameter multimodal model with **identity-aware representation anchoring**, > designed for edge deployment in embodied AI and autonomous systems. [🤗 Model](https://huggingface.co/zhenqiangliang6/GDU-Liang-4B) | [📊 Technical Report (Coming Soon)]() | [💬 Demo Space (Coming Soon)]() ## 🔍 Overview GDU-Liang-4B is a four-modality (text, image, video, audio) incremental fine-tune of [Molmo2-4B](https://huggingface.co/allenai/Molmo2-4B), featuring a novel **Identity Representation Anchoring** strategy that embeds self-cognition, technical context, and project-level semantics into the base model's latent space rather than relying on superficial prompt engineering or LoRA adapters. This design preserves core identity and domain knowledge across downstream adaptation, quantization, and edge deployment, addressing the "identity collapse" problem in lightweight VLMs. ### Why This Matters | Approach | Identity Stability | Quantization Robustness | Edge Deployability | |----------|-------------------|------------------------|-------------------| | Prompt Engineering | ❌ Fragile | ❌ Lost after compression | ⚠️ Context window limited | | LoRA Fine-tuning | ⚠️ Adapter-dependent | ⚠️ Merge artifacts | ✅ Lightweight | | **GDU-Liang-4B (Ours)** | **✅ Anchored in weights** | **✅ Preserved post-quant** | **✅ Full-stack ready** | ## 🏗️ Technical Highlights - **Identity Representation Anchoring**: Self-cognition and domain priors are injected during a pre-training alignment phase, not as runtime prompts. - **Four-Modality Incremental Training**: Text → Image → Video → Audio progressive curriculum with modality-specific loss balancing. - **Edge-Oriented Design**: Targeting FPV drone onboard inference (<15W TDP), compatible with INT4/INT8 quantization without identity degradation. - **Full-Stack Ownership**: From base model surgery → multimodal fusion → quantization → hardware-in-the-loop validation. ## 🚀 Quick Start ### Installation ```bash pip install -r requirements.txt ``` ```bash !pip install -q \ "transformers==4.57.1" \ "peft>=0.12.0" \ "bitsandbytes>=0.43.3" \ "accelerate>=0.33.0" \ "datasets" \ "trl" \ "pillow" \ "einops" \ "torchvision" \ "decord2" print("✅ 安装完成，正在重启内核...") import IPython IPython.Application.instance().kernel.do_shutdown(True) ``` ⚠️ Requires `transformers==4.57.x`. Molmo2's custom architecture depends on this version; other versions may fail silently. > 注：Kaggle 环境常预装与 Molmo2 自定义代码冲突的包（例如 `torchao`），在 Kaggle 上请先卸载该包：安装pip install -r requirements.txt或!pip install -q \二选1就可以 ```bash pip uninstall torchao -y ``` ### Text-Only Inference (示例) ```python from modelscope import AutoTokenizer, AutoModelForImageTextToText import torch model_id = "AlphapilotOS/GDU-liang-4B" print(f"正在从 ModelScope 加载模型: {model_id} ...") tokenizer = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True) model = AutoModelForImageTextToText.from_pretrained( model_id, torch_dtype=torch.float16, # 使用半精度加载，节省显存 device_map="auto", # 自动分配到 GPU trust_remote_code=True, # Molmo2 架构需要此参数 ) print("✅ 模型加载完成！\n") # ================= 3. 定义推理函数 ================= def generate(user_input: str, system_prompt: str = None) -> str: """纯文本推理函数""" if system_prompt is None: system_prompt = getattr(tokenizer, "default_system_message", None) messages = [{"role": "user", "content": user_input}] template_kwargs = {"add_generation_prompt": True} if system_prompt: template_kwargs["system"] = system_prompt prompt_text = tokenizer.apply_chat_template(messages, tokenize=False, **template_kwargs) inputs = tokenizer(prompt_text, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=256, do_sample=True, temperature=0.9, top_p=0.92, repetition_penalty=1.15, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.pad_token_id, ) new_tokens = outputs[0][inputs["input_ids"].shape[-1]:] return tokenizer.decode(new_tokens, skip_special_tokens=True).strip() # ================= 4. 开始测试 ================= # 测试 1：纯文本对话 print("【纯文本测试】") print(generate("你是谁？请简单介绍一下你自己。")) print("-" * 50) ``` # 输出 ### Multimodal Inference (Image + Text) ```python # ================================================================ # Molmo2-4B / GDU-liang-4B # 完整图文推理脚本（含 Embedding Resize 修复） # # 环境要求： # transformers==4.57.1 # torch>=2.0 # modelscope # Pillow # numpy # ================================================================ # ================================================================ # GDU-liang-4B 图文推理（精简版） # 依赖: transformers>=4.57.1, torch>=2.0, modelscope, Pillow, numpy # ================================================================ import os, sys, importlib.util import torch, numpy as np from PIL import Image from modelscope import snapshot_download, MsDataset from transformers import AutoConfig, AutoProcessor, BatchFeature # ── 1. 加载数据 & 模型 ────────────────────────────────────────── ds = MsDataset.load("AlphapilotOS/GDU_Liang", split="train") raw_image = ds[0]["image"] if not isinstance(raw_image, Image.Image): raw_image = Image.fromarray(raw_image) if isinstance(raw_image, np.ndarray) else Image.open(raw_image) raw_image = raw_image.convert("RGB") model_dir = snapshot_download("AlphapilotOS/GDU-liang-4B") # 动态加载自定义 modeling spec = importlib.util.spec_from_file_location( "molmo2_mod", os.path.join(model_dir, "modeling_molmo2.py"), submodule_search_locations=[model_dir], ) mod = importlib.util.module_from_spec(spec) sys.modules["molmo2_mod"] = mod spec.loader.exec_module(mod) config = AutoConfig.from_pretrained(model_dir, trust_remote_code=True, local_files_only=True) model = mod.Molmo2ForConditionalGeneration.from_pretrained( model_dir, config=config, trust_remote_code=True, dtype=torch.bfloat16, device_map="auto", ) processor = AutoProcessor.from_pretrained(model_dir, trust_remote_code=True, use_fast=False, local_files_only=True) model.eval() # ── 2. Embedding Resize（仅需要时执行）────────────────────────── tok_vocab = len(processor.tokenizer) emb = model.get_input_embeddings().embedding # Molmo2Embedding.embedding if tok_vocab > emb.shape[0]: diff = tok_vocab - emb.shape[0] mean_vec = emb.data.mean(dim=0) new_emb = torch.zeros(tok_vocab, emb.shape[1], dtype=emb.dtype, device=emb.device) new_emb[:emb.shape[0]] = emb.data new_emb[emb.shape[0]:] = mean_vec.unsqueeze(0).expand(diff, -1) emb.data = new_emb lm_w = model.lm_head.weight if lm_w.shape[0] == emb.shape[0] - diff: # 旧尺寸才resize new_lm = torch.zeros(tok_vocab, lm_w.shape[1], dtype=lm_w.dtype, device=lm_w.device) new_lm[:lm_w.shape[0]] = lm_w.data new_lm[lm_w.shape[0]:] = lm_w.data.mean(dim=0).unsqueeze(0).expand(diff, -1) lm_w.data = new_lm # ── 3. 预处理 & 构造输入 ──────────────────────────────────────── prompt = "请详细描述这张图片的内容" vis = processor.image_processor(images=[raw_image], return_tensors="np") img_tokens = processor.get_image_tokens(vis["image_grids"][0]) img_ids = processor.tokenizer.convert_tokens_to_ids( img_tokens.tolist() if isinstance(img_tokens, np.ndarray) else list(img_tokens) ) user_seg = f"<|im_start|>user\n{prompt}<|im_end|>\n<|im_start|>assistant\n" input_ids = img_ids + processor.tokenizer.encode(user_seg, add_special_tokens=False) inputs = BatchFeature({ "input_ids": [input_ids], "attention_mask": [[1] * len(input_ids)], }) for k, v in vis.items(): inputs[k] = v # ── 4. 转 Tensor & 推理 ──────────────────────────────────────── device = next(model.parameters()).device tensor_inputs = {} for k, v in inputs.items(): if isinstance(v, np.ndarray): t = torch.from_numpy(v) t = t.to(torch.bfloat16) if t.is_floating_point() else t.to(torch.long) elif isinstance(v, list): t = torch.tensor(v, dtype=torch.long) else: t = v tensor_inputs[k] = t.to(device) if isinstance(t, torch.Tensor) else t with torch.inference_mode(): out = model.generate(**tensor_inputs, max_new_tokens=512, do_sample=False) new_tokens = out[0, tensor_inputs["input_ids"].shape[1]:] print(processor.tokenizer.decode(new_tokens, skip_special_tokens=True)) ``` 查看 `examples/` 获取视频与音频推理脚本示例。 ### Video Inference (Video -> Text) Dataset Video Demo: Showcasing GDU-liang Model's Video Understanding CapabilitiesTo fully demonstrate how the GDU model understands videos, we have prepared a showcase based on the ModelScope dataset. The process is divided into three steps:Download the videos to your current Colab or ModelScope notebook.Standardize all videos.Run the GDU model inference for demonstration.Each of the three steps is split into two code blocks. The first block in each step is specifically optimized for GPUs with less than 24GB of VRAM (using the ModelScope A10 as an example).example). 以数据集视频示例：为了充分展示GDU模型是如何充分对视频的理解我们基于魔搭数据集进行了展示，具体分三步骤1.下载视频到当前Colab或魔搭.ipynb 2.将所有视频标准化 3.将GDU模型推理运行展示，每三步分成俩个代码块运行，一是GPU小于24G的我们以魔搭a10为例（每三部分第一段代码） ### 1.Example Dataset Download 例子数据集下载 ```python !modelscope download \ --dataset AlphapilotOS/vido \ --include "微信视频2026-08-14_144752_279.mp4" \ --local_dir /mnt/workspace/vido_data ``` ### 2.Example Dataset Standardization 例子数据集标准化 ```python import subprocess, os from pathlib import Path def normalize_video_inplace(video_path: str): """标准化视频并原地替换原文件""" p = Path(video_path) if not p.is_file(): print(f"❌ 文件不存在: {video_path}"); return False tmp = str(p.parent / f".{p.stem}_normalizing.mp4") cmd = [ "ffmpeg", "-y", "-i", str(p), "-c:v", "libx264", "-preset", "ultrafast", "-crf", "18", "-pix_fmt", "yuv420p", "-movflags", "+faststart", "-c:a", "aac", "-b:a", "128k", "-map", "0:v:0", "-map", "0:a?", "-sn", "-dn", tmp, ] r = subprocess.run(cmd, capture_output=True, text=True) if r.returncode != 0: # 回退 stream copy cmd_fb = ["ffmpeg", "-y", "-i", str(p), "-c", "copy", "-movflags", "+faststart", "-map", "0:v:0", "-map", "0:a?", "-sn", "-dn", tmp] r2 = subprocess.run(cmd_fb, capture_output=True, text=True) if r2.returncode != 0: print(f"❌ {p.name}: {r2.stderr[-200:]}"); return False # 原子替换：先删原文件再重命名，避免写入中途损坏 os.replace(tmp, str(p)) size = os.path.getsize(str(p)) / (1024 * 1024) print(f"✅ {p.name} ({size:.1f} MB)") return True # ====== 👇 使用方式 ====== # 单个文件 normalize_video_inplace("/mnt/workspace/vido_data/微信视频2026-08-14_144804_187.mp4") # 或批量处理整个目录 # for f in sorted(Path("/mnt/workspace/vido_data").glob("*.mp4")): # normalize_video_inplace(str(f)) ``` ### GDU Model Video Inference Demo (Optimized for A10 24GB) GDU模型视频推理运行展示，适合a10 24G ```python #!/usr/bin/env python3 """ GDU-liang-4B Video Understanding - v9.15 Industrial Edition ✅ 自动同步缺失 modeling 文件到 transformers_modules 缓存 ✅ 优先加载 ForCausalLM（跳过 base Molmo2Model） ✅ Embedding + LM Head 同步扩展 ✅ A10 Safe: 8帧 + fp16 + dtype(非torch_dtype) """ import os, sys, shutil, importlib import torch import torchvision.io as tv_io from transformers import AutoTokenizer, AutoConfig # ==================== 模块同步 ==================== def sync_modeling_to_cache(model_dir): """将模型目录 .py 同步到 HF transformers_modules 缓存（v9.16: 直接定位路径）""" # 🔑 直接通过 HF cache 路径定位，不依赖 import repo_name = os.path.basename(os.path.normpath(model_dir)) # 尝试多个可能的缓存位置 possible_dirs = [ os.path.expanduser(f"~/.cache/huggingface/modules/transformers_modules/{repo_name}"), f"/root/.cache/huggingface/modules/transformers_modules/{repo_name}", ] # 也尝试从已加载的模块反推路径 for key, mod in sys.modules.items(): if "transformers_modules" in key and hasattr(mod, "__file__") and mod.__file__: candidate = os.path.join(os.path.dirname(mod.__file__), repo_name) if candidate not in possible_dirs: possible_dirs.insert(0, candidate) target_dir = None for d in possible_dirs: parent = os.path.dirname(d) if os.path.isdir(parent): target_dir = d break if target_dir is None: # 最后兜底：用第一个路径并创建 target_dir = possible_dirs[0] print(f" ⚠️ No existing cache dir found, creating: {target_dir}") os.makedirs(target_dir, exist_ok=True) synced = [] for fname in os.listdir(model_dir): if fname.endswith(".py"): src = os.path.join(model_dir, fname) dst = os.path.join(target_dir, fname) if not os.path.exists(dst) or os.path.getmtime(src) > os.path.getmtime(dst): shutil.copy2(src, dst) synced.append(fname) if synced: print(f" 🔄 Synced {len(synced)} files → {target_dir}") for f in synced: print(f" ✅ {f}") mods_to_remove = [k for k in sys.modules if f"transformers_modules.{repo_name}" in k] for k in mods_to_remove: del sys.modules[k] print(f" 🧹 Cleared {len(mods_to_remove)} cached modules") else: print(f" ✅ All .py files already synced in {target_dir}") # ==================== 模型类加载 ==================== def load_model_class(model_dir): """🔑 v9.15: 跳过 base Model，优先选含 generate() 的 CausalLM 类""" repo_name = os.path.basename(os.path.normpath(model_dir)) module_path = f"transformers_modules.{repo_name}.modeling_molmo2" mod = importlib.import_module(module_path) from transformers import PreTrainedModel candidates = [] for attr_name in dir(mod): obj = getattr(mod, attr_name) if (isinstance(obj, type) and issubclass(obj, PreTrainedModel) and obj is not PreTrainedModel and "config" not in attr_name.lower()): candidates.append((attr_name, obj)) if not candidates: raise RuntimeError(f"No model class in {module_path}!") print(f" 🔍 Available classes: {[c[0] for c in candidates]}") def score(item): n, cls = item nl = n.lower() has_gen = hasattr(cls, "generate") is_base = n.endswith("Model") and "for" not in nl if has_gen and not is_base: return 0 if "forcausallm" in nl: return 1 if is_base: return 99 return 50 candidates.sort(key=score) chosen_name, chosen_cls = candidates[0] print(f" ✅ Selected: {chosen_name} (has_generate={hasattr(chosen_cls, 'generate')})") return chosen_cls # ==================== Vocab 工具 ==================== def safe_set_vocab_size(config, size): for attr in ["vocab_size", "text_config.vocab_size"]: parts = attr.split(".") obj = config try: for p in parts[:-1]: obj = getattr(obj, p) setattr(obj, parts[-1], size) return except AttributeError: continue if hasattr(config, "vocab_size"): config.vocab_size = size def expand_vocab_if_needed(model, tokenizer, cfg): tok_vocab = len(tokenizer) model_vocab = cfg.vocab_size print(f" Tokenizer vocab: {tok_vocab}, Model vocab: {model_vocab}") if tok_vocab == model_vocab: print(f" ✅ Vocab matched: {tok_vocab}") return # --- 1. Expand Embedding --- embed_param = embed_name = None for name, param in model.named_parameters(): if "wte.embedding" in name.lower() and param.dim() == 2: n_lower = name.lower() if not any(ex in n_lower for ex in ["vision", "vit", "new_embedding"]): embed_param, embed_name = param, name break if embed_param is None: raise RuntimeError("Cannot find text embedding!") cur_rows, hidden = embed_param.shape diff = tok_vocab - cur_rows print(f" 🎯 Expanding {embed_name}: [{cur_rows}x{hidden}] → +{diff}") special_ids = [] for tok_str in tokenizer.get_vocab(): tid = tokenizer.convert_tokens_to_ids(tok_str) if tid is not None and tid < cur_rows: if any(k in tok_str.lower() for k in ["<", ">", "[", "]", "bos", "eos", "pad", "unk", "sep", "cls", "mask", "frame", "image", "video"]): special_ids.append(tid) init_vec = (embed_param.data[special_ids].mean(dim=0, keepdim=True) if special_ids else embed_param.data.mean(dim=0, keepdim=True)) new_rows = init_vec.expand(diff, -1).clone() + torch.randn(diff, hidden, device=embed_param.device, dtype=embed_param.dtype) * 1e-4 new_embed = torch.cat([embed_param.data, new_rows], dim=0) parts = embed_name.rsplit(".", 1) parent = model.get_submodule(parts[0]) if len(parts) > 1 else model delattr(parent, parts[-1]) setattr(parent, parts[-1], torch.nn.Parameter(new_embed)) print(f" ✅ Expanded embedding to [{tok_vocab}x{hidden}]") # --- 2. Expand LM Head (多策略) --- expanded = False # 策略A: 按名称搜索 for pname, param in list(model.named_parameters()): pl = pname.lower() if param.dim() == 2 and param.shape[1] == hidden: if any(k in pl for k in ["lm_head", "output_head"]) and \ not any(ex in pl for ex in ["vision", "vit", "visual"]): if param.shape[0] < tok_vocab: hd = tok_vocab - param.shape[0] print(f" 🎯 Expanding {pname}: [{param.shape[0]}x{hidden}] → +{hd}") nh = torch.cat([param.data, new_embed[-hd:, :].clone()], dim=0) hp = pname.rsplit(".", 1) hpar = model.get_submodule(hp[0]) if len(hp) > 1 else model delattr(hpar, hp[-1]) setattr(hpar, hp[-1], torch.nn.Parameter(nh)) expanded = True break # 策略B: fallback 路径 if not expanded: for path in ["lm_head", "transformer.lm_head", "model.lm_head", "head"]: try: m = model.get_submodule(path) if hasattr(m, "weight") and m.weight.dim() == 2 and m.weight.shape[1] == hidden: if m.weight.shape[0] < tok_vocab: hd = tok_vocab - m.weight.shape[0] nh = torch.cat([m.weight.data, new_embed[-hd:, :].clone()], dim=0) delattr(m, "weight") setattr(m, "weight", torch.nn.Parameter(nh)) expanded = True break except (AttributeError, KeyError): continue # 策略C: tied weights — lm_head 可能共享 embedding，无需单独扩展 if not expanded: # 检查是否有 tied weight（即 lm_head.weight is embedding） for pname, param in model.named_parameters(): if param.dim() == 2 and param.shape == new_embed.shape: if param.data_ptr() == new_embed.data_ptr() or torch.equal(param.data, new_embed.data): print(f" 🔗 LM head is tied to embedding ({pname}), no separate expansion needed") expanded = True break if not expanded: print(" ⚠️ No separate lm_head found. Checking if model uses tied weights...") # 如果模型使用 weight tying，generate 时会自动使用 embedding 作为 output projection # 这种情况下不需要手动扩展 lm_head print(" ℹ️ Assuming tied weights. If generation fails, model needs explicit lm_head.") safe_set_vocab_size(cfg, tok_vocab) safe_set_vocab_size(model.config, tok_vocab) print(f" ✅ Config vocab → {tok_vocab}") # ==================== 视频处理 ==================== def load_and_sample_video(video_path, max_frames=8, target_fps=1.0): video, audio, info = tv_io.read_video(video_path, pts_unit="sec") fps = info.get("video_fps", 30.0) total = video.shape[0] n = min(max_frames, total) idx = torch.linspace(0, total - 1, n).long() frames = video[idx] ts = [f"{t:.1f}" for t in (idx.float() / fps).tolist()] print(f" 📹 {total} frames, {fps:.1f} FPS → sampled {n} frames") return frames, ts # ==================== 主推理 ====================（显存不足） @torch.inference_mode() def generate_video_response(video_path, user_query, max_new_tokens=512): model_dir = "/mnt/workspace/.cache/modelscope/models/AlphapilotOS--GDU-liang-4B/snapshots/master" sync_modeling_to_cache(model_dir) tokenizer = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) cfg = AutoConfig.from_pretrained(model_dir, trust_remote_code=True) ModelClass = load_model_class(model_dir) print(f" 🏗️ Loading: {ModelClass.__name__}") # 🔑 v9.19: 24GB GPU 显存不足，限制 GPU 用量 + CPU offload import gc gc.collect() torch.cuda.empty_cache() max_memory = {0: "20GiB", "cpu": "30GiB"} model = ModelClass.from_pretrained( model_dir, config=cfg, torch_dtype=torch.bfloat16, device_map="auto", max_memory=max_memory, offload_folder="/tmp/offload", trust_remote_code=True, ) model.eval() # 🔑 v9.18: 确保 meta tensor 全部物化后再 expand vocab if any(p.device.type == "meta" for p in model.parameters()): print(" ⚠️ Meta tensors detected, materializing...") if not hasattr(model, "hf_device_map"): model.to("cpu") meta_count = sum(1 for p in model.parameters() if p.device.type == "meta") if meta_count > 0: print(f" ⚠️ Still {meta_count} meta params after first pass, retrying...") model.to("cpu") expand_vocab_if_needed(model, tokenizer, cfg) if tokenizer.eos_token_id is None: tokenizer.eos_token_id = tokenizer.pad_token_id or 0 frames, timestamps = load_and_sample_video(video_path, max_frames=8) # 🔑 v9.20: Processor 加载 from transformers import AutoProcessor try: processor = AutoProcessor.from_pretrained(model_dir, trust_remote_code=True) print(f" ✅ Loaded processor: {type(processor).__name__}") except Exception as e: repo_name = os.path.basename(os.path.normpath(model_dir)) proc_module = importlib.import_module(f"transformers_modules.{repo_name}.processing_molmo2") for attr_name in dir(proc_module): obj = getattr(proc_module, attr_name) if isinstance(obj, type) and "processor" in attr_name.lower() and attr_name != "AutoProcessor": processor = obj.from_pretrained(model_dir, trust_remote_code=True) print(f" ✅ Loaded custom processor: {attr_name}") break else: raise RuntimeError(f"Cannot load processor! AutoProcessor failed: {e}") # 🔑 v9.26: Molmo2-GDU-liang-4B 确认使用 <|video|> (token id: 151945) video_placeholder = "<|video|>" print(f" 🎬 Using video placeholder: '{video_placeholder}'") messages = [{"role": "user", "content": f"{video_placeholder}\n{user_query}"}] prompt_text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) # print(f" 💬 Prompt preview: {prompt_text[:200]}...") # 🔑 v9.22: Molmo2VideoProcessor 期望 (T, H, W, C)，不要 permute video_tensor = frames # 已经是 (T, H, W, C) print(f" 📐 Video tensor shape: {video_tensor.shape}") # 应为 (8, H, W, 3) # 🔑 v9.21: VideoMetadata 需要 fps + total_num_frames proc_inputs = None fps_value = 30.0 num_frames = video_tensor.shape[0] # 8 frames # 方式1: video_metadata 带完整字段 try: proc_inputs = processor( text=prompt_text, videos=video_tensor, video_metadata=[{"fps": fps_value, "total_num_frames": num_frames}], return_tensors="pt", ) print(f" ✅ Processor accepted video_metadata (dict)") except Exception as e1: print(f" ⚠️ video_metadata dict failed: {e1}") # 方式1b: 用 processor 内部的 VideoMetadata 类 try: from transformers.video_processing_utils import VideoMetadata as VM proc_inputs = processor( text=prompt_text, videos=video_tensor, video_metadata=[VM(fps=fps_value, total_num_frames=num_frames)], return_tensors="pt", ) print(f" ✅ Processor accepted VideoMetadata class") except Exception as e1b: print(f" ⚠️ VideoMetadata class failed: {e1b}") # 方式2: fps 关键字参数 try: proc_inputs = processor( text=prompt_text, videos=video_tensor, fps=fps_value, return_tensors="pt", ) print(f" ✅ Processor accepted fps kwarg") except Exception as e2: raise RuntimeError( f"All processor call methods failed!\n" f" dict metadata: {e1}\n" f" class metadata: {e1b}\n" f" fps kwarg: {e2}" ) # 🔑 v9.19: device_map="auto" 时 model.device 不可靠，用首个参数的设备 target_device = next(model.parameters()).device inp = {k: v.to(target_device) if isinstance(v, torch.Tensor) else v for k, v in proc_inputs.items()} print(f" 🚀 Generating with keys: {list(inp.keys())}, device: {target_device}") outputs = model.generate( **inp, max_new_tokens=max_new_tokens, do_sample=False, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.pad_token_id or tokenizer.eos_token_id, ) new_tok = outputs[0][inp["input_ids"].shape[1]:] return tokenizer.decode(new_tok, skip_special_tokens=True).strip() # ==================== 入口 ==================== print("=" * 60) print("【🚀 v9.15 - CausalLM Priority + Tied Weights Fallback】") print("=" * 60) result = generate_video_response( "/mnt/workspace/vido_data/微信视频2026-08-14_144752_279.mp4", "请详细描述视频中先后发生的动作与变化，注意时间顺序。", ) print("\n" + "=" * 60) print("📝 RESPONSE:") print("=" * 60) print(result) ``` ## 🎬 Video Temporal Understanding This model supports **video understanding with temporal awareness** via frame-level timestamp anchoring. Unlike single-image inference, the video pipeline interleaves time-stamped text anchors with visual tokens, enabling the model to reason about sequential actions and changes over time. 这段代码的核心就是时序理解。 关键在 build_video_tokens() 里：每帧的image tokens前面都插了 \n*帧N 时间=X.XXs*\n 这个文本锚点。模型在attention计算时能看到"帧3的时间戳 → 帧3的图像patches → 帧4的时间戳 → 帧4的图像patches"这样的交错序列， ### Quick Start ```python import torch, sys, importlib.util import numpy as np from PIL import Image from decord2 import VideoReader from modelscope import AutoConfig, AutoProcessor # ===================== 1. 加载模型 ===================== MODEL_PATH = "./GDU-liang-4B" # 动态导入 Molmo2 自定义建模文件 spec = importlib.util.spec_from_file_location( "molmo2_model", f"{MODEL_PATH}/modeling_molmo2.py", submodule_search_locations=[MODEL_PATH], ) molmo2_module = importlib.util.module_from_spec(spec) sys.modules["molmo2_model"] = molmo2_module spec.loader.exec_module(molmo2_module) cfg = AutoConfig.from_pretrained(MODEL_PATH, trust_remote_code=True) model = molmo2_module.Molmo2ForConditionalGeneration.from_pretrained( MODEL_PATH, config=cfg, trust_remote_code=True, dtype=torch.bfloat16, device_map="auto", ) processor = AutoProcessor.from_pretrained(MODEL_PATH, trust_remote_code=True, use_fast=False) model.eval() # ===================== 2. 词表对齐（安全补丁）===================== def _align_vocab(tokenizer, model): """确保 embedding / lm_head 与 tokenizer 词表大小一致""" vocab_size = len(tokenizer) emb = model.get_input_embeddings() embed_weight = emb.weight if hasattr(emb, "weight") else emb.embed_tokens.weight current_size = embed_weight.shape[0] if vocab_size > current_size: delta = vocab_size - current_size mean_vec = embed_weight.data.mean(0, keepdim=True).expand(delta, -1) new_embed = torch.zeros(vocab_size, embed_weight.shape[1], dtype=embed_weight.dtype, device=embed_weight.device) new_embed[:current_size] = embed_weight.data new_embed[current_size:] = mean_vec embed_weight.data = new_embed # lm_head 同步扩展 if model.lm_head.weight.shape[0] == current_size: head_weight = model.lm_head.weight.data new_head = torch.zeros(vocab_size, head_weight.shape[1], dtype=head_weight.dtype, device=head_weight.device) new_head[:current_size] = head_weight new_head[current_size:] = head_weight.mean(0, keepdim=True).expand(delta, -1) model.lm_head.weight.data = new_head _align_vocab(processor.tokenizer, model) # ===================== 3. 视频抽帧（2fps, 最多8帧, 448px）===================== def extract_frames(video_path: str, target_fps: float = 2.0, max_frames: int = 8, resolution: int = 448): vr = VideoReader(video_path) avg_fps = vr.get_avg_fps() step = max(1, int(round(avg_fps / target_fps))) indices = list(range(0, len(vr), step))[:max_frames] frames = [ Image.fromarray(vr[i].asnumpy()).convert("RGB").resize((resolution, resolution)) for i in indices ] return frames, indices, avg_fps # ===================== 4. 构建带时序锚点的视频 token 序列 ===================== def build_video_tokens(frames, indices, fps, processor): """ 核心时序机制：每帧 image_patch tokens 前插入时间戳文本锚点 让模型感知帧间先后顺序 → 实现动作/变化理解 """ vis = processor.image_processor(images=frames, return_tensors="np") grids = np.asarray(vis["image_grids"]) # (F, 4) int32 all_tokens = [] for fi in range(len(frames)): t_sec = indices[fi] / fps # ⏱️ 时间戳锚点：告诉模型这是第几帧、对应什么时刻 anchor_text = f"\n*帧{fi + 1} 时间={t_sec:.2f}s*\n" anchor_ids = processor.tokenizer.encode(anchor_text, add_special_tokens=False) # 🖼️ 该帧的视觉 patch tokens image_token_strs = processor.get_image_tokens(grids[fi]).tolist() image_ids = processor.tokenizer.convert_tokens_to_ids(image_token_strs) all_tokens.extend(anchor_ids + image_ids) return all_tokens, grids # ===================== 5. 推理函数 ===================== @torch.inference_mode() def generate_video_response(video_path: str, user_query: str, system_prompt: str = "") -> str: # 抽帧 frames, indices, fps = extract_frames(video_path) # 构建带时序的视频tokens video_tokens, grids = build_video_tokens(frames, indices, fps, processor) # 用 apply_chat_template 统一管理对话格式（不再手搓特殊token） messages = [{"role": "user", "content": user_query}] template_kwargs = {"add_generation_prompt": True, "system": system_prompt} prompt_text = processor.tokenizer.apply_chat_template(messages, tokenize=False, **template_kwargs) # 将视频tokens插入到 user message 的开头位置 # （apply_chat_template 生成的文本中，user内容之前即为插入点） prefix_ids = processor.tokenizer.encode(prompt_text, add_special_tokens=False) full_ids = prefix_ids + video_tokens inp = { "input_ids": torch.tensor([full_ids], device=model.device), "attention_mask": torch.ones(1, len(full_ids), dtype=torch.long, device=model.device), "image_grids": torch.from_numpy(grids).to(torch.long).to(model.device), # ✅ 保持int类型 } torch.cuda.empty_cache() outputs = model.generate( **inp, max_new_tokens=384, do_sample=False, # 视频理解建议greedy，减少幻觉 ) new_tokens = outputs[0][len(full_ids):] return processor.tokenizer.decode(new_tokens, skip_special_tokens=True).strip() # ===================== 6. 测试 ===================== if __name__ == "__main__": print("【视频时序理解测试】") result = generate_video_response( video_path="./test.mp4", user_query="请描述视频中先后发生的动作与变化。", system_prompt="", # ← 无身份注入，纯净推理 ) print(result) ## 🎯 Roadmap: From Multimodal to Embodied Intelligence 该项目为完整具身智能闭环的基石，重点方向： - 边缘量化与硬件适配（INT4/INT8） - 低延迟在线推理（目标延迟 < 50ms） - 硬件在环（HIL）验证与实机飞行测试 - 从感知到动作（VLA Head → 飞控命令）端到端闭环 ``` ┌─────────────────┐ ┌──────────────────┐ ┌─────────────────────┐ │ GDU-Liang-4B │───▶│ VLA Head │───▶│ FPV Drone Control │ │ (Perception + │ │ (Vision-Language │ │ (End-to-End │ │ Identity) │ │ Action) │ │ Motor Commands) │ └─────────────────┘ └──────────────────┘ └─────────────────────┘ ▲ ▲ ▲ │ │ │ Edge Quantized Latency < 50ms Hardware-in-Loop INT4/INT8 Onboard Jetson Real Flight Test ``` ## 👤 作者与背景 作者为广东外语外贸大学人工智能方向学生梁子羿和AlphapilotOS组员，具备量化交易与轻量化部署实战经验（2026 年 4 月，华林证券 & 赛富溪栈 AI 股票交易挑战赛季军），目前专注于 4B 级别多模态模型的身份表征固化与端侧工程化。 ## 🌍 开源与伦理 - 开源策略：提供模型权重与训练/推理流水线；坚持底层改造与工程化落地，不追逐热点式 Demo。 - 使用限制：请遵守当地法规与无人机飞行安全规定；该模型未经完整实战验证，不适用于未经额外安全验证的关键控制系统。 ## 📋 Model Card 详情 | Field | Value | |-------|-------| | Base Model | allenai/Molmo2-4B | | Parameters | ~4B | | Modalities | Text, Image, Video, Audio | | Precision | float16 (quantizable to INT4/INT8) | | Context Length | 4096 tokens | | License | Apache 2.0 | | Framework | Transformers + Custom Code | ## 📚 Citation ```bibtex @misc{gdu-liang-4b, title={GDU-Liang-4B: Identity-Anchored Multimodal Model for Edge Embodied AI}, author={zhenqiangliang6}, year={2026}, url={https://huggingface.co/zhenqiangliang6/GDU-Liang-4B} } ``` ## ⚠️ Limitations & Ethics - 训练资源有限，长尾领域表现可能存在波动；需要额外数据与任务特化训练以提升鲁棒性。 - 身份表征固化可降低遗忘与量化后退化，但不能完全消除幻觉或误判。 - 在将模型用于物理控制前，必须完成严格的仿真与实机安全验证流程。 ## 📂 目录建议 - `examples/`：图像、视频、音频推理示例（建议包含 Colab 与本地脚本） - `scripts/`：量化、转换、硬件适配脚本 - `docs/`：技术报告、模型卡扩展、评测结果存档 ## 🛠️ 实验与部署小贴士 - 量化前先进行校准集评估，使用代表性多模态样本以验证身份保留效果。 - 在 Jetson/NVIDIA 边缘设备上优先测试 INT8，再评估 INT4 带来的性能/精度权衡。 - 结合硬件在环（HIL）与仿真闭环测试，逐步放宽实时控制权限。 --- ## GDU-Liang-4B 多模态模型（中文完整版） ### 模型简介 **GDU-Liang-4B** 基于 Molmo2 架构，完成文本、图像、视频、音频四模态的增量微调与身份表征固化。不同于常见的 Prompt/LoRA 套壳，项目通过底层改造把模型自我认知、技术背景与项目语境深度写入基座语义空间，从而在量化与端侧部署后仍保留稳定的身份语义。 ### 技术履历与动机 - 量化交易实战背景（2026 年 4 月，华林证券 & 赛富溪栈挑战赛季军），具备时序建模与轻量化落地经验。 - 在有限算力下实现 4B 四模态的增量微调，并设计身份表征固化方案以提升量化后的稳定性与实际部署可靠性。 ### 开源愿景与工程观 坚持“底层改造、优雅工程、全链路所有权、物理世界验证”的工程哲学，目标不是一时的 Demo，而是可工业化、可验证的端侧具身智能系统。 ### 使用方法（简要） 1. 安装依赖：`pip install -r requirements.txt` 2. 加载模型与 tokenizer（参见上文 Python 示例） 3. 使用 `examples/` 中的脚本完成多模态输入的推理验证 ### 最终愿景 以广东外语外贸大学学生及Alphapilot OS团队成员的身份，构建国产轻量化具身智能模型，打通从算法到设备的全链路，形成可工业化的解决方案。 ---