AirSim强化学习入门实战：从环境搭建到智能体训练全流程解析

1次阅读

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

AirSim（Aerial Informatics and Robotics Simulation）作为微软开源的无人机 / 自动驾驶仿真平台，在强化学习领域具有独特优势：

基于 Unreal Engine 的高保真物理引擎
原生支持 Python API 控制
可直接输出相机、激光雷达等传感器数据

但新手常遇到以下问题：

环境依赖复杂，UE4 引擎与 Python 包版本冲突频繁
AirSim API 调用方式与常规 RL 库（如 OpenAI Gym）差异大
训练效率低下，单机资源利用率不足

特性	AirSim	Gazebo
图形质量	高（UE4 渲染）	中等（OGRE 引擎）
物理精度	高（PhysX 引擎）	高（ODE/Bullet）
Python 支持	原生 API	需 ROS Bridge
部署难度	中等（需编译 UE4 插件）	简单（apt 安装）

PPO（Proximal Policy Optimization）因其以下特性成为首选：

支持连续 / 离散动作空间
样本效率高于传统 Policy Gradient
超参数鲁棒性强

CPU：至少 4 核（建议 8 核以上处理物理引擎）
GPU：GTX 1060 起步（需支持 UE4 实时渲染）
内存：16GB 最低配置

从 Epic Games Launcher 安装 UE4（建议 4.26 版本）
下载 AirSim 插件并放入引擎插件目录

关键参数（Settings.json）：

{
  "SettingsVersion": 1.2,
  "SimMode": "Car",  // 或 "Multirotor"
  "CameraDefaults": {
    "CaptureSettings": [{
      "Width": 256,
      "Height": 144,
      "FOV_Degrees": 90
    }]
  }
}

import airsim
import time

# 创建客户端
client = airsim.CarClient()  # 无人机使用 MultirotorClient
client.confirmConnection()  # 超时自动重试

# 获取状态示例
state = client.getCarState()
print("Speed:", state.speed)

# 异常处理
try:
    client.enableApiControl(True)
except airsim.exceptions.VehicleException as e:
    print("API 控制失败:", str(e))

import gym
from gym import spaces
import numpy as np

class AirSimEnv(gym.Env):
    def __init__(self):
        # 动作空间（转向, 油门）self.action_space = spaces.Box(low=np.array([-1, 0]),  
            high=np.array([1, 1]), dtype=np.float32)

        # 观察空间（前置摄像头 RGB）self.observation_space = spaces.Box(low=0, high=255, shape=(144, 256, 3), dtype=np.uint8)

    def step(self, action):
        client.setCarControls({"steering": action[0],
            "throttle": action[1]
        })
        time.sleep(0.1)  # 等待物理更新
        return self._get_obs(), reward, done, {}

import torch
from torch import nn

class PPONetwork(nn.Module):
    def __init__(self):
        super().__init__()
        self.cnn = nn.Sequential(nn.Conv2d(3, 32, 8, stride=4),  # [3,144,256] -> [32,35,63]
            nn.ReLU(),
            nn.Conv2d(32, 64, 4, stride=2),
            nn.ReLU())
        self.policy_head = nn.Linear(64*9*15, 2)  # 输出均值和方差

    def forward(self, x):
        x = x.float() / 255.0  # 归一化
        features = self.cnn(x.permute(0,3,1,2))
        return self.policy_head(features.flatten(1))

from multiprocessing import Process, Queue

def worker(input_q, output_q):
    env = AirSimEnv()
    while True:
        action = input_q.get()
        obs, reward = env.step(action)
        output_q.put((obs, reward))

# 主进程
processes = [Process(target=worker, args=(in_q, out_q)) 
             for _ in range(4)]