| """ | |
| KernelLLM | |
| This script provides a simple interface for the KernelLLM model. | |
| It allows users to input PyTorch models and let KernelLLM attempt to implement the corresponding Triton kernels. | |
| The KernelLLM class provides two types of methods: | |
| 1. Methods that instruct the model with a suitable prompt to generate Triton kernels. | |
| 2. "raw" methods that allow the user to interact with the model directly, without any additional prompt wrapping. | |
| For best results, use the `generate_triton` method to instruct the model the way it was trained. | |
| Example usage: | |
| To run the script from the command line: | |
| CUDA_VISIBLE_DEVICES=0 python kernelllm.py | |
| To use the class in an interactive Python session: | |
| $ ipython | |
| from kernelllm import KernelLLM | |
| model = KernelLLM() | |
| model.generate_triton("<your torch module here>", max_new_tokens=128) | |
| # or stream output directly | |
| model.stream_raw("<your text prompt>", max_new_tokens=128) | |
| Full example: | |
| ``` | |
| #Generate Triton-optimized code for a PyTorch model: | |
| from kernelllm import KernelLLM | |
| model = KernelLLM() | |
| pytorch_code = ''' | |
| import torch | |
| import torch.nn as nn | |
| class Model(nn.Module): | |
| def __init__(self): | |
| super(Model, self).__init__() | |
| def forward(self, x): | |
| return x * 2 | |
| def get_inputs(): | |
| return [torch.randn(1, 128).cuda()] | |
| def get_init_inputs(): | |
| return [] | |
| ''' | |
| optimized_code = model.generate_triton(pytorch_code, max_new_tokens=512) | |
| print(optimized_code) | |
| ``` | |
| """ | |
| import sys | |
| import torch | |
| from transformers import AutoModelForCausalLM, AutoTokenizer, TextStreamer | |
| HF_MODEL = "facebook/KernelLLM" | |
| REPL_INSTRUCTIONS = """ | |
| You can paste or write your nn.Module code below (and finish with Ctrl+D). | |
| The model will try to optimize it with Triton kernels. | |
| Make sure that you provide a `get_inputs()` and `get_init_inputs()` function such that your model can be run like this | |
| args, kwargs = get_inputs() | |
| model = Model(*args, **kwargs) | |
| out = model(get_inputs()) | |
| >>>""" | |
| DEFAULT_MODEL_CODE = """ | |
| import torch | |
| import torch.nn as nn | |
| class Model(nn.Module): | |
| \"\"\" | |
| A model that computes Hinge Loss for binary classification tasks. | |
| Parameters: | |
| None | |
| \"\"\" | |
| def __init__(self): | |
| super(Model, self).__init__() | |
| def forward(self, predictions, targets): | |
| return torch.mean(torch.clamp(1 - predictions * targets, min=0)) | |
| batch_size = 128 | |
| input_shape = (1,) | |
| dim = 1 | |
| def get_inputs(): | |
| return [torch.randn(batch_size, *input_shape), torch.randint(0, 2, (batch_size, 1)).float() * 2 - 1] | |
| def get_init_inputs(): | |
| return [] | |
| """ | |
| PROMPT_TEMPLATE = """ | |
| <|begin_of_text|>You write custom Triton kernels to replace the pytorch operators in the given architecture to get speedups. | |
| You have complete freedom to choose the set of operators you want to replace. You may make the decision to replace some operators with custom Triton kernels and leave others unchanged. You may replace multiple operators with custom implementations, consider operator fusion opportunities (combining multiple operators into a single kernel, for example, combining matmul+relu), or algorithmic changes (such as online softmax). You are only limited by your imagination. | |
| Here's an example to show you the syntax of inline embedding custom operators from the Triton DSL in torch: The example given architecture is: | |
| ``` | |
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| class Model(nn.Module): | |
| def __init__(self) -> None: | |
| super().__init__() | |
| def forward(self, a, b): | |
| return a + b | |
| def get_inputs(): | |
| # randomly generate input tensors based on the model architecture | |
| a = torch.randn(1, 128).cuda() | |
| b = torch.randn(1, 128).cuda() | |
| return [a, b] | |
| def get_init_inputs(): | |
| # randomly generate tensors required for initialization based on the model architecture | |
| return [] | |
| ``` | |
| The example new arch with custom Triton kernels looks like this: | |
| ``` | |
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| import triton | |
| import triton.language as tl | |
| @triton.jit | |
| def add_kernel( | |
| x_ptr, # Pointer to first input | |
| y_ptr, # Pointer to second input | |
| out_ptr, # Pointer to output | |
| n_elements, # Total number of elements in input/output | |
| BLOCK_SIZE: tl.constexpr, | |
| ): | |
| # Each program handles a contiguous block of data of size BLOCK_SIZE | |
| block_start = tl.program_id(0) * BLOCK_SIZE | |
| # Create a range of offsets [0..BLOCK_SIZE-1] | |
| offsets = block_start + tl.arange(0, BLOCK_SIZE) | |
| # Mask to ensure we don't go out of bounds | |
| mask = offsets < n_elements | |
| # Load input values | |
| x = tl.load(x_ptr + offsets, mask=mask, other=0.0) | |
| y = tl.load(y_ptr + offsets, mask=mask, other=0.0) | |
| # Perform the elementwise addition | |
| out = x + y | |
| # Store the result | |
| tl.store(out_ptr + offsets, out, mask=mask) | |
| def triton_add(x: torch.Tensor, y: torch.Tensor): | |
| \"\"\" | |
| This function wraps the Triton kernel call. It: | |
| 1. Ensures the inputs are contiguous on GPU. | |
| 2. Calculates the grid (blocks) needed. | |
| 3. Launches the Triton kernel. | |
| \"\"\" | |
| assert x.is_cuda and y.is_cuda, "Tensors must be on CUDA." | |
| x = x.contiguous() | |
| y = y.contiguous() | |
| # Prepare output tensor | |
| out = torch.empty_like(x) | |
| # Number of elements in the tensor | |
| n_elements = x.numel() | |
| BLOCK_SIZE = 128 # Tunable parameter for block size | |
| # Determine the number of blocks needed | |
| grid = lambda meta: ((n_elements + meta["BLOCK_SIZE"] - 1) // meta["BLOCK_SIZE"],) | |
| # Launch the Triton kernel | |
| add_kernel[grid](x, y, out, n_elements, BLOCK_SIZE=BLOCK_SIZE) | |
| return out | |
| class ModelNew(nn.Module): | |
| def __init__(self) -> None: | |
| super().__init__() | |
| def forward(self, a, b): | |
| # Instead of "return a + b", call our Triton-based addition | |
| return triton_add(a, b) | |
| ``` | |
| You are given the following architecture: | |
| ``` | |
| {} | |
| ``` | |
| Optimize the architecture named Model with custom Triton kernels! Name your optimized output architecture ModelNew. Output the new code in codeblocks. Please generate real code, NOT pseudocode, make sure the code compiles and is fully functional. Just output the new model code, no other text, and NO testing code! | |
| """ | |
| class KernelLLM: | |
| """ | |
| A simple wrapper around the KernelLLM model for generating Triton kernels that allows easy | |
| instruction of the model and a streamed repl interface to interact with the model. | |
| """ | |
| def __init__( | |
| self, | |
| model_name: str = HF_MODEL, | |
| device: str = "cuda" if torch.cuda.is_available() else "cpu", | |
| ): | |
| self.model_name = model_name | |
| self.device = device | |
| self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) | |
| self.model = AutoModelForCausalLM.from_pretrained( | |
| self.model_name, torch_dtype=torch.float16 | |
| ) | |
| self.model.to(self.device) | |
| def generate_raw( | |
| self, prompt: str, temperature: float = 0.6, max_new_tokens: int = 2048 | |
| ) -> str: | |
| """ | |
| Generate text from the model using the given prompt verbatim. | |
| Args: | |
| prompt (str): The prompt to generate text from. | |
| temperature (float): The temperature to use for sampling. | |
| max_new_tokens (int): The maximum length of the generated text. | |
| Returns: | |
| str: The generated text. | |
| """ | |
| inputs = self.tokenizer([prompt], return_tensors="pt") | |
| inputs = {k: v.to(self.device) for k, v in inputs.items()} | |
| outputs = self.model.generate( | |
| **inputs, | |
| max_new_tokens=max_new_tokens, | |
| temperature=temperature, | |
| top_k=0, | |
| top_p=0.95, | |
| do_sample=True, | |
| eos_token_id=self.tokenizer.eos_token_id, | |
| ) | |
| text = self.tokenizer.decode(outputs[0], skip_special_tokens=True) | |
| return text[len(prompt) :].strip() | |
| def stream_raw(self, prompt: str, max_new_tokens: int = 2048): | |
| """ | |
| Stream and print text from the model using the given prompt verbatim. | |
| Args: | |
| prompt (str): The prompt to generate text from. | |
| max_new_tokens (int): The maximum length of the generated text. | |
| """ | |
| inputs = self.tokenizer([prompt], return_tensors="pt") | |
| inputs = {k: v.cuda() for k, v in inputs.items()} | |
| streamer = TextStreamer( | |
| self.tokenizer, skip_prompt=True, skip_special_tokens=True | |
| ) | |
| self.model.generate( | |
| **inputs, | |
| streamer=streamer, | |
| max_new_tokens=max_new_tokens, | |
| do_sample=True, | |
| top_k=0, | |
| top_p=0.95, | |
| temperature=0.6, | |
| eos_token_id=self.tokenizer.eos_token_id, | |
| ) | |
| def generate_triton( | |
| self, code: str, temperature: float = 0.6, max_new_tokens: int = 2048 | |
| ) -> str: | |
| """ | |
| Generate Triton for the given torch module. | |
| The input code should be a python module that contains a torch Model(nn.Module) class and | |
| `get_inputs()` and `get_init_inputs()` functions such that your model can be run like this | |
| ``` | |
| args, kwargs = get_inputs() | |
| model = Model(*args, **kwargs) | |
| out = model(get_inputs()) | |
| ``` | |
| Args: | |
| code (str): The torch code to generate Triton for. | |
| temperature (float): The temperature to use for sampling. | |
| max_new_tokens (int): The maximum length of the generated text. | |
| Returns: | |
| str: The generated Triton module. | |
| """ | |
| prompt = PROMPT_TEMPLATE.format(code) | |
| return self.generate_raw(prompt, temperature, max_new_tokens) | |
| def run_repl(self): | |
| """ | |
| Run a REPL for the model. The user can input code and the model will try to optimize it with Triton kernels. | |
| """ | |
| while True: | |
| try: | |
| print(REPL_INSTRUCTIONS) | |
| code = sys.stdin.read().strip() | |
| if code.lower() == "exit": | |
| return | |
| except EOFError: | |
| pass | |
| if not code: | |
| print(f"Using default prompt:\n{DEFAULT_MODEL_CODE}\n") | |
| code = DEFAULT_MODEL_CODE | |
| prompt = PROMPT_TEMPLATE.format(DEFAULT_MODEL_CODE) | |
| try: | |
| self.stream_raw(prompt) | |
| except KeyboardInterrupt: | |
| print("Aborting...") | |
| if __name__ == "__main__": | |
| kernel_llm = KernelLLM() | |
| kernel_llm.run_repl() | |