PythonでLFU(Least Frequently Used)キャッシュを実装する方法

この記事では、PythonでLFU(Least Frequently Used:最低頻度使用)キャッシュを実装する方法を解説します。LFUキャッシュは、アクセス頻度が最も低いデータから優先的に追い出す(evictする)キャッシュ方式で、LRU(Least Recently Used)と並んでよく知られるアルゴリズムです。

LFUキャッシュに必要な操作

LFUキャッシュ用のデータ構造には、次の2つの操作が必要です。

• get(key) … キーがキャッシュに存在すればその値を返し、存在しない場合は -1 を返します。
• set(key, value) … キーがまだ存在しない場合に、値を挿入(または更新)します。

キャッシュが最大容量に達したときは、新しい要素を挿入する前に最も使用頻度の低い要素を無効化(削除)しなければなりません。頻度が同じ要素が複数ある場合は、挿入が早い方(FIFO順)を先に削除します。

動作例

容量2のLFUCacheに対して、次の一連の操作を行うとします。

cache.set(1, 1)
cache.set(2, 2)
cache.get(1)
cache.set(3, 3)
cache.get(2)
cache.set(4, 4)
cache.get(1)
cache.get(3)
cache.get(4)

このときの出力は 1, -1, 1, -1, 4 の順になります。これは、キー2とキー3がそれぞれ容量オーバー時に最も使用頻度の低い要素として削除されるためです。

実装の方針

実装は以下の手順で進めます。

1. コンストラクタ(__init__)で容量を受け取る
2. remain … 残り容量を保持
3. least_freq … 現在の最小使用頻度(初期値は1)
4. node_for_freq … 使用頻度ごとに、挿入順を保ったままデータを格納するマップ(defaultdict(OrderedDict))
5. node_for_key … キーから(値, 頻度)へのマップ

_update(key, value) の処理

• node_for_key[key] から現在の頻度 freq を取得する
• node_for_freq[freq] から該当キーの要素を削除する
• node_for_freq[least_freq] のサイズが0になったら、least_freq を +1 する
• node_for_freq[freq+1][key] と node_for_key[key] を、新しい値と頻度(freq+1)で更新する

get(key) の処理

• キーが node_for_key に存在しない場合は -1 を返す
• 値を取得し、_update(key, value) を呼び出して使用頻度を更新する
• 値を返す

set(key, value) の処理

• キーが既に存在する場合は _update(key, value) を呼び出す
• 存在しない場合は:
  • node_for_key[key] = (value, 1)、node_for_freq[1][key] = (value, 1) を設定する
  • remain == 0(容量いっぱい)の場合、node_for_freq[least_freq] からFIFO順に1つ取り除き(popitem(last=False))、対応するキーを node_for_key からも削除する
  • それ以外の場合は remain を -1 する
• 最後に least_freq を 1 に戻す

Pythonコード例

実際の実装は以下の通りです。collections モジュールの defaultdict と OrderedDict を活用することで、各頻度レベル内での挿入順を効率的に管理できます。

from collections import defaultdict, OrderedDict
class LFUCache:
    def __init__(self, capacity):
        self.remain = capacity
        self.least_freq = 1
        self.node_for_freq = defaultdict(OrderedDict)
        self.node_for_key = dict()

    def _update(self, key, value):
        _, freq = self.node_for_key[key]
        self.node_for_freq[freq].pop(key)
        if len(self.node_for_freq[self.least_freq]) == 0:
            self.least_freq += 1
        self.node_for_freq[freq+1][key] = (value, freq+1)
        self.node_for_key[key] = (value, freq+1)

    def get(self, key):
        if key not in self.node_for_key:
            return -1
        value = self.node_for_key[key][0]
        self._update(key, value)
        return value

    def set(self, key, value):
        if key in self.node_for_key:
            self._update(key, value)
        else:
            self.node_for_key[key] = (value, 1)
            self.node_for_freq[1][key] = (value, 1)
            if self.remain == 0:
                removed = self.node_for_freq[self.least_freq].popitem(last=False)
                self.node_for_key.pop(removed[0])
            else:
                self.remain -= 1
            self.least_freq = 1

cache = LFUCache(2)
cache.set(1, 1)
cache.set(2, 2)
print(cache.get(1))
cache.set(3, 3)
print(cache.get(2))
cache.set(4, 4)
print(cache.get(1))
print(cache.get(3))
print(cache.get(4))

入力

cache.set(1, 1)
cache.set(2, 2)
cache.get(1)
cache.set(3, 3)
cache.get(2)
cache.set(4, 4)
cache.get(1)
cache.get(3)
cache.get(4)

出力

1
-1
1
-1
4

まとめ

この実装では、OrderedDict を使って各頻度レベル内の要素を常に挿入順に保持しているため、get・set のどちらの操作も O(1) の計算量で実行できます。頻度の昇格は _update() で行われ、容量超過時には最小頻度グループの先頭要素(FIFO順)が確実に追い出されます。LeetCodeの「LFU Cache」問題などでも応用できる定番パターンなので、ぜひ理解しておきましょう。