第1章高并发与缓存策略¶

学习目标¶

理解 AI Agent 系统面临的高并发挑战与传统 Web 系统的差异
掌握三种经典缓存模式：Cache-Aside、Write-Through、Write-Behind
学会应对缓存穿透、缓存击穿、缓存雪崩三大经典问题
掌握限流（令牌桶/漏桶）与熔断降级的实现思路
理解 API 网关在 Agent 系统中的职责与选型
能够用 Python + Redis 实现一个可上生产的缓存层

问题导入¶

你做了一个很棒的智能客服 Agent，内部测得单次响应平均 1.2 秒。上线第一天，运营推了一波活动，瞬时涌入 5000 QPS。结果：LLM 上游 API 被打满限流，数据库连接池耗尽，用户等了 30 秒只看到一个转圈动画，最后 502。

这不是个例。AI Agent 系统的高并发有三大特殊性：其一，单次请求延迟高（秒级起步），传统 Web 的"加机器就行"思路在 LLM 调用面前效果有限；其二，成本与流量正相关，每一次缓存未命中都意味着真金白银的 Token 消耗；其三，上游 LLM API 本身有限流，你扩容再多，OpenAI/Anthropic 的 RPM 上限照样卡脖子。

缓存、限流、降级，是这类系统的"三板斧"。本章围绕这三件套，讲清楚怎么用、用在哪儿、避开哪些坑。

核心概念¶

1. 高并发的特殊性¶

传统高并发关注吞吐量与延迟的权衡（如 Little's Law：L = λW）。Agent 系统除此之外，还要面对：

长尾延迟：LLM 响应 P99 可能是 P50 的 5-10 倍，长尾会拖垮整体体验。
成本约束：每请求都计费，缓存命中率直接等于省钱比例。
上游不可控：LLM 提供商的限流、抖动、故障，会传导到你的系统。

因此 Agent 系统的并发设计，本质是在"延迟、成本、可用性"三角中做权衡。

2. 缓存模式¶

Cache-Aside（旁路缓存）：最常用。读时先查缓存，未命中再查 DB 并回填；写时先更 DB，再删缓存。优点是简单、缓存与 DB 解耦；缺点是存在短暂不一致窗口。

Write-Through（直写）：写时同时写缓存与 DB（通常由缓存层代理）。强一致，但写延迟高。

Write-Behind（回写）：写时只写缓存，异步刷回 DB。写延迟最低，但有丢数据风险，适合可容忍丢失的场景（如埋点）。

对 Agent 系统而言，Prompt 模板、Embedding 向量、高频问答结果适合 Cache-Aside；用户会话状态可用 Write-Through；非关键审计日志可用 Write-Behind。

3. 三大缓存问题¶

问题	现象	典型成因	应对
穿透	查一个根本不存在的 key，每次都打到 DB	恶意请求、查询不存在的 ID	布隆过滤器、缓存空值
击穿	单个热点 key 过期瞬间，海量请求同时打 DB	热点数据过期	互斥锁（只放一个请求回源）、永不过期+异步刷新
雪崩	大量 key 同时失效	批量预热+相同 TTL	TTL 加随机抖动、多级缓存

4. 限流算法¶

固定窗口：简单但有临界突刺。
滑动窗口：平滑，实现略复杂。
令牌桶：允许突发，按速率补充令牌。适合 API 网关。
漏桶：恒定速率出水，平滑突发。适合下游保护。

5. 熔断与降级¶

熔断器（Circuit Breaker）三态：Closed（正常）→ Open（熔断，快速失败）→ Half-Open（试探恢复）。当 LLM 上游错误率超阈值时熔断，避免雪崩。降级则是熔断后的兜底策略，如返回缓存的历史回复、降级到更便宜的模型、或返回"稍后重试"。

6. API 网关¶

Agent 系统的 API 网关通常承担：鉴权、限流、路由、协议转换、缓存、日志。常见选型：Kong（Lua 插件生态）、APISIX（高性能，etcd 配置）、Envoy（Service Mesh 友好）。对中小团队，FastAPI 内置中间件 + Redis 限流也能扛住中等流量。

原理图¶

flowchart TB
    CLIENT[客户端] --> GW[API 网关<br/>鉴权/限流/路由]
    GW --> RL{令牌桶限流}
    RL -->|通过| CB[熔断器]
    RL -->|拒绝| DEG[降级响应]
    CB -->|Closed/Half-Open| CACHE[(Redis 缓存)]
    CB -->|Open| DEG

    CACHE -->|命中| RESP[返回结果]
    CACHE -->|未命中| LOCK{互斥锁}
    LOCK -->|拿到| SRC[回源:LLM/DB]
    LOCK -->|未拿到| WAIT[等待重试缓存]
    SRC -->|回填| CACHE
    SRC --> RESP

    subgraph 后端
        LLM[LLM API<br/>上游限流]
        DB[(向量库/业务DB)]
    end
    SRC --> LLM
    SRC --> DB

    BF[布隆过滤器] -.->|过滤不存在key| CACHE
    DEG --> RESP

图：Agent 系统高并发架构。请求经网关鉴权限流后进入熔断器，命中缓存直接返回；未命中则通过互斥锁保护回源，回源结果回填缓存。布隆过滤器拦截穿透请求，熔断/降级在异常时兜底。

示例代码¶

下面用 Python + redis-py 实现一个生产可用的 Cache-Aside 层，包含互斥锁防击穿、空值缓存防穿透、TTL 抖动防雪崩。

import json
import time
import random
import hashlib
import redis
from typing import Optional, Callable

class AgentCache:
    """Agent 系统的 Cache-Aside 缓存层。

    特性：
    - 互斥锁防止缓存击穿（同一 key 只放一个回源请求）
    - 空值缓存防止缓存穿透
    - TTL 随机抖动防止缓存雪崩
    """

    # 空值标记，避免与真实空结果混淆
    _NULL_MARKER = "__NULL__"

    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.client = redis.Redis.from_url(redis_url, decode_responses=True)

    def _make_key(self, namespace: str, prompt: str, **tags) -> str:
        """根据 prompt 与标签生成稳定缓存 key，便于复用。"""
        raw = json.dumps({"prompt": prompt, "tags": tags}, sort_keys=True)
        digest = hashlib.sha256(raw.encode()).hexdigest()[:16]
        return f"agent:{namespace}:{digest}"

    def get_or_set(
        self,
        namespace: str,
        prompt: str,
        loader: Callable[[], str],
        ttl: int = 600,
        null_ttl: int = 60,
        **tags,
    ) -> Optional[str]:
        """读取缓存，未命中则通过 loader 回源并回填。

        Args:
            namespace: 业务命名空间，如 'qa'、'summary'
            prompt:   缓存依据的 prompt 文本
            loader:   回源函数（如调用 LLM）
            ttl:      正常缓存存活秒数
            null_ttl: 空值缓存存活秒数（较短，便于及时重试）
        """
        key = self._make_key(namespace, prompt, **tags)
        lock_key = f"lock:{key}"

        # 1. 先查缓存
        cached = self.client.get(key)
        if cached is not None:
            if cached == self._NULL_MARKER:
                return None  # 空值缓存命中，直接返回 None
            return cached

        # 2. 未命中：用 SET NX 抢互斥锁，防击穿
        #    抢到锁的请求回源，其余请求短暂自旋等缓存
        got_lock = self.client.set(lock_key, "1", nx=True, ex=10)
        if got_lock:
            try:
                # 双重检查：可能在抢锁期间别人已回填
                cached = self.client.get(key)
                if cached is not None:
                    return None if cached == self._NULL_MARKER else cached

                # 回源
                value = loader()
                if value is None:
                    # 空值缓存：防穿透
                    self.client.setex(key, null_ttl, self._NULL_MARKER)
                    return None
                # 正常缓存：TTL 加随机抖动，防雪崩
                jitter = random.randint(0, ttl // 10)
                self.client.setex(key, ttl + jitter, value)
                return value
            finally:
                self.client.delete(lock_key)
        else:
            # 未抢到锁：短暂自旋等待缓存写入
            for _ in range(20):
                time.sleep(0.05)
                cached = self.client.get(key)
                if cached is not None:
                    return None if cached == self._NULL_MARKER else cached
            # 等待超时，降级直接回源（少数情况）
            return loader()


# 使用示例：缓存 Agent 问答结果
def call_llm(prompt: str) -> str:
    """模拟昂贵的 LLM 调用。"""
    import os
    from openai import OpenAI
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content


if __name__ == "__main__":
    cache = AgentCache("redis://localhost:6379/0")
    question = "用一句话解释什么是 RAG。"

    # 第二次调用会命中缓存，省一次 LLM 费用
    answer = cache.get_or_set(
        namespace="qa",
        prompt=question,
        loader=lambda: call_llm(question),
        ttl=1800,         # 正常缓存 30 分钟
        null_ttl=60,      # 空值缓存 1 分钟
        model="gpt-4o-mini",
    )
    print(answer)

下面是一个简易令牌桶限流器，可用作 API 中间件：

import time
import redis

class TokenBucket:
    """基于 Redis 的分布式令牌桶限流器。"""

    def __init__(self, redis_url: str = "redis://localhost:6379/1"):
        self.client = redis.Redis.from_url(redis_url, decode_responses=True)

    def acquire(self, key: str, capacity: int, rate: float) -> bool:
        """尝试获取一个令牌。

        Args:
            key:       限流维度（如 user_id 或 api_key）
            capacity:  桶容量（允许突发）
            rate:      令牌补充速率（个/秒）
        Returns:
            True 表示放行，False 表示限流
        """
        bucket_key = f"bucket:{key}"
        now = time.time()
        # 用 Lua 保证"读-改-写"原子性
        lua = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])
        local data = redis.call('HMGET', key, 'tokens', 'ts')
        local tokens = tonumber(data[1]) or capacity
        local ts = tonumber(data[2]) or now
        -- 按经过时间补充令牌
        tokens = math.min(capacity, tokens + (now - ts) * rate)
        if tokens < 1 then
            redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
            redis.call('EXPIRE', key, math.ceil(capacity / rate) + 10)
            return 0
        else
            tokens = tokens - 1
            redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
            redis.call('EXPIRE', key, math.ceil(capacity / rate) + 10)
            return 1
        end
        """
        allowed = self.client.eval(lua, 1, bucket_key, capacity, rate, now)
        return bool(allowed)


# FastAPI 中间件用法
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
bucket = TokenBucket("redis://localhost:6379/1")

@app.middleware("http")
async def rate_limit(request: Request, call_next):
    api_key = request.headers.get("x-api-key", "anonymous")
    # 每个 key 每秒 5 个请求，突发上限 10
    if not bucket.acquire(api_key, capacity=10, rate=5):
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后重试")
    return await call_next(request)

企业案例¶

案例 1：高并发 Agent API 的缓存降级¶

某 SaaS 厂商提供"文档智能问答"API，峰值 3000 QPS。LLM 单次调用 P50 约 2 秒，成本约 ¥0.05/次。团队引入两级缓存：

L1 本地缓存（cachetools TTLCache，5 秒）：防同一热点请求穿透到 Redis。
L2 Redis（30 分钟 TTL + 抖动）：跨实例共享。

上线后缓存命中率 68%，LLM 调用量下降到原来的 1/3，单月省下近 ¥40 万 Token 费用。同时配置令牌桶限流（每租户 50 QPS）与熔断（LLM 错误率 > 30% 触发，返回"服务繁忙，已为您缓存历史回复"）。一次上游 LLM 故障期间，熔断器 30 秒内打开，系统靠缓存与降级撑过故障窗口，而非雪崩。

案例 2：用 API 网关统一治理¶

一家 AI 创业公司用 APISIX 作为统一网关，对接 10+ 个 Agent 服务。网关层完成：JWT 鉴权、按租户限流、灰度路由（新版本灰度 10% 流量）、请求日志打点到 Kafka。后端服务只关注业务，限流与鉴权逻辑下沉到网关，迭代效率显著提升。

常见问题¶

Q1：缓存与数据库不一致怎么办？ Cache-Aside 下，写操作推荐"先更 DB，再删缓存"（而非更新缓存），删除是幂等的，且避免并发写时缓存被旧值覆盖。若要求强一致，可用"双删"：更新后删一次，延迟几百毫秒再删一次。

Q2：热点 key 怎么处理？ 单 key 流量集中时，可：本地缓存兜底、key 分片（如 key:1~key:10 分散到多节点）、永不过期 + 异步刷新。

Q3：LLM 响应是流式的，怎么缓存？ 可缓存完整响应（流式结束后拼接存入），下次命中直接整体返回；或缓存首 token 后的部分。注意：流式场景下，缓存命中判断要在请求开始前完成。

Q4：限流维度怎么选？ 按 API Key（防止单租户滥用）、按用户、按 IP（防爬虫）多层叠加。Agent 系统还要按"模型 + 租户"维度限流，因为不同模型成本不同。

Q5：降级返回什么？ 避免返回错误码让用户困惑。可返回：缓存的历史相似回复、降级到更便宜模型（如 gpt-4o → gpt-4o-mini）、礼貌的"稍后重试"提示并附带任务 ID 供轮询。

面试题¶

1. 缓存穿透、击穿、雪崩的区别与解法？ 答：穿透是查不存在的 key（解法：布隆过滤器/空值缓存）；击穿是热点 key 过期瞬间高并发回源（解法：互斥锁/永不过期）；雪崩是大量 key 同时失效（解法：TTL 随机抖动/多级缓存）。

2. Cache-Aside 为什么"先更 DB 再删缓存"，而不是"先删缓存再更 DB"？ 答：先删缓存后更 DB，在删与更之间若有读请求，会把旧值回填缓存，导致不一致持续到 TTL 过期。先更 DB 再删缓存，即使删缓存失败也可通过重试或 binlog 订阅补偿。

3. 令牌桶与漏桶的区别？ 答：令牌桶按速率补充令牌、允许突发（桶满时一次取多个）；漏桶恒定速率出水、平滑输出。API 网关保护自己用令牌桶（允许突发流量），保护下游用漏桶（平滑到下游的请求）。

4. 熔断器的三态如何流转？ 答：Closed 正常放行并统计错误率；错误率超阈值转 Open，快速失败不调下游；超时后转 Half-Open 放少量试探请求，成功则回 Closed，失败则回 Open。

5. Agent 系统为什么特别需要缓存？ 答：LLM 调用既慢又贵，缓存命中既降延迟又省成本。且 Agent 场景存在大量相似/重复查询（FAQ、模板化提问），缓存命中率天然较高。

练习题¶

在示例代码基础上，给 AgentCache 增加"布隆过滤器"前置过滤，拒绝明显不存在的 key（提示：可用 redisbloom 模块或 pybloom_live）。
把令牌桶限流器改造成"按模型维度限流"：gpt-4o 每秒 2 个，gpt-4o-mini 每秒 10 个。
思考题：如果一个 Agent 的回答会随时间变化（如"今天天气"），缓存策略该如何调整？写出 TTL 与刷新方案的设计。
用 FastAPI 实现一个带熔断器的 Agent 接口：当 LLM 连续 5 次超时则熔断 30 秒，期间返回降级响应。

小结¶

本章围绕 AI Agent 系统的高并发挑战，系统讲解了缓存模式（Cache-Aside/Write-Through/Write-Behind）、三大缓存问题（穿透/击穿/雪崩）及其解法、限流算法（令牌桶/漏桶）与熔断降级，并用 Python + Redis 实现了带互斥锁、空值缓存、TTL 抖动的生产级缓存层。核心要点：Agent 系统的并发设计本质是"延迟-成本-可用性"的权衡，缓存既是性能优化更是成本优化，限流降级则是抵御雪崩的安全阀。下一章我们将把视角转向"耗时任务如何不拖垮主链路"，进入异步任务与消息队列的世界。

第1章 高并发与缓存策略¶