Deep Dive into DAMO ElementsClaw: How an AI Agent Discovered 4 New Superconductors in 28 GPU Hours

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch_geometric.nn import MessagePassing, global_mean_pool
from torch_geometric.data import Data
from typing import Optional, List


class AtomicMessagePassing(MessagePassing):
    """
    Geometry-aware message passing layer.
    Incorporates inter-atomic distances, bond angles, and periodic
    boundary conditions for 3D structural understanding.
    """
    def __init__(self, in_channels: int, out_channels: int, num_filters: int = 128):
        super().__init__(aggr='add')
        self.in_channels = in_channels
        self.out_channels = out_channels
        
        # Edge feature network: encodes geometry (distance, angle, periodicity)
        self.edge_mlp = nn.Sequential(
            nn.Linear(3, num_filters),
            nn.SiLU(),
            nn.Linear(num_filters, out_channels),
            nn.SiLU(),
        )
        
        # Node feature update network
        self.node_mlp = nn.Sequential(
            nn.Linear(in_channels + out_channels, out_channels * 2),
            nn.SiLU(),
            nn.Linear(out_channels * 2, out_channels),
        )
        
        self.residual = nn.Linear(in_channels, out_channels) if in_channels != out_channels else nn.Identity()
    
    def forward(self, x: torch.Tensor, edge_index: torch.Tensor, 
                edge_attr: torch.Tensor) -> torch.Tensor:
        return self.propagate(edge_index, x=x, edge_attr=edge_attr)
    
    def message(self, x_j: torch.Tensor, edge_attr: torch.Tensor) -> torch.Tensor:
        edge_features = self.edge_mlp(edge_attr)
        return edge_features * x_j  # Gating: geometry-modulated messages
    
    def update(self, aggr_out: torch.Tensor, x: torch.Tensor) -> torch.Tensor:
        combined = torch.cat([x, aggr_out], dim=-1)
        out = self.node_mlp(combined)
        return out + self.residual(x)


class ElementsFoundationModel(nn.Module):
    """
    1B-parameter atomic foundation model.
    Architecture: Embedding → 24-layer Geometric Message Passing → Readout
    Supports: superconductivity classification, Tc regression,
              energy prediction, and structure generation.
    """
    def __init__(
        self,
        num_atom_types: int = 118,
        hidden_dim: int = 1024,
        num_layers: int = 24,
        dropout: float = 0.1,
    ):
        super().__init__()
        self.hidden_dim = hidden_dim
        
        # Atom type embedding (118 elements + special token)
        self.atom_embedding = nn.Embedding(num_atom_types + 1, hidden_dim)
        
        # Position encoding: fractional coordinates within unit cell
        self.pos_mlp = nn.Sequential(
            nn.Linear(3, hidden_dim // 4),
            nn.SiLU(),
            nn.Linear(hidden_dim // 4, hidden_dim),
        )
        
        # Multi-layer geometric message passing
        self.layers = nn.ModuleList([
            AtomicMessagePassing(hidden_dim, hidden_dim)
            for _ in range(num_layers)
        ])
        
        self.layer_norms = nn.ModuleList([
            nn.LayerNorm(hidden_dim) for _ in range(num_layers)
        ])
        self.dropout = nn.Dropout(dropout)
        
        # Global readout
        self.readout = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim * 2),
            nn.SiLU(),
            nn.Linear(hidden_dim * 2, hidden_dim),
        )
        
        self.total_params = sum(p.numel() for p in self.parameters())
        print(f"Elements Foundation Model Parameters: {self.total_params:,}")
    
    def forward(self, data: Data) -> torch.Tensor:
        # Atom embedding + position encoding
        atom_feat = self.atom_embedding(data.x)
        pos_feat = self.pos_mlp(data.pos)
        h = atom_feat + pos_feat
        
        # Message passing with residual connections and layer norm
        for i in range(self.num_layers):
            h_new = self.layers[i](h, data.edge_index, data.edge_attr)
            h_new = self.layer_norms[i](h_new)
            h = h + self.dropout(h_new)
        
        # Global pooling: atom-level → crystal-level
        h_graph = global_mean_pool(h, data.batch)
        return self.readout(h_graph)


class ElementsMultiHead(nn.Module):
    """Four-branch output head for Elements model."""
    def __init__(self, hidden_dim: int = 1024):
        super().__init__()
        
        # Head 1: Elements-C (superconductivity classification)
        self.classifier = nn.Sequential(
            nn.Linear(hidden_dim, 512), nn.SiLU(), nn.Dropout(0.1),
            nn.Linear(512, 128), nn.SiLU(),
            nn.Linear(128, 1), nn.Sigmoid(),
        )
        
        # Head 2: Elements-T (critical temperature regression)
        self.tc_regressor = nn.Sequential(
            nn.Linear(hidden_dim, 512), nn.SiLU(), nn.Dropout(0.1),
            nn.Linear(512, 256), nn.SiLU(),
            nn.Linear(256, 1),
        )
        
        # Head 3: Elements-E (energy/stability prediction)
        self.energy_predictor = nn.Sequential(
            nn.Linear(hidden_dim, 512), nn.SiLU(),
            nn.Linear(512, 256), nn.SiLU(),
            nn.Linear(256, 1),
        )
        
        # Head 4: Elements-G (structure generation via diffusion head)
        self.structure_generator = nn.Sequential(
            nn.Linear(hidden_dim, 1024), nn.SiLU(),
            nn.Linear(1024, 2048), nn.SiLU(),
            nn.Linear(2048, 1024),
        )
    
    def forward(self, h: torch.Tensor) -> dict:
        return {
            'superconducting_prob': self.classifier(h),
            'critical_temp': self.tc_regressor(h),
            'energy_stability': self.energy_predictor(h),
            'structure_latent': self.structure_generator(h),
        }


class ElementsModel(nn.Module):
    """Complete 1B-parameter model: foundation + multi-head outputs."""
    def __init__(self, hidden_dim: int = 1024, num_layers: int = 24):
        super().__init__()
        self.foundation = ElementsFoundationModel(hidden_dim=hidden_dim, num_layers=num_layers)
        self.heads = ElementsMultiHead(hidden_dim=hidden_dim)
        
    def forward(self, data: Data) -> dict:
        h = self.foundation(data)
        return self.heads(h)


def validate_scaling_law():
    """Scaling Law verification: first confirmed on non-LLM architecture."""
    configs = [
        {'hidden_dim': 256, 'num_layers': 8},
        {'hidden_dim': 512, 'num_layers': 16},
        {'hidden_dim': 768, 'num_layers': 20},
        {'hidden_dim': 1024, 'num_layers': 24},
    ]
    
    for cfg in configs:
        model = ElementsModel(**cfg)
        params = sum(p.numel() for p in model.parameters())
        print(f"Config {cfg}: {params:,} params")


if __name__ == "__main__":
    model = ElementsModel(hidden_dim=1024, num_layers=24)
    total_params = sum(p.numel() for p in model.parameters())
    print(f"\nElements 1B Model: {total_params:,} total parameters")
    
    print("\nScaling Law Verification:")
    validate_scaling_law()

from dataclasses import dataclass
from typing import List, Optional
import torch


@dataclass
class CandidateMaterial:
    formula: str
    space_group: str
    predicted_tc: float
    superconducting_prob: float
    energy_above_hull: float
    source: str
    literature_matched: bool = False
    synthesis_score: float = 0.0


class ElementsClawAgent:
    """
    Core agent class integrating specialized models, literature retrieval,
    database queries, and synthesis evaluation into an automated pipeline.
    """
    def __init__(self, elements_model, llm_backend):
        self.model = elements_model
        self.llm = llm_backend
        self.candidates: List[CandidateMaterial] = []
        self.verified_materials: List[CandidateMaterial] = []
        
    def screen_crystals(self, crystal_data) -> List[CandidateMaterial]:
        """Phase 1: Large-scale screening of 2.4M crystals."""
        print(f"[Phase 1] Screening {len(crystal_data)} crystals...")
        candidates = []
        
        with torch.no_grad():
            predictions = self.model(crystal_data)
        
        for prob, tc, energy in zip(
            predictions['superconducting_prob'],
            predictions['critical_temp'],
            predictions['energy_stability']
        ):
            if (prob.item() > 0.8 and tc.item() > 1.0 and energy.item() < 50.0):
                candidates.append(CandidateMaterial(
                    formula="dummy",
                    space_group="P6/mmm",
                    predicted_tc=round(tc.item(), 2),
                    superconducting_prob=round(prob.item(), 4),
                    energy_above_hull=round(energy.item(), 2),
                    source="screening",
                ))
        
        candidates.sort(key=lambda x: x.superconducting_prob, reverse=True)
        self.candidates = candidates[:68000]
        return self.candidates
    
    def cross_validate(self, candidates: List[CandidateMaterial]) -> List[CandidateMaterial]:
        """Phase 2: Cross-validate against literature and databases."""
        validated = []
        for mat in candidates:
            if not self._is_known(mat.formula):
                mat.synthesis_score = self._evaluate_synthesis(mat)
                validated.append(mat)
        return validated
    
    def _is_known(self, formula: str) -> bool:
        """Check if material exists in SuperCon or other databases."""
        known = {"YBa2Cu3O7", "MgB2", "Nb3Sn", "FeSe"}
        return formula in known
    
    def _evaluate_synthesis(self, mat: CandidateMaterial) -> float:
        """Evaluate synthesis feasibility based on element abundance and structure."""
        score = 0.5
        abundant = {'Hf', 'Zr', 'Re', 'Sc', 'Ti', 'V', 'Nb'}
        elements = set(filter(str.isalpha, mat.formula))
        
        for elem in elements:
            if elem in abundant:
                score += 0.1
        if mat.energy_above_hull < 10:
            score += 0.2
        return min(max(score, 0.0), 1.0)
    
    def self_evolve(self, new_data: List[dict]):
        """Self-evolution: fine-tune model on new literature data."""
        print(f"[Self-Evolution] Ingesting {len(new_data)} new data points...")
        # In production: trigger incremental training
        print("[Self-Evolution] Model updated. Ready for next cycle.")


def simulate_discovery_pathways():
    """Simulate the 4 discovery pathways of ElementsClaw."""
    pathways = {
        "Hf21Re25": {
            "method": "Missed Catch (漏网之鱼)",
            "tc": 2.5,
            "insight": "Existing in database but never synthesized—AI spotted the overlooked gem",
        },
        "Zr4VRe7": {
            "method": "Structural Correction (沉冤得雪)",
            "tc": 3.5,
            "insight": "Database had wrong space group—AI predicted correct structure",
        },
        "HfZrRe4": {
            "method": "De Novo Design (无中生有)",
            "tc": 5.9,
            "insight": "AI generated entirely new crystal structure from scratch",
        },
        "Zr3ScRe8": {
            "method": "Analogical Reasoning (举一反三)",
            "tc": 6.5,
            "insight": "AI generalized structural motif: P6/mmm Re-rich framework",
        },
    }
    
    for formula, info in pathways.items():
        print(f"■ {formula} | Tc={info['tc']}K | Method: {info['method']}")
        print(f"  Key Insight: {info['insight']}\n")


if __name__ == "__main__":
    simulate_discovery_pathways()

def analyze_motif(formula: str, tc: float):
    """Validate the P6/mmm Re-rich hexagonal framework structural motif."""
    print(f"■ {formula} (Tc={tc}K)")
    
    checks = {
        "P6/mmmm symmetry": True,
        "Re sublattice intact": "Re" in formula,
        "Transition metal mixing": True,
    }
    
    for check, present in checks.items():
        print(f"  {check}: {'✓' if present else '✗'}")
    
    if all(checks.values()):
        print(f"  → Motif validated for {formula}\n")
    return all(checks.values())


materials = [
    ("Hf21Re25", 2.5), ("Zr4VRe7", 3.5),
    ("HfZrRe4", 5.9), ("Zr3ScRe8", 6.5),
]
for formula, tc in materials:
    analyze_motif(formula, tc)