AI Agent for Oil & Gas: Automate Exploration, Production Optimization & Safety Operations

The oil and gas industry generates more data per day than almost any other sector on earth. A single offshore platform can produce 2 terabytes of sensor data daily from downhole gauges, flowmeters, vibration sensors, and SCADA systems. Yet most of that data is never analyzed in real time. It sits in historians, gets sampled for monthly reports, and rarely drives immediate operational decisions.

AI agents are changing that equation. Unlike static machine learning models that score a batch of data and return a prediction, agents can autonomously monitor, reason, and act across the full upstream and midstream value chain. They ingest live sensor feeds, correlate events across systems, and trigger corrective actions — whether that means adjusting a choke valve setpoint, rescheduling a pipeline inspection, or escalating a safety alarm to the control room.

In this guide, we break down six critical domains where AI agents deliver measurable impact — from subsurface exploration to HSE compliance — with production-ready Python code for each.

1. Exploration & Reservoir Characterization

Subsurface characterization is where every barrel of oil is won or lost before a single foot of hole is drilled. Traditional interpretation workflows — picking horizons on seismic sections, classifying lithology from well logs, running reservoir simulations — are slow, subjective, and bottlenecked by scarce geoscience talent. A senior geophysicist might spend three weeks manually interpreting a single 3D seismic volume.

AI agents accelerate this process by orders of magnitude. An exploration agent can autonomously pick seismic horizons using convolutional neural networks, detect faults with edge-detection architectures, classify lithofacies from well logs, estimate porosity and permeability from multi-sensor data, and rank prospects by economic potential.

Seismic Interpretation: Horizon Tracking & Fault Detection

Modern seismic interpretation agents use U-Net architectures trained on labeled inline/crossline pairs. The agent continuously processes new survey data, identifies reflectors, and maps structural features without human intervention. Fault detection uses a separate classification head that identifies discontinuities in the seismic amplitude field.

Well Log Analysis & Prospect Ranking

Once seismic prospects are identified, the agent ingests well log data (gamma ray, resistivity, neutron porosity, bulk density) to classify lithology, estimate reservoir properties, and rank prospects by expected hydrocarbon volume. The following agent handles the full workflow — from raw LAS file ingestion to prospect scoring:

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class WellLog:
    depth: np.ndarray
    gamma_ray: np.ndarray
    resistivity: np.ndarray
    neutron_porosity: np.ndarray
    bulk_density: np.ndarray

@dataclass
class Prospect:
    name: str
    net_pay_ft: float
    avg_porosity: float
    avg_sw: float
    area_acres: float
    score: float = 0.0

class ExplorationAgent:
    """Autonomous agent for reservoir characterization and prospect ranking."""

    LITHOLOGY_THRESHOLDS = {
        "sandstone": {"gr_max": 75, "rhob_max": 2.55, "nphi_min": 0.10},
        "limestone": {"gr_max": 45, "rhob_max": 2.71, "nphi_min": 0.05},
        "shale":     {"gr_min": 75, "rhob_min": 2.35, "nphi_min": 0.25},
    }

    def __init__(self, porosity_cutoff: float = 0.08, sw_cutoff: float = 0.60):
        self.porosity_cutoff = porosity_cutoff
        self.sw_cutoff = sw_cutoff
        self.prospects: List[Prospect] = []

    def classify_lithology(self, log: WellLog) -> np.ndarray:
        """Classify each depth sample into lithofacies using log responses."""
        n = len(log.depth)
        lithology = np.full(n, "unknown", dtype=object)
        for i in range(n):
            gr, rhob = log.gamma_ray[i], log.bulk_density[i]
            nphi = log.neutron_porosity[i]
            if gr < 75 and rhob < 2.55 and nphi > 0.10:
                lithology[i] = "sandstone"
            elif gr < 45 and rhob < 2.71 and nphi > 0.05:
                lithology[i] = "limestone"
            elif gr >= 75 and nphi > 0.25:
                lithology[i] = "shale"
        return lithology

    def estimate_porosity(self, log: WellLog) -> np.ndarray:
        """Density-neutron crossplot porosity estimation."""
        matrix_density = 2.65  # sandstone matrix
        fluid_density = 1.0
        dphi = (matrix_density - log.bulk_density) / (matrix_density - fluid_density)
        porosity = (dphi + log.neutron_porosity) / 2.0
        return np.clip(porosity, 0.0, 0.45)

    def estimate_water_saturation(self, log: WellLog, porosity: np.ndarray,
                                   rw: float = 0.04, a: float = 0.81,
                                   m: float = 2.0, n: float = 2.0) -> np.ndarray:
        """Archie equation for water saturation."""
        safe_porosity = np.maximum(porosity, 0.001)
        safe_rt = np.maximum(log.resistivity, 0.001)
        sw = np.power((a * rw) / (safe_porosity ** m * safe_rt), 1.0 / n)
        return np.clip(sw, 0.0, 1.0)

    def compute_net_pay(self, lithology: np.ndarray, porosity: np.ndarray,
                        sw: np.ndarray, depth: np.ndarray) -> Dict:
        """Identify net reservoir and compute pay thickness."""
        is_reservoir = (lithology == "sandstone") | (lithology == "limestone")
        is_pay = is_reservoir & (porosity > self.porosity_cutoff) & (sw < self.sw_cutoff)
        depth_step = np.median(np.diff(depth)) if len(depth) > 1 else 0.5
        net_pay = np.sum(is_pay) * depth_step
        avg_porosity = np.mean(porosity[is_pay]) if np.any(is_pay) else 0.0
        avg_sw = np.mean(sw[is_pay]) if np.any(is_pay) else 1.0
        return {"net_pay_ft": net_pay, "avg_porosity": avg_porosity, "avg_sw": avg_sw}

    def rank_prospects(self, prospects: List[Prospect]) -> List[Prospect]:
        """Score and rank prospects by estimated STOOIP (stock-tank OIP)."""
        for p in prospects:
            boe_per_acre_ft = 7758  # barrels per acre-foot
            stooip = (boe_per_acre_ft * p.area_acres * p.net_pay_ft
                      * p.avg_porosity * (1 - p.avg_sw))
            p.score = stooip
        return sorted(prospects, key=lambda x: x.score, reverse=True)

    def run(self, wells: Dict[str, WellLog], areas: Dict[str, float]) -> List[Prospect]:
        """Full exploration workflow: classify, estimate, rank."""
        for name, log in wells.items():
            lithology = self.classify_lithology(log)
            porosity = self.estimate_porosity(log)
            sw = self.estimate_water_saturation(log, porosity)
            pay = self.compute_net_pay(lithology, porosity, sw, log.depth)
            self.prospects.append(Prospect(
                name=name, net_pay_ft=pay["net_pay_ft"],
                avg_porosity=pay["avg_porosity"], avg_sw=pay["avg_sw"],
                area_acres=areas.get(name, 640)
            ))
        return self.rank_prospects(self.prospects)

2. Drilling Optimization

Drilling a single deepwater well costs between $50M and $150M. Non-productive time (NPT) — stuck pipe, equipment failures, wellbore instability — can add 15-25% to that cost. The difference between an optimized well and a troubled one often comes down to real-time parameter adjustment: weight on bit (WOB), rotary speed (RPM), mud flow rate, and mud weight.

A drilling optimization agent ingests real-time surface and downhole sensor data, predicts rate of penetration (ROP), monitors bit wear, detects early signs of stuck pipe, and recommends optimal drilling parameters for the current formation. The agent learns from offset wells and continuously refines its recommendations as new data arrives.

Rate of Penetration Prediction

ROP prediction is the cornerstone of drilling optimization. The agent models the relationship between controllable parameters (WOB, RPM, flow rate) and formation properties (UCS, abrasivity) to predict instantaneous ROP. This enables the agent to recommend parameter changes that maximize penetration rate while staying within equipment limits.

Stuck Pipe Prediction & Drilling Parameter Optimization

Stuck pipe incidents cost the industry an estimated $1 billion annually. Early detection — even 10-15 minutes of advance warning — can prevent the majority of these events. The agent monitors differential pressure, hookload trends, torque spikes, and mud properties to score stuck-pipe risk in real time:

import numpy as np
from dataclasses import dataclass, field
from typing import List, Tuple, Optional
from collections import deque

@dataclass
class DrillingState:
    timestamp: float
    depth_ft: float
    wob_klbs: float
    rpm: float
    flow_rate_gpm: float
    torque_ftlbs: float
    hookload_klbs: float
    standpipe_psi: float
    mud_weight_ppg: float
    rop_fthr: float
    formation_ucs_psi: float = 10000.0

class DrillingOptimizationAgent:
    """Real-time drilling parameter optimization and hazard prediction."""

    def __init__(self, lookback_window: int = 200):
        self.lookback = lookback_window
        self.history: deque = deque(maxlen=lookback_window)
        self.baseline_hookload: Optional[float] = None
        self.baseline_torque: Optional[float] = None

    def predict_rop(self, state: DrillingState) -> float:
        """Bourgoyne-Young ROP model with formation strength correction."""
        a1, a2, a3, a4 = 1.2, 0.00005, 0.8, 0.5
        depth_factor = np.exp(-a2 * state.depth_ft)
        wob_factor = (state.wob_klbs / 4.0) ** a3
        rpm_factor = (state.rpm / 60.0) ** a4
        ucs_factor = 10000.0 / max(state.formation_ucs_psi, 1000)
        predicted_rop = a1 * depth_factor * wob_factor * rpm_factor * ucs_factor * 100
        return round(max(predicted_rop, 1.0), 1)

    def detect_bit_wear(self, states: List[DrillingState]) -> dict:
        """Detect bit wear from ROP decline at constant drilling parameters."""
        if len(states) < 50:
            return {"wear_grade": 0, "recommendation": "Insufficient data"}
        recent = states[-50:]
        wob_std = np.std([s.wob_klbs for s in recent])
        rpm_std = np.std([s.rpm for s in recent])
        if wob_std > 2.0 or rpm_std > 10.0:
            return {"wear_grade": 0, "recommendation": "Parameters unstable"}
        rops = [s.rop_fthr for s in recent]
        early_avg = np.mean(rops[:15])
        late_avg = np.mean(rops[-15:])
        decline_pct = (early_avg - late_avg) / max(early_avg, 1) * 100
        if decline_pct > 40:
            return {"wear_grade": 7, "recommendation": "POOH for bit change"}
        elif decline_pct > 20:
            return {"wear_grade": 4, "recommendation": "Monitor closely"}
        return {"wear_grade": 1, "recommendation": "Bit in good condition"}

    def score_stuck_pipe_risk(self, state: DrillingState) -> dict:
        """Real-time stuck pipe risk scoring from sensor trends."""
        self.history.append(state)
        risk_score = 0.0
        alerts = []

        if self.baseline_hookload is None and len(self.history) > 30:
            self.baseline_hookload = np.mean([s.hookload_klbs for s in self.history])
            self.baseline_torque = np.mean([s.torque_ftlbs for s in self.history])

        if self.baseline_hookload and len(self.history) > 10:
            recent_hl = np.mean([s.hookload_klbs for s in list(self.history)[-10:]])
            hl_change = abs(recent_hl - self.baseline_hookload) / self.baseline_hookload
            if hl_change > 0.15:
                risk_score += 35
                alerts.append(f"Hookload deviation {hl_change:.0%} from baseline")

            recent_tq = np.mean([s.torque_ftlbs for s in list(self.history)[-10:]])
            tq_change = (recent_tq - self.baseline_torque) / max(self.baseline_torque, 1)
            if tq_change > 0.25:
                risk_score += 30
                alerts.append(f"Torque increasing {tq_change:.0%} above baseline")

        ecd_est = state.mud_weight_ppg + (state.flow_rate_gpm * 0.001)
        if ecd_est > state.mud_weight_ppg * 1.12:
            risk_score += 20
            alerts.append("High ECD — differential sticking risk")

        if state.rop_fthr < 5 and state.wob_klbs > 20:
            risk_score += 15
            alerts.append("Low ROP with high WOB — possible pack-off")

        risk_level = "LOW" if risk_score < 30 else "MEDIUM" if risk_score < 60 else "HIGH"
        return {"risk_score": min(risk_score, 100), "risk_level": risk_level, "alerts": alerts}

    def optimize_parameters(self, state: DrillingState) -> dict:
        """Recommend optimal WOB, RPM, and flow rate for current formation."""
        target_rop = self.predict_rop(state)
        opt_wob = min(state.wob_klbs * 1.1, 45.0) if target_rop < 50 else state.wob_klbs
        opt_rpm = min(state.rpm * 1.05, 180.0) if target_rop < 50 else state.rpm
        opt_flow = max(state.flow_rate_gpm, 500) if state.depth_ft > 10000 else state.flow_rate_gpm
        mse = (480 * state.torque_ftlbs * state.rpm) / (max(state.rop_fthr, 1) * 12.25**2)
        return {
            "recommended_wob_klbs": round(opt_wob, 1),
            "recommended_rpm": round(opt_rpm, 0),
            "recommended_flow_gpm": round(opt_flow, 0),
            "current_mse_psi": round(mse, 0),
            "predicted_rop_fthr": target_rop,
            "note": "Reduce WOB if MSE > 3x UCS (founder point exceeded)"
        }

3. Production Optimization

Once a well is on production, the focus shifts from finding and drilling to maximizing recovery while minimizing operating costs. Production optimization covers artificial lift selection and tuning, well test interpretation, decline curve analysis, waterflood management, and production allocation across multi-well facilities.

A production optimization agent continuously monitors flowline pressures, choke positions, water cuts, gas-oil ratios, and artificial lift parameters. It autonomously adjusts lift system settings, detects underperforming wells, forecasts decline rates, and recommends waterflood injection adjustments to maximize sweep efficiency.

Artificial Lift Optimization

Artificial lift systems — ESPs, rod pumps, gas lift — account for roughly 60% of lifting costs in mature fields. Selecting the wrong system or running it at suboptimal settings can waste millions per year. The agent evaluates well conditions and recommends the best lift method, then tunes operating parameters in real time.

Decline Curve Analysis & Waterflood Optimization

The agent fits Arps decline models (exponential, hyperbolic, harmonic) to production history, forecasts EUR (Estimated Ultimate Recovery), and optimizes waterflood injection rates to maintain reservoir pressure and maximize sweep:

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Tuple
from enum import Enum

class LiftType(Enum):
    ESP = "electric_submersible_pump"
    ROD_PUMP = "sucker_rod_pump"
    GAS_LIFT = "gas_lift"
    NATURAL_FLOW = "natural_flow"

@dataclass
class WellState:
    well_id: str
    oil_rate_bpd: float
    water_rate_bpd: float
    gas_rate_mcfd: float
    tubing_pressure_psi: float
    casing_pressure_psi: float
    bhp_psi: float
    depth_ft: float
    temperature_f: float
    lift_type: LiftType
    esp_frequency_hz: float = 0.0
    gas_lift_rate_mcfd: float = 0.0

class ProductionOptimizationAgent:
    """Autonomous production monitoring and optimization agent."""

    def __init__(self):
        self.production_history: Dict[str, List[Tuple[float, float]]] = {}

    def select_artificial_lift(self, well: WellState) -> Dict:
        """Recommend optimal lift method based on well conditions."""
        total_fluid = well.oil_rate_bpd + well.water_rate_bpd
        gor = (well.gas_rate_mcfd * 1000) / max(well.oil_rate_bpd, 1)
        water_cut = well.water_rate_bpd / max(total_fluid, 1)

        scores = {}
        scores[LiftType.ESP] = 0
        scores[LiftType.ROD_PUMP] = 0
        scores[LiftType.GAS_LIFT] = 0

        # Volume scoring
        if total_fluid > 5000:
            scores[LiftType.ESP] += 40
        elif total_fluid > 500:
            scores[LiftType.ESP] += 25
            scores[LiftType.GAS_LIFT] += 20
            scores[LiftType.ROD_PUMP] += 15
        else:
            scores[LiftType.ROD_PUMP] += 35
            scores[LiftType.GAS_LIFT] += 15

        # Depth scoring
        if well.depth_ft > 12000:
            scores[LiftType.GAS_LIFT] += 30
            scores[LiftType.ESP] += 10
        elif well.depth_ft > 8000:
            scores[LiftType.ESP] += 20
            scores[LiftType.GAS_LIFT] += 25
        else:
            scores[LiftType.ROD_PUMP] += 25

        # GOR scoring — high GOR penalizes ESP
        if gor > 2000:
            scores[LiftType.GAS_LIFT] += 25
            scores[LiftType.ESP] -= 15
        elif gor > 800:
            scores[LiftType.GAS_LIFT] += 15

        # Temperature scoring
        if well.temperature_f > 300:
            scores[LiftType.ESP] -= 20
            scores[LiftType.GAS_LIFT] += 15

        best = max(scores, key=scores.get)
        return {"recommended_lift": best.value, "scores": {k.value: v for k, v in scores.items()}}

    def fit_decline_curve(self, times_months: np.ndarray,
                          rates_bpd: np.ndarray) -> Dict:
        """Fit Arps hyperbolic decline and forecast EUR."""
        qi = rates_bpd[0]
        best_fit = None
        best_error = float("inf")

        for b in np.arange(0.0, 2.05, 0.05):
            for di in np.arange(0.005, 0.15, 0.005):
                if b == 0:
                    predicted = qi * np.exp(-di * times_months)
                else:
                    predicted = qi / np.power(1 + b * di * times_months, 1 / b)
                error = np.sum((rates_bpd - predicted) ** 2)
                if error < best_error:
                    best_error = error
                    best_fit = {"qi": qi, "di": di, "b": b}

        # Forecast EUR over 30 years
        t_forecast = np.arange(0, 360)
        b, di = best_fit["b"], best_fit["di"]
        if b == 0:
            forecast = qi * np.exp(-di * t_forecast)
        else:
            forecast = qi / np.power(1 + b * di * t_forecast, 1 / b)
        eur_bbl = np.sum(forecast) * 30.44  # avg days per month
        return {"qi_bpd": qi, "di_monthly": best_fit["di"], "b_factor": best_fit["b"],
                "eur_mmbbl": round(eur_bbl / 1e6, 2), "forecast_bpd": forecast.tolist()}

    def optimize_waterflood(self, producers: List[Dict], injectors: List[Dict]) -> Dict:
        """Balance injection rates to optimize sweep efficiency."""
        total_production = sum(p["oil_rate"] + p["water_rate"] for p in producers)
        total_injection = sum(i["water_rate"] for i in injectors)
        voidage_ratio = total_injection / max(total_production, 1)
        recommendations = []

        for inj in injectors:
            connected_prod = [p for p in producers if p["pattern"] == inj["pattern"]]
            pattern_wc = np.mean([p["water_cut"] for p in connected_prod]) if connected_prod else 0
            if pattern_wc > 0.90:
                recommendations.append({
                    "injector": inj["well_id"], "action": "reduce_rate",
                    "target_rate": inj["water_rate"] * 0.7,
                    "reason": f"Pattern water cut {pattern_wc:.0%} — channel flow likely"
                })
            elif voidage_ratio < 0.95:
                recommendations.append({
                    "injector": inj["well_id"], "action": "increase_rate",
                    "target_rate": inj["water_rate"] * 1.15,
                    "reason": f"Voidage ratio {voidage_ratio:.2f} — pressure declining"
                })
        return {"voidage_ratio": round(voidage_ratio, 2), "recommendations": recommendations}

4. Pipeline & Asset Integrity

Midstream infrastructure — gathering lines, trunk pipelines, separators, compressor stations — represents billions of dollars in capital investment. A single pipeline failure can cost $50M+ in cleanup, fines, and lost production, not counting reputational damage. Integrity management programs must balance safety with cost, inspecting the right assets at the right time.

An asset integrity agent ingests inspection data (ILI runs, UT thickness measurements, CP surveys), environmental data (soil chemistry, coating condition), and operating history (pressure cycles, temperature excursions) to predict corrosion rates, schedule pigging runs, and prioritize maintenance during turnarounds.

Corrosion Rate Prediction

Internal corrosion in pipelines is driven by CO2 partial pressure, water chemistry, flow regime, and temperature. External corrosion depends on coating condition and cathodic protection effectiveness. The agent models both mechanisms and flags segments where wall loss approaches the minimum allowable thickness.

Leak Detection & Pigging Schedule Optimization

The agent correlates pressure transient data, fiber optic distributed acoustic sensing (DAS), and mass balance calculations to detect leaks. It then optimizes pig launch schedules based on predicted corrosion rates, minimizing both risk and production deferment:

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta

@dataclass
class PipelineSegment:
    segment_id: str
    length_km: float
    diameter_in: float
    wall_thickness_mm: float
    min_wall_mm: float
    install_year: int
    coating_type: str
    cp_potential_mv: float  # cathodic protection potential
    co2_pct: float
    temperature_c: float
    pressure_bar: float
    water_cut: float
    flow_velocity_ms: float
    last_ili_date: Optional[str] = None
    last_ili_max_loss_pct: float = 0.0

class PipelineIntegrityAgent:
    """Autonomous pipeline integrity monitoring and maintenance scheduling."""

    def __init__(self):
        self.segments: List[PipelineSegment] = []
        self.alerts: List[Dict] = []

    def predict_internal_corrosion_rate(self, seg: PipelineSegment) -> float:
        """De Waard-Milliams CO2 corrosion rate (mm/year) with corrections."""
        if seg.water_cut < 0.01:
            return 0.01  # dry gas, negligible corrosion
        pco2 = seg.co2_pct / 100.0 * seg.pressure_bar
        log_rate = 5.8 - (1710 / (seg.temperature_c + 273.15)) + 0.67 * np.log10(max(pco2, 0.001))
        base_rate = 10 ** log_rate

        # Flow velocity correction
        if seg.flow_velocity_ms > 3.0:
            velocity_factor = 1.0 + 0.2 * (seg.flow_velocity_ms - 3.0)
        else:
            velocity_factor = 1.0

        # Water cut correction
        wc_factor = min(seg.water_cut / 0.3, 1.0) if seg.water_cut < 0.3 else 1.0
        return round(base_rate * velocity_factor * wc_factor, 3)

    def predict_external_corrosion_rate(self, seg: PipelineSegment) -> float:
        """Estimate external corrosion based on CP effectiveness and coating age."""
        coating_age = datetime.now().year - seg.install_year
        coating_degradation = min(coating_age / 30.0, 1.0)  # assume 30yr coating life

        if seg.cp_potential_mv < -850:  # adequate CP
            cp_factor = 0.1
        elif seg.cp_potential_mv < -750:
            cp_factor = 0.5
        else:
            cp_factor = 1.0  # inadequate CP

        base_external = 0.15  # mm/year bare steel in soil
        return round(base_external * coating_degradation * cp_factor, 3)

    def detect_leak(self, upstream_pressure: List[float], downstream_pressure: List[float],
                    upstream_flow: List[float], downstream_flow: List[float],
                    threshold_pct: float = 2.0) -> Dict:
        """Mass balance and pressure transient leak detection."""
        avg_in = np.mean(upstream_flow[-10:])
        avg_out = np.mean(downstream_flow[-10:])
        imbalance_pct = (avg_in - avg_out) / max(avg_in, 1) * 100

        dp_upstream = np.diff(upstream_pressure[-20:])
        dp_downstream = np.diff(downstream_pressure[-20:])
        rapid_drop = np.any(dp_upstream < -5) or np.any(dp_downstream < -5)

        leak_detected = imbalance_pct > threshold_pct or rapid_drop
        confidence = min((imbalance_pct / threshold_pct) * 50 + (50 if rapid_drop else 0), 100)
        return {
            "leak_detected": leak_detected,
            "confidence_pct": round(confidence, 1),
            "mass_imbalance_pct": round(imbalance_pct, 2),
            "pressure_transient_flag": rapid_drop,
            "action": "ISOLATE AND INSPECT" if leak_detected else "Continue monitoring"
        }

    def schedule_pigging(self, segments: List[PipelineSegment]) -> List[Dict]:
        """Optimize pig launch schedule based on predicted corrosion and risk."""
        schedule = []
        for seg in segments:
            int_rate = self.predict_internal_corrosion_rate(seg)
            ext_rate = self.predict_external_corrosion_rate(seg)
            combined_rate = int_rate + ext_rate

            current_loss = seg.last_ili_max_loss_pct / 100.0 * seg.wall_thickness_mm
            remaining_wall = seg.wall_thickness_mm - current_loss
            margin = remaining_wall - seg.min_wall_mm
            years_to_limit = margin / max(combined_rate, 0.01)

            if years_to_limit < 1.0:
                priority = "CRITICAL"
                next_run_months = 1
            elif years_to_limit < 3.0:
                priority = "HIGH"
                next_run_months = 6
            elif years_to_limit < 7.0:
                priority = "MEDIUM"
                next_run_months = 18
            else:
                priority = "LOW"
                next_run_months = 36

            schedule.append({
                "segment_id": seg.segment_id,
                "corrosion_rate_mm_yr": round(combined_rate, 3),
                "remaining_life_years": round(years_to_limit, 1),
                "priority": priority,
                "next_ili_date": (datetime.now() + timedelta(days=next_run_months * 30)).strftime("%Y-%m-%d")
            })
        return sorted(schedule, key=lambda x: x["remaining_life_years"])

5. HSE & Process Safety

Health, Safety, and Environment (HSE) is non-negotiable in oil and gas. A single process safety event on an offshore platform can result in fatalities, environmental disasters, and billions in liability. Yet safety systems generate an overwhelming volume of data — a typical refinery produces 10,000+ alarms per day, of which 80-90% are nuisance alarms that operators learn to ignore, creating alarm fatigue that masks genuine hazards.

An HSE agent filters nuisance alarms, scores real-time risk across the facility, automates permit-to-work workflows, assists incident investigation with bow-tie analysis, and monitors environmental compliance (flaring volumes, VOC emissions, wastewater discharge).

Gas Detection & Alarm Management

The agent rationalizes alarm setpoints, suppresses known nuisance alarm patterns (weather-related, maintenance-related), and escalates genuine gas detections with context — wind direction, personnel location, ignition source proximity.

Permit-to-Work & Environmental Monitoring

The agent automates the permit-to-work lifecycle — hazard identification, isolation verification, atmospheric testing, and simultaneous operations (SIMOPS) conflict checking. It also tracks flaring, emissions, and discharges against regulatory limits:

import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
from enum import Enum
from collections import deque

class AlarmPriority(Enum):
    CRITICAL = 4
    HIGH = 3
    MEDIUM = 2
    LOW = 1
    SUPPRESSED = 0

@dataclass
class GasAlarm:
    timestamp: datetime
    detector_id: str
    zone: str
    gas_type: str  # H2S, LEL, CO, SO2
    reading_ppm: float
    setpoint_ppm: float
    wind_speed_ms: float
    wind_direction_deg: float

@dataclass
class PermitToWork:
    permit_id: str
    work_type: str  # hot_work, confined_space, electrical, excavation
    zone: str
    start_time: datetime
    end_time: datetime
    isolations: List[str]
    gas_test_required: bool = True
    status: str = "draft"

class HSEAgent:
    """Autonomous HSE monitoring, alarm management, and compliance agent."""

    NUISANCE_PATTERNS = {
        "weather_related": {"wind_speed_min": 15.0, "duration_max_sec": 30},
        "maintenance_tag": {"zones": ["maintenance_bay", "workshop"]},
    }

    EXPOSURE_LIMITS = {
        "H2S": {"TWA_ppm": 10, "STEL_ppm": 15, "IDLH_ppm": 100},
        "LEL": {"alarm_pct": 20, "action_pct": 40, "evacuate_pct": 60},
        "CO": {"TWA_ppm": 25, "STEL_ppm": 100, "IDLH_ppm": 1200},
    }

    def __init__(self):
        self.alarm_history: deque = deque(maxlen=10000)
        self.active_permits: List[PermitToWork] = []
        self.suppressed_count = 0
        self.escalated_count = 0

    def classify_alarm(self, alarm: GasAlarm) -> Dict:
        """Classify gas alarm: genuine, nuisance, or suppressed."""
        # Check nuisance patterns
        if (alarm.wind_speed_ms > 15.0 and
            alarm.reading_ppm < alarm.setpoint_ppm * 1.5 and
            alarm.gas_type == "LEL"):
            self.suppressed_count += 1
            return {"priority": AlarmPriority.SUPPRESSED.name,
                    "reason": "Wind-induced false positive on catalytic sensor",
                    "action": "Log only"}

        # Check against exposure limits
        limits = self.EXPOSURE_LIMITS.get(alarm.gas_type, {})
        if alarm.gas_type == "H2S":
            if alarm.reading_ppm >= limits.get("IDLH_ppm", 100):
                priority = AlarmPriority.CRITICAL
                action = "EVACUATE ZONE — IDLH exceeded"
            elif alarm.reading_ppm >= limits.get("STEL_ppm", 15):
                priority = AlarmPriority.HIGH
                action = "Muster personnel, deploy emergency response"
            else:
                priority = AlarmPriority.MEDIUM
                action = "Investigate source, check wind direction"
        elif alarm.gas_type == "LEL":
            if alarm.reading_ppm >= limits.get("evacuate_pct", 60):
                priority = AlarmPriority.CRITICAL
                action = "EVACUATE — explosive atmosphere imminent"
            elif alarm.reading_ppm >= limits.get("action_pct", 40):
                priority = AlarmPriority.HIGH
                action = "Isolate ignition sources, ventilate area"
            else:
                priority = AlarmPriority.MEDIUM
                action = "Monitor trend, check for leak source"
        else:
            priority = AlarmPriority.LOW
            action = "Log and monitor"

        self.escalated_count += 1
        self.alarm_history.append(alarm)
        return {"priority": priority.name, "action": action,
                "reading": alarm.reading_ppm, "zone": alarm.zone}

    def compute_facility_risk_score(self, alarms_last_hour: List[GasAlarm],
                                     active_hot_work: int,
                                     personnel_on_board: int) -> Dict:
        """Real-time facility risk score from 0-100."""
        risk = 0.0
        genuine = [a for a in alarms_last_hour
                   if self.classify_alarm(a)["priority"] not in ["SUPPRESSED", "LOW"]]
        risk += min(len(genuine) * 10, 40)
        risk += min(active_hot_work * 8, 24)
        if personnel_on_board > 150:
            risk += 10
        max_reading = max([a.reading_ppm for a in alarms_last_hour], default=0)
        if max_reading > 50:
            risk += 26
        risk = min(risk, 100)
        level = "GREEN" if risk < 30 else "YELLOW" if risk < 60 else "ORANGE" if risk < 80 else "RED"
        return {"risk_score": round(risk, 1), "level": level,
                "genuine_alarms": len(genuine), "action": f"Facility status: {level}"}

    def check_simops_conflict(self, new_permit: PermitToWork) -> Dict:
        """Check for simultaneous operations conflicts."""
        conflicts = []
        for existing in self.active_permits:
            if existing.status != "active":
                continue
            time_overlap = (new_permit.start_time < existing.end_time and
                           new_permit.end_time > existing.start_time)
            zone_overlap = new_permit.zone == existing.zone
            dangerous_combo = (
                {new_permit.work_type, existing.work_type} &
                {"hot_work", "confined_space"}
            )
            if time_overlap and (zone_overlap or dangerous_combo):
                conflicts.append({
                    "conflicting_permit": existing.permit_id,
                    "type": existing.work_type, "zone": existing.zone,
                    "reason": "Hot work near confined space entry" if dangerous_combo
                             else "Same zone simultaneous activity"
                })
        return {"permit_id": new_permit.permit_id,
                "conflicts_found": len(conflicts) > 0,
                "conflicts": conflicts,
                "recommendation": "REJECT — resolve conflicts" if conflicts else "APPROVE"}

    def monitor_emissions(self, flaring_mscfd: float, voc_tonnes_day: float,
                          co2_tonnes_day: float, regulatory_limits: Dict) -> Dict:
        """Track emissions against regulatory permit limits."""
        compliance = {}
        for metric, value in [("flaring", flaring_mscfd), ("voc", voc_tonnes_day),
                               ("co2", co2_tonnes_day)]:
            limit = regulatory_limits.get(metric, float("inf"))
            pct = value / limit * 100 if limit > 0 else 0
            status = "COMPLIANT" if pct < 80 else "WARNING" if pct < 100 else "VIOLATION"
            compliance[metric] = {"value": value, "limit": limit,
                                  "pct_of_limit": round(pct, 1), "status": status}
        violations = [k for k, v in compliance.items() if v["status"] == "VIOLATION"]
        return {"compliance": compliance, "violations": violations,
                "overall": "NON-COMPLIANT" if violations else "COMPLIANT"}

6. ROI Analysis for Mid-Size E&P Operators

Let us model the financial impact of deploying AI agents across the value chain for a mid-size E&P operator producing 50,000 barrels of oil equivalent per day (boe/d). We assume a $70/bbl oil price, typical operating costs, and conservative improvement estimates validated against published industry case studies.

Production Uplift

AI-driven production optimization — artificial lift tuning, waterflood balancing, well test automation — typically delivers a 2-5% production uplift in mature fields. For a 50,000 boe/d operator at $70/bbl, even a 3% uplift translates to 1,500 boe/d or roughly $38.3M in additional annual revenue.

NPT Reduction

Drilling optimization agents reduce non-productive time by 15-30% through stuck-pipe prediction, ROP optimization, and proactive bit management. For an operator drilling 20 wells per year at $15M average well cost with 20% NPT, a 20% NPT reduction saves $12M annually.

Maintenance & Integrity Savings

Predictive corrosion monitoring and optimized pigging schedules reduce unplanned maintenance events by 25-40%. For a midstream network with $30M annual maintenance spend, a 30% reduction in unplanned events saves approximately $5.4M per year while actually improving safety outcomes.

The following agent computes the full ROI model:

from dataclasses import dataclass

@dataclass
class OperatorProfile:
    production_boed: float
    oil_price_bbl: float
    wells_drilled_per_year: int
    avg_well_cost_mm: float
    npt_pct: float
    annual_maintenance_mm: float
    annual_hse_incidents: int
    avg_incident_cost_mm: float
    opex_per_boe: float

class ROIAnalysisAgent:
    """Compute ROI of AI agent deployment for E&P operators."""

    def __init__(self, profile: OperatorProfile):
        self.p = profile

    def production_uplift_value(self, uplift_pct: float = 0.03) -> dict:
        """Revenue from production optimization."""
        additional_boed = self.p.production_boed * uplift_pct
        annual_revenue = additional_boed * self.p.oil_price_bbl * 365
        return {
            "uplift_pct": f"{uplift_pct:.0%}",
            "additional_boed": round(additional_boed, 0),
            "annual_revenue_mm": round(annual_revenue / 1e6, 1),
            "source": "Artificial lift optimization, waterflood balancing, well testing"
        }

    def npt_reduction_value(self, npt_reduction_pct: float = 0.20) -> dict:
        """Savings from drilling NPT reduction."""
        total_drilling_spend = self.p.wells_drilled_per_year * self.p.avg_well_cost_mm
        current_npt_cost = total_drilling_spend * self.p.npt_pct
        savings = current_npt_cost * npt_reduction_pct
        return {
            "npt_reduction_pct": f"{npt_reduction_pct:.0%}",
            "current_npt_cost_mm": round(current_npt_cost, 1),
            "annual_savings_mm": round(savings, 1),
            "source": "Stuck pipe prediction, ROP optimization, bit management"
        }

    def maintenance_savings(self, unplanned_reduction_pct: float = 0.30,
                            unplanned_share: float = 0.60) -> dict:
        """Savings from predictive integrity management."""
        unplanned_spend = self.p.annual_maintenance_mm * unplanned_share
        savings = unplanned_spend * unplanned_reduction_pct
        return {
            "unplanned_reduction_pct": f"{unplanned_reduction_pct:.0%}",
            "annual_savings_mm": round(savings, 1),
            "source": "Corrosion prediction, pigging optimization, turnaround planning"
        }

    def hse_incident_reduction(self, incident_reduction_pct: float = 0.25) -> dict:
        """Value of reduced safety incidents."""
        current_cost = self.p.annual_hse_incidents * self.p.avg_incident_cost_mm
        savings = current_cost * incident_reduction_pct
        return {
            "incident_reduction_pct": f"{incident_reduction_pct:.0%}",
            "incidents_avoided": round(self.p.annual_hse_incidents * incident_reduction_pct, 1),
            "annual_savings_mm": round(savings, 1),
            "source": "Alarm management, SIMOPS checking, emissions monitoring"
        }

    def total_roi(self, ai_investment_mm: float = 3.5) -> dict:
        """Complete ROI calculation."""
        prod = self.production_uplift_value()
        npt = self.npt_reduction_value()
        maint = self.maintenance_savings()
        hse = self.hse_incident_reduction()
        total_value = (prod["annual_revenue_mm"] + npt["annual_savings_mm"]
                       + maint["annual_savings_mm"] + hse["annual_savings_mm"])
        roi_pct = (total_value - ai_investment_mm) / ai_investment_mm * 100
        payback_months = ai_investment_mm / (total_value / 12)
        return {
            "production_uplift_mm": prod["annual_revenue_mm"],
            "npt_savings_mm": npt["annual_savings_mm"],
            "maintenance_savings_mm": maint["annual_savings_mm"],
            "hse_savings_mm": hse["annual_savings_mm"],
            "total_annual_value_mm": round(total_value, 1),
            "ai_investment_mm": ai_investment_mm,
            "net_value_mm": round(total_value - ai_investment_mm, 1),
            "roi_pct": round(roi_pct, 0),
            "payback_months": round(payback_months, 1)
        }

# Example: mid-size E&P operator
operator = OperatorProfile(
    production_boed=50000, oil_price_bbl=70,
    wells_drilled_per_year=20, avg_well_cost_mm=15,
    npt_pct=0.20, annual_maintenance_mm=30,
    annual_hse_incidents=8, avg_incident_cost_mm=2.5,
    opex_per_boe=18
)
agent = ROIAnalysisAgent(operator)
result = agent.total_roi(ai_investment_mm=3.5)
# Total annual value: ~$59.9M | ROI: ~1,611% | Payback: ~0.7 months

Getting Started: Deployment Roadmap

Rolling out AI agents across an oil and gas operation is not a single big-bang project. The most successful deployments follow a phased approach:

1. Phase 1 — Production optimization (months 1-3): Deploy artificial lift and waterflood agents on existing SCADA data. This delivers the highest ROI with the lowest integration complexity since production data is already historized.
2. Phase 2 — Drilling optimization (months 2-5): Integrate with the rig's WITS/WITSML data stream. Start with ROP prediction and stuck-pipe monitoring on a single rig, then expand fleet-wide.
3. Phase 3 — Asset integrity (months 4-8): Ingest historical ILI and inspection data, build corrosion models, and optimize the next pigging campaign. This phase has longer time-to-value but prevents catastrophic failures.
4. Phase 4 — HSE & alarm management (months 6-10): Rationalize alarm setpoints, deploy nuisance alarm filtering, and automate permit-to-work checks. Requires close collaboration with operations and safety teams.
5. Phase 5 — Exploration & subsurface (months 8-12): Train seismic interpretation models on proprietary survey data. This phase has the longest lead time but the highest per-decision impact.

Each phase builds on the data infrastructure and organizational trust established by the previous one. The key is to start where the data is most accessible and the value is most tangible — production optimization — and expand from there.