gqt
86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: UTF-8 -*-
|
|
###########################################################################
|
|
# Copyright © 1998 - 2026 Tencent. All Rights Reserved.
|
|
###########################################################################
|
|
"""
|
|
Author: Tencent AI Arena Authors
|
|
|
|
Data definition and GAE computation for Robot Vacuum.
|
|
清扫大作战数据类定义与 GAE 计算。
|
|
"""
|
|
|
|
import numpy as np
|
|
from common_python.utils.common_func import create_cls
|
|
from agent_ppo.conf.conf import Config
|
|
|
|
|
|
# ObsData: feature vector + legal action mask
|
|
# 观测数据：feature 为特征向量，legal_action 为合法动作掩码
|
|
ObsData = create_cls("ObsData", feature=None, legal_action=None)
|
|
|
|
# ActData: sampled action, greedy action, action probabilities, state value
|
|
# 动作数据：action 为采样动作，d_action 为贪心动作，prob 为动作概率，value 为状态价值
|
|
ActData = create_cls(
|
|
"ActData",
|
|
action=None,
|
|
d_action=None,
|
|
prob=None,
|
|
value=None,
|
|
)
|
|
|
|
# SampleData: int values are treated as dimensions by the framework
|
|
# 训练样本数据：字段值为 int 时框架自动按维度处理
|
|
SampleData = create_cls(
|
|
"SampleData",
|
|
obs=Config.DIM_OF_OBSERVATION, # feature vector / 特征向量
|
|
legal_action=Config.ACTION_NUM, # 8D legal action mask / 合法动作掩码
|
|
act=1, # action index / 执行的动作
|
|
reward=Config.VALUE_NUM, # 1D reward / 奖励
|
|
reward_sum=Config.VALUE_NUM, # GAE td-lambda return
|
|
done=1,
|
|
value=Config.VALUE_NUM, # 1D value estimate / 价值估计
|
|
next_value=Config.VALUE_NUM,
|
|
advantage=Config.VALUE_NUM, # 1D GAE advantage / GAE 优势
|
|
prob=Config.ACTION_NUM, # 8D action probabilities / 动作概率
|
|
)
|
|
|
|
|
|
def sample_process(list_sample_data):
|
|
"""Fill next_value and compute GAE advantage.
|
|
|
|
计算 GAE 并填充 next_value。
|
|
"""
|
|
for i in range(len(list_sample_data) - 1):
|
|
list_sample_data[i].next_value = list_sample_data[i + 1].value
|
|
|
|
_calc_gae(list_sample_data)
|
|
return list_sample_data
|
|
|
|
|
|
def _calc_gae(list_sample_data):
|
|
"""Compute advantage and cumulative return using GAE(λ).
|
|
|
|
使用 GAE(λ) 计算优势函数与累积回报。
|
|
"""
|
|
gae = 0.0
|
|
gamma = Config.GAMMA
|
|
lamda = Config.LAMDA
|
|
for sample in reversed(list_sample_data):
|
|
value = _scalar(sample.value)
|
|
reward = _scalar(sample.reward)
|
|
next_value = _scalar(sample.next_value)
|
|
nonterminal = 1.0 - _scalar(sample.done)
|
|
delta = reward + gamma * next_value * nonterminal - value
|
|
gae = delta + gamma * lamda * nonterminal * gae
|
|
sample.advantage = gae
|
|
sample.reward_sum = gae + value
|
|
|
|
|
|
def _scalar(value):
|
|
"""Read the first scalar from numpy/tensor/list values."""
|
|
if hasattr(value, "detach"):
|
|
value = value.detach().cpu().numpy()
|
|
arr = np.asarray(value, dtype=np.float32).reshape(-1)
|
|
return float(arr[0]) if arr.size else 0.0
|