把卡牌游戏的考试关卡变成 Gym 环境

游戏自动化里，能写规则就尽量写规则——便宜、可控、好调。但有些场景规则真写不动。

比如卡牌游戏的考试阶段，出牌顺序、资源分配会被手牌、对手压制、剩余回合等十几个变量同时拉扯，写 if-else 写到最后你只想砸键盘。

所以这个项目里把”培育”和”考试”两段都封成了 Gymnasium 环境，让 PPO 自己去摸索策略。训练好的模型通过 HTTP 服务对外暴露，主程序跑自动化的时候按需 query。下面是中间几个选择背后的考虑。

观测空间

游戏状态怎么塞进神经网络，是最头疼的一步。最后定的是三段结构：全局状态、每个动作的特征、动作掩码。

import gymnasium as gym
from gymnasium import spaces

class GameExamEnv(gym.Env):
    def __init__(self):
        super().__init__()
        self.global_dim = 60          # 全局状态维度
        self.action_feature_dim = 100 # 每个动作的特征维度
        self.max_actions = 50         # 手牌 + 饮料 + 结束回合

        self.observation_space = spaces.Dict({
            'global': spaces.Box(-20.0, 20.0, shape=(self.global_dim,), dtype=np.float32),
            'action_features': spaces.Box(
                -20.0, 20.0,
                shape=(self.max_actions, self.action_feature_dim),
                dtype=np.float32,
            ),
            'action_mask': spaces.Box(0.0, 1.0, shape=(self.max_actions,), dtype=np.float32),
        })

“每个动作一个向量”这种 action_features 设计，是为了让模型在动作集变化时还能复用——手牌每回合都不一样，但每张牌的特征结构是统一的。

全局状态

全局状态主要装”剩余进度”和”当前资源”。为了让训练稳一点，所有值都归一化到大致 [0, 1]：

def _global_observation(self) -> np.ndarray:
    state = self.runtime.state

    global_values = {
        # 进度
        'step_ratio': state['step'] / max(self.scenario.steps, 1),
        'remaining_step_ratio': max(self.scenario.steps - state['step'], 0) / max(self.scenario.steps, 1),
        'audition_progress': state['audition_index'] / max(len(self.scenario.audition_sequence), 1),

        # 资源
        'stamina_ratio': state['stamina'] / max(state['max_stamina'], 1.0),
        'produce_point_ratio': state['produce_points'] / 150.0,
        'fan_vote_ratio': state['fan_votes'] / 5000.0,

        # 三维参数
        'vocal_ratio': state['vocal'] / parameter_scale,
        'dance_ratio': state['dance'] / parameter_scale,
        'visual_ratio': state['visual'] / parameter_scale,

        # 卡组
        'deck_quality': state['deck_quality'] / 20.0,
        'drink_quality': state['drink_quality'] / 10.0,
        'deck_size_ratio': len(self.runtime.deck) / 40.0,
    }

    return np.array(
        [float(global_values[name]) for name in self.global_feature_names],
        dtype=np.float32,
    )

那个 60 维不是精心算出来的——是把”模型可能用得上的状态量”全往里塞之后凑出的。事后看，估计有一半是冗余的，但训练效果还能看，就懒得动了。

每个动作的特征

每个候选动作（出哪张牌、喝啥饮料、结束回合）都编成定长向量。类型 one-hot，效果 one-hot，剩下都是数值：

def _candidate_feature(self, candidate) -> np.ndarray:
    action_type_vec = self.taxonomy.encode_actions([candidate.action_type])
    effect_types = self._produce_effect_types(candidate)
    effect_vec = self.taxonomy.encode_produce_effects(effect_types)

    numeric = np.array([
        candidate.stamina_delta / max(state['max_stamina'], 1.0),
        candidate.produce_point_delta / 100.0,
        candidate.success_probability,
        len(candidate.produce_effect_ids) / 8.0,
        1.0 if candidate.available else 0.0,
        state['stamina'] / max(state['max_stamina'], 1.0),
    ], dtype=np.float32)

    return np.concatenate([action_type_vec, effect_vec, numeric]).astype(np.float32)

动作掩码

非法动作（已经打掉的牌、体力不够发动的牌）必须屏蔽。不然模型会傻乎乎地一直选它，然后撞墙撞个没完：

def action_masks(self) -> np.ndarray:
    return np.array([
        bool(candidate.payload.get('available', False))
        for candidate in self._candidates
    ], dtype=bool)

用的是 sb3-contrib 的 MaskablePPO，掩码直接进 policy 网络，违法动作的概率会被强行压成 0——干净利落。

课程学习

你要是头铁，直接把模型扔进最难的”NIA Master 全流程”开训，那场面挺惨的——它一路吃负奖励，啥也学不会，跟个迷茫的萌新一头撞墙似的。

所以搞了套课程，从最简单的”初中间考试”起步，一关一关往上爬：

CURRICULUM_STAGES = [
    "初中间考试",
    "初最终考试",
    "NIA中间考试",
    "NIA最终考试",
    "NIA选拔",
    "初Regular全流程",
    "初Master全流程",
    "NIA Pro全流程",
    "NIA Master全流程",
]

每个阶段训若干 timesteps，然后评估、存 checkpoint，再进下一关：

def run_curriculum(self, stages, timesteps_per_stage=131072):
    for stage_name in stages:
        env = self._create_env_for_stage(stage_name)
        model = MaskablePPO("MultiInputPolicy", env, verbose=1)
        model.learn(total_timesteps=timesteps_per_stage)
        quality = self._evaluate_model(model, stage_name)
        self._save_checkpoint(model, stage_name, quality)

timesteps_per_stage=131072 是 2^17，没啥特殊原因，纯属 PPO 跑下来手感比较顺的一个量级。

自举训练

没人类示范数据，也不想花钱让 GPT 当老师。于是搞了一套”自己教自己”：多个 checkpoint 在固定 seed 上各跑一遍，挑表现最好的轨迹做行为克隆，再 RL 微调一把：

def self_bootstrap(self, iterations=3):
    for iteration in range(iterations):
        # 多 checkpoint 跑固定种子
        candidates = []
        for checkpoint in self.checkpoints:
            for seed in self.evaluation_seeds:
                episode = self._run_episode(checkpoint, seed)
                candidates.append(episode)

        # 排序选最优
        candidates.sort(key=lambda c: c.quality_key(), reverse=True)
        best_trajectories = candidates[:10]

        # BC 蒸馏
        bc_model = self._bc_train(best_trajectories)

        # RL 微调
        rl_model = self._rl_finetune(bc_model, timesteps=50000)

        # 谁稳留谁
        if self._evaluate(rl_model) > self._evaluate(bc_model):
            self.best_model = rl_model
        else:
            self.best_model = bc_model

排序 key 这事得多说一句——优先级是”无效动作越少越好 > 排名越高越好 > 分数越高越好”。

为啥不把”分数高”放第一位？因为那样会挑出一堆”靠运气打高分”的轨迹，BC 学完之后会被这些路径带歪，模型反倒更不稳了。

@dataclass
class EpisodeCandidate:
    checkpoint_path: Path
    seed: int
    records: list
    total_reward: float
    terminal_score: float
    invalid_actions: int
    clear_rank: int

    def quality_key(self):
        return (
            -int(self.invalid_actions),
            int(self.clear_rank),
            float(self.terminal_score),
            float(self.total_reward),
        )

推理服务化

训练好的模型没塞进主程序——stable_baselines3 那套依赖太重，整进 GUI 应用包能膨胀得吓人。

所以单独起了个 FastAPI 服务，主程序通过 HTTP 调：

from fastapi import FastAPI
from sb3_contrib import MaskablePPO

app = FastAPI()
model = MaskablePPO.load("best_model.zip")

@app.post("/api/inference/predict")
async def predict(request: PredictRequest):
    obs = build_observation(request.state, request.legal_actions)
    action_mask = build_action_mask(request.legal_actions)

    action_value, _ = model.predict(
        obs,
        deterministic=True,
        action_masks=action_mask,
    )

    action_index = int(action_value)
    return {
        "action_index": action_index,
        "action_id": request.legal_actions[action_index]["action_id"],
        "confidence": float(action_value),
    }

主程序这边就是个朴素 HTTP 客户端，超时定短点，失败了就 fallback 到规则策略：

class RLInferenceClient:
    def __init__(self, base_url="http://127.0.0.1:8100"):
        self._base_url = base_url
        self._session = requests.Session()

    def predict(self, exam_state, legal_actions, deterministic=True):
        payload = {
            **exam_state,
            "legal_actions": legal_actions,
            "deterministic": deterministic,
        }
        try:
            resp = self._session.post(
                f"{self._base_url}/api/inference/predict",
                json=payload,
                timeout=10.0,
            )
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as exc:
            logger.warning(f"RL 推理请求失败: {exc}")
            return None

score 模式 vs clear 模式

考试有两个互相打架的目标：分数尽量高、能否过线。一套 reward 配置很难两头都讨好，所以干脆分了两套：

def build_reward_config(mode: str) -> RewardConfig:
    if mode == "score":
        # 拼分数
        return RewardConfig(
            score_weight=1.0,
            clear_bonus=0.5,
            invalid_action_penalty=-0.25,
        )
    elif mode == "clear":
        # 保过线
        return RewardConfig(
            score_weight=0.3,
            clear_bonus=2.0,
            overclear_penalty=-0.1,  # 过线后再硬冲会被惩罚
            invalid_action_penalty=-0.25,
        )

overclear_penalty 这个是后加的——一开始 clear 模式训出来的策略，过线了还在硬打、停不下来。因为对它来说反正 reward 越多越好嘛。

加了过线后的负反馈，它才学会”得了得了，过了就停”。

中间最容易踩的一个大坑：reward 一改就得重训。课程学习的 checkpoint 不能跨 reward 配置复用——前期学到的”打高分”行为会污染”保过线”目标。我就是没注意这点，白白浪费了好几个小时的训练时间。