把lru_cache直接加在async def上，缓存的不是await之后的结果。普通装饰器看到的是函数调用立即返回的协程对象，而不是协程将来返回的值。相同参数再次命中时，拿到的仍是原来那个协程；它一旦被等待到结束，就不能再次执行。

异步检索、配置读取或模型请求的包装函数，都可能遇到这个错误。下面用不访问网络的完整示例，把三件容易混淆的事拆开：复用协程、复用已完成值，以及多个调用者等待同一个Task。Read in English。

缓存命中发生在函数体运行之前

普通同步函数返回一个元组，lru_cache可以保存这个元组。async def的调用则先创建协程对象，调用这一步并不等于运行函数体。外层lru_cache没有自动插入await，也不会把协程的return值提取出来保存。这是两个返回时刻，不能根据源代码里写着return就把它们视为同一个时刻。

可以从CPython 3.13.0的协程文档核对调用与执行的区别。同版本functools文档也明确将async函数列为不适合直接使用这种缓存的对象。这里引用固定历史版本，是为了让补发文章的资料时间可追溯，不是在介绍一个2026年才出现的新行为。

示例里连续调用bad(‘demo’)两次，尚未await。两次返回对象的身份相同，cache_info().hits已经是1，函数体里的calls却仍为0。这个组合很有诊断价值：命中率增加只能证明装饰器复用了一个返回对象，不能证明一次远程查询已经执行或成功。

第一次await成功，第二次为何失败

第一次await first把协程执行完，得到(‘demo’,42)，calls变为1。接着await second会触发RuntimeError，因为second指向的就是刚刚用完的那个原生协程对象。代码只检查异常类型，不把具体英文报错字符串写成跨版本接口约定。

这与“相同参数会重新创建一个等价协程”不同。缓存命中绕过了原函数调用，也就没有创建新的协程。清空缓存后下次调用可以产生新对象，但“每次使用前清缓存”会放弃跨调用复用，也不能修复已经交给其他调用者的旧引用。

排查时先记录对象类型、是否为同一对象、函数体实际进入次数，再看cache_info。不要打印完整请求参数或返回文档；在检索和模型应用中，它们可能含私人内容。本文使用固定虚构键与元组，足以检验机制，不需要真实请求日志。

只缓存成功完成的值，能解决什么

get_value先查values，未命中则await load，再把result写入字典。返回给调用者的是完成后的元组，不是协程。两次顺序await都成功，底层load只进入一次，而且结果身份相同。使用不可变元组也避免了这个示例里一个调用者修改共享列表、影响另一个调用者的问题。

写入发生在await成功之后，所以这段实现不会把load抛出的异常当作成功结果存入values。若加载失败，下次调用仍会尝试加载。这只是代码的行为，不代表所有业务都应无条件重试；外部服务的限流、错误类型和退避策略需要另外决定。

这个最小实现没有容量限制、TTL、失效管理，也没有并发合并。它的用途是解释缓存位置，不是生产环境的异步LRU库。若结果还受模型版本、租户或权限条件影响，应把影响结果的条件纳入设计，不能仅用问题文本作通用共享键。否则缓存正确运行，也可能返回不适合当前请求的数据。

结果缓存不等于并发请求合并

两个调用者可以在首个结果写入前同时发现未命中。即使事件循环在同一线程运行，await仍为另一个任务提供了进入未完成区间的机会。示例中的race用两个Event固定这个顺序：两个任务都进入加载阶段后，both_entered才放行主任务；主任务确认字典仍没有结果，再设置release允许两者完成。这样能稳定看到entered为2，无需靠sleep碰运气。

所以“完成后只存成功值”和“同一键同时只执行一次”是两项不同契约。前者处理已经完成的数据，后者还需要管理正在进行的工作。给字典换一个名字或增加命中统计，不会凭空产生后者。之前lru_cache线程安全仍可能重复计算讨论的是同步多线程；这里的新增问题是协程对象生命周期与异步未完成状态。

共享Task，失败以后还会重试吗

Task负责执行协程并保存完成状态。在同一个运行中的事件循环内，示例创建一个task，两次await都拿到相同内容，load只多执行一次。它不是让原生协程重新运行，而是在等待同一次任务的完成。本文不创建跨事件循环的全局Task缓存。

失败的Task也有完成状态。failed只执行一次fail，随后两次await都得到ValueError。把Task保存在字典里而从不移除失败项，会让后续调用继续遇到同一次失败，不会自动变成重试。这正是“换成Task就好了”遗漏的决策：成功项留多久，失败项何时清理，由谁发起下一次工作。淘汰正在运行的项时，还应决定只是移除缓存引用，还是终止工作；这两个动作不能混为一谈。

取消还涉及所有权：一个等待者离开时，是否允许它取消共享工作？其他等待者是否仍需要结果？这里不提供未经验证的通用取消封装，也没有演示屏蔽取消。真正实现请求合并前，应把调用者取消、任务失败和服务关闭分别设计；持有Task引用并消费最终结果或异常，不能只把它放入容器就不管。

完整程序与十九项检查

下面只用标准库，实测环境为Python 3.15.0。十九项检查覆盖提前命中但未执行、顺序重用协程报错、完成值复用、并发双加载、Task重复等待和失败状态保留。程序没有访问检索服务或模型，没有测吞吐与成本。所有创建的Task都会被等待，事件也由明确条件释放。

import asyncio
from functools import lru_cache

checks = 0
def check(condition):
    global checks
    assert condition
    checks += 1

async def main():
    calls = 0
    @lru_cache(maxsize=8)
    async def bad(key):
        nonlocal calls
        calls += 1
        return (key, 42)

    first = bad('demo')
    second = bad('demo')
    check(first is second)
    check(calls == 0)
    check(bad.cache_info().hits == 1)
    check(await first == ('demo', 42))
    check(calls == 1)
    try:
        await second
    except RuntimeError:
        check(True)
    else:
        raise AssertionError('expected coroutine reuse failure')
    bad.cache_clear()

    values = {}
    value_calls = 0
    async def load(key):
        nonlocal value_calls
        value_calls += 1
        return (key, 42)
    async def get_value(key):
        if key in values:
            return values[key]
        result = await load(key)
        values[key] = result
        return result

    a = await get_value('demo')
    b = await get_value('demo')
    check(a is b)
    check(value_calls == 1)
    check(values['demo'] == ('demo', 42))

    entered = 0
    both_entered = asyncio.Event()
    release = asyncio.Event()
    concurrent_values = {}
    async def race(key):
        nonlocal entered
        if key in concurrent_values:
            return concurrent_values[key]
        entered += 1
        if entered == 2:
            both_entered.set()
        await release.wait()
        result = (key, 42)
        concurrent_values[key] = result
        return result
    tasks = [asyncio.create_task(race('same')) for _ in range(2)]
    await both_entered.wait()
    check(entered == 2)
    check('same' not in concurrent_values)
    release.set()
    check(await asyncio.gather(*tasks) == [('same', 42)] * 2)

    task = asyncio.create_task(load('shared'))
    check(await task == ('shared', 42))
    check(await task == ('shared', 42))
    check(value_calls == 2)

    failures = 0
    async def fail():
        nonlocal failures
        failures += 1
        raise ValueError('example')
    failed = asyncio.create_task(fail())
    for _ in range(2):
        try:
            await failed
        except ValueError:
            check(True)
        else:
            raise AssertionError('expected retained failure')
    check(failures == 1)
    check(failed.done())
    print(f'{checks} checks passed')

asyncio.run(main())

输出为19 checks passed。要把它改为项目里的回归检查，可以保留相同参数连续等待、并发冷启动和失败后第二次调用这三种场景，再按业务契约补上容量、失效和取消测试。不要只测试第一次调用成功：第一次成功恰恰掩盖了缓存着一次性协程的问题。

判断缓存设计时，先问容器里到底放了什么，再问它能复用多久。协程对象、Task和已完成数据分别代表执行过程、可等待的任务状态与业务结果，不能因为都能出现在异步函数附近就互相替换。

资料依据为CPython 3.13.0的functools文档与asyncio文档；图形、事件控制的对照和检查程序为原创教学示例。文章日期保留北京时间2026年10月25日23:30，实际于2026年10月26日补发。