Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA8_neural_program_synthesis /neural_program_synthesis.py
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| from abc import ABC, abstractmethod | |
| from typing import List, Dict, Tuple, Optional, Union | |
| import ast | |
| import math | |
| import numpy as np | |
| import random | |
| import re | |
| import subprocess | |
| import tempfile | |
| import os | |
| from dataclasses import dataclass | |
| from enum import Enum | |
| class ProgramEncoder(nn.Module): | |
| def __init__( | |
| self, vocab_size: int, embedding_dim: int, hidden_dim: int, num_layers: int = 2 | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.embedding_dim = embedding_dim | |
| self.hidden_dim = hidden_dim | |
| self.num_layers = num_layers | |
| self.embedding = nn.Embedding(vocab_size, embedding_dim) | |
| self.lstm = nn.LSTM( | |
| input_size=embedding_dim, | |
| hidden_size=hidden_dim, | |
| num_layers=num_layers, | |
| dropout=0.1 if num_layers > 1 else 0, | |
| batch_first=True, | |
| ) | |
| def forward(self, input_ids: torch.Tensor, lengths: torch.Tensor): | |
| embedded = self.embedding(input_ids) | |
| packed = nn.utils.rnn.pack_padded_sequence( | |
| embedded, lengths, batch_first=True, enforce_sorted=False | |
| ) | |
| output, (hidden, cell) = self.lstm(packed) | |
| output, _ = nn.utils.rnn.pad_packed_sequence(output, batch_first=True) | |
| return output, hidden, cell | |
| class ProgramDecoder(nn.Module): | |
| def __init__( | |
| self, vocab_size: int, embedding_dim: int, hidden_dim: int, num_layers: int = 2 | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.embedding_dim = embedding_dim | |
| self.hidden_dim = hidden_dim | |
| self.num_layers = num_layers | |
| self.embedding = nn.Embedding(vocab_size, embedding_dim) | |
| self.lstm = nn.LSTM( | |
| input_size=embedding_dim, | |
| hidden_size=hidden_dim, | |
| num_layers=num_layers, | |
| dropout=0.1 if num_layers > 1 else 0, | |
| batch_first=True, | |
| ) | |
| self.attention = nn.Linear(hidden_dim * 2, 1) | |
| self.output_proj = nn.Linear(hidden_dim, vocab_size) | |
| def forward( | |
| self, | |
| decoder_input_ids: torch.Tensor, | |
| encoder_outputs: torch.Tensor, | |
| hidden: torch.Tensor, | |
| cell: torch.Tensor, | |
| ): | |
| embedded = self.embedding(decoder_input_ids) | |
| lstm_out, (hidden, cell) = self.lstm(embedded, (hidden, cell)) | |
| batch_size, decoder_len, _ = lstm_out.shape | |
| _, encoder_len, _ = encoder_outputs.shape | |
| lstm_out_expanded = lstm_out.unsqueeze(2).expand(-1, -1, encoder_len, -1) | |
| encoder_outputs_expanded = encoder_outputs.unsqueeze(1).expand( | |
| -1, decoder_len, -1, -1 | |
| ) | |
| concat = torch.cat([lstm_out_expanded, encoder_outputs_expanded], dim=-1) | |
| attn_scores = self.attention(concat).squeeze(-1) | |
| attn_weights = F.softmax(attn_scores, dim=-1) | |
| context = torch.bmm( | |
| attn_weights.reshape(-1, encoder_len), | |
| encoder_outputs.reshape(-1, encoder_len, -1), | |
| ) | |
| context = context.reshape(batch_size, decoder_len, -1) | |
| logits = self.output_proj(lstm_out) | |
| return logits, hidden, cell, attn_weights | |
| class Seq2SeqProgramSynthesizer(nn.Module): | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int = 256, | |
| hidden_dim: int = 512, | |
| num_layers: int = 2, | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.encoder = ProgramEncoder(vocab_size, embedding_dim, hidden_dim, num_layers) | |
| self.decoder = ProgramDecoder(vocab_size, embedding_dim, hidden_dim, num_layers) | |
| def forward( | |
| self, | |
| input_ids: torch.Tensor, | |
| input_lengths: torch.Tensor, | |
| target_ids: torch.Tensor, | |
| ): | |
| enc_output, hidden, cell = self.encoder(input_ids, input_lengths) | |
| dec_output, _, _, _ = self.decoder(target_ids[:, :-1], enc_output, hidden, cell) | |
| return dec_output | |
| def generate( | |
| self, | |
| input_ids: torch.Tensor, | |
| input_lengths: torch.Tensor, | |
| max_length: int = 100, | |
| bos_id: int = 1, | |
| eos_id: int = 2, | |
| ): | |
| self.eval() | |
| with torch.no_grad(): | |
| enc_output, hidden, cell = self.encoder(input_ids, input_lengths) | |
| batch_size = input_ids.size(0) | |
| prev_ids = torch.full( | |
| (batch_size, 1), bos_id, dtype=torch.long, device=input_ids.device | |
| ) | |
| generated_tokens = [] | |
| for _ in range(max_length): | |
| dec_output, hidden, cell, _ = self.decoder( | |
| prev_ids, enc_output, hidden, cell | |
| ) | |
| prev_ids = torch.argmax(dec_output[:, -1:, :], dim=-1) | |
| generated_tokens.append(prev_ids.squeeze(-1)) | |
| if (prev_ids == eos_id).all(): | |
| break | |
| return torch.stack(generated_tokens, dim=1) | |
| class TransformerProgramSynthesizer(nn.Module): | |
| def __init__( | |
| self, vocab_size: int, d_model: int = 512, nhead: int = 8, num_layers: int = 6 | |
| ): | |
| super().__init__() | |
| self.d_model = d_model | |
| self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=0) | |
| self.pos_encoding = PositionalEncoding(d_model) | |
| self.transformer = nn.Transformer( | |
| d_model=d_model, | |
| nhead=nhead, | |
| num_encoder_layers=num_layers, | |
| num_decoder_layers=num_layers, | |
| ) | |
| self.output_proj = nn.Linear(d_model, vocab_size) | |
| def _generate_square_subsequent_mask(sz: int) -> torch.Tensor: | |
| mask = torch.triu(torch.ones(sz, sz) * float("-inf"), diagonal=1) | |
| return mask | |
| def forward(self, src_ids: torch.Tensor, tgt_ids: torch.Tensor): | |
| src_padding_mask = src_ids == 0 | |
| tgt_padding_mask = tgt_ids == 0 | |
| src_embedded = self.embedding(src_ids).transpose(0, 1) | |
| tgt_embedded = self.embedding(tgt_ids).transpose(0, 1) | |
| src = self.pos_encoding(src_embedded) | |
| tgt = self.pos_encoding(tgt_embedded) | |
| tgt_mask = self._generate_square_subsequent_mask(tgt.size(0)).to(tgt.device) | |
| transformer_out = self.transformer( | |
| src, | |
| tgt, | |
| tgt_mask=tgt_mask, | |
| src_key_padding_mask=src_padding_mask, | |
| tgt_key_padding_mask=tgt_padding_mask, | |
| ) | |
| return self.output_proj(transformer_out.transpose(0, 1)) | |
| class PositionalEncoding(nn.Module): | |
| def __init__(self, d_model: int, dropout: float = 0.1, max_len: int = 5000): | |
| super().__init__() | |
| self.dropout = nn.Dropout(dropout) | |
| pe = torch.zeros(max_len, d_model) | |
| position = torch.arange(0, max_len).unsqueeze(1).float() | |
| div_term = torch.exp( | |
| torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model) | |
| ) | |
| pe[:, 0::2] = torch.sin(position * div_term) | |
| pe[:, 1::2] = torch.cos(position * div_term) | |
| self.register_buffer("pe", pe.unsqueeze(1)) | |
| def forward(self, x: torch.Tensor) -> torch.Tensor: | |
| x = x + self.pe[: x.size(0), :].transpose(0, 1) | |
| return self.dropout(x) | |
| class CodeGraphEncoder(nn.Module): | |
| def __init__( | |
| self, vocab_size: int, embedding_dim: int, hidden_dim: int, num_layers: int = 3 | |
| ): | |
| super().__init__() | |
| self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=0) | |
| self.gnn_layers = nn.ModuleList( | |
| [ | |
| nn.Linear(embedding_dim if i == 0 else hidden_dim, hidden_dim) | |
| for i in range(num_layers) | |
| ] | |
| ) | |
| self.activation = nn.ReLU() | |
| def forward(self, node_features: torch.Tensor, adjacency: torch.Tensor): | |
| x = self.embedding(node_features) | |
| for layer in self.gnn_layers: | |
| x = layer(x) | |
| x = self.activation(x) | |
| if adjacency.size(-1) > 0: | |
| x = adjacency @ x / (torch.sum(adjacency, dim=-1, keepdim=True) + 1e-6) | |
| return x | |
| class ProgramExecutor(ABC): | |
| def execute( | |
| self, program_code: str, input_data: Dict | |
| ) -> Tuple[bool, Dict, Optional[str]]: | |
| pass | |
| class PythonExecutor(ProgramExecutor): | |
| def execute( | |
| self, program_code: str, input_data: Dict | |
| ) -> Tuple[bool, Dict, Optional[str]]: | |
| try: | |
| namespace = {**input_data} | |
| exec(program_code, namespace) | |
| return True, namespace, None | |
| except Exception as e: | |
| return False, {}, str(e) | |
| class JavaScriptExecutor(ProgramExecutor): | |
| def execute( | |
| self, program_code: str, input_data: Dict | |
| ) -> Tuple[bool, Dict, Optional[str]]: | |
| return False, {}, "Not implemented" | |
| class ProgramTokenizer(ABC): | |
| def __init__(self, vocab_size: int): | |
| self.vocab_size = vocab_size | |
| def tokenize(self, program_code: str) -> List[int]: | |
| pass | |
| def detokenize(self, token_ids: List[int]) -> str: | |
| pass | |
| class SimpleTokenizer(ProgramTokenizer): | |
| def __init__(self, vocab_size: int = 5000): | |
| super().__init__(vocab_size) | |
| self.word_to_id = {} | |
| self.id_to_word = {} | |
| def tokenize(self, program_code: str) -> List[int]: | |
| tokens = program_code.split() | |
| return [self.word_to_id.get(tok, 0) for tok in tokens if tok] | |
| def detokenize(self, token_ids: List[int]) -> str: | |
| return " ".join(self.id_to_word.get(id, "<UNK>") for id in token_ids) | |
| class ProgramTemplate(ABC): | |
| def instantiate(self, hole_values: Dict[str, str]) -> str: | |
| pass | |
| def get_holes(self) -> List[str]: | |
| pass | |
| class SimpleTemplate(ProgramTemplate): | |
| def __init__(self, template_str: str, holes: List[str]): | |
| self.template_str = template_str | |
| self.holes = holes | |
| def instantiate(self, hole_values: Dict[str, str]) -> str: | |
| return self.template_str.format( | |
| **{k: v for k, v in hole_values.items() if k in self.holes} | |
| ) | |
| def get_holes(self) -> List[str]: | |
| return self.holes.copy() | |
| from typing import List | |
| class TemplateSynthesizer(nn.Module): | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int, | |
| hidden_dim: int, | |
| templates: List[ProgramTemplate], | |
| ): | |
| super().__init__() | |
| self.templates = templates | |
| self.template_predictor = nn.Sequential( | |
| nn.Linear(hidden_dim, hidden_dim), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim, len(templates)), | |
| ) | |
| self.hole_fillers = nn.ModuleDict() | |
| for i, template in enumerate(templates): | |
| if template.holes: | |
| self.hole_fillers[f"template_{i}"] = nn.ModuleDict( | |
| { | |
| hole: nn.Sequential( | |
| nn.Linear(hidden_dim, hidden_dim // 2), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim // 2, vocab_size), | |
| ) | |
| for hole in template.holes | |
| } | |
| ) | |
| def forward(self, example_embedding: torch.Tensor): | |
| template_logits = self.template_predictor(example_embedding.squeeze()) | |
| template_probs = F.softmax(template_logits, dim=-1) | |
| hole_predictions = {} | |
| for i, prob in enumerate(template_probs): | |
| if prob > 0.1: | |
| template_holes = self.hole_fillers.get(f"template_{i}", {}) | |
| for hole_name, hole_predictor in template_holes.items(): | |
| hole_predictions[f"{i}_{hole_name}"] = F.softmax( | |
| hole_predictor(example_embedding), dim=-1 | |
| ) | |
| return template_logits, hole_predictions | |
| class MultiModalCodeEncoder(nn.Module): | |
| def __init__( | |
| self, | |
| code_vocab_size: int, | |
| text_vocab_size: int, | |
| embedding_dim: int, | |
| hidden_dim: int, | |
| ): | |
| super().__init__() | |
| self.code_embedding = nn.Embedding(code_vocab_size, embedding_dim) | |
| self.text_embedding = nn.Embedding(text_vocab_size, embedding_dim) | |
| self.code_encoder = nn.LSTM( | |
| input_size=embedding_dim, hidden_size=hidden_dim, batch_first=True | |
| ) | |
| self.text_encoder = nn.LSTM( | |
| input_size=embedding_dim, hidden_size=hidden_dim, batch_first=True | |
| ) | |
| self.fusion_layer = nn.Sequential( | |
| nn.Linear(2 * hidden_dim, hidden_dim), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim, hidden_dim // 2), | |
| ) | |
| def forward( | |
| self, | |
| code_ids: torch.Tensor, | |
| text_ids: torch.Tensor, | |
| code_lengths: torch.Tensor, | |
| text_lengths: torch.Tensor, | |
| ): | |
| code_embedded = self.code_embedding(code_ids) | |
| text_embedded = self.text_embedding(text_ids) | |
| packed_code = nn.utils.rnn.pack_padded_sequence( | |
| code_embedded, code_lengths, batch_first=True, enforce_sorted=False | |
| ) | |
| packed_text = nn.utils.rnn.pack_padded_sequence( | |
| text_embedded, text_lengths, batch_first=True, enforce_sorted=False | |
| ) | |
| _, (code_hidden, _) = self.code_encoder(packed_code) | |
| _, (text_hidden, _) = self.text_encoder(packed_text) | |
| fused = torch.cat([code_hidden[-1], text_hidden[-1]], dim=-1) | |
| fused_representation = self.fusion_layer(fused) | |
| return fused_representation | |
| class ProgramSynthesisTrainer(ABC): | |
| def __init__( | |
| self, | |
| synthesizer: nn.Module, | |
| optimizer: torch.optim.Optimizer, | |
| device: torch.device, | |
| ): | |
| self.synthesizer = synthesizer | |
| self.optimizer = optimizer | |
| self.device = device | |
| def train_step(self, batch: Dict) -> Dict[str, float]: | |
| pass | |
| def evaluate(self, batch: Dict) -> Dict[str, float]: | |
| pass | |
| class Seq2SeqTrainer(ProgramSynthesisTrainer): | |
| def __init__( | |
| self, | |
| synthesizer: Seq2SeqProgramSynthesizer, | |
| optimizer: torch.optim.Optimizer, | |
| device: torch.device, | |
| criterion: nn.CrossEntropyLoss = nn.CrossEntropyLoss(reduction="mean"), | |
| ): | |
| super().__init__(synthesizer, optimizer, device) | |
| self.criterion = criterion | |
| def train_step(self, batch: Dict) -> Dict[str, float]: | |
| self.optimizer.zero_grad() | |
| outputs = self.synthesizer( | |
| batch["src_ids"], batch["src_lengths"], batch["tgt_ids"] | |
| ) | |
| loss = self.criterion( | |
| outputs.reshape(-1, outputs.size(-1)), batch["tgt_ids"][:, 1:].reshape(-1) | |
| ) | |
| loss.backward() | |
| self.optimizer.step() | |
| return {"loss": loss.item()} | |
| def evaluate(self, batch: Dict) -> Dict[str, float]: | |
| with torch.no_grad(): | |
| outputs = self.synthesizer( | |
| batch["src_ids"], batch["src_lengths"], batch["tgt_ids"] | |
| ) | |
| loss = self.criterion( | |
| outputs.reshape(-1, outputs.size(-1)), | |
| batch["tgt_ids"][:, 1:].reshape(-1), | |
| ) | |
| return {"loss": loss.item()} | |
| class ProgramEvaluator: | |
| def __init__(self): | |
| self.executor = PythonExecutor() | |
| def evaluate_syntax(self, program_code: str) -> bool: | |
| try: | |
| ast.parse(program_code) | |
| return True | |
| except: | |
| return False | |
| def evaluate_functional( | |
| self, program_code: str, test_cases: List[Dict] | |
| ) -> List[bool]: | |
| results = [] | |
| for test_case in test_cases: | |
| inp = test_case.get("input", {}) | |
| exp_out = test_case.get("expected_output", {}) | |
| success, out_data, _ = self.executor.execute(program_code, inp) | |
| match = False | |
| try: | |
| if success: | |
| match = str(out_data).strip() == str(exp_out).strip() | |
| results.append(match) | |
| except: | |
| results.append(False) | |
| return results | |
| class DSLProgramSynthesizer(ABC): | |
| def synthesize(self, specification: str) -> str: | |
| return NotImplemented | |
| class ArithmeticDSLSynthesizer(DSLProgramSynthesizer): | |
| def __init__(self): | |
| super().__init__() | |
| def synthesize(self, specification: str) -> str: | |
| return "result = var1 + var2" | |
| def compare_programs( | |
| programs: List[str], | |
| evaluator: ProgramEvaluator, | |
| test_cases: List[Tuple[Dict, Dict, bool]], | |
| ) -> List[int]: | |
| scores = [] | |
| for prog in programs: | |
| if not evaluator.evaluate_syntax(prog): | |
| scores.append(0.0) | |
| continue | |
| func_tests = evaluator.evaluate_functional(prog, test_cases) | |
| avg_success = sum(func_tests) / len(func_tests) if func_tests else 0.0 | |
| scores.append(avg_success) | |
| indices = np.argsort(scores)[::-1].tolist() | |
| return indices | |
| # ============================================================================ | |
| # ADVANCED PROGRAM SYNTHESIS TECHNIQUES | |
| # ============================================================================ | |
| class ProgramNode: | |
| """Node in hierarchical program tree""" | |
| code: str | |
| parent: Optional["ProgramNode"] = None | |
| children: List["ProgramNode"] = None | |
| depth: int = 0 | |
| def __post_init__(self): | |
| if self.children is None: | |
| self.children = [] | |
| class HierarchicalProgramSynthesizer(nn.Module): | |
| """ | |
| Hierarchical Neural Program Synthesis | |
| Synthesizes complex programs by composing simpler sub-programs | |
| Based on: https://arxiv.org/abs/2303.06018 | |
| """ | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int = 256, | |
| hidden_dim: int = 512, | |
| max_depth: int = 3, | |
| num_layers: int = 2, | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.embedding_dim = embedding_dim | |
| self.hidden_dim = hidden_dim | |
| self.max_depth = max_depth | |
| # Encoder for specifications | |
| self.spec_encoder = ProgramEncoder( | |
| vocab_size, embedding_dim, hidden_dim, num_layers | |
| ) | |
| # Hierarchical decomposition network | |
| self.decomposer = nn.Sequential( | |
| nn.Linear(hidden_dim, hidden_dim), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim, max_depth * 2), | |
| ) | |
| # Sub-program synthesizers for each level | |
| self.level_synthesizers = nn.ModuleList( | |
| [ | |
| ProgramDecoder(vocab_size, embedding_dim, hidden_dim, num_layers) | |
| for _ in range(max_depth) | |
| ] | |
| ) | |
| # Composition network | |
| self.composer = nn.LSTM( | |
| input_size=hidden_dim, | |
| hidden_size=hidden_dim, | |
| num_layers=num_layers, | |
| batch_first=True, | |
| ) | |
| self.output_proj = nn.Linear(hidden_dim, vocab_size) | |
| def forward( | |
| self, | |
| spec_ids: torch.Tensor, | |
| spec_lengths: torch.Tensor, | |
| target_ids: torch.Tensor = None, | |
| ): | |
| # Encode specification | |
| spec_output, spec_hidden, spec_cell = self.spec_encoder(spec_ids, spec_lengths) | |
| # Predict hierarchical structure | |
| structure_logits = self.decomposer(spec_hidden[-1]) | |
| # Generate sub-programs at each level | |
| level_outputs = [] | |
| for level_idx, synthesizer in enumerate(self.level_synthesizers): | |
| if target_ids is not None: | |
| level_out, _, _, _ = synthesizer( | |
| target_ids[:, :-1], spec_output, spec_hidden, spec_cell | |
| ) | |
| else: | |
| level_out = None | |
| level_outputs.append(level_out) | |
| return level_outputs, structure_logits | |
| def synthesize_hierarchical( | |
| self, | |
| spec_ids: torch.Tensor, | |
| spec_lengths: torch.Tensor, | |
| max_length: int = 100, | |
| ) -> ProgramNode: | |
| """Synthesize program hierarchically""" | |
| self.eval() | |
| with torch.no_grad(): | |
| spec_output, spec_hidden, spec_cell = self.spec_encoder( | |
| spec_ids, spec_lengths | |
| ) | |
| root = self._build_program_tree( | |
| spec_output, spec_hidden, spec_cell, depth=0, max_length=max_length | |
| ) | |
| return root | |
| def _build_program_tree( | |
| self, | |
| context: torch.Tensor, | |
| hidden: torch.Tensor, | |
| cell: torch.Tensor, | |
| depth: int, | |
| max_length: int, | |
| ) -> ProgramNode: | |
| """Recursively build program tree""" | |
| if depth >= self.max_depth: | |
| code = self._generate_code(context, hidden, cell, max_length) | |
| return ProgramNode(code=code, depth=depth) | |
| node = ProgramNode(code="", depth=depth) | |
| num_children = random.randint(2, 3) | |
| for _ in range(num_children): | |
| child = self._build_program_tree( | |
| context, hidden, cell, depth + 1, max_length | |
| ) | |
| child.parent = node | |
| node.children.append(child) | |
| node.code = self._compose_code([c.code for c in node.children]) | |
| return node | |
| def _generate_code(self, context, hidden, cell, max_length) -> str: | |
| return "result = a + b" | |
| def _compose_code(self, sub_codes: List[str]) -> str: | |
| return "\n".join(sub_codes) | |
| class DifferentiableProgramFixer(nn.Module): | |
| """ | |
| Differentiable Program Fixer | |
| Iteratively refines generated programs based on execution results | |
| Based on: https://arxiv.org/abs/2006.10924 | |
| """ | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int = 256, | |
| hidden_dim: int = 512, | |
| num_layers: int = 2, | |
| max_fix_iterations: int = 5, | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.max_fix_iterations = max_fix_iterations | |
| self.program_encoder = ProgramEncoder( | |
| vocab_size, embedding_dim, hidden_dim, num_layers | |
| ) | |
| self.error_encoder = nn.LSTM( | |
| input_size=embedding_dim, | |
| hidden_size=hidden_dim, | |
| num_layers=num_layers, | |
| batch_first=True, | |
| ) | |
| self.error_embedding = nn.Embedding(vocab_size, embedding_dim) | |
| self.fixer = nn.TransformerDecoder( | |
| nn.TransformerDecoderLayer(d_model=hidden_dim, nhead=8), | |
| num_layers=num_layers, | |
| ) | |
| self.output_proj = nn.Linear(hidden_dim, vocab_size) | |
| def forward( | |
| self, | |
| program_ids: torch.Tensor, | |
| program_lengths: torch.Tensor, | |
| error_ids: torch.Tensor, | |
| error_lengths: torch.Tensor, | |
| ): | |
| prog_output, prog_hidden, prog_cell = self.program_encoder( | |
| program_ids, program_lengths | |
| ) | |
| error_embedded = self.error_embedding(error_ids) | |
| error_output, (error_hidden, _) = self.error_encoder(error_embedded) | |
| fixed_output = self.fixer( | |
| prog_output.transpose(0, 1), | |
| error_output.transpose(0, 1), | |
| ).transpose(0, 1) | |
| logits = self.output_proj(fixed_output) | |
| return logits | |
| class REPLGuidedSynthesizer(nn.Module): | |
| """ | |
| REPL-Guided Program Synthesis | |
| Uses Read-Eval-Print Loop to guide synthesis with execution feedback | |
| Based on: https://arxiv.org/abs/1906.04604 | |
| """ | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int = 256, | |
| hidden_dim: int = 512, | |
| num_layers: int = 2, | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.synthesizer = Seq2SeqProgramSynthesizer( | |
| vocab_size, embedding_dim, hidden_dim, num_layers | |
| ) | |
| self.feedback_encoder = nn.LSTM( | |
| input_size=embedding_dim, | |
| hidden_size=hidden_dim, | |
| num_layers=num_layers, | |
| batch_first=True, | |
| ) | |
| self.feedback_embedding = nn.Embedding(vocab_size, embedding_dim) | |
| self.state_fusion = nn.Sequential( | |
| nn.Linear(hidden_dim * 2, hidden_dim), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim, hidden_dim), | |
| ) | |
| class TreeOfThoughtsProgramSynthesizer(nn.Module): | |
| """ | |
| Tree-of-Thoughts for Program Synthesis | |
| Explores multiple reasoning paths to synthesize programs | |
| """ | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int = 256, | |
| hidden_dim: int = 512, | |
| num_layers: int = 2, | |
| num_thoughts: int = 3, | |
| max_depth: int = 4, | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.num_thoughts = num_thoughts | |
| self.max_depth = max_depth | |
| self.synthesizer = Seq2SeqProgramSynthesizer( | |
| vocab_size, embedding_dim, hidden_dim, num_layers | |
| ) | |
| self.thought_evaluator = nn.Sequential( | |
| nn.Linear(hidden_dim, hidden_dim // 2), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim // 2, 1), | |
| nn.Sigmoid(), | |
| ) | |
| self.thought_generator = nn.ModuleList( | |
| [ | |
| ProgramDecoder(vocab_size, embedding_dim, hidden_dim, num_layers) | |
| for _ in range(num_thoughts) | |
| ] | |
| ) | |
| class ReinforcementLearningProgramSynthesizer(nn.Module): | |
| """ | |
| RL-based Program Synthesis | |
| Uses reinforcement learning to optimize program generation | |
| """ | |
| def __init__( | |
| self, | |
| vocab_size: int, | |
| embedding_dim: int = 256, | |
| hidden_dim: int = 512, | |
| num_layers: int = 2, | |
| ): | |
| super().__init__() | |
| self.vocab_size = vocab_size | |
| self.policy_net = Seq2SeqProgramSynthesizer( | |
| vocab_size, embedding_dim, hidden_dim, num_layers | |
| ) | |
| self.value_net = nn.Sequential( | |
| nn.Linear(hidden_dim, hidden_dim // 2), | |
| nn.ReLU(), | |
| nn.Linear(hidden_dim // 2, 1), | |
| ) | |
| def demo_neural_program_synthesis(): | |
| vocab_size = 1000 | |
| embedding_dim = 100 | |
| hidden_dim = 200 | |
| synthesizer = Seq2SeqProgramSynthesizer(vocab_size, embedding_dim, hidden_dim) | |
| print("Neural Program Synthesis Model Initialized") | |
| print("\n=== Advanced Models ===") | |
| hierarchical = HierarchicalProgramSynthesizer(vocab_size, embedding_dim, hidden_dim) | |
| print("✓ Hierarchical Program Synthesizer") | |
| fixer = DifferentiableProgramFixer(vocab_size, embedding_dim, hidden_dim) | |
| print("✓ Differentiable Program Fixer") | |
| repl = REPLGuidedSynthesizer(vocab_size, embedding_dim, hidden_dim) | |
| print("✓ REPL-Guided Synthesizer") | |
| tot = TreeOfThoughtsProgramSynthesizer(vocab_size, embedding_dim, hidden_dim) | |
| print("✓ Tree-of-Thoughts Synthesizer") | |
| rl = ReinforcementLearningProgramSynthesizer(vocab_size, embedding_dim, hidden_dim) | |
| print("✓ RL-based Synthesizer") | |
| if __name__ == "__main__": | |
| demo_neural_program_synthesis() | |
Xet Storage Details
- Size:
- 26.9 kB
- Xet hash:
- faab1810f6d1a0d5589603892b3013d338c75c6ed688345352612138c780e4e5
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.