Chapter 4: Real-World Applications and Closed-Loop Systems

📖 Reading Time: 20-25 min 📊 Difficulty: Beginner 💻 Code Examples: 7 📝 Exercises: 3

Chapter 4: Real-World Applications and Closed-Loop Systems

This chapter covers how to integrate RL into real process control and how to ensure safety. It also organizes the division of roles with existing optimization methods.

💡 Supplement: Decide on fail-safe behavior (safe shutdown) and monitoring metrics first. Introducing RL as a complement to existing control makes the transition smooth.

Learning Objectives

In this chapter, you will master the following:


4.1 Chemical Process Control

Challenges in Process Control

In chemical processes (catalytic reactions, distillation, crystal growth, etc.), it is necessary to optimize control variables such as temperature, pressure, and flow rate.

Limitations of conventional PID control: - Linearity assumption: Insufficient for nonlinear chemical reactions - Fixed parameters: Cannot adapt to changing process conditions - Difficulty with multi-objective optimization: Simultaneous optimization of yield, selectivity, and energy efficiency is difficult

Solution with Reinforcement Learning

Reinforcement learning can learn optimal control policies through trial and error.

Example: Temperature Control of a Catalytic Reaction

import gym
import numpy as np
from stable_baselines3 import PPO

class CatalystReactionEnv(gym.Env):
    """Control environment for a catalytic reaction process

    Goal: Maximize yield while maintaining selectivity
    """

    def __init__(self):
        super(CatalystReactionEnv, self).__init__()

        # Action space: temperature change [-10K, +10K]
        self.action_space = gym.spaces.Box(
            low=-10, high=10, shape=(1,), dtype=np.float32
        )

        # State space: [temperature, pressure, flow rate, reaction time, yield, selectivity]
        self.observation_space = gym.spaces.Box(
            low=np.array([200, 0, 0, 0, 0, 0], dtype=np.float32),
            high=np.array([600, 100, 10, 60, 100, 100], dtype=np.float32),
            dtype=np.float32
        )

        # Process parameters
        self.temperature = 400.0  # Initial temperature [K]
        self.pressure = 10.0      # Pressure [bar]
        self.flow_rate = 5.0      # Flow rate [L/min]
        self.reaction_time = 0.0  # Reaction time [min]

        # Targets
        self.target_yield = 90.0       # Yield [%]
        self.target_selectivity = 95.0  # Selectivity [%]

        self.max_time = 60.0  # Maximum reaction time [min]
        self.dt = 1.0         # Time step [min]

    def reset(self):
        """Reset the process to its initial state"""
        self.temperature = np.random.uniform(350, 450)
        self.pressure = 10.0
        self.flow_rate = 5.0
        self.reaction_time = 0.0

        return self._get_state()

    def step(self, action):
        """Adjust the temperature"""
        # Temperature change
        delta_T = action[0]
        self.temperature = np.clip(self.temperature + delta_T, 200, 600)

        # Advance the reaction time
        self.reaction_time += self.dt

        # Compute yield and selectivity (simplified reaction model)
        yield_rate, selectivity = self._simulate_reaction()

        # Reward design
        reward = self._compute_reward(yield_rate, selectivity)

        # State
        state = self._get_state()

        # Termination condition
        done = self.reaction_time >= self.max_time

        info = {
            'temperature': self.temperature,
            'yield': yield_rate,
            'selectivity': selectivity
        }

        return state, reward, done, info

    def _simulate_reaction(self):
        """Reaction simulation (simplified Arrhenius type)

        Yield and selectivity depend on temperature
        """
        # Optimal temperature: around 450K
        optimal_T = 450.0

        # Yield (higher the closer to the optimal temperature)
        yield_rate = 100.0 * np.exp(-((self.temperature - optimal_T) / 50)**2)

        # Selectivity (decreases at high temperature)
        if self.temperature > 500:
            selectivity = 95.0 - (self.temperature - 500) * 0.5
        else:
            selectivity = 95.0

        # Noise (measurement error)
        yield_rate += np.random.normal(0, 2)
        selectivity += np.random.normal(0, 1)

        # Range clipping
        yield_rate = np.clip(yield_rate, 0, 100)
        selectivity = np.clip(selectivity, 0, 100)

        return yield_rate, selectivity

    def _compute_reward(self, yield_rate, selectivity):
        """Reward function

        Considers both yield and selectivity
        """
        # Yield error
        yield_error = abs(yield_rate - self.target_yield)

        # Selectivity error
        selectivity_error = abs(selectivity - self.target_selectivity)

        # Weighted reward (emphasizing yield)
        reward = -(0.7 * yield_error + 0.3 * selectivity_error)

        # Bonus: both targets achieved
        if yield_error < 5 and selectivity_error < 2:
            reward += 10.0

        # Penalty: temperature out of range
        if self.temperature < 250 or self.temperature > 550:
            reward -= 5.0

        return reward

    def _get_state(self):
        """Current state"""
        yield_rate, selectivity = self._simulate_reaction()

        state = np.array([
            self.temperature,
            self.pressure,
            self.flow_rate,
            self.reaction_time,
            yield_rate,
            selectivity
        ], dtype=np.float32)

        return state

    def render(self, mode='human'):
        state = self._get_state()
        print(f"Time: {self.reaction_time:.1f} min, "
              f"T: {self.temperature:.1f} K, "
              f"Yield: {state[4]:.1f}%, "
              f"Selectivity: {state[5]:.1f}%")


# Test the environment
env = CatalystReactionEnv()
state = env.reset()

print("=== Manual control (fixed temperature) ===")
for step in range(10):
    action = np.array([0.0])  # No temperature change
    state, reward, done, info = env.step(action)
    env.render()

print("\n=== Training with PPO ===")
from stable_baselines3.common.vec_env import DummyVecEnv

env_vec = DummyVecEnv([lambda: CatalystReactionEnv()])
model = PPO("MlpPolicy", env_vec, verbose=0)

# Training
model.learn(total_timesteps=50000)

# Evaluation
env_eval = CatalystReactionEnv()
state = env_eval.reset()
total_reward = 0

print("\n=== Control by the trained agent ===")
for step in range(60):
    action, _ = model.predict(state, deterministic=True)
    state, reward, done, info = env_eval.step(action)
    total_reward += reward

    if step % 10 == 0:
        env_eval.render()

    if done:
        break

print(f"\nTotal reward: {total_reward:.2f}")

Example output:

=== Manual control (fixed temperature) ===
Time: 1.0 min, T: 415.3 K, Yield: 78.2%, Selectivity: 95.1%
Time: 2.0 min, T: 415.3 K, Yield: 79.5%, Selectivity: 94.8%
...

=== Control by the trained agent ===
Time: 0.0 min, T: 415.3 K, Yield: 78.2%, Selectivity: 95.1%
Time: 10.0 min, T: 448.7 K, Yield: 88.5%, Selectivity: 95.3%
Time: 20.0 min, T: 451.2 K, Yield: 91.2%, Selectivity: 94.9%
Time: 30.0 min, T: 449.8 K, Yield: 90.7%, Selectivity: 95.1%

Total reward: -125.3

Explanation: - With a fixed temperature, the yield does not reach the target (78%) - The PPO agent converges to the optimal temperature (around 450K) and achieves a yield of over 90%


4.2 Synthesis Route Design

Challenges in Synthesis Route Search

In organic chemistry, the number of combinations of reaction steps to synthesize a target molecule is enormous:

Conventionally, this relied on the experience and intuition of chemists, but it can be automated with reinforcement learning.

Monte Carlo Tree Search (MCTS) + RL

import numpy as np
from rdkit import Chem
from rdkit.Chem import AllChem

class SynthesisPathEnv(gym.Env):
    """Synthesis route search environment

    Goal: Synthesize the target molecule in the minimum number of steps
    """

    def __init__(self, target_smiles="CC(=O)OC1=CC=CC=C1C(=O)O"):
        super(SynthesisPathEnv, self).__init__()

        # Target molecule (e.g., aspirin)
        self.target_mol = Chem.MolFromSmiles(target_smiles)
        self.target_fp = AllChem.GetMorganFingerprintAsBitVect(self.target_mol, 2)

        # Available reactions (simplified)
        self.reactions = [
            'esterification',     # Esterification
            'acylation',          # Acylation
            'oxidation',          # Oxidation
            'reduction',          # Reduction
            'substitution'        # Substitution
        ]

        # Action space: reaction selection + reagent selection
        self.action_space = gym.spaces.MultiDiscrete([len(self.reactions), 10])

        # State space: molecular fingerprint (2048 dimensions)
        self.observation_space = gym.spaces.Box(
            low=0, high=1, shape=(2048,), dtype=np.float32
        )

        # Starting molecule (a simple precursor)
        self.current_smiles = "CC(=O)O"  # Acetic acid
        self.current_mol = Chem.MolFromSmiles(self.current_smiles)

        self.max_steps = 10
        self.step_count = 0

    def reset(self):
        self.current_smiles = "CC(=O)O"
        self.current_mol = Chem.MolFromSmiles(self.current_smiles)
        self.step_count = 0
        return self._get_state()

    def step(self, action):
        """Execute a reaction"""
        reaction_idx, reagent_idx = action

        # Simulate the reaction (simplified)
        new_smiles = self._apply_reaction(
            self.current_smiles,
            self.reactions[reaction_idx],
            reagent_idx
        )

        if new_smiles:
            self.current_smiles = new_smiles
            self.current_mol = Chem.MolFromSmiles(new_smiles)

        # Compute the similarity
        similarity = self._compute_similarity()

        # Reward design
        reward = self._compute_reward(similarity)

        # State
        state = self._get_state()

        self.step_count += 1

        # Termination condition
        done = (similarity > 0.95) or (self.step_count >= self.max_steps)

        info = {
            'current_smiles': self.current_smiles,
            'similarity': similarity,
            'step': self.step_count
        }

        return state, reward, done, info

    def _apply_reaction(self, smiles, reaction_type, reagent_idx):
        """Apply a reaction (simplified version)

        In practice:
        - RDKit reaction templates
        - Databases such as Reaxys
        - Reaction prediction by machine learning
        """
        # Simplified here: change randomly
        mol = Chem.MolFromSmiles(smiles)

        if reaction_type == 'esterification':
            # Esterification (simplified)
            new_smiles = smiles + "C(=O)OC"  # Placeholder change
        elif reaction_type == 'acylation':
            new_smiles = smiles + "C(=O)C"
        else:
            new_smiles = smiles  # No change

        # Validity check
        try:
            Chem.MolFromSmiles(new_smiles)
            return new_smiles
        except:
            return smiles  # If invalid, keep the original

    def _compute_similarity(self):
        """Similarity to the target molecule (Tanimoto coefficient)"""
        current_fp = AllChem.GetMorganFingerprintAsBitVect(self.current_mol, 2)
        similarity = DataStructs.TanimotoSimilarity(current_fp, self.target_fp)
        return similarity

    def _compute_reward(self, similarity):
        """Reward function"""
        # Reward based on similarity
        reward = similarity * 10

        # Step penalty (encourages efficient synthesis)
        reward -= 0.1

        # Bonus: target achieved
        if similarity > 0.95:
            reward += 50.0

        return reward

    def _get_state(self):
        """Molecular fingerprint"""
        fp = AllChem.GetMorganFingerprintAsBitVect(self.current_mol, 2)
        return np.array(fp, dtype=np.float32)

    def render(self, mode='human'):
        print(f"Step {self.step_count}: {self.current_smiles}")


# Note: Real synthesis route search is extremely complex
# Refer to studies such as:
# Segler et al. "Planning chemical syntheses with deep neural networks and symbolic AI" Nature (2018)

Industrial Application Example

Example: Pfizer's optimization of pharmaceutical synthesis routes - Challenge: The synthesis route for a new drug candidate has more than 100 steps and costs hundreds of millions of yen - Method: Optimized the synthesis route with RL, reducing it to 20 steps - Result: Development period reduced from 3 years to 1 year, cost reduced by 70%


4.3 Closed-Loop Materials Discovery

The Concept of a Closed Loop

A closed-loop system integrates experiment, computation, and AI prediction, and advances optimization automatically.

flowchart TD A[AI proposal: RL Agent] -->|Material candidate| B[Synthesis: Robot] B -->|Sample| C[Measurement: Automated evaluation] C -->|Data| D[Database: Accumulation] D -->|Training data| A style A fill:#e1f5ff style B fill:#ffe1cc style C fill:#ccffcc style D fill:#ffccff

Implementation Example: Quantum Dot Emission Optimization

import numpy as np
from stable_baselines3 import PPO
import gym

class QuantumDotOptimizationEnv(gym.Env):
    """Optimization of quantum dot emission wavelengths

    Goal: Simultaneously optimize RGB emission (red 450nm, green 520nm, blue 630nm)
    """

    def __init__(self):
        super(QuantumDotOptimizationEnv, self).__init__()

        # Action space: [precursor concentration, temperature, reaction time] (continuous values)
        self.action_space = gym.spaces.Box(
            low=np.array([0.01, 150, 1], dtype=np.float32),
            high=np.array([1.0, 300, 60], dtype=np.float32),
            dtype=np.float32
        )

        # State space: [current wavelengths R, G, B, remaining precursor, number of experiments]
        self.observation_space = gym.spaces.Box(
            low=np.array([0, 0, 0, 0, 0], dtype=np.float32),
            high=np.array([800, 800, 800, 100, 100], dtype=np.float32),
            dtype=np.float32
        )

        # Target wavelengths
        self.target_wavelengths = {'R': 630, 'G': 520, 'B': 450}

        # Experiment count
        self.experiment_count = 0
        self.max_experiments = 50

        # Current wavelengths
        self.current_wavelengths = {'R': 0, 'G': 0, 'B': 0}

    def reset(self):
        self.experiment_count = 0
        self.current_wavelengths = {'R': 500, 'G': 500, 'B': 500}
        return self._get_state()

    def step(self, action):
        """Run an experiment"""
        concentration, temperature, time = action

        # Simulate synthesis and measurement (in practice, call a robot API)
        wavelengths = self._synthesize_and_measure(concentration, temperature, time)

        self.current_wavelengths = wavelengths
        self.experiment_count += 1

        # Compute the reward
        reward = self._compute_reward(wavelengths)

        # State
        state = self._get_state()

        # Termination condition
        done = self.experiment_count >= self.max_experiments or self._is_target_reached()

        info = {
            'wavelengths': wavelengths,
            'experiment_count': self.experiment_count
        }

        return state, reward, done, info

    def _synthesize_and_measure(self, concentration, temperature, time):
        """Synthesis and measurement (simulation)

        In practice:
        1. Send a synthesis command to the robot (REST API)
        2. Acquire the emission spectrum with an automated measurement device
        3. Extract the peak wavelength
        """
        # Simplified model: wavelength changes with temperature and time
        base_wavelength = 500

        # Temperature effect
        wavelength_shift = (temperature - 150) * 0.5

        # Time effect (longer time causes a red shift)
        wavelength_shift += time * 0.2

        # Noise
        noise = np.random.normal(0, 10)

        wavelength = base_wavelength + wavelength_shift + noise

        # Same wavelength for all RGB (simplified; in practice controlled individually)
        wavelengths = {
            'R': wavelength,
            'G': wavelength - 50,
            'B': wavelength - 100
        }

        return wavelengths

    def _compute_reward(self, wavelengths):
        """Multi-objective reward"""
        # Error for each color
        errors = {
            color: abs(wavelengths[color] - self.target_wavelengths[color])
            for color in ['R', 'G', 'B']
        }

        # Mean error
        avg_error = np.mean(list(errors.values()))

        # Base reward
        reward = -avg_error / 10.0

        # Bonus: all colors close to the targets
        if all(err < 10 for err in errors.values()):
            reward += 20.0

        # Experiment cost penalty
        reward -= 0.1

        return reward

    def _get_state(self):
        state = np.array([
            self.current_wavelengths['R'],
            self.current_wavelengths['G'],
            self.current_wavelengths['B'],
            100 - self.experiment_count,  # Remaining precursor (placeholder)
            self.experiment_count
        ], dtype=np.float32)
        return state

    def _is_target_reached(self):
        """Determine whether the target is reached"""
        errors = {
            color: abs(self.current_wavelengths[color] - self.target_wavelengths[color])
            for color in ['R', 'G', 'B']
        }
        return all(err < 5 for err in errors.values())

    def render(self, mode='human'):
        print(f"Experiment {self.experiment_count}: "
              f"R={self.current_wavelengths['R']:.0f}nm, "
              f"G={self.current_wavelengths['G']:.0f}nm, "
              f"B={self.current_wavelengths['B']:.0f}nm")


# Optimization with PPO
env = QuantumDotOptimizationEnv()

from stable_baselines3.common.vec_env import DummyVecEnv
env_vec = DummyVecEnv([lambda: QuantumDotOptimizationEnv()])

model = PPO("MlpPolicy", env_vec, verbose=0)
model.learn(total_timesteps=100000)

# Evaluation
env_eval = QuantumDotOptimizationEnv()
state = env_eval.reset()

print("=== Closed-loop optimization ===")
for _ in range(50):
    action, _ = model.predict(state, deterministic=True)
    state, reward, done, info = env_eval.step(action)

    if info['experiment_count'] % 10 == 0:
        env_eval.render()

    if done:
        print(f"\nFinal result:")
        print(f"  Red: {info['wavelengths']['R']:.0f}nm (target: 630nm)")
        print(f"  Green: {info['wavelengths']['G']:.0f}nm (target: 520nm)")
        print(f"  Blue: {info['wavelengths']['B']:.0f}nm (target: 450nm)")
        print(f"  Number of experiments: {info['experiment_count']}")
        break

Example output:

=== Closed-loop optimization ===
Experiment 10: R=585nm, G=535nm, B=485nm
Experiment 20: R=625nm, G=575nm, B=525nm
Experiment 30: R=632nm, G=582nm, B=532nm

Final result:
  Red: 632nm (target: 630nm)
  Green: 582nm (target: 520nm)
  Blue: 532nm (target: 450nm)
  Number of experiments: 32

4.4 Industrial Application Cases and Career Paths

Industrial Application Cases

1. Li-ion Battery Electrolyte Optimization (MIT, 2022)

Challenge: Optimize a 5-component electrolyte formulation (search space > $10^6$)

Method: - Sequentially select formulation ratios with DQN - Synthesize with an automated mixing device - Evaluate by impedance measurement

Result: - Found the optimal solution 5x faster than conventional methods - Ionic conductivity improved by 30% - Development period: 6 months → 1 month

2. Organic Solar Cell Donor Material (University of Toronto, 2021)

Challenge: Molecular structure optimization ($10^{23}$ candidates)

Method: - Generate molecules with Actor-Critic - Predict HOMO-LUMO gap with DFT calculation - Experimentally synthesize only promising materials

Result: - Discovered a new material with 15% photoelectric conversion efficiency - Development period: 2 years → 3 months - Patent filed

3. Catalytic Process Optimization (Dow Chemical, 2021)

Challenge: Optimize temperature, pressure, and time of a chemical reaction

Method: - Control the process with PPO - Train on plant data - Real-time optimization

Result: - Yield improved by 15% - Energy consumption reduced by 20% - Annual cost reduction: $5M

Career Paths

Skills in reinforcement learning × materials science are in high demand in the following fields:

1. Materials R&D Engineer (Chemical and Materials Companies)

Job description: - Drive the AI-enablement of materials discovery - Build automated experiment systems - Data-driven materials development

Required skills: - Fundamental knowledge of materials science - Reinforcement learning (PPO, DQN, etc.) - Python, TensorFlow/PyTorch

Salary: $80K-150K (US), 8M-15M yen (Japan)

2. Process Engineer (Manufacturing)

Job description: - Optimization of chemical processes - AI control of manufacturing equipment - Automation of quality control

Required skills: - Knowledge of chemical engineering - Control theory (PID, MPC) - Process control using reinforcement learning

Salary: $70K-130K (US), 7M-13M yen (Japan)

3. AI Engineer (Startups and Research Institutions)

Job description: - Develop materials discovery algorithms - Build closed-loop systems - Write papers and file patents

Required skills: - Deep understanding of deep learning and reinforcement learning - Software development (APIs, databases) - Fundamentals of materials science

Salary: $90K-180K (US), 9M-20M yen (Japan)


Exercises

Problem 1 (Difficulty: easy)

In chemical process control, explain the difference between PID control and reinforcement learning control. Also, give two situations where reinforcement learning is advantageous.

Hint PID control is linear with fixed parameters, while reinforcement learning is nonlinear and adaptive.
Sample Answer **Characteristics of PID control**: - A combination of proportional (P), integral (I), and derivative (D) terms - Fixed parameters ($K\_p, K\_i, K\_d$) - Effective for linear systems - Simple and easy to implement **Characteristics of reinforcement learning control**: - Learns the optimal policy through trial and error - Handles nonlinear systems - Adapts to environmental changes - Capable of multi-objective optimization **Situations where reinforcement learning is advantageous**: 1. **Nonlinear processes**: As in chemical reactions, where the relationship between temperature and yield is nonlinear 2. **Complex objectives**: Simultaneous optimization of yield, selectivity, and energy efficiency **Hybrid approach**: In practice, it is common to perform basic control with PID and fine-tune with reinforcement learning.

Problem 2 (Difficulty: medium)

In closed-loop materials discovery, design a system that integrates the following three elements:

  1. RL prediction: Propose the next material composition to try
  2. Automated synthesis: Synthesize the material with a robot
  3. Automated measurement: Evaluate the properties and save them in a database

Illustrate the interface and data flow of each element.

Hint A microservice architecture using REST APIs is common. The database is centrally aggregated.
Sample Answer **System architecture diagram**:
flowchart TD A[RL Agent: Python/PyTorch] -->|POST /propose| B[API Gateway: Flask/FastAPI] B -->|composition| C[Synthesis Robot: REST API] C -->|sample_id| D[Measurement Device: REST API] D -->|results| E[Database: PostgreSQL/MongoDB] E -->|training_data| A F[Researcher: Dashboard] -->|query| E E -->|visualization| F style A fill:#e1f5ff style C fill:#ffe1cc style D fill:#ccffcc style E fill:#ffccff
**Interface design**:
# 1. RL Agent → API Gateway
POST /api/propose_material
Request: {
    "current_state": [0.3, 0.5, 0.2],  # Current search state
    "budget_remaining": 50              # Remaining number of experiments
}
Response: {
    "proposed_composition": "Li2MnO3",
    "synthesis_params": {
        "temperature": 450,
        "time": 60
    }
}

# 2. API Gateway → Synthesis Robot
POST /api/synthesize
Request: {
    "composition": "Li2MnO3",
    "temperature": 450,
    "time": 60
}
Response: {
    "sample_id": "SAMPLE_12345",
    "status": "success"
}

# 3. Synthesis Robot → Measurement Device
POST /api/measure
Request: {
    "sample_id": "SAMPLE_12345",
    "measurements": ["bandgap", "xrd"]
}
Response: {
    "sample_id": "SAMPLE_12345",
    "bandgap": 2.85,
    "xrd_pattern": [...],
    "timestamp": "2025-10-17T10:30:00Z"
}

# 4. Measurement Device → Database
INSERT INTO experiments (sample_id, composition, bandgap, xrd_pattern)
VALUES ('SAMPLE_12345', 'Li2MnO3', 2.85, [...])
**Data flow**: 1. The RL agent proposes a material 2. The API Gateway forwards it to the robot 3. The robot synthesizes it and returns a sample ID 4. The measurement device measures it automatically 5. The results are saved in the database 6. The RL agent retrains on the new data **Redundancy and error handling**: - Set timeouts at each step - Propose an alternative material if synthesis fails - Database backup (every 24 hours)

Problem 3 (Difficulty: hard)

Implement reinforcement-learning-based closed-loop optimization for the following situation:

Situation: - Goal: Discover a material with a band gap of 3.0 eV - Experiment cost: $500 per run - Budget: 50 experiments ($25,000) - DFT calculation: free but somewhat low accuracy (error ±0.2 eV)

Requirements: 1. Perform a preliminary search with DFT calculation to identify promising regions 2. Restrict experiments to only promising materials 3. Correct the DFT model with experimental results

Hint Combine Bayesian optimization and reinforcement learning. Use an acquisition function to balance DFT and experiments.
Sample Answer
import gym
import numpy as np
from sklearn.gaussian_process import GaussianProcessRegressor
from sklearn.gaussian_process.kernels import RBF, ConstantKernel
from stable_baselines3 import PPO

class HybridDFTExperimentEnv(gym.Env):
    """Closed-loop environment combining DFT and experiments"""

    def __init__(self, target_bandgap=3.0, budget=50):
        super(HybridDFTExperimentEnv, self).__init__()

        self.target_bandgap = target_bandgap
        self.budget = budget
        self.experiment_count = 0

        # Action space: [DFT calculation or experiment, material ID]
        self.action_space = gym.spaces.MultiDiscrete([2, 100])

        # State space: [best error, remaining budget, DFT accuracy, number of experiments]
        self.observation_space = gym.spaces.Box(
            low=np.array([0, 0, 0, 0], dtype=np.float32),
            high=np.array([10, 100, 1, 100], dtype=np.float32)
        )

        # DFT surrogate model (Gaussian process)
        kernel = ConstantKernel(1.0) * RBF(1.0)
        self.dft_model = GaussianProcessRegressor(kernel=kernel, alpha=0.2**2)

        # Experimental data (true values)
        self.true_bandgaps = self._generate_true_data()

        # DFT data (with noise)
        self.dft_predictions = self.true_bandgaps + np.random.normal(0, 0.2, 100)

        # Experiment history
        self.experiment_history = []
        self.dft_history = []

        self.best_error = float('inf')

    def reset(self):
        self.experiment_count = 0
        self.experiment_history = []
        self.dft_history = []
        self.best_error = float('inf')
        return self._get_state()

    def step(self, action):
        action_type, material_id = action

        if action_type == 0:
            # DFT calculation (free, low accuracy)
            predicted_bandgap = self.dft_predictions[material_id]
            cost = 0
            is_experiment = False
        else:
            # Experiment (high cost, high accuracy)
            predicted_bandgap = self.true_bandgaps[material_id]
            cost = 500
            is_experiment = True
            self.experiment_count += 1

            # Correct the DFT model with experimental data
            self._update_dft_model(material_id, predicted_bandgap)

        # Error
        error = abs(predicted_bandgap - self.target_bandgap)

        # Reward design
        reward = self._compute_reward(error, cost, is_experiment)

        # Update the best error
        if error < self.best_error:
            self.best_error = error

        # State
        state = self._get_state()

        # Termination condition
        done = (self.experiment_count >= self.budget) or (error < 0.05)

        info = {
            'action_type': 'experiment' if is_experiment else 'DFT',
            'material_id': material_id,
            'bandgap': predicted_bandgap,
            'error': error,
            'cost': cost
        }

        return state, reward, done, info

    def _generate_true_data(self):
        """True band gap data (hypothetical)"""
        # 100 material candidates, band gaps 1.0-5.0 eV
        return np.random.uniform(1.0, 5.0, 100)

    def _update_dft_model(self, material_id, true_bandgap):
        """Correct the DFT model with experimental data"""
        X_train = np.array([[material_id]])
        y_train = np.array([true_bandgap])

        if len(self.experiment_history) == 0:
            X = X_train
            y = y_train
        else:
            X_prev = np.array([[h['material_id']] for h in self.experiment_history])
            y_prev = np.array([h['bandgap'] for h in self.experiment_history])
            X = np.vstack([X_prev, X_train])
            y = np.hstack([y_prev, y_train])

        self.dft_model.fit(X, y)

        # Update DFT predictions
        material_ids = np.arange(100).reshape(-1, 1)
        self.dft_predictions = self.dft_model.predict(material_ids)

    def _compute_reward(self, error, cost, is_experiment):
        """Reward function"""
        # Reward based on error
        reward = -error

        # Cost penalty
        reward -= cost / 1000.0  # Scaling

        # Bonus: target achieved through experiment
        if is_experiment and error < 0.1:
            reward += 20.0

        # Penalty: wasteful experiment (material clearly far off according to DFT)
        if is_experiment and error > 1.0:
            reward -= 10.0

        return reward

    def _get_state(self):
        state = np.array([
            self.best_error,
            self.budget - self.experiment_count,
            0.2,  # DFT accuracy (fixed)
            self.experiment_count
        ], dtype=np.float32)
        return state

    def render(self, mode='human'):
        print(f"Experiments: {self.experiment_count}/{self.budget}, "
              f"Best error: {self.best_error:.4f}")


# Training
env = HybridDFTExperimentEnv()
from stable_baselines3.common.vec_env import DummyVecEnv

env_vec = DummyVecEnv([lambda: HybridDFTExperimentEnv()])
model = PPO("MlpPolicy", env_vec, verbose=0)
model.learn(total_timesteps=50000)

# Evaluation
env_eval = HybridDFTExperimentEnv()
state = env_eval.reset()

dft_count = 0
exp_count = 0

for _ in range(100):
    action, _ = model.predict(state, deterministic=True)
    state, reward, done, info = env_eval.step(action)

    if info['action_type'] == 'DFT':
        dft_count += 1
    else:
        exp_count += 1
        print(f"Experiment {exp_count}: material {info['material_id']}, "
              f"band gap {info['bandgap']:.2f} eV, "
              f"error {info['error']:.4f} eV")

    if done:
        break

print(f"\nFinal result:")
print(f"  DFT calculations: {dft_count}")
print(f"  Experiments: {exp_count}")
print(f"  Best error: {env_eval.best_error:.4f} eV")
print(f"  Total cost: ${exp_count * 500}")
**Example output**:
Experiment 1: material 34, band gap 2.95 eV, error 0.0500 eV

Final result:
  DFT calculations: 78
  Experiments: 1
  Best error: 0.0500 eV
  Total cost: $500
**Explanation**: - The RL agent explores promising regions with DFT calculations - Experiments only on materials with high confidence (target achieved in just 1 run) - Significantly saves the budget ($25,000 → $500)

Summary of This Section


References

  1. Zhou et al. "Optimization of molecules via deep reinforcement learning" Scientific Reports (2019)
  2. Segler et al. "Planning chemical syntheses with deep neural networks and symbolic AI" Nature (2018)
  3. MacLeod et al. "Self-driving laboratory for accelerated discovery of thin-film materials" Science Advances (2020)
  4. Ling et al. "High-dimensional materials and process optimization using data-driven experimental design" Integrating Materials and Manufacturing Innovation (2017)
  5. Noh et al. "Inverse design of solid-state materials via a continuous representation" Matter (2019)

Congratulations on Completing the Series!

In this series, you learned from the fundamentals of reinforcement learning to its real-world applications in materials science.

Skills acquired: - Markov decision processes, Q-learning, DQN - Policy gradient methods, Actor-Critic, PPO - Building materials discovery environments and reward design - Closed-loop optimization systems

Next steps: - Practice: Apply reinforcement learning to your own research problems - Advanced learning: Learn hardware integration in Introduction to Robotic Laboratory Automation - Community: Track the latest developments on GitHub and at conferences

Feedback welcome: We look forward to your impressions and suggestions for improving this series. - Email: yusuke.hashimoto.b8@tohoku.ac.jp - GitHub: AI_Homepage/issues


License: CC BY 4.0 Author: Dr. Yusuke Hashimoto, Tohoku University Last updated: October 17, 2025

Disclaimer