前面的文章已经介绍了模型选择、错误处理和成本估算。原型能跑起来之后，新的问题通常是：请求一多，响应变慢，账单也跟着上涨。很多时候，优化并不始于更换模型，而是先减少不必要的调用，再让剩余调用以可控的并发运行。本篇只聚焦一个核心知识点：如何用缓存和并发控制改善 AI 应用的吞吐与成本。示例使用 Python 标准库模拟模型调用，不需要密钥，读者可以直接运行并观察缓存命中和并发限制。

为什么缓存和并发要一起考虑

一次模型调用通常包含网络等待、排队和生成时间。若相同输入被重复提交，例如用户刷新页面、定时任务重复执行，直接再次调用既浪费钱，也增加延迟。缓存可以把“输入相同、允许复用”的结果保存起来，命中时不再访问模型。

但缓存不能解决所有问题。面对大量不同请求时，同时发出太多调用可能触发服务商限流，甚至让本机连接、内存和下游系统过载。因此需要并发上限：允许多个请求同时进行以提高吞吐，但最多只运行固定数量。可以把它理解成一个有容量的服务窗口，而不是无限扩张的线程数。

一个实际的调用层至少要回答四个问题：缓存键是什么、结果能保存多久、最多允许多少并发、失败结果是否缓存。缓存键必须包含会影响结果的因素，例如模型名、系统提示词版本和用户输入；否则不同配置可能错误地复用同一个结果。失败结果通常不应该缓存，否则一次临时故障会被保留成长期错误。

最小可运行示例

下面的代码用 asyncio 模拟一个耗时的模型调用。fake_model_call 代表真实 SDK 的异步请求；实际接入时，只需要替换这个函数，并保留缓存和信号量的边界。程序还记录了模拟成本：缓存命中不产生新的模型调用，未命中则按一次输入和输出 token 估算。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import asyncio
import hashlib
import time

CACHE = {}
CONCURRENCY = asyncio.Semaphore(2)
INPUT_PRICE = 0.20       # 每百万输入 token 的示例价格
OUTPUT_PRICE = 1.20      # 每百万输出 token 的示例价格


def cache_key(model: str, prompt_version: str, question: str) -> str:
    raw = f"{model}\n{prompt_version}\n{question}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


async def fake_model_call(question: str) -> tuple[str, int, int]:
    await asyncio.sleep(0.2)  # 模拟网络和生成延迟
    input_tokens = len(question) * 2
    output_tokens = 20
    return f"回答：{question}", input_tokens, output_tokens


async def ask(question: str) -> tuple[str, bool, float]:
    key = cache_key("demo-model", "prompt-v1", question)
    if key in CACHE:
        return CACHE[key], True, 0.0

    async with CONCURRENCY:
        # 获取信号量后再次检查，避免并发请求重复计算同一个问题
        if key in CACHE:
            return CACHE[key], True, 0.0
        answer, input_tokens, output_tokens = await fake_model_call(question)
        cost = (
            input_tokens / 1_000_000 * INPUT_PRICE
            + output_tokens / 1_000_000 * OUTPUT_PRICE
        )
        CACHE[key] = answer
        return answer, False, cost


async def main() -> None:
    questions = ["什么是缓存？", "如何限制并发？", "什么是缓存？", "如何限制并发？"]
    started = time.perf_counter()
    results = await asyncio.gather(*(ask(q) for q in questions))
    elapsed = time.perf_counter() - started

    for question, (answer, hit, cost) in zip(questions, results):
        state = "命中缓存" if hit else "调用模型"
        print(f"{state} | {question} | {answer} | 成本估算 ${cost:.8f}")
    print(f"总耗时: {elapsed:.3f}s，实际模型调用: {sum(not hit for _, hit, _ in results)}")


if __name__ == "__main__":
    asyncio.run(main())

运行方式：

1
python cache_demo.py

这段程序有三个关键点。第一，cache_key 使用 SHA-256 生成稳定键，但哈希并不会自动让缓存正确：真正重要的是把模型和提示词版本也纳入键。修改系统提示词后，应升级 prompt-v1，避免旧答案污染新逻辑。第二，asyncio.gather 同时提交多个任务，Semaphore(2) 则把实际模型调用限制为最多两个并行任务。第三，信号量内部的第二次缓存检查是必要的：两个相同请求可能同时发现缓存为空，排队后必须再次检查，才能避免重复调用。

从内存缓存走向真实系统

示例中的字典只适合演示。进程重启后数据会消失，多进程部署时每个进程也各自维护一份缓存。生产环境通常会使用带过期时间的缓存服务或数据库，并为每条记录设置 TTL（生存时间）。TTL 太短，命中率低；太长，答案可能过时。适合缓存的往往是确定性较高、重复率高且允许短暂陈旧的任务。

缓存还要控制容量和隐私。不要把完整的敏感原文、访问令牌或用户个人信息无期限保存；可以只保存必要结果，并设置最大容量、过期时间和访问权限。多租户应用必须把租户标识纳入缓存键，否则可能把一个用户的结果返回给另一个用户。对于包含实时数据、权限信息或随机生成要求的请求，应明确禁止缓存，或者让这些因素参与键计算。

并发上限也不能凭感觉设定。先记录请求数、平均延迟、P95 延迟、限流错误和缓存命中率，再逐步调整。并发数增大不一定更快：当服务商限流或网络成为瓶颈时，继续增加只会带来更多重试和更高失败率。实际调用还应结合前文的超时与指数退避，并区分可重试错误和业务错误；不要对所有失败无限重试。

常见问题

缓存命中后还要计费吗？ 如果确实没有发起新的模型请求，通常不会产生这一请求的模型调用费用；但缓存基础设施本身可能有存储和网络成本，计费规则仍应以所用服务商为准。

是不是并发越大越好？ 不是。并发只是把等待时间重叠起来，不能突破服务商配额。应从较小值开始，结合限流、错误率和尾延迟压测。

为什么代码要在信号量内再次检查缓存？ 多个相同请求可能同时通过第一次检查。第一个任务写入结果后，后续任务如果不复查，仍会重复调用模型。这个模式常称为“缓存击穿保护”的最小版本。

什么时候不应该缓存？ 当结果必须实时、与用户权限强相关、包含敏感数据，或每次都要求模型重新随机生成时，不应直接复用旧结果。即使能缓存，也要先完成脱敏、隔离和过期设计。

小结

  • 缓存通过复用相同请求的结果，减少重复调用、延迟和成本。
  • 缓存键必须包含模型、提示词版本以及所有影响结果的输入。
  • 异步并发可以提高吞吐，但信号量能把并发控制在服务商和系统可承受的范围内。
  • 在并发控制区域内再次检查缓存，可以避免相同请求同时穿透缓存。
  • 真实系统还要补充 TTL、容量、隐私隔离、超时、限流监控和成本记录。

到这里，AI 应用的基础工程化路线已经覆盖了成本、并发和性能。下一篇可在已有知识上综合实现一个小项目，但应先明确数据边界、失败策略和可验证指标，再把多个组件组合起来。