«`html
Build an Autonomous Wet-Lab Protocol Planner and Validator Using Salesforce CodeGen for Agentic Experiment Design and Safety Optimization
In this tutorial, we build a Wet-Lab Protocol Planner & Validator that acts as an intelligent agent for experimental design and execution. We design the system using Python and integrate Salesforce’s CodeGen-350M-mono model for natural language reasoning. The pipeline is structured into modular components:
- ProtocolParser: Extracts structured data, such as steps, durations, and temperatures, from textual protocols.
- InventoryManager: Validates reagent availability and expiry.
- Schedule Planner: Generates timelines and parallelization.
- Safety Validator: Identifies biosafety or chemical hazards.
The LLM is then used to generate optimization suggestions, effectively closing the loop between perception, planning, validation, and refinement.
Technical Implementation
We begin by importing essential libraries and loading the Salesforce CodeGen-350M-mono model locally for lightweight, API-free inference. The model is initialized with float16 precision and automatic device mapping to ensure compatibility and speed on Colab GPUs.
import re, json, pandas as pd
from datetime import datetime, timedelta
from collections import defaultdict
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
MODEL_NAME = "Salesforce/codegen-350M-mono"
print("Loading CodeGen model (30 seconds)...")
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
tokenizer.pad_token = tokenizer.eos_token
model = AutoModelForCausalLM.from_pretrained(
MODEL_NAME, torch_dtype=torch.float16, device_map="auto"
)
print("✓ Model loaded!")
Next, we define the ProtocolParser and InventoryManager classes to extract structured experimental details and verify reagent inventory. The ProtocolParser parses each protocol step for duration, temperature, and safety markers, while the InventoryManager validates stock levels, expiry dates, and reagent availability through fuzzy matching.
class ProtocolParser:
def read_protocol(self, text):
steps = []
lines = text.split('\n')
for i, line in enumerate(lines, 1):
step_match = re.search(r'^(\d+)\.\s+(.+)', line.strip())
if step_match:
num, name = step_match.groups()
context = '\n'.join(lines[i:min(i+4, len(lines))])
duration = self._extract_duration(context)
temp = self._extract_temp(context)
safety = self._check_safety(context)
steps.append({
'step': int(num), 'name': name, 'duration_min': duration,
'temp': temp, 'safety': safety, 'line': i, 'details': context[:200]
})
return steps
Scheduling and Safety Validation
We implement the SchedulePlanner and SafetyValidator to design efficient experiment timelines and enforce lab safety standards. The SchedulePlanner dynamically generates daily schedules, identifies parallelizable steps, and validates potential risks, such as unsafe pH levels, hazardous chemicals, or biosafety-level requirements.
class SchedulePlanner:
def make_schedule(self, steps, start_time="09:00"):
schedule = []
current = datetime.strptime(f"2025-01-01 {start_time}", "%Y-%m-%d %H:%M")
day = 1
for step in steps:
end = current + timedelta(minutes=step['duration_min'])
if step['duration_min'] > 480:
day += 1
current = datetime.strptime(f"2025-01-0{day} 09:00", "%Y-%m-%d %H:%M")
end = current
schedule.append({
'step': step['step'], 'name': step['name'][:40],
'start': current.strftime("%H:%M"), 'end': end.strftime("%H:%M"),
'duration': step['duration_min'], 'temp': step['temp'],
'day': day, 'can_parallelize': step['duration_min'] > 60,
'safety': ', '.join(step['safety']) if step['safety'] else 'None'
})
if step['duration_min'] <= 480:
current = end
return schedule
Agent Loop Integration
We construct the agent loop, integrating perception, planning, validation, and revision into a single, coherent flow. The CodeGen model is used for reasoning-based optimization to refine step sequencing and propose practical improvements for efficiency and parallel execution.
def agent_loop(protocol_text, inventory_csv, start_time="09:00"):
print("\n AGENT STARTING PROTOCOL ANALYSIS...\n")
parser = ProtocolParser()
steps = parser.read_protocol(protocol_text)
print(f" Parsed {len(steps)} protocol steps")
inventory = InventoryManager(inventory_csv)
reagents = inventory.extract_reagents(protocol_text)
print(f" Identified {len(reagents)} reagents: {', '.join(reagents[:5])}...")
inv_issues = inventory.check_availability(reagents)
validator = SafetyValidator()
safety_risks = validator.validate(steps)
planner = SchedulePlanner()
schedule = planner.make_schedule(steps, start_time)
parallel_opts, time_saved = planner.optimize_parallelization(schedule)
total_time = sum(s['duration'] for s in schedule)
optimized_time = total_time - time_saved
opt_prompt = f"Protocol has {len(steps)} steps, {total_time} min total. Key bottleneck optimization:"
optimization = llm_call(opt_prompt, max_tokens=80)
return {
'steps': steps, 'schedule': schedule, 'inventory_issues': inv_issues,
'safety_risks': safety_risks, 'parallelization': parallel_opts,
'time_saved': time_saved, 'total_time': total_time,
'optimized_time': optimized_time, 'ai_optimization': optimization,
'reagents': reagents
}
Output Generation
We create output generators that transform results into human-readable checklists and Gantt-compatible CSVs. Every execution produces clear summaries of reagents, time savings, and safety or inventory alerts for streamlined lab operations.
def generate_checklist(results):
md = "# WET-LAB PROTOCOL CHECKLIST\n\n"
md += f"**Total Steps:** {len(results['schedule'])}\n"
md += f"**Estimated Time:** {results['total_time']} min ({results['total_time']//60}h {results['total_time']%60}m)\n"
md += f"**Optimized Time:** {results['optimized_time']} min (save {results['time_saved']} min)\n\n"
md += "## TIMELINE\n"
current_day = 1
for item in results['schedule']:
if item['day'] > current_day:
md += f"\n### Day {item['day']}\n"
current_day = item['day']
parallel = " " if item['can_parallelize'] else ""
md += f"- [ ] **{item['start']}-{item['end']}** | Step {item['step']}: {item['name']} ({item['temp']}){parallel}\n"
md += "\n## REAGENT PICK-LIST\n"
for reagent in results['reagents']:
md += f"- [ ] {reagent}\n"
md += "\n## SAFETY & INVENTORY ALERTS\n"
all_issues = results['safety_risks'] + results['inventory_issues']
if all_issues:
for risk in all_issues:
md += f"- {risk}\n"
else:
md += "- No critical issues detected\n"
md += "\n## OPTIMIZATION TIPS\n"
for tip in results['parallelization']:
md += f"- {tip}\n"
md += f"- AI Suggestion: {results['ai_optimization']}\n"
return md
Conclusion
We demonstrated how agentic AI principles can enhance reproducibility and safety in wet-lab workflows. By parsing free-form experimental text into structured, actionable plans, we automated protocol validation, reagent management, and temporal optimization in a single pipeline. The integration of CodeGen enables on-device reasoning about bottlenecks and safety conditions, allowing for self-contained, data-secure operations.
Check out the FULL CODES here. Feel free to check out our GitHub Page for Tutorials, Codes, and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! Are you on Telegram? Now you can join us on Telegram as well.
```