Build tensor inputs from text that is interleaved with tokens from other modalities Supports multi-annotation data where we have many annotations for one multi-modal input.
| 71 | |
| 72 | @dataclasses.dataclass |
| 73 | class InterleavedTextPreprocessor: |
| 74 | """ |
| 75 | Build tensor inputs from text that is interleaved with tokens from other modalities |
| 76 | |
| 77 | Supports multi-annotation data where we have many annotations for one multi-modal input. |
| 78 | """ |
| 79 | tokenizer: Any = None |
| 80 | max_text_tokens: Optional[int] = None |
| 81 | max_sequence_length: Optional[int] = None |
| 82 | last_message_loss_only: bool = False |
| 83 | max_answer_len: int = None |
| 84 | default_message_weight: Optional[MessageWeight] = dataclasses.field(default_factory=MessageWeight) |
| 85 | |
| 86 | def tokenize_message(self, message_list: List[str], weight, bos=True, add_last_eos=True): |
| 87 | if bos: |
| 88 | bos = self.tokenizer.bos_token_id or self.tokenizer.eos_token_id |
| 89 | text_token_ids = [bos] |
| 90 | loss_mask = [0.0] |
| 91 | else: |
| 92 | text_token_ids = [] |
| 93 | loss_mask = [] |
| 94 | for msg_ix, message in enumerate(message_list): |
| 95 | message_ids = self.tokenizer.encode(message) |
| 96 | is_model = msg_ix % 2 == 1 |
| 97 | if is_model and (add_last_eos or msg_ix != len(message_list) - 1): |
| 98 | message_ids.append(self.tokenizer.eos_token_id) |
| 99 | |
| 100 | if is_model and self.max_answer_len: |
| 101 | message_ids = message_ids[:self.max_answer_len] |
| 102 | |
| 103 | has_loss = is_model and ( |
| 104 | not self.last_message_loss_only or (msg_ix == (len(message_list) - 1))) |
| 105 | loss_mask += [has_loss] * len(message_ids) |
| 106 | text_token_ids += message_ids |
| 107 | text_token_ids = np.array(text_token_ids) |
| 108 | is_prompt = text_token_ids == self.tokenizer.image_prompt_token_id |
| 109 | loss_mask = np.array(loss_mask, dtype=np.float32) |
| 110 | if weight.root_length: |
| 111 | if loss_mask.sum() > 0: |
| 112 | loss_mask *= 2 / np.sqrt(loss_mask.sum()) |
| 113 | if weight.weight is not None: |
| 114 | loss_mask *= weight.weight |
| 115 | return text_token_ids, loss_mask |
| 116 | |
| 117 | def tokenize_message_list( |
| 118 | self, |
| 119 | message_list: Union[List[str], List[List[str]]], |
| 120 | n_mm_tokens: int, |
| 121 | num_images: int = 1, |
| 122 | weights: List[MessageWeight] = None, |
| 123 | ): |
| 124 | """Handle multi-annotation data where we have many annotations for one multi-modal input""" |
| 125 | assert len(message_list) > 0, "Given empty messages" |
| 126 | # Multi-annotation data where we have many annotations for one multi-modal input |
| 127 | before_ids = [] |
| 128 | after_ids = [] |
| 129 | before_losses = [] |
| 130 | after_losses = [] |
no outgoing calls