GDU-Liang-4B is a four-modality (text, image, video, audio) incremental fine-tune of 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
#或者运行下面：
!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 uninstall torchao -y

Text-Only Inference (示例)

python

import torch
from transformers import AutoTokenizer, AutoModelForImageTextToText

model_id = "zhenqiangliang6/GDU-Liang-4B"

# 加载模型和分词器
tokenizer = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True)
model = AutoModelForImageTextToText.from_pretrained(
    model_id,
    torch_dtype=torch.float16,
    device_map="auto",
    trust_remote_code=True,
)

# 定义推理函数
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)

    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()

# 测试
print(generate("你是谁？"))

输出

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

#!/usr/bin/env python3
"""
GDU-liang-4B Video Understanding - v9.26 Production Edition
✅ 自动同步缺失 modeling 文件到 transformers_modules 缓存
✅ 优先加载 ForCausalLM（跳过 base Molmo2Model）
✅ Embedding + LM Head 同步扩展
✅ A10 Safe: 8帧 + bf16 + CPU offload (24GB GPU)
✅ <|video|> placeholder 硬编码
这是显存大于24GB的时候最优运行代码（显存富裕请使用这代码）
"""

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 缓存"""
    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:
        if os.path.isdir(os.path.dirname(d)):
            target_dir = d
            break

    if target_dir is None:
        target_dir = possible_dirs[0]
        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}")
        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")


# ==================== 模型类加载 ====================

def load_model_class(model_dir):
    """优先选含 generate() 的 CausalLM 类，跳过 base Model"""
    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}!")

    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"   ✅ Model: {chosen_name}")
    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

    if tok_vocab == model_vocab:
        return

    # --- 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

    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: [{cur_rows}x{hidden}] → [{tok_vocab}x{hidden}]")

    # --- Expand LM Head ---
    expanded = False

    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]
                    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

    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

    if not expanded:
        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):
                    expanded = True
                    break

    safe_set_vocab_size(cfg, tok_vocab)
    safe_set_vocab_size(model.config, tok_vocab)
    print(f"   ✅ Vocab aligned to {tok_vocab}")


# ==================== 视频处理 ====================

def load_and_sample_video(video_path, max_frames=8):
    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"   📹 Sampled {n}/{total} frames @ {fps:.1f} FPS")
    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)

    # 🔑 24GB GPU 显存安全策略
    import gc
    gc.collect()
    torch.cuda.empty_cache()

    max_memory = {0: "20GiB", "cpu": "30GiB"}

    model = ModelClass.from_pretrained(
        model_dir,
        config=cfg,
        dtype=torch.bfloat16,
        device_map="auto",
        max_memory=max_memory,
        offload_folder="/tmp/offload",
        trust_remote_code=True,
    )
    model.eval()

    # 确保 meta tensor 全部物化后再 expand vocab
    if any(p.device.type == "meta" for p in model.parameters()):
        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:
            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)

    # Processor 加载
    from transformers import AutoProcessor
    try:
        processor = AutoProcessor.from_pretrained(model_dir, trust_remote_code=True)
    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)
                break
        else:
            raise RuntimeError(f"Cannot load processor! AutoProcessor failed: {e}")

    print(f"   ✅ Processor: {type(processor).__name__}")

    # 🔑 Molmo2-GDU-liang-4B 确认使用 <|video|>
    video_placeholder = "<|video|>"
    messages = [{"role": "user", "content": f"{video_placeholder}\n{user_query}"}]
    prompt_text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)

    # Molmo2VideoProcessor 期望 (T, H, W, C)
    video_tensor = frames
    fps_value = 30.0
    num_frames = video_tensor.shape[0]

    # 多策略调用 processor
    proc_inputs = None
    try:
        proc_inputs = processor(
            text=prompt_text, videos=video_tensor,
            video_metadata=[{"fps": fps_value, "total_num_frames": num_frames}],
            return_tensors="pt",
        )
    except Exception:
        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",
            )
        except Exception:
            proc_inputs = processor(
                text=prompt_text, videos=video_tensor,
                fps=fps_value, return_tensors="pt",
            )

    # device_map="auto" 时用首个参数的设备
    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 ({list(inp.keys())})...")

    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()


# ==================== 入口 ====================

if __name__ == "__main__":
    print("=" * 60)
    print("【🚀 v9.26 - GDU-liang-4B Video Understanding】")
    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)
## 🎯 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，而是可工业化、可验证的端侧具身智能系统。

使用方法（简要）

安装依赖：pip install -r requirements.txt
加载模型与 tokenizer（参见上文 Python 示例）
使用 examples/ 中的脚本完成多模态输入的推理验证

最终愿景

以广东外语外贸大学学生及Alphapilot OS团队成员的身份，构建国产轻量化具身智能模型，打通从算法到设备的全链路，形成可工业化的解决方案。

zhenqiangliang6/GDU-liang-4B

GDU-Liang-4B

🔍 Overview

Why This Matters

🏗️ Technical Highlights

🚀 Quick Start

Installation

Text-Only Inference (示例)

输出

Multimodal Inference (Image + Text)图文理解

1.Example Dataset Download 例子数据集下载

2.Example Dataset Standardization 例子数据集标准化

GDU Model Video Inference Demo (Optimized for A10 24GB) GDU模型视频推理运行展示，适合a10 24G

🎬 Video Temporal Understanding

Quick Start

⚠️ Limitations & Ethics

📂 目录建议

🛠️ 实验与部署小贴士

GDU-Liang-4B 多模态模型（中文完整版）

模型简介

技术履历与动机

开源愿景与工程观

使用方法（简要）

最终愿景

zhenqiangliang6/
GDU-liang-4B