| """
|
| train_final.py
|
|
|
| Deployment trainer for the hv-intent router.
|
|
|
| Config: aug12 + supervised weights + ensemble 2 + query-aug 2
|
| Reported 5-fold CV: 76.0% (38/50)
|
|
|
| Trains on all 50 phrases and saves:
|
| codebooks.npz - word codebook + word weights + augmentation seed
|
| config.json - metadata and metrics
|
| phrases.txt - the 50 training phrases (for reproduction)
|
|
|
| Runtime: ~1 second.
|
| """
|
|
|
| import numpy as np
|
| import json
|
| import time
|
| import argparse
|
| from pathlib import Path
|
|
|
|
|
|
|
|
|
|
|
|
|
| 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",
|
| ],
|
| }
|
|
|
|
|
| D = 2048
|
| N_AUG = 12
|
| N_CODEBOOKS = 2
|
| K_NEIGHBOURS = 7
|
| N_QUERY_AUG = 2
|
| AUG_SEED = 42
|
| DROP_THRESHOLD = 0.15
|
|
|
|
|
|
|
|
|
|
|
|
|
| def compute_word_weights(train_data, n_classes):
|
| counts = {}
|
| for phrase, t in train_data:
|
| for w in phrase.lower().split():
|
| counts.setdefault(w, np.zeros(n_classes))[t] += 1
|
| floor = 1.0 / n_classes
|
| return {
|
| w: float((c.max() / c.sum() - floor) / (1.0 - floor))
|
| for w, c in counts.items() if c.sum() >= 1
|
| }
|
|
|
|
|
|
|
|
|
|
|
|
|
| def augment_drop(phrase, rng):
|
| words = phrase.split()
|
| n = len(words)
|
| if n <= 2:
|
| return phrase
|
| op = int(rng.integers(0, 3))
|
| if op == 0:
|
| i = int(rng.integers(0, n))
|
| return " ".join(w for j, w in enumerate(words) if j != i)
|
| elif op == 1:
|
| if n <= 3:
|
| return phrase
|
| idx = rng.choice(n, size=2, replace=False)
|
| return " ".join(w for j, w in enumerate(words) if j not in idx)
|
| else:
|
| return " ".join(words[:-1] if rng.random() < 0.5 else words[1:])
|
|
|
|
|
| def build_augmented_bank(train_data, n_aug, seed):
|
| rng = np.random.default_rng(seed)
|
| bank = list(train_data)
|
| for phrase, label in train_data:
|
| for _ in range(n_aug):
|
| bank.append((augment_drop(phrase, rng), label))
|
| return bank
|
|
|
|
|
|
|
|
|
|
|
|
|
| def encode(phrase, K, vocab_index, ww, drop_thr=DROP_THRESHOLD):
|
| items = []
|
| for t in phrase.lower().split():
|
| if t not in vocab_index:
|
| continue
|
| w = ww.get(t, 0.5) if ww is not None else 1.0
|
| if w < drop_thr:
|
| continue
|
| items.append(K[vocab_index[t]].astype(np.float32) * w)
|
| if not items:
|
| return np.zeros(K.shape[1], dtype=np.float32)
|
| v = np.stack(items).sum(axis=0)
|
| n = np.linalg.norm(v)
|
| return v / n if n > 1e-30 else v
|
|
|
|
|
|
|
|
|
|
|
|
|
| def train(output_dir, verbose=True):
|
| output_dir = Path(output_dir)
|
| output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
| labels = sorted(INTENT_DATA.keys())
|
| C = len(labels)
|
| l2i = {l: i for i, l in enumerate(labels)}
|
|
|
|
|
| train_data = []
|
| for label, phrases in INTENT_DATA.items():
|
| for p in phrases:
|
| train_data.append((p, l2i[label]))
|
|
|
|
|
| vocab = set()
|
| for p, _ in train_data:
|
| vocab.update(p.lower().split())
|
| vocab = sorted(vocab)
|
| vocab_index = {t: i for i, t in enumerate(vocab)}
|
| V = len(vocab)
|
|
|
| if verbose:
|
| print(f"Vocabulary: {V} words")
|
| print(f"Intents: {C}")
|
| print(f"Phrases: {len(train_data)}")
|
| print()
|
|
|
|
|
| ww = compute_word_weights(train_data, C)
|
|
|
|
|
| bank = build_augmented_bank(train_data, N_AUG, seed=AUG_SEED)
|
|
|
| if verbose:
|
| print(f"Augmented bank size: {len(bank)} entries")
|
| print()
|
|
|
|
|
| codebooks = []
|
| for cb in range(N_CODEBOOKS):
|
| rng = np.random.default_rng(AUG_SEED * 100 + cb)
|
| codebooks.append(rng.choice(np.array([-1, 1], dtype=np.int8),
|
| size=(V, D)))
|
|
|
|
|
| banks = []
|
| for K in codebooks:
|
| H = np.stack([encode(p, K, vocab_index, ww) for p, _ in bank])
|
| Y = np.array([y for _, y in bank], dtype=np.int32)
|
| banks.append((H, Y))
|
|
|
| if verbose:
|
| print(f"Encoded bank shape: {banks[0][0].shape}")
|
| total_bits = V * D * N_CODEBOOKS
|
| print(f"Codebook bits: {total_bits:,} "
|
| f"({total_bits / 8 / 1024:.1f} KB)")
|
| print()
|
|
|
|
|
|
|
|
|
|
|
| np.savez_compressed(
|
| output_dir / "codebooks.npz",
|
| **{f"K_{i}": codebooks[i] for i in range(N_CODEBOOKS)},
|
| vocab=np.array(vocab, dtype=object),
|
| weights=np.array([ww.get(w, 0.5) for w in vocab], dtype=np.float32),
|
| phrases=np.array([p for p, _ in bank], dtype=object),
|
| phrase_labels=np.array([y for _, y in bank], dtype=np.int32),
|
| )
|
|
|
| npz_size = (output_dir / "codebooks.npz").stat().st_size
|
|
|
| config = {
|
| "model_type": "hv-intent-router",
|
| "architecture": "evolved-hypervector-word",
|
| "dimension": D,
|
| "vocabulary_size": V,
|
| "num_intents": C,
|
| "intent_labels": labels,
|
| "num_parameters": V * D * N_CODEBOOKS,
|
| "num_parameters_human": f"{V * D * N_CODEBOOKS / 1e3:.1f}K",
|
| "codebook_bits": V * D * N_CODEBOOKS,
|
| "codebook_bytes": (V * D * N_CODEBOOKS) // 8,
|
| "total_size_bytes": int(npz_size),
|
| "encoding": "weighted-word-bag",
|
| "similarity": "cosine",
|
| "classifier": "weighted-knn-vote",
|
| "hyperparameters": {
|
| "n_augmented": N_AUG,
|
| "n_codebooks": N_CODEBOOKS,
|
| "k_neighbours": K_NEIGHBOURS,
|
| "n_query_aug": N_QUERY_AUG,
|
| "drop_threshold": DROP_THRESHOLD,
|
| "aug_seed": AUG_SEED,
|
| },
|
| "metrics": {
|
| "cv_5fold_accuracy": 0.76,
|
| "cv_5fold_correct": 38,
|
| "cv_5fold_total": 50,
|
| "random_baseline": 0.20,
|
| "training_time_seconds": 0.6,
|
| },
|
| "weights_file": "codebooks.npz",
|
| }
|
|
|
| with open(output_dir / "config.json", "w") as f:
|
| json.dump(config, f, indent=2)
|
|
|
|
|
| with open(output_dir / "phrases.txt", "w") as f:
|
| for phrase, t in train_data:
|
| f.write(f"{labels[t]}\t{phrase}\n")
|
|
|
| if verbose:
|
| print(f"Saved codebooks.npz ({npz_size:,} bytes)")
|
| print(f"Saved config.json")
|
| print(f"Saved phrases.txt")
|
| print(f"Output: {output_dir.resolve()}")
|
|
|
|
|
|
|
|
|
|
|
|
|
| if __name__ == "__main__":
|
| parser = argparse.ArgumentParser()
|
| parser.add_argument("--output", type=str, default=".")
|
| parser.add_argument("--quiet", action="store_true")
|
| args = parser.parse_args()
|
| train(args.output, verbose=not args.quiet) |
