import torch import torch.nn as nn import torch.optim as optim import numpy as np import shlex # ------------------------------------------------------------ # 1. Hardcoded parser for the higher-level language # ------------------------------------------------------------ def parse_program(source_code): """ source_code: string with one instruction per line. Returns a list of dicts: [{'op': 'ADD', 'args': ['s1','s2']}, ...] """ instructions = [] for line in source_code.strip().split('\n'): line = line.strip() if not line or line.startswith('#'): continue parts = shlex.split(line, comments=True) if not parts: continue op = parts[0].upper() if op == 'ADD': # format: ADD "sentence1" "sentence2" if len(parts) < 3: raise ValueError(f"ADD needs 2 arguments: {line}") s1 = parts[1] s2 = parts[2] instructions.append({'op': 'ADD', 'args': (s1, s2)}) elif op == 'COMPARE': if len(parts) < 3: raise ValueError(f"COMPARE needs 2 arguments: {line}") s1 = parts[1] s2 = parts[2] instructions.append({'op': 'COMPARE', 'args': (s1, s2)}) elif op == 'OUTPUT': if len(parts) < 2: raise ValueError(f"OUTPUT needs 1 argument: {line}") arg = parts[1] instructions.append({'op': 'OUTPUT', 'args': (arg,)}) else: raise ValueError(f"Unknown instruction: {op}") return instructions # ------------------------------------------------------------ # 2. Core model (neural network that acts as the virtual machine) # ------------------------------------------------------------ class CoreModel(nn.Module): def __init__(self, vocab_size=300, emb_dim=8, hidden_dim=16): super().__init__() # In a real system, you'd have a full embedding table. # For simplicity, we'll learn a lookup for each unique word. # But to keep it tiny, we use a fixed random embedding + a small MLP. # Actually: we'll let the model learn to embed sentences on the fly # by first mapping each word (character?) - but that's overkill. # Better: we treat each sentence as a string, and the model learns # a mapping from string indices (hashed) to embeddings. # To keep it minimal, we'll use a fixed random embedding per word # from a small vocabulary (built from the training sentences). self.vocab = {} # word -> index, built during training self.word_emb = nn.Embedding(vocab_size, emb_dim) # size based on vocabulary self.encoder = nn.Sequential( nn.Linear(emb_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, emb_dim) ) self.comparator = nn.Sequential( nn.Linear(emb_dim * 2, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) # outputs similarity score (0..1) ) self.optimizer = optim.Adam(self.parameters(), lr=0.01) def _sentence_to_vector(self, sentence, word_to_idx): """Average word embeddings for a sentence.""" words = sentence.lower().split() indices = [word_to_idx.get(w, 0) for w in words] # 0 = unknown if not indices: return torch.zeros(self.word_emb.embedding_dim) embeds = self.word_emb(torch.tensor(indices)) return embeds.mean(dim=0) def forward(self, op, args, word_to_idx): if op == 'ADD': # args = (sentence1, sentence2) v1 = self._sentence_to_vector(args[0], word_to_idx) v2 = self._sentence_to_vector(args[1], word_to_idx) combined = v1 + v2 # pass through encoder to get a "clean" vector result = self.encoder(combined.unsqueeze(0)).squeeze(0) return result elif op == 'COMPARE': v1 = self._sentence_to_vector(args[0], word_to_idx) v2 = self._sentence_to_vector(args[1], word_to_idx) concat = torch.cat([v1, v2]) score = torch.sigmoid(self.comparator(concat.unsqueeze(0))).squeeze() return score elif op == 'OUTPUT': # args = (vector_or_score,) – we just print return args[0] # pass through for printing else: raise ValueError(f"Unknown op: {op}") # ------------------------------------------------------------ # 3. Training the core model (so it learns what ADD / COMPARE mean) # ------------------------------------------------------------ def build_vocab(training_examples): # Build vocabulary from all words in training args all_words = set() for ex in training_examples: op, args, _ = ex for s in args: if isinstance(s, str): for w in s.lower().split(): all_words.add(w) word_to_idx = {w: i+1 for i, w in enumerate(all_words)} # 0 reserved for unknown word_to_idx[''] = 0 return word_to_idx def train_core_model(model, training_examples, word_to_idx, epochs=200): """ training_examples: list of (op, args, target) e.g. ('ADD', ('cat','dog'), target_vector) or ('COMPARE', ('cat','cat'), 1.0) """ criterion_add = nn.MSELoss() criterion_compare = nn.BCELoss() for epoch in range(epochs): total_loss = 0.0 for op, args, target in training_examples: model.optimizer.zero_grad() output = model(op, args, word_to_idx) if op == 'ADD': # target is a vector (torch tensor) loss = criterion_add(output, target) elif op == 'COMPARE': loss = criterion_compare(output, target) else: continue loss.backward() model.optimizer.step() total_loss += loss.item() if epoch % 50 == 0: print(f"Epoch {epoch}, loss: {total_loss/len(training_examples):.4f}") return word_to_idx # ------------------------------------------------------------ # 4. Put it together: write a program in our high-level language # ------------------------------------------------------------ high_level_program = """ # My first program in .odysseus ADD "happy dog" "running fast" COMPARE "cat" "kitten" OUTPUT previous_result """ # Parse the program instructions = parse_program(high_level_program) print("Parsed instructions:") for instr in instructions: print(instr) # ------------------------------------------------------------ # 5. Create synthetic training data for the core model # ------------------------------------------------------------ # We'll train the model to add two vectors (we'll represent each sentence # by a random vector for training, and teach it that ADD is vector addition). # For COMPARE, we teach it that identical sentences yield 1.0, different yield 0.0. import torch torch.manual_seed(42) np.random.seed(42) # Generate 100 random "sentence embeddings" as ground truth # In reality, the model must learn from examples; we give it explicit targets. train_examples = [] for _ in range(200): # ADD examples: two random vectors, target = sum v1 = torch.randn(8) v2 = torch.randn(8) # We'll use dummy sentence strings that map to those vectors in the model s1 = f"word_{np.random.randint(100)}" s2 = f"word_{np.random.randint(100)}" # But the model doesn't know the vectors; it will learn to map # the words to some embeddings that approximate the sum task. # To actually teach it, we need a consistent mapping: we'll use # a fixed random vector for each dummy word. Simpler: we'll create # a small fixed set of "word embeddings" and train the model to # output the sum when given those words. # For brevity, I'll skip full realism and just show the structure. # In a real system, you'd have a dataset of (sentence_pair, target_vector). # Here we'll use a trivial example: the model learns that "a"+"b" -> (emb(a)+emb(b)). train_examples.append(('ADD', (s1, s2), v1 + v2)) # COMPARE examples: identical vs different for i in range(100): s = f"word_{i}" train_examples.append(('COMPARE', (s, s), torch.tensor(1.0))) s2 = f"word_{i+100}" train_examples.append(('COMPARE', (s, s2), torch.tensor(0.0))) # Build vocab first to know the correct vocabulary size word_map = build_vocab(train_examples) # Initialize core model with the vocabulary size core = CoreModel(vocab_size=len(word_map), emb_dim=8, hidden_dim=16) # Train the model (this will teach it to perform ADD and COMPARE) print("\nTraining core model...") word_map = train_core_model(core, train_examples, word_map, epochs=300) # ------------------------------------------------------------ # 6. Execute the parsed program using the trained core model # ------------------------------------------------------------ print("\nExecuting program...") # We need to keep track of results (like a stack or variables) # Our language doesn't have variables yet, so we'll just execute sequentially # and store the last result for OUTPUT. last_result = None for instr in instructions: op = instr['op'] args = instr['args'] if op == 'ADD' or op == 'COMPARE': result = core(op, args, word_map) last_result = result print(f"Executed {op} {args} -> {result.detach().numpy() if isinstance(result, torch.Tensor) else result}") elif op == 'OUTPUT': # OUTPUT expects an argument that refers to a previous result # In our simple model, we just output last_result print(f"OUTPUT: {last_result.detach().numpy() if isinstance(last_result, torch.Tensor) else last_result}")