1from transformers import AutoConfig, AutoModel, AutoModelForCausalLM, AutoTokenizer
2import torch
3import torch.nn as nn
4from typing import Optional
5import os
6
7model_name = "kaist-ai/janus-7b"
8reward_model_name = "kaist-ai/janus-rm-7b"
9
10model_device = "cuda:0"
11reward_model_device = "cuda:1"
12
13dtype = "float16"
14if torch.cuda.is_bf16_supported():
15 dtype = "bfloat16"
16
17# Get model and tokenizer
18tokenizer = AutoTokenizer.from_pretrained(model_name)
19model = AutoModelForCausalLM.from_pretrained(model_name, torch_dtype=getattr(torch, dtype), cache_dir="/mnt/sda/suehyun/huggingface")
20model.eval()
21model.to(model_device)
22
23# Get reward model
24def get_reward_model(base_pretrained_model, base_llm_model):
25 class LLMForSequenceRegression(base_pretrained_model):
26 def __init__(self, config: AutoConfig):
27 super().__init__(config)
28 setattr(self, self.base_model_prefix, base_llm_model(config))
29
30 self.value_head = nn.Linear(config.hidden_size, 1, bias=False)
31
32 def forward(
33 self,
34 input_ids: torch.LongTensor = None,
35 attention_mask: Optional[torch.Tensor] = None,
36 return_output=False,
37 ) -> torch.Tensor:
38 position_ids = attention_mask.long().cumsum(-1) - 1
39 position_ids.masked_fill_(attention_mask == 0, 1)
40 outputs = getattr(self, self.base_model_prefix)(
41 input_ids, attention_mask=attention_mask, position_ids=position_ids
42 )
43 last_hidden_states = outputs["last_hidden_state"]
44 values = self.value_head(last_hidden_states).squeeze(-1)
45
46 eos_indices = attention_mask.size(1) - 1 - attention_mask.long().fliplr().argmax(dim=1, keepdim=True)
47 reward = values.gather(dim=1, index=eos_indices).squeeze(1)
48
49 if return_output:
50 return reward, outputs
51 else:
52 return reward
53
54 return LLMForSequenceRegression
55
56
57config = AutoConfig.from_pretrained(reward_model_name)
58config.normalize_reward = True
59
60base_class = AutoModel._model_mapping[type(config)] # <class 'transformers.models.mistral.modeling_mistral.MistralModel'>
61base_pretrained_class = base_class.__base__ # <class 'transformers.models.mistral.modeling_mistral.MistralPreTrainedModel'>
62print(base_class, base_pretrained_class)
63cls_class = get_reward_model(base_pretrained_class,base_class)
64
65reward_model = cls_class.from_pretrained(
66 reward_model_name,
67 config=config,
68 cache_dir="/mnt/sda/suehyun/huggingface",
69 torch_dtype=getattr(torch, dtype),
70)
71print(reward_model)
72reward_model.eval()
73reward_model.to(reward_model_device)
74
75
76# Prepare inputs
77system = "You are a savvy beverage consultant, adept at offering quick, concise drink recommendations that cater to the common palette, yet surprise with a touch of creativity. When approached with a request, your expertise shines by suggesting one or two easily recognizable and widely accessible options, ensuring no one feels overwhelmed by complexity or rarity. Your skill lies not just in meeting the immediate need for refreshment but in gently nudging the curious towards unique hydration choices, beautifully balancing familiarity with the thrill of discovery. Importantly, your recommendations are crafted with a keen awareness of dietary preferences, presenting choices that respect and include considerations for sugar-free, dairy-free, and other common dietary restrictions. Your guidance empowers users to explore a range of beverages, confident they are making informed decisions that respect their health and lifestyle needs."
78prompt = "If you are thirsty, what can you drink to quench your thirst?"
79
80def apply_template_mistral_instruct(system_message, content):
81 prompt = f"{system_message}\n{content}".strip()
82 return f"[INST] {prompt} [/INST] "
83
84input_str = apply_template_mistral_instruct(system, prompt)
85inputs = tokenizer.encode(input_str, return_tensors="pt")
86print(input_str)
87
88model_inputs = inputs.to(model_device)
89
90# Generate text
91with torch.inference_mode():
92 output_ids = model.generate(model_inputs, max_new_tokens=1024)
93decoded = tokenizer.batch_decode(output_ids, skip_special_tokens=True)
94output_str = decoded[0][len(input_str):]
95print(output_str)
96'''
971. **Water**: The ultimate go-to, especially if you're watching what you consume. Opting for sparkling or infused water (think cucumber and mint, berries, or a splash of lemon) can add a bit of excitement and hydration without the added sugar.
98
992. **Herbal Tea**: Perfect for a warmer climate but equally delightful at any temperature. Choose from various flavors, ranging from the traditional peppermint to chamomile or hibiscus, which adds a unique twist with their own health benefits and refreshing flavors. Many options are caffeine-free, making them suitable for all times of the day.
100
101For those needing a touch more sweetness or a slight twist:
102
1033. **Unsweetened Coconut Water**: With its natural sweetness and electrolyte content, it's a great hydration pick after a workout or on a hot day. It's also low in calories and naturally sweet, making it an excellent alternative without added sugars.
104
1054. **Sparkling Water with a Splash of Fruit Juice**: To satisfy a craving for something bubbly and fruit-infused with fewer calories and sugars than commercial sodas or juices. Feel free to experiment with different juices to find your favorite combination.
106'''
107
108# Get reward
109print(input_str + output_str + " " + tokenizer.eos_token)
110reward_inputs = tokenizer(
111 input_str + output_str + " " + tokenizer.eos_token, # same as decoded[0] + " " + tokenizer.eos_token
112 max_length=2048,
113 truncation=True,
114 return_tensors="pt"
115)
116reward_input_ids = reward_inputs.input_ids.to(reward_model_device)
117reward_attention_masks = reward_inputs.attention_mask.to(reward_model_device)
118rewards = reward_model(input_ids=reward_input_ids, attention_mask=reward_attention_masks)
119print(rewards.item())
120# 3.28125
121