lamm-mit Campaign Manager
Goal-oriented protein design campaign planning. Converts abstract objectives into concrete computational pipelines with cost/time estimates, health monitoring, and adaptive decision-making.
Campaign Types
| Goal | Pipeline | Expected Timeline |
|---|---|---|
| Proof-of-concept binder | BoltzGen → Chai → ipSAE | 1–2 weeks compute |
| High-throughput screen | RFdiffusion → MPNN → ESMFold → AF2 | 2–4 weeks compute |
| High-quality therapeutic | BindCraft → ipSAE → affinity maturation | 4–8 weeks compute |
| Enzyme design | RFdiffusion + LigandMPNN → Boltz | 3–6 weeks compute |
| Quick exploration | BoltzGen 50 designs → Chai | 3–5 days compute |
Campaign Planning
from dataclasses import dataclass, field
from typing import List, Optional
import json
@dataclass
class DesignCampaign:
name: str
target_pdb: str
goal: str # "binder", "enzyme", "scaffold"
hotspot_residues: List[str]
binder_length_range: tuple = (60, 100)
# Pipeline config
n_backbone_designs: int = 200
n_seqs_per_backbone: int = 8
structure_prediction_tool: str = "colabfold" # or "chai", "boltz"
sequence_design_tool: str = "proteinmpnn" # or "ligandmpnn", "solublempnn"
# Quality thresholds
iptm_threshold: float = 0.6
plddt_threshold: float = 75.0
ipsae_top_n: int = 20
# Budget
gpu_hours_budget: Optional[float] = None
designs_to_order: int = 10
def estimate_pipeline(self) -> dict:
"""Estimate compute requirements and expected yields."""
total_seqs = self.n_backbone_designs * self.n_seqs_per_backbone
# Empirical pass rates
esm_pass_rate = 0.65
af2_pass_rate = 0.25
ipsae_final = self.ipsae_top_n
esm_survivors = int(total_seqs * esm_pass_rate)
af2_survivors = int(esm_survivors * af2_pass_rate)
final_candidates = min(af2_survivors, ipsae_final)
# GPU time estimates (A100)
rfdiffusion_hours = self.n_backbone_designs * 0.1 / 60
mpnn_hours = 0.1 # fast
esm_hours = esm_survivors * 0.01 / 60
af2_hours = esm_survivors * 2.5 / 60 # ~2.5 min per pair
return {
"total_sequences_designed": total_seqs,
"esm_survivors": esm_survivors,
"af2_survivors": af2_survivors,
"expected_final_candidates": final_candidates,
"gpu_hours": {
"rfdiffusion": round(rfdiffusion_hours, 1),
"mpnn": round(mpnn_hours, 1),
"esmfold": round(esm_hours, 1),
"af2_validation": round(af2_hours, 1),
"total": round(rfdiffusion_hours + esm_hours + af2_hours, 1),
},
"wall_time_days": round(
(rfdiffusion_hours + esm_hours + af2_hours) / 24, 1
),
"meets_budget": (
self.gpu_hours_budget is None or
(rfdiffusion_hours + af2_hours) <= self.gpu_hours_budget
)
}
Campaign Health Assessment
def assess_campaign_health(campaign_dir: str, stage: str) -> dict:
"""
Check campaign progress and flag issues.
stage: "post_rfdiffusion" | "post_mpnn" | "post_esm" | "post_af2"
"""
import os, glob, json
import numpy as np
health = {"stage": stage, "issues": [], "recommendations": []}
if stage == "post_af2":
json_files = glob.glob(f"{campaign_dir}/af2/**/*.json", recursive=True)
iptm_vals = []
plddt_vals = []
for f in json_files:
with open(f) as fp:
r = json.load(fp)
iptm_vals.append(r.get("iptm", 0))
plddt_vals.append(np.array(r.get("plddt", [0])).mean())
if not iptm_vals:
health["issues"].append("No AF2 results found")
return health
pass_rate = sum(1 for x in iptm_vals if x > 0.6) / len(iptm_vals)
health["metrics"] = {
"n_predictions": len(iptm_vals),
"mean_iptm": np.mean(iptm_vals),
"pass_rate_iptm06": pass_rate,
"mean_plddt": np.mean(plddt_vals),
}
if pass_rate < 0.05:
health["issues"].append("Very low pass rate (<5%) — redesign needed")
health["recommendations"].extend([
"Check hotspot accessibility in target structure",
"Try BoltzGen instead of RFdiffusion",
"Reduce binder length constraints",
"Verify target PDB quality",
])
elif pass_rate < 0.15:
health["issues"].append("Low pass rate (<15%) — consider adjustments")
health["recommendations"].append("Try partial diffusion for diversity")
return health
Adaptive Pipeline Decisions
def decide_next_step(health: dict) -> str:
"""
Given campaign health assessment, recommend next action.
"""
issues = health.get("issues", [])
metrics = health.get("metrics", {})
pass_rate = metrics.get("pass_rate_iptm06", 0)
if pass_rate == 0:
return "ABORT: Zero passing designs. Review target preparation and hotspot selection."
elif pass_rate < 0.05:
return (
"REDESIGN: Switch to BindCraft for built-in AF2 validation during design, "
"or verify hotspot residues are solvent-exposed."
)
elif pass_rate < 0.10:
return (
"EXPAND: Generate 2x more backbones with partial diffusion from best current designs. "
"Also try SolubleMPNN for better expression."
)
elif metrics.get("mean_iptm", 0) > 0.75:
return (
"ADVANCE: Strong campaign. Run ipSAE ranking on top designs, "
"run biophysical QC, and prepare ordering list."
)
else:
return (
"CONTINUE: Moderate progress. Apply ipSAE ranking, "
"consider affinity maturation on top 5 designs via partial diffusion."
)
Cost Estimation
def estimate_costs(campaign: DesignCampaign,
gpu_cost_per_hour: float = 2.50, # USD, A100 on-demand
gene_synthesis_per_seq: float = 150.0,
expression_per_sample: float = 80.0) -> dict:
estimates = campaign.estimate_pipeline()
gpu_hours = estimates["gpu_hours"]["total"]
return {
"compute_cost_usd": round(gpu_hours * gpu_cost_per_hour, 0),
"gene_synthesis_usd": round(campaign.designs_to_order * gene_synthesis_per_seq, 0),
"expression_screening_usd": round(campaign.designs_to_order * expression_per_sample, 0),
"total_estimated_usd": round(
gpu_hours * gpu_cost_per_hour +
campaign.designs_to_order * (gene_synthesis_per_seq + expression_per_sample),
0
),
}
Affinity Maturation (Second Round)
After identifying binders, improve affinity via computational iteration:
# Partial diffusion on best designs
python3 RFdiffusion/scripts/run_inference.py \
inference.input_pdb=best_binder.pdb \
'contigmap.contigs=[A1-999/0 B1-80]' \
diffuser.partial_T=20 \
inference.num_designs=50 \
inference.output_prefix=matured/binder
# Then: ProteinMPNN → AF2 → ipSAE (same pipeline)