LLM Batch Splitter

What It Does

Installation

pip install numpy==1.23.5 transformers==4.33.2

Usage

python llm_batch_splitter.py --inputs "Hello, world!" "How are you?" --max_chunk_size 2

Source Code

import numpy as np
from transformers import PreTrainedModel, PreTrainedTokenizer
from typing import List, Any

def process_in_chunks(model: PreTrainedModel, tokenizer: PreTrainedTokenizer, inputs: List[str], max_chunk_size: int = 5) -> List[Any]:
    """
    Splits input texts into smaller chunks, processes them through the model, and reassembles the outputs.

    Args:
        model (PreTrainedModel): The pre-trained language model to use for inference.
        tokenizer (PreTrainedTokenizer): The tokenizer corresponding to the model.
        inputs (List[str]): A list of input texts to process.
        max_chunk_size (int): The maximum number of inputs per chunk.

    Returns:
        List[Any]: The processed outputs reassembled in the original order.
    """
    if not inputs:
        return []

    if max_chunk_size <= 0:
        raise ValueError("max_chunk_size must be greater than 0")

    # Split inputs into chunks
    chunks = [inputs[i:i + max_chunk_size] for i in range(0, len(inputs), max_chunk_size)]

    outputs = []

    for chunk in chunks:
        try:
            # Tokenize the chunk
            tokenized = tokenizer(chunk, padding=True, truncation=True, return_tensors="pt")

            # Perform model inference
            with model.no_grad():
                model_output = model(**tokenized)

            # Decode the outputs (assuming the model generates text)
            decoded_outputs = tokenizer.batch_decode(model_output.logits.argmax(dim=-1), skip_special_tokens=True)

            outputs.extend(decoded_outputs)
        except Exception as e:
            raise RuntimeError(f"Error processing chunk: {chunk}") from e

    return outputs

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="LLM Batch Splitter: Process input texts in memory-efficient chunks.")
    parser.add_argument("--inputs", nargs="*", required=True, help="List of input texts to process.")
    parser.add_argument("--max_chunk_size", type=int, default=5, help="Maximum number of inputs per chunk.")

    args = parser.parse_args()

    # Example usage with a dummy model and tokenizer (replace with actual ones in real use)
    from transformers import AutoModelForCausalLM, AutoTokenizer

    model_name = "gpt2"  # Replace with your model
    model = AutoModelForCausalLM.from_pretrained(model_name)
    tokenizer = AutoTokenizer.from_pretrained(model_name)

    results = process_in_chunks(model, tokenizer, args.inputs, args.max_chunk_size)
    print("Processed Outputs:", results)