Explainable Causal Reinforcement Learning for wildfire evacuation logistics networks with inverse simulation verification
The Spark: A Learning Journey into Crisis AI
It started with a simulation that refused to behave. I was tinkering with a simple evacuation model—a grid of roads, a spreading fire, and a few thousand virtual agents trying to escape. The reinforcement learning agent I'd trained kept sending evacuees toward the fire. Not because it was broken, but because my reward function had inadvertently rewarded "movement" over "survival."
That failure was my gateway into a deeper question: how do we build AI systems that not only optimize complex logistics under extreme uncertainty but also explain why they make those decisions? And how do we verify those explanations when we can't run real-world experiments?
This article chronicles my exploration of Explainable Causal Reinforcement Learning (XC-RL) applied to the nightmarish complexity of wildfire evacuation logistics—and the inverse simulation framework I built to verify what the models claim to have learned.
Why Wildfire Evacuation is the Perfect AI Stress Test
Wildfire evacuation is a "wicked problem" for AI. It combines:
- Non-stationary dynamics: Fire fronts move unpredictably, wind shifts constantly
- Strict temporal constraints: Every second of delay compounds risk exponentially
- Resource scarcity: Limited vehicles, fuel, road capacity, and personnel
- Human irrationality: Panic, hesitation, and herd behavior defy classical models
- Causal complexity: Correlation ≠ causation in life-or-death scenarios
Traditional reinforcement learning approaches—like DQN or PPO—can optimize routes, but they operate as opaque black boxes. When an agent reroutes a bus convoy through a smoke-filled canyon, we need to know why. Is it because the canyon is genuinely safer, or because the agent learned a spurious correlation with "shorter distance"?
During my research of causal inference in RL, I discovered that the standard toolkit (SHAP values, LIME, attention maps) provides attributions, not explanations. They tell you which inputs mattered, but not how the system would behave if those inputs were causally intervened upon. For evacuation logistics, that distinction is existential.
The Architecture: Causal RL with an Explanation Layer
My journey led me to a three-tier architecture that I believe represents the future of safety-critical RL:
- Causal World Model: A learned graph that encodes the true causal structure of evacuation dynamics (fire spread, road closure cascades, population movement)
- Policy Network: A PPO-based agent that acts through the causal model rather than directly from raw observations
- Inverse Simulation Verifier: A separate system that runs "counterfactual rollouts" to test whether the policy's stated reasons for actions match actual behavior
Let me walk you through each component as I built it.
Tier 1: The Causal World Model
The first challenge was representing the environment causally. I moved beyond simple state-action pairs to a Structural Causal Model (SCM) with a graph structure G = (V, E), where nodes represent:
- Fire intensity at location
(x, y)at timet - Road capacity (dynamic, affected by debris and smoke)
- Population density in each evacuation zone
- Available vehicle resources
- Wind vector and humidity
Edges represent causal relationships. For instance, Fire(x,y,t) → RoadCapacity(x,y,t+1) captures the causal link between fire proximity and road usability.
import networkx as nx
import numpy as np
from typing import Dict, Tuple
class CausalEvacuationModel:
def __init__(self, road_network: nx.Graph, fire_sources: list):
self.road_network = road_network
self.causal_graph = nx.DiGraph()
self._build_causal_structure(fire_sources)
def _build_causal_structure(self, fire_sources):
# Add fire nodes
for i, source in enumerate(fire_sources):
self.causal_graph.add_node(f"fire_{i}",
type="fire",
pos=source,
intensity=0.8)
# Add road nodes with causal edges from fire
for (u, v, data) in self.road_network.edges(data=True):
node_id = f"road_{u}_{v}"
self.causal_graph.add_node(node_id, type="road", capacity=data['capacity'])
# Causal edge: fire → road (fire reduces capacity)
for i, source in enumerate(fire_sources):
dist = np.linalg.norm(np.array(source) - np.array(data['midpoint']))
if dist < 5000: # 5km causal radius
self.causal_graph.add_edge(f"fire_{i}", node_id,
effect="capacity_reduction",
strength=1.0 / (1.0 + dist/1000))
def causal_intervention(self, node: str, value: float) -> Dict[str, float]:
"""Do-calculus style intervention: set node value and propagate."""
# This is where we use Pearl's do-operator
intervened = {}
intervened[node] = value
# Propagate through causal graph (simplified linear approximation)
for descendant in nx.descendants(self.causal_graph, node):
# Apply causal effect along edges
parents = list(self.causal_graph.predecessors(descendant))
effect = sum(self.causal_graph[parent][descendant]['strength'] *
intervened.get(parent, 0.0)
for parent in parents if parent in intervened)
intervened[descendant] = effect
return intervened
Learning Insight: The critical realization here was that standard do-calculus requires knowing the full causal graph a priori. In real wildfire scenarios, we don't. I had to implement a causal discovery layer using PC algorithm variants on historical fire data, then refine with expert knowledge from fire engineers.
Tier 2: Policy Optimization Through the Causal Lens
The policy network doesn't see raw pixels or sensor readings. Instead, it receives the causal embeddings—representations of the current state that have been transformed through the causal graph. This forces the agent to reason about causes rather than correlations.
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
class CausalPolicyNetwork(nn.Module):
def __init__(self, causal_dim: int, action_dim: int, hidden_dim: int = 128):
super().__init__()
self.causal_encoder = nn.Sequential(
nn.Linear(causal_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU()
)
# Separate heads for action and explanation
self.action_head = nn.Linear(hidden_dim, action_dim)
self.explanation_head = nn.Linear(hidden_dim, causal_dim) # Predicts causal factors
def forward(self, causal_state: torch.Tensor):
encoded = self.causal_encoder(causal_state)
action_logits = self.action_head(encoded)
# The explanation head predicts which causal factors drove this decision
# This is our "explainability" hook
explanation_weights = torch.softmax(self.explanation_head(encoded), dim=-1)
return action_logits, explanation_weights
class CausalPPOAgent:
def __init__(self, causal_model: CausalEvacuationModel):
self.causal_model = causal_model
self.policy = CausalPolicyNetwork(
causal_dim=causal_model.causal_graph.number_of_nodes(),
action_dim=4 # [north, south, east, west] routing decisions
)
self.optimizer = optim.Adam(self.policy.parameters(), lr=3e-4)
def select_action(self, state: np.ndarray, explain: bool = False):
# Transform raw state through causal model
causal_state = self._to_causal_embedding(state)
causal_tensor = torch.FloatTensor(causal_state).unsqueeze(0)
action_logits, explanation_weights = self.policy(causal_tensor)
if explain:
# Return both action and causal explanation
action = Categorical(logits=action_logits).sample()
explanation = explanation_weights.squeeze().detach().numpy()
# Map explanation weights back to causal graph nodes
node_names = list(self.causal_model.causal_graph.nodes())
explanation_dict = {node_names[i]: float(explanation[i])
for i in range(len(node_names))}
return action.item(), explanation_dict
return Categorical(logits=action_logits).sample().item()
Key Discovery: During my experimentation with this architecture, I found that forcing the policy to output explanation weights before the action (as a kind of "reasoning scratchpad") dramatically improved both performance and interpretability. The agent couldn't "cheat" by making decisions without committing to causal reasons.
Tier 3: Inverse Simulation Verification
This is the part I'm most proud of. The problem with RL explanations is that they're post-hoc—the agent acts, then we try to explain. But what if we could verify the explanation by running the policy backward?
Inverse simulation works like this:
- Take a policy decision: "Route Bus 7 through Canyon Road because fire intensity at Ridge Road exceeds threshold"
- Run a counterfactual simulation where we intervene on the causal model: set fire intensity at Ridge Road to below threshold
- Check if the policy changes its decision
- If yes → the explanation is causally valid
- If no → the explanation is spurious
import simpy
from typing import List, Dict, Any
class InverseSimulationVerifier:
def __init__(self, causal_model: CausalEvacuationModel,
policy_agent: CausalPPOAgent):
self.causal_model = causal_model
self.agent = policy_agent
def verify_explanation(self, original_state: Dict,
original_action: int,
explanation: Dict[str, float],
num_counterfactuals: int = 100) -> Dict[str, float]:
"""
Verify if the explanation is causally valid by running
inverse simulations with counterfactual interventions.
"""
verification_results = {}
# For each causal factor in the explanation
for factor, weight in explanation.items():
if weight < 0.1: # Skip negligible factors
continue
# Run counterfactual: intervene on this factor
intervention_value = original_state[factor] * 0.5 # Halve it
# Run multiple counterfactual simulations
action_changes = 0
total_sims = 0
for _ in range(num_counterfactuals):
# Create counterfactual state
cf_state = self.causal_model.causal_intervention(factor, intervention_value)
# Get policy action under counterfactual
cf_action = self.agent.select_action(cf_state)
if cf_action != original_action:
action_changes += 1
total_sims += 1
# Causal validity score: how often does changing this factor
# actually change the policy's decision?
causal_validity = action_changes / total_sims
verification_results[factor] = {
'causal_validity': causal_validity,
'explained_weight': weight,
'verified': causal_validity > 0.7 # Threshold
}
return verification_results
def generate_verified_explanation(self, state: Dict,
action: int) -> Dict[str, Any]:
"""Generate and verify explanations in one pass."""
# Get initial explanation from policy
_, raw_explanation = self.agent.select_action(state, explain=True)
# Verify through inverse simulation
verified = self.verify_explanation(state, action, raw_explanation)
# Filter to only verified causal factors
verified_factors = {
factor: data for factor, data in verified.items()
if data['verified']
}
# Build human-readable explanation
explanation_text = self._format_explanation(verified_factors)
return {
'action': action,
'verified_causal_factors': verified_factors,
'explanation_text': explanation_text,
'confidence': np.mean([d['causal_validity'] for d in verified_factors.values()])
}
During my investigation of inverse verification, I discovered something counterintuitive: the most confident explanations (high weight in the policy's explanation head) were often the least causally valid. The policy had learned to "explain" decisions using salient but non-causal features—like smoke color instead of fire proximity. The inverse simulation caught these spurious explanations and forced the policy to rely on true causal drivers.
Real-World Implementation: A Case Study
Let me show you a complete implementation for a simplified but realistic scenario: evacuating a small town with two evacuation routes, one mountain road and one coastal road.
python
import numpy as np
import matplotlib.pyplot as plt
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class EvacuationScenario:
population: int = 5000
vehicles: int = 1500
road_capacity: List[int] = None # vehicles per minute
fire_speed: float = 0.5 # km per minute
wind_direction: float = 45 # degrees
def __post_init__(self):
if self.road_capacity is None:
self.road_capacity = [30, 25] # mountain, coastal
class WildfireEvacuationSystem:
def __init__(self, scenario: EvacuationScenario):
self.scenario = scenario
self.causal_model = self._build_causal_model()
self.agent = CausalPPOAgent(self.causal_model)
self.verifier = InverseSimulationVerifier(self.causal_model, self.agent)
def _build_causal_model(self) -> CausalEvacuationModel:
# Build road network graph
road_network = nx.Graph()
road_network.add_edge("town", "mountain_pass",
capacity=self.scenario.road_capacity[0],
midpoint=(10, 20))
road_network.add_edge("town", "coastal_route",
capacity=self.scenario.road_capacity[1],
midpoint=(5, 15))
road_network.add_edge("mountain_pass", "safe_zone",
capacity=self.scenario.road_capacity[0],
midpoint=(20, 25))
road_network.add_edge("coastal_route", "safe_zone",
capacity=self.scenario.road_capacity[1],
midpoint=(10, 10))
# Fire source near coastal route
fire_sources = [(8, 12)]
return CausalEvacuationModel(road_network, fire_sources)
def run_evacuation(self, timesteps: int = 100) -> Dict[str, Any]:
"""Run the full evacuation with explainable decisions."""
state = self._initialize_state()
decisions_log = []
for t in range(timesteps):
# Agent makes decision with explanation
action, explanation = self.agent.select_action(state, explain=True)
# Verify the explanation through inverse simulation
verified = self.verifier.verify_explanation(state, action, explanation)
# Log decision with verification
decisions_log.append({
'timestep': t,
'action': action,
'explanation': verified,
'state': state.copy()
})
# Update state based on action and fire dynamics
state = self._update_state(state, action, t)
# Emergency stop if fire reaches town
if state['fire_proximity'] < 1.0:
print(f"EVACUATION COMPLETE at timestep {t}")
break
return {
'decisions': decisions_log,
'total_evacuated': state['evacuated'],
'casualties': state['population'] - state['evacuated']
}
def _update_state(self, state: Dict, action: int, t: int) -> Dict:
"""Update evacuation state based on action and fire dynamics."""
# Action 0: use mountain route (safer but slower)
# Action 1: use coastal route (faster but fire risk)
# Action 2: hold position
# Action 3: split evacuation
new_state = state.copy()
fire_advance = self.scenario.fire_speed * np.cos(np.radians(self.scenario.wind_direction))
# Update fire position
new_state['fire_proximity'] -= fire_advance
if action == 0:
evacuated = min(state['road_capacity'][0], state['population'])
new_state['evacuated'] += evacuated
new_state['population'] -= evacuated
elif action == 1:
# Coastal route is faster but riskier
risk_factor = 1.0 / (state['fire_proximity'] + 0.1)
evacuated = min(state['road_capacity'][1] * risk_factor, state['population'])
new_state['evacuated'] += evacuated
new_state['population'] -= evacuated
# Fire risk causes casualties
casualties = int(evacuated * 0.1 * risk_factor)
new_state['evacuated'] -= casualties
new_state['casualties'] += casualties
elif action == 3:
# Split evacuation
mountain_cap = state['road_capacity'][0] * 0.6
coastal_cap = state['road_capacity'][
Top comments (0)