论文《信息不完整的游戏中的遗憾最小化》介绍了反事实的遗憾,以及如何利用通过自我游戏最大限度地减少反事实遗憾来达到纳什平衡。该算法称为反事实遗憾最小化(CFR)。
论文《在广泛游戏中尽量减少遗憾的蒙特卡洛抽样》介绍了蒙特卡洛反事实遗憾最小化(MCCFR),我们在其中从游戏树中进行采样并估计遗憾。
我们试图让我们的Python实现像教程一样易于理解。我们在一个名为 Kuhn poker 的非常简单的不完美信息游戏上运行它。
我们通过机会抽样(CS)实施了蒙特卡洛反事实遗憾最小化(MCCFR)。它通过尝试所有玩家动作来反复探索游戏树的一部分,但采样机会事件。机会事件就像发牌一样;每次迭代都会对它们进行一次采样。然后,它会计算每个动作遵循当前策略而不是采取该操作的遗憾。然后,它会根据这些遗憾更新策略,使用遗憾匹配进行下一次迭代。最后,它计算整个迭代期间策略的平均值,如果我们进行了足够的迭代,则该平均值与纳什均衡非常接近。
我们将首先介绍数学符号和理论。
玩家用表示,其中是玩家集合。
历史是包括偶然事件在内的一系列动作,是所有历史的集合。
是一组终端历史记录(游戏结束)。
操作,其中是非终端历史记录。
为玩家设置的信息与历史记录类似,但仅包含玩家可见的动作。也就是说,历史记录将包含动作/事件,例如发给对方玩家的牌,而不会有。
被称为玩家的信息分区。
是属于给定信息集的所有历史记录的集合;也就是说,所有这些历史在玩家眼中看起来都是一样的。
玩家的策略,是对动作的分布,其中是玩家所有策略的集合。第-th 次迭代的策略用表示。
策略被定义为针对给定信息集采取行动的概率,
是策略配置文件,由所有玩家的策略组成
是所有玩家的策略
是使用策略配置文件进入历史记录的概率。是在没有玩家贡献的情况下到达的概率;即玩家采取了要遵循的动作概率为。
是仅通过玩家的贡献达到目标的概率。也就是说,
达到信息集的概率是
终端实用程序是玩家对终端历史记录的实用程序(或回报)。
在哪里
是具有策略配置文件的玩家的预期效用(回报)。
纳什均衡是一种状态,在这种状态下,任何参与者都无法仅通过改变策略来增加预期的效用(或回报)。
对于两个参与者来说,纳什均衡是一种策略配置文件,其中
-纳什均衡是,
遗憾的是玩家因为没有遵循最佳策略或采取最佳行动而没有获得的效用(或回报)。
玩家的平均总体遗憾是在所有回合迭代中都没有遵循最佳策略的平均遗憾。
迭代中所有玩家的策略配置文件在哪里,以及
是将玩家的策略替换为的策略配置文件。
平均策略是所有回合中遵循的策略的平均值
这是没有使用最佳策略的卑鄙遗憾。
如果对所有玩家来说都是-Nash 均衡。
因为这是一场零和博弈,所以我们可以加和,第二个术语将被取消。
一组策略的公用事业平均值等于平均策略的效用。
因此,
从定义来看,
那么,
这是纳什均衡。同样,你可以为超过 2 名玩家的游戏进行证明。
因此,我们需要最小化才能接近纳什均衡。
玩家尝试到达(采取了行动),则反事实值是玩家的预期效用导致的概率为)。
wh ere 是可从中访问的终端历史记录集,并且是 up 的前缀。是从到达 z 的概率。
直接的反事实遗憾是,
在哪里
其中是修改为始终在信息集中采取行动的策略配置文件。
本文证明(定理3),
在哪里
该策略是使用后悔匹配计算得出的。
对每个信息集和行动对的遗憾仍然存在,
策略是用遗憾匹配来计算的,
在哪里
这篇论文《信息不完整的游戏中的遗憾最小化》论文证明,如果根据上述方程式选择策略,则与,因此达到-纳什均衡。
计算需要在每次迭代时扩展完整的游戏树。
论文《在广泛游戏中尽量减少遗憾的蒙特卡洛抽样》表明,我们可以从游戏树中进行采样并估计遗憾。
是 () 的一组子集,我们在迭代中只查看单个块。所有子集的联合 spans ()。是选取区块的概率。
是当前迭代中选取的概率;即-where 的总和。
然后我们得到区块的反事实值采样,
这篇论文表明
用一个简单的证据。
因此,我们可以对游戏树的一部分进行采样并计算遗憾。我们计算遗憾的估计值
然后用它来更新和计算每次迭代的策略。最后,我们计算整体平均策略。
这里有一个库恩扑克实现,可以在库恩扑克上试用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)
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
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()
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):
Walk tree 并更新每位玩家的遗憾
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)