跳转到内容

扩展环境:Android/Minecraft/Werewolf

学习目标

理解 MetaGPT 如何通过 ExtEnv 将框架扩展到非软件工程场景,以及各游戏环境的具体实现。


项目实践

ExtEnv 架构

ExtEnv(External Environment)是 MetaGPT 的游戏环境基类,继承 BaseEnvironment + Pydantic BaseModel

class ExtEnv(BaseEnvironment, BaseModel):
action_space: spaces.Space[ActType] # 动作空间(Gymnasium)
observation_space: spaces.Space[ObsType] # 观测空间(Gymnasium)
# Gymnasium 标准接口
def reset(self, seed=None, options=None) -> tuple[dict, dict]: ...
def observe(self, obs_params=None) -> Any: ...
def step(self, action: BaseEnvAction) -> tuple[dict, float, bool, bool, dict]: ...
# MetaGPT 消息接口
def publish_message(self, message, peekable=True) -> bool: ...
async def run(self, k=1): ...

API 注册机制

ExtEnv 使用装饰器注册可读写 API:

env_read_api_registry = ReadAPIRegistry()
env_write_api_registry = WriteAPIRegistry()
def mark_as_readable(func):
"""标记为环境中的可读 API"""
env_read_api_registry[func.__name__] = get_function_schema(func)
return func
def mark_as_writeable(func):
"""标记为环境中的可写 API"""
env_write_api_registry[func.__name__] = get_function_schema(func)
return func

角色通过统一接口调用环境能力:

# 从环境读取
async def read_from_api(self, env_action):
if isinstance(env_action, str):
env_read_api = env_read_api_registry.get(api_name=env_action)["func"]
return env_read_api(self)
elif isinstance(env_action, EnvAPIAbstract):
return env_read_api(self, *env_action.args, **env_action.kwargs)
# 写入环境
async def write_thru_api(self, env_action):
if isinstance(env_action, Message):
self.publish_message(env_action)
elif isinstance(env_action, EnvAPIAbstract):
return env_write_api(self, *env_action.args, **env_action.kwargs)

AndroidEnv

Android 操控环境,通过 ADB 控制手机:

class AndroidEnv(ExtEnv):
# 可读 API
@mark_as_readable
def get_screenshot(self): ... # 获取屏幕截图
@mark_as_readable
def get_ui_tree(self): ... # 获取 UI 层级结构
@mark_as_readable
def get_clipboard(self): ... # 获取剪贴板内容
# 可写 API
@mark_as_writeable
def tap(self, x, y): ... # 点击坐标
@mark_as_writeable
def swipe(self, x1, y1, x2, y2): ... # 滑动
@mark_as_writeable
def input_text(self, text): ... # 输入文本
@mark_as_writeable
def press_key(self, key): ... # 按键(back/home/enter)
@mark_as_writeable
def adb_shell(self, command): ... # 执行 ADB shell 命令

依赖:

  • ADB(Android Debug Bridge)连接设备
  • uiautomator2 获取 UI 树
  • GroundingDINO 视觉定位(可选)

使用场景:自动化测试、APP 操作演示、无障碍辅助。

MinecraftEnv

Minecraft 游戏环境,通过 Mineflayer JS 库控制游戏内 Bot:

class MinecraftEnv(ExtEnv):
@mark_as_writeable
def chat(self, message): ... # 发送聊天消息
@mark_as_writeable
def move(self, direction): ... # 移动
@mark_as_writeable
def place_block(self, block_type, position): ... # 放置方块
@mark_as_writeable
def dig_block(self, position): ... # 挖掘方块
@mark_as_writeable
def attack(self): ... # 攻击
@mark_as_readable
def get_inventory(self): ... # 获取物品栏
@mark_as_readable
def get_health(self): ... # 获取生命值
@mark_as_readable
def get_position(self): ... # 获取位置

依赖:

  • Node.js + Mineflayer(JS Minecraft Bot 库)
  • 运行中的 Minecraft 服务器(本地或远程)

使用场景:建筑指令、自动化任务、RL 训练环境。

WerewolfEnv

狼人杀游戏环境:

class WerewolfEnv(ExtEnv):
# 角色类型
roles = ["werewolf", "villager", "seer", "witch", "guard", "moderator"]
@mark_as_writeable
def vote(self, voter, target): ... # 投票
@mark_as_writeable
def kill(self, werewolf, target): ... # 狼人杀人
@mark_as_writeable
def use_skill(self, player): ... # 使用技能(预言家/女巫)
@mark_as_readable
def get_game_state(self): ... # 获取当前游戏状态
@mark_as_readable
def get_alive_players(self): ... # 获取存活玩家
@mark_as_readable
def get_night_actions(self): ... # 获取夜间行动

对应的角色定义在 metagpt/ext/werewolf/roles/

角色文件能力
Werewolfwerewolf.py夜间杀人
Seerseer.py查验身份
Witchwitch.py救人/毒人
Guardguard.py守护玩家
Villagervillager.py投票
Moderatormoderator.py主持流程

StanfordTownEnv

斯坦福小镇模拟环境,基于 Generative Agents 论文:

class StanfordTownEnv(ExtEnv):
@mark_as_writeable
def move(self, agent, location): ... # 移动 Agent
@mark_as_writeable
def interact(self, agent_a, agent_b): ... # 两个 Agent 交互
@mark_as_readable
def get_agent_state(self, agent): ... # 获取 Agent 状态
@mark_as_readable
def get_world_state(self): ... # 获取世界状态

使用场景:社交行为模拟、多 Agent 交互研究。


问题与规避

Android 环境依赖

  • 需要连接 Android 设备或模拟器
  • 对策:使用 adb devices 检查连接状态,环境初始化时验证

Minecraft 需要 Node.js

  • Mineflayer 是 JS 库,需要 Node.js 运行时
  • 对策:在环境初始化时检查 nodenpm 是否可用

环境状态同步

  • ExtEnv 的状态变化可能快于 Role 的观察频率
  • 对策:observe() 接口支持 obs_params 过滤,只获取关心的状态变化

设计取舍

Gymnasium 接口的通用性

  • reset() / observe() / step() 是强化学习标准接口
  • 使得 MetaGPT Agent 可以直接使用 RL 框架训练
  • 代价:对于非 RL 场景(如纯 Agent 操控),这些接口可能过于复杂

API 注册 vs 直接方法调用

  • API 注册允许通过字符串名称调用环境能力,适合 LLM 动态选择
  • 代价:失去了 IDE 的类型检查和自动补全
  • 折中:mark_as_readable/writeable 装饰器保留 Python 方法签名的同时注册到字典

ExtEnv vs 标准 Environment

  • ExtEnv 继承 BaseEnvironment,增加了 Gymnasium 接口和 API 注册
  • 标准 Environment 继承 ExtEnv,增加了角色管理和消息路由
  • 分层设计允许只使用 ExtEnv(无角色)进行 RL 训练,或两者结合使用

参考来源

  • 源码验证: metagpt/environment/base_env.py:53 — ExtEnv 类
  • 源码验证: metagpt/environment/android/android_env.py — AndroidEnv
  • 源码验证: metagpt/environment/minecraft/minecraft_env.py — MinecraftEnv
  • 源码验证: metagpt/environment/werewolf/werewolf_env.py — WerewolfEnv
  • 源码验证: metagpt/environment/stanford_town/stanford_town_env.py — StanfordTownEnv
  • 源码验证: metagpt/ext/werewolf/roles/ — 狼人杀角色