zeechimp
/

zee

@@ -1,391 +0,0 @@
-"""
-train_hv_intent.py
-Single-file trainer for the hv-intent-router model, with multi-seed
-genetic algorithm training.  Runs N independent GA instances with
-different seeds and keeps the best.
-Produces:
-    codebooks.npz   - vocab_codebook, intent_codebook, memory
-    config.json     - model metadata including parameter counts
-Run:
-    python train_hv_intent.py
-    python train_hv_intent.py --output ./my_model_dir --seeds 8
-Runtime: ~50 seconds for 8 seeds on a laptop CPU.
-Dependencies: numpy only.
-"""
-import numpy as np
-import json
-import time
-import argparse
-from pathlib import Path
-# ===========================================================================
-#  Configuration
-# ===========================================================================
-D_DEFAULT = 4096
-POP_SIZE = 12
-GENERATIONS = 50
-ELITE = 3
-MUTATION_RATE = 0.005
-TRAIN_PER_INTENT = 7
-N_SEEDS = 8
-SEED_BASE = 1
-SEED_STEP = 7919
-SEED_DATA = 42
-VOCAB = "abcdefghijklmnopqrstuvwxyz"
-# ===========================================================================
-#  Training data
-# ===========================================================================
-INTENT_DATA = {
-    "greeting": [
-        "hello there",
-        "hi how are you",
-        "good morning",
-        "good evening",
-        "hey what is up",
-        "nice to meet you",
-        "how are you doing",
-        "what is going on",
-        "hello friend",
-        "good afternoon",
-    ],
-    "math": [
-        "what is two plus two",
-        "calculate ten times three",
-        "what is the square root of nine",
-        "add five and seven",
-        "subtract three from ten",
-        "what is twenty divided by four",
-        "multiply six by eight",
-        "what is one hundred minus forty",
-        "calculate the sum of one and two",
-        "what is the product",
-    ],
-    "reminder": [
-        "remind me to call mom",
-        "set an alarm for six",
-        "remind me to buy milk",
-        "add a reminder for tomorrow",
-        "set a timer for ten minutes",
-        "remind me about the meeting",
-        "create a to do item",
-        "remind me to water the plants",
-        "set an alarm for morning",
-        "remind me to pick up kids",
-    ],
-    "time": [
-        "what time is it",
-        "tell me the time",
-        "how late is it",
-        "what is the current hour",
-        "when is the meeting",
-        "what day is today",
-        "what is the date",
-        "how many hours until noon",
-        "when does the sun set",
-        "when does the store close",
-    ],
-    "weather": [
-        "what is the weather today",
-        "will it rain tomorrow",
-        "is it sunny outside",
-        "what is the forecast",
-        "how hot is it",
-        "is it cold today",
-        "will it snow tonight",
-        "how windy is it",
-        "what is the temperature",
-        "is it cloudy",
-    ],
-}
-# ===========================================================================
-#  Primitives
-# ===========================================================================
-def bundle(stacked_int32):
-    """Majority vote over ±1 array.  stack shape (N, D)."""
-    s = stacked_int32.sum(axis=0)
-    return np.where(s >= 0, np.int8(1), np.int8(-1))
-def mutate(K, rng, rate):
-    mask = rng.random(K.shape) < rate
-    return np.where(mask, -K, K).astype(np.int8)
-def crossover(K1, K2, rng):
-    mask = rng.random(K1.shape) < 0.5
-    return np.where(mask, K1, K2).astype(np.int8)
-# ===========================================================================
-#  Trainer
-# ===========================================================================
-def train(output_dir, D=D_DEFAULT, n_seeds=N_SEEDS, verbose=True):
-    output_dir = Path(output_dir)
-    output_dir.mkdir(parents=True, exist_ok=True)
-    labels = sorted(INTENT_DATA.keys())
-    chars = sorted(set("".join("".join(v) for v in INTENT_DATA.values())))
-    V = len(chars)
-    C = len(labels)
-    c2i = {c: i for i, c in enumerate(chars)}
-    if verbose:
-        print(f"Vocabulary: {V} characters")
-        print(f"Intents:    {C} ({', '.join(labels)})")
-        print(f"Dimension:  {D}")
-        print()
-    # Train / test split
-    train_pairs = []
-    test_pairs = []
-    for idx, label in enumerate(labels):
-        for i, phrase in enumerate(INTENT_DATA[label]):
-            if i < TRAIN_PER_INTENT:
-                train_pairs.append((phrase, idx))
-            else:
-                test_pairs.append((phrase, idx))
-    if verbose:
-        print(f"Train pairs: {len(train_pairs)}")
-        print(f"Test pairs:  {len(test_pairs)}")
-        print()
-    def char_indices(phrase):
-        return np.array([c2i[c] for c in phrase if c in c2i], dtype=np.int32)
-    train_idx = [(char_indices(p), t) for p, t in train_pairs]
-    test_idx = [(char_indices(p), t) for p, t in test_pairs]
-    # Fixed intent codebook (same across seeds)
-    rng_intents = np.random.default_rng(SEED_DATA)
-    intent_codebook = rng_intents.choice(
-        np.array([-1, 1], dtype=np.int8), size=(C, D))
-    def encode(indices, K):
-        if len(indices) == 0:
-            return np.zeros(D, dtype=np.int8)
-        v = K[indices].astype(np.int32).sum(axis=0)
-        return np.where(v >= 0, np.int8(1), np.int8(-1))
-    def build_memory(K):
-        items = []
-        for idx_arr, target in train_idx:
-            p_hv = encode(idx_arr, K)
-            bound = p_hv.astype(np.int32) * intent_codebook[target].astype(np.int32)
-            items.append(bound)
-        return bundle(np.stack(items).astype(np.int32))
-    def evaluate(K, mem=None):
-        if mem is None:
-            mem = build_memory(K)
-        correct = 0
-        for idx_arr, target in test_idx:
-            p_hv = encode(idx_arr, K)
-            retrieved = mem.astype(np.int32) * p_hv.astype(np.int32)
-            sims = (retrieved.astype(np.float32) @
-                    intent_codebook.astype(np.float32).T)
-            if int(np.argmax(sims)) == target:
-                correct += 1
-        return correct / len(test_idx)
-    # Random baseline (average over 10 random codebooks)
-    rng_base = np.random.default_rng(SEED_BASE + 999999)
-    base_accs = []
-    for _ in range(10):
-        K_rand = rng_base.choice(np.array([-1, 1], dtype=np.int8), size=(V, D))
-        base_accs.append(evaluate(K_rand))
-    acc_random = float(np.mean(base_accs))
-    if verbose:
-        print(f"Random baseline: {acc_random:.1%} "
-              f"(mean of 10 random codebooks)")
-        print()
-    # -----------------------------------------------------------------------
-    #  Multi-seed GA
-    # -----------------------------------------------------------------------
-    def train_one_seed(seed):
-        rng = np.random.default_rng(seed)
-        pop = [rng.choice(np.array([-1, 1], dtype=np.int8), size=(V, D))
-               for _ in range(POP_SIZE)]
-        best_K_local = None
-        best_acc_local = 0.0
-        best_mem_local = None
-        for gen in range(GENERATIONS):
-            fits = np.array([evaluate(K) for K in pop])
-            order = np.argsort(fits)[::-1]
-            if fits[order[0]] > best_acc_local:
-                best_acc_local = float(fits[order[0]])
-                best_K_local = pop[order[0]].copy()
-                best_mem_local = build_memory(best_K_local)
-            survivors = [pop[i] for i in order[:ELITE]]
-            while len(survivors) < POP_SIZE:
-                i1, i2 = rng.choice(POP_SIZE, 2, replace=False)
-                p1 = pop[i1] if fits[i1] >= fits[i2] else pop[i2]
-                j1, j2 = rng.choice(POP_SIZE, 2, replace=False)
-                p2 = pop[j1] if fits[j1] >= fits[j2] else pop[j2]
-                child = crossover(p1, p2, rng)
-                child = mutate(child, rng, MUTATION_RATE)
-                survivors.append(child)
-            pop = survivors
-        return best_acc_local, best_K_local, best_mem_local
-    if verbose:
-        print(f"Running GA with {n_seeds} seeds, "
-              f"{GENERATIONS} generations each")
-        print()
-        print(f"  {'seed':>8} | {'accuracy':>9} | {'time':>7}")
-        print("  " + "-" * 32)
-    best_acc = 0.0
-    best_K = None
-    best_memory = None
-    all_accs = []
-    seed_results = []
-    t_all = time.perf_counter()
-    for s in range(n_seeds):
-        seed = SEED_BASE + s * SEED_STEP
-        t0 = time.perf_counter()
-        acc_s, K_s, mem_s = train_one_seed(seed)
-        dt = time.perf_counter() - t0
-        all_accs.append(acc_s)
-        seed_results.append({"seed": seed, "accuracy": acc_s, "time": dt})
-        if verbose:
-            print(f"  {seed:>8} | {acc_s:>9.1%} | {dt:>6.1f}s")
-        if acc_s > best_acc:
-            best_acc = acc_s
-            best_K = K_s
-            best_memory = mem_s
-    total_time = time.perf_counter() - t_all
-    if verbose:
-        print()
-        print(f"Best accuracy:  {best_acc:.1%}")
-        print(f"Mean accuracy:  {np.mean(all_accs):.1%}")
-        print(f"Std:            {np.std(all_accs):.1%}")
-        print(f"Min / Max:      {np.min(all_accs):.1%} / {np.max(all_accs):.1%}")
-        print(f"Total GA time:  {total_time:.1f}s")
-        print()
-    # -----------------------------------------------------------------------
-    #  Parameter counts
-    # -----------------------------------------------------------------------
-    n_vocab_params = V * D
-    n_intent_params = C * D
-    n_total_params = n_vocab_params + n_intent_params
-    # -----------------------------------------------------------------------
-    #  Save
-    # -----------------------------------------------------------------------
-    np.savez_compressed(
-        output_dir / "codebooks.npz",
-        vocab_codebook=best_K,
-        intent_codebook=intent_codebook,
-        memory=best_memory,
-    )
-    npz_size = (output_dir / "codebooks.npz").stat().st_size
-    config = {
-        "model_type": "hv-intent-router",
-        "architecture": "evolved-hypervector",
-        "dimension": D,
-        "vocabulary_size": V,
-        "num_intents": C,
-        "intent_labels": labels,
-        "vocabulary": "".join(chars),
-        "encoding": "sum-bundle",
-        "bundling": "majority-vote",
-        "similarity": "hamming",
-        "num_parameters": n_total_params,
-        "num_parameters_human": (
-            f"{n_total_params/1e6:.1f}M" if n_total_params >= 1e6
-            else f"{n_total_params/1e3:.1f}K" if n_total_params >= 1e3
-            else str(n_total_params)
-        ),
-        "model_size_bytes": int(D // 8),
-        "codebook_size_bytes": int(n_total_params // 8),
-        "total_size_bytes": int(npz_size),
-        "training": {
-            "method": "multi-seed-genetic-algorithm",
-            "population_size": POP_SIZE,
-            "generations": GENERATIONS,
-            "elite_count": ELITE,
-            "mutation_rate": MUTATION_RATE,
-            "crossover_rate": 0.5,
-            "selection": "tournament-2",
-            "n_seeds": n_seeds,
-            "seed_base": SEED_BASE,
-            "seed_step": SEED_STEP,
-            "seed_data": SEED_DATA,
-        },
-        "metrics": {
-            "test_accuracy": float(best_acc),
-            "test_accuracy_mean": float(np.mean(all_accs)),
-            "test_accuracy_std": float(np.std(all_accs)),
-            "test_accuracy_min": float(np.min(all_accs)),
-            "test_accuracy_max": float(np.max(all_accs)),
-            "random_baseline": float(acc_random),
-            "total_training_time_seconds": float(total_time),
-        },
-        "seed_results": seed_results,
-        "weights_file": "codebooks.npz",
-    }
-    with open(output_dir / "config.json", "w") as f:
-        json.dump(config, f, indent=2)
-    if verbose:
-        print(f"Saved codebooks.npz ({npz_size:,} bytes)")
-        print(f"Saved config.json")
-        print(f"Output directory: {output_dir.resolve()}")
-    return config
-# ===========================================================================
-#  CLI
-# ===========================================================================
-def main():
-    parser = argparse.ArgumentParser(
-        description="Train the hv-intent-router model with multi-seed GA.")
-    parser.add_argument(
-        "--output", type=str, default=".",
-        help="Output directory for codebooks.npz and config.json")
-    parser.add_argument(
-        "--dim", type=int, default=D_DEFAULT,
-        help=f"Hypervector dimension (default: {D_DEFAULT})")
-    parser.add_argument(
-        "--seeds", type=int, default=N_SEEDS,
-        help=f"Number of GA seeds to run (default: {N_SEEDS})")
-    parser.add_argument(
-        "--quiet", action="store_true",
-        help="Suppress progress output")
-    args = parser.parse_args()
-    train(args.output, D=args.dim, n_seeds=args.seeds, verbose=not args.quiet)
-if __name__ == "__main__":
-    main()