feat+test: select=ep_waste 白费比例惩罚(罚度按当前误差缩放);单测抓到我把净位移对时间轴求和写成了对轴求和(frac 恒 0)
This commit is contained in:
+1
-1
@@ -98,7 +98,7 @@ def main():
|
||||
ap.add_argument('--cert_batch', type=int, default=0, help='1=一次 gaps 调用覆盖全部候选(压最坏情况,典型路径略贵)')
|
||||
ap.add_argument('--steps', type=int, default=1, help='MeanFlow 采样段数;1=线上一口径(历史发布数字全是 1)')
|
||||
ap.add_argument('--z_scale', type=float, default=1.0, help='候选噪声缩放:>1 让 K 条候选更分散(测"全拒是没有备选"还是"真的没路")')
|
||||
ap.add_argument('--len_w', type=float, default=0.0, help="select=ep_len 时行程惩罚权重(mm / 每轴归一 STEP 的累计路程)")
|
||||
ap.add_argument('--len_w', type=float, default=0.0, help="ep_len=罚累计归一行程;ep_waste=罚(1−净/总)×当前误差;0 与 endpoint 逐位相同")
|
||||
ap.add_argument('--backoff', type=int, default=0, help='全拒时把首选候选按 1/2、1/4…缩步重试的档数(需 --cert_batch 1)')
|
||||
ap.add_argument('--max_sec', type=float, default=None)
|
||||
ap.add_argument('--seed', type=int, default=0)
|
||||
|
||||
@@ -284,13 +284,19 @@ def rollout(model, kin, tasks, device, K=8, safety=False, seed=0, max_sec=None,
|
||||
fut = qe[:, None, None] + torch.cumsum(inc[:, :, r:], 2) # [B,K,H-r,6]
|
||||
if select == 'first':
|
||||
score = torch.zeros(B, K, device=device); score[:, 0] = 1
|
||||
elif select in ('endpoint', 'adapt', 'ep_len'):
|
||||
elif select in ('endpoint', 'adapt', 'ep_len', 'ep_waste'):
|
||||
qa = fut[:, :, -1].reshape(B * K, 6)
|
||||
e_end, _ = goal_err(kin, qa, T.repeat_interleave(K, 0), R.repeat_interleave(K, 0))
|
||||
score = -e_end.view(B, K)
|
||||
if select == 'ep_len' and len_w > 0: # 行程惩罚:每轴按 STEP 归一的累计路程。§9.2 要"行程≤规划器×1.10",
|
||||
if select == 'ep_len' and len_w > 0: # 行程惩罚:每轴按 STEP 归一的累计路程。§9.2 要“行程≤规划器×1.10”,
|
||||
ln = ((fut[:, :, 1:] - fut[:, :, :-1]).abs() / step).sum(-1).sum(-1) # 而现在 2.4 倍没人管
|
||||
score = score - len_w * ln
|
||||
if select == 'ep_waste' and len_w > 0: # 轮次 39 的分解证明行程超出主要是同一轴来回抖(换向步占 23.5%、
|
||||
dv = fut[:, :, 1:] - fut[:, :, :-1] # 净位移只占总行程 44.5%),绕远只占一小部分 ⇒ 罚"白费的比例":
|
||||
tot = dv.abs().sum(-2).sum(-1) # 同轴来回会在净位移里相互抵消,所以 waste = 1 − 净/总 ∈ [0,1]。
|
||||
net = dv.sum(-2).abs().sum(-1) # dv 是 [B,K,H-1,6]:时间在第 -2 轴(写错成 -1 就变成"对轴求和",等于总行程)
|
||||
frac = (1.0 - net / tot.clamp_min(1e-6)).clamp(0, 1)
|
||||
score = score - len_w * frac * e.view(B, 1) # 罚度按"当前离目标多远"缩放:近端不会把进度项压死
|
||||
if select == 'adapt' and last is not None:
|
||||
n = min(fut.shape[2], last.shape[1] - (commit_n + 1))
|
||||
w = 0.85 ** torch.arange(n, device=device, dtype=torch.float32)
|
||||
|
||||
+15
-2
@@ -795,7 +795,7 @@ class TravelPenaltySelectTest(unittest.TestCase):
|
||||
return (self.act.expand(B, K, 16, 6).clone(), torch.zeros(B, device=feat.device),
|
||||
torch.zeros(B, 13, device=feat.device))
|
||||
|
||||
def run1(self, len_w):
|
||||
def run1(self, len_w, select='ep_len'):
|
||||
class SafeKin:
|
||||
def clamp_box(self, q):
|
||||
return q
|
||||
@@ -811,7 +811,7 @@ class TravelPenaltySelectTest(unittest.TestCase):
|
||||
patch('ml.sim.closed_loop.goal_err',
|
||||
side_effect=lambda kin, q, T, R: (torch.full((q.shape[0],), 10.0), torch.zeros(q.shape[0]))):
|
||||
_, per = rollout(self.StubModel(acts), SafeKin(), tasks, 'cpu', K=2, safety=True, max_sec=0.06,
|
||||
return_traj=True, use_sub=False, select='ep_len', replan='empty', commit_n=2,
|
||||
return_traj=True, use_sub=False, select=select, replan='empty', commit_n=2,
|
||||
safety_partial=1, len_w=len_w)
|
||||
return per
|
||||
|
||||
@@ -821,6 +821,19 @@ class TravelPenaltySelectTest(unittest.TestCase):
|
||||
p1 = self.run1(1.0) # 加惩罚:行程 10.8 vs 3.6 ⇒ 选匀速那条
|
||||
self.assertAlmostEqual(p1['traj'][0, 1, 0].item(), 0.225 * 1.2, places=5)
|
||||
|
||||
def test_waste_penalty_also_separates_them(self):
|
||||
"""换成"白费比例"(1 − 同轴净位移/累计行程)也必须能分开,且罚度按当前误差缩放:
|
||||
这里 goal_err 恒为 10 ⇒ 抖动条 waste≈0.78、匀速条 0 ⇒ 差 7.8×len_w 的当量。"""
|
||||
p0 = self.run1(0.0, select='ep_waste')
|
||||
self.assertAlmostEqual(p0['traj'][0, 1, 0].item(), 0.9 * 1.2, places=5)
|
||||
p1 = self.run1(1.0, select='ep_waste')
|
||||
self.assertAlmostEqual(p1['traj'][0, 1, 0].item(), 0.225 * 1.2, places=5)
|
||||
|
||||
def test_zero_weight_ep_waste_is_the_old_endpoint_path(self):
|
||||
p0 = self.run1(0.0, select='ep_waste')
|
||||
p1 = self.run1(0.0, select='endpoint')
|
||||
torch.testing.assert_close(p0['traj'], p1['traj'], rtol=0, atol=0)
|
||||
|
||||
def test_zero_weight_is_the_old_endpoint_path(self):
|
||||
"""len_w=0 且 select='ep_len' 必须与 select='endpoint' 逐位相同(不许悄悄改默认行为)"""
|
||||
class SafeKin:
|
||||
|
||||
Reference in New Issue
Block a user