# Scientific Agents

Agents for scientific discovery: hypothesis generation, experimental design, simulation, and automated discovery.

---

## Scientific Agent Capabilities

| Capability | Description | Domains |
|---|---|---|
| **Hypothesis Generation** | Propose novel, testable hypotheses | All sciences |
| **Literature Mining** | Extract knowledge from papers | All sciences |
| **Experimental Design** | Plan experiments, power analysis | Wet lab, clinical, field |
| **Simulation** | Molecular dynamics, CFD, FEM, climate | Physics, chemistry, biology, climate |
| **Data Analysis** | Process experimental data | All sciences |
| **Model Discovery** | Symbolic regression, equation discovery | Physics, chemistry, biology |
| **Automated Experimentation** | Lab automation, robotics | Chemistry, biology, materials |
| **Scientific Writing** | Papers, grants, reports | All sciences |
| **Peer Review Assistance** | Review, replicate, verify | All sciences |

---

## Scientific Agent Architectures

### 1. Hypothesis Generation Agent
```python
class ScientificHypothesisAgent:
    def __init__(self, knowledge_graph: ScientificKnowledgeGraph):
        self.kg = knowledge_graph
        self.embedder = ScientificEmbedder("specter2")
        self.analogical = AnalogicalReasoning()
    
    async def generate_hypotheses(self, 
                                  domain: str, 
                                  observation: str,
                                  num_hypotheses: int = 5) -> list[ScientificHypothesis]:
        # 1. Map to knowledge graph
        context = await self.kg.query_subgraph(domain, observation)
        
        # 2. Identify anomalies & gaps
        gaps = await self.identify_knowledge_gaps(context, observation)
        
        # 3. Generate via multiple strategies
        hypotheses = []
        
        # Strategy 1: Analogical transfer
        for gap in gaps:
            analogies = await self.analogical.find_analogies(gap, self.kg)
            for analogy in analogies[:2]:
                h = await self.transfer_hypothesis(analogy, gap)
                hypotheses.append(h)
        
        # Strategy 2: Causal reasoning
        for gap in gaps:
            causal_h = await self.causal_reasoning(gap, context)
            hypotheses.append(causal_h)
        
        # Strategy 3: Combinatorial
        for gap in gaps:
            combo_h = await self.combinatorial_hypothesis(gap, context)
            hypotheses.append(combo_h)
        
        # 4. Filter & rank
        filtered = await self.filter_hypotheses(hypotheses)
        ranked = await self.rank_by_criteria(filtered)
        
        return ranked[:num_hypotheses]
    
    def evaluate_hypothesis(self, h: ScientificHypothesis) -> HypothesisScore:
        return HypothesisScore(
            novelty=self.assess_novelty(h),
            testability=self.assess_testability(h),
            plausibility=self.assess_plausibility(h),
            impact=self.assess_impact(h),
            falsifiability=self.assess_falsifiability(h)
        )
```

### 2. Experimental Design Agent
```python
class ExperimentalDesignAgent:
    def __init__(self):
        self.stats = StatisticalDesign()
        self.simulator = ExperimentSimulator()
        self.safety = SafetyChecker()
    
    async def design_experiment(self, hypothesis: ScientificHypothesis) -> ExperimentProtocol:
        # 1. Identify variables
        iv, dv, controls = self.identify_variables(hypothesis)
        
        # 2. Select design
        design = self.select_design(iv, dv, controls)
        
        # 3. Power analysis
        power = await self.stats.power_analysis(
            effect_size=hypothesis.expected_effect_size,
            alpha=0.05,
            power=0.8,
            design=design
        )
        
        # 4. Randomization & blocking
        rand_plan = self.design_randomization(iv, dv, controls)
        
        # 5. Safety & ethics
        safety = await self.safety.assess(design)
        
        # 5. Protocol generation
        protocol = ExperimentProtocol(
            hypothesis=hypothesis,
            design=design,
            sample_size=power.sample_size,
            randomization=rand_plan,
            procedures=self.generate_procedures(design),
            materials=self.list_materials(design),
            safety=safety,
            timeline=self.estimate_timeline(design),
            budget=self.estimate_budget(design)
        )
        
        return protocol
    
    def select_design(self, iv, dv, controls) -> ExperimentalDesign:
        if iv.categorical and dv.continuous:
            if len(iv.levels) == 2:
                return DesignType.TWO_GROUP_RCT
            else:
                return DesignType.MULTI_GROUP_RCT
        elif iv.continuous and dv.continuous:
            return DesignType.DOSE_RESPONSE
        elif iv.time_series:
            return DesignType.LONGITUDINAL
        ...
```

### 3. Laboratory Automation Agent
```python
class LabAutomationAgent:
    def __init__(self, lab_interface: LabInterface):
        self.lab = lab_interface
        self.planner = ProtocolPlanner()
        self.scheduler = ExperimentScheduler()
    
    async def execute_protocol(self, protocol: ExperimentProtocol) -> ExperimentResult:
        # 1. Validate protocol
        validation = await self.validate_protocol(protocol)
        
        # 2. Prepare materials
        await self.prepare_materials(protocol.materials)
        
        # 3. Schedule runs
        schedule = await self.scheduler.schedule(protocol)
        
        # 4. Execute with monitoring
        results = []
        for step in protocol.steps:
            result = await self.execute_step(step)
            results.append(result)
            
            # Safety check
            if not await self.safety.check(result):
                await self.emergency_stop()
                raise SafetyViolation(f"Step {step.id} failed safety check")
        
        # 4. Data collection
        data = await self.collect_data(protocol)
        
        return ExperimentResult(
            protocol=protocol,
            steps=results,
            data=data,
            metadata=self.generate_metadata(protocol)
        )
    
    async def execute_step(self, step: ProtocolStep) -> StepResult:
        if step.type == "liquid_handling":
            return await self.lab.liquid_handler.execute(step.params)
        elif step.type == "incubation":
            return await self.lab.incubator.execute(step.params)
        elif step.type == "measurement":
            return await self.lab.instrument.execute(step.params)
        elif step.type == "imaging":
            return await self.lab.imager.execute(step.params)
        ...
```

### 4. Simulation Agent
```python
class SimulationAgent:
    def __init__(self):
        self.engines = {
            "molecular_dynamics": MDEngine("openmm"),
            "quantum_chemistry": QCEngine("psi4"),
            "cfd": CFDEngine("openfoam"),
            "fem": FEMengine("fenics"),
            "climate": ClimateEngine("cesm"),
            "agent_based": ABMEngine("mesa"),
            "stochastic": GillespieEngine()
        }
    
    async def run_simulation(self, spec: SimulationSpec) -> SimulationResult:
        engine = self.engines[spec.engine_type]
        
        # 1. Prepare input
        input_files = await self.prepare_input(spec)
        
        # 2. Submit job
        job = await engine.submit(
            input_files=input_files,
            parameters=spec.parameters,
            resources=spec.resources
        )
        
        # 3. Monitor
        result = await self.monitor(job)
        
        # 3. Post-process
        return await self.post_process(result, spec.output_format)
    
    async def parameter_sweep(self, spec: SweepSpec) -> SweepResult:
        # Latin hypercube sampling
        param_sets = LatinHypercubeSampling(spec.param_space, spec.n_samples)
        
        # Batch submit
        jobs = await asyncio.gather(*[
            self.engines[spec.engine_type].submit_batch(
                base_params=spec.base_params,
                variations=params
            ) for params in param_sets
        ])
        
        return await self.aggregate_results(jobs)
```

---

## Domain-Specific Scientific Agents

### 1. Drug Discovery Agent
```python
class DrugDiscoveryAgent:
    def __init__(self):
        self.target_id = TargetIdentification()
        self.hit_finding = HitFinding()
        self.lead_opt = LeadOptimization()
        self.admet = ADMETPrediction()
        self.docking = MolecularDocking()
    
    async def discover(self, target: Target, config: DiscoveryConfig) -> DiscoveryResult:
        # 1. Target validation
        validation = await self.target_id.validate(target)
        
        # 2. Hit finding
        hits = await self.hit_finding.find(target, config.library)
        
        # 3. Hit validation
        validated = await self.validate_hits(hits, target)
        
        # 4. Lead optimization
        leads = await self.lead_opt.optimize(validated, config.objectives)
        
        # 5. ADMET prediction
        admet = await self.admet.predict(leads)
        
        # 6. Prioritization
        candidates = self.prioritize(leads, admet, config.criteria)
        
        return DiscoveryResult(
            target=target,
            hits=hits,
            leads=leads,
            candidates=candidates,
            admet=admet
        )
    
    async def design_molecule(self, target: Target, constraints: Constraints) -> Molecule:
        # Generative model
        molecule = await self.generative_model.generate(
            target=target,
            constraints=constraints
        )
        
        # Validate
        if not await self.validate_molecule(molecule, target):
            return await self.design_molecule(target, constraints)
        
        return molecule
```

### 2. Materials Discovery Agent
```python
class MaterialsDiscoveryAgent:
    def __init__(self):
        self.crystal_predictor = CrystalStructurePredictor()
        self.property_predictor = PropertyPredictor()
        self.synthesis_planner = SynthesisPlanner()
        self.phase_diagram = PhaseDiagramAnalyzer()
    
    async def discover_material(self, target_properties: PropertyTarget) -> MaterialCandidate:
        # 1. Inverse design
        structures = await self.inverse_design(target_properties)
        
        # 2. Stability check
        stable = await self.filter_stable(structures)
        
        # 3. Property prediction
        properties = await self.property_predictor.predict(stable)
        
        # 4. Synthesis feasibility
        feasible = await self.synthesis_planner.plan(properties)
        
        # 5. Cost estimation
        cost = self.estimate_cost(feasible)
        
        return MaterialCandidate(
            structure=feasible,
            properties=properties,
            synthesis_route=feasible.route,
            estimated_cost=cost
        )
```

### 3. Climate Science Agent
```python
class ClimateScienceAgent:
    def __init__(self):
        self.models = {
            "earth_system": EarthSystemModel("CESM"),
            "atmosphere": AtmosphereModel("WRF"),
            "ocean": OceanModel("MOM6"),
            "ice": IceSheetModel("ISSM"),
            "carbon_cycle": CarbonCycleModel("CLM")
        }
        self.analysis = ClimateAnalysis()
    
    async def run_projection(self, scenario: Scenario, config: ProjectionConfig) -> ProjectionResult:
        # 1. Select model
        model = self.models[config.model_type]
        
        # 2. Initialize
        initial_state = await self.initialize_state(scenario, config)
        
        # 3. Run ensemble
        ensemble = await self.run_ensemble(model, initial_state, config.ensemble_size)
        
        # 3. Analyze
        analysis = await self.analysis.analyze(ensemble)
        
        return ProjectionResult(
            ensemble=ensemble,
            analysis=analysis,
            uncertainty=self.quantify_uncertainty(ensemble)
        )
```

---

## Automated Discovery Systems

### 1. AI Scientist (End-to-End)
```python
class AIScientist:
    def __init__(self):
        self.hypothesis_agent = HypothesisAgent()
        self.design_agent = ExperimentalDesignAgent()
        self.lab_agent = LabAutomationAgent()
        self.analysis_agent = DataAnalysisAgent()
        self.writing_agent = ScientificWritingAgent()
        self.review_agent = PeerReviewAgent()
    
    async def run_research_cycle(self, domain: str, budget: Budget) -> ResearchOutput:
        # 1. Idea generation
        hypotheses = await self.hypothesis_agent.generate(domain)
        
        # 2. Select best
        selected = self.select_hypotheses(hypotheses, budget)
        
        # 3. For each hypothesis
        results = []
        for h in selected:
            # Design experiment
            protocol = await self.design_agent.design(h)
            
            # Execute (simulation or lab)
            if protocol.requires_lab:
                result = await self.lab_agent.execute(protocol)
            else:
                result = await self.simulation_agent.run(protocol)
            
            # Analyze
            analysis = await self.analysis_agent.analyze(result)
            
            # Write paper
            paper = await self.writing_agent.write(h, result, analysis)
            
            # Self-review
            review = await self.review_agent.review(paper)
            
            results.append(ResearchOutput(
                hypothesis=h,
                protocol=protocol,
                result=result,
                analysis=analysis,
                paper=paper,
                review=review
            ))
        
        return ResearchOutput(results=results)
```

### 2. Coscientist (Chemistry)
```python
class Coscientist:
    def __init__(self):
        self.planner = SynthesisPlanner()
        self.executor = LabAutomationAgent()
        self.analyzer = SpectralAnalyzer()
    
    async def synthesize_target(self, target: Molecule) -> SynthesisResult:
        # 1. Retrosynthesis
        routes = await self.planner.retrosynthesize(target)
        
        # 2. Select best route
        best_route = self.select_route(routes)
        
        # 3. Execute
        result = await self.executor.execute_route(best_route)
        
        # 4. Analyze
        analysis = await self.analyzer.analyze(result.product)
        
        # 5. Verify
        verified = self.verify(analysis, target)
        
        return SynthesisResult(
            route=best_route,
            yield=result.yield,
            purity=analysis.purity,
            verified=verified
        )
```

---

## Scientific Benchmarks

| Benchmark | Domain | Task |
|---|---|---|
| **LitReview** | All | Literature review quality |
| **HypoGen** | All | Hypothesis novelty/testability |
| **ExpDesign** | All | Design validity, power |
| **MolGen** | Chemistry | Molecule validity, properties |
| **RetroSyn** | Chemistry | Route correctness, yield |
| **ProteinDesign** | Biology | Structure, function, stability |
| **MaterialsGen** | Materials | Stability, target properties |
| **TheoremProving** | Math | Proof correctness |
| **Simulation** | Physics | Accuracy, speed |
| **CausalInference** | All | ATE accuracy, coverage |

---

## Safety & Ethics

### Responsible AI in Science
```python
class ScientificEthicsChecker:
    def __init__(self):
        self.dual_use = DualUseDetector()
        self.biosafety = BiosafetyClassifier()
        self.ethics = EthicsReviewer()
    
    async def review(self, research: ResearchProposal) -> EthicsReview:
        # Dual-use concern
        dual_use = await self.dual_use.assess(research)
        
        # Biosafety
        biosafety = await self.biosafety.classify(research)
        
        # Ethics
        ethics = await self.ethics.review(research)
        
        return EthicsReview(
            approved=dual_use.safe and biosafety.safe and ethics.approved,
            concerns=[dual_use.concerns, biosafety.concerns, ethics.concerns],
            mitigations=[dual_use.mitigations, biosafety.mitigations, ethics.mitigations],
            required_approvals=self.determine_approvals(dual_use, biosafety, ethics)
        )
```

### Reproducibility
```python
class ReproducibilityManager:
    def __init__(self):
        self.container = ContainerManager()
        self.versioning = DataVersioning()
        self.workflow = WorkflowManager()
    
    async def ensure_reproducible(self, experiment: Experiment) -> ReproducibilityReport:
        # 1. Containerize environment
        container = await self.container.build(experiment.environment)
        
        # 2. Version data
        data_version = await self.versioning.snapshot(experiment.data)
        
        # 3. Capture workflow
        workflow = await self.workflow.capture(experiment)
        
        # 4. Test re-execution
        reproduced = await self.re_execute(container, workflow, data_version)
        
        return ReproducibilityReport(
            container_digest=container.digest,
            data_version=data_version.version,
            workflow_hash=workflow.hash,
            reproduced=reproduced.success,
            differences=reproduced.differences
        )
```

---

## Related Notes

- [[Domains/Research-Agents.md]] — Research methods
- [[Domains/Coding-Agents.md]] — Code for science
- [[Domains/Data-Analysis-Agents.md]] — Data analysis methods
- [[ModelArchitectures/Reasoning-Models.md]] — Reasoning for science
- [[ModelArchitectures/Mixture-of-Experts.md]] — MoE for science
- [[ContextEngineering/Knowledge-Graphs.md]] — KG for science
- [[ContextEngineering/Advanced-RAG.md]] — RAG for literature
- [[ToolUse/Code-Execution.md]] — Simulation environments
- [[ToolUse/API-Integration.md]] — Lab APIs, databases
- [[ContextEngineering/Knowledge-Graphs.md]] — KG for science
- [[Frontiers/Self-Improving-Agents.md]] — Self-improving science
- [[Frontiers/Research-Directions.md]] — Future directions
- [[Frontiers/Neurosymbolic-Integration.md]] — Neurosymbolic science
- [[Frontiers/Formal-Verification.md]] — Verified science
- [[Evaluation/Agent-Benchmarks.md]] — Benchmarks
- [[Production/Observability.md]] — Experiment monitoring
- [[Production/Reliability.md]] — Experiment reliability
- [[Production/Security.md]] — Lab security
- [[Security/Supply-Chain-Security.md]] — Reagent supply chain
- [[Frontiers/Research-Directions.md]] — Future directions
- [[Frontiers/Self-Improving-Agents.md]] — Self-improving science

---

*Sources: "The AI Scientist" (Lu et al.), "Coscientist" (Boiko et al.), "AutoML for Science", "AlphaFold" (DeepMind), "AlphaFold-Multimer", "ESM" (Meta), "AlphaFold-Multimer", "RoseTTAFold", "DiffDock", "EquiBind", "MolGPT", "ChemBERTa", "MolBERT", "Graphormer", "Uni-Mol", "Molecular Dynamics (OpenMM, GROMACS, AMBER)", "Quantum Chemistry (Psi4, ORCA, VASP)", "CFD (OpenFOAM, SU2)", "FEM (FEniCS, deal.II)", "Climate Models (CESM, WRF, MOM6)", "Lab Automation (Opentrons, Tecan, Hamilton)", "Cloud Labs (Emerald, Strateos, Transcriptic)", "Electronic Lab Notebooks (Benchling, LabArchives)", "LIMS (LabWare, STARLIMS)", "DOE (Design of Experiments)", "Response Surface Methodology", "Bayesian Optimization", "Active Learning", "Scientific Machine Learning (SciML)", "Neural Operators", "Physics-Informed Neural Networks (PINNs)", "Symbolic Regression (PySR, Eureqa)", "Automated Theorem Proving (Lean, Isabelle)", "Formal Verification", "Causal Discovery (PC, GES, NOTEARS)", "Causal Inference (DoWhy, EconML, CausalML)", "Scientific Machine Learning (SciML)", "Julia (DifferentialEquations, ModelingToolkit)", "SciML Open Source Software", "Materials Project", "OQMD", "AFLOW", "NOMAD", "COD", "ICSD", "PubChem", "ChEMBL", "BindingDB", "PDB", "UniProt", "AlphaFold DB", "ESM Atlas", "Open Targets", "DrugBank", "ClinicalTrials.gov", "FAIR principles", "Reproducibility Crisis", "Open Science", "Plan S", "ORCID", "ROR", "DOI", "ORCID", "Research Data Management", "Data Management Plans", "Metadata Standards (Dublin Core, DataCite, Schema.org)", "RO-Crate", "FAIRshake", "FORCE11", "COAR", "RDA", "GO FAIR".*