File size: 5,644 Bytes
9eb0412
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
"""
Local inference script for MicroGPT.
Loads a trained checkpoint and generates text.

Usage:
    # Interactive mode (type prompts)
    python inference.py

    # Generate from a prompt
    python inference.py --prompt "Once upon a time"

    # Use a specific checkpoint
    python inference.py --checkpoint runs_out/micro_gpt_ckpt_step_100000.pt --prompt "Hello"
"""

import os
import sys
import argparse
from pathlib import Path

import torch
import torch.nn.functional as F

# Add current directory for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from models import get_model


def load_model_and_meta(checkpoint_path: str, device: str = 'cpu'):
    """Load a trained model checkpoint and return model + metadata."""
    print(f'Loading checkpoint: {checkpoint_path}')
    ckpt = torch.load(checkpoint_path, map_location=device, weights_only=False)

    config = ckpt['config']
    meta = ckpt['meta']
    step = ckpt['step']
    best_val_loss = ckpt.get('best_val_loss', 'N/A')

    print(f'  Architecture : {config["arch"]}')
    print(f'  Dataset      : {config["dataset"]}')
    print(f'  Vocab size   : {meta["vocab_size"]}')
    print(f'  Block size   : {config["block_size"]}')
    print(f'  Layers/Heads : {config["n_layer"]}/{config["n_head"]}')
    print(f'  Embed dim    : {config["n_embd"]}')
    print(f'  Parameters   : {ckpt["n_params"]:,}')
    print(f'  Training step: {step}')
    print(f'  Best val loss: {best_val_loss}')

    # Build model
    ModelClass = get_model(config['arch'])
    model = ModelClass(
        vocab_size=meta['vocab_size'],
        block_size=config['block_size'],
        n_layer=config['n_layer'],
        n_head=config['n_head'],
        n_embd=config['n_embd'],
        dropout=config.get('dropout', 0.1),
    )
    model.load_state_dict(ckpt['model'])
    model.to(device)
    model.eval()

    return model, meta, config


def encode_text(text: str, stoi: dict) -> torch.Tensor:
    """Encode a string into token indices using the char-level tokenizer."""
    indices = [stoi.get(c, 0) for c in text]  # 0 = newline (fallback)
    return torch.tensor([indices], dtype=torch.long)


def decode_tokens(tokens: list, itos: dict) -> str:
    """Decode token indices back into a string."""
    return ''.join(itos.get(i, '?') for i in tokens)


@torch.no_grad()
def generate_text(model, prompt: str, meta: dict, max_new_tokens: int = 256,
                  temperature: float = 0.9, top_k: int = 40, device: str = 'cpu'):
    """Generate text continuation from a prompt."""
    stoi = meta['stoi']
    itos = meta['itos']

    # Encode prompt
    prompt_ids = encode_text(prompt, stoi)
    prompt_ids = prompt_ids.to(device)

    # Generate
    output_ids = model.generate(
        prompt_ids,
        max_new_tokens=max_new_tokens,
        temperature=temperature,
        top_k=top_k,
    )

    # Decode
    full_text = decode_tokens(output_ids[0].tolist(), itos)
    return full_text


def interactive_mode(model, meta, device: str):
    """Interactive chat-like generation."""
    print('\n' + '=' * 60)
    print('  MicroGPT Interactive Mode')
    print('  Type your prompt and press Enter.')
    print('  Type "quit", "exit", or "q" to stop.')
    print('=' * 60)

    while True:
        try:
            prompt = input('\nPrompt: ').strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break

        if prompt.lower() in ('quit', 'exit', 'q'):
            break

        if not prompt:
            continue

        print('\nGenerating...\n')
        result = generate_text(model, prompt, meta, device=device)
        print(result)
        print()


def main():
    parser = argparse.ArgumentParser(description='MicroGPT Local Inference')
    parser.add_argument('--checkpoint', type=str,
                        default='runs_out/micro_gpt_ckpt_step_100000.pt',
                        help='Path to model checkpoint')
    parser.add_argument('--prompt', type=str, default=None,
                        help='Text prompt for generation')
    parser.add_argument('--max_new_tokens', type=int, default=256,
                        help='Maximum tokens to generate')
    parser.add_argument('--temperature', type=float, default=0.9,
                        help='Sampling temperature (higher = more random)')
    parser.add_argument('--top_k', type=int, default=40,
                        help='Top-k sampling threshold')
    parser.add_argument('--device', type=str, default='cpu',
                        help='Device to run on (cpu, cuda, mps)')
    args = parser.parse_args()

    # Resolve device
    if args.device == 'cuda' and not torch.cuda.is_available():
        print('CUDA not available, falling back to CPU')
        args.device = 'cpu'
    elif args.device == 'mps' and not hasattr(torch.backends, 'mps') or not torch.backends.mps.is_available():
        print('MPS not available, falling back to CPU')
        args.device = 'cpu'

    # Load model
    model, meta, config = load_model_and_meta(args.checkpoint, args.device)

    if args.prompt:
        # Single generation
        print(f'\nPrompt: {args.prompt}')
        print(f'Generating (temp={args.temperature}, top_k={args.top_k}, max_tokens={args.max_new_tokens})...\n')
        result = generate_text(
            model, args.prompt, meta,
            max_new_tokens=args.max_new_tokens,
            temperature=args.temperature,
            top_k=args.top_k,
            device=args.device,
        )
        print(result)
    else:
        # Interactive mode
        interactive_mode(model, meta, args.device)


if __name__ == '__main__':
    main()