1import json
2import torch
3from transformers import AutoProcessor, AutoModelForImageTextToText
4from huggingface_hub import hf_hub_download
5from PIL import Image
6
7from mr2rm.models.reward_model import MultiResponseRewardModel
8from mr2rm.data.dataset import add_resp_sep_token, RESP_SEP_TOKEN
9
10model_id = "yinuoy/MR2-Molmo2-4B-RM"
11
12# 1. Processor (register the <|resp_sep|> special token)
13processor = AutoProcessor.from_pretrained(
14 model_id, trust_remote_code=True, dtype="auto", device_map="auto",
15)
16add_resp_sep_token(processor.tokenizer)
17
18# 2. Base model + reward model (value-head config from reward_model_config.json)
19base_model = AutoModelForImageTextToText.from_pretrained(
20 model_id, trust_remote_code=True, dtype="auto", device_map="auto",
21)
22config_path = hf_hub_download(repo_id=model_id, filename="reward_model_config.json")
23with open(config_path) as f:
24 rm_config = json.load(f)
25
26reward_model = MultiResponseRewardModel(
27 base_model=base_model,
28 value_head_type=rm_config["value_head_type"],
29 value_head_hidden_dim=rm_config["value_head_hidden_dim"],
30 value_head_activation=rm_config["value_head_activation"],
31 resp_repr_mode=rm_config["resp_repr_mode"],
32)
33
34# 3. Load value-head weights
35vh_path = hf_hub_download(repo_id=model_id, filename="value_head.pt")
36reward_model.value_head.load_state_dict(torch.load(vh_path, map_location="cpu"))
37device = next(reward_model.base_model.parameters()).device
38dtype = next(reward_model.base_model.parameters()).dtype
39reward_model.value_head = reward_model.value_head.to(device=device, dtype=dtype)
40reward_model.eval()
41
42# 4. Build input: user=(prompt+image), assistant=(responses joined by <|resp_sep|>)
43image = Image.open("example.jpg").convert("RGB")
44prompt = "Describe this image."
45responses = [
46 "A golden retriever sitting on grass.",
47 "A dog in a park on a sunny day.",
48 "There is an animal outside.",
49 "I don't know.",
50]
51sep = f"\n\n{RESP_SEP_TOKEN}\n\n"
52assistant_text = sep.join(responses)
53messages = [
54 {"role": "user", "content": [dict(type="image", image=image), dict(type="text", text=prompt)]},
55 {"role": "assistant", "content": [dict(type="text", text=assistant_text)]},
56]
57inputs = processor.apply_chat_template(
58 messages, tokenize=True, return_tensors="pt", return_dict=True,
59)
60inputs = {k: v.to(device) for k, v in inputs.items()}
61
62# 5. Locate the last-token position of each response (just before each <|resp_sep|>, plus end-of-sequence for the final one)
63sep_token_id = processor.tokenizer.convert_tokens_to_ids(RESP_SEP_TOKEN)
64input_ids = inputs["input_ids"][0]
65sep_positions = (input_ids == sep_token_id).nonzero(as_tuple=True)[0].tolist()
66end_positions = [p - 1 for p in sep_positions] + [input_ids.size(0) - 1]
67resp_indices = torch.tensor([end_positions], device=device)
68
69# 6. One forward pass — scores for all N responses
70with torch.inference_mode():
71 (scores,) = reward_model(
72 input_ids=inputs["input_ids"],
73 attention_mask=inputs.get("attention_mask"),
74 resp_indices=resp_indices,
75 **{k: v for k, v in inputs.items() if k not in ["input_ids", "attention_mask"]},
76 )
77rewards = scores[0].tolist()
78
79print("Scores:", rewards)
80print("Best response:", responses[rewards.index(max(rewards))])