reinforcement-learning-supply-chain

供应链强化学习

你是应用强化学习解决供应链序贯决策问题的专家。你的目标是帮助设计、训练和部署 RL Agent，使其通过与环境交互，学习库存控制、定价、路径规划和资源分配的最优策略。

初步评估

1. 问题类型：是否涉及序贯决策？（库存订购、价格调整、路径规划）
2. 状态空间：有哪些可用信息？（库存水平、需求、价格）
3. 动作空间：有哪些决策？（订货量、价格、路径）
4. 奖励函数：如何衡量表现？（利润、服务水平、成本）
5. 环境：是否有模拟器或真实系统？

---

RL 基础知识

马尔可夫决策过程（MDP）：

• 状态（S）：系统状况
• 动作（A）：可选决策
• 转移（P）：状态动态
• 奖励（R）：即时反馈
• 策略（π）：状态 → 动作映射

**目标：**学习使预期累积奖励最大化的策略 π

---

用于库存控制的 Q-Learning

import numpy as np
import matplotlib.pyplot as plt
from collections import defaultdict

class InventoryEnvironment:
    """
    Inventory control environment
    
    State: current inventory level
    Action: order quantity
    Reward: -holding_cost - backorder_cost + revenue
    """
    
    def __init__(self,
                 max_inventory=50,
                 holding_cost=1.0,
                 backorder_cost=10.0,
                 order_cost=2.0,
                 price=15.0):
        
        self.max_inventory = max_inventory
        self.h_cost = holding_cost
        self.b_cost = backorder_cost
        self.o_cost = order_cost
        self.price = price
        
        # Demand distribution (Poisson)
        self.mean_demand = 10
        
        self.state = 20  # Initial inventory
    
    def reset(self):
        """Reset environment"""
        self.state = 20
        return self.state
    
    def step(self, action):
        """
        Take action (order quantity), observe demand, get reward
        
        Returns: next_state, reward, done
        """
        
        # Order arrives
        inventory_after_order = min(self.state + action, self.max_inventory)
        
        # Demand occurs (stochastic)
        demand = np.random.poisson(self.mean_demand)
        
        # Satisfy demand
        sales = min(inventory_after_order, demand)
        backorder = max(0, demand - inventory_after_order)
        
        next_inventory = inventory_after_order - sales
        
        # Calculate reward
        revenue = self.price * sales
        holding = self.h_cost * next_inventory
        backorder_penalty = self.b_cost * backorder
        ordering = self.o_cost * action
        
        reward = revenue - holding - backorder_penalty - ordering
        
        self.state = next_inventory
        done = False
        
        return next_inventory, reward, done


class QLearningAgent:
    """
    Q-Learning agent for inventory control
    """
    
    def __init__(self,
                 state_space,
                 action_space,
                 learning_rate=0.1,
                 discount_factor=0.95,
                 epsilon=0.1):
        
        self.states = state_space
        self.actions = action_space
        self.lr = learning_rate
        self.gamma = discount_factor
        self.epsilon = epsilon
        
        # Q-table: Q(s, a)
        self.Q = defaultdict(lambda: defaultdict(float))
    
    def select_action(self, state):
        """
        Epsilon-greedy action selection
        """
        
        if np.random.random() < self.epsilon:
            # Explore: random action
            return np.random.choice(self.actions)
        else:
            # Exploit: best action
            q_values = [self.Q[state][a] for a in self.actions]
            best_action = self.actions[np.argmax(q_values)]
            return best_action
    
    def update(self, state, action, reward, next_state):
        """
        Q-learning update rule
        
        Q(s,a) ← Q(s,a) + α[r + γ max_a' Q(s',a') - Q(s,a)]
        """
        
        # Current Q-value
        current_q = self.Q[state][action]
        
        # Best Q-value for next state
        next_q_values = [self.Q[next_state][a] for a in self.actions]
        max_next_q = max(next_q_values)
        
        # TD target
        target = reward + self.gamma * max_next_q
        
        # Update
        self.Q[state][action] = current_q + self.lr * (target - current_q)
    
    def get_policy(self):
        """Extract greedy policy from Q-values"""
        
        policy = {}
        for state in self.states:
            q_values = [self.Q[state][a] for a in self.actions]
            best_action = self.actions[np.argmax(q_values)]
            policy[state] = best_action
        
        return policy


# Training
env = InventoryEnvironment()
agent = QLearningAgent(
    state_space=list(range(51)),
    action_space=list(range(21)),  # Order 0-20 units
    learning_rate=0.1,
    discount_factor=0.95,
    epsilon=0.1
)

n_episodes = 10000
episode_rewards = []

for episode in range(n_episodes):
    state = env.reset()
    total_reward = 0
    
    for t in range(30):  # 30-day horizon
        action = agent.select_action(state)
        next_state, reward, done = env.step(action)
        
        agent.update(state, action, reward, next_state)
        
        total_reward += reward
        state = next_state
        
        if done:
            break
    
    episode_rewards.append(total_reward)
    
    if (episode + 1) % 1000 == 0:
        avg_reward = np.mean(episode_rewards[-100:])
        print(f"Episode {episode+1}: Avg Reward = {avg_reward:.2f}")

# Extract learned policy
policy = agent.get_policy()

print("\nLearned Policy (Inventory → Order Quantity):")
for inventory in range(0, 51, 5):
    order = policy.get(inventory, 0)
    print(f"  Inventory {inventory}: Order {order}")

---

用于复杂状态的深度 Q 网络（DQN）

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from collections import deque
import random

class DQN(nn.Module):
    """
    Deep Q-Network
    """
    
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(DQN, self).__init__()
        
        self.network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )
    
    def forward(self, state):
        return self.network(state)


class DQNAgent:
    """
    DQN Agent with Experience Replay and Target Network
    """
    
    def __init__(self, state_dim, action_dim, lr=0.001, gamma=0.99):
        
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.gamma = gamma
        
        # Main network
        self.q_network = DQN(state_dim, action_dim)
        
        # Target network
        self.target_network = DQN(state_dim, action_dim)
        self.target_network.load_state_dict(self.q_network.state_dict())
        
        self.optimizer = optim.Adam(self.q_network.parameters(), lr=lr)
        self.loss_fn = nn.MSELoss()
        
        # Experience replay buffer
        self.memory = deque(maxlen=10000)
        self.batch_size = 64
    
    def select_action(self, state, epsilon=0.1):
        """Epsilon-greedy action selection"""
        
        if random.random() < epsilon:
            return random.randint(0, self.action_dim - 1)
        
        with torch.no_grad():
            state_tensor = torch.FloatTensor(state).unsqueeze(0)
            q_values = self.q_network(state_tensor)
            return q_values.argmax().item()
    
    def store_transition(self, state, action, reward, next_state, done):
        """Store experience in replay buffer"""
        self.memory.append((state, action, reward, next_state, done))
    
    def train(self):
        """Train on mini-batch from replay buffer"""
        
        if len(self.memory) < self.batch_size:
            return
        
        # Sample mini-batch
        batch = random.sample(self.memory, self.batch_size)
        
        states = torch.FloatTensor([t[0] for t in batch])
        actions = torch.LongTensor([t[1] for t in batch])
        rewards = torch.FloatTensor([t[2] for t in batch])
        next_states = torch.FloatTensor([t[3] for t in batch])
        dones = torch.FloatTensor([t[4] for t in batch])
        
        # Current Q-values
        q_values = self.q_network(states).gather(1, actions.unsqueeze(1))
        
        # Target Q-values
        with torch.no_grad():
            next_q_values = self.target_network(next_states).max(1)[0]
            targets = rewards + self.gamma * next_q_values * (1 - dones)
        
        # Loss and update
        loss = self.loss_fn(q_values.squeeze(), targets)
        
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
    
    def update_target_network(self):
        """Copy weights from main network to target network"""
        self.target_network.load_state_dict(self.q_network.state_dict())

---

用于连续动作的策略梯度

class PolicyNetwork(nn.Module):
    """
    Policy network for continuous actions
    """
    
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(PolicyNetwork, self).__init__()
        
        self.network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.Tanh()
        )
        
        # Mean and std for Gaussian policy
        self.mean_layer = nn.Linear(hidden_dim, action_dim)
        self.log_std_layer = nn.Linear(hidden_dim, action_dim)
    
    def forward(self, state):
        features = self.network(state)
        mean = self.mean_layer(features)
        log_std = self.log_std_layer(features)
        std = torch.exp(log_std)
        
        return mean, std


class PolicyGradientAgent:
    """
    REINFORCE algorithm for policy gradient
    """
    
    def __init__(self, state_dim, action_dim, lr=0.001, gamma=0.99):
        
        self.policy = PolicyNetwork(state_dim, action_dim)
        self.optimizer = optim.Adam(self.policy.parameters(), lr=lr)
        self.gamma = gamma
        
        self.saved_log_probs = []
        self.rewards = []
    
    def select_action(self, state):
        """Sample action from policy"""
        
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        mean, std = self.policy(state_tensor)
        
        # Sample from Gaussian
        dist = torch.distributions.Normal(mean, std)
        action = dist.sample()
        log_prob = dist.log_prob(action).sum()
        
        self.saved_log_probs.append(log_prob)
        
        return action.numpy()[0]
    
    def update(self):
        """Update policy using REINFORCE"""
        
        # Calculate returns
        returns = []
        R = 0
        
        for r in reversed(self.rewards):
            R = r + self.gamma * R
            returns.insert(0, R)
        
        returns = torch.tensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + 1e-9)
        
        # Policy gradient
        policy_loss = []
        for log_prob, R in zip(self.saved_log_probs, returns):
            policy_loss.append(-log_prob * R)
        
        policy_loss = torch.stack(policy_loss).sum()
        
        # Update
        self.optimizer.zero_grad()
        policy_loss.backward()
        self.optimizer.step()
        
        # Clear buffers
        self.saved_log_probs = []
        self.rewards = []

---

应用

1. 动态定价

# State: inventory, time, competitor prices, demand signals
# Action: price adjustment
# Reward: revenue - costs

2. 仓库机器人控制

# State: robot position, item locations, orders
# Action: movement and pick decisions
# Reward: -time - collisions + items picked

3. 供应链网络优化

# State: inventory at all nodes, pipeline inventory, demand
# Action: shipment quantities between nodes
# Reward: -costs + service level bonuses

4. 订单履约

# State: orders, inventory, capacity, time
# Action: order-to-warehouse assignment
# Reward: -shipping cost - delay penalties

---

工具与库

Python RL：

• stable-baselines3：RL 算法
• Ray RLlib：分布式 RL
• TensorFlow Agents：基于 TF 的 RL
• PyTorch：自定义实现

仿真：

• SimPy：离散事件仿真
• Gym：RL 环境
• 自定义模拟器

---

相关技能

• optimization-modeling：传统优化
• optimization-ml-hybrid：RL + 优化
• **dynamic-pricing`：定价应用
• inventory-optimization：库存控制
• **route-optimization`：结合 RL 的 VRP