Tech Racho エンジニアの「?」を「!」に。
  • Ruby / Rails以外の開発一般

Python: AIエージェント用OSSフレームワーク"SAGA"でAI闘技場

🔗目次

  1. まえがき
  2. 今回作りたいもの
  3. 完成品
  4. SAGAって何?
  5. 環境構築
  6. 試運転
  7. 実装
  8. あとがき
  9. 参考記事

🔗まえがき

みなさんどうもこんにちは、株式会社ECN所属インターンFuseです。あけましておめでとうございます。
突然ですが、皆さんはこの記事を憶えていますか?

OpenAI:AIに自作したTRPGもどきを遊ばせる~AI対AIにAIを添えて~

AIにプロンプトでルールを与えて行動を生成させて殴り合わせようぜという企画だったのですが、その記事のあとがきの中で私はこんなことを言っていました。

そしてこの3つの課題点は、すべてプログラムに任せることで解決できます。
ダイス振りやダメージ計算といったシステム面をプログラムに任せてしまうことでプロンプトのテキスト量を節約しつつ、不確定要素を減らせます。
また、プロンプトに記入するには複雑すぎるアイテムやNPCのデータベースも、オブジェクトとして保存し必要に応じプロンプトに埋め込むといった形がとれます。

いつかやろうと思いながらも忙しくてなかなか取り組めなかったのですが、

  • 学校の冬休みとECNの冬休みが重なり、いい感じのまとまった休みが手に入ったこと
  • 会社全体でAIを活用する方向へと舵切りが行われ、それがモチベーションになった

などのこともあり、執筆に踏み切りました。
↑目次に戻る

🔗今回作りたいもの

  • 仮想空間上の闘技場で、複数のエージェントがタッグで戦う
  • エージェントたちは自分の番が回ると複数の行動から1つを選び実行する
  • 戦闘ロジックはプログラムとして書かれている
  • 決着がつくまで戦い続ける

↑目次に戻る

🔗完成品

↑目次に戻る

🔗SAGAって何?

fablestudio/fable-saga - GitHub

SAGASimulation Incによって作られたフレームワークで、あらゆるシミュレーションのアクションを生成するために使用できます。
必要なデータを与えることで、エージェントが行うアクションの候補をいくつか生成し、スコアリングして返してくれます。

SAGAはJoon Park氏の論文 "Generative Agents: Interactive Simulacra of Human Behavior"から部分的にインスピレーションを受けており、デモである"Murder in Thistle Gulch"(訳:シズルガルチ殺人事件)では保安官"カッパー"や犯罪組織のリーダー"ブラックジャック"などの複数のエージェントが独立した記憶を保持しながら定義されたそれぞれの目的のために世界にあるオブジェクトやほかのエージェントと相互作用する様子が確認できます。

それでは実際に使ってみましょう。
↑目次に戻る

🔗環境構築

まずはpyenvを用意します

pyenv/pyenv - GitHub

まずはPoetryと相性のいいバージョン管理ソフトpyenvを用意します。最新バージョンでないPythonが後々必要になるため、今のうちにバージョンを切り替えられるようにしておきましょう。

pipからインストールします。

pip install pyenv-win --target $HOME\\.pyenv

その後、公式のガイドに従いパスなどの環境変数を設定します。

[System.Environment]::SetEnvironmentVariable('PYENV',$env:USERPROFILE + "\.pyenv\pyenv-win\","User")
[System.Environment]::SetEnvironmentVariable('PYENV_ROOT',$env:USERPROFILE + "\.pyenv\pyenv-win\","User")
[System.Environment]::SetEnvironmentVariable('PYENV_HOME',$env:USERPROFILE + "\.pyenv\pyenv-win\","User")
[System.Environment]::SetEnvironmentVariable('path', $env:USERPROFILE + "\.pyenv\pyenv-win\bin;" + $env:USERPROFILE + "\.pyenv\pyenv-win\shims;" + [System.Environment]::GetEnvironmentVariable('path', "User"),"User")

その後、バージョン確認ついでにうまくパスが通ってるかを確認します。

pyenv --version

こんな感じにバージョンが表示されればOKです、お疲れさまでした。

pyenv 3.1.1

次にPoetryを用意します

python-poetry/poetry - GitHub

必要なパッケージ管理ソフトであるpoetryをインストールします。

ここで注意点ですが、Pythonは必ず公式サイトからインストールしたものを使用して下さい。
MicroSoft Storeのものを使うとpoetryのパスがバグで通らない挙句インストールされているという判定になり何とかしてpoetryを消さない限りインストーラーも動かなくなりにっちもさっちもいかなくなります。

(Invoke-WebRequest -Uri https://raw.githubusercontent.com/python-poetry/poetry/master/get-poetry.py -UseBasicParsing).Content | python -

その後はPoetryのバージョン確認ついでにうまくパスが通ってるかを確認します。

poetry --version

こんな感じにバージョンが出力されたら成功です。お疲れ様でした。

Poetry (version 1.7.1)

実行環境の準備

SAGAを動かすにはバージョンが3.11のPythonが必要です。先ほど入手したpyenvを使って環境を構築します。
まずはpyenv install --listでインストールできるバージョンのリストを確認しましょう。

~~中略~~
3.11.0a1-win32
3.11.0a1
3.11.0a2-win32
3.11.0a2
3.11.0a3-win32
3.11.0a3
3.11.0a4-win32
3.11.0a4
3.11.0a5-win32
3.11.0a5
3.11.0a6-win32
3.11.0a6
3.11.0a7-win32
3.11.0a7
3.11.0b1-win32
3.11.0b1
3.11.0b2
3.11.0b3-win32
3.11.0b4-win32
3.11.0b4

…なんか古くない?

2023年12月30日の安定版である3.12は含まれてないし、3.11も明らかにアルファやベータっぽいのしかないし、明らかに古いです。それもそのはず、インストールしたばっかりのpyenvはデータベースが古いので、aptみたいにupdateしてやる必要があります。

pyenv update

を実行してしばらく待つと、

:: [Info] ::  Mirror: https://www.python.org/ftp/python
:: [Info] ::  Scanned 204 pages and found 635 installers.

と表示され、データベースのアップデートが完了しました。
もう一度pyenv install --listで確認してみると。

~~中略~~
3.11.0a1-win32
3.11.0a1
3.11.0a2-win32
3.11.0a2
3.11.0a3-win32
3.11.0a3
3.11.0a4-win32
3.11.0a4
3.11.0a5-win32
3.11.0a5
3.11.0a6-win32
3.11.0a6
3.11.0a7-win32
3.11.0a7
3.11.0b1-win32
3.11.0b1
3.11.0b2-win32
3.11.0b2
3.11.0b3-win32
3.11.0b3
3.11.0b4-win32
3.11.0b4
3.11.0b5-win32
3.11.0b5
3.11.0rc1-win32
3.11.0rc1
3.11.0rc2-win32
3.11.0rc2
3.11.0-win32
3.11.0
3.11.1-win32
3.11.1
3.11.2-win32
3.11.2
3.11.3-win32
3.11.3
3.11.4-win32
3.11.4
3.11.5-win32
3.11.5
3.11.6-win32
3.11.6
3.11.7-win32
3.11.7
3.12.0a1-win32
3.12.0a1
3.12.0a2-win32
3.12.0a2
3.12.0a3-win32
3.12.0a3
3.12.0a4-win32
3.12.0a4
3.12.0a5-win32
3.12.0a5
3.12.0a6-win32
3.12.0a6
3.12.0a7-win32
3.12.0a7
3.12.0b1-win32
3.12.0b1
3.12.0b2-win32
3.12.0b2
3.12.0b3-win32
3.12.0b3
3.12.0b4-win32
3.12.0b4
3.12.0rc1-win32
3.12.0rc1
3.12.0rc2-win32
3.12.0rc2
3.12.0rc3-win32
3.12.0rc3
3.12.0-win32
3.12.0
3.12.1-win32
3.12.1
3.13.0a1-win32
3.13.0a1
3.13.0a2-win32
3.13.0a2

ばっちり最新の情報が表示されましたね。
それでは、さっそく仮想環境を用意していきましょう。

pyenv install 3.11.7 #3.11.7をインストールして
pyenv global 3.11.7 #3.11.7をデフォルトのバージョンに指定する
poetry env use 3.11.7 #poetryに3.11.7を使わせる

こんな表示が最終的に出ればOKです。

Creating virtualenv fable-saga-QE3P02ha-py3.11 in C:\Users\ユーザー名\AppData\Local\pypoetry\Cache\virtualenvs
Using virtualenv: C:\Users\ユーザー名\AppData\Local\pypoetry\Cache\virtualenvs\fable-saga-QE3P02ha-py3.11

SAGAの準備

必要な物はそろったので、試しにSAGAのリポジトリ内にある宇宙船デモを動かしてみましょう。
まずはリポジトリをローカルにクローンします。

git clone https://github.com/fablestudio/fable-saga.git

次に依存関係を解決します。

poetry install

これで準備完了、さっそく動かしてみましょう。
↑目次に戻る

🔗 試運転

poetry shell

で仮想環境内のシェルに入り、

python demos/space_colony/simulation.py

で実行します!

====  SHIP TIME: 2060-01-01 08:00:00  ====
---- elara_sundeep ----
  ROLE: As the Captain, Elara is responsible for making the strategic decisions for the ship's routes and trade agreements. She also serves as the public face of their independent shipping operations and ensures the safety and morale of her crew.
  LOCATION: Crew Quarters Corridor
  ACTION: Idle
  MEMORIES:

---- vance_tormun ----
  ROLE: Vance is the chief engineer and is responsible for the maintenance and repair of the ship, ensuring that it can endure the long treks between colonies. His ingenuity has saved the "Stellar Runner" from more than a few scrapes.
  LOCATION: Crew Quarters Corridor
  ACTION: Idle
  MEMORIES:

---- nyah_kobari ----
  ROLE: As the head of security, Nyah ensures the safety of the cargo and crew from pirates, smugglers, and other potential threats. She also serves as a tactical advisor to Captain Sundeep and leads boarding and inspection procedures.
  LOCATION: Crew Quarters Corridor
  ACTION: Idle
  MEMORIES:

---- jax_emerson ----
  ROLE: The ship's main pilot, Jax handles the controls with an expert touch and is responsible for navigating through debris fields, asteroid belts, and the occasional high-speed getaway.
  LOCATION: Crew Quarters Corridor
  ACTION: Idle
  MEMORIES:

---- linh_le ----
  ROLE: As the medical officer, Linh is responsible for the physical and mental health of the crew, managing everything from routine check-ups to emergency medical procedures. Additionally, she oversees the hydroponic gardens that supplement the ship's food supply and contribute to oxygen regeneration.
  LOCATION: Crew Quarters Corridor
  ACTION: Idle
  MEMORIES:

Generating actions for elara_sundeep ...
Generating actions for vance_tormun ...
Generating actions for nyah_kobari ...
Generating actions for jax_emerson ...
Generating actions for linh_le ...

このように、エージェントごとにアクションの生成が行われていますね。
それでは、これを使って闘技場プログラムを作っていきましょう。
↑目次に戻る

🔗実装

それでは早速闘技場のプログラムを作っていきましょう。
方針としては同梱されている宇宙船デモに倣い

  • プログラムはモデル・アクション・本体で分ける
  • シミュレーションに使うデータはハードコーディングせず、yamlファイルに書く

事にします。

lf_models.py

まずはシミュレーションに使うデータの構造を定義していきます。
具体的には

  • 戦士・魔法使いのような職業
  • 個人名やほかのファイターとの関係などを含むファイター
  • 攻撃や魔法などのスキル

を定義していきます。
宇宙船デモのdemos\space_colony\sim_models.pyをベースに改造していきましょう。

from abc import ABC, abstractmethod
from typing import List, Dict, Tuple
from attr import define

EntityId = str


class EntityInterface(ABC):
    @abstractmethod
    def id(self) -> EntityId:
        pass


@define(slots=True)
class Job(EntityInterface):
    guid: EntityId
    name: str
    max_health: int
    min_strength: int
    max_strength: int
    min_armor: int
    max_armor: int
    min_agility: int
    max_agility: int
    min_magic: int
    max_magic: int
    skill_names: List[str]  # 使えるスキルの名前一覧
    role: str  # 行動指針

    def id(self) -> EntityId:
        return self.guid


@define(slots=True)
class Fighter(EntityInterface):
    guid: EntityId
    job: EntityId  # 職業のID
    partner: EntityId  # タッグ相手のID
    enemy: List[EntityId]  # 敵のID一覧
    name: str

    def id(self) -> EntityId:
        return self.guid


@define(slots=True)
class Skill(EntityInterface):
    guid: EntityId
    description: str
    mp_cost: int  # 必要なMP
    parameters: Dict[str, str]

    def id(self) -> EntityId:
        return self.guid

キャラクターごとに個性を持たせるために職業のデータに最低値と最大値を持たせ、ランダムにステータスを生成できるようにしています。
データ構造を定義したら、さっそく実データをyamlファイルに記述していきましょう。

resources\skills.yaml

- name: attack
  description: "Attack to Enemy.Damage inflicted depends on STR."
  parameters:
    target: "<str: targets fighter_guid>"
  mp_cost: 0

- name: magic_attack
  description: "Attack to Enemy With Magic.Damage inflicted depends on MAG. (require 1 MP)"
  parameters:
    target: "<str: targets fighter_guid>"
  mp_cost: 1

- name: cover
  description: "Shield your partner from attacks this turn."
  parameters: {}
  mp_cost: 0

- name: heal
  description: "Heal fighter's health.Use when partner or own Health is less than half. (require 2 MP)"
  parameters:
    target: "<str: targets fighter_guid>"
  mp_cost: 2

名前と説明文、必要ならば引数の型と説明に加え、必要MPを書きました。

resources\jobs.yaml

- guid: swordsman
  name: 剣士
  max_health: 30
  min_strength: 7
  max_strength: 10
  min_armor: 5
  max_armor: 8
  min_agility: 3
  max_agility: 6
  min_magic: 1
  max_magic: 4
  skill_names: 
    - attack
    - cover
  role: You are a swordsman. You must cooperate with your partner, a wizard, to defeat all your enemies.You must use your high durability to protect your partner and attack enemies with your sword.

- guid: wizard
  name: 魔法使い
  max_health: 20
  min_strength: 1
  max_strength: 4
  min_armor: 2
  max_armor: 5
  min_agility: 1
  max_agility: 4
  min_magic: 6
  max_magic: 9
  skill_names: 
    - attack
    - magic_attack
    - heal
  role: You are a wizard. You must work together with your partner, a swordsman, to defeat all your enemies.Use magic to attack enemies and Recover when you or your partner's health is depleted.But watch out for your MP, because if you run out, you will have very little to do.

各種ステータスの最低値、最大値や使用可能スキル一覧、行動指針などを書きました。

resources\fighters.yaml

- guid: alice
  name: アリス
  job: swordsman
  partner: bob
  enemy:
    - cherry
    - dave
- guid: bob
  name: ボブ
  job: wizard
  partner: alice
  enemy:
    - cherry
    - dave
- guid: cherry
  name: チェリー
  job: swordsman
  partner: dave
  enemy:
    - alice
    - bob
- guid: dave
  name: デイヴ
  job: wizard
  partner: cherry
  enemy:
    - alice
    - bob

ファイターごとのジョブや名前、パートナーや敵を定義しています。
今回は2対2ですが、おそらく3チーム戦や4チーム戦も可能だと思います。

lf_main.py

import asyncio
import json
import pathlib
from typing import List, Dict, Optional, Any
import cattrs
import yaml
import fable_saga
import lf_models
import lf_actions
import random


class FighterAgent:  # 実際に戦闘するエージェント
    def __init__(self, guid: lf_models.EntityId) -> None:
        self.guid = guid
        self.fighter: Optional[lf_models.Fighter] = None
        self.skills: List[lf_models.Skill] = []
        self.max_health = 0  # 最大HP
        self.health = 0  # HP
        self.mp = 0  # MP
        self.strength = 0  # 物理攻撃
        self.armor = 0  # 物理防御
        self.agility = 0  # すばやさ
        self.magic = 0  # 魔法攻撃兼魔法防御
        self.alive = True  # 戦闘可能か?
        self.is_human = False  # 人間操作か?
        self.covering = False  # パートナーをかばう準備はできているか?
        pass

    async def turn(self, sim: "Simulation"):  # ターンが回ってきたら
        self.covering = False

        def list_actions(actions: fable_saga.GeneratedActions):  # アクションの一覧を標準出力
            for i, action in enumerate(actions.options):
                output = f"#{i} -- {action.skill} ({actions.scores[i]})\n"
                for key, value in action.parameters.items():
                    output += f"  {key}: {value}"
                print(output)

        def choose_action():
            item = input(f"{self.fighter.id()}のコマンドを選んでください...")
            return int(item)

        def handle_action(action: fable_saga.Action):  # アクションインスタンスの生成と実行
            from lf_actions import Attack, MagicAttack, Cover, Heal

            n_action: Optional[lf_actions.SimAction] = None
            # action.skillの値に応じてインスタンス生成
            if action.skill == "attack":
                n_action = Attack(self, action)
            elif action.skill == "magic_attack":
                n_action = MagicAttack(self, action)
            elif action.skill == "cover":
                n_action = Cover(self, action)
            elif action.skill == "heal":
                n_action = Heal(self, action)
            else:
                print(f"存在しないアクションです。 {action.skill}.")
            if not n_action is None:
                n_action.run(sim)  # 存在するアクションなら実行

        if self.is_human:
            actions: fable_saga.GeneratedActions = await sim.generate_actions(
                self, verbose=False
            )
            print(f"\n========== {self.fighter.id()} ===========")
            actions.sort()
            list_actions(actions)
            while True:
                idx = choose_action()
                if 0 <= idx < len(actions.options):
                    print(f"{actions.options[idx].skill}が選ばれました。")
                    handle_action(actions.options[idx])
                    break
                else:
                    print(f"不正な入力です {idx}.")
        else:
            actions = await sim.generate_actions(self, verbose=False)
            print(f"\n========== {self.fighter.id()} ===========")
            actions.sort()
            handle_action(actions.options[0])

    def hurt(self, val: int):  # 被ダメージ処理
        print(f"{self.fighter.name}は{val}のダメージを受けた!")
        self.health = max(0, self.health - val)  # 0未満にはならない
        print(f"残りHP:{self.health}")
        if self.health <= 0:  # HP0以下なら
            print(f"{self.fighter.name}は倒れた!")
            self.covering = False
            self.alive = False  # 戦闘不能

    def heal(self, val: int):  # 回復処理
        print(f"{self.fighter.name}のHPが{val}回復した!")
        self.health = min(self.max_health, self.health + val)  # 最大HPを超えることはない
        print(f"残りHP:{self.health}")


class Simulation:  # シミュレーション本体
    def __init__(self):
        self.turn = 0  # 現在のターン数
        self.agents: Dict[lf_models.EntityId, FighterAgent] = {}  # エージェントの一覧
        self.jobs: Dict[lf_models.EntityId, lf_models.Job] = {}  # 職業の一覧
        self.skills: Dict[str, lf_models.Skill] = {}  # スキルの一覧
        self.saga_agent = fable_saga.Agent()  # こっちはアクションを生成するエージェント

    def load(self):  # データを読み込む
        path = pathlib.Path(__file__).parent.resolve()

        with open(path / "resources/jobs.yaml", "r") as f:  # 職業を読み込む
            for job_data in yaml.load(f, Loader=yaml.FullLoader):
                obj = cattrs.structure(job_data, lf_models.Job)
                self.jobs[obj.id()] = obj

        with open(path / "resources/fighters.yaml", "r") as f:  # ファイターを読み込む
            for fighter_data in yaml.load(f, Loader=yaml.FullLoader):
                fighter = cattrs.structure(fighter_data, lf_models.Fighter)
                agent = FighterAgent(fighter.id())
                agent.fighter = fighter
                job = self.jobs[agent.fighter.job]
                # 職業のデータからステータスを生成する
                m_health = job.max_health
                str = random.randrange(job.min_strength, job.max_strength)
                arm = random.randrange(job.min_armor, job.max_armor)
                agi = random.randrange(job.min_agility, job.max_agility)
                mag = random.randrange(job.min_magic, job.max_magic)
                agent.max_health = m_health
                agent.health = agent.max_health
                agent.strength = str
                agent.armor = arm
                agent.agility = agi
                agent.magic = mag
                agent.mp = agent.magic
                self.agents[fighter.id()] = agent

        with open(path / "resources/skills.yaml", "r") as f:  # スキルを読み込む
            for skill_data in yaml.load(f, Loader=yaml.FullLoader):
                print(skill_data)
                skill = cattrs.structure(skill_data, lf_models.Skill)
                self.skills[skill.name] = skill
            for agent in self.agents.values():  # エージェントのスキルを設定する
                result = []
                for n in self.jobs[agent.fighter.job].skill_names:
                    result.append(self.skills[n])
                agent.skills = result

    async def tick(self):  # ターンを回す
        self.turn += 1
        for agent in self.agents.values():
            if agent.alive:
                await agent.turn(self)
            if not (self.agents["alice"].alive or self.agents["bob"].alive):  # アリボブ全滅
                print("=== 決着がつきました ===")
                break
            if not (
                self.agents["cherry"].alive or self.agents["dave"].alive
            ):  # チェリデイヴ全滅
                print("=== 決着がつきました ===")
                break

    async def generate_actions(
        self, sim_agent: FighterAgent, retries=0, verbose=False
    ) -> [List[Dict[str, Any]]]:  # アクションの精鋭
        print(f"Generating actions for {sim_agent.fighter.id()} ...")
        context = ""
        fighter = []
        for f in self.agents.values():
            if f.alive:
                fighter.append(
                    {
                        "guid": f.fighter.guid,
                        "Health": f"{f.health}/{f.max_health}",
                        "stat": {
                            "STR": f.strength,
                            "CON": f.armor,
                            "AGI": f.agility,
                            "MAG": f.magic,
                        },
                        "MP": f.mp,
                        "job": cattrs.unstructure(self.jobs[f.fighter.job]),
                    }
                )  # エージェントのデータを生成
        context += (
            "FIGHTER: They are warriors who fight in the arena of the region:\n"
            + f"{json.dumps(fighter)}\n"
        )

        if sim_agent.fighter is not None:
            context += f"You are a warrior in the arena. Join forces with your partner to reduce the HP of all enemies to zero."
            context += f"You are {sim_agent.fighter.id()}.\n\n"
            if self.agents[sim_agent.fighter.partner].alive:
                context += f"You are partner is {sim_agent.fighter.partner}.\n\n"
            enm = " and ".join(
                filter(lambda x: self.agents[x].alive, sim_agent.fighter.enemy)
            )
            context += f"You are enemy is {enm}.\n\n"
        useable = []
        for v in sim_agent.skills:
            if sim_agent.mp >= v.mp_cost:  # MP足りてたら
                useable.append(
                    fable_saga.Skill(
                        name=v.name, description=v.description, parameters=v.parameters
                    )
                )  # fable_saga.Skillに詰め込む
        return await self.saga_agent.generate_actions(
            context, useable, max_tries=retries, verbose=verbose
        )


async def main():
    sim = Simulation()
    sim.load()

    keep_running = True
    while keep_running:
        print("====  ターン" + str(sim.turn) + "  ====")
        for agent in sim.agents.values():
            print(f"---- {agent.fighter.name} ----")
            print(f"  HP: {agent.health}/{agent.max_health}")
            print(f"  MP: {agent.mp}")
        await sim.tick()
        if not (sim.agents["alice"].alive or sim.agents["bob"].alive):  # アリボブ全滅
            keep_running = False
        if not (sim.agents["cherry"].alive or sim.agents["dave"].alive):  # チェリデイヴ全滅
            keep_running = False

    print("終了します")
    exit(0)


if __name__ == "__main__":
    asyncio.run(main())

このファイルには

  • 実際に戦うエージェントのクラス FighterAgent
  • シミュレーションを総括するクラスSimulation
  • main処理

を書いています。内容が濃いので順を追って説明します。

FighterAgent

class FighterAgent:  # 実際に戦闘するエージェント
    def __init__(self, guid: lf_models.EntityId) -> None:
        self.guid = guid
        self.fighter: Optional[lf_models.Fighter] = None
        self.skills: List[lf_models.Skill] = []
        self.max_health = 0  # 最大HP
        self.health = 0  # HP
        self.mp = 0  # MP
        self.strength = 0  # 物理攻撃
        self.armor = 0  # 物理防御
        self.agility = 0  # すばやさ
        self.magic = 0  # 魔法攻撃兼魔法防御
        self.alive = True  # 戦闘可能か?
        self.is_human = False  # 人間操作か?
        self.covering = False  # パートナーをかばう準備はできているか?
        pass

    async def turn(self, sim: "Simulation"):  # ターンが回ってきたら
        self.covering = False

        def list_actions(actions: fable_saga.GeneratedActions):  # アクションの一覧を標準出力
            for i, action in enumerate(actions.options):
                output = f"#{i} -- {action.skill} ({actions.scores[i]})\n"
                for key, value in action.parameters.items():
                    output += f"  {key}: {value}"
                print(output)

        def choose_action():
            item = input(f"{self.fighter.id()}のコマンドを選んでください...")
            return int(item)

        def handle_action(action: fable_saga.Action):  # アクションインスタンスの生成と実行
            from lf_actions import Attack, MagicAttack, Cover, Heal

            n_action: Optional[lf_actions.SimAction] = None
            # action.skillの値に応じてインスタンス生成
            if action.skill == "attack":
                n_action = Attack(self, action)
            elif action.skill == "magic_attack":
                n_action = MagicAttack(self, action)
            elif action.skill == "cover":
                n_action = Cover(self, action)
            elif action.skill == "heal":
                n_action = Heal(self, action)
            else:
                print(f"存在しないアクションです。 {action.skill}.")
            if not n_action is None:
                n_action.run(sim)  # 存在するアクションなら実行

        if self.is_human:
            actions: fable_saga.GeneratedActions = await sim.generate_actions(
                self, verbose=False
            )
            print(f"\n========== {self.fighter.id()} ===========")
            actions.sort()
            list_actions(actions)
            while True:
                idx = choose_action()
                if 0 <= idx < len(actions.options):
                    print(f"{actions.options[idx].skill}が選ばれました。")
                    handle_action(actions.options[idx])
                    break
                else:
                    print(f"不正な入力です {idx}.")
        else:
            actions = await sim.generate_actions(self, verbose=False)
            print(f"\n========== {self.fighter.id()} ===========")
            actions.sort()
            handle_action(actions.options[0])

    def hurt(self, val: int):  # 被ダメージ処理
        print(f"{self.fighter.name}は{val}のダメージを受けた!")
        self.health = max(0, self.health - val)  # 0未満にはならない
        print(f"残りHP:{self.health}")
        if self.health <= 0:  # HP0以下なら
            print(f"{self.fighter.name}は倒れた!")
            self.covering = False
            self.alive = False  # 戦闘不能

    def heal(self, val: int):  # 回復処理
        print(f"{self.fighter.name}のHPが{val}回復した!")
        self.health = min(self.max_health, self.health + val)  # 最大HPを超えることはない
        print(f"残りHP:{self.health}")

FighterAgentクラスは戦うファイター1人につき1つ生成されるエージェントです。
ファイターごとのGUIDやステータスといったファイターごとの情報を保持しています。
また、HPの増減処理やアクションのハンドルもここで行っています。

Simuration

class Simulation:  # シミュレーション本体
    def __init__(self):
        self.turn = 0  # 現在のターン数
        self.agents: Dict[lf_models.EntityId, FighterAgent] = {}  # エージェントの一覧
        self.jobs: Dict[lf_models.EntityId, lf_models.Job] = {}  # 職業の一覧
        self.skills: Dict[str, lf_models.Skill] = {}  # スキルの一覧
        self.saga_agent = fable_saga.Agent()  # こっちはアクションを生成するエージェント

    def load(self):  # データを読み込む
        path = pathlib.Path(__file__).parent.resolve()

        with open(path / "resources/jobs.yaml", "r") as f:  # 職業を読み込む
            for job_data in yaml.load(f, Loader=yaml.FullLoader):
                obj = cattrs.structure(job_data, lf_models.Job)
                self.jobs[obj.id()] = obj

        with open(path / "resources/fighters.yaml", "r") as f:  # ファイターを読み込む
            for fighter_data in yaml.load(f, Loader=yaml.FullLoader):
                fighter = cattrs.structure(fighter_data, lf_models.Fighter)
                agent = FighterAgent(fighter.id())
                agent.fighter = fighter
                job = self.jobs[agent.fighter.job]
                # 職業のデータからステータスを生成する
                m_health = job.max_health
                str = random.randrange(job.min_strength, job.max_strength)
                arm = random.randrange(job.min_armor, job.max_armor)
                agi = random.randrange(job.min_agility, job.max_agility)
                mag = random.randrange(job.min_magic, job.max_magic)
                agent.max_health = m_health
                agent.health = agent.max_health
                agent.strength = str
                agent.armor = arm
                agent.agility = agi
                agent.magic = mag
                agent.mp = agent.magic
                self.agents[fighter.id()] = agent

        with open(path / "resources/skills.yaml", "r") as f:  # スキルを読み込む
            for skill_data in yaml.load(f, Loader=yaml.FullLoader):
                print(skill_data)
                skill = cattrs.structure(skill_data, lf_models.Skill)
                self.skills[skill.name] = skill
            for agent in self.agents.values():  # エージェントのスキルを設定する
                result = []
                for n in self.jobs[agent.fighter.job].skill_names:
                    result.append(self.skills[n])
                agent.skills = result

    async def tick(self):  # ターンを回す
        self.turn += 1
        for agent in self.agents.values():
            if agent.alive:
                await agent.turn(self)
            if not (self.agents["alice"].alive or self.agents["bob"].alive):  # アリボブ全滅
                print("=== 決着がつきました ===")
                break
            if not (
                self.agents["cherry"].alive or self.agents["dave"].alive
            ):  # チェリデイヴ全滅
                print("=== 決着がつきました ===")
                break

    async def generate_actions(
        self, sim_agent: FighterAgent, retries=0, verbose=False
    ) -> [List[Dict[str, Any]]]:  # アクションの精鋭
        print(f"Generating actions for {sim_agent.fighter.id()} ...")
        context = ""
        fighter = []
        for f in self.agents.values():
            if f.alive:
                fighter.append(
                    {
                        "guid": f.fighter.guid,
                        "Health": f"{f.health}/{f.max_health}",
                        "stat": {
                            "STR": f.strength,
                            "CON": f.armor,
                            "AGI": f.agility,
                            "MAG": f.magic,
                        },
                        "MP": f.mp,
                        "job": cattrs.unstructure(self.jobs[f.fighter.job]),
                    }
                )  # エージェントのデータを生成
        context += (
            "FIGHTER: They are warriors who fight in the arena of the region:\n"
            + f"{json.dumps(fighter)}\n"
        )

        if sim_agent.fighter is not None:
            context += f"You are a warrior in the arena. Join forces with your partner to reduce the HP of all enemies to zero."
            context += f"You are {sim_agent.fighter.id()}.\n\n"
            if self.agents[sim_agent.fighter.partner].alive:
                context += f"You are partner is {sim_agent.fighter.partner}.\n\n"
            enm = " and ".join(
                filter(lambda x: self.agents[x].alive, sim_agent.fighter.enemy)
            )
            context += f"You are enemy is {enm}.\n\n"
        useable = []
        for v in sim_agent.skills:
            if sim_agent.mp >= v.mp_cost:  # MP足りてたら
                useable.append(
                    fable_saga.Skill(
                        name=v.name, description=v.description, parameters=v.parameters
                    )
                )  # fable_saga.Skillに詰め込む
        return await self.saga_agent.generate_actions(
            context, useable, max_tries=retries, verbose=verbose
        )

Simulationクラスはシミュレーションを総括するためのクラスです。
シミュレーションに必要なデータを読み込み、エージェントを生成したり、エージェントたちにターンを回したりプロンプトを作るためのコンテキストを組み立ててアクションを生成させたりするクラスです。

main処理

async def main():
    sim = Simulation()
    sim.load()

    keep_running = True
    while keep_running:
        print("====  ターン" + str(sim.turn) + "  ====")
        for agent in sim.agents.values():
            print(f"---- {agent.fighter.name} ----")
            print(f"  HP: {agent.health}/{agent.max_health}")
            print(f"  MP: {agent.mp}")
        await sim.tick()
        if not (sim.agents["alice"].alive or sim.agents["bob"].alive):  # アリボブ全滅
            keep_running = False
        if not (sim.agents["cherry"].alive or sim.agents["dave"].alive):  # チェリデイヴ全滅
            keep_running = False

    print("終了します")
    exit(0)


if __name__ == "__main__":
    asyncio.run(main())

main処理ではシミュレーションを生成したり、決着がつくまでターンを回したりとゲーム全体を管理しています。

lf_actions.py

from datetime import timedelta
from typing import Dict, Any

import fable_saga
import random
from lf_main import FighterAgent, Simulation
from lf_models import EntityId


class SimAction:
    def __init__(self, agent: FighterAgent, action_data: fable_saga.Action):
        self.agent = agent
        self.action_data: fable_saga.Action = action_data
        self.skill: str = action_data.skill
        self.parameters: Dict[str, Any] = action_data.parameters

    def run(self, sim: Simulation):
        pass


class Attack(SimAction):  # 物理攻撃
    def __init__(self, agent: FighterAgent, action_data: fable_saga.Action):
        super().__init__(agent, action_data)

        self.target = self.parameters["target"]

    def run(self, sim: Simulation):
        enemy = sim.agents[self.target]
        print(f"{self.agent.fighter.name}の攻撃!")
        norma = max(
            min(80 + (self.agent.agility - enemy.agility) * 5, 90), 10
        )  # 成功率は80%、速度差1につき5%の補正(-70~10)
        dice = random.randint(0, 99)
        hit = dice <= norma
        print(f"(1d100)<={norma}:{dice}={hit}")
        if hit:
            damage = max(
                self.agent.strength - enemy.armor + random.randint(1, 6), 1
            )  # 0以下にはならない
            print(f"damage={self.agent.strength-enemy.armor}+1d6={damage}")
            partner = sim.agents[enemy.fighter.partner]
            if partner.covering:  # パートナーがかばう準備をしているなら
                print(f"{partner.fighter.name}が{enemy.fighter.name}をかばった!")
                partner.hurt(damage)  # 代わりにパートナーが被弾
            else:
                enemy.hurt(damage)
        else:
            print(f"ミス!ダメージを与えられない。")


class MagicAttack(SimAction):  # 攻撃魔法
    def __init__(self, agent: FighterAgent, action_data: fable_saga.Action):
        super().__init__(agent, action_data)

        self.target = self.parameters["target"]

    def run(self, sim: Simulation):
        enemy = sim.agents[self.target]
        print(f"{self.agent.fighter.name}は攻撃魔法を唱えた!")
        if self.agent.mp < 1:
            print("しかしMPが足りない!")
            return
        damage = max(
            self.agent.magic - enemy.magic + random.randint(1, 6), 1
        )  # 0以下にはならない
        print(f"damage={self.agent.magic-enemy.magic}+1d6={damage}")
        partner = sim.agents[enemy.fighter.partner]
        if partner.covering:  # パートナーがかばう準備をしているなら
            print(f"{partner.fighter.name}が{enemy.fighter.name}をかばった!")
            partner.hurt(damage)  # 代わりにパートナーが被弾
        else:
            enemy.hurt(damage)
            self.agent.mp -= 1


class Heal(SimAction):  # 回復魔法
    def __init__(self, agent: FighterAgent, action_data: fable_saga.Action):
        super().__init__(agent, action_data)
        if "target" in self.parameters:
            self.target = self.parameters["target"]
        else:
            self.target = self.agent.guid

    def run(self, sim: Simulation):
        partner = sim.agents[self.target]
        print(f"{self.agent.fighter.name}は回復魔法を唱えた!")
        if self.agent.mp < 2:
            print("しかしMPが足りない!")
            return
        heal_amount = max(
            round(self.agent.magic / 2) + random.randint(1, 6), 1
        )  # MAG依存
        print(f"heal_amount={self.agent.magic/2}+1d6={heal_amount}")
        partner.heal(heal_amount)
        self.agent.mp -= 2


class Cover(SimAction):  # かばう
    def __init__(self, agent: FighterAgent, action_data: fable_saga.Action):
        super().__init__(agent, action_data)

    def run(self, sim: Simulation):
        print(
            f"{self.agent.fighter.name}は{sim.agents[self.agent.fighter.partner].fighter.name}をかばう準備をしている…"
        )
        self.agent.covering = True

このファイルには攻撃や魔法のようなアクションの実際の処理を実装しています。
ダメージ量や回復量の計算や的中判定を行い、HPの増減処理を呼び出したり、エージェントのデータを書き換えます。

↑目次に戻る

🔗あとがき

どうでしたか?今回はSAGAを使ってAI同士が戦う闘技場を作りました。
まだまだアイテムなどの拡張の余地はありそうですし、いつかまたこのテーマで記事を書いてみたいですね。
それでは、また次回の記事でお会いしましょう。
↑目次に戻る

🔗参考記事

↑目次に戻る


株式会社ECNはPHP、JavaScriptを中心にお客様のご要望に合わせたwebサービス、システム開発を承っております。
ビジネスの最初から最後までサポートを行い
お客様のイメージに合わせたWebサービス、システム開発、デザインを行います。

関連記事

python:pygameで生態系[第一章]植物編


CONTACT

TechRachoでは、パートナーシップをご検討いただける方からの
ご連絡をお待ちしております。ぜひお気軽にご意見・ご相談ください。