Game AI — LLM Vision + TAS 自动通关系统设计

LLM 看屏幕 · 模拟操控 · TAS 仿真训练 · MBE 闭环进化

创建时间: 2026-02-07 状态: 设计阶段

1. 核心思路

不走传统 RL（训练 CNN + 策略网络），而是：

截屏 → LLM Vision 理解画面 → LLM 推理决策 → 模拟键鼠执行 → 观察结果 → 循环

为什么这条路可行？

TAS 仿真的优势：

2. 系统架构

╔═══════════════════════════════════════════════════════════════════════╗
║                    Game AI TAS System                                ║
║                                                                      ║
║  ┌─────────────────────────────────────────────────────────────────┐  ║
║  │                     主控循环 (Agent Loop)                        │  ║
║  │                                                                 │  ║
║  │   ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌─────────┐  │  ║
║  │   │ ① 感知层  │──►│ ② 决策层  │──►│ ③ 执行层  │──►│ ④ 反馈层 │  │  ║
║  │   │  Screen   │    │  Brain   │    │  Hands   │    │ Memory  │  │  ║
║  │   └──────────┘    └──────────┘    └──────────┘    └────┬────┘  │  ║
║  │        ▲                                               │       │  ║
║  │        └───────────────────────────────────────────────┘       │  ║
║  └─────────────────────────────────────────────────────────────────┘  ║
║                              │                                        ║
║                    ┌─────────┴─────────┐                              ║
║                    │  ⑤ MBE 闭环层     │                              ║
║                    │  (按需调用)        │                              ║
║                    │                   │                              ║
║                    │  · 策略专家知识库  │                              ║
║                    │  · Eval 表现评估  │                              ║
║                    │  · HOPE 策略优化  │                              ║
║                    │  · 训练数据积累   │                              ║
║                    └───────────────────┘                              ║
║                                                                      ║
║  ┌───────────────────────────────────────────────────────────────┐    ║
║  │                   ⑥ TAS 仿真环境                               │    ║
║  │  BizHawk / RetroArch / mGBA / FCEUX / DeSmuME ...            │    ║
║  │  存档管理 · 帧步进 · RAM 读取 · 输入注入 · 加速               │    ║
║  └───────────────────────────────────────────────────────────────┘    ║
╚═══════════════════════════════════════════════════════════════════════╝

3. 六大模块详细设计

3.1 ① 感知层 (Perception) — "AI 的眼睛"

双通道感知：视觉 + 内存

┌─────────────────────────────────────────┐
│              感知层                       │
│                                          │
│  通道 A: LLM Vision (通用, 跨游戏)       │
│  ┌──────────────────────────────────┐    │
│  │  截屏 → 缩放/裁剪 → Base64       │    │
│  │  → 发送给 LLM Vision API         │    │
│  │  → 返回结构化场景描述             │    │
│  └──────────────────────────────────┘    │
│                                          │
│  通道 B: RAM 直读 (精确, TAS 专用)       │
│  ┌──────────────────────────────────┐    │
│  │  读取模拟器内存地址               │    │
│  │  → 解析: HP, 坐标, 分数, 状态    │    │
│  │  → 返回结构化数值数据             │    │
│  └──────────────────────────────────┘    │
│                                          │
│  融合输出: GameState 对象                │
│  {scene, entities, player, metrics...}   │
└─────────────────────────────────────────┘

通道 A — LLM Vision 感知（核心）：

class VisionPerception:
    """用 LLM 视觉能力理解游戏画面"""

    def __init__(self, llm_client, game_profile):
        self.llm = llm_client          # GPT-4o / Claude 3.5 / Gemini
        self.game = game_profile       # 游戏特定的提示词模板
        self.frame_buffer = []         # 最近 N 帧缓存 (用于理解动态)

    async def perceive(self, screenshot: bytes) -> GameState:
        """截屏 → 结构化游戏状态"""
        prompt = f"""你是 {self.game.name} 的游戏分析专家。
分析这张游戏截图，返回 JSON：
{{
  "scene_type": "战斗/探索/菜单/过场/死亡",
  "player": {{
    "position": "屏幕位置描述",
    "health": "血量状态",
    "state": "站立/跑动/跳跃/攻击/受伤"
  }},
  "enemies": [
    {{"type": "类型", "position": "位置", "threat_level": "高/中/低"}}
  ],
  "environment": {{
    "obstacles": ["描述"],
    "items": ["描述"],
    "paths": ["可行进方向"]
  }},
  "ui_info": {{
    "score": null,
    "lives": null,
    "level": null,
    "special_meter": null
  }},
  "situation_summary": "一句话总结当前局面",
  "urgency": "高/中/低"
}}"""
        response = await self.llm.vision_query(
            image=screenshot,
            prompt=prompt,
            max_tokens=500
        )
        return GameState.from_dict(json.loads(response))

通道 B — RAM 直读（TAS 专用，精确补充）：

class RAMPerception:
    """直接读取模拟器 RAM 获取精确数值"""

    # 每个游戏需要一份内存地址映射表
    # 例: 超级马里奥兄弟 (NES)
    MEMORY_MAP = {
        "super_mario_bros": {
            "player_x":     0x0086,  # 玩家 X 坐标
            "player_y":     0x00CE,  # 玩家 Y 坐标
            "player_state": 0x000E,  # 状态 (0=小, 1=大, 2=火球)
            "lives":        0x075A,  # 生命数
            "coins":        0x075E,  # 金币数
            "world":        0x075F,  # 当前世界
            "level":        0x0760,  # 当前关卡
            "score":        0x07DD,  # 分数 (BCD)
            "time":         0x07F8,  # 剩余时间
            "enemy_drawn":  0x000F,  # 敌人是否在画面上
        }
    }

    def __init__(self, emulator_api, game_id):
        self.emu = emulator_api
        self.mem_map = self.MEMORY_MAP.get(game_id, {})

    def read_state(self) -> dict:
        """读取所有已映射的 RAM 值"""
        state = {}
        for name, addr in self.mem_map.items():
            state[name] = self.emu.read_memory(addr)
        return state

双通道融合：

class FusedPerception:
    """融合 LLM Vision + RAM 数据"""

    async def perceive(self, screenshot, emulator) -> GameState:
        # 两路并行
        vision_task = self.vision.perceive(screenshot)
        ram_state = self.ram.read_state()

        game_state = await vision_task

        # RAM 数据修正/补充 LLM 的理解
        # LLM 可能估错血量, RAM 给出精确值
        if "lives" in ram_state:
            game_state.player.lives = ram_state["lives"]
        if "player_x" in ram_state:
            game_state.player.exact_position = (
                ram_state["player_x"], ram_state["player_y"]
            )
        if "score" in ram_state:
            game_state.ui_info.score = ram_state["score"]

        return game_state

3.2 ② 决策层 (Brain) — "AI 的大脑"

LLM 推理决策，支持多级思考：

┌───────────────────────────────────────────────────────┐
│                     决策层                              │
│                                                        │
│  输入:                                                 │
│  ├── 当前 GameState (来自感知层)                       │
│  ├── 最近 N 步行动历史 + 结果                          │
│  ├── 游戏攻略知识 (来自 MBE 知识库, 按需)              │
│  └── 当前目标/子目标                                   │
│                                                        │
│  思考模式:                                             │
│  ┌──────────────────────────────────────────────┐      │
│  │ Level 1: 反射式 (简单场景, 低延迟)             │      │
│  │   "前面有坑 → 跳"                             │      │
│  │   "敌人接近 → 攻击"                            │      │
│  │   延迟: ~200ms, 适用: 80% 场景                │      │
│  ├──────────────────────────────────────────────┤      │
│  │ Level 2: 策略式 (复杂场景, 中延迟)             │      │
│  │   "Boss 战第二阶段 → 分析攻击模式 → 选择策略"  │      │
│  │   延迟: ~1-2s, 适用: 15% 场景                 │      │
│  ├──────────────────────────────────────────────┤      │
│  │ Level 3: 规划式 (卡关/迷宫, 高延迟+存档)      │      │
│  │   "迷宫地图推理 → 路径规划 → 分步执行"         │      │
│  │   延迟: ~5-10s, 适用: 5% 场景                 │      │
│  │   → 此时 TAS 暂停帧步进, 让 AI 慢慢想         │      │
│  └──────────────────────────────────────────────┘      │
│                                                        │
│  输出: ActionPlan                                      │
│  {actions: [...], confidence, reasoning}               │
└───────────────────────────────────────────────────────┘

class GameBrain:
    """LLM 驱动的游戏决策引擎"""

    def __init__(self, llm_client, game_profile, mbe_client=None):
        self.llm = llm_client
        self.game = game_profile
        self.mbe = mbe_client             # 可选, 调用 MBE 获取攻略知识
        self.action_history = deque(maxlen=20)
        self.current_goal = None
        self.sub_goals = []

    async def decide(self, state: GameState) -> ActionPlan:
        """根据当前状态决定下一步行动"""

        # 1. 判断思考级别
        think_level = self._assess_complexity(state)

        # 2. 如果需要攻略知识, 调用 MBE
        knowledge = ""
        if think_level >= 2 and self.mbe:
            knowledge = await self._query_mbe_strategy(state)

        # 3. 构建决策 Prompt
        prompt = self._build_decision_prompt(state, knowledge, think_level)

        # 4. LLM 推理
        response = await self.llm.query(
            prompt=prompt,
            temperature=0.3 if think_level == 1 else 0.7,
            max_tokens=300 if think_level == 1 else 800
        )

        # 5. 解析行动计划
        plan = ActionPlan.parse(response)
        self.action_history.append((state.summary, plan))
        return plan

    def _build_decision_prompt(self, state, knowledge, level):
        return f"""你是 {self.game.name} 的游戏高手 AI。

## 当前状态
{state.to_prompt()}

## 最近行动历史
{self._format_history()}

## 当前目标
{self.current_goal or "通关当前关卡"}

{f"## 攻略参考 (来自 MBE 知识库)" + chr(10) + knowledge if knowledge else ""}

## 可用操作
{self.game.action_space_description}

## 请决策
返回 JSON：
{{
  "reasoning": "简短思考过程 (1-2句话)",
  "actions": [
    {{"button": "按键名", "duration_frames": 帧数, "delay_frames": 延迟帧数}}
  ],
  "confidence": 0.0-1.0,
  "need_savestate": true/false  // 是否需要先存档 (不确定时存档)
}}"""

    async def _query_mbe_strategy(self, state: GameState) -> str:
        """调用 MBE 查询攻略知识"""
        query = f"{self.game.name} {state.situation_summary}"
        # 调用 MBE 专家系统 API
        response = await self.mbe.query_expert(
            expert_id=f"game_strategy_{self.game.id}",
            question=query
        )
        return response.answer if response else ""

3.3 ③ 执行层 (Hands) — "AI 的手"

class ActionExecutor:
    """操作执行器 — 支持 TAS 模拟器和真实键鼠两种模式"""

    def __init__(self, mode="tas", emulator=None):
        self.mode = mode
        if mode == "tas":
            self.backend = TASExecutor(emulator)
        elif mode == "realtime":
            self.backend = RealtimeExecutor()

    async def execute(self, plan: ActionPlan):
        """执行行动计划"""
        # 存档 (如果 AI 觉得不确定)
        if plan.need_savestate:
            await self.backend.save_state()

        for action in plan.actions:
            await self.backend.press(
                button=action.button,
                duration=action.duration_frames,
                delay=action.delay_frames
            )

        # 等待动作生效
        await self.backend.advance_frames(plan.total_frames)


class TASExecutor:
    """TAS 模拟器执行器"""

    def __init__(self, emulator):
        self.emu = emulator  # BizHawk / RetroArch API

    async def press(self, button, duration_frames, delay_frames):
        """精确到帧的按键输入"""
        if delay_frames > 0:
            await self.emu.advance_frames(delay_frames)
        # 按下
        self.emu.set_input(button, True)
        await self.emu.advance_frames(duration_frames)
        # 松开
        self.emu.set_input(button, False)

    async def save_state(self):
        self.emu.save_state(slot="auto")

    async def load_state(self, slot="auto"):
        self.emu.load_state(slot=slot)

    async def advance_frames(self, n):
        """步进 N 帧"""
        self.emu.frame_advance(n)

    async def get_screenshot(self) -> bytes:
        """获取当前帧截图"""
        return self.emu.screenshot()


class RealtimeExecutor:
    """真实键鼠执行器 (实操模式)"""

    def __init__(self):
        import pydirectinput
        self.input = pydirectinput

    async def press(self, button, duration_frames, delay_frames):
        """模拟键盘按键 (帧转换为时间, 假设 60fps)"""
        frame_time = 1.0 / 60
        if delay_frames > 0:
            await asyncio.sleep(delay_frames * frame_time)
        self.input.keyDown(button)
        await asyncio.sleep(duration_frames * frame_time)
        self.input.keyUp(button)

3.4 ④ 反馈层 (Memory) — "AI 的记忆"

class GameMemory:
    """游戏记忆 — 记录经验, 识别模式, 避免重复错误"""

    def __init__(self):
        self.episode_log = []         # 当前回合的完整记录
        self.death_patterns = []      # 死亡模式库
        self.success_patterns = []    # 成功模式库
        self.area_knowledge = {}      # 区域知识图谱
        self.boss_patterns = {}       # Boss 行为模式
        self.stuck_detector = StuckDetector()

    def record_step(self, state, action, result_state):
        """记录每一步"""
        step = StepRecord(
            state=state,
            action=action,
            result=result_state,
            reward=self._compute_reward(state, result_state),
            timestamp=time.time()
        )
        self.episode_log.append(step)

        # 检测卡关
        if self.stuck_detector.is_stuck(self.episode_log):
            return StepFeedback(stuck=True, suggestion="尝试不同策略")

        return StepFeedback(stuck=False)

    def on_death(self, state, action):
        """死亡回调 — 记录死亡模式"""
        pattern = DeathPattern(
            scene=state.scene_type,
            situation=state.situation_summary,
            last_actions=[s.action for s in self.episode_log[-5:]],
            cause=state.death_cause
        )
        self.death_patterns.append(pattern)

    def on_checkpoint(self, state):
        """通关/到达检查点 — 记录成功模式"""
        # 提取从上个检查点到这里的操作序列
        success_trace = self.episode_log[self.last_checkpoint_idx:]
        self.success_patterns.append(SuccessPattern(
            area=state.current_area,
            actions_trace=success_trace,
            total_steps=len(success_trace)
        ))

    def get_context_for_decision(self, state) -> str:
        """为决策层提供记忆上下文"""
        context_parts = []

        # 1. 这个区域之前死过吗?
        relevant_deaths = [d for d in self.death_patterns
                          if d.scene == state.scene_type]
        if relevant_deaths:
            context_parts.append(
                f"⚠️ 注意: 在类似场景死过 {len(relevant_deaths)} 次, "
                f"死因: {relevant_deaths[-1].cause}"
            )

        # 2. 这个区域有成功经验吗?
        relevant_success = self.area_knowledge.get(state.current_area)
        if relevant_success:
            context_parts.append(
                f"✅ 经验: 上次通过此区域用了 {relevant_success.strategy}"
            )

        # 3. Boss 模式识别
        if state.scene_type == "boss_fight":
            boss_id = state.boss_identifier
            if boss_id in self.boss_patterns:
                context_parts.append(
                    f"🎯 Boss 攻击模式: {self.boss_patterns[boss_id]}"
                )

        return "\n".join(context_parts)

    def _compute_reward(self, before, after) -> float:
        """计算奖励信号"""
        reward = 0.0
        # 进度推进 (+)
        if after.progress > before.progress:
            reward += (after.progress - before.progress) * 10
        # 分数增加 (+)
        if after.score > before.score:
            reward += (after.score - before.score) * 0.01
        # 生命减少 (-)
        if after.lives < before.lives:
            reward -= 5.0
        # 血量减少 (-)
        if after.player.health < before.player.health:
            reward -= 2.0
        # 死亡 (--)
        if after.is_dead:
            reward -= 20.0
        # 通关 (++)
        if after.level_completed:
            reward += 100.0
        return reward

3.5 ⑤ MBE 闭环层 — "需要时调用"

不是把 MBE 嵌进来，而是在关键节点调用 MBE 的服务：

┌──────────────────────────────────────────────────────────┐
│                  MBE 调用点                                │
│                                                           │
│  ① 策略知识查询                                           │
│  ├── 卡关时 → 查询 MBE 攻略专家知识库                      │
│  ├── Boss 战 → 查询 Boss 弱点/攻略                        │
│  └── 新游戏 → 加载 MBE 中该游戏的策略知识                  │
│                                                           │
│  ② 表现评估 (Eval)                                        │
│  ├── 每关结束 → 调用 MBE Eval 评估通关质量                 │
│  │   (用时、死亡次数、得分、操作精度)                       │
│  ├── 对比历史 → 调用 MBE 趋势分析看是否进步                │
│  └── 门禁检查 → 策略更新后必须通过 Eval 才能"上线"         │
│                                                           │
│  ③ 策略进化 (HOPE + Training)                             │
│  ├── 收集死亡模式 → 存入 MBE 回归用例库                    │
│  ├── 成功策略 → 存入 MBE 训练数据                          │
│  ├── HOPE 学习 → 哪些策略对哪些场景更有效                  │
│  └── Prompt 优化 → 基于 HOPE 反馈微调决策 Prompt          │
│                                                           │
│  ④ A/B 测试                                               │
│  ├── 新策略 Prompt vs 旧策略 Prompt                       │
│  ├── 不同 LLM (GPT-4o vs Claude) 对比                    │
│  └── 不同感知频率/精度对比                                 │
│                                                           │
│  ⑤ 监控告警                                               │
│  ├── 胜率突然下降 → 告警                                   │
│  ├── 同一位置反复死亡 > 10 次 → 告警 + 自动切换策略        │
│  └── LLM API 延迟飙升 → 降级到纯 RAM 感知模式             │
└──────────────────────────────────────────────────────────┘

class MBEIntegration:
    """MBE 按需调用集成"""

    def __init__(self, mbe_base_url="http://localhost:8000"):
        self.base_url = mbe_base_url

    # ---- 策略知识查询 ----

    async def query_strategy(self, game_id, situation) -> str:
        """卡关时查询 MBE 攻略知识库"""
        resp = await httpx.post(f"{self.base_url}/api/consult", json={
            "question": f"游戏 {game_id} 中遇到 {situation}，应该怎么做？",
            "expert_id": f"game_{game_id}",
        })
        return resp.json().get("answer", "")

    # ---- 表现评估 ----

    async def evaluate_run(self, game_id, run_stats: dict):
        """评估一次通关/闯关的表现"""
        resp = await httpx.post(
            f"{self.base_url}/api/evaluation/expert",
            json={
                "expert_id": f"game_agent_{game_id}",
                "eval_type": "game_run",
                "report_json": run_stats,
                "overall_score": run_stats.get("score", 0),
            }
        )
        return resp.json()

    # ---- 回归用例 (死亡模式) ----

    async def report_failure(self, game_id, death_pattern: dict):
        """上报死亡模式到 MBE 回归用例库"""
        await httpx.post(
            f"{self.base_url}/api/evaluation/regression/"
            f"game_agent_{game_id}/report",
            json=death_pattern
        )

    # ---- A/B 测试 ----

    async def create_strategy_test(self, game_id, v1_prompt, v2_prompt):
        """创建策略 A/B 测试"""
        await httpx.post(
            f"{self.base_url}/api/evaluation/ab-test/create",
            params={
                "expert_id": f"game_agent_{game_id}",
                "control_version": "v1",
                "treatment_version": "v2",
            }
        )

3.6 ⑥ TAS 仿真环境层

支持的模拟器和游戏平台：

BizHawk 接口封装：

class BizHawkBridge:
    """BizHawk 模拟器桥接 (通过 Socket 通信)"""

    def __init__(self, host="localhost", port=9999):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((host, port))

    def read_memory(self, addr: int, size: int = 1) -> int:
        """读取 RAM 内存"""
        self._send(f"READ {addr} {size}")
        return int(self._recv())

    def write_memory(self, addr: int, value: int):
        """写入 RAM (调试用)"""
        self._send(f"WRITE {addr} {value}")

    def set_input(self, button: str, pressed: bool):
        """设置按键状态"""
        self._send(f"INPUT {button} {'1' if pressed else '0'}")

    def frame_advance(self, n: int = 1):
        """步进 N 帧"""
        self._send(f"ADVANCE {n}")

    def save_state(self, slot: str = "auto"):
        """存档"""
        self._send(f"SAVESTATE {slot}")

    def load_state(self, slot: str = "auto"):
        """读档"""
        self._send(f"LOADSTATE {slot}")

    def screenshot(self) -> bytes:
        """截屏 (返回 PNG bytes)"""
        self._send("SCREENSHOT")
        size = int(self._recv())
        return self._recv_bytes(size)

    def get_frame_count(self) -> int:
        """当前帧数"""
        self._send("FRAMECOUNT")
        return int(self._recv())

    def set_speed(self, multiplier: float):
        """设置模拟速度 (1.0=正常, 0=最快)"""
        self._send(f"SPEED {multiplier}")

4. 主控循环

class GameAIAgent:
    """游戏 AI 主控循环"""

    def __init__(self, config: GameConfig):
        # 核心组件
        self.perception = FusedPerception(config)
        self.brain = GameBrain(config)
        self.executor = ActionExecutor(config)
        self.memory = GameMemory()

        # MBE 集成 (可选)
        self.mbe = MBEIntegration(config.mbe_url) if config.use_mbe else None

        # 运行状态
        self.running = False
        self.total_deaths = 0
        self.current_level = None
        self.run_stats = RunStats()

    async def run(self, max_episodes=100):
        """主循环"""
        self.running = True

        for episode in range(max_episodes):
            logger.info(f"=== Episode {episode + 1} ===")
            await self._run_episode()

            # 每个 episode 结束后, 评估表现
            if self.mbe:
                await self.mbe.evaluate_run(
                    self.game.id, self.run_stats.to_dict()
                )

            # 重置
            self.run_stats.reset()

    async def _run_episode(self):
        """单个回合"""
        step = 0
        while self.running:
            # ① 感知
            screenshot = await self.executor.backend.get_screenshot()
            state = await self.perception.perceive(screenshot)

            # 检测是否通关
            if state.level_completed:
                logger.info(f"🎉 通关! Level {state.current_level}")
                self.memory.on_checkpoint(state)
                break

            # 检测是否死亡
            if state.is_dead:
                self.total_deaths += 1
                logger.info(f"💀 死亡 #{self.total_deaths}: {state.death_cause}")
                self.memory.on_death(state, last_action)

                # 上报 MBE 回归库
                if self.mbe:
                    await self.mbe.report_failure(self.game.id, {
                        "scene": state.scene_type,
                        "cause": state.death_cause,
                        "step": step
                    })

                # 读档
                await self.executor.backend.load_state()
                continue

            # ② 获取记忆上下文
            memory_context = self.memory.get_context_for_decision(state)
            self.brain.inject_context(memory_context)

            # ③ 决策
            plan = await self.brain.decide(state)
            last_action = plan

            # 低置信度时查询 MBE 攻略
            if plan.confidence < 0.3 and self.mbe:
                strategy = await self.mbe.query_strategy(
                    self.game.id, state.situation_summary
                )
                if strategy:
                    plan = await self.brain.decide_with_hint(state, strategy)

            # ④ 执行
            await self.executor.execute(plan)

            # ⑤ 记录
            new_screenshot = await self.executor.backend.get_screenshot()
            new_state = await self.perception.perceive(new_screenshot)
            feedback = self.memory.record_step(state, plan, new_state)

            # 卡关检测
            if feedback.stuck:
                logger.warning("🔄 检测到卡关, 尝试新策略")
                self.brain.switch_strategy()
                if self.mbe:
                    # 查 MBE 有没有别人的攻略
                    hint = await self.mbe.query_strategy(
                        self.game.id, f"卡在 {state.situation_summary}"
                    )
                    self.brain.inject_hint(hint)

            step += 1
            self.run_stats.update(state, plan, new_state)

            # 控制感知频率 (不需要每帧都调 LLM)
            # 简单场景: 每 10 帧感知一次
            # 复杂场景: 每 3 帧
            # Boss 战: 每帧
            skip = self._get_perception_interval(state)
            if skip > 0:
                await self.executor.backend.advance_frames(skip)

    def _get_perception_interval(self, state: GameState) -> int:
        """根据场景复杂度调整感知频率"""
        if state.urgency == "高":
            return 0      # 每帧
        elif state.urgency == "中":
            return 3      # 每 3 帧
        else:
            return 10     # 每 10 帧

5. 游戏配置文件 (Game Profile)

每个游戏一个配置文件，零代码切换游戏：

# profiles/super_mario_bros.yaml

game:
  id: super_mario_bros
  name: "超级马里奥兄弟"
  platform: nes
  rom: "roms/Super Mario Bros (JU).nes"

emulator:
  type: bizhawk
  core: NesHawk

# 按键映射
action_space:
  - { name: "right",  button: "Right",  description: "向右走" }
  - { name: "left",   button: "Left",   description: "向左走" }
  - { name: "jump",   button: "A",      description: "跳跃" }
  - { name: "run",    button: "B",      description: "加速跑" }
  - { name: "down",   button: "Down",   description: "蹲下/进水管" }
  - { name: "up",     button: "Up",     description: "向上看" }
  - { name: "run_jump", button: "A+B+Right", description: "加速跳" }

# RAM 内存地址映射
memory_map:
  player_x:     0x0086
  player_y:     0x00CE
  player_state: 0x000E
  lives:        0x075A
  coins:        0x075E
  world:        0x075F
  level:        0x0760
  score:        0x07DD
  time:         0x07F8

# 游戏特定知识 (注入 Prompt)
game_knowledge: |
  - 马里奥可以通过跳上敌人头顶消灭它们 (乌龟、板栗仔)
  - 问号砖块可以撞出金币或蘑菇
  - 大蘑菇让马里奥变大, 火花让马里奥能发射火球
  - 星星让马里奥无敌一段时间
  - 关底旗杆: 跳得越高分数越高
  - 1-2、4-2 有隐藏水管可以跳关

# 奖励设计
rewards:
  progress_forward: 1.0      # 每向右推进 1 像素
  coin_collected: 5.0        # 每个金币
  enemy_killed: 10.0         # 每消灭一个敌人
  powerup_collected: 20.0    # 获得道具
  level_complete: 100.0      # 通关
  death: -50.0               # 死亡
  time_penalty: -0.1         # 每帧时间惩罚 (鼓励速通)

# 感知配置
perception:
  vision_interval_easy: 10    # 简单场景每 10 帧看一次
  vision_interval_hard: 2     # 困难场景每 2 帧看一次
  ram_interval: 1             # RAM 每帧都读

# MBE 集成
mbe:
  enabled: true
  expert_id: game_strategy_mario
  knowledge_base: "mario_walkthrough"
  eval_metrics:
    - total_score
    - deaths
    - completion_time
    - levels_cleared

6. 训练流程

╔═══════════════════════════════════════════════════════════════╗
║                   训练流程 (渐进式)                            ║
║                                                               ║
║  Stage 1: 零样本探索 (Zero-shot)                              ║
║  ┌──────────────────────────────────────────┐                 ║
║  │  · 只用 LLM Vision + 基础游戏知识         │                 ║
║  │  · TAS 帧步进模式, 不限思考时间            │                 ║
║  │  · 收集: 死亡模式、卡关位置、成功片段       │                 ║
║  │  · 产出: 初始经验库                        │                 ║
║  └──────────────────────────────────────────┘                 ║
║                          ↓                                    ║
║  Stage 2: 经验回放 (Experience Replay)                        ║
║  ┌──────────────────────────────────────────┐                 ║
║  │  · 分析 Stage 1 的死亡/卡关模式            │                 ║
║  │  · 注入攻略知识到 MBE 知识库               │                 ║
║  │  · 优化决策 Prompt (加入经验规则)          │                 ║
║  │  · 读档重跑困难段落, 尝试不同策略           │                 ║
║  │  · 产出: 优化后的决策 Prompt               │                 ║
║  └──────────────────────────────────────────┘                 ║
║                          ↓                                    ║
║  Stage 3: 策略 A/B 测试                                      ║
║  ┌──────────────────────────────────────────┐                 ║
║  │  · 多个策略变体同时跑同一关                 │                 ║
║  │  · MBE A/B 测试框架对比表现                │                 ║
║  │  · 自动选择最优策略                        │                 ║
║  │  · 产出: 最优策略 Prompt + 配置            │                 ║
║  └──────────────────────────────────────────┘                 ║
║                          ↓                                    ║
║  Stage 4: 速度优化 (Speed Run)                                ║
║  ┌──────────────────────────────────────────┐                 ║
║  │  · 能通关后, 优化通关速度                   │                 ║
║  │  · 减少感知频率, 提高操作精度               │                 ║
║  │  · 记录最优操作序列 (TAS input log)        │                 ║
║  │  · 产出: 可回放的 TAS 通关录像             │                 ║
║  └──────────────────────────────────────────┘                 ║
║                          ↓                                    ║
║  Stage 5: 实时挑战 (Real-time)                                ║
║  ┌──────────────────────────────────────────┐                 ║
║  │  · 关闭帧步进, 正常速度运行                 │                 ║
║  │  · 用积累的经验减少 LLM 调用频率           │                 ║
║  │  · 简单场景用缓存策略, 复杂场景调 LLM      │                 ║
║  │  · 产出: 真正的"实时打游戏" AI             │                 ║
║  └──────────────────────────────────────────┘                 ║
╚═══════════════════════════════════════════════════════════════╝

7. 目录结构

game-ai/
├── README.md
├── requirements.txt
├── config.yaml                    # 全局配置
│
├── profiles/                      # 游戏配置
│   ├── super_mario_bros.yaml
│   ├── zelda.yaml
│   ├── megaman.yaml
│   └── ...
│
├── src/
│   ├── __init__.py
│   ├── agent.py                   # GameAIAgent 主控循环
│   │
│   ├── perception/
│   │   ├── __init__.py
│   │   ├── vision.py              # LLM Vision 感知
│   │   ├── ram.py                 # RAM 直读
│   │   └── fusion.py             # 双通道融合
│   │
│   ├── brain/
│   │   ├── __init__.py
│   │   ├── decision.py            # LLM 决策引擎
│   │   ├── goals.py               # 目标管理
│   │   └── prompts/               # 决策 Prompt 模板
│   │       ├── base.jinja2
│   │       ├── combat.jinja2
│   │       └── exploration.jinja2
│   │
│   ├── execution/
│   │   ├── __init__.py
│   │   ├── executor.py            # 操作执行器
│   │   ├── tas.py                 # TAS 模拟器后端
│   │   └── realtime.py            # 真实键鼠后端
│   │
│   ├── memory/
│   │   ├── __init__.py
│   │   ├── game_memory.py         # 游戏记忆
│   │   ├── death_patterns.py      # 死亡模式库
│   │   └── stuck_detector.py      # 卡关检测
│   │
│   ├── emulators/
│   │   ├── __init__.py
│   │   ├── bizhawk.py             # BizHawk 桥接
│   │   ├── retroarch.py           # RetroArch 桥接
│   │   └── base.py                # 模拟器基类
│   │
│   ├── mbe/
│   │   ├── __init__.py
│   │   └── integration.py         # MBE 调用封装
│   │
│   └── utils/
│       ├── __init__.py
│       ├── screenshot.py           # 截屏工具
│       └── models.py               # 数据模型
│
├── scripts/
│   ├── play.py                    # 启动游戏 AI
│   ├── train.py                   # 训练模式
│   ├── replay.py                  # 回放 TAS 录像
│   └── setup_bizhawk.py           # 模拟器配置向导
│
├── roms/                          # (gitignore) ROM 文件
├── savestates/                    # 存档
├── recordings/                    # 操作录像
└── logs/                          # 运行日志

8. 技术选型

成本估算 (以超级马里奥为例)

9. 快速启动计划

10. 首选验证游戏

建议从"超级马里奥兄弟 1-1"开始 — 最经典、资源最多、复杂度适中。

附录: BizHawk Lua 端脚本

-- bizhawk_bridge.lua
-- 在 BizHawk 中运行, 与 Python 通过 Socket 通信

local socket = require("socket")
local server = socket.tcp()
server:bind("localhost", 9999)
server:listen(1)
server:settimeout(0)  -- 非阻塞

local client = nil

while true do
    -- 接受连接
    if not client then
        client = server:accept()
        if client then
            client:settimeout(0)
            console.log("Python connected!")
        end
    end

    -- 处理命令
    if client then
        local line, err = client:receive()
        if line then
            local parts = {}
            for word in line:gmatch("%S+") do
                table.insert(parts, word)
            end
            local cmd = parts[1]

            if cmd == "READ" then
                local addr = tonumber(parts[2])
                local val = memory.read_u8(addr)
                client:send(tostring(val) .. "\n")

            elseif cmd == "INPUT" then
                local button = parts[2]
                local pressed = parts[3] == "1"
                joypad.set({[button] = pressed}, 1)
                client:send("OK\n")

            elseif cmd == "ADVANCE" then
                local frames = tonumber(parts[2]) or 1
                for i = 1, frames do
                    emu.frameadvance()
                end
                client:send("OK\n")

            elseif cmd == "SAVESTATE" then
                local slot = tonumber(parts[2]) or 0
                savestate.saveslot(slot)
                client:send("OK\n")

            elseif cmd == "LOADSTATE" then
                local slot = tonumber(parts[2]) or 0
                savestate.loadslot(slot)
                client:send("OK\n")

            elseif cmd == "SCREENSHOT" then
                local path = "screenshot_temp.png"
                client.screenshot(path)
                local f = io.open(path, "rb")
                local data = f:read("*a")
                f:close()
                client:send(tostring(#data) .. "\n")
                client:send(data)

            elseif cmd == "FRAMECOUNT" then
                client:send(tostring(emu.framecount()) .. "\n")

            elseif cmd == "SPEED" then
                local speed = tonumber(parts[2]) or 100
                client.setspeed(speed)
                client:send("OK\n")
            end

        elseif err == "closed" then
            client = nil
            console.log("Python disconnected")
        end
    end

    emu.frameadvance()
end