Views
No views yet
pip install transformers torch huggingface_hub1import json, torch
2from huggingface_hub import hf_hub_download
3from transformers import Qwen2_5OmniThinkerForConditionalGeneration, AutoProcessor
4
5MODEL_ID = "keentomato/omnisapiens_bam_sentiment_polarity_mosei"
6
7# 1. Load backbone and processor
8model = Qwen2_5OmniThinkerForConditionalGeneration.from_pretrained(
9 MODEL_ID, torch_dtype=torch.float16, device_map="auto"
10)
11processor = AutoProcessor.from_pretrained(MODEL_ID)
12
13# 2. Load classification heads and label scheme
14heads_path = hf_hub_download(MODEL_ID, "heads.bin")
15label_path = hf_hub_download(MODEL_ID, "label_scheme.json")
16heads_sd = torch.load(heads_path, map_location="cpu")
17with open(label_path) as f:
18 label_scheme = json.load(f)
19
20# 3. Reconstruct domain heads
21global_classes = label_scheme["meta"]["global_classes"] # {domain: [{index, label}, ...]}
22hidden_size = model.config.hidden_size
23domain_names = list(global_classes.keys())
24domain_heads = torch.nn.ModuleList([
25 torch.nn.Linear(hidden_size, len(global_classes[d])) for d in domain_names
26])
27domain_heads.load_state_dict({k.replace("heads.", ""): v for k, v in heads_sd.items()})
28domain_heads.eval().to(model.device).to(torch.float16)
29domain_to_id = {d: i for i, d in enumerate(domain_names)}
30
31# 4. Prepare multimodal inputs
32# video_tensor: [T, C, H, W] tensor or list of PIL images
33# audio_waveform: 1-D numpy array / tensor at 16 kHz
34domain = "sentiment_intensity"
35messages = [{"role": "user", "content": [
36 {"type": "video"},
37 {"type": "audio"},
38 {"type": "text", "text": "Classify the human behavior expressed."},
39]}]
40text = processor.apply_chat_template(messages, add_generation_prompt=False, tokenize=False)
41inputs = processor(text=[text], videos=[video_tensor], audio=[audio_waveform], return_tensors="pt")
42inputs = {k: v.to(model.device) for k, v in inputs.items()}
43
44# 5. Forward pass — pool penultimate hidden layer, route through domain head
45with torch.no_grad():
46 out = model(**inputs, output_hidden_states=True, use_cache=False)
47 h = out.hidden_states[-2] # [B, T, H]
48 mask = inputs["attention_mask"].unsqueeze(-1).float()
49 pooled = (h * mask).sum(1) / mask.sum(1) # [B, H]
50 logits = domain_heads[domain_to_id[domain]](pooled.float()) # [B, K_d]
51 pred_idx = logits.argmax(dim=-1).item()
52
53label_name = global_classes[domain][pred_idx]["label"]
54print(f"Predicted {domain}: {label_name}")adapters.bin is present in the repo, the model supports side-channel
behavioral descriptors extracted from OpenPose (video) and OpenSmile (audio).
These replace the raw video/audio inputs to the backbone with pre-computed
behavioral feature vectors that are injected via lightweight MLP adapters.pose, face, left_hand, right_hand,
each a [T, K, 2or3] tensor (T frames, K keypoints, x/y/conf).1def prepare_video_feats(openpose_dict, temporal_mode="meanstd"):
2 """OpenPose dict → pooled feature vector [D_v_pooled]."""
3 parts = []
4 for key in ("pose", "face", "left_hand", "right_hand"):
5 t = openpose_dict.get(key) # [T, K, 2or3]
6 if t is None: continue
7 t = torch.as_tensor(t).float()[..., :2] # drop confidence, keep x/y
8 parts.append(t.reshape(t.shape[0], -1)) # [T, K*2]
9 seq = torch.cat(parts, dim=-1).float() # [T, D_v]
10 if temporal_mode == "meanstd":
11 return torch.cat([seq.mean(0), seq.std(0)]) # [D_v*2]
12 return seq.mean(0) # [D_v]
13
14video_feats = prepare_video_feats(openpose_dict).unsqueeze(0) # [1, D_v_pooled]features → [T, D_a] or [D_a].1def prepare_audio_feats(opensmile_dict):
2 """OpenSmile dict → L2-normalised feature vector [D_a]."""
3 x = torch.as_tensor(opensmile_dict["features"]).float()
4 if x.ndim == 2: x = x.squeeze(0) # [D_a] (single frame assumed)
5 return x / x.norm(p=2).clamp_min(1e-6)
6
7audio_feats = prepare_audio_feats(opensmile_dict).unsqueeze(0) # [1, D_a]1import torch, torch.nn as nn
2from huggingface_hub import hf_hub_download
3
4adapters_sd = torch.load(hf_hub_download(MODEL_ID, "adapters.bin"), map_location="cpu")
5
6# Infer architecture from saved weight shapes — no config needed
7def _make_adapter(prefix, sd):
8 w0 = sd[f"{prefix}.mlp.0.weight"] # [hidden, feat_dim]
9 w2 = sd[f"{prefix}.mlp.2.weight"] # [out_dim, hidden]
10 feat_dim, hidden, out_dim = w0.shape[1], w0.shape[0], w2.shape[0]
11 mlp = nn.Sequential(nn.Linear(feat_dim, hidden), nn.ReLU(), nn.Linear(hidden, out_dim))
12 alpha = nn.Parameter(sd[f"{prefix}.alpha"])
13 class _Adapter(nn.Module):
14 def __init__(self): super().__init__(); self.mlp = mlp; self.alpha = alpha
15 def forward(self, x): return self.mlp(x) * self.alpha
16 m = _Adapter()
17 m.load_state_dict({k[len(prefix)+1:]: v for k, v in sd.items() if k.startswith(prefix)}, strict=False)
18 return m.eval()
19
20video_adapter = _make_adapter("video_adapter", adapters_sd).to(model.device).half()
21audio_adapter = _make_adapter("audio_adapter", adapters_sd).to(model.device).half()
22
23# Augment pooled repr with BAM deltas before the classification head
24with torch.no_grad():
25 out = model(**inputs, output_hidden_states=True, use_cache=False)
26 h = out.hidden_states[-2]
27 mask = inputs["attention_mask"].unsqueeze(-1).float()
28 pooled = (h * mask).sum(1) / mask.sum(1) # [B, H]
29 pooled = pooled + video_adapter(video_feats.to(model.device).half())
30 pooled = pooled + audio_adapter(audio_feats.to(model.device).half())
31 logits = domain_heads[domain_to_id[domain]](pooled.float())
32 pred_idx = logits.argmax(dim=-1).item()
33
34label_name = global_classes[domain][pred_idx]["label"]
35print(f"Predicted {domain}: {label_name}")