Q-Learning：表格型强化学习

import numpy as np

# V 函数 vs Q 函数的数据结构对比

class ValueFunction:
    """状态值函数 V(s)：每个状态一个值"""
    def __init__(self, n_states: int):
        self.V = np.zeros(n_states)

    def get_value(self, state: int) -> float:
        return self.V[state]

    def update(self, state: int, td_error: float, lr: float = 0.1):
        """TD 更新：V(s) += alpha * delta"""
        self.V[state] += lr * td_error


class QFunction:
    """动作值函数 Q(s,a)：每个状态-动作对一个值"""
    def __init__(self, n_states: int, n_actions: int):
        self.Q = np.zeros((n_states, n_actions))

    def get_q(self, state: int, action: int) -> float:
        return self.Q[state, action]

    def best_action(self, state: int) -> int:
        """返回最优动作（贪婪选择）"""
        return int(np.argmax(self.Q[state]))

# Q 函数如何指导决策：从值到策略

def derive_policy(Q: np.ndarray) -> np.ndarray:
    """从 Q 表提取确定性贪婪策略"""
    n_states = Q.shape[0]
    policy = np.zeros(n_states, dtype=int)
    for s in range(n_states):
        policy[s] = np.argmax(Q[s])
    return policy


# 示例：3 状态 2 动作的小环境
Q_example = np.array([
    [1.5, 2.3],  # 状态 0: 动作 1 更好
    [0.8, 0.8],  # 状态 1: 两个动作等价
    [3.1, 1.2],  # 状态 2: 动作 0 更好
])
V_example = np.max(Q_example, axis=1)  # 最优策略下 V*(s) = max_a Q*(s,a)
pi_example = derive_policy(Q_example)
print(f"最优策略: {pi_example}")  # [1 0 0]
print(f"最优值函数: {V_example}")  # [2.3 0.8 3.1]

# Bellman 最优 Q 方程的直观实现

def bellman_optimal_Q(state: int, action: int, Q: np.ndarray,
                      P: dict, R: dict, gamma: float = 0.9) -> float:
    """
    Q*(s,a) = sum_s' P(s'|s,a) * [R(s,a,s') + gamma * max_a' Q*(s',a')]
    """
    q_value = 0.0
    for next_state, prob in P[state][action].items():
        reward = R[state][action].get(next_state, 0.0)
        # 从下一状态出发，取最优动作的 Q 值
        best_future_q = np.max(Q[next_state])
        q_value += prob * (reward + gamma * best_future_q)
    return q_value


def synchronous_value_update(Q: np.ndarray, P: dict, R: dict,
                              gamma: float = 0.9) -> np.ndarray:
    """同步更新：用旧的 Q 计算所有新的 Q 值（理论推导用）"""
    n_states, n_actions = Q.shape
    Q_new = np.zeros_like(Q)
    for s in range(n_states):
        for a in range(n_actions):
            Q_new[s, a] = bellman_optimal_Q(s, a, Q, P, R, gamma)
    return Q_new

# 从 Bellman 最优方程到值迭代（表格法）

def value_iteration_q(P: dict, R: dict, n_states: int, n_actions: int,
                      gamma: float = 0.9, theta: float = 1e-6) -> np.ndarray:
    """用 Bellman 最优方程迭代求解 Q*"""
    Q = np.zeros((n_states, n_actions))

    iteration = 0
    while True:
        Q_new = synchronous_value_update(Q, P, R, gamma)
        delta = np.max(np.abs(Q_new - Q))
        Q = Q_new.copy()
        iteration += 1

        if iteration % 50 == 0:
            print(f"迭代 {iteration}: max delta = {delta:.6f}")

        if delta < theta:
            print(f"收敛于迭代 {iteration}")
            break

    return Q

# Q-Learning 核心更新：一步步拆解

def q_learning_update(Q: np.ndarray, state: int, action: int,
                       reward: float, next_state: int, done: bool,
                       alpha: float = 0.1, gamma: float = 0.99) -> float:
    """
    Q(s,a) <- Q(s,a) + alpha * [r + gamma * max_a' Q(s',a') - Q(s,a)]
    """
    if done:
        # 终止状态：没有未来回报
        td_target = reward
    else:
        # 非终止：即时奖励 + 折现后的最优未来值
        td_target = reward + gamma * np.max(Q[next_state])

    # TD 误差：目标值 - 当前估计
    td_error = td_target - Q[state, action]

    # 更新 Q 表
    Q[state, action] += alpha * td_error

    return td_error

# 完整的 Q-Learning 训练框架

def q_learning(env, episodes: int = 1000, alpha: float = 0.1,
               gamma: float = 0.99, epsilon: float = 0.1) -> tuple:
    """Q-Learning 训练循环"""
    n_states = env.observation_space.n
    n_actions = env.action_space.n
    Q = np.zeros((n_states, n_actions))
    history = []

    for ep in range(episodes):
        state, _ = env.reset()
        total_reward = 0.0
        done = False
        truncated = False

        while not (done or truncated):
            # epsilon-greedy 选择动作
            if np.random.random() < epsilon:
                action = env.action_space.sample()
            else:
                action = int(np.argmax(Q[state]))

            # 执行动作，获取反馈
            next_state, reward, done, truncated, _ = env.step(action)

            # Q-Learning 更新（离线策略：用 max Q 做目标）
            if done:
                td_target = reward
            else:
                td_target = reward + gamma * np.max(Q[next_state])

            td_error = td_target - Q[state, action]
            Q[state, action] += alpha * td_error

            state = next_state
            total_reward += reward

        history.append(total_reward)

    return Q, history

# epsilon-greedy 策略的完整实现

class EpsilonGreedyPolicy:
    """带衰减的 epsilon-greedy 策略"""

    def __init__(self, n_actions: int, epsilon_start: float = 1.0,
                 epsilon_end: float = 0.01, decay_rate: float = 0.995):
        self.n_actions = n_actions
        self.epsilon = epsilon_start
        self.epsilon_start = epsilon_start
        self.epsilon_end = epsilon_end
        self.decay_rate = decay_rate

    def choose_action(self, Q: np.ndarray, state: int) -> int:
        """根据当前 Q 表和 epsilon 选择动作"""
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)  # 探索
        return int(np.argmax(Q[state]))  # 利用

    def decay(self):
        """衰减 epsilon"""
        self.epsilon = max(self.epsilon_end,
                          self.epsilon * self.decay_rate)

# 不同探索策略对比

class SoftmaxPolicy:
    """Softmax/Boltzmann 探索：按 Q 值的概率分布选择动作"""

    def __init__(self, n_actions: int, temperature: float = 1.0):
        self.n_actions = n_actions
        self.temperature = temperature

    def choose_action(self, Q: np.ndarray, state: int) -> int:
        q_values = Q[state] / self.temperature
        # 数值稳定性：减去最大值
        q_values = q_values - np.max(q_values)
        probs = np.exp(q_values) / np.sum(np.exp(q_values))
        return np.random.choice(self.n_actions, p=probs)


def compare_exploration_strategies(Q: np.ndarray, state: int,
                                    n_trials: int = 1000):
    """比较 epsilon-greedy 和 Softmax 的动作选择分布"""
    n_actions = Q.shape[1]

    # epsilon-greedy (epsilon=0.1)
    eg_counts = np.zeros(n_actions)
    for _ in range(n_trials):
        if np.random.random() < 0.1:
            eg_counts[np.random.randint(n_actions)] += 1
        else:
            eg_counts[np.argmax(Q[state])] += 1

    # Softmax
    sm_policy = SoftmaxPolicy(n_actions, temperature=0.5)
    sm_counts = np.zeros(n_actions)
    for _ in range(n_trials):
        sm_counts[sm_policy.choose_action(Q, state)] += 1

    print(f"Q 值: {Q[state]}")
    print(f"Epsilon-greedy: {eg_counts / n_trials}")
    print(f"Softmax:        {sm_counts / n_trials}")

# 学习率衰减策略对比

import numpy as np

def constant_lr(t: int, alpha0: float = 0.1) -> float:
    """固定学习率"""
    return alpha0

def decay_lr(t: int, alpha0: float = 0.5, k: float = 100) -> float:
    """衰减学习率：alpha_t = alpha0 * k / (k + t)"""
    return alpha0 * k / (k + t)

def step_lr(t: int, alpha0: float = 0.1,
            decay_steps: int = 500, decay_rate: float = 0.5) -> float:
    """阶梯式衰减"""
    return alpha0 * (decay_rate ** (t // decay_steps))


# 可视化不同学习率策略
total_steps = 2000
steps = np.arange(total_steps)
lr_constant = [constant_lr(t) for t in steps]
lr_decay = [decay_lr(t) for t in steps]
lr_step = [step_lr(t) for t in steps]

print(f"初始: {lr_constant[0]:.3f}, {lr_decay[0]:.3f}, {lr_step[0]:.3f}")
print(f"500步: {lr_constant[500]:.3f}, {lr_decay[500]:.3f}, {lr_step[500]:.3f}")
print(f"2000步:{lr_constant[1999]:.3f}, {lr_decay[1999]:.3f}, {lr_step[1999]:.3f}")

# 折扣因子对策略选择的影响分析

def analyze_gamma_impact(P: dict, R: dict, n_states: int,
                          n_actions: int) -> dict:
    """分析不同 gamma 对最优策略的影响"""
    gammas = [0.0, 0.5, 0.9, 0.99, 0.999]
    results = {}

    for gamma in gammas:
        Q = np.zeros((n_states, n_actions))

        # 值迭代求解 Q*
        for _ in range(1000):
            Q_new = Q.copy()
            for s in range(n_states):
                for a in range(n_actions):
                    q = 0.0
                    for sp, prob in P[s][a].items():
                        r = R[s][a].get(sp, 0)
                        q += prob * (r + gamma * np.max(Q[sp]))
                    Q_new[s, a] = q
            if np.max(np.abs(Q_new - Q)) < 1e-6:
                break
            Q = Q_new

        policy = np.argmax(Q, axis=1)
        results[gamma] = {"policy": policy.tolist(), "V_start": float(np.max(Q[0]))}
        print(f"gamma={gamma:.3f}: 策略={policy.tolist()}, 起点值={np.max(Q[0]):.3f}")

    return results

# 监控 Q-Learning 收敛过程

def monitor_convergence(env, episodes: int = 5000, alpha: float = 0.1,
                         gamma: float = 0.99, epsilon: float = 0.3) -> dict:
    """监控 Q-Learning 的收敛行为"""
    n_states = env.observation_space.n
    n_actions = env.action_space.n
    Q = np.zeros((n_states, n_actions))

    metrics = {
        "episode_rewards": [],
        "max_q_changes": [],
        "state_visit_counts": np.zeros(n_states),
    }

    for ep in range(episodes):
        state, _ = env.reset()
        total_reward = 0.0
        max_q_change = 0.0
        done = False
        truncated = False

        while not (done or truncated):
            if np.random.random() < epsilon:
                action = np.random.randint(n_actions)
            else:
                action = int(np.argmax(Q[state]))

            next_state, reward, done, truncated, _ = env.step(action)
            metrics["state_visit_counts"][state] += 1

            if done:
                td_target = reward
            else:
                td_target = reward + gamma * np.max(Q[next_state])

            old_q = Q[state, action]
            td_error = td_target - old_q
            Q[state, action] += alpha * td_error
            max_q_change = max(max_q_change, abs(td_error))

            state = next_state
            total_reward += reward

        metrics["episode_rewards"].append(total_reward)
        metrics["max_q_changes"].append(max_q_change)

    return metrics

# 验证 Robbins-Monro 条件

def check_robbins_monro(alpha_fn, max_steps: int = 100000) -> dict:
    """
    检查学习率函数是否满足 Robbins-Monro 条件:
    1. sum(alpha_t) = 无穷（足够大的更新）
    2. sum(alpha_t^2) < 无穷（方差趋于零）
    """
    sum_alpha = 0.0
    sum_alpha_sq = 0.0

    for t in range(1, max_steps + 1):
        alpha = alpha_fn(t)
        sum_alpha += alpha
        sum_alpha_sq += alpha ** 2

    converged = {
        "sum_alpha_diverges": sum_alpha > 100,
        "sum_alpha_sq_converges": sum_alpha_sq < 100,
    }
    return converged


# 测试不同学习率策略
def lr_classic(t: int) -> float:
    return 1.0 / t  # 经典 RM 学习率

def lr_fixed(t: int) -> float:
    return 0.1  # 固定学习率

print("=== 学习率 1/t ===")
r1 = check_robbins_monro(lr_classic)
print(f"满足 RM 条件: {r1}")
print("\n=== 固定学习率 0.1 ===")
r2 = check_robbins_monro(lr_fixed)
print(f"满足 RM 条件: {r2}")

import gymnasium as gym
import numpy as np

class QLearningAgent:
    """完整的 Q-Learning 智能体"""

    def __init__(self, n_states: int, n_actions: int,
                 alpha: float = 0.1, gamma: float = 0.99,
                 epsilon: float = 0.3, epsilon_min: float = 0.01,
                 epsilon_decay: float = 0.999):
        self.Q = np.zeros((n_states, n_actions))
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_min = epsilon_min
        self.epsilon_decay = epsilon_decay
        self.n_actions = n_actions

    def choose_action(self, state: int) -> int:
        """epsilon-greedy 策略选择动作"""
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)
        return int(np.argmax(self.Q[state]))

    def update(self, state: int, action: int, reward: float,
               next_state: int, done: bool) -> float:
        """Q-Learning 更新"""
        if done:
            td_target = reward
        else:
            td_target = reward + self.gamma * np.max(self.Q[next_state])

        td_error = td_target - self.Q[state, action]
        self.Q[state, action] += self.alpha * td_error
        return td_error

    def decay_epsilon(self):
        """衰减 epsilon"""
        self.epsilon = max(self.epsilon_min,
                          self.epsilon * self.epsilon_decay)

    def get_policy(self) -> np.ndarray:
        """提取最优策略"""
        return np.argmax(self.Q, axis=1)

# 完整训练流程 + 评估

def train_and_evaluate(env, agent: QLearningAgent,
                        train_episodes: int = 5000,
                        eval_episodes: int = 100) -> dict:
    """训练 Q-Learning 智能体并评估"""
    rewards_history = []
    window = 100  # 滑动窗口

    # 训练阶段
    for ep in range(train_episodes):
        state, _ = env.reset()
        total_reward = 0.0
        done = False
        truncated = False

        while not (done or truncated):
            action = agent.choose_action(state)
            next_state, reward, done, truncated, _ = env.step(action)
            agent.update(state, action, reward, next_state, done or truncated)
            state = next_state
            total_reward += reward

        agent.decay_epsilon()
        rewards_history.append(total_reward)

        if (ep + 1) % window == 0:
            recent = rewards_history[-window:]
            success_rate = np.mean(recent)
            print(f"Episode {ep+1}/{train_episodes} | "
                  f"epsilon={agent.epsilon:.3f} | "
                  f"成功率={success_rate:.1%}")

    # 评估阶段（关闭探索）
    epsilon_save = agent.epsilon
    agent.epsilon = 0.0
    eval_rewards = []
    for _ in range(eval_episodes):
        state, _ = env.reset()
        total_reward = 0.0
        done = False
        truncated = False
        while not (done or truncated):
            action = agent.choose_action(state)
            state, reward, done, truncated, _ = env.step(action)
            total_reward += reward
        eval_rewards.append(total_reward)
    agent.epsilon = epsilon_save

    eval_success = np.mean(eval_rewards)
    policy = agent.get_policy()
    action_names = ["左", "下", "右", "上"]
    print(f"评估成功率: {eval_success:.1%}")
    print(f"最优策略: {[action_names[a] for a in policy]}")

    return {"Q": agent.Q, "policy": policy, "eval_success_rate": eval_success}