AirSim与Gym强化学习环境集成实战：从环境搭建到避坑指南

1次阅读

共计 3226 个字符，预计需要花费 9 分钟才能阅读完成。

强化学习在机器人控制、自动驾驶等领域展现出巨大潜力，但传统 Gym 环境（如 CartPole、MountainCar）过于简化，难以模拟真实世界的复杂物理交互。开发者常面临三大挑战：

场景真实性不足：简单几何体无法模拟无人机 / 车辆的空气动力学
传感器缺失：缺乏相机、LIDAR 等多模态数据接口
计算效率低下：物理引擎与训练流程耦合导致吞吐量瓶颈

平台	物理精度	传感器支持	扩展性	学习曲线
AirSim	高	相机 /LIDAR/IMU	强	中等
Gazebo	中	需插件扩展	一般	陡峭
CARLA	高	专注自动驾驶	较弱	中等
PyBullet	中	基础传感器	较强	平缓

选择 AirSim 的核心优势：

基于 Unreal/Unity 的逼真渲染
内置无人机 / 车辆动力学模型
原生 Python API 支持
可扩展的多智能体接口

flowchart TD
    A[Gym Env] -->| 动作 | B[AirSimAdapter]
    B -->| 状态 / 奖励 | A
    B <-->|gRPC| C[AirSim Server]
    C <--> D[Unreal Engine]

import gym
from gym import spaces
import airsim
import numpy as np

class AirSimDroneEnv(gym.Env):
    """
    无人机悬停控制环境
    Observation: 位置(x,y,z) + 姿态角(roll,pitch,yaw)
    Action: 各轴向速度(vx,vy,vz) + 偏航速率
    """
    def __init__(self):
        # 连接 AirSim
        self.client = airsim.MultirotorClient()
        self.client.confirmConnection()

        # 定义动作空间（归一化到[-1,1]）self.action_space = spaces.Box(low=np.array([-1,-1,-1,-1]),
            high=np.array([1,1,1,1]),
            dtype=np.float32
        )

        # 定义状态空间
        self.observation_space = spaces.Box(low=np.array([-np.inf]*6),
            high=np.array([np.inf]*6),
            dtype=np.float32
        )

        # 目标位置
        self.target_pos = np.array([0, 0, -5])  

    def _get_obs(self):
        """获取当前状态"""
        state = self.client.getMultirotorState()
        pos = state.kinematics_estimated.position
        orient = state.kinematics_estimated.orientation

        return np.array([
            pos.x_val, pos.y_val, pos.z_val,
            orient.x_val, orient.y_val, orient.z_val
        ])

    def reset(self):
        self.client.reset()
        self.client.enableApiControl(True)
        self.client.armDisarm(True)

        # 起飞到初始高度
        self.client.takeoffAsync().join()
        self.client.moveToZAsync(-5, 1).join()

        return self._get_obs()

    def step(self, action):
        # 动作解归一化
        vx, vy, vz, yaw_rate = action * 2  # 映射到[-2,2]m/s

        # 执行动作（持续时间 0.1 秒）self.client.moveByVelocityBodyFrameAsync(
            vx, vy, vz, 0.1,
            drivetrain=airsim.DrivetrainType.MaxDegreeOfFreedom,
            yaw_mode=airsim.YawMode(True, yaw_rate)
        )

        # 获取新状态
        obs = self._get_obs()

        # 计算奖励（距离目标越近奖励越高）pos_error = np.linalg.norm(obs[:3] - self.target_pos)
        reward = -pos_error  

        # 终止条件：碰撞或超出边界
        done = self.client.simGetCollisionInfo().has_collision

        return obs, reward, done, {}

动作归一化 ：将连续控制量映射到[-1,1] 区间
坐标系转换：注意 AirSim 使用 NED 坐标系（z 轴向下为正）
异步控制 ：使用Async 方法避免阻塞训练循环

gRPC 压缩 ：在settings.json 中启用"Compression": true
批量传输：合并图像 / 点云数据请求
本地模式 ：优先使用LocalHost 连接

# 在环境初始化时设置
self.client.simSetFrameRate(20)  # 与算法步频匹配
self.client.simPause(False)  # 确保不处于暂停状态

对于图像输入，建议使用 AirSim 的 simGetImages() 参数：

responses = client.simGetImages([airsim.ImageRequest("0", airsim.ImageType.Scene, False, False)  # 不保存到磁盘
])
img = np.frombuffer(responses[0].image_data_uint8, dtype=np.uint8)
img = img.reshape(responses[0].height, responses[0].width, 3)
img = cv2.resize(img, (84, 84))  # 降采样到经典 RL 输入尺寸

连接失败 ：检查 AirSim 是否以-RenderOffscreen 模式运行
控制不稳定 ：调整PhysicsEngineLoop 频率（默认 120Hz 可能过高）
内存泄漏：定期调用client.simFlushPersistentMarkers()
同步异常 ：确保每次reset() 后调用simPause(False)
奖励震荡：在奖励函数中加入平滑项（如速度惩罚）

class MultiAgentWrapper(gym.Env):
    def __init__(self, num_agents=2):
        self.agents = [AirSimDroneEnv() for _ in range(num_agents)]

        # 合并观测空间
        self.observation_space = spaces.Tuple([agent.observation_space for agent in self.agents])

    def step(self, actions):
        obs_n, rew_n, done_n = [], [], []
        for agent, action in zip(self.agents, actions):
            o, r, d, _ = agent.step(action)
            obs_n.append(o)
            rew_n.append(r)
            done_n.append(d)

        return obs_n, rew_n, any(done_n), {}