推理策略

策略把策略预测出的动作chunk转成机器人消费的逐步命令流。EVA内置五种——sync / async / naive / act / rtc——可在配置文件里指定，也可在控制台里实时切换。

流水线

端到端：观测 → 策略返回chunk → 推理策略合并进动作缓冲区 → 每个周期弹出一条命令 → 机器人。五种策略的差别仅在合并这一步。实时切换在DEBUG中完成（见DEBUG）。

五种策略

键	行为	适用场景
`sync`	阻塞式。取chunk、执行其最前面几步、再取下一个。无后台线程、不平滑；每次预测时机器人短暂停顿。	最简单、最可预期的行为；可接受短暂停顿。
`async`	后台预测；在新旧chunk的重叠区做线性交叉淡化。默认。	连续平滑动作——通常的选择。
`naive`	后台预测；新chunk整体替换缓冲区，轨迹直接跳到最新预测。	需要立刻用上最新预测；可接受轻微跳变。
`act`	后台预测；ACT式时序集成，对重叠的chunk按步加权平均。	需要最平滑、最经过平均的动作；模型预测偏抖。
`rtc`	后台预测，配合Real-Time Chunking：把新chunk对齐到已在执行中的动作，预测速率较高时仍可无缝衔接。	策略服务器较快；需对延迟敏感的衔接。

act和naive是独立策略，不是async的子选项。切换会替换整个合并行为。

后台预测循环

只有sync在策略上阻塞。其余四种在后台持续预测：工作单元按推理速率（默认3.0Hz）取新chunk，裁剪到最前面的exec steps步，再合并进缓冲区。控制循环每个周期弹出一条已就绪的命令，从不阻塞。无新观测的轮次直接跳过；工作单元按设定速率自我配速，而非全速运行。

为什么有两种速率？策略每秒产出几次chunk；机器人需要数十Hz的命令。缓冲区填平这道落差。

合并行为

合并方式	使用者	效果
交叉淡化（线性重叠）	`async`、`rtc`	命令在重叠区从旧轨迹线性过渡到新轨迹。平滑，无可见跳变。
替换	`naive`	丢弃队列、用新chunk重新填满。响应即时；衔接处可能有跳变。
平均（时序集成）	`act`	保留最近多个重叠chunk；每一步取它们对该步预测的加权平均。非常平滑，但灵敏度略低。

三种方式都会先从新chunk的前端丢掉若干已过时的步：等预测返回时机器人已往前走。这一前端裁剪就是latency k旋钮。

建议配图。交叉淡化／替换／平均三种行为用一张「轨迹随时间」的小图会更清晰。

可调参数

每种策略暴露若干旋钮，作为DEBUG调参面板里的实时输入框；填值后按APPLY立即生效。同一面板也调整全局预测速率。

旋钮（控制台标签）	适用策略	含义 · 取值范围 · 默认值
`execute_horizon`（`exec steps`）	全部	取下一个chunk前每个chunk保留的最前几步。整数 ≥ 1。不填⇒执行整个chunk；`sync`预设出厂为`5`。越小⇒越快响应新预测；越大⇒越久依赖同一chunk。
`inference_rate`	async / naive / act / rtc	后台取chunk的速率，Hz。默认`3.0`。仅当策略服务器跟得上时调高。
`latency_k`（`latency k`）	async / naive / rtc	从新chunk前端裁掉的过时步数上限。整数 ≥ 0。`naive`预设出厂为`4`；基础默认为不裁。越大⇒跳得更靠前；过大⇒会跳过真实动作。
`exp_weight_m`	act	较旧预测在平均中的衰减速率。数值 ≥ 0，默认`0.01`。越大⇒最旧的幸存预测越主导；越小⇒平均越均匀。

latency_k由async、naive、rtc共享，不绑定任何一种合并方式。

APPLY在下一轮预测时生效（默认速率下亚秒级）。非法值会弹回原值或在面板报错——请检查取值范围。机器人若完全不动，最常见原因是策略服务器，请查看DEBUG日志。

选择与切换策略

实时：DEBUG的策略选择器可在运行中在五个键之间切换。见DEBUG。

配置：默认值在共享配置里；你的部署配置只覆盖你想改的旋钮。从configs/01_deploy/<robot>/（例如openpi_qpos.py）起步，加上一个inference_strategies块：

python

# 仅覆盖 async/rtc 的 latency_k，其余字段保持共享默认值。
inference_strategies = {
    'async': dict(args=dict(
        latency_k=<步数>,   # 整数 >= 0
    )),
    'rtc': dict(args=dict(
        latency_k=<步数>,
    )),
}

其他旋钮以同样方式放进同一args——任意键上的execute_horizon=<步数>，或act上的exp_weight_m=<数值>。各键完整字段清单见配置页。

评测运行在评测配置里通过eval.inference_strategy='async'（五个键中任一）固定一种策略；位置见配置。

RTC对齐由策略后端实现，不在推理策略中。rtc策略仅调度后台循环；真正的对齐由openpi_rtc策略后端完成。见策略后端。

如何添加新的推理策略

推理策略放在src/strategy/下。任何固定Hz的异步变体都继承strategy.base_strategy里的BackgroundLoopInferStrategy，共享的后台循环、chunk裁剪、缓冲区接线都已就绪；要写阻塞式同步模式则继承BaseInferStrategy。dataclass字段就是YAML args里的键。

新建src/strategy/<your>_strategy.py。继承BackgroundLoopInferStrategy（或BaseInferStrategy），把要新增的参数声明为dataclass字段；若需要别的合并方式，覆写_make_buffer()返回不同的buffer。
用@STRATEGY_REGISTRY.register("YourInferStrategy")装饰类，STRATEGY_REGISTRY从core.registry导入。注册名就是配置里type=要写的字符串。
无需手写import样板——src/strategy/__init__.py启动时会自动import包内的每一个子模块，文件放进去即可。
在部署配置的inference_strategies里加一条："your_key": dict(type="YourInferStrategy", args=dict(...))。短键your_key就是DEBUG策略选择器和eval.inference_strategy使用的标识。

python

# src/strategy/your_strategy.py
from __future__ import annotations
import dataclasses
from typing import ClassVar

from core.registry import STRATEGY_REGISTRY
from strategy.base_strategy import BackgroundLoopInferStrategy


@STRATEGY_REGISTRY.register("YourInferStrategy")   # ← 此名对应配置里的 type=
@dataclasses.dataclass
class YourInferStrategy(BackgroundLoopInferStrategy):
    """一句话描述你的合并行为。"""

    your_knob: float = 0.5                       # ← 自动成为 YAML args 里的键

    _thread_name: str = dataclasses.field(default="eva-your-loop", init=False)
    _log_label: str = dataclasses.field(default="your", init=False)

    # ← 控制台调参面板按此清单渲染实时输入框
    tune_fields: ClassVar[list[dict]] = [
        {"key": "execute_horizon", "label": "exec steps", "min": 1, "step": 1},
        {"key": "latency_k",       "label": "latency k",  "min": 0, "step": 1},
        {"key": "your_knob",       "label": "your knob",  "min": 0, "step": 0.01},
    ]

    def __post_init__(self) -> None:
        self.your_knob = max(0.0, float(self.your_knob))
        super().__post_init__()                  # ← 父类负责钳制 inference_rate / latency_k 并建 buffer

    # 想换合并方式就覆写 _make_buffer() 返回别的 buffer。
    # 默认的固定Hz取chunk逻辑不够用时再覆写 _inference_loop()。

要写纯同步变体，改继承BaseInferStrategy并覆写take_or_fetch_chunk即可——start_loop/pop_next_action用父类的空实现就行。仓库里现成的样例——async_strategy.py、naive_strategy.py、act_strategy.py、rtc_strategy.py——都和base_strategy.py放在一起；具体的合并语义则在strategy/action_buffer.py里的buffer类中。