Skip to content

Repository files navigation

AgentBenchFrame

统一游戏 AI 实验框架,支持规则迭代、RL 训练、对抗竞技场、透明追踪与可视化。

安装

git clone git@github.com:SAST-agent/AgentBenchFramework.git
cd AgentBenchFramework
uv sync                          # 零硬依赖,uv 管理环境

可选依赖按需安装:

uv sync --extra rl               # + torch, numpy (RL 训练)
uv sync --extra report           # + jinja2 (报告生成)
uv sync --extra all              # 全部

快速开始

5 行跑一场对战

from agentbench_frame.env import GeneralsEnv, EnvMode, register_env, make_env
from agentbench_frame.agent import RuleBasedAgent, RandomAgent
from agentbench_frame.arena import Match

register_env("generals", GeneralsEnv)
env = make_env("generals", mode=EnvMode.DIRECT)
a1 = RuleBasedAgent("expansionist", rules=[...])
a2 = RandomAgent("random")

match = Match(env, a1, a2)
result = match.run(n_games=100)
print(f"{result.agent1_name} win rate: {result.win_rate:.0%}")

跑一场锦标赛

from agentbench_frame.arena import Arena

agents = [agent_a, agent_b, agent_c, agent_d]
arena = Arena(env, agents)
result = arena.round_robin(n_games=20)
for rank, name, elo in result.rankings:
    print(f"#{rank} {name} Elo={elo:.0f}")

带透明追踪的训练

from agentbench_frame.tracking import Run

run = Run.start(game="28_generals", agent="ppo_v3", run_type="rl")
env = run.wrap_env(GeneralsEnv())
agent = run.wrap_agent(my_agent)
run.start_sampler()          # 后台采集 CPU/内存

for ep in range(1000):
    obs = env.reset()
    done = False
    while not done:
        action = agent.act(obs.to_dict())
        obs, reward, done, _ = env.step(action)
    run.log_episode(reward, obs.round_num, obs.state["winner"])
    if ep % 50 == 0:
        run.log_elo(evaluate(my_agent))   # 记录 Elo 变化

run.finish()                  # 自动写入 run.toml + summary.json
# → agentbench_data/runs/28_generals/ppo_v3/{run_id}/

推送数据到可视化

export AGENTBENCH_DATA=/path/to/AgentBenchResults
# 框架自动写入上述目录
cd $AGENTBENCH_DATA
git add runs/ && git commit -m "ppo_v3: Elo 1520" && git push
# → CI 自动聚合 → https://sast-agent.github.io/AgentBenchResults/

架构

┌──────────────────────────────────────────────┐
│                  runner/                      │  ← 策略基类(可扩展)
│  BaseRunner → BaseRLRunner / BaseRuleRunner  │
├──────────────────────────────────────────────┤
│                  training/                    │  ← 训练算法
│  PPOTrainer, RLTrainer, RuleIterator         │
├──────────────────────────────────────────────┤
│  agent/       │  skills/    │  mcp/           │  ← 决策层
│  RuleBasedAgent│ ReplayReader│ MCPTool        │
│  RLAgent      │ MapAnalyzer │ MCPServer      │
├───────────────┼─────────────┼────────────────┤
│              env/                             │  ← 环境层
│  BaseEnv, GeneralsEnv, StdioProtocol         │
├──────────────────────────────────────────────┤
│  arena/       │  eval/      │  tracking/      │  ← 评估与追踪
│  Match,Arena  │ Trajectory  │  Run,Sampler    │
└──────────────────────────────────────────────┘

核心基类与扩展

框架设计原则:所有关键行为都通过基类约束,子类只需重写 1-2 个方法。

1. 扩展游戏环境 — BaseEnv

from agentbench_frame.env import BaseEnv, Observation, ActionSpace, EnvMode

class MyGameEnv(BaseEnv):
    game_name = "MyGame"
    num_players = 2

    @property
    def action_space(self) -> ActionSpace:
        return ActionSpace(type="discrete", n=4)

    @property
    def observation_space(self) -> dict:
        return {"type": "dict", "keys": ["board", "score"]}

    def _reset_direct(self, seed=None) -> Observation:
        self._state = self._init_game(seed)
        return self._build_obs()

    def _step_direct(self, action) -> tuple[Observation, float, bool, dict]:
        self._apply(action)
        reward = self._calc_reward()
        done = self._is_terminal()
        return self._build_obs(), reward, done, {}

注册后即可使用:

from agentbench_frame.env import register_env, make_env
register_env("mygame", MyGameEnv)
env = make_env("mygame", mode=EnvMode.DIRECT)

2. 扩展 Agent — BaseAgent

规则 Agent(组合已有规则):

from agentbench_frame.agent import RuleBasedAgent

def my_custom_rule(obs, state):
    if obs["state"]["round"] < 5:
        return [[1, 0, 0, 4, 2]]   # 开局 rush
    return None                      # 交给下一条规则

agent = RuleBasedAgent(
    name="rush_agent",
    skills=[MapAnalyzerSkill()],     # 可选:附加技能
    rules=[my_custom_rule, expand_rule, end_turn_rule],
)

RL Agent(自定义策略网络):

from agentbench_frame.agent import RLAgent, PolicyNetwork

class MyPolicy(PolicyNetwork):
    def predict(self, obs):
        # 你的推理逻辑
        return action_id, action_probs

agent = RLAgent(name="my_rl", policy=MyPolicy())

3. 扩展训练策略 — BaseRunner

from agentbench_frame.runner import BaseRunner, BaseRLRunner

class MyTrainer(BaseRLRunner):
    """自定义 RL 训练器"""
    def _execute(self, env, agent):
        # env 和 agent 已被框架自动包装(追踪 + 计时 + 采样)
        for episode in range(self.config.total_episodes):
            obs = env.reset()
            done = False
            while not done:
                action = agent.act(obs.to_dict())
                obs, reward, done, _ = env.step(action)
            self._run.log_episode(reward, obs.round_num, obs.state["winner"])
            if episode % 100 == 0:
                self._run.log_elo(self._evaluate(agent))
        # run.finish() 由 BaseRunner 自动调用

# 使用
trainer = MyTrainer(env, agent, config={"total_episodes": 1000})
run = trainer.run()   # 返回 Run 对象,可直接查询
print(run.run_dir)    # agentbench_data/runs/mygame/myagent/{run_id}/

4. 扩展迭代策略 — BaseRuleRunner

from agentbench_frame.runner import BaseRuleRunner

class EvolutionIterator(BaseRuleRunner):
    def _execute(self, env, agent):
        best = agent
        for gen in range(self.config["generations"]):
            variants = self._mutate(best)
            winner = self._evaluate_population(env, variants)
            self._run.write("generation", gen=gen, winner=winner.name,
                            win_rate=self._last_win_rate)
            if winner.win_rate > self._baseline + 0.05:
                best = winner
                self._mark_accepted(gen)     # 写 accepted 标记文件

5. 扩展技能 — Skill

from agentbench_frame.skills import Skill, SkillMeta

class MyAnalyzer(Skill):
    def __init__(self):
        super().__init__(SkillMeta(
            name="my_analyzer", version="1.0",
            description="Custom game analysis",
            game="28_generals",
        ))

    def can_activate(self, obs, ctx):
        return obs.get("round_num", 0) % 5 == 0

    def execute(self, obs, ctx):
        ctx["analysis"] = self._analyze(obs)
        return None    # 纯信息技能,不产生动作

6. 扩展 MCP 工具 — MCPTool

from agentbench_frame.mcp import MCPTool

class QueryDatabaseTool(MCPTool):
    def __init__(self):
        super().__init__(name="query_db", description="Query game database")

    def get_input_schema(self):
        return {"type": "object", "properties": {"query": {"type": "string"}}}

    def call(self, **kwargs):
        return {"results": db.query(kwargs["query"])}

数据契约

框架自动产出 CI 兼容的数据。详见 CONVENTIONS.md

# 验证本地数据格式
agentbench data check

# 列出所有 run
agentbench data list

数据目录结构:

$AGENTBENCH_DATA/
└── runs/{game}/{agent}/{run_id}/
    ├── run.toml         # 元信息(type, created, git_commit)
    └── summary.json     # 聚合指标(best_elo, elo_history, h2h, wall_hours)

CLI

agentbench train   --game 28_generals --agent ppo_v3    # RL 训练
agentbench eval    --game 28_generals --agent ppo_v3    # 评估
agentbench iterate --game 28_generals --agent rules_v1  # 规则迭代
agentbench arena   --game 28_generals --agents a,b,c    # 锦标赛
agentbench report  --data-dir ./data --output ./_site   # 生成静态站点
agentbench mcp                                           # 启动 MCP 服务器
agentbench data check --data-dir ./data                  # 验证数据格式
agentbench data list --data-dir ./data                   # 列出所有 run

环境变量

变量 说明 默认值
AGENTBENCH_DATA 数据根目录 ./agentbench_data

依赖策略

  • 核心零依赖env, agent, arena, skills, mcp 可直接使用
  • torch, numpy — RL 训练时需要 ([rl] extra)
  • jinja2 — 报告生成时需要 ([report] extra)
  • psutil — 资源追踪时需要 ([tracking] extra)
  • 无对应库时自动降级,不报错

About

No description, website, or topics provided.

Resources

Stars

1 star

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages