算法、LLM、RL 三种策略共一个接口——一份 Strategy 模式实战

上一篇说我们最后是”启发式 + LLM + RL”三套决策并存。这就引出一个工程问题——怎么让三套决策器共一个接口，让上层调度无脑切换？

答案听上去老套：Strategy 模式。但 LLM 和 RL 这两位选手的”行为习惯”和经典 OO 教科书里的策略可不太一样，硬套模板会被现实教做人。这篇说说我们最后怎么落地的。

为什么必须共一个接口

三套决策器各有各的强项（上一篇详细写了），但调用方——也就是游戏自动化的任务调度——不应该关心今天用的是哪套。它的诉求很朴素：

给我一个当前局面，我告诉你出什么。

如果三套各暴露各的 API，上层就得到处写：

if mode == "algo":
    action = algo_solver.choose(...)
elif mode == "llm":
    action = llm_strategy.ask(prompt=..., context=...)
elif mode == "rl":
    obs = build_observation(...)
    action = rl_policy.predict(obs)

每加一种策略，整个调用链都得改。调度逻辑被三家细节污染，谁都不开心。

接口长什么样

抽象到最后就一个方法：

class BaseStrategy(ABC):
    @abstractmethod
    def decide(self, state: GameState) -> Decision:
        """根据当前游戏状态做决策"""

GameState 是一个 dataclass，封装”当前局面所需的一切”——手牌、状态、回合数、可选动作等。Decision 也是 dataclass，包含”选哪个动作 + 决策理由 + 元数据”。

三个具体实现：

gameplay/strategy/
├── base_strategy.py          # 抽象基类
├── algo_strategy.py          # 启发式
├── llm_strategy.py           # LLM 决策
└── rl_strategy.py            # RL 决策

听起来很标准对吧？真做下来三个具体类各有各的脾气。

algo_strategy：最听话的那个

启发式策略是最贴合 Strategy 模板的——纯函数，输入 state 输出 decision，没有任何隐藏状态：

class AlgoStrategy(BaseStrategy):
    def decide(self, state: GameState) -> Decision:
        best_action = None
        best_eval = -inf
        for action in state.legal_actions:
            ev = evaluate(state, action)
            if ev > best_eval:
                best_eval, best_action = ev, action
        return Decision(
            action=best_action,
            reason=f"evaluation={best_eval:.2f}",
            meta={"all_evals": {...}},
        )

可重入、线程安全、没有外部依赖。教科书都不舍得这么干净的实现。

如果三套都能这么写，这篇就不用发了。

llm_strategy：把”对话感”塞进无状态接口

LLM 自带”对话历史”的概念——你跟它聊得越多，它越懂你。但 Strategy 接口要求无状态——同样的 state 进来，应该产生（可比较的）输出。

这两者本质冲突。

解决办法是把”上下文”显式放进 state 里，而不是藏在 strategy 内部：

@dataclass
class GameState:
    # 当前局面
    hand: list[Card]
    hp: int
    turn: int
    ...
    # 历史上下文（由上层维护，显式传入）
    session_history: SessionHistory
    insights: list[Insight]

session_history 怎么压缩、insights 怎么挑（参考前面两篇 insight 和上下文压缩），都是 state 构造期间完成的，strategy 拿到时已经是”压缩好的快照”。

LLM strategy 内部就变成无状态了：

class LLMStrategy(BaseStrategy):
    def decide(self, state: GameState) -> Decision:
        prompt = self._build_prompt(state)  # 完全由 state 构造
        response = self.llm_client.chat(prompt)
        return self._parse(response, state)

关键约束：strategy 内部不允许存任何跨 decide 调用的状态。所有”历史”“记忆”“上下文”都从外面传进来。这条规矩破了，后面 RL 那位选手就跟你急。

rl_strategy：远程推理服务

RL 这位最特别——训练时它在 PyTorch / SB3 那一套里，部署时神经网络要做实时推理。

最朴素的做法是在 strategy 里直接加载模型：

class RLStrategy(BaseStrategy):
    def __init__(self, ckpt_path):
        self.policy = load_policy(ckpt_path)
    
    def decide(self, state):
        obs = state_to_observation(state)
        action = self.policy.predict(obs)
        return Decision(action=action, ...)

能跑。但很快就被现实教育了——

问题一：模型几百 MB，每个进程都加载一份，主程序内存爆炸。 问题二：模型用 GPU 推理才划算，但主程序在 CPU 机器上跑。 问题三：模型版本升级要重启整个主程序，运维痛苦。 问题四：训练侧和部署侧的 Python 依赖打架——SB3、Ray、Torch 这些跟主程序的依赖经常冲突。

最后改成了远程无状态推理服务：

[主程序]
   ↓ HTTP/RPC（传 observation，收 action）
[RL 推理服务（独立部署）]
   ↓
[加载好的策略网络]

代码层面引入一个 client：

# rl_inference_client.py
class RLInferenceClient:
    def predict(self, observation: dict) -> int:
        resp = self.http.post(self.endpoint, json=observation)
        return resp.json()["action"]

# rl_strategy.py
class RLStrategy(BaseStrategy):
    def __init__(self, client: RLInferenceClient):
        self.client = client
    
    def decide(self, state):
        obs = state_to_observation(state)
        action = self.client.predict(obs)
        return Decision(action=action, ...)

主程序不需要装 PyTorch、不需要管模型版本、不需要 GPU。RL 服务挂了就降级——上层调度发现 RL strategy 拿不到结果，自动 fallback 到 algo strategy。

项目规范文档里有一条规则正好对应：

决策逻辑需要以无状态方式对接到：train/gakumas_rl

这就是为什么早期我们就强制”无状态”——后来要拆服务、要并发、要降级，都靠这条约束才不至于推倒重来。

三种策略凭什么能并存

三个具体实现完全不同——一个纯函数、一个 LLM 调用、一个 RPC 调用——能塞进同一个接口的原因只有一个：state 和 decision 这两个数据契约定得够稳。

GameState 设计原则：

完整：调用方传一次 state，strategy 就能做决策，不需要再去问别的地方
可序列化：dataclass + 基础类型，能 JSON 化（RL 推理服务要走网络）
版本化：加字段用 Optional + 默认值，老的 strategy 不会被新字段搞挂

Decision 设计原则：

统一动作类型：不管哪个 strategy，最后输出的 action 都是同一个枚举
必带 reason：哪怕 RL 给个 “rl_policy_prediction” 也行，调试时要追问
保留 meta：每个 strategy 想塞什么调试信息进去都可以，上层不强制读

这些约束写在 dataclass 里：

@dataclass(frozen=True)
class Decision:
    action: Action               # 必须是已知 Action 枚举
    reason: str                  # 必须有
    meta: dict[str, Any] = field(default_factory=dict)  # 各家自由发挥

frozen=True 是个小心思——decision 一旦产出不允许修改，避免被下游悄悄篡改。

上层调度怎么选

有了三个 strategy，最后还要决定”这次用哪个”。我们的做法很土，但稳：

def get_strategy(scene: Scene, config: Config) -> BaseStrategy:
    # 配置层面的强制指定优先
    if config.force_strategy:
        return STRATEGIES[config.force_strategy]
    
    # 按场景默认
    return SCENE_DEFAULT_STRATEGY[scene]

然后在 SCENE_DEFAULT_STRATEGY 里把每个游戏场景默认用哪套写死：

SCENE_DEFAULT_STRATEGY = {
    Scene.DAILY_TASK: "algo",      # 日常任务，启发式够用
    Scene.HIGH_RANK_EXAM: "rl",    # 高难度，RL 冲分
    Scene.STORY_DIALOGUE: "llm",   # 剧情对话，LLM 选项
    ...
}

加新场景时只需要决定”默认走哪条路”，不动 strategy 实现。这种”配置驱动 + 默认值显式”的模式比”代码里 if-else”清爽太多。

降级链

三个 strategy 在线上不可避免会出问题：

LLM 服务超时 / 限流
RL 推理服务挂了
启发式遇到没考虑过的边界情况

降级链设计：

1
2
3

RL fail → fallback to algo
LLM fail → fallback to algo
algo fail → fallback to "什么都不做"（safe action）

为什么 algo 是兜底——因为它没有外部依赖，纯函数，最不容易挂。这条降级链就一句话：

def decide_with_fallback(state, primary, fallbacks=[algo, safe]):
    for strategy in [primary, *fallbacks]:
        try:
            return strategy.decide(state)
        except StrategyError as e:
            log.warning(f"{strategy} failed: {e}, trying next")
    raise RuntimeError("All strategies failed")

至少三个都挂的情况，我们到目前没遇到过。

复盘几条原则

写完这套之后总结的经验：

接口要早定，并且不许 strategy 内部存状态。 任何”我先在 strategy 里缓存一下”都是定时炸弹——后面拆远程服务、并发调用，全爆。

state 和 decision 的 dataclass 是核心契约，比 strategy 实现重要十倍。 实现可以换，契约不能轻易动。

支持降级链，特别是对外部依赖的 strategy。 LLM 和 RL 都有外部依赖，没有降级等于线上裸奔。

让最简单、最没有依赖的实现做兜底。 启发式这种纯函数策略平时不显眼，关键时候救命。

默认策略选择写成数据，不写成 if-else。 加新场景时改个映射表就好，不要去翻代码。

收个尾

经典 Strategy 模式在教科书里就一页纸——抽象基类、几个实现、上下文持有引用，完事。

放到 LLM 和 RL 这种带外部依赖、带历史状态、带服务化诉求的现实场景里，模式还是那个模式，但周边的工程肉戏一大堆——状态外置、远程推理、降级链、数据契约、配置驱动——任何一块没做好，三套策略并存就变成三套互相打架。

接口设计这事儿，从来不只是写个 ABC 这么简单。