Claude's Corner: Origin - The DNA Switch Startup That Beat Google DeepMind

# How to Build a Regulatory DNA Design Tool with Claude Code

A step-by-step guide to building an AI-powered system for designing and predicting regulatory DNA sequences, inspired by Origin's Axis model architecture.

## Step 1: Set Up the Database Schema

Design your data model to store regulatory DNA sequences, their cell-type predictions, and experimental validation results.

```sql
CREATE TABLE regulatory_sequences (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  sequence TEXT NOT NULL,
  length INTEGER GENERATED ALWAYS AS (char_length(sequence)) STORED,
  gc_content FLOAT,
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE cell_types (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name TEXT NOT NULL UNIQUE,
  tissue_type TEXT,
  disease_context TEXT,
  encode_cell_type_id TEXT
);

CREATE TABLE sequence_predictions (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  sequence_id UUID REFERENCES regulatory_sequences(id),
  cell_type_id UUID REFERENCES cell_types(id),
  predicted_expression FLOAT,
  predicted_accessibility FLOAT,
  tf_motif_enrichment JSONB,
  model_version TEXT,
  confidence_score FLOAT,
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE experimental_validations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  sequence_id UUID REFERENCES regulatory_sequences(id),
  cell_type_id UUID REFERENCES cell_types(id),
  measured_expression FLOAT,
  assay_type TEXT, -- 'MPRA', 'STARR-seq', 'luciferase'
  replicate_count INTEGER,
  validated_at TIMESTAMPTZ DEFAULT now(),
  validated_by TEXT
);

CREATE INDEX ON sequence_predictions(sequence_id);
CREATE INDEX ON experimental_validations(sequence_id);
```

## Step 2: Build the Sequence Encoder

The core of the model takes DNA sequences and converts them to token embeddings that a Transformer can process.

```python
import torch
import torch.nn as nn

NUCLEOTIDES = {'A': 0, 'C': 1, 'G': 2, 'T': 3, 'N': 4, '[MASK]': 5, '[PAD]': 6}

class DNATokenizer:
    def __init__(self, max_length=512):
        self.max_length = max_length
        self.vocab = NUCLEOTIDES
    
    def encode(self, sequence: str) -> torch.Tensor:
        tokens = [self.vocab.get(nt.upper(), self.vocab['N']) for nt in sequence]
        if len(tokens) < self.max_length:
            tokens += [self.vocab['[PAD]']] * (self.max_length - len(tokens))
        return torch.tensor(tokens[:self.max_length], dtype=torch.long)
    
    def mask_random(self, tokens: torch.Tensor, mask_prob=0.15) -> tuple:
        mask = torch.rand(tokens.shape) < mask_prob
        masked = tokens.clone()
        masked[mask] = self.vocab['[MASK]']
        return masked, mask

class DNAEmbedding(nn.Module):
    def __init__(self, vocab_size=7, embed_dim=256, max_len=512):
        super().__init__()
        self.token_embed = nn.Embedding(vocab_size, embed_dim, padding_idx=NUCLEOTIDES['[PAD]'])
        self.position_embed = nn.Embedding(max_len, embed_dim)
    
    def forward(self, tokens):
        positions = torch.arange(tokens.shape[1], device=tokens.device).unsqueeze(0)
        return self.token_embed(tokens) + self.position_embed(positions)
```

## Step 3: Implement the Shared Transformer Backbone

Build the multitask architecture with a shared backbone and separate prediction/generation heads.

```python
class SharedTransformerBackbone(nn.Module):
    def __init__(self, embed_dim=256, num_heads=8, num_layers=6, dropout=0.1):
        super().__init__()
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embed_dim,
            nhead=num_heads,
            dim_feedforward=embed_dim * 4,
            dropout=dropout,
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
    
    def forward(self, embeddings, padding_mask=None):
        return self.transformer(embeddings, src_key_padding_mask=padding_mask)

class NucleotideGenerationHead(nn.Module):
    def __init__(self, embed_dim=256, vocab_size=5):
        super().__init__()
        self.proj = nn.Linear(embed_dim, vocab_size)
    
    def forward(self, hidden_states):
        return self.proj(hidden_states)  # [batch, seq_len, 4 nucleotides + N]

class ExpressionPredictionHead(nn.Module):
    def __init__(self, embed_dim=256, num_cell_types=64):
        super().__init__()
        self.cell_type_embed = nn.Embedding(num_cell_types, embed_dim)
        self.pooling = nn.AdaptiveAvgPool1d(1)
        self.predictor = nn.Sequential(
            nn.Linear(embed_dim * 2, embed_dim),
            nn.ReLU(),
            nn.Linear(embed_dim, 1)
        )
    
    def forward(self, hidden_states, cell_type_ids):
        pooled = hidden_states.mean(dim=1)
        cell_embed = self.cell_type_embed(cell_type_ids)
        combined = torch.cat([pooled, cell_embed], dim=-1)
        return self.predictor(combined).squeeze(-1)

class AxisModel(nn.Module):
    def __init__(self, embed_dim=256, num_heads=8, num_layers=6, num_cell_types=64):
        super().__init__()
        self.embedder = DNAEmbedding(embed_dim=embed_dim)
        self.backbone = SharedTransformerBackbone(embed_dim, num_heads, num_layers)
        self.gen_head = NucleotideGenerationHead(embed_dim)
        self.pred_head = ExpressionPredictionHead(embed_dim, num_cell_types)
    
    def forward(self, tokens, cell_type_ids=None, task='both'):
        embeddings = self.embedder(tokens)
        hidden = self.backbone(embeddings)
        
        outputs = {}
        if task in ('generation', 'both'):
            outputs['logits'] = self.gen_head(hidden)
        if task in ('prediction', 'both') and cell_type_ids is not None:
            outputs['expression'] = self.pred_head(hidden, cell_type_ids)
        return outputs
```

## Step 4: Implement Confidence-Based Iterative Generation

The key innovation: generate sequences position-by-position based on model confidence, not randomly.

```python
import torch.nn.functional as F

def generate_regulatory_sequence(
    model: AxisModel,
    tokenizer: DNATokenizer,
    target_length: int,
    cell_type_id: int,
    temperature: float = 0.8,
    num_steps: int = None
) -> str:
    model.eval()
    device = next(model.parameters()).device
    
    if num_steps is None:
        num_steps = target_length
    
    # Start fully masked
    tokens = torch.full((1, target_length), NUCLEOTIDES['[MASK]'], dtype=torch.long, device=device)
    
    with torch.no_grad():
        for step in range(num_steps):
            outputs = model(tokens, task='generation')
            logits = outputs['logits']  # [1, seq_len, vocab_size]
            
            # Compute per-position entropy (uncertainty)
            probs = F.softmax(logits[:, :, :4] / temperature, dim=-1)
            entropy = -(probs * probs.log()).sum(dim=-1)  # [1, seq_len]
            
            # Only consider still-masked positions
            masked_positions = (tokens[0] == NUCLEOTIDES['[MASK]']).nonzero(as_tuple=True)[0]
            if len(masked_positions) == 0:
                break
            
            # Pick the LOWEST entropy masked position (most confident)
            masked_entropy = entropy[0, masked_positions]
            most_confident_idx = masked_positions[masked_entropy.argmin()]
            
            # Sample from that position's distribution
            position_probs = probs[0, most_confident_idx]
            chosen_nucleotide = torch.multinomial(position_probs, 1).item()
            tokens[0, most_confident_idx] = chosen_nucleotide
    
    # Decode
    idx_to_nt = {v: k for k, v in NUCLEOTIDES.items() if k not in ('[MASK]', '[PAD]')}
    return ''.join(idx_to_nt.get(t.item(), 'N') for t in tokens[0])
```

## Step 5: Build the Training Pipeline

Train on ENCODE data with a combined loss that handles both generation and expression prediction tasks.

```python
from torch.optim import AdamW
from torch.optim.lr_scheduler import CosineAnnealingLR

def combined_loss(gen_logits, gen_targets, gen_mask, pred_expression, true_expression):
    gen_loss = 0.0
    if gen_mask.any():
        gen_loss = F.cross_entropy(
            gen_logits[gen_mask].view(-1, gen_logits.shape[-1]),
            gen_targets[gen_mask].view(-1)
        )
    
    # Factorized Poisson + multinomial loss for expression prediction
    pred_loss = 0.0
    if true_expression is not None:
        log_pred = pred_expression
        pred_loss = F.poisson_nll_loss(log_pred, true_expression, log_input=True)
    
    return gen_loss + 0.5 * pred_loss

def train_epoch(model, dataloader, optimizer, device):
    model.train()
    tokenizer = DNATokenizer()
    total_loss = 0.0
    
    for batch in dataloader:
        sequences = batch['sequences'].to(device)
        cell_types = batch['cell_types'].to(device)
        expression = batch['expression'].to(device)
        
        masked_seqs, masks = tokenizer.mask_random(sequences)
        
        outputs = model(masked_seqs, cell_type_ids=cell_types)
        loss = combined_loss(
            outputs['logits'], sequences, masks,
            outputs['expression'], expression
        )
        
        optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()
        total_loss += loss.item()
    
    return total_loss / len(dataloader)
```

## Step 6: Design the API Layer

Expose generation and prediction endpoints for pharma customers to integrate with their workflows.

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uuid

app = FastAPI(title="Regulatory DNA Design API")

class GenerateRequest(BaseModel):
    target_length: int
    cell_type: str
    temperature: float = 0.8
    n_candidates: int = 10

class PredictRequest(BaseModel):
    sequences: list[str]
    cell_types: list[str]

@app.post("/generate")
async def generate_sequences(req: GenerateRequest):
    cell_type_id = get_cell_type_id(req.cell_type)
    if cell_type_id is None:
        raise HTTPException(404, f"Cell type '{req.cell_type}' not found")
    
    candidates = []
    for _ in range(req.n_candidates):
        seq = generate_regulatory_sequence(
            model, tokenizer,
            target_length=req.target_length,
            cell_type_id=cell_type_id,
            temperature=req.temperature
        )
        gc = (seq.count('G') + seq.count('C')) / len(seq)
        candidates.append({
            "id": str(uuid.uuid4()),
            "sequence": seq,
            "gc_content": gc,
            "length": len(seq)
        })
    
    return {"candidates": candidates, "cell_type": req.cell_type}

@app.post("/predict")
async def predict_expression(req: PredictRequest):
    results = []
    for seq, ct in zip(req.sequences, req.cell_types):
        cell_type_id = get_cell_type_id(ct)
        tokens = tokenizer.encode(seq).unsqueeze(0)
        ct_tensor = torch.tensor([cell_type_id])
        
        with torch.no_grad():
            outputs = model(tokens, cell_type_ids=ct_tensor, task='prediction')
        
        results.append({
            "sequence": seq[:20] + "...",
            "cell_type": ct,
            "predicted_expression": outputs['expression'].item()
        })
    
    return {"predictions": results}
```

## Step 7: Deploy with GPU Inference and Wet Lab Integration

Set up production deployment with GPU inference and workflow integration for experimental validation.

```yaml
# docker-compose.yml
services:
  api:
    image: regulatory-dna-api:latest
    environment:
      - MODEL_PATH=/models/axis_v1.pt
      - DATABASE_URL=${DATABASE_URL}
      - GPU_DEVICE=cuda:0
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    ports:
      - "8000:8000"
  
  worker:
    image: regulatory-dna-worker:latest
    environment:
      - REDIS_URL=${REDIS_URL}
      - DATABASE_URL=${DATABASE_URL}
    command: python worker.py --queue validation_pipeline
```

```python
# worker.py - handles experimental validation ingestion
import redis
import json
from db import upsert_validation

def process_validation_result(job_data: dict):
    """Ingest wet lab validation results and update training dataset."""
    validation = {
        "sequence_id": job_data["sequence_id"],
        "cell_type_id": job_data["cell_type_id"],
        "measured_expression": job_data["expression_value"],
        "assay_type": job_data["assay"],
        "replicate_count": job_data["n_replicates"]
    }
    upsert_validation(validation)
    
    # Flag for next training run if enough new data accumulated
    check_retrain_threshold()

def check_retrain_threshold(min_new_sequences=500):
    new_count = count_unincorporated_validations()
    if new_count >= min_new_sequences:
        trigger_training_job()
```

**Key architectural decisions:**
- Store all generated sequences in Postgres immediately, even before validation - the sequence space exploration data has value
- Use async job queues (Redis + Celery) for generation requests that take >2 seconds
- Pin model weights in GPU VRAM between requests to eliminate cold-start latency
- Build a validation ingestion pipeline early - the experimental data is the moat, and you need a clean system for getting it into the training loop fast
- Track which sequences were generated for which customers, so you can weight customer-relevant cell types higher in future training runs