File size: 9,890 Bytes
9e62d8e
 
40ba0ea
9f5d69c
9e62d8e
4a9e46f
babcd78
9e62d8e
 
40ba0ea
9e62d8e
40ba0ea
30421b7
 
214fb7b
4a9e46f
4a28ca5
9e62d8e
77b5a47
4a28ca5
9e62d8e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4a9e46f
 
 
 
 
 
 
30421b7
 
4a28ca5
 
 
 
 
 
 
 
4a9e46f
 
 
77b5a47
 
 
 
 
4a28ca5
9e62d8e
 
4a28ca5
30421b7
395e196
30421b7
 
 
 
 
 
 
 
 
 
 
 
 
dc3b7a9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3b00a19
77b5a47
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3b00a19
4a9e46f
 
b9a642a
65da212
8697de8
 
 
 
4a9e46f
 
30421b7
3965e1f
 
 
 
9e62d8e
30421b7
9e62d8e
0b6e0c9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
9e62d8e
 
77b5a47
395e196
3b00a19
3965e1f
3b00a19
4a28ca5
9e62d8e
 
 
4a28ca5
9e62d8e
 
 
30421b7
 
9e62d8e
 
 
 
 
 
 
0b6e0c9
 
 
 
 
 
 
 
 
 
 
 
9f5d69c
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
import re
from pprint import pprint

from transformers import AutoTokenizer

from constants.models import AVAILABLE_MODELS, MODEL_MAP
from tclogger import logger


class MessageComposer:
    def __init__(self, model: str = None):
        if model in AVAILABLE_MODELS:
            self.model = model
        else:
            self.model = "nous-mixtral-8x7b"
        self.model_fullname = MODEL_MAP[self.model]
        self.system_roles = ["system"]
        self.inst_roles = ["user", "system", "inst"]
        self.answer_roles = ["assistant", "bot", "answer", "model"]
        self.default_role = "user"

    def concat_messages_by_role(self, messages):
        def is_same_role(role1, role2):
            if (
                (role1 == role2)
                or (role1 in self.inst_roles and role2 in self.inst_roles)
                or (role1 in self.answer_roles and role2 in self.answer_roles)
            ):
                return True
            else:
                return False

        concat_messages = []
        for message in messages:
            role = message["role"]
            content = message["content"]
            if concat_messages and is_same_role(role, concat_messages[-1]["role"]):
                concat_messages[-1]["content"] += "\n" + content
            else:
                if role in self.inst_roles:
                    message["role"] = "inst"
                elif role in self.answer_roles:
                    message["role"] = "answer"
                else:
                    message["role"] = "inst"
                concat_messages.append(message)
        return concat_messages

    def merge(self, messages) -> str:
        # Templates for Chat Models
        # - https://huggingface.co/docs/transformers/main/en/chat_templating
        #   - https://huggingface.co/mistralai/Mixtral-8x7B-Instruct-v0.1#instruction-format
        #   - https://huggingface.co/NousResearch/Nous-Hermes-2-Mixtral-8x7B-DPO#prompt-format
        #   - https://huggingface.co/openchat/openchat-3.5-0106
        #   - https://huggingface.co/google/gemma-7b-it#chat-template

        # Mistral and Mixtral:
        #   <s> [INST] Instruction [/INST] Model answer </s> [INST] Follow-up instruction [/INST]

        # Nous Mixtral:
        #   <|im_start|>system
        #   You are "Hermes 2".<|im_end|>
        #   <|im_start|>user
        #   Hello, who are you?<|im_end|>
        #   <|im_start|>assistant

        # OpenChat:
        #   GPT4 Correct User: Hello<|end_of_turn|>GPT4 Correct Assistant: Hi<|end_of_turn|>GPT4 Correct User: How are you today?<|end_of_turn|>GPT4 Correct Assistant:

        # Google Gemma-it
        # <start_of_turn>user
        # How does the brain work?<end_of_turn>
        # <start_of_turn>model

        self.messages = messages
        self.merged_str = ""

        # https://huggingface.co/mistralai/Mixtral-8x7B-Instruct-v0.1#instruction-format
        if self.model in ["mixtral-8x7b", "mistral-7b"]:
            self.messages = self.concat_messages_by_role(messages)
            self.cached_str = ""
            for message in self.messages:
                role = message["role"]
                content = message["content"]
                if role in self.inst_roles:
                    self.cached_str = f"[INST] {content} [/INST]"
                elif role in self.answer_roles:
                    self.merged_str += f"<s> {self.cached_str} {content} </s>\n"
                    self.cached_str = ""
                else:
                    self.cached_str = f"[INST] {content} [/INST]"
            if self.cached_str:
                self.merged_str += f"{self.cached_str}"
        # https://huggingface.co/NousResearch/Nous-Hermes-2-Mixtral-8x7B-DPO#prompt-format
        elif self.model in ["nous-mixtral-8x7b"]:
            self.merged_str_list = []
            for message in self.messages:
                role = message["role"]
                content = message["content"]
                if role not in ["system", "user", "assistant"]:
                    role = self.default_role
                message_line = f"<|im_start|>{role}\n{content}<|im_end|>"
                self.merged_str_list.append(message_line)
            self.merged_str_list.append("<|im_start|>assistant")
            self.merged_str = "\n".join(self.merged_str_list)
        # https://huggingface.co/openchat/openchat-3.5-0106
        elif self.model in ["openchat-3.5"]:
            self.messages = self.concat_messages_by_role(messages)
            self.merged_str_list = []
            self.end_of_turn = "<|end_of_turn|>"
            for message in self.messages:
                role = message["role"]
                content = message["content"]
                if role in self.inst_roles:
                    self.merged_str_list.append(
                        f"GPT4 Correct User:\n{content}{self.end_of_turn}"
                    )
                elif role in self.answer_roles:
                    self.merged_str_list.append(
                        f"GPT4 Correct Assistant:\n{content}{self.end_of_turn}"
                    )
                else:
                    self.merged_str_list.append(
                        f"GPT4 Correct User: {content}{self.end_of_turn}"
                    )
            self.merged_str_list.append(f"GPT4 Correct Assistant:\n")
            self.merged_str = "\n".join(self.merged_str_list)
        # https://huggingface.co/google/gemma-1.1-7b-it#chat-template
        elif self.model in ["gemma-7b"]:
            self.messages = self.concat_messages_by_role(messages)
            self.merged_str_list = []
            self.end_of_turn = "<end_of_turn>"
            self.start_of_turn = "<start_of_turn>"
            for message in self.messages:
                role = message["role"]
                content = message["content"]
                if role in self.inst_roles:
                    self.merged_str_list.append(
                        f"{self.start_of_turn}user\n{content}{self.end_of_turn}"
                    )
                elif role in self.answer_roles:
                    self.merged_str_list.append(
                        f"{self.start_of_turn}model\n{content}{self.end_of_turn}"
                    )
                else:
                    self.merged_str_list.append(
                        f"{self.start_of_turn}user\n{content}{self.end_of_turn}"
                    )
            self.merged_str_list.append(f"{self.start_of_turn}model\n")
            self.merged_str = "<bos>" + "\n".join(self.merged_str_list)
        # https://huggingface.co/NousResearch/Nous-Hermes-2-Mixtral-8x7B-DPO#prompt-format
        # https://huggingface.co/openchat/openchat-3.5-0106
        # https://huggingface.co/01-ai/Yi-1.5-34B-Chat
        elif self.model in ["openchat-3.5", "command-r-plus", "gemma-7b", "yi-1.5-34b", "llama3-8b", "llama3-70b"]:
            # https://discuss.huggingface.co/t/error-with-new-tokenizers-urgent/2847/5
            tokenizer = AutoTokenizer.from_pretrained(
                self.model_fullname, use_fast=False
            )
            self.merged_str = tokenizer.apply_chat_template(
                messages, tokenize=False, add_generation_prompt=True
            )
        else:
            self.merged_str = "\n\n".join(
                [f"{message['role']}: {message['content']}" for message in messages]
            )

        return self.merged_str

    def decompose_to_system_and_input_prompt(
        self, messages: list[dict], append_assistant=True
    ):
        system_prompt_list = []
        user_and_assistant_messages = []
        for message in messages:
            role = message["role"]
            content = message["content"]
            if role in self.system_roles:
                system_prompt_list.append(content)
            else:
                user_and_assistant_messages.append(message)
        system_prompt = "\n".join(system_prompt_list)

        input_prompt_list = []
        input_messages = self.concat_messages_by_role(user_and_assistant_messages)
        for message in input_messages:
            role = message["role"]
            content = message["content"]
            if role in self.answer_roles:
                role_content_str = f"`assistant`:\n{content}"
            else:
                role_content_str = f"`user`:\n{content}"
            input_prompt_list.append(role_content_str)
        input_prompt = "\n\n".join(input_prompt_list)

        if append_assistant:
            input_prompt += "\n\n`assistant`:"

        return system_prompt, input_prompt


if __name__ == "__main__":
    # model = "mixtral-8x7b"
    # model = "nous-mixtral-8x7b"
    model = "gemma-7b"
    # model = "openchat-3.5"
    # model = "command-r-plus"
    composer = MessageComposer(model)
    messages = [
        {
            "role": "system",
            "content": "You are a LLM developed by OpenAI.\nYour name is GPT-4.",
        },
        {"role": "user", "content": "Hello, who are you?"},
        {"role": "assistant", "content": "I am a bot."},
        {"role": "user", "content": "What is your name?"},
        # {"role": "assistant", "content": "My name is Bing."},
        # {"role": "user", "content": "Tell me a joke."},
        # {"role": "assistant", "content": "What is a robot's favorite type of music?"},
        # {
        #     "role": "user",
        #     "content": "How many questions have I asked? Please list them.",
        # },
    ]
    # logger.note(f"model: {composer.model}")
    # merged_str = composer.merge(messages)
    # logger.note("merged_str:")
    # logger.mesg(merged_str)

    system_prompt, input_prompt = composer.decompose_to_system_and_input_prompt(
        messages
    )
    logger.note("system_prompt:")
    logger.mesg(system_prompt)
    logger.note("input_prompt:")
    logger.mesg(input_prompt)

    # python -m messagers.message_composer