Add a generate module for OpenELM models.

generate_openelm.py

@@ -0,0 +1,239 @@
+"""Module to generate OpenELM output given a model and an input prompt."""
+import os
+import logging
+import time
+import argparse
+from typing import Optional, Union
+import torch
+
+from transformers import AutoTokenizer, AutoModelForCausalLM
+
+
+def generate(
+    prompt: str,
+    model: Union[str, AutoModelForCausalLM],
+    hf_security_token: str = None,
+    tokenizer: Union[str, AutoTokenizer] = 'meta-llama/Llama-2-7b-hf',
+    device: Optional[str] = None,
+    max_length: int = 1024,
+    speculative_model: Optional[Union[str, AutoModelForCausalLM]] = None,
+    generate_kwargs: Optional[dict] = None,
+) -> str:
+    """ Generates output given a prompt.
+
+    Args:
+        prompt: The string prompt.
+        model: The LLM Model. If a string is passed, it should be the path to
+            the hf converted checkpoint.
+        hf_security_token: Hugging face security token.
+        tokenizer: Tokenizer instance. If model is set as a string path,
+            the tokenizer will be loaded from the checkpoint.
+        device: String representation of device to run the model on. If None
+            and cuda available it would be set to cuda:0 else cpu.
+        max_length: Maximum length of tokens, input prompt + generated tokens.
+        speculative_model: If set, this model will be used for
+            speculative generation. If a string is passed, it should be the
+            path to the hf converted checkpoint.
+        generate_kwargs: Extra kwargs passed to the generate function.
+
+    Returns:
+        output_text: output generated as a string.
+        generation_time: generation time in seconds.
+
+    Raises:
+        ValueError: If device is set to CUDA but no CUDA device is detected.
+        FileNotFoundError: If model or speculative_model are strings but
+            the model paths do not exist.
+        ValueError: If hf_security_token is not specified.
+    """
+    if not device:
+        if torch.cuda.is_available() and torch.cuda.device_count():
+            device = "cuda:0"
+            logging.warning(
+                'inference device is not set, using cuda:0, %s',
+                torch.cuda.get_device_name(0)
+            )
+        else:
+            device = 'cpu'
+            logging.warning('no CUDA device detected, using cpu, expect slower speeds.')
+
+    if 'cuda' in device and not torch.cuda.is_available():
+        raise ValueError('CUDA device requested but no CUDA device detected.')
+
+    if isinstance(model, str) and (not model or not os.path.exists(model)):
+        raise FileNotFoundError(f'Model checkpoint does not exist at {model}.')
+
+    if (isinstance(speculative_model, str) and (
+            not speculative_model and not os.path.exists(speculative_model))):
+        raise FileNotFoundError(
+            (
+                'Speculative checkpoint path does not exist at '
+                f'{speculative_model}.'
+            )
+        )
+    if not tokenizer and not isinstance(model, str):
+        raise ValueError('Tokenizer is not set in the generate function.')
+
+    if not hf_security_token:
+        raise ValueError((
+            'Hugging face security key needs to be specified. '
+            'Please refer to https://huggingface.co/docs/hub/security-tokens'
+            ' to obtain one.'
+            )
+        )
+
+    if isinstance(model, str):
+        checkpoint_path = model
+        model = AutoModelForCausalLM.from_pretrained(
+            checkpoint_path,
+            trust_remote_code=True
+        )
+    model.to(device).eval()
+    if isinstance(tokenizer, str):
+        tokenizer = AutoTokenizer.from_pretrained(
+            tokenizer,
+            token=hf_security_token,
+        )
+
+    # Speculative mode
+    draft_model = None
+    if speculative_model:
+        draft_model = speculative_model
+        if isinstance(speculative_model, str):
+            draft_model = AutoModelForCausalLM.from_pretrained(
+                speculative_model,
+                trust_remote_code=True
+            )
+        draft_model.to(device).eval()
+
+    # Prepare the prompt
+    tokenized_prompt = tokenizer(prompt)
+    tokenized_prompt = torch.tensor(
+        tokenized_prompt['input_ids'],
+        device=device
+    )
+
+    tokenized_prompt = tokenized_prompt.unsqueeze(0)
+
+    # Generate
+    stime = time.time()
+    output_ids = model.generate(
+        tokenized_prompt,
+        max_length=max_length,
+        pad_token_id=0,
+        assistant_model=draft_model,
+        **(generate_kwargs if generate_kwargs else {}),
+    )
+    generation_time = time.time() - stime
+
+    output_text = tokenizer.decode(
+        output_ids[0].tolist(),
+        skip_special_tokens=True
+    )
+
+    return output_text, generation_time
+
+
+def openelm_generate_parser():
+    """Argument Parser"""
+
+    class KwargsParser(argparse.Action):
+        """Parser action class to parse kwargs of form key=value"""
+        def __call__(self, parser, namespace, values, option_string=None):
+            setattr(namespace, self.dest, dict())
+            for val in values:
+                if '=' not in val:
+                    raise ValueError(
+                        (
+                            'Argument parsing error, kwargs are expected in'
+                            ' the form of key=value.'
+                        )
+                    )
+                kwarg_k, kwarg_v = val.split('=')
+                try:
+                    converted_v = int(kwarg_v)
+                except ValueError:
+                    try:
+                        converted_v = float(kwarg_v)
+                    except ValueError:
+                        converted_v = kwarg_v            
+                getattr(namespace, self.dest)[kwarg_k] = converted_v
+
+    parser = argparse.ArgumentParser('OpenELM Generate Module')
+    parser.add_argument(
+        '--checkpoint',
+        dest='checkpoint_path',
+        help='Path to the model hf converted checkpoint.',
+        required=True,
+        type=str,
+    )
+    parser.add_argument(
+        '--hf_security_token',
+        dest='hf_security_token',
+        help='HF security token, starting with "hf_".',
+        type=str,
+    )
+    parser.add_argument(
+      '--prompt',
+      dest='prompt',
+      help='Prompt for LLM call. Ignored if demo is True.',
+      default='',
+      type=str,
+    )
+    parser.add_argument(
+        '--device',
+        dest='device',
+        help='Device used for inference.',
+        type=str,
+    )
+    parser.add_argument(
+        '--max_length',
+        dest='max_length',
+        help='Maximum length of tokens.',
+        default=256,
+        type=int,
+    )
+    parser.add_argument(
+        '--speculative_model_ckpt',
+        dest='speculative_model_ckpt',
+        help=(
+            'If set, this is used as a draft model for speculative generation.'
+        ),
+        type=str,
+    )
+    parser.add_argument(
+        '--generate_kwargs',
+        dest='generate_kwargs',
+        help='additional kwargs passed to the HF generate function.',
+        type=str,
+        nargs='*',
+        action=KwargsParser,
+    )
+    return parser.parse_args()
+
+
+if __name__ == '__main__':
+    args = openelm_generate_parser()
+    prompt = args.prompt
+
+    output_text, genertaion_time = generate(
+        prompt=prompt,
+        model=args.checkpoint_path,
+        device=args.device,
+        max_length=args.max_length,
+        speculative_model=args.speculative_model_ckpt,
+        generate_kwargs=args.generate_kwargs,
+        hf_security_token=args.hf_security_token,
+    )
+
+    print_txt = (
+        f'\r\n{"=" * os.get_terminal_size().columns}\r\n'
+        '\033[1m Prompt + Generated Output\033[0m\r\n'
+        f'{"-" * os.get_terminal_size().columns}\r\n'
+        f'{output_text}\r\n'
+        f'{"-" * os.get_terminal_size().columns}\r\n'
+        '\r\nGeneration took'
+        f'\033[1m\033[92m {round(genertaion_time, 2)} \033[0m'
+        'seconds.\r\n'
+    )
+    print(print_txt)