Views
No views yet
Move: Move to the adjacent node closest to the center of the field of view.Left: Turn 30° to the left.Right: Turn 30° to the right.Stop: Select when the agent believes it has reached the goal.1 (with gradient accumulation over each episode)1e-50.1bfloat161import json
2import torch
3from torch.utils.data import Dataset, DataLoader
4from datasets import Dataset as DT
5from transformers import Qwen2_5_VLForConditionalGeneration, AutoTokenizer, AutoProcessor
6from PIL import Image
7
8class CustomDataset(Dataset):
9 def __init__(self, data):
10 self.text = data["text"]
11 self.images = data["images"]
12
13 def __len__(self):
14 return len(self.text)
15
16 def __getitem__(self, index):
17 return self.text[index], self.images[index]
18
19class CollateFunctor:
20 # No batch, therefore no max length
21 def __init__(self, processor, width, height):
22 self.processor = processor
23 self.width = width
24 self.height = height
25
26 def __call__(self, batch):
27 text, images = batch[0]
28 label_start = processor.tokenizer("<|im_start|>assistant\nAction: ", return_tensors="pt").input_ids
29
30 images = [Image.open(img).resize((self.width, self.height), Image.Resampling.LANCZOS) for img in images]
31
32 processed = processor(text=text, images=[images], return_tensors="pt")
33
34 prompt_input_ids = processed["input_ids"]
35 input_ids = torch.cat([prompt_input_ids, label_start], dim=1)
36
37 attention_mask = torch.ones(1, input_ids.shape[1])
38 processed["input_ids"] = input_ids
39 processed["attention_mask"] = attention_mask
40
41 return processed
42
43def format_prompt(images_path, step_id, route_instruction, distance_traveled, previous_actions, move_possible, processor, system_prompt):
44 images = os.listdir(images_path)
45 images = [os.path.join(images_path, img) for img in images]
46 images = sorted(images, key=lambda x: int(x.split("_")[-1].split(".")[0]))
47
48 current_image = images.pop(-1)
49
50 content = [
51 {
52 "type" : "text",
53 #"text" : f"Route instruction: {sample['instructions'][instruction_index]}\nPrevious images: "
54 "text" : f"Route Instruction: {route_instruction}\nCurrent Step: {step_id}\nCummulative Distance Traveled: {distance_traveled}\nImages from Previous Steps: "
55 },
56 ]
57
58 for img in images:
59 content.append({"type" : "image", "image" : img})
60
61 if len(images) == 0:
62 content[0]["text"] += f"[]"
63
64 content.append(
65 {
66 "type" : "text",
67 "text" : f"\nActions performed at Previous Steps: {previous_actions.__str__()}\nCurrent image:"
68 }
69 )
70 content.append(
71 {
72 "type" : "image",
73 "image" : current_image
74 }
75 )
76 if move_possible:
77 possible_actions = ["Left", "Right", "Move", "Stop"]
78
79 else:
80 possible_actions = ["Left", "Right", "Stop"]
81
82 content.append(
83 {
84 "type" : "text",
85 "text" : f"\nPossible actions: {possible_actions.__str__()}\nNow predict the next action based on the input you have recived. Answer on the format: Action: (an the action you choose)"
86 }
87 )
88
89 messages = [
90 {"role" : "system", "content" : [{"type" : "text", "text" : system_prompt}]},
91 {"role" : "user", "content" : content},
92 ]
93
94 text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=False)
95 images.extend([current_image])
96
97 formatted_sample = {}
98 formatted_sample["text"] = text
99 formatted_sample["images"] = images
100
101 formatted_data = [formatted_sample]
102 formatted_data = DT.from_list(formatted_data)
103 return formatted_data
104
105# Load model and processor
106processor = AutoProcessor.from_pretrained("Vebbern/Qwen2.5-VL-3B-R2R-low-level")
107model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
108 "Vebbern/Qwen2.5-VL-3B-R2R-low-level",
109 torch_dtype=torch.bfloat16,
110 attn_implementation="flash_attention_2",
111 device_map="cuda"
112)
113
114# remember to set the correct image resolution (however a higher might still work as the vision encoder is not trained)
115collate_fn = CollateFunctor(processor, 320, 240)
116
117# Load mandatory system prompt
118with open("system_prompt.txt", "r") as f:
119 system_prompt = f.read()
120
121path_id = 1021 # id for the R2R path
122route_instruction = "Turn around and keep walking on the hallway across the first doorway and wait at the top of some stairs. "
123images_path = f"./images/{path_id}" # paths to images for the whole episode, images are on the format: step_0.png, step_1.png....
124step_id = 2
125distance = 8.223
126previous_actions = ["Left", "Move"]
127move_possible = True # if there are no nodes within the field of view this should be set to False
128
129# This code will load all images in the path from step 0 up to the current step.
130prompt = format_prompt(images_path, step_id, route_instruction, distance, previous_actions, move_possible, processor, system_prompt)
131
132dataset = CustomDataset(prompt)
133data_loader = DataLoader(
134 dataset,
135 batch_size=1,
136 collate_fn=collate_fn
137)
138
139# Run inference
140for batch in data_loader:
141 batch.to("cuda")
142
143 outputs = model(**batch)
144 argmax = torch.argmax(outputs.logits, dim=2)[0]
145 model_prediction = processor.decode(argmax[-1]) # is -1 because it does not predict one more
146 print(f"Predicted action: {model_prediction}")
147⚠️ Sorry for the rough code — the goal here is to show how the system prompt and inputs should be structured for inference. The system prompt is included in the repo.
| Metric | Val Seen | Val Unseen | Test |
|---|---|---|---|
| Path Length (↓) | 10.27 | 10.50 | 10.59 |
| Navigation Error (↓) | 7.14 | 7.84 | 7.99 |
| Oracle Success Rate (↑) | 41% | 34% | 34% |
| Success Rate (↑) | 35% | 27% | 26% |
| SPL (↑) | 32% | 24% | 24% |