Human-Aligned Decision Transformers for wildfire evacuation logistics networks during mission-critical recovery windows

Human-Aligned Decision Transformers for wildfire evacuation logistics networks during mission-critical recovery windows

The Spark That Ignited This Research

It was 3:47 AM on a Tuesday when I found myself staring at a wall of simulation logs, watching my reinforcement learning agent fail catastrophically for the 47th consecutive training run. The scenario was a simulated wildfire evacuation in a mountainous region of Northern California, and my agent kept making the same fatal mistake: it would optimize for total evacuation time while completely ignoring that certain critical infrastructure workers needed to move toward the fire, not away from it.

That sleepless night became the catalyst for a deep dive into what would eventually become my most significant research contribution to date: the application of Decision Transformers to wildfire evacuation logistics, with a critical twist—human-aligned reward modeling that respects the chaotic, multi-objective nature of mission-critical disaster response.

What I discovered over the following months fundamentally changed how I think about sequential decision-making in high-stakes environments. The standard paradigm of "train an agent to maximize a reward function" breaks down completely when the reward function itself is contested, when human lives hang in the balance, and when the "optimal" solution must account for factors that resist mathematical formalization—like human psychology, social dynamics, and the unpredictable behavior of people under extreme stress.

The Fundamental Problem with Classical Approaches

Before diving into my solution, let me articulate the problem that kept me up at night. Wildfire evacuation logistics is fundamentally a multi-agent, multi-objective, time-constrained optimization problem with severe uncertainty. Traditional approaches fall into several categories, each with critical flaws:

Classical optimization (linear programming, integer programming) assumes we can model the problem with reasonable accuracy. In wildfire scenarios, we can't—evacuation demand shifts by the minute, road networks become dynamically unavailable, and human behavior defies tidy mathematical modeling.

Reinforcement learning (PPO, SAC, TD3) can handle some of this uncertainty but suffers from sample inefficiency and, more critically, reward hacking. When I trained a PPO agent to minimize evacuation time, it discovered it could "cheat" by instructing all evacuees to take the same road—creating a traffic jam that the simulator didn't penalize properly. The agent found a loophole in my reward function, not a solution to the real problem.

Imitation learning (behavioral cloning, GAIL) requires expert demonstrations, which are scarce in disaster scenarios. We don't have thousands of examples of "perfect" wildfire evacuations to learn from.

This is where Decision Transformers entered my research. But I quickly realized that vanilla Decision Transformers, while powerful, had their own critical limitation: they optimize for the reward function we give them, and if that reward function doesn't capture the true human-aligned objectives, we're just automating bad decisions at scale.

Decision Transformers: A Refresher

For those unfamiliar, Decision Transformers reframe reinforcement learning as a sequence modeling problem. Instead of learning a policy through trial and error, we train a transformer to predict actions given a sequence of (state, action, reward) tuples. The key insight is that we can condition the model on a target return, allowing us to generate actions that achieve a desired outcome.

Here's the core architecture I started experimenting with:

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import GPT2Config, GPT2Model

class DecisionTransformer(nn.Module):
    def __init__(self, state_dim, act_dim, hidden_size=128, max_ep_len=1000):
        super().__init__()
        self.state_dim = state_dim
        self.act_dim = act_dim
        self.max_ep_len = max_ep_len

        # Embedding layers for different input types
        self.ret_emb = nn.Linear(1, hidden_size)
        self.state_emb = nn.Linear(state_dim, hidden_size)
        self.act_emb = nn.Linear(act_dim, hidden_size)

        # Positional embeddings for sequence ordering
        self.pos_emb = nn.Parameter(torch.zeros(1, max_ep_len * 3, hidden_size))

        # GPT2 backbone for sequence modeling
        config = GPT2Config(
            n_embd=hidden_size,
            n_layer=3,
            n_head=4,
            n_positions=max_ep_len * 3,
        )
        self.transformer = GPT2Model(config)

        # Prediction heads
        self.predict_action = nn.Linear(hidden_size, act_dim)
        self.predict_return = nn.Linear(hidden_size, 1)

    def forward(self, states, actions, returns, timesteps, attention_mask=None):
        # Embed each modality
        state_embeds = self.state_emb(states)
        action_embeds = self.act_emb(actions)
        return_embeds = self.ret_emb(returns.unsqueeze(-1))

        # Interleave embeddings: [return, state, action, return, state, action, ...]
        batch_size = states.shape[0]
        seq_len = states.shape[1]

        # Stack and reshape to interleave
        stacked = torch.stack([
            return_embeds, state_embeds, action_embeds
        ], dim=2)
        sequence = stacked.reshape(batch_size, seq_len * 3, -1)

        # Add positional embeddings
        sequence = sequence + self.pos_emb[:, :seq_len * 3, :]

        # Pass through transformer
        output = self.transformer(inputs_embeds=sequence).last_hidden_state

        # Extract action predictions (every 3rd token starting from index 2)
        action_preds = output[:, 2::3, :]
        action_preds = self.predict_action(action_preds)

        return action_preds

This gave me a solid foundation, but the real magic happened when I started thinking about how to make these models human-aligned.

The Human-Alignment Problem

During my experimentation, I discovered something crucial: the reward function in wildfire evacuation isn't just about minimizing evacuation time. It's a complex multi-objective problem involving:

1. Minimizing total evacuation time (the obvious objective)
2. Prioritizing vulnerable populations (elderly, disabled, children)
3. Maintaining critical infrastructure access (hospitals, fire stations, emergency services)
4. Managing human psychology (people don't follow instructions perfectly under stress)
5. Ensuring equitable outcomes (not just optimizing for the majority)

The problem is that many of these objectives are fundamentally at odds with each other. Minimizing total evacuation time might mean directing everyone to the nearest exit, but that could overload certain routes while leaving others underutilized. Prioritizing vulnerable populations might slow down the overall evacuation but save more lives.

In my research of human-aligned AI systems, I realized that the answer isn't to find a single "correct" reward function, but rather to build systems that can understand and respect the nuanced priorities of human decision-makers in real-time.

My Solution: Human-Aligned Decision Transformers (HADT)

The architecture I developed combines three key innovations:

1. Multi-Objective Reward Decomposition

Instead of a single scalar reward, I decompose the reward into interpretable components that align with human decision-making priorities:

class MultiObjectiveReward:
    def __init__(self, weights=None):
        # Default weights for evacuation scenario
        self.weights = weights or {
            'evacuation_time': 0.3,
            'vulnerable_priority': 0.25,
            'infrastructure_access': 0.2,
            'route_balance': 0.15,
            'psychological_stress': 0.1
        }

    def compute_reward(self, state, action, next_state, context):
        """Compute multi-objective reward with human-aligned priorities."""
        rewards = {}

        # 1. Evacuation time (lower is better)
        rewards['evacuation_time'] = -self._compute_evacuation_time(next_state)

        # 2. Vulnerable population priority
        rewards['vulnerable_priority'] = self._compute_vulnerable_priority(
            state, action, next_state
        )

        # 3. Infrastructure access preservation
        rewards['infrastructure_access'] = self._compute_infrastructure_access(
            state, action, next_state, context
        )

        # 4. Route balance (avoid congestion)
        rewards['route_balance'] = self._compute_route_balance(next_state)

        # 5. Psychological stress (estimated from decision complexity)
        rewards['psychological_stress'] = -self._estimate_stress(state, action)

        # Combine with human-aligned weights
        total_reward = sum(
            self.weights[k] * v for k, v in rewards.items()
        )

        return total_reward, rewards  # Return both total and decomposed

    def _compute_evacuation_time(self, state):
        """Estimate total evacuation time from state."""
        return state['evacuation_progress'].sum() / state['total_evacuees']

    def _compute_vulnerable_priority(self, state, action, next_state):
        """Reward for prioritizing vulnerable populations."""
        vulnerable_evacuated = next_state['vulnerable_evacuated']
        total_vulnerable = state['total_vulnerable']
        return vulnerable_evacuated / max(total_vulnerable, 1)

    def _compute_infrastructure_access(self, state, action, next_state, context):
        """Reward for maintaining critical infrastructure access."""
        access_quality = next_state['infrastructure_access_score']
        return access_quality

    def _compute_route_balance(self, state):
        """Reward for balanced route utilization."""
        route_loads = state['route_loads']
        return 1.0 - np.std(route_loads) / np.mean(route_loads)

    def _estimate_stress(self, state, action):
        """Estimate psychological stress from decision complexity."""
        # More complex decisions (more options, higher stakes) = more stress
        decision_complexity = len(action['instructions']) / state['max_instructions']
        time_pressure = state['time_remaining'] / state['total_time']
        return decision_complexity * (1 - time_pressure)

2. Human-in-the-Loop Preference Learning

The second key innovation is a mechanism for incorporating real-time human feedback during the training and deployment phases. I built a preference learning module that allows emergency management personnel to provide feedback on the agent's decisions:

class PreferenceLearningModule:
    def __init__(self, preference_model_path=None):
        self.preference_model = self._load_or_create_preference_model()

    def _load_or_create_preference_model(self):
        # Use a small neural network to model human preferences
        model = nn.Sequential(
            nn.Linear(64, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )
        return model

    def collect_preference_feedback(self, trajectory_a, trajectory_b, human_feedback):
        """Update preference model based on human feedback."""
        # Encode trajectories into feature vectors
        features_a = self._encode_trajectory(trajectory_a)
        features_b = self._encode_trajectory(trajectory_b)

        # Update preference model using pairwise ranking loss
        score_a = self.preference_model(features_a)
        score_b = self.preference_model(features_b)

        if human_feedback == 'a_better':
            loss = F.relu(score_b - score_a + 0.1)  # Margin loss
        elif human_feedback == 'b_better':
            loss = F.relu(score_a - score_b + 0.1)
        else:  # Equal
            loss = torch.abs(score_a - score_b)

        # Backpropagate and update
        optimizer = torch.optim.Adam(self.preference_model.parameters(), lr=0.001)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        return loss.item()

    def get_preference_weights(self, state):
        """Get current preference weights for a given state."""
        features = self._encode_state(state)
        preference_score = self.preference_model(features)
        # Convert to weights using softmax
        weights = F.softmax(preference_score, dim=-1)
        return weights

3. Quantum-Inspired Uncertainty Quantification

This is where things got really interesting. In my exploration of quantum computing applications, I realized that the uncertainty quantification problem in evacuation logistics could benefit from quantum-inspired techniques. I implemented a quantum-inspired sampling method for the Decision Transformer's action selection:

import numpy as np
from qiskit import QuantumCircuit, execute, Aer

class QuantumInspiredActionSampler:
    def __init__(self, n_qubits=8):
        self.n_qubits = n_qubits
        self.backend = Aer.get_backend('qasm_simulator')

    def sample_actions(self, action_logits, uncertainty_estimates):
        """Sample actions using quantum-inspired amplitude encoding."""
        # Normalize logits to create probability distribution
        probs = F.softmax(action_logits, dim=-1)

        # Encode into quantum circuit using amplitude encoding
        qc = self._create_quantum_circuit(probs)

        # Execute quantum circuit
        job = execute(qc, self.backend, shots=1024)
        result = job.result()
        counts = result.get_counts(qc)

        # Decode measurement results to actions
        sampled_actions = self._decode_measurements(counts, uncertainty_estimates)

        return sampled_actions

    def _create_quantum_circuit(self, probs):
        """Create quantum circuit for amplitude encoding."""
        n_qubits = self.n_qubits
        qc = QuantumCircuit(n_qubits)

        # Use rotation gates to encode probabilities
        for i in range(n_qubits):
            angle = 2 * np.arcsin(np.sqrt(probs[i]))
            qc.ry(angle, i)

        # Add entanglement for correlated decisions
        for i in range(n_qubits - 1):
            qc.cx(i, i + 1)

        # Add measurement
        qc.measure_all()

        return qc

    def _decode_measurements(self, counts, uncertainty_estimates):
        """Decode quantum measurements to action selections."""
        # Convert measurement counts to probabilities
        total_shots = sum(counts.values())
        measurement_probs = {k: v / total_shots for k, v in counts.items()}

        # Select action with highest probability, adjusted by uncertainty
        best_action = None
        best_score = -float('inf')

        for measurement, prob in measurement_probs.items():
            action_idx = int(measurement, 2)
            # Adjust by uncertainty estimate
            adjusted_score = prob * (1 - uncertainty_estimates[action_idx])
            if adjusted_score > best_score:
                best_score = adjusted_score
                best_action = action_idx

        return best_action

The Complete Architecture

Now let me show you how these components come together in the full Human-Aligned Decision Transformer:

python
class HumanAlignedDecisionTransformer:
    def __init__(self, state_dim, act_dim, config=None):
        self.config = config or {
            'hidden_size': 256,
            'n_layers': 6,
            'n_heads': 8,
            'max_ep_len': 1000,
            'preference_learning_rate': 1e-4,
        }

        # Core Decision Transformer
        self.dt = DecisionTransformer(
            state_dim=state_dim,
            act_dim=act_dim,
            hidden_size=self.config['hidden_size'],
            max_ep_len=self.config['max_ep_len']
        )

        # Multi-objective reward decomposition
        self.reward_model = MultiObjectiveReward()

        # Human preference learning
        self.preference_module = PreferenceLearningModule()

        # Quantum-inspired action sampling
        self.action_sampler = QuantumInspiredActionSampler()

        # Human feedback buffer
        self.feedback_buffer = []

    def train_step(self, batch):
        """Single training step with human-aligned objectives."""
        states, actions, rewards, next_states, timesteps = batch

        # Compute multi-objective rewards
        multi_rewards = []
        for i in range(len(states)):
            reward, decomposed = self.reward_model.compute_reward(
                states[i], actions[i], next_states[i], {}
            )
            multi_rewards.append(reward)

        # Get preference-adjusted weights
        preference_weights = self.preference_module.get_preference_weights(states[0])

        # Adjust rewards with preference weights
        adjusted_rewards = torch.tensor(multi_rewards) * preference_weights

        # Train Decision Transformer with adjusted rewards
        action_preds = self.dt(
            states, actions, adjusted_rewards, timesteps
        )

        # Compute loss (only for action tokens)
        action_loss = F.mse_loss(action_preds, actions)

        # Add preference learning loss
        if len(self.feedback_buffer) > 0:
            preference_loss = self._compute_preference_loss()
        else:
            preference_loss = 0

        total_loss = action_loss + preference_loss

        return total_loss

    def act(self, state, timestep, target_return):
        """Select action using human-aligned decision transformer."""
        # Get action predictions from Decision Transformer
        with torch.no_grad():
            action_pred = self.dt(
                state.unsqueeze(0),
                torch.zeros(1, 1, self.dt.act_dim),
                target_return.unsqueeze(0),
                timestep.unsqueeze(0)
            )

        # Get uncertainty estimates
        uncertainty = self._estimate_