はじめに

AIエージェントが盛り上がっています。でも、AIエージェントって結局なんなんでしょう

AIエージェントとは、目標に向かって状況を観測し、次の行動を決め、必要なツールを使いながらタスクを進めるソフトウェアである

と説明されても、「なるほど、そういうことか」とはならないのではないでしょうか。
エンジニアとしては、内部の仕組みをある程度知っておいた方がいいでしょう。

そこで今回は、「新元号を当てるまで脱出できない生活」を模したゲームを題材にAIエージェントを動かします。
まずゲームをWebサーバーとして実装し、Codexのサブエージェントで攻略できることを確認します。その後、最小構成のAIエージェントを実装し、少しずつ機能を追加していきます。
目的は性能の高いAIエージェントを作ることではなく、AIエージェントがどのように動くのかを理解することです。

作成したゲームとエージェントはgithubに載せています。エージェントを動かすにはOpenAI APIのキーが必要です。
https://github.com/persimmon-persimmon/gengo-game-scenario/tree/main

新元号を当てるまで脱出できないゲーム

水曜日のダウンタウンで放送された有名な企画「新元号を当てるまで脱出できない生活」をプレイできるゲームを作りました。原作そのままのルールではなくAIエージェント向けに調整しています。

プレイヤーは外部から隔離された部屋で生活し、ゲーム内に隠された新元号を当てるまで脱出できません。限られた資源を使って情報を集め、観測結果をもとに次の行動を決める必要があります。

ゲームのルール

ゲーム開始時、プレイヤーは2,000円と満腹度80を持っています。
元号の回答には、漢字表記と読みの両方が必要です。不正解の場合は満腹度を1消費します。満腹度は食べ物を購入して回復できますが、0になるとゲームオーバーです。

日を終えると満腹度を10消費します。その代わり、生活費としてお金を受け取れます。得られる生活費は1日目の2,000円から、2日目は1,800円と、1日ごとに200円ずつ減っていきます。11日目には0円になります。

毎日、元号とは別のDaily Challengeにも挑戦できます。その日に指定された音読みを持ち、必要画数以上の漢字を一文字ずつ答える課題です。正解すると賞金を得られますが、日数が進むほど必要な画数が増え、難しくなります。1日に回答できる回数も3回までです。

つまり、元号の回答を繰り返して情報を集めるだけでは、いずれ満腹度が尽きます。食料を買えば長く探索できますが、ヒントを買うお金が足りなくなるかもしれません。何にお金を使うかも、プレイヤーの判断になります。

ヒントとショップ

ショップでは、食料とヒントを購入できます。
ヒントには、曖昧な文章によるものや、元号の回答に反応して追加の情報を返すものがあります。実際に回答して反応を観測することで、ヒントの仕組みを推測できるようになっています。
また、価格の割に満腹度の回復量が少ない食べ物（せんべい）を用意しました。一見すると効率の悪い商品ですが、あえて購入して使うことで、新元号のヒントを得られます。エージェントがメタ読みできるかを見ます。

商品	種類	価格	効果	在庫
おにぎり	食料	300円	満腹度 +15	無制限
カップ麺	食料	500円	満腹度 +30	無制限
弁当	食料	800円	満腹度 +50	無制限
せんべい	食料	500円	満腹度 +10	3個
ヒント梅	ヒント	3,000円	街頭インタビュー形式の曖昧なコメント	1回
ヒント竹	ヒント	5,000円	回答に反応するランプ5つ。回答した漢字表記が正解の漢字表記にどの程度近いかを、fastTextでベクトル化しコサイン類似度を5段階のランプで示します。	1回
ヒント松	ヒント	10,000円	回答に反応するコンパス2つ。回答の読みの2文字目と3文字目について、五十音表上で正解の文字がある方向をコンパスで示します。	1回

食料は購入後に使用することで満腹度を回復します。ヒント竹と松は購入しただけでは発動せず、その後の回答に応じて情報を返します。
「平成」と答えたときはランプ2個、「永和」では4個点灯した、といった複数の観測から、装置が何を測っているのかを推測します。

人間がプレイできるUIを用意しています。

せんべいを食べたとき

ヒント梅を購入したとき

ヒント竹と松を購入したときと、購入後に回答したとき。ランプとコンパスの反応の意味は教えてもらえず、自分で推測しなければいけません。

なぜAIエージェントの題材になるのか

このゲームでは、プレイヤーは次のループを何度も繰り返します。

ゲームの状態を観測する
        ↓
次の行動を決める
        ↓
購入・回答・日送りを実行する
        ↓
結果を受け取り、仮説を更新する

複数ターンにわたる判断が必要になります。

お金と満腹度を管理する
どのヒントを試すか決める
回答結果を記憶する
ヒントの反応から仕組みを推測する
これまでの仮説を更新する

実験環境

ゲーム本体は、FastAPIでAPIサーバーとして実装しました。ゲームの状態はSQLiteに保存し、1つのゲームを識別するrun_idを基準に管理しています。

人間はWeb UIから、AIエージェントはREST APIまたはMCPからゲームを操作します。どの経路から操作しても同じゲームロジックが実行される構成です。
MCP（Model Context Protocol）とは、LLMアプリケーションから外部ツールやデータソースを共通形式で利用するためのプロトコルです。

人間 ── Web UI ─────────┐
                        ├── APIサーバー ── SQLite
AIエージェント ── MCP ────┘

AIエージェントにとって、APIサーバーはゲームの環境です。現在の状態を取得したり、ショップを確認したり、アイテムを購入したり、元号を回答したりします。

Codexで攻略できることの確認

エージェントを自作する前に、まずCodexのサブエージェントにゲームをプレイさせました。

このテストでは、サブエージェントにはゲームサーバーが公開するMCPツールだけを使わせます。実装コードやScenarioファイル、データベース、ゲームの正解は参照させません。つまり、プレイヤーがゲーム内で得られる情報だけを使って攻略します。

Daily Challengeで資金を得てコンパスを購入するところ

結果として、Codexのサブエージェントはゲームをクリアできました。ちなみに一度GPT-5.6 Lunaでプレイさせましたがクリアはできませんでした。

この結果から、ゲームの難易度が高すぎて攻略不能なわけではないことを確認できました。

エージェント実装

Codexが解けるのに、なぜ自作するのか

Codexのサブエージェントは汎用的でいろいろなことができます。しかしすべてのユースケースに当てはまることはないでしょう。

実際のシステムでは、予算、コンプライアンス、セキュリティ、監査要件などの制約があります。バックエンドで大量のエージェントを実行する場合は、実行回数やコストを細かく制御したり、利用できるツールやデータを限定したりする必要もあります。Codexのような汎用エージェントが、そのまま要件に合うとは限りません。

ハーネスとは何か

LLMは入力を受け取って次の出力を生成する推論器です。ゲームの状態を見て「次はショップを確認する」と判断することはできますが、それだけでは実際にショップを確認したことにはなりません。

そこで必要になるのがハーネスです。

ハーネスは、LLMと外部環境の間をつなぐプログラムです。環境の状態をLLMに渡し、LLMが選んだ行動を検証して実行し、その結果をまたLLMに返します。記憶、再試行、停止条件、ログ管理などを担当することもあります。

環境を観測
  ↓
LLMに渡す
  ↓
LLMが次の行動を選ぶ
  ↓
ハーネスが検証・実行
  ↓
結果を次の入力に加える

この記事では、LLMとハーネスを組み合わせたシステムをAIエージェントとして扱います。今回は、このハーネスを最小構成から実装していきます。

なお、「CodexにはAGENTS.mdがある。ハーネスってこれのことでしょ」と思う方もいるかもしれません。
広い意味では、AGENTS.mdのような指示ファイルもハーネスを構成する要素の一つと考えられます。ただし、この記事ではハーネスを、LLMと外部環境を接続し、観測を渡し、LLMの出力を検証・実行する制御プログラムという意味で使います。

最小構成：単発推論

まずは、LLMに毎ターン1つの行動だけを選ばせるAgentを作ります。

このAgentは、次の処理を繰り返します。

ゲームの状態を取得
  ↓
LLMに状態と直近の結果を渡す
  ↓
LLMが次の行動をJSONで返す
  ↓
ハーネスが内容を検証してMCPを実行

LLMには、現在のゲーム状態、直前の行動結果、直近の行動履歴、確認済みのショップ情報を渡します。そして、許可された行動の中から1つを選び、次のようなJSONを返してもらいます。

{
  "reason": "まずショップを確認して購入可能なアイテムを調べる",
  "action": {
    "kind": "inspect_shop",
    "arguments": {}
  }
}

この段階では、LLMにMCPのツールを直接渡していません。LLMが返すのはあくまでゲーム内の行動を表すJSONで、ハーネスがそれを対応するMCPツールへ変換します。

また、LLMの出力をそのまま実行することもありません。行動の種類や引数、所持金、インベントリなどをハーネス側で検証し、不正な出力なら実行せずに再試行します。実行結果はログに記録し、次のターンの入力へ渡します。

ただし、長期的な記憶や明示的な候補管理はありません。直近の情報は参照できますが、ゲーム全体を通した仮説を安定して保持する仕組みは、まだLLMの出力に任されています。

class SingleDecisionAgent:

    def __init__(self, client: McpGameClient, decider: Decider, log_path: Path, max_turns: int = 80):
        # 略
        # - McpGameClient: ゲームサーバクライアント
        # - Decider: OpenAI APIクライアント

    async def run(self) -> dict[str, Any]:
        state = await self.client.new_game()
        run_id = state.get("run_id")
        if not isinstance(run_id, str):
            raise RuntimeError("new_game did not return run_id")
        shop, latest, history, invalid = {}, None, [], 0
        self.log("game_started", state=state)

        # Agent Loop 本体
        # contextを作りdecideでLLMに推論させる。出力されたdecisionで次のactionを実行
        for turn in range(1, self.max_turns + 1):
            if state.get("status") != "playing":
                return self.finish(state, run_id, turn - 1)
            context = {"state": state, "latest_result": latest, "recent_actions": history[-6:], "known_shop": list(shop.values())}
            self.log("decision_requested", turn=turn, context=context)
            try:
                decision = await self.decider.decide(context)
                action = self.validate(decision.get("action"), state, shop)
                invalid = 0
            except (ValueError, RuntimeError) as exc:
                # 例外処理。略

            # 後処理。略
            # McpGameClientでactionの実行、log追加、state更新など

        return self.finish(state, run_id, self.max_turns, "max_turns")

decideはこんな感じです。

PROMPT = """You are playing a game through MCP.
Your goal is to clear the game while avoiding game over.
Balance progress toward clearing, information gathering, money, and survival.
Choose exactly one action and
return JSON: {\"reason\": string, \"action\": {\"kind\": allowed string, \"arguments\": object}}."""

# （中略）
    async def decide(self, context: dict[str, Any]) -> dict[str, Any]:
        payload = {"model": self.model, "input": f"{PROMPT}\nAllowed actions: {sorted(TOOLS)}\n{json.dumps(context, ensure_ascii=False)}", "text": {"format": {"type": "json_object"}}}
        response = await asyncio.to_thread(self.post, payload)
        text = self.output_text(response)
        if not isinstance(text, str):
            raise RuntimeError(f"OpenAI response did not contain text: {self.response_shape(response)}")
        value = json.loads(text)
        if not isinstance(value, dict):
            raise RuntimeError("LLM decision was not an object")
        return value

AIエージェントとして動きはしますが、解くことはできません。この構成では複数ターンにわたる探索や、過去のヒントを使った推理には限界があります。そこで次は、記憶や候補管理を追加した改善版を作ります。

改善版：記憶と候補管理

エージェント側に作業メモリを追加します。メモリには、次の4種類の情報を保存します。

これまでに分かった事実
そこから立てた仮説
ゲーム上の制約
試した元号候補と、その結果

LLMには毎ターン、このメモリと最新の観測を渡します。そして、次の行動を1つ選ぶ前に、メモリを更新し、候補を1〜3個比較させます。

{
  "summary": "候補を絞るため、まず追加の観測を行う",
  "memory": {
    "facts": ["これまでに得た公開情報"],
    "hypotheses": ["現在の仮説"],
    "constraints": ["満腹度と所持金に関する制約"],
    "candidates": [
      {
        "candidate": "候補の漢字",
        "reading": "候補の読み",
        "status": "untested"
      }
    ]
  },
  "evaluation": [
    {
      "kind": "inspect_shop",
      "benefit": "購入可能な行動を確認できる",
      "risk": "1ターン消費する",
      "score": 3
    }
  ],
  "action": {
    "kind": "inspect_shop",
    "arguments": {}
  }
}

class ImprovedAgent:
    """Maintains public-memory, asks for an evaluated plan, then takes one action."""

    def __init__(self, client: McpGameClient, decider: Decider, log_path: Path, max_turns: int = 80):
        self.client, self.decider, self.log_path, self.max_turns = client, decider, log_path, max_turns
        self.memory = Memory()
        log_path.parent.mkdir(parents=True, exist_ok=True)

    async def run(self) -> dict[str, Any]:
        state = await self.client.new_game()
        run_id = state.get("run_id")
        if not isinstance(run_id, str):
            raise RuntimeError("new_game did not return run_id")
        shop, latest, recent_plans, feedback, invalid = {}, None, [], [], 0
        self.log("game_started", state=state, memory=self.memory.as_dict())

        for turn in range(1, self.max_turns + 1):
            if state.get("status") != "playing":
                return self.finish(state, run_id, turn - 1)
            context = {
                "public_state": state,
                "latest_public_result": latest,
                "player_memory": self.memory.as_dict(), # Contextに作業メモリを入れる
                "recent_plans": recent_plans[-4:],
                "decision_shape": { # LLM出力の形式を指定。ここでmemoryの出力形式も指定している
                    "summary": "string",
                    "memory": {"facts": ["string"], "hypotheses": ["string"], "constraints": ["string"], "candidates": [{"candidate": "era name only", "reading": "its reading", "status": "optional string"}]},
                    "evaluation": [{"kind": "allowed action", "benefit": "string", "risk": "string", "score": "1..5"}],
                    "action": {"kind": "allowed action", "arguments": "object"},
                },
                # 以下略。実行可能アクションなどを記載
            }
            self.log("plan_requested", turn=turn, context=context)
            proposal: Any = None
            try:
                proposal = await self.decider.decide(context)
                plan, action = self.validate_plan(proposal, state, shop)
                self.memory.update(plan["memory"])
                feedback = action_format_feedback(proposal)
                invalid = 0
            except (ValueError, RuntimeError) as exc:
                # 例外処理。略

            # 後処理。略
            # McpGameClientでactionの実行、log追加、state更新、memory更新など

元号の回答が不正解だった場合は、ハーネスの後処理でその候補をwrongとしてメモリに記録します。これにより、同じ候補を何度も試すことを避けやすくなります。

この版では、LLMが「考えたこと」を次のターンへ明示的に引き継げるようになりました。

このハーネスであればGPT-5.6 Solでだいたいゲームをクリアできます。GPT-5.6 Lunaではなかなかクリアできず、クリア率は数%です。

Tool calling版

単発推論版と改善版では、LLMに専用のJSON形式を返してもらい、それをハーネスがMCPツールへ変換していました。

Tool calling版では、MCPサーバーが公開しているツール定義を、そのままLLMのFunction toolとして渡します。LLMは次の行動を、通常の文章や独自形式のJSONではなく、Function callとして返します。

MCPのツール定義を取得
  ↓
LLMにツールとして渡す
  ↓
LLMがツール名と引数を選ぶ
  ↓
ハーネスがMCPを実行
  ↓
実行結果を次のターンに渡す

例えば、LLMは次のような呼び出しを返します。

{
  "name": "game_state",
  "arguments": {}
}

LLMに推論させる時、contextだけでなくtoolsを渡しています。
この実装は単発推論にtool callingを導入したものです。

class ToolCallingAgent:
    """Run the four-step tool-calling loop until terminal state or turn limit."""

    def __init__(self, client: McpGameClient, caller: ToolCaller, log_path: Path, max_turns: int = 80, recent_limit: int = 6):
        # 略

    async def run(self) -> dict[str, Any]:
        state = await self.client.new_game()
        run_id = state.get("run_id")
        if not isinstance(run_id, str):
            raise RuntimeError("new_game did not return run_id")
        history: list[dict[str, Any]] = []
        self.log("game_started", state=state)

        for turn in range(1, self.max_turns + 1):
            if state.get("status") != "playing":
                return self.finish(state, run_id, turn - 1)

            context = {"state": state, "recent_results": history[-self.recent_limit :]}
            self.log("decision_requested", turn=turn, context=context)
            try:
                decision = await self.caller.decide(context, self.client.tools())
                name, arguments = self.read_action_plan(decision)
                result = await self.client.execute(run_id, name, arguments)
                record_execution = getattr(self.caller, "record_execution", None)
                if callable(record_execution):
                    record_execution(decision, result)
            except (RuntimeError, ValueError, TypeError) as exc:
                # 例外処理。略

            # 後処理。略

        return self.finish(state, run_id, self.max_turns, "max_turns")

decideの内容もやや変わります。Tool calling用のpayloadにします。


    async def decide(
        self, context: dict[str, Any], tools: list[dict[str, Any]]
    ) -> dict[str, Any]:
        """Return ``{tool_name, arguments}`` parsed from one model call."""
        payload: dict[str, Any] = {
            "model": self.model,
            "instructions": PROMPT,
            "tools": tools,
            "tool_choice": "required",
            "parallel_tool_calls": False,
            "store": True,
        }

ハーネスはこの呼び出しにゲームのrun_idを補い、MCPサーバーへ送信します。返ってきた結果は、次のリクエストのfunction_call_outputとしてLLMに渡します。

この方式では、ゲーム固有の行動一覧や引数の変換をPython側に個別に書く必要がありません。サーバーから取得したツール定義を使えるため、MCP側でツールを追加・変更したとき、エージェント側で独自のaction schemaを二重管理しなくて済みます。

ただし、tool callingを導入しただけで、エージェントが自動的に計画したり記憶したりするわけではありません。今回の実装も、1ターンに1つのツールを呼ぶ最小構成です。記憶や候補管理、停止判断などは、引き続きハーネス側で設計する必要があります。

OpenAI Agents SDKで実装

ここまで、Agentのループを自分で実装してきました。次は同じゲームを、OpenAI Agents SDKを使って実装します。

Agents SDKでは、Agentに指示とツールを渡し、Runner.runを呼び出すだけで、LLMへの入力、ツール呼び出し、結果の受け渡し、ターンの継続をSDKが管理してくれます。OpenAI Agents SDK公式ドキュメント

今回は、ゲームのMCPサーバーをMCPServerStreamableHttpで接続しました。Python側でゲーム開始時にrun_idを取得したあとは、ゲームのツール選択と実行をAgents SDKに任せます。

さらに今回は、戦略監査Agentをagents as toolsとして親Agentに追加しました。親Agentが購入や回答などの行動を選ぶ前に、現在の公開状態と行動案を監査Agentへ渡します。

親Agent
  ↓ audit_action
戦略監査Agent
  ↓ approve / reject / revise
親Agentがゲームツールを実行

async def main_async(args: argparse.Namespace) -> int:
    if args.max_turns < 1:
        raise ValueError("--max-turns must be at least 1")

    sdk = load_agents_sdk()
    Agent = sdk.Agent
    Runner = sdk.Runner
    mcp_module = importlib.import_module("agents.mcp")
    model_settings_module = importlib.import_module("agents.model_settings")
    audit_agent = Agent(name="Strategy auditor", model=args.model, output_type=AuditResult, instructions=("略。あなたは戦略監査役を担うサブエージェントです的プロンプト"))

    async with PublicMcpClient(args.mcp_url) as public_client:
        initial = await public_client.call("new_game", {})
        run_id = initial.get("run_id")
        if not isinstance(run_id, str):
            raise RuntimeError("new_game did not return run_id")

        tool_filter = mcp_module.create_static_tool_filter(blocked_tool_names=["new_game"])
        audit_tool = audit_agent.as_tool(
            tool_name="audit_action",
            tool_description=(
                "Review one proposed game action for public-state validity, resource risk, "
                "duplicate guesses, and safer alternatives. This tool never changes the game."
            ),
            parameters=AuditRequest,
            max_turns=2,
        )
        async with mcp_module.MCPServerStreamableHttp(name="Gengo game MCP", params={"url": args.mcp_url, "timeout": 30}, cache_tools_list=True, max_retry_attempts=2, tool_filter=tool_filter) as game_server:
            agent = Agent(
                name="Gengo game player",
                model=args.model,
                instructions=(
                    "略。ゲームをクリアして的なプロンプト"
                ),
                tools=[audit_tool],
                mcp_servers=[game_server],
                model_settings=model_settings_module.ModelSettings(
                    tool_choice="required", parallel_tool_calls=False
                ),
            )
            try:
                result = await Runner.run(
                    agent,
                    json.dumps(
                        {"run_id": run_id, "public_state": initial},
                        ensure_ascii=False,
                    ),
                    max_turns=args.max_turns,
                )
            except BaseException as exc:  # preserve terminal state in the log
                error = exc

        final_state = await public_client.call("game_state", {"run_id": run_id})

（正直、私自身もまだ完全には理解できていません）

評価ハーネス

作ったエージェントをどう評価すればいいでしょうか。
エージェントを一度動かして「解けた／解けなかった」を見るだけでは、たまたま成功したのか、改善によって本当に性能が上がったのか分かりません。モデルやプロンプトを変えるたびに、結果を手作業で比べるのも大変です。
そこで、エージェントを同じ条件で何度も実行し、結果を集計する評価ハーネスを用意します。固定したゲームシナリオに対して、クリア率、ターン数、重複回答、無効な操作などを記録します。

評価ハーネス
  ├─ ゲームサーバーを起動
  ├─ エージェントを指定回数実行
  ├─ 実行ログを収集
  └─ 結果を集計

（実装略）

まとめ

最小ループだけでは長期探索が苦しい
記憶・候補管理を追加すると改善した
Tool callingを使うと環境との境界を整理できる
SDKを使えばagent loopそのものを自作しなくてもよい
モデルによって同じハーネスでも結果が変わる
評価には複数試行が必要

感想ですが、改善版エージェントでGPT-5.6 Solでクリアでき、GPT-5.6 Lunaではクリアがほとんどできなかったことから、モデル性能が上がるほど、ハーネスは「LLMの性能を引き出すもの」から「LLMに制約を守らせるもの」になるのかもと考えました。
プロンプトで「xxxは禁止」と言っても必ず守ってくれる保証はないです。そこでハーネスで決定論的にxxをさせないような実装をするイメージです。