### Configure Environment File Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Prepares the environment variables by copying the example configuration file. ```bash cd ARC-AGI-3-Agents cp .env-example .env # Then edit .env file and replace the empty ARC_API_KEY= with your actual API key cd .. ``` -------------------------------- ### Setup Experiment Directories Source: https://context7.com/driessmit/arc3-solution/llms.txt Creates timestamped directories and tracks git state for reproducibility. Returns the base directory path and log file location. ```python from datetime import datetime import subprocess import logging import os def setup_experiment_directory(base_output_dir='runs'): """Create directories for outputs and logging.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") base_dir = os.path.join(base_output_dir, timestamp) os.makedirs(base_dir, exist_ok=True) log_file = os.path.join(base_dir, 'logs.log') # Save git commit and diff info commit = subprocess.check_output(['git', 'rev-parse', 'HEAD']).decode().strip() diff = subprocess.check_output(['git', 'diff']).decode() with open(os.path.join(base_dir, 'git_info.txt'), 'w') as f: f.write(f"commit: {commit}\n\n{diff}") return base_dir, log_file def get_environment_directory(base_dir, game_id): """Get or create environment-specific directory for a game_id.""" env_dir = os.path.join(base_dir, game_id) os.makedirs(env_dir, exist_ok=True) return env_dir # Example usage base_dir, log_file = setup_experiment_directory() # Output: runs/20240115_143022/ env_dir = get_environment_directory(base_dir, "vc33") # Output: runs/20240115_143022/vc33/ ``` -------------------------------- ### Install Dependencies Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Installs the required Python dependencies using the Makefile. ```bash make install ``` -------------------------------- ### Clone Repository Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Initializes the project by cloning the repository and its submodules. ```bash git clone --recurse-submodules git@github.com:DriesSmit/ARC3-solution.git cd ARC3-solution ``` -------------------------------- ### Monitor Training Metrics Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Launches TensorBoard to visualize training progress and metrics. ```bash # View training metrics make tensorboard # Open http://localhost:6006 in browser ``` -------------------------------- ### Initialize Action Agent Source: https://context7.com/driessmit/arc3-solution/llms.txt Initializes the Action agent, setting up reinforcement learning components, experiment directories, and TensorBoard logging. Ensure necessary libraries like `torch`, `collections`, and `torch.optim` are imported. ```python from agents.agent import Agent from agents.structs import FrameData, GameAction, GameState from collections import deque from torch.utils.tensorboard import SummaryWriter import torch.optim as optim class Action(Agent): """Agent using action model to predict which actions lead to new frames.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Seed for reproducibility seed = int(time.time() * 1000000) + hash(self.game_id) % 1000000 random.seed(seed) np.random.seed(seed % (2**32 - 1)) torch.manual_seed(seed % (2**32 - 1)) # Device configuration self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # Setup experiment directory and TensorBoard self.base_dir, log_file = setup_experiment_directory() env_dir = get_environment_directory(self.base_dir, self.game_id) tensorboard_dir = os.path.join(env_dir, 'tensorboard') self.writer = SummaryWriter(tensorboard_dir) # Model configuration self.grid_size = 64 self.num_coordinates = self.grid_size * self.grid_size self.num_colours = 16 self.action_model = ActionModel(input_channels=16, grid_size=64).to(self.device) self.optimizer = optim.Adam(self.action_model.parameters(), lr=0.0001) # Experience buffer with hash-based deduplication self.experience_buffer = deque(maxlen=200000) self.experience_hashes = set() self.batch_size = 64 self.train_frequency = 5 # Action mapping self.action_list = [GameAction.ACTION1, GameAction.ACTION2, GameAction.ACTION3, GameAction.ACTION4, GameAction.ACTION5] ``` -------------------------------- ### Project File Structure Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Overview of the project directory layout. ```text ARC3/ ├── ARC-AGI-3-Agents/ # Competition framework (submodule) ├── custom_agents/ │ ├── __init__.py # Agent registration │ ├── action.py # Main action learning agent │ └── view_utils.py # Visualization utilities ├── custom_agents.py # Agent imports ├── Makefile # Build commands ├── README.md # This file ├── requirements.txt # Python dependencies └── utils.py # Shared utilities ``` -------------------------------- ### Run Action Agent Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Executes the action agent using the configured Makefile command. ```bash make action ``` -------------------------------- ### Train Action Model with Supervised Learning Source: https://context7.com/driessmit/arc3-solution/llms.txt Implements binary cross-entropy loss and entropy regularization for training. Requires an experience buffer and a configured optimizer. ```python def _train_action_model(self): """Train the action model on collected experiences.""" if len(self.experience_buffer) < self.batch_size: return # Sample batch from experience buffer batch_indices = np.random.choice(len(self.experience_buffer), self.batch_size, replace=False) batch = [self.experience_buffer[i] for i in batch_indices] # Prepare batch data states = torch.stack([torch.from_numpy(exp['state']).float().to(self.device) for exp in batch]) action_indices = torch.tensor([exp['action_idx'] for exp in batch], dtype=torch.long, device=self.device) rewards = torch.tensor([exp['reward'] for exp in batch], dtype=torch.float32, device=self.device) self.optimizer.zero_grad() # Forward pass combined_logits = self.action_model(states) # Main loss - binary cross-entropy on selected actions selected_logits = combined_logits.gather(1, action_indices.unsqueeze(1)).squeeze(1) main_loss = F.binary_cross_entropy_with_logits(selected_logits, rewards) # Entropy regularization for exploration all_probs = torch.sigmoid(combined_logits) action_probs = all_probs[:, :5] coord_probs = all_probs[:, 5:] action_entropy = action_probs.mean() coord_entropy = coord_probs.mean() # Total loss with entropy bonus total_loss = main_loss - 0.0001 * action_entropy - 0.00001 * coord_entropy # Backward pass total_loss.backward() self.optimizer.step() # Calculate accuracy accuracy = ((torch.sigmoid(selected_logits) > 0.5) == rewards).float().mean() print(f"Loss: {total_loss.item():.4f}, Accuracy: {accuracy.item():.4f}") # Training is triggered every N actions if action_counter % train_frequency == 0: _train_action_model() ``` -------------------------------- ### Additional Usage Commands Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Commonly used commands for running the agent, viewing logs, and cleaning up files. ```bash # Standard competition run make action # Run with specific game ID uv run ARC-AGI-3-Agents/main.py --agent=action --game=vc33 # View logs and metrics make tensorboard # Clean generated files make clean ``` -------------------------------- ### Define and Initialize ActionModel Source: https://context7.com/driessmit/arc3-solution/llms.txt Implements a shared convolutional backbone with separate heads for discrete action classification and coordinate prediction. Requires PyTorch and expects 16-channel input tensors. ```python import torch import torch.nn as nn import torch.nn.functional as F class ActionModel(nn.Module): """CNN that predicts which actions will result in new frames with shared conv backbone.""" def __init__(self, input_channels=16, grid_size=64): super().__init__() self.grid_size = grid_size self.num_action_types = 5 # ACTION1-ACTION5 # Shared convolutional backbone self.conv1 = nn.Conv2d(input_channels, 32, kernel_size=3, padding=1) self.conv2 = nn.Conv2d(32, 64, kernel_size=3, padding=1) self.conv3 = nn.Conv2d(64, 128, kernel_size=3, padding=1) self.conv4 = nn.Conv2d(128, 256, kernel_size=3, padding=1) # Action head - preserve spatial information self.action_pool = nn.MaxPool2d(4, 4) # Reduce to 16x16 action_flattened_size = 256 * 16 * 16 # 65,536 self.action_fc = nn.Linear(action_flattened_size, 512) self.action_head = nn.Linear(512, self.num_action_types) # Coordinate head - spatial reasoning for 64x64 action space self.coord_conv1 = nn.Conv2d(256, 128, kernel_size=3, padding=1) self.coord_conv2 = nn.Conv2d(128, 64, kernel_size=3, padding=1) self.coord_conv3 = nn.Conv2d(64, 32, kernel_size=1) self.coord_conv4 = nn.Conv2d(32, 1, kernel_size=1) self.dropout = nn.Dropout(0.2) def forward(self, x): # x shape: (batch_size, 16, 64, 64) x = F.relu(self.conv1(x)) x = F.relu(self.conv2(x)) x = F.relu(self.conv3(x)) conv_features = F.relu(self.conv4(x)) # (batch, 256, 64, 64) # Action head output action_features = self.action_pool(conv_features) action_features = action_features.view(action_features.size(0), -1) action_features = F.relu(self.action_fc(action_features)) action_features = self.dropout(action_features) action_logits = self.action_head(action_features) # (batch, 5) # Coordinate head output coord_features = F.relu(self.coord_conv1(conv_features)) coord_features = F.relu(self.coord_conv2(coord_features)) coord_features = F.relu(self.coord_conv3(coord_features)) coord_logits = self.coord_conv4(coord_features) coord_logits = coord_logits.view(coord_logits.size(0), -1) # (batch, 4096) # Combined output: [5 action logits, 4096 coordinate logits] return torch.cat([action_logits, coord_logits], dim=1) # Example usage device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = ActionModel(input_channels=16, grid_size=64).to(device) # Create sample input (batch of 4 frames, 16 color channels, 64x64 grid) sample_input = torch.randn(4, 16, 64, 64).to(device) combined_logits = model(sample_input) print(f"Output shape: {combined_logits.shape}") # Output: torch.Size([4, 4101]) ``` -------------------------------- ### Configure Agent Submodule Source: https://github.com/driessmit/arc3-solution/blob/main/README.md Updates the agent initialization and data structures to support the custom agent implementation. ```python import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from custom_agent import * ``` ```python available_actions: list[GameAction] = Field(default_factory=list) ``` -------------------------------- ### Manage Experience Buffer with Deduplication Source: https://context7.com/driessmit/arc3-solution/llms.txt Uses MD5 hashing to ensure uniqueness of frame-action pairs in the experience buffer. Requires numpy arrays for frame data. ```python import hashlib def _compute_experience_hash(self, frame: np.array, action_idx: int) -> str: """Compute hash for frame+action combination to ensure uniqueness.""" assert frame.shape == (self.num_colours, self.grid_size, self.grid_size) frame_bytes = frame.tobytes() hash_input = frame_bytes + str(action_idx).encode('utf-8') return hashlib.md5(hash_input).hexdigest() # Example: Adding experience to buffer prev_frame = current_frame.cpu().numpy().astype(bool) # Boolean for comparison prev_action_idx = 3 # ACTION4 # On next frame, compute hash and store if unique experience_hash = _compute_experience_hash(prev_frame, prev_action_idx) if experience_hash not in experience_hashes: # Check if frame changed current_frame_np = current_frame.cpu().numpy().astype(bool) frame_changed = not np.array_equal(prev_frame, current_frame_np) experience = { 'state': prev_frame, 'action_idx': prev_action_idx, 'reward': 1.0 if frame_changed else 0.0 } experience_buffer.append(experience) experience_hashes.add(experience_hash) print(f"Buffer size: {len(experience_buffer)}, Unique hashes: {len(experience_hashes)}") ``` -------------------------------- ### Implement Hierarchical Action Sampling Source: https://context7.com/driessmit/arc3-solution/llms.txt Handles action masking and coordinate selection for a combined action space. Requires a torch.Tensor of logits and an optional list of available actions. ```python def _sample_from_combined_output(self, combined_logits: torch.Tensor, available_actions: list = None) -> tuple: """Sample from combined 5 + 64x64 action space with masking for invalid actions.""" # Split logits action_logits = combined_logits[:5] # First 5 coord_logits = combined_logits[5:] # Remaining 4096 # Apply masking based on available_actions if available_actions is not None and len(available_actions) > 0: action_mask = torch.full_like(action_logits, float('-inf')) action6_available = False for action in available_actions: action_id = action.value if 1 <= action_id <= 5: # ACTION1-ACTION5 action_mask[action_id - 1] = 0.0 elif action_id == 6: action6_available = True action_logits = action_logits + action_mask if not action6_available: coord_logits = coord_logits + torch.full_like(coord_logits, float('-inf')) # Apply sigmoid for probabilities action_probs = torch.sigmoid(action_logits) coord_probs_raw = torch.sigmoid(coord_logits) # Scale coordinate probs for fair sampling coord_probs_scaled = coord_probs_raw / 4096 # Combine and normalize for sampling all_probs_sampling = torch.cat([action_probs, coord_probs_scaled]) all_probs_sampling = all_probs_sampling / all_probs_sampling.sum() all_probs_sampling_np = all_probs_sampling.cpu().numpy() # Sample action selected_idx = np.random.choice(len(all_probs_sampling_np), p=all_probs_sampling_np) if selected_idx < 5: return selected_idx, None, None, all_probs_sampling_np else: coord_idx = selected_idx - 5 y_idx = coord_idx // 64 x_idx = coord_idx % 64 return 5, (y_idx, x_idx), coord_idx, all_probs_sampling_np # Example: Sample action from model output with torch.no_grad(): combined_logits = model(sample_input[0:1]).squeeze(0) action_idx, coords, coord_idx, probs = _sample_from_combined_output(combined_logits) if action_idx < 5: print(f"Selected ACTION{action_idx + 1}") else: print(f"Selected ACTION6 at coordinates ({coords[1]}, {coords[0]})") ``` -------------------------------- ### Visualize ARC Grids Source: https://context7.com/driessmit/arc3-solution/llms.txt Converts numpy arrays into PIL images using a predefined color palette. Useful for debugging grid-based tasks. ```python from PIL import Image, ImageDraw import numpy as np # ARC color palette (16 colors) key_colors = { 0: "#FFFFFF", 1: "#CCCCCC", 2: "#999999", 3: "#666666", 4: "#333333", 5: "#000000", 6: "#E53AA3", 7: "#FF7BCC", 8: "#F93C31", 9: "#1E93FF", 10: "#88D8F1", 11: "#FFDC00", 12: "#FF851B", 13: "#921231", 14: "#4FCC30", 15: "#A356D6" } def create_grid_image(grid: np.ndarray, cell_size: int = 8, border_width: int = 1): """Create a PIL image from an ARC grid.""" if len(grid.shape) == 3 and grid.shape[0] == 1: grid = grid.squeeze(0) height, width = grid.shape img_width = width * cell_size + (width + 1) * border_width img_height = height * cell_size + (height + 1) * border_width img = Image.new('RGB', (img_width, img_height), 'white') draw = ImageDraw.Draw(img) for y in range(height): for x in range(width): color_idx = int(grid[y, x]) hex_color = key_colors[color_idx].lstrip('#') color = tuple(int(hex_color[i:i+2], 16) for i in (0, 2, 4)) left = x * (cell_size + border_width) + border_width top = y * (cell_size + border_width) + border_width draw.rectangle([left, top, left + cell_size, top + cell_size], fill=color) return img # Example: Create and save grid visualization sample_grid = np.random.randint(0, 16, size=(64, 64)) grid_img = create_grid_image(sample_grid, cell_size=8) grid_img.save("sample_grid.png") ``` -------------------------------- ### Convert ARC Frame to Tensor Source: https://context7.com/driessmit/arc3-solution/llms.txt Converts ARC frame data into a one-hot encoded PyTorch tensor. This method is crucial for feeding frame information into the action model, ensuring the correct shape and data type for the CNN. ```python def _frame_to_tensor(self, frame_data: FrameData) -> torch.Tensor: """Convert frame data to tensor format for the model.""" # Convert frame to numpy array with color indices 0-15 frame = np.array(frame_data.frame, dtype=np.int64) # Take the last frame (in case of an animation of frames) frame = frame[-1] assert frame.shape == (self.grid_size, self.grid_size) # One-hot encode: (64, 64) -> (16, 64, 64) tensor = torch.zeros(self.num_colours, self.grid_size, self.grid_size, dtype=torch.float32) tensor.scatter_(0, torch.from_numpy(frame).unsqueeze(0), 1) return tensor.to(self.device) ``` ```python # Example usage with mock frame data import numpy as np # Simulate a 64x64 ARC grid with color indices 0-15 mock_frame = np.random.randint(0, 16, size=(1, 64, 64), dtype=np.int64) # Create mock FrameData object class MockFrameData: def __init__(self, frame): self.frame = frame frame_data = MockFrameData(mock_frame) # Convert to tensor (16 channels, 64x64) tensor = _frame_to_tensor(frame_data) print(f"Tensor shape: {tensor.shape}") # Output: torch.Size([16, 64, 64]) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.