1. 引言

AFSIM（Advanced Framework for Simulation, Integration, and Modeling）是美军广泛使用的作战仿真框架，具备高保真的平台建模、传感器建模与交战规则引擎。近年来，把强化学习（RL）智能体接入 AFSIM 做空战/对抗训练成为热门方向。但直接在 AFSIM 上训练端到端 RL 智能体面临两个现实问题：

• 动作空间巨大：一架战机同时涉及航向、高度、速度、雷达开关、武器发射等多个维度，扁平策略难以收敛。
• 奖励稀疏且时序跨度长：一次有效交战往往需要几十秒甚至数分钟的战术机动铺垫，单层 RL 很难把"战术意图"和"底层操纵"关联起来。

分层强化学习（Hierarchical RL, HRL）正是应对这类问题的有效范式：把"打什么、怎么打"的高层决策与"杆舵油门怎么动"的低层控制解耦，分别训练、协同优化。本文结合 AFSIM 的仿真接口，给出一套可落地的 HRL 智能体训练方案。

2. 为什么在 AFSIM 上选择 HRL

2.1 扁平 RL 的瓶颈

问题	扁平 RL 表现	HRL 改善方式
动作维度高	探索效率低，样本需求爆炸	高层输出离散战术选项，低层只负责跟踪
奖励稀疏	难以把终局奖励回传到早期动作	高层用战术级奖励，低层用跟踪误差奖励
时序跨度长	信用分配困难	分层时间尺度，高层决策周期更长
可解释性差	策略黑盒	高层选项语义明确，便于复盘

2.2 HRL 与 AFSIM 的契合点

AFSIM 天然提供了分层可观测的状态：

• 平台级：位置、姿态、速度、油量、挂载。
• 传感器级：雷达跟踪状态、目标距离/角度、威胁告警。
• 任务级：任务指令、编队状态、交战结果。

这些恰好可以分别喂给高层与低层策略，形成"任务层—战术层—控制层"的天然分层。

3. 整体架构设计

架构分为四层：

1. 仿真层：AFSIM 负责物理与交战裁决，通过外部接口（如 AFSIM 的 WsfScript 或外部控制插件）暴露状态、接收指令。
2. 适配层：把 AFSIM 的原始状态归一化为固定维度的观测向量，把离散/连续动作翻译为 AFSIM 可执行的指令。
3. 策略层：高层 Manager 与低层 Worker 两级策略网络。
4. 训练层：经验回放、奖励塑形、课程学习调度。

4. 分层策略建模

4.1 高层策略：战术选项选择

高层策略在较长的决策周期（例如每 2 秒或每 10 个仿真步）输出一个选项（Option），选项是预定义的战术原语集合：

• PURSUE：追击当前目标
• EVADE：规避来袭威胁
• MAINTAIN：保持当前态势
• REPOSITION：调整占位/高度优势
• ENGAGE：进入武器发射包线并攻击
• RETREAT：脱离交战

高层观测包含：目标相对方位/距离/速度、威胁等级、剩余弹药、油量、当前选项及其持续步数。

下面给出一个可直接运行的 PyTorch 高层 Manager 策略网络实现：输入为高层观测，输出 6 个战术选项的离散概率分布与状态价值，并附选项采样与终止判断的调用示例。

import torch
import torch.nn as nn
import torch.nn.functional as F

# ===== 维度定义 =====
# 高层观测（已归一化到 [-1, 1] 或 [0, 1]）：
#   0     : 目标相对方位角 sin
#   1     : 目标相对方位角 cos
#   2     : 目标相对距离 / 最大探测距离
#   3     : 目标相对接近速度 / 最大速度（正为接近）
#   4     : 威胁等级 [0, 1]
#   5     : 剩余弹药比例 [0, 1]
#   6     : 剩余油量比例 [0, 1]
#   7-12  : 当前选项 one-hot（PURSUE/EVADE/MAINTAIN/REPOSITION/ENGAGE/RETREAT）
#   13    : 当前选项已持续步数 / 最大持续步数
HIGH_OBS_DIM = 14

# 战术选项数量，与 4.1 节 6 个战术原语一一对应
OPTION_DIM = 6

# 选项名称，便于日志与调试
OPTION_NAMES = ["PURSUE", "EVADE", "MAINTAIN", "REPOSITION", "ENGAGE", "RETREAT"]


class HighLevelManager(nn.Module):
    """高层策略网络：高层观测 -> 6 个战术选项的离散分布 + 状态价值"""

    def __init__(self, obs_dim: int = HIGH_OBS_DIM,
                 option_dim: int = OPTION_DIM,
                 hidden: int = 128):
        super().__init__()
        # 共享特征提取：14 维观测 -> 128 维特征
        self.backbone = nn.Sequential(
            nn.Linear(obs_dim, hidden),   # 14 -> 128
            nn.ReLU(),
            nn.Linear(hidden, hidden),    # 128 -> 128
            nn.ReLU(),
        )

        # 策略头：输出每个选项的 logits，softmax 后即为离散概率分布
        self.actor = nn.Linear(hidden, option_dim)   # 128 -> 6

        # 价值头：输出状态价值，供 PPO/A2C 使用
        self.critic = nn.Sequential(
            nn.Linear(hidden, hidden // 2),  # 128 -> 64
            nn.ReLU(),
            nn.Linear(hidden // 2, 1),       # 64 -> 1
        )

    def forward(self, obs: torch.Tensor):
        """
        obs: [B, HIGH_OBS_DIM] 归一化高层观测
        返回: probs [B, OPTION_DIM] 选项概率分布, value [B, 1] 状态价值
        """
        feat = self.backbone(obs)              # [B, 128]
        logits = self.actor(feat)              # [B, 6]
        probs = F.softmax(logits, dim=-1)      # [B, 6]，和为 1
        value = self.critic(feat)              # [B, 1]
        return probs, value

    def act(self, obs: torch.Tensor, deterministic: bool = False):
        """采样一个战术选项，返回 (option_idx, log_prob, value)"""
        probs, value = self.forward(obs)
        dist = torch.distributions.Categorical(probs)
        if deterministic:
            option = torch.argmax(probs, dim=-1)   # 贪心，用于评估
        else:
            option = dist.sample()                 # 按概率采样，用于训练探索
        log_prob = dist.log_prob(option).unsqueeze(-1)   # [B, 1]
        return option, log_prob, value


# ===== 选项终止判断（规则实现，可后续替换为学习） =====
def should_terminate(option_idx: int, obs: dict, duration: int) -> bool:
    """
    option_idx: 当前选项索引
    obs:        原始（未归一化）状态字典，含 distance/threat/ammo 等
    duration:   当前选项已持续步数
    """
    name = OPTION_NAMES[option_idx]
    if name == "PURSUE":
        return obs["distance"] < 3000 or not obs["radar_lock"]
    if name == "EVADE":
        return obs["threat"] < 0.2 or duration > 100
    if name == "ENGAGE":
        return obs["ammo"] <= 0 or obs["target_destroyed"]
    if name == "RETREAT":
        return obs["threat"] < 0.1 or duration > 200
    # MAINTAIN / REPOSITION 用最大持续步数兜底，避免死锁
    return duration > 150


if __name__ == "__main__":
    manager = HighLevelManager()
    obs = torch.randn(4, HIGH_OBS_DIM)          # 模拟一个 batch 的高层观测
    probs, value = manager.forward(obs)
    print("probs shape:", probs.shape)          # torch.Size([4, 6])
    print("probs sum:", probs.sum(dim=-1))      # 每行和为 1
    print("value shape:", value.shape)          # torch.Size([4, 1])

    # 采样选项并判断是否终止
    option, logp, val = manager.act(obs)
    print("sampled options:", [OPTION_NAMES[i] for i in option.tolist()])
    raw_state = {"distance": 2500, "threat": 0.1, "ammo": 2,
                 "radar_lock": True, "target_destroyed": False}
    print("terminate PURSUE:", should_terminate(0, raw_state, duration=10))

几点实现说明：

• 离散策略头：actor 直接输出 6 维 logits，经 softmax 得到选项概率分布，配合 Categorical 分布即可完成采样与 log_prob 计算，天然适配 PPO 的离散动作形式。
• 观测编码：把「当前选项 one-hot + 持续步数」一并放进观测，让高层网络感知自己已执行了多久，有助于学习「避免频繁切换」的行为。
• 终止判断解耦：should_terminate 先用规则实现，与 4.3 节的终止条件对应；后续可把终止函数也参数化，用学习方式替代。
• 决策周期：高层每 2 秒（或每 10 个仿真步）调用一次 act，低层在每个仿真步调用低层策略，形成分层时间尺度。

下面给出配套的基于 PPO 的高层 Manager 训练循环，覆盖 rollout 采集、GAE 优势估计、策略与价值损失计算、梯度裁剪与学习率调度，可直接与上面的 HighLevelManager 对接。

import torch
import torch.nn as nn
import torch.nn.functional as F

# ===== PPO 超参数 =====
GAMMA = 0.99            # 折扣因子：越接近 1 越看重长期战术收益
LAM = 0.95              # GAE 的 lambda：平衡偏差与方差
CLIP_EPS = 0.2          # PPO 策略裁剪范围，限制单次更新幅度
VALUE_COEF = 0.5        # 价值损失权重
ENTROPY_COEF = 0.01     # 熵奖励系数，鼓励探索、防止过早收敛
MAX_GRAD_NORM = 0.5     # 梯度裁剪阈值，防止梯度爆炸
LR = 3e-4               # 初始学习率
PPO_EPOCHS = 4          # 每批数据重复训练轮数
MINI_BATCH = 64         # 小批量大小
ROLLOUT_STEPS = 2048    # 每次 rollout 采集的决策步数


def compute_gae(rewards, values, dones, last_value, gamma=GAMMA, lam=LAM):
    """GAE 优势估计：把稀疏的战术级奖励平滑地回传到每个高层决策步"""
    T = len(rewards)
    advantages = torch.zeros(T, 1)
    gae = 0.0
    for t in reversed(range(T)):
        # 终止步之后不再有未来价值
        next_value = last_value if t == T - 1 else values[t + 1]
        next_non_terminal = 1.0 - dones[t]
        # TD 残差：当前奖励 + 折扣后的下一状态价值 - 当前状态价值
        delta = rewards[t] + gamma * next_value * next_non_terminal - values[t]
        # 递归累积：兼顾多步回报，降低方差
        gae = delta + gamma * lam * next_non_terminal * gae
        advantages[t] = gae
    # 回报 = 优势 + 价值基线，作为价值网络的回归目标
    returns = advantages + values
    return advantages, returns


def collect_rollout(manager, env, steps=ROLLOUT_STEPS):
    """rollout 采集：按高层决策周期调用 act，并用 should_terminate 判断选项切换"""
    obs_buf, act_buf, logp_buf, val_buf = [], [], [], []
    rew_buf, done_buf = [], []

    obs = env.reset()                 # 返回归一化后的高层观测 [HIGH_OBS_DIM]
    option_duration = 0               # 当前选项已持续步数
    current_option = 0

    for _ in range(steps):
        obs_t = torch.as_tensor(obs, dtype=torch.float32).unsqueeze(0)
        with torch.no_grad():
            option, logp, value = manager.act(obs_t)   # 采样战术选项
        option_idx = int(option.item())

        # 执行一个高层决策周期（内部由低层策略跟踪该选项）
        next_obs, reward, done, info = env.step(option_idx)
        option_duration += 1

        # 选项终止判断：满足终止条件则允许下一步重新选择
        if should_terminate(option_idx, info["raw_state"], option_duration):
            option_duration = 0
            current_option = option_idx

        obs_buf.append(obs_t.squeeze(0))
        act_buf.append(option)
        logp_buf.append(logp.squeeze(0))
        val_buf.append(value.squeeze(0))
        rew_buf.append(float(reward))
        done_buf.append(float(done))

        obs = env.reset() if done else next_obs

    # 用最后一步的价值估计 bootstrap 未结束的轨迹
    with torch.no_grad():
        last_obs = torch.as_tensor(obs, dtype=torch.float32).unsqueeze(0)
        _, last_value = manager.forward(last_obs)

    obs_buf = torch.stack(obs_buf)                 # [T, HIGH_OBS_DIM]
    act_buf = torch.stack(act_buf)                 # [T]
    logp_buf = torch.stack(logp_buf)               # [T, 1]
    val_buf = torch.stack(val_buf)                 # [T, 1]
    rew_buf = torch.tensor(rew_buf).unsqueeze(-1)  # [T, 1]
    done_buf = torch.tensor(done_buf).unsqueeze(-1)

    advantages, returns = compute_gae(rew_buf, val_buf, done_buf, last_value)
    # 优势标准化：稳定策略梯度尺度
    advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
    return obs_buf, act_buf, logp_buf, val_buf, advantages, returns


def ppo_update(manager, optimizer, scheduler, batch):
    """PPO 更新：多轮小批量迭代，计算裁剪策略损失 + 价值损失 + 熵奖励"""
    obs, actions, old_logp, old_values, advantages, returns = batch
    n = obs.shape[0]

    for _ in range(PPO_EPOCHS):
        idx = torch.randperm(n)                    # 打乱样本，降低相关性
        for start in range(0, n, MINI_BATCH):
            mb = idx[start:start + MINI_BATCH]
            mb_obs = obs[mb]
            mb_act = actions[mb]
            mb_old_logp = old_logp[mb]
            mb_adv = advantages[mb]
            mb_ret = returns[mb]

            probs, values = manager.forward(mb_obs)
            dist = torch.distributions.Categorical(probs)
            new_logp = dist.log_prob(mb_act).unsqueeze(-1)
            entropy = dist.entropy().mean()

            # 重要性采样比率：新策略与旧策略在同一动作上的概率比
            ratio = torch.exp(new_logp - mb_old_logp)
            # 裁剪策略损失：限制 ratio 在 [1-eps, 1+eps]，避免策略突变
            surr1 = ratio * mb_adv
            surr2 = torch.clamp(ratio, 1.0 - CLIP_EPS, 1.0 + CLIP_EPS) * mb_adv
            policy_loss = -torch.min(surr1, surr2).mean()

            # 价值损失：用 GAE 回报作为回归目标
            value_loss = F.mse_loss(values, mb_ret)

            # 总损失：策略 + 价值 - 熵奖励（熵越大探索越充分）
            loss = policy_loss + VALUE_COEF * value_loss - ENTROPY_COEF * entropy

            optimizer.zero_grad()
            loss.backward()
            # 梯度裁剪：防止长时序 rollout 带来的梯度爆炸
            nn.utils.clip_grad_norm_(manager.parameters(), MAX_GRAD_NORM)
            optimizer.step()

        # 学习率调度：每轮 epoch 后线性衰减，后期更新更稳
        scheduler.step()


if __name__ == "__main__":
    manager = HighLevelManager()
    optimizer = torch.optim.Adam(manager.parameters(), lr=LR)
    # 余弦退火调度：从 LR 平滑衰减到接近 0
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)

    # 伪代码：env 需实现 reset()/step(option_idx)，返回 (obs, reward, done, info)
    # for iteration in range(100):
    #     batch = collect_rollout(manager, env)
    #     ppo_update(manager, optimizer, scheduler, batch)
    print("PPO 训练循环已就绪：collect_rollout -> compute_gae -> ppo_update")

几点训练实现说明：

• rollout 采集：高层按决策周期调用 act 采样选项，每个选项执行若干步后由 should_terminate 判断是否切换，保证采集到的轨迹符合分层时间尺度。
• GAE 优势估计：用 gamma 与 lam 在反向遍历中累积 TD 残差，把稀疏的战术级奖励平滑分配到每个高层决策步，显著降低策略梯度方差。
• 裁剪策略损失：ratio 被限制在 [1-eps, 1+eps]，避免一次更新把策略推得太远导致性能崩塌，这是 PPO 稳定性的核心。
• 梯度裁剪与学习率调度：clip_grad_norm_ 控制梯度范数，余弦退火让学习率随训练平滑下降，二者配合可明显提升长时序训练的收敛稳定性。

4.2 低层策略：飞行控制

低层策略接收"当前选项 + 局部观测"，输出连续动作（航向角速度、俯仰角速度、油门、雷达开关、发射指令）。低层决策周期更短（例如每 0.1 秒），保证控制平滑。

下面给出一个可直接运行的 PyTorch 低层策略网络实现，输入为「当前选项 + 局部观测」，输出经 tanh 激活后再按物理包线缩放的连续动作（航向角速度、俯仰角速度、油门）。

import torch
import torch.nn as nn

# ===== 维度定义 =====
# 局部观测（已归一化到 [-1, 1]）：
#   0-2   : 自身姿态角速度 (roll_rate, pitch_rate, yaw_rate)  单位 rad/s
#   3     : 自身速度 / 最大速度
#   4     : 自身高度归一化 (alt - 500) / (12000 - 500)
#   5-7   : 目标相对位置 (dx, dy, dz) / 10000，机体系
#   8     : 目标相对速度 / 最大速度
#   9     : 目标方位角 sin
#   10    : 目标方位角 cos
#   11    : 目标进入角 sin
#   12    : 目标进入角 cos
#   13    : 威胁等级 [0, 1]
#   14    : 剩余油量比例 [0, 1]
#   15    : 雷达跟踪标志 {0, 1}
OBS_DIM = 16

# 选项（Option）one-hot 编码，与 4.1 节 6 个战术原语一一对应
#   PURSUE / EVADE / MAINTAIN / REPOSITION / ENGAGE / RETREAT
OPTION_DIM = 6

# 连续动作维度：航向角速度、俯仰角速度、油门
ACTION_DIM = 3

# 动作物理上限，用于把 tanh 输出缩放到真实控制量
ACTION_SCALE = torch.tensor([0.5, 0.3, 1.0])   # rad/s, rad/s, 油门 [0,1] 区间
ACTION_BIAS = torch.tensor([0.0, 0.0, 0.5])    # 油门中位偏置，使初始输出约 0.5


class LowLevelPolicy(nn.Module):
    """低层策略网络：观测 + 选项 -> 连续飞控动作"""

    def __init__(self, obs_dim: int = OBS_DIM,
                 option_dim: int = OPTION_DIM,
                 action_dim: int = ACTION_DIM,
                 hidden: int = 256):
        super().__init__()
        in_dim = obs_dim + option_dim

        # 共享特征提取：两层全连接，逐步把 22 维输入压到 256 维特征
        self.backbone = nn.Sequential(
            nn.Linear(in_dim, hidden),   # 22 -> 256
            nn.ReLU(),
            nn.Linear(hidden, hidden),   # 256 -> 256
            nn.ReLU(),
        )

        # 策略头：输出动作均值，tanh 保证落在 [-1, 1]
        self.actor = nn.Sequential(
            nn.Linear(hidden, hidden // 2),  # 256 -> 128
            nn.ReLU(),
            nn.Linear(hidden // 2, action_dim),  # 128 -> 3
            nn.Tanh(),
        )

        # 价值头：输出状态价值，供 PPO/A2C 使用
        self.critic = nn.Sequential(
            nn.Linear(hidden, hidden // 2),  # 256 -> 128
            nn.ReLU(),
            nn.Linear(hidden // 2, 1),       # 128 -> 1
        )

        # 可学习对数标准差，与动作维度对齐，便于连续动作探索
        self.log_std = nn.Parameter(torch.zeros(action_dim))

        # 注册缩放/偏置，随模型一起迁移到 GPU
        self.register_buffer("action_scale", ACTION_SCALE)
        self.register_buffer("action_bias", ACTION_BIAS)

    def forward(self, obs: torch.Tensor, option_onehot: torch.Tensor):
        """
        obs:          [B, OBS_DIM]      归一化局部观测
        option_onehot:[B, OPTION_DIM]   当前选项 one-hot
        返回: action [B, ACTION_DIM] 已缩放到物理量纲, value [B, 1]
        """
        x = torch.cat([obs, option_onehot], dim=-1)   # [B, 22]
        feat = self.backbone(x)                        # [B, 256]

        raw = self.actor(feat)                         # [B, 3], 取值 [-1, 1]
        # 缩放：raw * scale + bias，得到真实控制量
        action = raw * self.action_scale + self.action_bias

        value = self.critic(feat)                      # [B, 1]
        return action, value

    def act(self, obs: torch.Tensor, option_onehot: torch.Tensor, deterministic: bool = False):
        """采样动作，返回 (action, log_prob, value)，供训练循环调用"""
        action, value = self.forward(obs, option_onehot)
        if deterministic:
            return action, None, value

        std = self.log_std.exp().expand_as(action)
        dist = torch.distributions.Normal(action, std)
        sample = dist.sample()
        # 采样后再次裁剪到物理包线内，避免越界指令
        sample = torch.clamp(sample, self.action_bias - self.action_scale,
                             self.action_bias + self.action_scale)
        log_prob = dist.log_prob(sample).sum(dim=-1, keepdim=True)
        return sample, log_prob, value


if __name__ == "__main__":
    policy = LowLevelPolicy()
    obs = torch.randn(4, OBS_DIM)                       # 模拟一个 batch 的观测
    option = torch.eye(OPTION_DIM)[torch.randint(0, OPTION_DIM, (4,))]  # 随机选项
    act, logp, val = policy.act(obs, option)
    print("action shape:", act.shape)   # torch.Size([4, 3])
    print("log_prob shape:", logp.shape)
    print("value shape:", val.shape)

几点实现说明：

• 输入拼接：把选项 one-hot 与观测直接拼接，让低层网络显式感知「当前在执行哪个战术」，避免不同选项共用同一套控制策略导致行为混淆。
• tanh + 缩放：tanh 把输出限制在 [-1, 1]，再乘以 ACTION_SCALE 并加 ACTION_BIAS，映射到航向角速度 ±0.5 rad/s、俯仰角速度 ±0.3 rad/s、油门 [0, 1] 的真实包线。
• 双头结构：actor 与 critic 共享 backbone，减少参数量并加速收敛，适合 PPO 这类 on-policy 算法。
• 可学习标准差：log_std 作为独立参数，训练中自动调节探索强度，比固定噪声更稳定。

下面给出配套的基于 PPO 的低层策略训练循环，覆盖连续动作的 rollout 采集、GAE 优势估计、带 log_prob 修正的裁剪策略损失、价值损失与熵奖励，可直接与上面的 LowLevelPolicy 对接。

import torch
import torch.nn as nn
import torch.nn.functional as F

# ===== PPO 超参数（低层控制级） =====
GAMMA = 0.99            # 折扣因子：低层决策周期短，可适当调小（如 0.95）
LAM = 0.95              # GAE 的 lambda：平衡偏差与方差
CLIP_EPS = 0.2          # PPO 策略裁剪范围，限制单次更新幅度
VALUE_COEF = 0.5        # 价值损失权重
ENTROPY_COEF = 0.005    # 熵奖励系数：连续动作熵天然较大，系数比离散动作更小
MAX_GRAD_NORM = 0.5     # 梯度裁剪阈值，防止梯度爆炸
LR = 3e-4               # 初始学习率
PPO_EPOCHS = 4          # 每批数据重复训练轮数
MINI_BATCH = 256        # 小批量大小：低层步数多，batch 可比高层更大
ROLLOUT_STEPS = 4096    # 每次 rollout 采集的控制步数（低层频率高，采集更多）


def compute_gae(rewards, values, dones, last_value, gamma=GAMMA, lam=LAM):
    """GAE 优势估计：把控制级奖励平滑回传到每个低层控制步"""
    T = len(rewards)
    advantages = torch.zeros(T, 1)
    gae = 0.0
    for t in reversed(range(T)):
        next_value = last_value if t == T - 1 else values[t + 1]
        next_non_terminal = 1.0 - dones[t]
        # TD 残差：当前奖励 + 折扣后的下一状态价值 - 当前状态价值
        delta = rewards[t] + gamma * next_value * next_non_terminal - values[t]
        # 递归累积多步回报，降低方差
        gae = delta + gamma * lam * next_non_terminal * gae
        advantages[t] = gae
    # 回报 = 优势 + 价值基线，作为价值网络的回归目标
    returns = advantages + values
    return advantages, returns


def collect_rollout(policy, env, steps=ROLLOUT_STEPS):
    """连续动作 rollout 采集：每个控制步都调用 act 采样连续动作"""
    obs_buf, opt_buf, act_buf, logp_buf, val_buf = [], [], [], [], []
    rew_buf, done_buf = [], []

    obs, option_onehot = env.reset()   # 返回 (局部观测 [OBS_DIM], 当前选项 one-hot [OPTION_DIM])

    for _ in range(steps):
        obs_t = torch.as_tensor(obs, dtype=torch.float32).unsqueeze(0)
        opt_t = torch.as_tensor(option_onehot, dtype=torch.float32).unsqueeze(0)
        with torch.no_grad():
            # 连续动作：返回的是 [B, ACTION_DIM] 的动作向量，而非离散索引
            action, logp, value = policy.act(obs_t, opt_t)

        # 执行一个控制步（0.1 秒），返回下一观测、奖励、终止标志与信息
        next_obs, next_option, reward, done, info = env.step(action.squeeze(0).numpy())

        obs_buf.append(obs_t.squeeze(0))
        opt_buf.append(opt_t.squeeze(0))
        act_buf.append(action.squeeze(0))       # [ACTION_DIM] 连续动作
        logp_buf.append(logp.squeeze(0))        # [1] 该动作的联合 log_prob
        val_buf.append(value.squeeze(0))
        rew_buf.append(float(reward))
        done_buf.append(float(done))

        if done:
            obs, option_onehot = env.reset()
        else:
            obs, option_onehot = next_obs, next_option

    # 用最后一步的价值估计 bootstrap 未结束的轨迹
    with torch.no_grad():
        last_obs = torch.as_tensor(obs, dtype=torch.float32).unsqueeze(0)
        last_opt = torch.as_tensor(option_onehot, dtype=torch.float32).unsqueeze(0)
        _, last_value = policy.forward(last_obs, last_opt)

    obs_buf = torch.stack(obs_buf)                 # [T, OBS_DIM]
    opt_buf = torch.stack(opt_buf)                 # [T, OPTION_DIM]
    act_buf = torch.stack(act_buf)                 # [T, ACTION_DIM]
    logp_buf = torch.stack(logp_buf)               # [T, 1]
    val_buf = torch.stack(val_buf)                 # [T, 1]
    rew_buf = torch.tensor(rew_buf).unsqueeze(-1)  # [T, 1]
    done_buf = torch.tensor(done_buf).unsqueeze(-1)

    advantages, returns = compute_gae(rew_buf, val_buf, done_buf, last_value)
    # 优势标准化：稳定策略梯度尺度
    advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
    return obs_buf, opt_buf, act_buf, logp_buf, val_buf, advantages, returns


def ppo_update(policy, optimizer, scheduler, batch):
    """PPO 更新：连续动作版本，策略损失需用高斯分布重新计算 log_prob"""
    obs, option, actions, old_logp, old_values, advantages, returns = batch
    n = obs.shape[0]

    for _ in range(PPO_EPOCHS):
        idx = torch.randperm(n)                    # 打乱样本，降低相关性
        for start in range(0, n, MINI_BATCH):
            mb = idx[start:start + MINI_BATCH]
            mb_obs, mb_opt = obs[mb], option[mb]
            mb_act, mb_old_logp = actions[mb], old_logp[mb]
            mb_adv, mb_ret = advantages[mb], returns[mb]

            # 前向得到动作均值与价值；注意这里拿到的是缩放后的动作均值
            mean_action, values = policy.forward(mb_obs, mb_opt)

            # 连续动作的关键差异：用当前策略的高斯分布重新计算 log_prob
            # 注意 std 必须与采样时一致（同一个 log_std），否则 ratio 会失真
            std = policy.log_std.exp().expand_as(mean_action)
            dist = torch.distributions.Normal(mean_action, std)
            # 对每个动作维度求 log_prob 后求和，得到该动作向量的联合 log_prob
            new_logp = dist.log_prob(mb_act).sum(dim=-1, keepdim=True)
            # 连续动作的熵是各维度高斯熵之和
            entropy = dist.entropy().sum(dim=-1).mean()

            # 重要性采样比率：新策略与旧策略在同一连续动作上的概率密度比
            ratio = torch.exp(new_logp - mb_old_logp)
            # 裁剪策略损失：限制 ratio 在 [1-eps, 1+eps]，避免策略突变
            surr1 = ratio * mb_adv
            surr2 = torch.clamp(ratio, 1.0 - CLIP_EPS, 1.0 + CLIP_EPS) * mb_adv
            policy_loss = -torch.min(surr1, surr2).mean()

            # 价值损失：用 GAE 回报作为回归目标
            value_loss = F.mse_loss(values, mb_ret)

            # 总损失：策略 + 价值 - 熵奖励（熵越大探索越充分）
            loss = policy_loss + VALUE_COEF * value_loss - ENTROPY_COEF * entropy

            optimizer.zero_grad()
            loss.backward()
            # 梯度裁剪：防止长时序 rollout 带来的梯度爆炸
            nn.utils.clip_grad_norm_(policy.parameters(), MAX_GRAD_NORM)
            optimizer.step()

        # 学习率调度：每轮 epoch 后衰减，后期更新更稳
        scheduler.step()


if __name__ == "__main__":
    policy = LowLevelPolicy()
    optimizer = torch.optim.Adam(policy.parameters(), lr=LR)
    # 余弦退火调度：从 LR 平滑衰减到接近 0
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)

    # 伪代码：env 需实现 reset()/step(action)，返回 (obs, option_onehot, reward, done, info)
    # for iteration in range(100):
    #     batch = collect_rollout(policy, env)
    #     ppo_update(policy, optimizer, scheduler, batch)
    print("低层 PPO 训练循环已就绪：collect_rollout -> compute_gae -> ppo_update")

几点训练实现说明（重点对比连续动作与离散动作的 PPO 差异）：

• 动作表示不同：高层 act 返回离散选项索引，低层返回 [ACTION_DIM] 的连续动作向量；因此 rollout 缓冲区里 act_buf 是 [T, ACTION_DIM] 而非 [T]，env.step 也要接收动作向量。
• 分布不同：离散动作用 Categorical，log_prob 直接对索引取值；连续动作用 Normal(mean, std)，需要对每个动作维度求 log_prob 再 sum，得到动作向量的联合对数概率。
• log_prob 修正：PPO 的 ratio = exp(new_logp - old_logp) 依赖新旧策略在同一动作上的概率密度。连续动作必须用当前策略的均值与同一个 log_std 重新构造分布计算 new_logp，否则 ratio 失真、裁剪失效。
• 熵奖励：离散动作熵是单个分布的熵；连续动作熵是各维度高斯熵之和，量级更大，因此 ENTROPY_COEF 通常取得更小（如 0.005）。
• 动作裁剪：采样后仍需 clamp 到物理包线，但要注意裁剪会改变实际执行动作，计算 log_prob 时应使用裁剪后的动作，保持训练与执行一致。

4.3 选项终止条件

每个选项需要明确的终止函数，例如：

• PURSUE 在目标进入武器包线或丢失跟踪时终止；
• EVADE 在威胁解除或达到最大规避时长时终止；
• ENGAGE 在弹药耗尽或目标被击毁时终止。

终止函数的设计直接决定分层结构的有效性，建议先用规则实现，再逐步学习。

5. 奖励设计

HRL 的关键在于分层奖励，避免高层与低层目标冲突。

5.1 高层奖励（战术级）

• 态势优势变化：角度优势、能量优势的增量；
• 达成战术目标：进入发射包线、成功规避；
• 终局奖励：击毁目标 +100，被击毁 -100；
• 时间惩罚：每步 -0.01，鼓励快速决策。

5.2 低层奖励（控制级）

• 跟踪误差：实际航向/高度与选项期望值的偏差；
• 平滑性惩罚：动作变化率过大扣分；
• 边界惩罚：超出高度/速度包线扣分。

def low_level_reward(state, action, option_target):
    track_err = abs(state.heading - option_target.heading) \
              + abs(state.altitude - option_target.altitude) / 100.0
    smooth_pen = 0.01 * abs(action - state.last_action).sum()
    boundary_pen = 1.0 if state.altitude < 500 or state.altitude > 12000 else 0.0
    return -track_err - smooth_pen - boundary_pen

6. 训练流程与课程学习

6.1 分阶段训练

1. 低层预训练：固定若干规则化选项，用脚本生成期望轨迹，训练低层策略跟踪能力。
2. 高层冷启动：冻结低层，用规则或模仿学习初始化高层选项选择。
3. 联合微调：解冻低层，用较低学习率联合优化，防止低层能力退化。
4. 对抗自博弈：引入对手智能体，逐步提升难度。

6.2 课程学习调度

curriculum = [
    {"stage": 1, "opponent": "static",   "init_dist": 8000, "episodes": 2000},
    {"stage": 2, "opponent": "scripted", "init_dist": 6000, "episodes": 5000},
    {"stage": 3, "opponent": "selfplay", "init_dist": 4000, "episodes": 20000},
]

从静止目标到脚本对手再到自博弈，逐步增加对抗强度，能显著提升收敛稳定性。

7. 与 AFSIM 的工程对接要点

• 仿真加速：AFSIM 支持无界面批量运行，训练时关闭渲染，用多进程并行采样。
• 时间同步：确保策略决策周期与仿真步长对齐，避免动作延迟导致的状态错位。
• 状态归一化：不同量纲（米、度、节）必须归一化，否则网络难以学习。
• 接口稳定性：把 AFSIM 交互封装为独立进程，通过共享内存或 socket 通信，避免仿真崩溃影响训练主进程。
• 可复现性：固定随机种子，记录每局初始态势，便于回放分析。

8. 常见问题与调优建议

现象	可能原因	调优建议
高层频繁切换选项	选项终止条件过松	增加最小持续步数、切换惩罚
低层跟踪抖动	动作空间过大	限制动作范围、加平滑惩罚
训练不收敛	奖励尺度失衡	归一化奖励、检查分层奖励权重
智能体"摆烂"	时间惩罚过重	降低时间惩罚、增加探索奖励
仿真速度慢	单进程串行	多进程并行、降低仿真精度

9. 总结

在 AFSIM 上训练 HRL 智能体，核心是把"战术决策"与"飞行控制"解耦：高层用离散选项在长周期上做战术选择，低层用连续动作在短周期上做精确控制，再通过分层奖励与课程学习把两者协同起来。工程上要重点解决仿真加速、状态归一化与接口稳定性三个问题。按本文的分阶段训练流程推进，可以在保证可解释性的同时，显著提升智能体在复杂对抗场景下的学习效率。