Token Streamer

What It Does

Installation

pip install transformers rich pytest

Usage

python token_streamer.py --model-path <model_path> --input <input_prompt> [--stream-speed <seconds>] [--output-file <file_path>]

python token_streamer.py --model-path ./models/gpt2 --input "Once upon a time" --stream-speed 0.2 --output-file output.txt

Source Code

import argparse
import time
from transformers import AutoTokenizer, AutoModelForCausalLM
from rich.console import Console

def stream_tokens(model_path, input_prompt, stream_speed, output_file=None):
    """
    Streams tokens generated by a language model in real-time.

    Args:
        model_path (str): Path to the locally hosted model.
        input_prompt (str): Input prompt for the model.
        stream_speed (float): Delay in seconds between streaming tokens.
        output_file (str, optional): File path to save the output. Defaults to None.

    Returns:
        None
    """
    console = Console()

    try:
        console.print("[bold green]Loading model and tokenizer...[/bold green]")
        tokenizer = AutoTokenizer.from_pretrained(model_path)
        model = AutoModelForCausalLM.from_pretrained(model_path)

        console.print("[bold green]Generating tokens...[/bold green]")
        input_ids = tokenizer.encode(input_prompt, return_tensors="pt")
        output_ids = model.generate(input_ids, max_new_tokens=50, do_sample=True)

        generated_tokens = tokenizer.decode(output_ids[0], skip_special_tokens=True)
        generated_text = ""

        for token in generated_tokens[len(input_prompt):]:
            generated_text += token
            console.print(token, end="", style="bold blue")
            time.sleep(stream_speed)

        console.print("\n[bold green]Streaming complete.[/bold green]")

        if output_file:
            with open(output_file, "w") as f:
                f.write(generated_text)

    except Exception as e:
        console.print(f"[bold red]Error: {e}[/bold red]")
        raise

def main():
    parser = argparse.ArgumentParser(description="Token Streamer: Stream token-by-token completions from a locally hosted LLM.")
    parser.add_argument("--model-path", required=True, help="Path to the locally hosted model.")
    parser.add_argument("--input", required=True, help="Input prompt for the model.")
    parser.add_argument("--stream-speed", type=float, default=0.5, help="Delay in seconds between streaming tokens.")
    parser.add_argument("--output-file", help="Optional file path to save the output.")

    args = parser.parse_args()

    stream_tokens(
        model_path=args.model_path,
        input_prompt=args.input,
        stream_speed=args.stream_speed,
        output_file=args.output_file
    )

if __name__ == "__main__":
    main()