exp: 轮次45——tie(等长安全前缀按沿路最小间隙定夺)判掉

两臂 n=256 同 seed 背靠背(配方=轮次42 K8+cert_batch=1+inc_cap=1):到达 0.5234→0.5039
(-1.95pp,逐题配对丢6赢1净-5,McNemar 精确 p=0.125 与零不可分),时限内 +1.17pp、
刹停 -9%、行程比 -0.076、执行碰撞仍为 0(判据未放松)。丢的 6 题 5 题在 A 类(开阔区)
=买余量卖进度的签名。

仪表与离线数交叉验证:改选率 5.58% 对上轮次44 独立测得的"执行点间隙落在 0.2–0.5mm 占
5.6%",并列率 79.0% 对上"满前缀 79.4%" ⇒ 分辨率上界被宽相位截顶 gcap=max(0.5,margin+0.5)
锁死,5.6% 是这个家族在 cap=0.5 下的天花板,不是实现没调好。

结构性发现:#24 作用在错的人群——tie 只在 maxlen>0 触发,而失败主人群是全拒(maxlen=0,
未到达题 22% 的规划);全拒那条已由 backoff(轮次43)判掉。前置条件记为 #28(gcap 与
gap_margin 解耦)。

实现账:新增 tie 与触发计数器(生产端 per 数组+dump 键、消费端汇总 tie_* 同批改完,
tie=0 时 22/22 老键逐位复现);修掉 tie 分支局部名 nmax 顶掉主循环总步数的遮蔽 bug,
守卫单测做过变异检查;计数器正对照可手算(tie_opp=tie_changed=n_plans)。47 项测试绿。
This commit is contained in:
cbb
2026-10-21 00:33:44 +08:00
parent 350f386c02
commit 72c8835229
8 changed files with 564 additions and 5 deletions
+77
View File
@@ -889,5 +889,82 @@ class TravelPenaltySelectTest(unittest.TestCase):
self.assertAlmostEqual(float(per['moving_share'][0]), 2.0 / 3.0, places=5)
class TieByClearanceTest(unittest.TestCase):
"""#24:两条候选的安全前缀一样长时,现在按"沿路最小间隙更大"定夺(零额外几何、判据不变)。"""
class StubModel:
act_scale = 1.0
def __init__(self, act):
self.act = act
def sample_k(self, feat, K, z, steps=1):
B = feat.shape[0]
return (self.act.expand(B, K, 16, 6).clone(), torch.zeros(B, device=feat.device),
torch.zeros(B, 13, device=feat.device))
class ClearKin:
"""间隙 = 0.4 − 0.2·q[0](恒为正 ⇒ 两条候选都全安全;走得越远间隙越小)。全部低于 0.5 的宽相位截顶值,
否则一切都被 clamp 成 0.5 就分不出高低了。"""
def clamp_box(self, q):
return q
def gaps(self, q, cap):
gm = (0.4 - 0.2 * q[:, 0]).clamp(min=0.0)
return {'gm': torch.stack([gm] * 49, 1)}
def run1(self, tie, cert_batch=1):
big = [[0.2, 0, 0, 0, 0, 0]] * 16 # 走得远:间隙小,但分数并列第一
small = [[0.1, 0, 0, 0, 0, 0]] * 16 # 走得近:间隙大
acts = torch.tensor([big, small], dtype=torch.float32)
with patch('ml.sim.closed_loop.features', return_value=torch.zeros(1, 83)), \
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), self.ClearKin(), [{'q0': [0] * 6, 'T': [10, 0, 0]}], 'cpu',
K=2, safety=True, max_sec=0.04, return_traj=True, use_sub=False,
select='endpoint', replan='empty', commit_n=2, safety_partial=2,
cert_batch=cert_batch, tie=tie)
return per
def test_tie_prefers_the_wider_clearance(self):
p0 = self.run1(0) # 分数并列 ⇒ 稳定排序取第 0 条(走得远、余量小)
self.assertAlmostEqual(float(p0['traj'][0, 1, 0]), 0.2 * 1.2, places=5)
p1 = self.run1(1) # 改由余量定夺 ⇒ 取间隙更大的那条(0.1×STEP=0.12°)
self.assertAlmostEqual(float(p1['traj'][0, 1, 0]), 0.1 * 1.2, places=5)
torch.testing.assert_close(p0['brakes'], p1['brakes'], rtol=0, atol=0) # 安全性判据没动
torch.testing.assert_close(p0['plans']['certified'], p1['plans']['certified'], rtol=0, atol=0)
def test_tie_requires_batched_certification(self):
with self.assertRaisesRegex(ValueError, 'cert_batch'):
self.run1(1, cert_batch=0)
def test_tie_counters_are_exact(self):
"""钩子自检的期望值可手算:夹具里两条候选永远全安全 ⇒ 每次规划都是并列,且每次都改选 ⇒
tie_opp = tie_changed = n_plans,两个比率恒 1。tie=0 时两个计数恒 0。"""
p0 = self.run1(0)
self.assertGreater(float(p0['n_plans'][0]), 0) # 夹具得真的跑过规划,否则下面的相等是 0==0 的假通过
torch.testing.assert_close(p0['tie_opp'], torch.zeros_like(p0['tie_opp']), rtol=0, atol=0)
torch.testing.assert_close(p0['tie_changed'], torch.zeros_like(p0['tie_changed']), rtol=0, atol=0)
p1 = self.run1(1)
torch.testing.assert_close(p1['tie_opp'], p1['n_plans'], rtol=0, atol=0)
torch.testing.assert_close(p1['tie_changed'], p1['n_plans'], rtol=0, atol=0)
self.assertAlmostEqual(float(p1['tie_opp_rate'][0]), 1.0, places=6)
self.assertAlmostEqual(float(p1['tie_changed_rate'][0]), 1.0, places=6)
def test_tie_block_does_not_shadow_loop_state(self):
"""tie 分支里新增的局部名一旦顶掉主循环状态(真实踩过 nmax=总步数被改成张量),
进度回调会立刻炸;所以 tie=1 必须带 progress 跑通,且 total 保持标量。"""
seen = []
acts = torch.tensor([[[0.2, 0, 0, 0, 0, 0]] * 16, [[0.1, 0, 0, 0, 0, 0]] * 16], dtype=torch.float32)
with patch('ml.sim.closed_loop.features', return_value=torch.zeros(1, 83)), \
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]))):
rollout(self.StubModel(acts), self.ClearKin(), [{'q0': [0] * 6, 'T': [10, 0, 0]}], 'cpu',
K=2, safety=True, max_sec=0.04, use_sub=False, select='endpoint', replan='empty',
commit_n=2, safety_partial=2, cert_batch=1, tie=1,
progress=lambda step, total: seen.append((step, total)))
self.assertTrue(seen)
self.assertTrue(all(isinstance(s, int) and isinstance(t, int) for s, t in seen))
if __name__ == '__main__':
unittest.main()