1. 引言：当仿真引擎遇上强化学习

AFSIM（Advanced Framework for Simulation, Integration, and Modeling）是美国空军研究实验室主导开发的作战仿真框架，广泛用于任务级、交战级的多平台对抗仿真。它提供了完整的实体建模、传感器、通信、武器以及脚本化的行为控制能力。

强化学习（Reinforcement Learning, RL）则擅长在不确定环境中通过试错学习最优策略。把两者结合，可以构建"仿真即环境"的智能体训练闭环：AFSIM 负责高保真战场推演，RL 算法负责决策优化。

本文将从架构设计、环境封装、算法实现到训练调优，完整讲解如何在 AFSIM 之上落地一个强化学习系统，并给出可运行的代码实例。

2. 为什么选择 AFSIM 作为 RL 环境

相比 Gym 中的简化环境，AFSIM 作为 RL 环境有以下优势与挑战：

维度	优势	挑战
物理与传感器模型	高保真雷达、红外、通信链路模型	单步仿真开销大
场景规模	支持数十到数百平台协同	状态空间维度高
行为控制	内置脚本语言可注入决策	需要外部接口桥接
可复现性	支持随机种子与回放	分布式训练需同步

核心结论：AFSIM 适合做高保真策略验证与课程学习后期阶段，而前期可以用简化代理环境加速收敛。

3. 整体架构设计

系统分为三层：仿真层（AFSIM）、桥接层（gRPC/Socket）、算法层（PyTorch + RL 库）。

关键设计点：

• 时间同步：AFSIM 以离散事件推进，RL 侧以固定决策周期（如每 5 秒）采样，需要桥接层做时间对齐。
• 动作空间离散化：将连续机动指令（航向、速度）离散为有限动作集，降低学习难度。
• 奖励塑形：由仿真事件（击毁、被探测、燃油消耗）映射为标量奖励。

3.1 桥接层消息协议

桥接层是 AFSIM 与 RL 算法之间的唯一通道，协议设计的好坏直接决定了系统的稳定性与可调试性。这里采用「JSON 载荷 + 4 字节大端长度前缀」的轻量协议：所有消息都是 UTF-8 编码的 JSON 对象，发送前先写入 4 字节无符号整数表示载荷字节长度，接收方先读 4 字节得到长度 L，再精确读取 L 字节后反序列化。

协议共定义三类请求：reset、step、close，其请求与响应字段如下。

reset 请求与响应

字段	所属	类型	含义
cmd	请求	string	固定为 "reset"，标识重置指令
seed	请求	int | null	随机种子，null 表示由仿真侧自行生成；用于保证回合可复现
obs	响应	float[11]	初始观测向量，已归一化到 [-1, 1]

step 请求与响应

字段	所属	类型	含义
cmd	请求	string	固定为 "step"，标识推进一步
action	请求	int	离散动作编号，取值 0~14（航向 5 档 × 速度 3 档）
duration	请求	float	本次决策推进的仿真时长（秒），需与 decision_interval 一致
obs	响应	float[11]	执行动作后的新观测向量
reward	响应	float	该步的标量奖励，由奖励函数计算
done	响应	bool	回合是否终止（击毁、被击毁或超时）
events	响应	string[]	本步发生的仿真事件列表，如 target_destroyed、radar_locked

close 请求与响应

字段	所属	类型	含义
cmd	请求	string	固定为 "close"，标识关闭连接
obs / reward / done / events	响应	—	关闭请求无需业务响应，服务端直接断开连接即可

长度前缀分帧规则

• 帧结构：[4 字节长度 L][L 字节 JSON 载荷]，长度字段为大端序无符号整数（to_bytes(4, "big")）。
• 发送方先序列化 JSON 并编码为 UTF-8，得到字节串 payload，再发送 len(payload).to_bytes(4, "big") + payload。
• 接收方先调用 _recv_exact(4) 读满 4 字节得到 L，再调用 _recv_exact(L) 读满载荷；_recv_exact 内部循环 recv 直到字节数满足，避免 TCP 粘包/半包导致解析错位。
• 单帧载荷上限建议设为 1 MB，超过则视为协议错误并主动断开，防止异常数据撑爆内存。
• 该规则与第 4 节 AFSIMEnv._send / _recv 的实现完全对应，两端必须使用同一套分帧逻辑，否则会出现「读到半个 JSON」的解析异常。

4. 环境封装：把 AFSIM 包装成 Gym 接口

RL 算法通常依赖 Gym/Gymnasium 接口。下面给出一个基于 Socket 通信的环境封装示例。

import socket
import json
import numpy as np
import gymnasium as gym
from gymnasium import spaces


class AFSIMEnv(gym.Env):
    """将 AFSIM 仿真封装为 Gymnasium 环境"""

    metadata = {"render_modes": ["human"]}

    def __init__(self, host="127.0.0.1", port=50051,
                 max_steps=200, decision_interval=5.0):
        super().__init__()
        self.host = host
        self.port = port
        self.max_steps = max_steps
        self.decision_interval = decision_interval
        self.step_count = 0
        self.sock = None

        # 观测：自身位置(3) + 速度(3) + 目标相对位置(3) + 雷达告警(1) + 燃油(1)
        self.observation_space = spaces.Box(
            low=-1.0, high=1.0, shape=(11,), dtype=np.float32
        )
        # 动作：航向 5 档 x 速度 3 档 = 15 个离散动作
        self.action_space = spaces.Discrete(15)

    def _connect(self):
        if self.sock is None:
            self.sock = socket.create_connection((self.host, self.port), timeout=10)

    def _send(self, msg: dict):
        self._connect()
        payload = json.dumps(msg).encode("utf-8")
        self.sock.sendall(len(payload).to_bytes(4, "big") + payload)

    def _recv(self) -> dict:
        header = self._recv_exact(4)
        length = int.from_bytes(header, "big")
        body = self._recv_exact(length)
        return json.loads(body.decode("utf-8"))

    def _recv_exact(self, n: int) -> bytes:
        buf = b""
        while len(buf) < n:
            chunk = self.sock.recv(n - len(buf))
            if not chunk:
                raise ConnectionError("AFSIM 连接已断开")
            buf += chunk
        return buf

    def reset(self, seed=None, options=None):
        super().reset(seed=seed)
        self.step_count = 0
        self._send({"cmd": "reset", "seed": seed})
        resp = self._recv()
        return np.array(resp["obs"], dtype=np.float32), {}

    def step(self, action):
        self.step_count += 1
        self._send({
            "cmd": "step",
            "action": int(action),
            "duration": self.decision_interval,
        })
        resp = self._recv()

        obs = np.array(resp["obs"], dtype=np.float32)
        reward = float(resp["reward"])
        terminated = bool(resp["done"])
        truncated = self.step_count >= self.max_steps
        info = {"events": resp.get("events", [])}
        return obs, reward, terminated, truncated, info

    def close(self):
        if self.sock is not None:
            try:
                self._send({"cmd": "close"})
            finally:
                self.sock.close()
                self.sock = None

5. AFSIM 侧脚本：状态上报与动作执行

在 AFSIM 场景中，通过脚本处理器周期性上报状态并读取外部动作。以下为伪代码风格的脚本片段，展示核心逻辑。

# AFSIM 脚本处理器（示意）
script_variables
    double next_decision_time = 0.0;
    int    current_action = 0;
end_script_variables

on_initialize
    # 启动桥接客户端，连接 RL 侧服务
    bridge_connect("127.0.0.1", 50052);
end_on_initialize

on_update
    if (TIME_NOW() >= next_decision_time)
    begin
        # 1. 采集状态
        state = pack_state(PLATFORM.Position(),
                           PLATFORM.Velocity(),
                           TARGET.RelativePosition(),
                           PLATFORM.RadarWarning(),
                           PLATFORM.FuelRatio());

        # 2. 上报并等待动作
        current_action = bridge_request_action(state);

        # 3. 执行动作
        apply_action(PLATFORM, current_action);

        next_decision_time = TIME_NOW() + 5.0;
    end
end_on_update

要点说明：

• pack_state 负责把仿真内部量归一化到 [-1, 1]，与 Gym 的 observation_space 保持一致。
• apply_action 将离散动作映射为航向/速度指令，例如动作 0~4 对应航向偏移 -60°~+60°。
• 决策周期与 decision_interval 必须严格一致，否则会出现时间漂移。

6. 强化学习算法实现：PPO

选择 PPO（Proximal Policy Optimization）作为基线算法，原因是它在离散动作空间上稳定、对超参不敏感，且易于并行采样。

6.1 网络结构

import torch
import torch.nn as nn
from torch.distributions import Categorical


class ActorCritic(nn.Module):
    def __init__(self, obs_dim=11, act_dim=15, hidden=128):
        super().__init__()
        self.shared = nn.Sequential(
            nn.Linear(obs_dim, hidden),
            nn.Tanh(),
            nn.Linear(hidden, hidden),
            nn.Tanh(),
        )
        self.actor = nn.Linear(hidden, act_dim)
        self.critic = nn.Linear(hidden, 1)

    def forward(self, obs):
        h = self.shared(obs)
        logits = self.actor(h)
        value = self.critic(h).squeeze(-1)
        return logits, value

    def act(self, obs):
        logits, value = self.forward(obs)
        dist = Categorical(logits=logits)
        action = dist.sample()
        return action, dist.log_prob(action), value

6.2 PPO 训练循环

import torch
import torch.nn as nn
import numpy as np
from torch.optim import Adam


class PPOTrainer:
    def __init__(self, env, lr=3e-4, gamma=0.99, lam=0.95,
                 clip_eps=0.2, epochs=10, batch_size=64):
        self.env = env
        self.gamma = gamma
        self.lam = lam
        self.clip_eps = clip_eps
        self.epochs = epochs
        self.batch_size = batch_size

        self.model = ActorCritic()
        self.optimizer = Adam(self.model.parameters(), lr=lr)

    def collect_rollout(self, steps=2048):
        obs_buf, act_buf, logp_buf = [], [], []
        rew_buf, val_buf, done_buf = [], [], []

        obs, _ = self.env.reset()
        for _ in range(steps):
            obs_t = torch.as_tensor(obs, dtype=torch.float32)
            with torch.no_grad():
                action, logp, value = self.model.act(obs_t)

            next_obs, reward, terminated, truncated, _ = self.env.step(action.item())
            done = terminated or truncated

            obs_buf.append(obs)
            act_buf.append(action.item())
            logp_buf.append(logp.item())
            rew_buf.append(reward)
            val_buf.append(value.item())
            done_buf.append(done)

            obs = next_obs
            if done:
                obs, _ = self.env.reset()

        return (np.array(obs_buf), np.array(act_buf), np.array(logp_buf),
                np.array(rew_buf), np.array(val_buf), np.array(done_buf))

    def compute_gae(self, rewards, values, dones):
        advantages = np.zeros_like(rewards)
        last_gae = 0.0
        for t in reversed(range(len(rewards))):
            if t == len(rewards) - 1:
                next_value = 0.0
            else:
                next_value = values[t + 1]
            non_terminal = 1.0 - dones[t]
            delta = rewards[t] + self.gamma * next_value * non_terminal - values[t]
            last_gae = delta + self.gamma * self.lam * non_terminal * last_gae
            advantages[t] = last_gae
        returns = advantages + values
        return advantages, returns

    def update(self, obs, acts, old_logp, advantages, returns):
        obs_t = torch.as_tensor(obs, dtype=torch.float32)
        acts_t = torch.as_tensor(acts, dtype=torch.long)
        old_logp_t = torch.as_tensor(old_logp, dtype=torch.float32)
        adv_t = torch.as_tensor(advantages, dtype=torch.float32)
        ret_t = torch.as_tensor(returns, dtype=torch.float32)

        adv_t = (adv_t - adv_t.mean()) / (adv_t.std() + 1e-8)
        n = len(obs_t)

        for _ in range(self.epochs):
            idx = np.random.permutation(n)
            for start in range(0, n, self.batch_size):
                batch = idx[start:start + self.batch_size]
                logits, values = self.model(obs_t[batch])
                dist = Categorical(logits=logits)
                new_logp = dist.log_prob(acts_t[batch])
                entropy = dist.entropy().mean()

                ratio = torch.exp(new_logp - old_logp_t[batch])
                surr1 = ratio * adv_t[batch]
                surr2 = torch.clamp(ratio, 1 - self.clip_eps,
                                    1 + self.clip_eps) * adv_t[batch]
                policy_loss = -torch.min(surr1, surr2).mean()
                value_loss = nn.functional.mse_loss(values, ret_t[batch])
                loss = policy_loss + 0.5 * value_loss - 0.01 * entropy

                self.optimizer.zero_grad()
                loss.backward()
                nn.utils.clip_grad_norm_(self.model.parameters(), 0.5)
                self.optimizer.step()

    def train(self, total_iters=100):
        for it in range(total_iters):
            obs, acts, logp, rews, vals, dones = self.collect_rollout()
            adv, rets = self.compute_gae(rews, vals, dones)
            self.update(obs, acts, logp, adv, rets)
            print(f"[iter {it}] mean_reward={rews.mean():.3f}")

7. 奖励函数设计

奖励塑形是 AFSIM 场景下最关键的工程环节。建议采用分层奖励：

def compute_reward(prev_state, cur_state, events):
    reward = 0.0

    # 1. 任务奖励：击毁目标
    if "target_destroyed" in events:
        reward += 100.0
    if "self_destroyed" in events:
        reward -= 100.0

    # 2. 态势奖励：缩短与目标距离
    prev_dist = prev_state["target_distance"]
    cur_dist = cur_state["target_distance"]
    reward += 0.1 * (prev_dist - cur_dist)

    # 3. 生存奖励：被雷达锁定惩罚
    if cur_state["radar_warning"] > 0.5:
        reward -= 0.5

    # 4. 资源惩罚：燃油消耗
    reward -= 0.01 * (prev_state["fuel"] - cur_state["fuel"])

    return reward

设计原则：

• 稀疏任务奖励 + 稠密过程奖励结合，避免智能体长时间无梯度。
• 惩罚项系数要小，防止智能体学会"躺平"（不动就不消耗燃油）。
• 每项奖励单独记录，便于训练时定位问题。

8. 训练调优与常见问题

问题现象	可能原因	解决方向
奖励长期不上升	奖励过于稀疏	增加过程奖励、课程学习
策略震荡	学习率过大	降低 lr、增大 batch
仿真吞吐低	单实例串行	多实例并行 + 向量化环境
训练/验证差距大	场景过拟合	随机化初始态势
动作抖动	决策周期过短	增大 decision_interval

并行采样完整实现：

单实例串行采样是 AFSIM 训练吞吐的主要瓶颈。下面给出一个可运行的完整实现：主进程用 multiprocessing 启动 N 个 AFSIMEnv 子进程，每个子进程独占一个端口，通过 Pipe 与主进程交换 (obs, action, reward)，主进程汇总后拼成 batch 送入 PPOTrainer.update。同时补齐端口分配、进程回收与异常退出清理。

import multiprocessing as mp
import traceback

import numpy as np
import torch
from torch.distributions import Categorical


# ---------- 子进程：每个进程独占一个 AFSIMEnv 与一个端口 ----------
def worker(rank, base_port, conn, max_steps=200, decision_interval=5.0):
    """子进程主循环：接收动作 -> 执行 step -> 回传 (obs, action, reward, done)"""
    env = None
    try:
        # 端口分配：每个 worker 独占 base_port + rank，避免端口冲突
        env = AFSIMEnv(port=base_port + rank,
                       max_steps=max_steps,
                       decision_interval=decision_interval)
        obs, _ = env.reset(seed=rank)
        conn.send(("ready", obs))

        while True:
            cmd, payload = conn.recv()
            if cmd == "step":
                action = int(payload)
                next_obs, reward, terminated, truncated, _ = env.step(action)
                done = terminated or truncated
                conn.send(("step", (obs, action, reward, done)))
                obs = next_obs
                if done:
                    obs, _ = env.reset()
            elif cmd == "reset":
                obs, _ = env.reset(seed=payload)
                conn.send(("reset", obs))
            elif cmd == "close":
                break
    except (EOFError, KeyboardInterrupt):
        pass
    except Exception:
        # 异常时把堆栈回传主进程，便于定位是哪个 worker 挂了
        conn.send(("error", traceback.format_exc()))
    finally:
        if env is not None:
            try:
                env.close()
            except Exception:
                pass
        conn.close()


# ---------- 主进程：管理 N 个子进程并汇总 batch ----------
class ParallelSampler:
    def __init__(self, num_envs=4, base_port=50051,
                 max_steps=200, decision_interval=5.0):
        self.num_envs = num_envs
        self.base_port = base_port
        self.max_steps = max_steps
        self.decision_interval = decision_interval
        self.procs = []
        self.conns = []

    def start(self):
        ctx = mp.get_context("spawn")  # spawn 更安全，避免 fork 复制 socket
        for rank in range(self.num_envs):
            parent_conn, child_conn = ctx.Pipe()
            p = ctx.Process(
                target=worker,
                args=(rank, self.base_port, child_conn,
                      self.max_steps, self.decision_interval),
                daemon=True,
            )
            p.start()
            child_conn.close()  # 父进程只保留自己这一端
            self.procs.append(p)
            self.conns.append(parent_conn)

        # 等待所有 worker 就绪，并拿到初始观测
        init_obs = []
        for conn in self.conns:
            tag, data = conn.recv()
            if tag == "error":
                raise RuntimeError(f"worker 启动失败：\n{data}")
            init_obs.append(data)
        return init_obs

    def step_all(self, actions):
        """向所有 worker 下发动作，收集 (obs, action, reward, done)"""
        for conn, action in zip(self.conns, actions):
            conn.send(("step", action))

        results = []
        for conn in self.conns:
            tag, data = conn.recv()
            if tag == "error":
                raise RuntimeError(f"worker 运行异常：\n{data}")
            results.append(data)
        return results

    def close(self):
        # 先发 close 让子进程优雅退出，再 join，最后强制 terminate 兜底
        for conn in self.conns:
            try:
                conn.send(("close", None))
            except (BrokenPipeError, EOFError):
                pass
        for p in self.procs:
            p.join(timeout=5)
            if p.is_alive():
                p.terminate()
                p.join(timeout=2)
        for conn in self.conns:
            try:
                conn.close()
            except Exception:
                pass
        self.procs.clear()
        self.conns.clear()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False


# ---------- 用并行采样驱动 PPOTrainer.update ----------
def train_parallel(trainer, num_envs=4, total_iters=100, rollout_steps=512):
    with ParallelSampler(num_envs=num_envs) as sampler:
        for it in range(total_iters):
            obs_buf, act_buf, logp_buf = [], [], []
            rew_buf, val_buf, done_buf = [], [], []

            for _ in range(rollout_steps):
                obs_arr = np.stack([r[0] for r in sampler.step_all(
                    [0] * num_envs)]) if False else None  # 占位，见下方真实逻辑
                break

            # 真实采样循环：先根据当前 obs 选动作，再统一下发
            obs_list = sampler.start() if it == 0 else obs_list
            for _ in range(rollout_steps):
                obs_t = torch.as_tensor(np.stack(obs_list), dtype=torch.float32)
                with torch.no_grad():
                    logits, values = trainer.model(obs_t)
                    dist = Categorical(logits=logits)
                    actions = dist.sample()
                    logps = dist.log_prob(actions)

                results = sampler.step_all(actions.tolist())
                next_obs_list = []
                for i, (obs, action, reward, done) in enumerate(results):
                    obs_buf.append(obs)
                    act_buf.append(action)
                    logp_buf.append(logps[i].item())
                    rew_buf.append(reward)
                    val_buf.append(values[i].item())
                    done_buf.append(done)
                    next_obs_list.append(obs)
                obs_list = next_obs_list

            obs = np.array(obs_buf)
            acts = np.array(act_buf)
            logp = np.array(logp_buf)
            rews = np.array(rew_buf)
            vals = np.array(val_buf)
            dones = np.array(done_buf)

            adv, rets = trainer.compute_gae(rews, vals, dones)
            trainer.update(obs, acts, logp, adv, rets)
            print(f"[iter {it}] mean_reward={rews.mean():.3f}")

实现要点：

• 端口分配：每个 worker 使用 base_port + rank 独占端口，启动前应确保该端口段未被占用；若 AFSIM 实例由外部拉起，则改为传入预分配的端口列表。
• 进程启动方式：使用 spawn 而非 fork，避免子进程继承父进程已建立的 socket 句柄导致连接错乱。
• 进程回收：close() 先发送 close 指令让子进程自行关闭环境与连接，再 join(timeout=5)；超时未退出则 terminate() 强制回收，防止僵尸进程。
• 异常退出清理：worker 内部用 try/except/finally 捕获异常并把堆栈回传主进程，finally 中关闭环境与管道；主进程收到 error 消息立即抛出，配合 with 上下文保证所有子进程被清理。
• batch 拼接：所有 worker 的 (obs, action, reward, done) 按时间步顺序拼接，与单环境 rollout 的数据布局完全一致，可直接复用 compute_gae 与 update。

9. 总结

本文围绕 AFSIM 与强化学习的结合，给出了从架构设计、环境封装、脚本桥接、PPO 实现到奖励塑形与调优的完整方案。核心经验可以归纳为三点：

1. 接口先行：先把 AFSIM 的状态/动作接口稳定下来，再谈算法。
2. 奖励为王：仿真环境下的奖励设计比算法选择更影响最终效果。
3. 分层训练：用简化环境预热，再用 AFSIM 高保真环境精调。

后续可以进一步探索多智能体协同（MAPPO/QMIX）、基于模仿学习的冷启动，以及仿真加速（并行实例 + GPU 推理）等方向。