lru_cacheでasync関数を包むと2回目が落ちる

キャッシュを足したら、2回目だけ落ちた。

今日、LLMの疑似呼び出しで検証した。lru_cacheを付けると、最初のawaitは通り、次で落ちる。ヒットしているのに使えないんだよね。

自分はTaskを共有したい。同じ入力の並列2件を1回にまとめ、失敗後の再実行まで確認した。

lru_cacheが保存するのは何か

async defを呼ぶとコルーチンが返り、awaitで実行結果を受け取る。lru_cacheは待機せず戻り値を保存するので、使い切ったコルーチンが残る。

同梱のLib/functools.pyも確認した。Python版の実装も、関数の戻り値を保存していた。

コルーチンとTask、名前だけだとどうにも混ざる。

Pythonで再現し、Taskを共有する

CPython 3.14.8で実行済み。async_cache_check.pyに保存し、python3 async_cache_check.pyで動く。追加パッケージ不要。外部LLMの料金や性能は測っていない。

import asyncio
from collections import Counter
from functools import lru_cache

async def main():
    calls = Counter()

    async def infer(key):
        calls[key] += 1
        await asyncio.sleep(0)
        if key == "retry" and calls[key] == 1:
            raise ValueError("temporary failure")
        return key.upper()

    bad = lru_cache()(infer)
    print("first:", await bad("hi"))
    try:
        await bad("hi")
    except RuntimeError as e:
        print(type(e).__name__ + ":", e)

    tasks = {}

    async def once(key):
        if key not in tasks:
            tasks[key] = asyncio.create_task(infer(key))
        task = tasks[key]
        try:
            return await task
        except Exception:
            if tasks.get(key) is task:
                del tasks[key]
            raise

    values = await asyncio.gather(once("same"), once("same"))
    assert values == ["SAME", "SAME"] and calls["same"] == 1
    assert await once("same") == "SAME" and calls["same"] == 1
    print("shared:", values, "calls:", calls["same"])
    try:
        await once("retry")
    except ValueError:
        pass
    assert await once("retry") == "RETRY" and calls["retry"] == 2
    print("retry:", "RETRY", "calls:", calls["retry"])

asyncio.run(main())

first: HI
RuntimeError: cannot reuse already awaited coroutine
shared: ['SAME', 'SAME'] calls: 1
retry: RETRY calls: 2

sleep(0)で処理を譲り、完了前にもう1件を入れる。Taskは複数回待てる。完了後の呼び出しも含め、処理回数は1だった。

辞書の確認からTaskの登録までにawaitはない。同じイベントループでは、この途中で別のonceへ処理を譲らず、二重作成を防げる。

失敗したTaskは削除する。残すと次も同じ例外になる。Taskの同一性を確かめてから消し、次の呼び出しで再実行する。

結果だけ保存すれば十分では？

順番に呼ぶなら、await後の値を保存する方法で足りる。並列では両方がキャッシュの空振りを見て処理を始める。実行中もまとめたいときにTaskを共有する。

有限件数で、途中キャンセルをしない単一イベントループのバッチ向け。常駐サービスでは上限やTTL、キャンセル時の扱いが要る。別のasyncio.runへ辞書を持ち越さない。

キーにはモデル名・生成設定・入力を含める。毎回違う回答が欲しい用途では共有を選ばない。

バッチへ入れる前に

並列に流し、実処理の回数を数える。成功後の再利用と、失敗後の再実行も確認する。LLMの一括処理にも入れたい。手元では何件重複しているだろう。