cleanup: trim open_clip to inference-only (remove loss, tokenizer, linear_probe, bert)

models/CLAP/open_clip/__init__.py

@@ -1,25 +1,3 @@
-from .factory import (
-    list_models,
-    create_model,
-    create_model_and_transforms,
-    add_model_config,
-)
-from .loss import ClipLoss, gather_features, LPLoss, lp_gather_features, LPMetrics
-from .model import (
-    CLAP,
-    CLAPTextCfg,
-    CLAPVisionCfg,
-    CLAPAudioCfp,
-    convert_weights_to_fp16,
-    trace_model,
-)
-from .openai import load_openai_model, list_openai_models
-from .pretrained import (
-    list_pretrained,
-    list_pretrained_tag_models,
-    list_pretrained_model_tags,
-    get_pretrained_url,
-    download_pretrained,
-)
-from .tokenizer import SimpleTokenizer, tokenize
-from .transform import image_transform
+from .factory import create_model
+from .model import CLAP, CLAPTextCfg, CLAPVisionCfg, CLAPAudioCfp, convert_weights_to_fp16
+# removed: loss, tokenizer, openai, pretrained, transform re-exports — training/loading helpers not used at inference

models/CLAP/open_clip/bert.py

@@ -1,40 +0,0 @@
-from transformers import BertTokenizer, BertModel
-
-tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
-model = BertModel.from_pretrained("bert-base-uncased")
-text = "Replace me by any text you'd like."
-
-
-def bert_embeddings(text):
-    # text = "Replace me by any text you'd like."
-    encoded_input = tokenizer(text, return_tensors="pt")
-    output = model(**encoded_input)
-    return output
-
-
-from transformers import RobertaTokenizer, RobertaModel
-
-tokenizer = RobertaTokenizer.from_pretrained("roberta-base")
-model = RobertaModel.from_pretrained("roberta-base")
-text = "Replace me by any text you'd like."
-
-
-def Roberta_embeddings(text):
-    # text = "Replace me by any text you'd like."
-    encoded_input = tokenizer(text, return_tensors="pt")
-    output = model(**encoded_input)
-    return output
-
-
-from transformers import BartTokenizer, BartModel
-
-tokenizer = BartTokenizer.from_pretrained("facebook/bart-base")
-model = BartModel.from_pretrained("facebook/bart-base")
-text = "Replace me by any text you'd like."
-
-
-def bart_embeddings(text):
-    # text = "Replace me by any text you'd like."
-    encoded_input = tokenizer(text, return_tensors="pt")
-    output = model(**encoded_input)
-    return output

models/CLAP/open_clip/linear_probe.py

@@ -1,66 +0,0 @@
-import numpy as np
-import torch.nn.functional as F
-from torch import nn
-from .model import MLPLayers
-
-
-class LinearProbe(nn.Module):
-    def __init__(self, model, mlp, freeze, in_ch, out_ch, act=None):
-        """
-        Args:
-            model: nn.Module
-            mlp: bool, if True, then use the MLP layer as the linear probe module
-            freeze: bool, if True, then freeze all the CLAP model's layers when training the linear probe
-            in_ch: int, the output channel from CLAP model
-            out_ch: int, the output channel from linear probe (class_num)
-            act: torch.nn.functional, the activation function before the loss function
-        """
-        super().__init__()
-        in_ch = 512
-        self.clap_model = model
-        self.clap_model.text_branch = None  # to save memory
-        self.freeze = freeze
-        if mlp:
-            self.lp_layer = MLPLayers(units=[in_ch, in_ch * 2, out_ch])
-        else:
-            self.lp_layer = nn.Linear(in_ch, out_ch)
-
-        if self.freeze:
-            for param in self.clap_model.parameters():
-                param.requires_grad = False
-
-        if act == "None":
-            self.act = None
-        elif act == "relu":
-            self.act = nn.ReLU()
-        elif act == "elu":
-            self.act = nn.ELU()
-        elif act == "prelu":
-            self.act = nn.PReLU(num_parameters=in_ch)
-        elif act == "softmax":
-            self.act = nn.Softmax(dim=-1)
-        elif act == "sigmoid":
-            self.act = nn.Sigmoid()
-
-    def forward(self, x, mix_lambda=None, device=None):
-        """
-        Args:
-            x: waveform, torch.tensor [batch, t_samples] / batch of mel_spec and longer list
-            mix_lambda: torch.tensor [batch], the mixup lambda
-        Returns:
-            class_prob: torch.tensor [batch, class_num]
-
-        """
-        # batchnorm cancel grandient
-        if self.freeze:
-            self.clap_model.eval()
-
-        x = self.clap_model.audio_projection(
-            self.clap_model.audio_branch(x, mixup_lambda=mix_lambda, device=device)[
-                "embedding"
-            ]
-        )
-        out = self.lp_layer(x)
-        if self.act is not None:
-            out = self.act(out)
-        return out

models/CLAP/open_clip/loss.py

@@ -1,398 +0,0 @@
-from multiprocessing.sharedctypes import Value
-import torch
-import torch.distributed.nn
-from torch import distributed as dist, nn as nn
-from torch.nn import functional as F
-import numpy as np
-from sklearn.metrics import average_precision_score, roc_auc_score, accuracy_score
-
-try:
-    import horovod.torch as hvd
-except ImportError:
-    hvd = None
-
-
-def gather_features(
-    audio_features,
-    text_features,
-    audio_features_mlp=None,
-    text_features_mlp=None,
-    local_loss=False,
-    gather_with_grad=False,
-    rank=0,
-    world_size=1,
-    use_horovod=False,
-    mlp_loss=False,
-):
-    if use_horovod:
-        assert hvd is not None, "Please install horovod"
-        if gather_with_grad:
-            all_audio_features = hvd.allgather(audio_features)
-            all_text_features = hvd.allgather(text_features)
-            if mlp_loss:
-                all_audio_features_mlp = hvd.allgather(audio_features_mlp)
-                all_text_features_mlp = hvd.allgather(text_features_mlp)
-        else:
-            with torch.no_grad():
-                all_audio_features = hvd.allgather(audio_features)
-                all_text_features = hvd.allgather(text_features)
-                if mlp_loss:
-                    all_audio_features_mlp = hvd.allgather(audio_features_mlp)
-                    all_text_features_mlp = hvd.allgather(text_features_mlp)
-            if not local_loss:
-                # ensure grads for local rank when all_* features don't have a gradient
-                gathered_audio_features = list(
-                    all_audio_features.chunk(world_size, dim=0)
-                )
-                gathered_text_features = list(
-                    all_text_features.chunk(world_size, dim=0)
-                )
-                gathered_audio_features[rank] = audio_features
-                gathered_text_features[rank] = text_features
-                all_audio_features = torch.cat(gathered_audio_features, dim=0)
-                all_text_features = torch.cat(gathered_text_features, dim=0)
-                if mlp_loss:
-                    gathered_audio_features_mlp = list(
-                        all_audio_features_mlp.chunk(world_size, dim=0)
-                    )
-                    gathered_text_features_mlp = list(
-                        all_text_features_mlp.chunk(world_size, dim=0)
-                    )
-                    gathered_audio_features_mlp[rank] = audio_features_mlp
-                    gathered_text_features_mlp[rank] = text_features_mlp
-                    all_audio_features_mlp = torch.cat(
-                        gathered_audio_features_mlp, dim=0
-                    )
-                    all_text_features_mlp = torch.cat(gathered_text_features_mlp, dim=0)
-    else:
-        # We gather tensors from all gpus
-        if gather_with_grad:
-            all_audio_features = torch.cat(
-                torch.distributed.nn.all_gather(audio_features), dim=0
-            )
-            all_text_features = torch.cat(
-                torch.distributed.nn.all_gather(text_features), dim=0
-            )
-            if mlp_loss:
-                all_audio_features_mlp = torch.cat(
-                    torch.distributed.nn.all_gather(audio_features_mlp), dim=0
-                )
-                all_text_features_mlp = torch.cat(
-                    torch.distributed.nn.all_gather(text_features_mlp), dim=0
-                )
-        else:
-            gathered_audio_features = [
-                torch.zeros_like(audio_features) for _ in range(world_size)
-            ]
-            gathered_text_features = [
-                torch.zeros_like(text_features) for _ in range(world_size)
-            ]
-            dist.all_gather(gathered_audio_features, audio_features)
-            dist.all_gather(gathered_text_features, text_features)
-            if mlp_loss:
-                gathered_audio_features_mlp = [
-                    torch.zeros_like(audio_features_mlp) for _ in range(world_size)
-                ]
-                gathered_text_features_mlp = [
-                    torch.zeros_like(text_features_mlp) for _ in range(world_size)
-                ]
-                dist.all_gather(gathered_audio_features_mlp, audio_features_mlp)
-                dist.all_gather(gathered_text_features_mlp, text_features_mlp)
-            if not local_loss:
-                # ensure grads for local rank when all_* features don't have a gradient
-                gathered_audio_features[rank] = audio_features
-                gathered_text_features[rank] = text_features
-                if mlp_loss:
-                    gathered_audio_features_mlp[rank] = audio_features_mlp
-                    gathered_text_features_mlp[rank] = text_features_mlp
-
-            all_audio_features = torch.cat(gathered_audio_features, dim=0)
-            all_text_features = torch.cat(gathered_text_features, dim=0)
-            if mlp_loss:
-                all_audio_features_mlp = torch.cat(gathered_audio_features_mlp, dim=0)
-                all_text_features_mlp = torch.cat(gathered_text_features_mlp, dim=0)
-    if mlp_loss:
-        return (
-            all_audio_features,
-            all_text_features,
-            all_audio_features_mlp,
-            all_text_features_mlp,
-        )
-    else:
-        return all_audio_features, all_text_features
-
-
-class ClipLoss(nn.Module):
-    def __init__(
-        self,
-        local_loss=False,
-        gather_with_grad=False,
-        cache_labels=False,
-        rank=0,
-        world_size=1,
-        use_horovod=False,
-        mlp_loss=False,
-        weight_loss_kappa=0,
-    ):
-        super().__init__()
-        self.local_loss = local_loss
-        self.gather_with_grad = gather_with_grad
-        self.cache_labels = cache_labels
-        self.rank = rank
-        self.world_size = world_size
-        self.use_horovod = use_horovod
-        self.mlp_loss = mlp_loss
-        self.weighted_loss = bool(weight_loss_kappa != 0)
-        self.weight_loss_kappa = weight_loss_kappa
-        # cache state
-        self.prev_num_logits = 0
-        self.labels = {}
-
-    def forward(
-        self,
-        audio_features,
-        text_features,
-        logit_scale_a,
-        logit_scale_t=None,
-        audio_features_mlp=None,
-        text_features_mlp=None,
-    ):
-        device = audio_features.device
-        if self.mlp_loss:
-            if self.world_size > 1:
-                (
-                    all_audio_features,
-                    all_text_features,
-                    all_audio_features_mlp,
-                    all_text_features_mlp,
-                ) = gather_features(
-                    audio_features=audio_features,
-                    text_features=text_features,
-                    audio_features_mlp=audio_features_mlp,
-                    text_features_mlp=text_features_mlp,
-                    local_loss=self.local_loss,
-                    gather_with_grad=self.gather_with_grad,
-                    rank=self.rank,
-                    world_size=self.world_size,
-                    use_horovod=self.use_horovod,
-                    mlp_loss=self.mlp_loss,
-                )
-                if self.local_loss:
-                    a_logits_per_audio = (
-                        logit_scale_a * audio_features @ all_text_features_mlp.T
-                    )
-                    a_logits_per_text = (
-                        logit_scale_a * text_features_mlp @ all_audio_features.T
-                    )
-                    t_logits_per_audio = (
-                        logit_scale_t * audio_features_mlp @ all_text_features.T
-                    )
-                    t_logits_per_text = (
-                        logit_scale_t * text_features @ all_audio_features_mlp.T
-                    )
-                else:
-                    a_logits_per_audio = (
-                        logit_scale_a * all_audio_features @ all_text_features_mlp.T
-                    )
-                    a_logits_per_text = a_logits_per_audio.T
-                    t_logits_per_audio = (
-                        logit_scale_t * all_audio_features_mlp @ all_text_features.T
-                    )
-                    t_logits_per_text = t_logits_per_audio.T
-            else:
-                a_logits_per_audio = (
-                    logit_scale_a * audio_features @ text_features_mlp.T
-                )
-                a_logits_per_text = logit_scale_a * text_features_mlp @ audio_features.T
-                t_logits_per_audio = (
-                    logit_scale_t * audio_features_mlp @ text_features.T
-                )
-                t_logits_per_text = logit_scale_t * text_features @ audio_features_mlp.T
-
-            # calculated ground-truth and cache if enabled
-            num_logits = a_logits_per_audio.shape[0]
-            if self.prev_num_logits != num_logits or device not in self.labels:
-                labels = torch.arange(num_logits, device=device, dtype=torch.long)
-                if self.world_size > 1 and self.local_loss:
-                    labels = labels + num_logits * self.rank
-                if self.cache_labels:
-                    self.labels[device] = labels
-                    self.prev_num_logits = num_logits
-            else:
-                labels = self.labels[device]
-
-            if not self.weighted_loss:
-                total_loss = (
-                    F.cross_entropy(a_logits_per_audio, labels)
-                    + F.cross_entropy(a_logits_per_text, labels)
-                    + F.cross_entropy(t_logits_per_audio, labels)
-                    + F.cross_entropy(t_logits_per_text, labels)
-                ) / 4
-            else:
-                audio_weight = (audio_features @ audio_features.T).detach()
-                audio_weight = (
-                    torch.exp(
-                        torch.sum(audio_weight, axis=1)
-                        / (self.weight_loss_kappa * len(audio_weight))
-                    )
-                ).detach()
-                text_weight = (text_features @ text_features.T).detach()
-                text_weight = (
-                    torch.exp(
-                        torch.sum(text_weight, axis=1)
-                        / (self.weight_loss_kappa * len(text_features))
-                    )
-                ).detach()
-                total_loss = (
-                    F.cross_entropy(a_logits_per_audio, labels, weight=audio_weight)
-                    + F.cross_entropy(a_logits_per_text, labels, weight=audio_weight)
-                    + F.cross_entropy(t_logits_per_audio, labels, weight=text_weight)
-                    + F.cross_entropy(t_logits_per_text, labels, weight=text_weight)
-                ) / 4
-        else:
-            if self.world_size > 1:
-                all_audio_features, all_text_features = gather_features(
-                    audio_features=audio_features,
-                    text_features=text_features,
-                    local_loss=self.local_loss,
-                    gather_with_grad=self.gather_with_grad,
-                    rank=self.rank,
-                    world_size=self.world_size,
-                    use_horovod=self.use_horovod,
-                    mlp_loss=self.mlp_loss,
-                )
-
-                if self.local_loss:
-                    logits_per_audio = (
-                        logit_scale_a * audio_features @ all_text_features.T
-                    )
-                    logits_per_text = (
-                        logit_scale_a * text_features @ all_audio_features.T
-                    )
-                else:
-                    logits_per_audio = (
-                        logit_scale_a * all_audio_features @ all_text_features.T
-                    )
-                    logits_per_text = logits_per_audio.T
-            else:
-                logits_per_audio = logit_scale_a * audio_features @ text_features.T
-                logits_per_text = logit_scale_a * text_features @ audio_features.T
-
-            # calculated ground-truth and cache if enabled
-            num_logits = logits_per_audio.shape[0]
-            if self.prev_num_logits != num_logits or device not in self.labels:
-                labels = torch.arange(num_logits, device=device, dtype=torch.long)
-                if self.world_size > 1 and self.local_loss:
-                    labels = labels + num_logits * self.rank
-                if self.cache_labels:
-                    self.labels[device] = labels
-                    self.prev_num_logits = num_logits
-            else:
-                labels = self.labels[device]
-            if not self.weighted_loss:
-                total_loss = (
-                    F.cross_entropy(logits_per_audio, labels)
-                    + F.cross_entropy(logits_per_text, labels)
-                ) / 2
-            else:
-                audio_weight = (all_audio_features @ all_audio_features.T).detach()
-                audio_weight = (
-                    torch.exp(
-                        torch.sum(audio_weight, axis=1)
-                        / (self.weight_loss_kappa * len(all_audio_features))
-                    )
-                ).detach()
-                text_weight = (all_text_features @ all_text_features.T).detach()
-                text_weight = (
-                    torch.exp(
-                        torch.sum(text_weight, axis=1)
-                        / (self.weight_loss_kappa * len(all_text_features))
-                    )
-                ).detach()
-                total_loss = (
-                    F.cross_entropy(logits_per_audio, labels, weight=text_weight)
-                    + F.cross_entropy(logits_per_text, labels, weight=audio_weight)
-                ) / 2
-        return total_loss
-
-
-def lp_gather_features(pred, target, world_size=1, use_horovod=False):
-    if use_horovod:
-        assert hvd is not None, "Please install horovod"
-        with torch.no_grad():
-            all_preds = hvd.allgather(pred)
-            all_targets = hvd.allgath(target)
-    else:
-        gathered_preds = [torch.zeros_like(pred) for _ in range(world_size)]
-        gathered_targets = [torch.zeros_like(target) for _ in range(world_size)]
-
-        dist.all_gather(gathered_preds, pred)
-        dist.all_gather(gathered_targets, target)
-        all_preds = torch.cat(gathered_preds, dim=0)
-        all_targets = torch.cat(gathered_targets, dim=0)
-
-    return all_preds, all_targets
-
-
-def get_map(pred, target):
-    pred = torch.sigmoid(pred).numpy()
-    target = target.numpy()
-    return np.mean(average_precision_score(target, pred, average=None))
-
-
-def get_acc(pred, target):
-    pred = torch.argmax(pred, 1).numpy()
-    target = torch.argmax(target, 1).numpy()
-    return accuracy_score(target, pred)
-
-
-def get_mauc(pred, target):
-    pred = torch.sigmoid(pred).numpy()
-    target = target.numpy()
-    return np.mean(roc_auc_score(target, pred, average=None))
-
-
-class LPMetrics(object):
-    def __init__(self, metric_names=["map", "acc", "mauc"]):
-        self.metrics = []
-        for name in metric_names:
-            self.metrics.append(self.get_metric(name))
-        self.metric_names = metric_names
-
-    def get_metric(self, name):
-        if name == "map":
-            return get_map
-        elif name == "acc":
-            return get_acc
-        elif name == "mauc":
-            return get_mauc
-        else:
-            raise ValueError(f"the metric should be at least one of [map, acc, mauc]")
-
-    def evaluate_mertics(self, pred, target):
-        metric_dict = {}
-        for i in range(len(self.metric_names)):
-            metric_dict[self.metric_names[i]] = self.metrics[i](pred, target)
-        return metric_dict
-
-
-def calc_celoss(pred, target):
-    target = torch.argmax(target, 1).long()
-    return nn.CrossEntropyLoss()(pred, target)
-
-
-class LPLoss(nn.Module):
-    def __init__(self, loss_name):
-        super().__init__()
-        if loss_name == "bce":
-            self.loss_func = nn.BCEWithLogitsLoss()
-        elif loss_name == "ce":
-            self.loss_func = calc_celoss
-        elif loss_name == "mse":
-            self.loss_func = nn.MSELoss()
-        else:
-            raise ValueError(f"the loss func should be at least one of [bce, ce, mse]")
-
-    def forward(self, pred, target):
-        loss = self.loss_func(pred, target)
-        return loss

models/CLAP/open_clip/tokenizer.py

@@ -1,197 +0,0 @@
-""" CLIP tokenizer
-
-Copied from https://github.com/openai/CLIP. Originally MIT License, Copyright (c) 2021 OpenAI.
-"""
-import gzip
-import html
-import os
-from functools import lru_cache
-from typing import Union, List
-
-import ftfy
-import regex as re
-import torch
-
-
-@lru_cache()
-def default_bpe():
-    return os.path.join(
-        os.path.dirname(os.path.abspath(__file__)), "bpe_simple_vocab_16e6.txt.gz"
-    )
-
-
-@lru_cache()
-def bytes_to_unicode():
-    """
-    Returns list of utf-8 byte and a corresponding list of unicode strings.
-    The reversible bpe codes work on unicode strings.
-    This means you need a large # of unicode characters in your vocab if you want to avoid UNKs.
-    When you're at something like a 10B token dataset you end up needing around 5K for decent coverage.
-    This is a significant percentage of your normal, say, 32K bpe vocab.
-    To avoid that, we want lookup tables between utf-8 bytes and unicode strings.
-    And avoids mapping to whitespace/control characters the bpe code barfs on.
-    """
-    bs = (
-        list(range(ord("!"), ord("~") + 1))
-        + list(range(ord("¡"), ord("¬") + 1))
-        + list(range(ord("®"), ord("ÿ") + 1))
-    )
-    cs = bs[:]
-    n = 0
-    for b in range(2**8):
-        if b not in bs:
-            bs.append(b)
-            cs.append(2**8 + n)
-            n += 1
-    cs = [chr(n) for n in cs]
-    return dict(zip(bs, cs))
-
-
-def get_pairs(word):
-    """Return set of symbol pairs in a word.
-    Word is represented as tuple of symbols (symbols being variable-length strings).
-    """
-    pairs = set()
-    prev_char = word[0]
-    for char in word[1:]:
-        pairs.add((prev_char, char))
-        prev_char = char
-    return pairs
-
-
-def basic_clean(text):
-    text = ftfy.fix_text(text)
-    text = html.unescape(html.unescape(text))
-    return text.strip()
-
-
-def whitespace_clean(text):
-    text = re.sub(r"\s+", " ", text)
-    text = text.strip()
-    return text
-
-
-class SimpleTokenizer(object):
-    def __init__(self, bpe_path: str = default_bpe(), special_tokens=None):
-        self.byte_encoder = bytes_to_unicode()
-        self.byte_decoder = {v: k for k, v in self.byte_encoder.items()}
-        merges = gzip.open(bpe_path).read().decode("utf-8").split("\n")
-        merges = merges[1 : 49152 - 256 - 2 + 1]
-        merges = [tuple(merge.split()) for merge in merges]
-        vocab = list(bytes_to_unicode().values())
-        vocab = vocab + [v + "</w>" for v in vocab]
-        for merge in merges:
-            vocab.append("".join(merge))
-        if not special_tokens:
-            special_tokens = ["<start_of_text>", "<end_of_text>"]
-        else:
-            special_tokens = ["<start_of_text>", "<end_of_text>"] + special_tokens
-        vocab.extend(special_tokens)
-        self.encoder = dict(zip(vocab, range(len(vocab))))
-        self.decoder = {v: k for k, v in self.encoder.items()}
-        self.bpe_ranks = dict(zip(merges, range(len(merges))))
-        self.cache = {t: t for t in special_tokens}
-        special = "|".join(special_tokens)
-        self.pat = re.compile(
-            special + r"""|'s|'t|'re|'ve|'m|'ll|'d|[\p{L}]+|[\p{N}]|[^\s\p{L}\p{N}]+""",
-            re.IGNORECASE,
-        )
-
-        self.vocab_size = len(self.encoder)
-        self.all_special_ids = [self.encoder[t] for t in special_tokens]
-
-    def bpe(self, token):
-        if token in self.cache:
-            return self.cache[token]
-        word = tuple(token[:-1]) + (token[-1] + "</w>",)
-        pairs = get_pairs(word)
-
-        if not pairs:
-            return token + "</w>"
-
-        while True:
-            bigram = min(pairs, key=lambda pair: self.bpe_ranks.get(pair, float("inf")))
-            if bigram not in self.bpe_ranks:
-                break
-            first, second = bigram
-            new_word = []
-            i = 0
-            while i < len(word):
-                try:
-                    j = word.index(first, i)
-                    new_word.extend(word[i:j])
-                    i = j
-                except:
-                    new_word.extend(word[i:])
-                    break
-
-                if word[i] == first and i < len(word) - 1 and word[i + 1] == second:
-                    new_word.append(first + second)
-                    i += 2
-                else:
-                    new_word.append(word[i])
-                    i += 1
-            new_word = tuple(new_word)
-            word = new_word
-            if len(word) == 1:
-                break
-            else:
-                pairs = get_pairs(word)
-        word = " ".join(word)
-        self.cache[token] = word
-        return word
-
-    def encode(self, text):
-        bpe_tokens = []
-        text = whitespace_clean(basic_clean(text)).lower()
-        for token in re.findall(self.pat, text):
-            token = "".join(self.byte_encoder[b] for b in token.encode("utf-8"))
-            bpe_tokens.extend(
-                self.encoder[bpe_token] for bpe_token in self.bpe(token).split(" ")
-            )
-        return bpe_tokens
-
-    def decode(self, tokens):
-        text = "".join([self.decoder[token] for token in tokens])
-        text = (
-            bytearray([self.byte_decoder[c] for c in text])
-            .decode("utf-8", errors="replace")
-            .replace("</w>", " ")
-        )
-        return text
-
-
-_tokenizer = SimpleTokenizer()
-
-
-def tokenize(
-    texts: Union[str, List[str]], context_length: int = 77
-) -> torch.LongTensor:
-    """
-    Returns the tokenized representation of given input string(s)
-
-    Parameters
-    ----------
-    texts : Union[str, List[str]]
-        An input string or a list of input strings to tokenize
-    context_length : int
-        The context length to use; all CLIP models use 77 as the context length
-
-    Returns
-    -------
-    A two-dimensional tensor containing the resulting tokens, shape = [number of input strings, context_length]
-    """
-    if isinstance(texts, str):
-        texts = [texts]
-
-    sot_token = _tokenizer.encoder["<start_of_text>"]
-    eot_token = _tokenizer.encoder["<end_of_text>"]
-    all_tokens = [[sot_token] + _tokenizer.encode(text) + [eot_token] for text in texts]
-    result = torch.zeros(len(all_tokens), context_length, dtype=torch.long)
-
-    for i, tokens in enumerate(all_tokens):
-        if len(tokens) > context_length:
-            tokens = tokens[:context_length]  # Truncate
-        result[i, : len(tokens)] = torch.tensor(tokens)
-
-    return result