Overview
The quest for fully autonomous vehicles (AVs) capable of navigating complex real-world scenarios with human-like understanding and responsiveness. In this paper, we introduce Dolphins, a novel vision-language model architected to imbibe human-like driving abilities. Dolphins is adept at processing multimodal inputs comprising video (or image) data, text instructions, and historical control signals to generate informed outputs corresponding to the provided instructions. Building upon the open-sourced pretrained Vision-Language Model, OpenFlamingo, we tailored Dolphins to the driving domain by constructing driving-specific instruction data and conducting instruction tuning. Through the utilization of the BDD-X dataset, we designed and consolidated four distinct AV tasks into Dolphins to foster a holistic understanding of intricate driving scenarios. As a result, the distinctive features of Dolphins are delineated into two dimensions: (1) the ability to provide a comprehensive understanding of complex and long-tailed open-world driving scenarios and solve a spectrum of AV tasks, and (2) the emergence of human-like capabilities including gradient-free rapid learning and adaptation via in-context learning, reflection and error recovery, and interoperability.
Initialization
from mllm.src.factory import create_model_and_transforms
from configs.lora_config import openflamingo_tuning_config
peft_config, peft_model_id = None, None
peft_config = LoraConfig(**openflamingo_tuning_config)
model, image_processor, tokenizer = create_model_and_transforms(
clip_vision_encoder_path="ViT-L-14-336",
clip_vision_encoder_pretrained="openai",
lang_encoder_path="anas-awadalla/mpt-7b", # anas-awadalla/mpt-7b
tokenizer_path="anas-awadalla/mpt-7b", # anas-awadalla/mpt-7b
cross_attn_every_n_layers=4,
use_peft=True,
peft_config=peft_config,
)
# grab model checkpoint from huggingface hub
from huggingface_hub import hf_hub_download
import torch
checkpoint_path = hf_hub_download("gray311/Dolphins", "checkpoint. pt")
model.load_state_dict(torch.load(checkpoint_path), strict=False)
Generation example
Below is an example of generating text conditioned on driving videos.
import os
import json
import argparse
import pandas as pd
from tqdm import tqdm
from typing import Union
from PIL import Image
import mimetypes
import cv2
import torch
from torch.utils.data import DataLoader
import transformers
from transformers import LlamaTokenizer, CLIPImageProcessor
from configs.dataset_config import DATASET_CONFIG
from configs.lora_config import openflamingo_tuning_config, otter_tuning_config
from mllm.src.factory import create_model_and_transforms
from mllm.otter.modeling_otter import OtterConfig, OtterForConditionalGeneration
from huggingface_hub import hf_hub_download
from peft import (
get_peft_model,
LoraConfig,
get_peft_model_state_dict,
PeftConfig,
PeftModel
)
def setup_seed(seed):
torch.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
np.random.seed(seed)
random.seed(seed)
torch.backends.cudnn.deterministic = True
def get_content_type(file_path):
content_type, _ = mimetypes.guess_type(file_path)
return content_type
# ------------------- Image and Video Handling Functions -------------------
def extract_frames(video_path, num_frames=16):
video = cv2.VideoCapture(video_path)
total_frames = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
frame_step = total_frames // num_frames
frames = []
for i in range(num_frames):
video.set(cv2.CAP_PROP_POS_FRAMES, i * frame_step)
ret, frame = video.read()
if ret:
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = Image.fromarray(frame).convert("RGB")
frames.append(frame)
video.release()
return frames
def get_image(url: str) -> Union[Image.Image, list]:
if "://" not in url: # Local file
content_type = get_content_type(url)
else: # Remote URL
content_type = requests.head(url, stream=True, verify=False).headers.get("Content-Type")
if "image" in content_type:
if "://" not in url: # Local file
return Image.open(url)
else: # Remote URL
return Image.open(requests.get(url, stream=True, verify=False).raw)
elif "video" in content_type:
video_path = "temp_video.mp4"
if "://" not in url: # Local file
video_path = url
else: # Remote URL
with open(video_path, "wb") as f:
f.write(requests.get(url, stream=True, verify=False).content)
frames = extract_frames(video_path)
if "://" in url: # Only remove the temporary video file if it was downloaded
os.remove(video_path)
return frames
else:
raise ValueError("Invalid content type. Expected image or video.")
def load_pretrained_modoel():
peft_config, peft_model_id = None, None
peft_config = LoraConfig(**openflamingo_tuning_config)
model, image_processor, tokenizer = create_model_and_transforms(
clip_vision_encoder_path="ViT-L-14-336",
clip_vision_encoder_pretrained="openai",
lang_encoder_path="anas-awadalla/mpt-7b", # anas-awadalla/mpt-7b
tokenizer_path="anas-awadalla/mpt-7b", # anas-awadalla/mpt-7b
cross_attn_every_n_layers=4,
use_peft=True,
peft_config=peft_config,
)
checkpoint_path = hf_hub_download("gray311/Dolphins", "checkpoint.pt")
model.load_state_dict(torch.load(checkpoint_path), strict=False)
model.half().cuda()
return model, image_processor, tokenizer
def get_model_inputs(video_path, instruction, model, image_processor, tokenizer):
frames = get_image(video_path)
vision_x = torch.stack([image_processor(image) for image in frames], dim=0).unsqueeze(0).unsqueeze(0)
assert vision_x.shape[2] == len(frames)
prompt = [
f"USER: <image> is a driving video. {instruction} GPT:<answer>"
]
inputs = tokenizer(prompt, return_tensors="pt", ).to(model.device)
return vision_x, inputs
if __name__ == "__main__":
video_path = "path/to/your/video"
instruction = "Please describe this video in detail."
model, image_processor, tokenizer = load_pretrained_modoel()
vision_x, inputs = get_model_inputs(video_path, instruction, model, image_processor, tokenizer)
generation_kwargs = {'max_new_tokens': 512, 'temperature': 1,
'top_k': 0, 'top_p': 1, 'no_repeat_ngram_size': 3, 'length_penalty': 1,
'do_sample': False,
'early_stopping': True}
generated_tokens = model.generate(
vision_x=vision_x.half().cuda(),
lang_x=inputs["input_ids"].cuda(),
attention_mask=inputs["attention_mask"].cuda(),
num_beams=3,
**generation_kwargs,
)
generated_tokens = generated_tokens.cpu().numpy()
if isinstance(generated_tokens, tuple):
generated_tokens = generated_tokens[0]
generated_text = tokenizer.batch_decode(generated_tokens)
print(
f"Dolphin output:\n\n{generated_text}"
)
- Downloads last month
- 10