- 라이브러리 불러오기
- 랜덤 에이전트가 이용한 환경 정보
- 환경 정의 및 설정
- Behavior 이름 불러오기 및 timescale 설정
- 전체 진행을 위한 Loop
- 에피소드 진행을 위한 Loop
- 누적보상 출력 및 환경 종료
- 전체 코드
- Reference
from mlagents_envs.environment import UnityEnvironment, ActionTuple
from mlagents_envs.side_channel.engine_configuration_channel import EngineConfigurationChannel
import numpy as np
UnityEnvironment
: Unity 환경과 Python Code를 연결해주는 메인 인터페이스- 이를 통해 Python 코드가 Unity 환경의 정보를 받으며, Agent의 Action 정보를 Unity 환경에게 전송
ActionTuple
: numpy array를 Unity에 전송하기 위해 ActionTuple로 변환해서 사용
챌린지 환경인 RL Village에 대한 더 자세한 정보는 RL Village Infomation를 참고해주세요!
- Observation: 6개의 관측 데이터
- 정면 이미지
- 오른쪽 이미지
- 뒤쪽 이미지
- 왼쪽 이미지
- 아래쪽 이미지
- 벡터 관측
- Action: 3개의 continuous 벡터 (range: -1 ~ 1 float)
if __name__ == '__main__':
# 환경 정의 및 설정
engine_configuration_channel = EngineConfigurationChannel()
env = UnityEnvironment(file_name='환경의 경로',
worker_id=np.random.randint(65535),
side_channels=[engine_configuration_channel])
env.reset()
EngineConfigurationChannel
: 환경에 대한 타임 스케일, 해상도, 그래픽 품질 등에 대한 변경 가능UnityEnvironment
를 통해 환경 정의file_name
: 환경의 경로 입력worker_id
: 환경과 통신하기 위한 포트 -> 여러 환경을 사용하는 경우 값을 다르게 설정해야함side_channels
: 강화학습 루프와 상관없는 데이터를 환경과 주고받기 위한 방법을 제공
env.reset()
: 환경 초기화
# behavior 이름 불러오기 및 timescale 설정
behavior_name = list(env.behavior_specs)[0]
engine_configuration_channel.set_configuration_parameters(time_scale=1)
engine_configuration_channel
의set_configuration_parameters
로time_scale
을 1로 설정- 1프레임 마다 한 번씩 화면 업데이트
# 전체 진행을 위한 반복문 (10 에피소드 반복)
for ep in range(10):
# 환경 초기화
env.reset()
# decision_steps와 terminal_steps 정의
decision_steps, terminal_steps = env.get_steps(behavior_name)
# 파라미터 초기화
done = False
ep_rewards = 0
get_steps
:decision_step
과terminal_step
을 tuple로 반환decision_step
: 환경을 에피소드가 진행중인 경우 Agent에 대한 정보를 포함terminal_step
: 환경의 에피소드가 종료된 경우 Agent에 대한 정보를 포함
# 에피소드 진행을 위한 while문
while not done:
# 랜덤 행동 설정
random_action = np.random.randn(len(decision_steps),3)
action_tuple = ActionTuple()
action_tuple.add_continuous(random_action)
env.set_actions(behavior_name, action_tuple)
# 행동 수행
env.step()
# 행동 수행 후 에이전트의 정보 (상태, 보상, 종료 여부) 취득
decision_steps, terminal_steps = env.get_steps(behavior_name)
done = len(terminal_steps.agent_id)>0
reward = terminal_steps.reward[0] if done else decision_steps.reward[0]
if done:
next_state = [terminal_steps.obs[i][0] for i in range(6)]
else:
next_state = [decision_steps.obs[i][0] for i in range(6)]
# 매 스텝 보상을 에피소드에 대한 누적보상에 더해줌
ep_rewards += reward
- Random Action 설정: Action 수 = 3
action_tuple.add_continuous(random_action)
:ActionTuple
을 이용하여random_action
을 변환
env.step()
: 1 번의 Action 실행- 에피소드 종료 여부에 따라
- 에피소드가 종료된 (
done=True
)인 경우terminal_steps
의 정보를 취득 - 에피소드를 진행중인 경우 (
done=False
)decision_step
의 정보를 취득
- 에피소드가 종료된 (
# 누적 보상 출력
print('total reward for ep {} is {}'.format(ep, ep_rewards))
# 환경 종료
env.close()
from mlagents_envs.environment import UnityEnvironment, ActionTuple
from mlagents_envs.side_channel.engine_configuration_channel import EngineConfigurationChannel
import numpy as np
if __name__ == '__main__':
# 환경 정의 및 설정
engine_configuration_channel = EngineConfigurationChannel()
env = UnityEnvironment(file_name='환경의 경로',
worker_id=np.random.randint(65535),
side_channels=[engine_configuration_channel])
env.reset()
# behavior 이름 불러오기 및 timescale 설정
behavior_name = list(env.behavior_specs)[0]
engine_configuration_channel.set_configuration_parameters(time_scale=1)
# 전체 진행을 위한 반복문 (10 에피소드 반복)
for ep in range(10):
# 환경 초기화
env.reset()
# decision_steps와 terminal_steps 정의
decision_steps, terminal_steps = env.get_steps(behavior_name)
# 파라미터 초기화
done = False
ep_rewards = 0
# 에피소드 진행을 위한 while문
while not done:
# 랜덤 행동 설정
random_action = np.random.randn(len(decision_steps),3)
action_tuple = ActionTuple()
action_tuple.add_continuous(random_action)
env.set_actions(behavior_name, action_tuple)
# 행동 수행
env.step()
# 행동 수행 후 에이전트의 정보 (상태, 보상, 종료 여부) 취득
decision_steps, terminal_steps = env.get_steps(behavior_name)
done = len(terminal_steps.agent_id)>0
reward = terminal_steps.reward[0] if done else decision_steps.reward[0]
if done:
next_state = [terminal_steps.obs[i][0] for i in range(6)]
else:
next_state = [decision_steps.obs[i][0] for i in range(6)]
# 매 스텝 보상을 에피소드에 대한 누적보상에 더해줌
ep_rewards += reward
# 누적 보상 출력
print('total reward for ep {} is {}'.format(ep, ep_rewards))
# 환경 종료
env.close()