論文「情報が不完全なゲームにおける後悔の最小化」では、反事実に基づく後悔と、セルフプレイを通じて反事実に基づく後悔を最小限に抑えることでナッシュ均衡を実現する方法を紹介しています。
このアルゴリズムは、反事実に基づく後悔最小化 (CFR) と呼ばれています。論文「大規模ゲームにおける後悔最小化のためのモンテカルロサンプリング」では、ゲームツリーからサンプリングして後悔を推定するモンテカルロ反事実的後悔最小化(MCCFR)を紹介しています。
Pythonの実装はチュートリアルのようにわかりやすいものにしようとしました。私たちはそれをクーンポーカーという非常に単純で不完全な情報ゲームで運営しています
。モンテカルロ反事実後悔最小化(MCCFR)とチャンスサンプリング(CS)を実装しています。プレイヤーのすべてのアクションを試しながら、偶然の出来事をサンプリングすることで、ゲームツリーの一部を探索する反復処理を行います。チャンスイベントはディーリングカードのようなもので、イテレーションごとに1回サンプリングされます。次に、各アクションについて、そのアクションを実行する代わりに現在の戦略に従った場合の後悔を計算します。次に、後悔マッチングを使用して、これらの後悔に基づいて戦略を更新し、次のイテレーションに備えます。最後に、反復全体の戦略の平均を計算します。これは、十分な反復を実行した場合、ナッシュ均衡に非常に近い値になります
。最初に数学的表記法と理論を紹介します。
プレーヤーはで表されます。ここで、はプレーヤーの集合です。
履歴は偶然の出来事を含む一連の行動であり、すべての履歴の集合です。
端末履歴 (ゲームオーバー) のセットです。
アクション。は端末以外の履歴です。
プレーヤーに設定される情報は履歴に似ていますが、プレーヤーに表示されるアクションのみが含まれます。つまり、履歴には相手プレイヤーに配られたカードなどのアクション/イベントが含まれますが、
それらはありません。プレイヤーの情報パーティションとして知られています。
は、特定の情報セットに属するすべての履歴の集合です。つまり、これらの履歴はすべてプレイヤーの目には同じように見えます。
プレイヤーのストラテジーとは、アクションを分散させたものです。ここで、はプレイヤーのすべてのストラテジーのセットです。-回目の反復のストラテジーはで表されます
。ストラテジーとは、与えられた情報に対して何らかのアクションを起こす確率として定義されます。
すべてのプレイヤーの戦略で構成されるストラテジープロファイルです
を除くすべてのプレイヤーの戦略です
はストラテジープロファイルでヒストリーに到達する確率です。は、プレイヤーの貢献なしにリーチする確率です。つまり、プレイヤーが次のアクションを実行した確率です。
プレイヤーの貢献度だけでリーチする確率です。つまり、
情報セットに到達する確率は、
ターミナルユーティリティは、プレイヤーのターミナル履歴に対する効用(または報酬)です。
どこ
ストラテジープロファイルを持つプレイヤーに期待される効用(ペイオフ)です。
ナッシュ均衡とは、どのプレイヤーも戦略を変えるだけで期待される効用(または見返り)を上げることができない状態のことです。
2人のプレイヤーにとって、ナッシュ均衡は次のような戦略プロファイルです。
-ナッシュ均衡は、
後悔とは、プレイヤーが最適な戦略に従わなかったり、最善の行動をとらなかったために得られなかった効用(または報酬)です。
プレイヤーの全体的な後悔の平均は、すべてのラウンドで最適な戦略に従わなかったことに対する平均的な後悔です。
イテレーション中の全プレイヤーのストラテジープロファイルと
プレイヤーのストラテジーがに置き換えられたストラテジープロファイルです。
平均戦略とは、すべてのラウンドで実施した戦略の平均です
それこそが、最適な戦略でプレイしなかったことに対するあからさまな後悔です。
すべてのプレイヤーに当てはまるなら、 -Nash均衡です。
ゼロサムゲームなので、とを加算すると、2 番目のタームがキャンセルされます。
一連の戦略における効用の平均は、平均的な戦略の効用と等しくなります。
したがって、
の定義から、
次に、
これが -ナッシュ均衡です。同じように、2人以上でプレイするゲームでも証明できます。
そのため、最小化してナッシュ均衡に近づける必要があります。
カウンターファクチュアルバリューは、プレイヤーがリーチしようとした場合(確率でそれに繋がるアクションを取った)プレイヤーにとって期待される効用です。
where はからアクセス可能な端末履歴のセットで、は up to のプレフィックスです。 は z に到達する確率です。
即時の反事実的後悔は、
どこ
情報セットで常にアクションを取るように修正したストラテジープロファイルがどこにあるか
論文はそれを証明しています(定理3)、
どこ
ストラテジーはリグレットマッチングを使用して計算されます。
それぞれの情報セット、アクションペアに対する後悔は保たれ、
そして戦略は後悔マッチングで計算されます
どこ
論文「情報が不完全なゲームにおける後悔の最小化」という論文は、上記の方程式に従って戦略を選択した場合、ナッシュ均衡に比例して小さくなり、したがってナッシュ均衡に達することを証明しています。
コンピューティングでは、イテレーションのたびにゲームツリー全体を拡張する必要があります。
大規模ゲームにおける後悔最小化のためのモンテカルロサンプリングという論文は、ゲームツリーからサンプリングして後悔を推定できることを示しています。
は () のサブセットの集合で、1 回の反復で 1 つのブロックしか見ることができません。すべてのサブセットのスパン () の和集合です。ブロックを選ぶ確率です。
は現在のイテレーションでピッキングする確率、つまり -の合計です。
次に、ブロックから反事実値をサンプリングし、
論文はそれを示しています
簡単な証拠で。
そのため、ゲームツリーの一部をサンプリングして後悔を計算することができます。後悔の推定値を計算します
そして、それを使って各イテレーションのストラテジーを更新し、計算します。最後に、全体の平均戦略を計算します。
ここでは、クーンポーカーで CFR を試すためのクーンポーカーの実装を紹介します。
さっそくコードを見ていきましょう!
328from typing import NewType, Dict, List, Callable, cast
329
330from labml import monit, tracker, logger, experiment
331from labml.configs import BaseConfigs, option
プレイヤーの集合がどこにいるか
334Player = NewType('Player', int)
336Action = NewType('Action', str)
339class History:
端末の履歴、つまりゲームオーバーかどうか。
351 def is_terminal(self):
356 raise NotImplementedError()
358 def terminal_utility(self, i: Player) -> float:
364 raise NotImplementedError()
現在のプレーヤーを取得する (という文字で表される)。これをプレーヤー関数と呼びます。
現在の出来事が偶然の出来事であることを意味する場合。カードを配ったり、ポーカーで一般的なカードを開いたりするようなものです。
366 def player(self) -> Player:
373 raise NotImplementedError()
次のステップがチャンスステップ、つまり新しいカードを配るようなものかどうか。
375 def is_chance(self) -> bool:
380 raise NotImplementedError()
次の機会にぜひお試しください.
382 def sample_chance(self) -> Action:
386 raise NotImplementedError()
履歴にアクションを追加します。
388 def __add__(self, action: Action):
392 raise NotImplementedError()
394 def info_set_key(self) -> str:
398 raise NotImplementedError
400 def new_info_set(self) -> 'InfoSet':
404 raise NotImplementedError()
人間が読める表現
406 def __repr__(self):
410 raise NotImplementedError()
413class InfoSet:
情報セットを識別するユニークなキー
421 key: str
438 regret: Dict[Action, float]
445 cumulative_strategy: Dict[Action, float]
[初期化]
447 def __init__(self, key: str):
451 self.key = key
452 self.regret = {a: 0 for a in self.actions()}
453 self.cumulative_strategy = {a: 0 for a in self.actions()}
454 self.calculate_strategy()
アクション
456 def actions(self) -> List[Action]:
460 raise NotImplementedError()
保存したディクショナリから情報セットを読み込む
462 @staticmethod
463 def from_dict(data: Dict[str, any]) -> 'InfoSet':
467 raise NotImplementedError()
情報セットをディクショナリに保存する
469 def to_dict(self):
473 return {
474 'key': self.key,
475 'regret': self.regret,
476 'average_strategy': self.cumulative_strategy,
477 }
保存したディクショナリからデータを読み込む
479 def load_dict(self, data: Dict[str, any]):
483 self.regret = data['regret']
484 self.cumulative_strategy = data['average_strategy']
485 self.calculate_strategy()
487 def calculate_strategy(self):
506 regret = {a: max(r, 0) for a, r in self.regret.items()}
508 regret_sum = sum(regret.values())
もし、
510 if regret_sum > 0:
513 self.strategy = {a: r / regret_sum for a, r in regret.items()}
それ以外の場合は、
515 else:
517 count = len(list(a for a in self.regret))
520 self.strategy = {a: 1 / count for a, r in regret.items()}
522 def get_average_strategy(self):
531 cum_strategy = {a: self.cumulative_strategy.get(a, 0.) for a in self.actions()}
535 strategy_sum = sum(cum_strategy.values())
もし、
537 if strategy_sum > 0:
541 return {a: s / strategy_sum for a, s in cum_strategy.items()}
それ以外の場合は、
543 else:
545 count = len(list(a for a in cum_strategy))
548 return {a: 1 / count for a, r in cum_strategy.items()}
人間が読める表現
550 def __repr__(self):
554 raise NotImplementedError()
すべてのチャンスイベント(ノード)をサンプリングし、他のすべてのイベント(ノード)を調査するチャンスサンプリング(CS)を行います。
この用語は無視できます。なぜなら、チャンスサンプリングを行っていて、(分子と分母で共通)戦略を計算するときに相殺されてしまうので、すべての端末履歴で同じだからです。
557class CFR:
すべての情報セットのセット。
570 info_sets: Dict[str, InfoSet]
create_new_history
新しい空の履歴を作成しますepochs
はトレーニングする反復回数です n_players
はプレイヤーの数572 def __init__(self, *,
573 create_new_history: Callable[[], History],
574 epochs: int,
575 n_players: int = 2):
581 self.n_players = n_players
582 self.epochs = epochs
583 self.create_new_history = create_new_history
すべての情報セットのセット用の辞書
585 self.info_sets = {}
分析用トラッカー
587 self.tracker = InfoSetTracker()
特定の履歴における現在のプレーヤーの情報セットを返します
589 def _get_info_set(self, h: History):
593 info_set_key = h.info_set_key()
594 if info_set_key not in self.info_sets:
595 self.info_sets[info_set_key] = h.new_info_set()
596 return self.info_sets[info_set_key]
この関数はゲームツリーを歩き回ります。
h
現在の履歴です i
後悔の気持ちを込めて計算している選手はpi_i
は pi_neg_i
は この関数は、プレフィックス付きの端末履歴の集合である履歴に対して、期待どおりのユーティリティを返します。
ティーを歩いていると、後悔の気持ちが全部アップデートされます。
598 def walk_tree(self, h: History, i: Player, pi_i: float, pi_neg_i: float) -> float:
端末履歴の場合は、端末ユーティリティを返します。
619 if h.is_terminal():
620 return h.terminal_utility(i)
それが偶然の出来事なら、 aをサンプルして次のステップに進んでください。
622 elif h.is_chance():
623 a = h.sample_chance()
624 return self.walk_tree(h + a, i, pi_i, pi_neg_i)
現在のプレイヤーの情報設定を取得
627 I = self._get_info_set(h)
保存する
629 v = 0
アクションごとに保存するには
633 va = {}
すべてのアクションを反復処理
636 for a in I.actions():
現在のプレーヤーが
638 if i == h.player():
643 va[a] = self.walk_tree(h + a, i, pi_i * I.strategy[a], pi_neg_i)
それ以外の場合は、
645 else:
650 va[a] = self.walk_tree(h + a, i, pi_i, pi_neg_i * I.strategy[a])
655 v = v + I.strategy[a] * va[a]
現在のプレイヤーがそうなら、累積攻略法と総後悔を更新
659 if h.player() == i:
累積戦略の更新
664 for a in I.actions():
665 I.cumulative_strategy[a] = I.cumulative_strategy[a] + pi_i * I.strategy[a]
678 for a in I.actions():
679 I.regret[a] += pi_neg_i * (va[a] - v)
ストラテジーを更新
682 I.calculate_strategy()
プレイヤーに期待通りのユーティリティを返し、
686 return v
688 def iterate(self):
epochs
ループ・フォー・タイムズ
696 for t in monit.iterate('Train', self.epochs):
ツリーを歩き、各プレイヤーの悔い改めを更新
698 for i in range(self.n_players):
699 self.walk_tree(self.create_new_history(), cast(Player, i), 1, 1)
分析用トラッキングデータ
702 tracker.add_global_step()
703 self.tracker(self.info_sets)
704 tracker.save()
反復ごとにチェックポイントを保存
707 if (t + 1) % 1_000 == 0:
708 experiment.save_checkpoint()
情報セットを印刷
711 logger.inspect(self.info_sets)
714class InfoSetTracker:
トラッキングインジケーターを設定
720 def __init__(self):
724 tracker.set_histogram(f'strategy.*')
725 tracker.set_histogram(f'average_strategy.*')
726 tracker.set_histogram(f'regret.*')
すべての情報セットのデータを追跡
728 def __call__(self, info_sets: Dict[str, InfoSet]):
732 for I in info_sets.values():
733 avg_strategy = I.get_average_strategy()
734 for a in I.actions():
735 tracker.add({
736 f'strategy.{I.key}.{a}': I.strategy[a],
737 f'average_strategy.{I.key}.{a}': avg_strategy[a],
738 f'regret.{I.key}.{a}': I.regret[a],
739 })
742class CFRConfigs(BaseConfigs):
746 create_new_history: Callable[[], History]
747 epochs: int = 1_00_000
748 cfr: CFR = 'simple_cfr'
CFR アルゴリズムを初期化
751@option(CFRConfigs.cfr)
752def simple_cfr(c: CFRConfigs):
756 return CFR(create_new_history=c.create_new_history,
757 epochs=c.epochs)