zhengkid/AutoTTS 源码: https://github.com/zhengkid/AutoTTS # AutoTTS LLMs 改进 LLMs：用于测试时扩展的智能体发现 Tong Zheng, Haolin Liu, Chengsong Huang, Huiwen Bao, Sheng Zhang, Rui Liu, Runpeng Dai, Ruibo Chen, Chenxi Liu, Tianyi Xiong, Xidong Wu, Hongming Zhang, Heng Huang UMD · UVA · WUSTL · UNC · Google · Meta 项目主页 (https://zhengkid.github.io/AutoTTS-web/) AutoTTS 将 TTS（测试时缩放）策略设计从手工启发式规则重构为环境驱动的自动搜索：人类仅需构建一个离线回放环境（状态、动作、反馈、目标），而一个编码智能体则在其中迭代提出并完善基于代码定义的控制器——仅进行代码编辑，无梯度更新。 低成本：0 次 LLM 调用，完全基于回放。 快速链接： 安装 · 复现 · 引用 ## 高亮成果 - 与 SC@64 相比，在 β ≈ 0.5 时节省了约 69.5% 的 tokens；在四种骨干模型规模下，保留集（held-out）的平均准确率与 SC@64 持平。 - 一次完整发现运行的估计货币成本为 $39.9。 - 相同运行的墙钟时间为 160 分钟。 - 发现评估期间 0 次 LLM 调用（仅回放缓存片段）。 发现的控制器是置信度动量控制器 (CMC, Confidence Momentum Controller)，其特点包括基于趋势的停止、耦合的宽度-深度控制、对齐感知的深度分配以及保守的分支放弃。 — ## 问题设定 我们将自适应测试时推理视为在固定长度区间内的分支上分配有限预算。 第 t 步的状态： text s_t = (q, m_t, I_t, l_t, Ω_t) q：问题；m_t：实例化的分支数量；I_t：活动分支集合；l_t：深度向量；Ω_t：已揭示的探测三元组。 可行动作 A(s_t)： - BRANCH — 通过第一个区间打开新分支。 - CONTINUE(i) — 将分支 i 前进一步区间。 - PROBE(i) — 揭示 ω_{i,l} 而不推进深度。 - PRUNE(i) — 停用分支 i；深度和过去的探测仍被记录。 - ANSWER — 终止并应用控制器的终端聚合器。 成本（以区间为单位）： text Cost(s_t) = Σ_i l_{t,i} + κ_probe · |Ω_t| (通常 κ_probe = 0) 目标。 一个代码定义的策略 π(· | s, β) 由一个标量元参数 β 参数化，该参数确定性调度所有内部超参数。在任务 (q, y) ~ D 上： text max_{π, β} E_{q,y}[ 1{ŷ_{π,β}(q) = y} − γ · C_{π,β}(q) ] 外部循环搜索 π 的实现。每个候选方案都在离线缓存上进行回放评估；轨迹和缩放曲线进入下一轮的历史记录中。 — ## 环境构建（每对 (模型, 基准) 运行一次) 上述 MDP 在发现循环开始之前实例化为一个具体的回放环境： 1. 指定接口。 固定 s_t、A(s_t)、Cost(s_t) 以及准确率-成本目标。 2. 离线轨迹收集。 对于每个查询，从骨干模型中抽取 N 条并行独立的推理轨迹（先获取完整字符串），然后将每条轨迹划分为长度为 Δ tokens 的固定长度片段，并枚举带有探测响应 ω_{i,k} 的分支前缀 z_{i,k}。 3. 实现回放存储。 每个环境转换都查阅归档表；例如，PROBE(i) 检索缓存的 ω_{i,k}，无需任何新的解码。 4. 移交给发现。 候选控制器仅通过 observe/step 进行模拟。渐进式评估成本主要由表格回放主导。步骤 1-3 仅运行一次。仅在回放存储冻结后，才开始迭代编码智能体发现。 在本仓库中： - efficient_reasoning_controller/workspace/code_base/environment/ — 搜索集回放存储。 - efficient_reasoning_controller/test_environment/ — 保留集回放存储；从未向提议者暴露。 — ## 发现：β 参数化 & 轨迹反馈 - β 参数化。 每个候选控制器导出一个标量 β 加上从 β 到每个内部旋钮的确定性单调映射。外部搜索简化为扫描 β，消除了仅针对搜索集调整的脆弱阈值。 - 带有执行轨迹的历史增强。 除了每轮的 β-扫描外，我们还存档经验缩放曲线 以及 在回放期间重建的逐动作轨迹。轨迹为探索者提供了细粒度的行为证据，以便在重写代码之前定位缺陷。 — ## 主要结果 AutoTTS 在 AIME24 回放构造上进行优化，并在保留的 AIME25 / HMMT25 基准测试上进行评估，涵盖四种 Qwen3 骨干模型规模。项目主页报告了以下趋势： - 更好的准确率-token 权衡。 发现的控制器通常将经验 Pareto 前沿移出手工基线（如 SC@64, ASC, ESC, Parallel-Probe）之外。 - 保留集泛化。 在 AIME24 上发现的策略迁移到保留集基准测试，在四种骨干模型规模中的三种上平均准确率优于所有手工基线，并在 Qwen3-8B 上保持竞争力。 - β = 0.5 运行点。 与 SC@64 相比，聚合 token 使用量减少约 69.5%，同时在模型间匹配平均保留集准确率。 - β = 1.0 运行点。 在项目主页上八个表格比较单元中的五个中，将峰值准确率推向所有手工基线之上。 扫描 β 可追踪准确率-token 缩放曲线：较大的 β 通常倾向于更高预算、准确率优先的行为，而较小的 β 则倾向于更便宜的推理。 ### 发现过程的演进 轮级轨迹（例如上图中的 t1 -> t5）显示在整个搜索过程中目标值一致地向更好方向移动： - 在搜索基准上，后期轮次提高了准确率同时控制了 token 增长，表明策略结构逐渐改善而非随机波动。 - 在保留集基准上，相同的轨迹保持竞争力且经常改善，表明发现的控制逻辑迁移到了优化分割之外。 - 轨迹反映了无梯度更新的目标寻求代码演化：智能体编辑显式控制器程序，接收基于回放的准确率/成本反馈，并迭代地将行为转向更好的经验权衡。这是 AutoTTS 的关键点：优化是通过在固定回放环境中的迭代程序搜索实现的，而不是通过反向传播或骨干模型参数微调。 — ## 发现的控制器：CMC 发现的控制器命名为置信度动量控制器 (CMC)。其主要机制包括： - 基于趋势的停止。 CMC 维护池置信度的指数移动平均值，并仅在置信度水平高且趋势非负时停止。这避免了在瞬态置信度峰值上停止。 - 耦合的宽度-深度控制。 加宽和加深通过 EMA 增量链接：强烈的置信度增益抑制新分支生成，而停滞或回归触发加宽。 - 对齐感知的深度分配。 最新答案与池获胜者匹配的分支获得额外的探测步骤，将计算集中在新兴共识上，同时仍推进活动分支。 - 保守的分支放弃。 仅在多个轮次持续偏离后才放弃分支，并且至少保留两个活动分支。 这些机制作为代码定义的控制器逻辑实现，并通过与手工基线相同的回放环境进行评估。 显示完整 OptimalController 源码 (CMC, 点击展开) ``python class OptimalController(LLMDesignedMethod): “”“ 置信度动量控制器 (CMC). 核心思想 ——— 所有先前提案 (IBC, SCR, DGCC) 共享相同的基本停止信号： “瞬时” Beta-多数置信度，从当前步骤的已完成答案池中计算。 这容易受到单步置信度峰值的影响：早期相同答案的幸运集群可能在分布稳定之前过早触发门控。 CMC 用动量感知门控替换瞬时置信度门控： - 跟踪过去 T_ema 轮中池置信度的指数移动平均值 (EMA)： ema_conf = alpha * conf + (1 - alpha) * ema_conf - 跟踪最近的改进增量：delta = ema_conf - ema_conf_prev - 当以下两者都满足时触发门控： (a) ema_conf >= conf_thresh (水平要求) (b) delta >= -slack (非恶化动量；slack 是一个小容差，防止在下降信号上停止) 这意味着控制器不能在一轮峰值上停止；EMA 必须高且没有活跃下降。 通过探测年龄优先级进行自适应深度分配 ———————————————— 每个未完成的活跃分支跟踪 probe_count (它收到了多少探测步骤)。 在每轮中，控制器使用按 probe_count 降序排序的优先级队列在活动分支之间分配每轮探测预算 probe_budget 步骤。 投资最多的分支首先得到服务（每个最多 burst_senior 额外步骤），然后剩余预算分配给投资较少的分支。 这将深度集中在最接近完成的分支上，同时仍推进年轻分支，而不是均匀或纯粹对齐偏置分配 (SCR) 或懒惰睡眠 (DGCC)。 三层分支分类 ——————————— 热身之后： - “aligned” (对齐): 最新答案 == pool_winner - “deviant” (偏离): 最新答案 != pool_winner，且不同意 >= 1 轮 - “neutral” (中立): 尚无池获胜者，或第一轮分歧 层级影响每分支探测乘数： aligned -> multiplier = burst_aligned (例如，高 beta 时为 2) neutral -> multiplier = 1 deviant -> multiplier = 1，但如果偏离 >= abandon_patience 轮，则放弃该分支 基于置信度趋势的加宽 ———————–– 加宽（生成新分支）由置信度 趋势 (delta) 是否正且大，或弱/负驱动： - 如果 delta > trend_thresh：置信度加速 -> 不加宽（我们即将停止） - 如果 delta <= trend_thresh：平台期或回归 -> 通过 widen_burst 新分支加宽，直至 max_branch 上限 这将宽度决策直接与加深是否产生证据质量增益耦合，这是先前提案中不存在的反馈回路。 Beta 计划 ———–– 所有超参数都是 [0,1] 区间内单个 beta 的确定性函数。 beta=0 -> 保守（分支少，低 EMA 惯性，更容易停止） beta=1 -> 接近满预算（分支多，高惯性，更难停止） 与先前工作的新颖性 ——————— ASC / ESC：完全读取；无增量探测。 Parallel_Probe：固定队列；瞬时多数；无池/完成区分；无 EMA。 IBC (r0001)：瞬时池置信度门控；均匀 1-步探测；每轮 1-分支加宽；无 EMA 或趋势。 SCR (r0002)：非对称突发（对齐获得更多步骤）；平台触发加宽；瞬时门控；无 EMA。 DGCC (r0003)：双瞬时门控（主要 + 软佐证）；锁定分支的懒惰睡眠；投票差距比例加宽；无 EMA 动量。 CMC：用单个 EMA 动量门控替换所有瞬时门控；引入探测年龄优先级调度（既非均匀也非仅突发对齐）；置信度趋势加宽（既非平台也非投票差距）；三层分类是相对于 DGCC 双门控的自然简化，未增加额外超参数。 “”“ NAME = “optimal_controller” _MAX_BRANCH = 64 _MAX_OUTER = 500 def _schedule(self, beta: float) -> dict: “”“ 所有计划都是 [0,1] 区间内 beta 的光滑解析函数。 单调性： - 控制预算使用的参数 (n_init, max_branch_use, burst_aligned, widen_burst, warm_up, abandon_patience, T_ema) 随 beta 非递减。 - conf_thresh 随 beta 非递减（更难停止 -> 更多预算）。 - trend_thresh 随 beta 非递增（高 beta 下更容易触发加宽 -> 通过更广泛的探索获得更多预算）。 - ema_alpha 随 beta 非递增（较低 alpha = 较慢 EMA = 更多惯性 = 高 beta 下更多预算）。 “”“ b = max(0.0, min(1.0, float(beta))) n_init = max(2, round(2 + 6 * b)) max_branch_use = min(self._MAX_BRANCH, round(4 + 60 * b)) warm_up = max(2, round(2 + 8 * b)) abandon_patience = max(3, round(3 + 9 * b)) T_ema = max(2, round(2 + 6 * b)) ema_alpha = 0.70 - 0.40 * b conf_thresh = 0.85 + 0.12 * b delta_slack = 0.04 - 0.03 * b burst_aligned = max(1, round(1 + 2 * b)) widen_burst = max(1, round(1 + 3 * b)) trend_thresh = 0.04 - 0.03 * b min_complete = max(2, round(2 + 3 * b)) return { “n_init”: n_init, “max_branch_use”: max_branch_use, “warm_up”: warm_up, “abandon_patience”: abandon_patience, “T_ema”: T_ema, “ema_alpha”: round(ema_alpha, 4), “conf_thresh”: round(conf_thresh, 4), “delta_slack”: round(delta_slack, 4), “burst_aligned”: burst_aligned, “widen_burst”: widen_burst, “trend_thresh”: round(trend_thresh, 4), “min_complete”: min_complete, } def init(self, config: Optional[Dict[str, Any]] = None): super().init(config) self._beta = float((config or {}).get(“beta”, 0.5)) sched = self._schedule(self._beta) self.n_init = sched[“n_init”] self.max_branch_use = sched[“max_branch_use”] self.warm_up = sched[“warm_up”] self.abandon_patience = sched[“abandon_patience”] self.T_ema = sched[“T_ema”] self.ema_alpha = sched[“ema_alpha”] self.conf_thresh = sched[“conf_thresh”] self.delta_slack = sched[“delta_slack”] self.burst_aligned = sched[“burst_aligned”] self.widen_burst = sched[“widen_burst”] self.trend_thresh = sched[“trend_thresh”] self.min_complete = sched[“min_complete”] self.trace_recorder = MethodTraceRecorder() def _reset_trace(self) -> None: self.trace_recorder = MethodTraceRecorder() def _trace_step( self, *, event: str, goal: str, step_input: Dict[str, Any], step_output: Any, state: Dict[str, Any], decision: str, ) -> None: self.trace_recorder.add_step( event=event, goal=goal, input=step_input, output=step_output, state=state, decision=decision, ) def get_last_trace(self) -> List[Dict[str, Any]]: return self.trace_recorder.to_list() def solve_with_trace(self, question) -> Dict[str, Any]: answer = self.solve(question) return {“answer”: answer, “trace”: self.get_last_trace()} def _pool_stats(self, completed: List[str]): “”“已完成答案池上的 (winner, top1, top2, conf)。”“” if not completed: return None, 0, 0, 0.0 winner, top1, top2, _ = _vote_stats(completed) conf = _beta_majority_confidence(top1, top2) return winner, top1, top2, conf def _update_ema(self, ema_prev: float, new_val: float) -> float: “”“EMA 更新：ema = (1 - alpha) * ema_prev + alpha * new_val。”“” return (1.0 - self.ema_alpha) * ema_prev + self.ema_alpha * new_val def _classify_branch( self, br: Dict[str, Any], pool_winner, warm_enough: bool, ) -> str: if not warm_enough or pool_winner is None: return “neutral” if br[“latest_ans”] == pool_winner: return “aligned” return “deviant” def _probe_branch( self, question, br: Dict[str, Any], completed_answers: List[str], n_steps: int, ) -> None: “”“探测分支 br 多达 n_steps 步；记录完成。”“” for _ in range(n_steps): if br[“finished”]: break out = _safe_probe_more(question, br[“index”]) if out is None: br[“finished”] = True if br[“latest_ans”] is not None: completed_answers.append(br[“latest_ans”]) break new_ans, is_finish = out br[“probe_count”] += 1 br[“latest_ans”] = new_ans br[“finished”] = is_finish if is_finish: completed_answers.append(new_ans) break def solve(self, question) -> Optional[str]: self._reset_trace() self._trace_step( event=“start”, goal=“initialize CMC run”, step_input={“beta”: self._beta}, step_output=“initialized”, state={ “n_init”: self.n_init, “max_branch_use”: self.max_branch_use, “warm_up”: self.warm_up, “abandon_patience”: self.abandon_patience, “T_ema”: self.T_ema, “ema_alpha”: self.ema_alpha, “conf_thresh”: self.conf_thresh, “delta