[ PROMPT_NODE_27070 ]
environments
[ SKILL_DOCUMENTATION ]
# PufferLib 环境指南
## 概述
PufferLib 提供了用于创建高性能自定义环境的 PufferEnv API,以及包含 20 多个预构建环境的 Ocean 套件。环境支持单智能体和多智能体场景,并具备原生向量化功能。
## PufferEnv API
### 核心特性
PufferEnv 通过原地(in-place)操作实现高性能:
- 观测值、动作和奖励从共享缓冲区对象初始化
- 所有操作原地进行,避免创建和复制数组
- 原生支持单智能体和多智能体环境
- 平铺的观测/动作空间以实现高效向量化
### 创建 PufferEnv
python
import numpy as np
import pufferlib
from pufferlib import PufferEnv
class MyEnvironment(PufferEnv):
def __init__(self, buf=None):
super().__init__(buf)
# 定义观测和动作空间
self.observation_space = self.make_space({
'image': (84, 84, 3),
'vector': (10,)
})
self.action_space = self.make_discrete(4) # 4 个离散动作
# 初始化状态
self.reset()
def reset(self):
"""将环境重置为初始状态。"""
# 重置内部状态
self.agent_pos = np.array([0, 0])
self.step_count = 0
# 返回初始观测
obs = {
'image': np.zeros((84, 84, 3), dtype=np.uint8),
'vector': np.zeros(10, dtype=np.float32)
}
return obs
def step(self, action):
"""执行一个环境步。"""
# 根据动作更新状态
self.step_count += 1
# 计算奖励
reward = self._compute_reward()
# 检查回合是否结束
done = self.step_count >= 1000
# 生成观测
obs = self._get_observation()
# 附加信息
info = {'episode': {'r': reward, 'l': self.step_count}} if done else {}
return obs, reward, done, info
def _compute_reward(self):
"""计算当前状态的奖励。"""
return 1.0
def _get_observation(self):
"""从当前状态生成观测。"""
return {
'image': np.random.randint(0, 256, (84, 84, 3), dtype=np.uint8),
'vector': np.random.randn(10).astype(np.float32)
}
### 观测空间
#### 离散空间
python
# 单个离散值
self.observation_spa