June 23, 2026

03. 缓存与对象存储:Redis、S3、向量库

读完这篇你能:

• 用 Redis 给 LLM 接口加缓存,重复请求秒回
• 用 MinIO/S3 存用户上传的文件,生成预签名 URL 给前端直传
• 用 pgvector 做向量检索,搭一个最小 RAG
• 知道缓存穿透、击穿、雪崩是什么,怎么防

对应架构层:数据层(扩展)

前置知识:01. Web 基础、02. 数据库

预计用时:3 小时(含动手)

---

0. 为什么算法工程师要学这个

你的 LLM 应用上线了,用户问"FastAPI 是什么",你调一次 OpenAI 花 2 秒 + $0.001。

然后 1000 个用户问了同样的问题——你调了 1000 次 OpenAI,花 2000 秒 + $1。

但你完全可以只调第一次,后面 999 次直接返回缓存的结果。这就是 Redis 缓存。

另一个场景:用户上传一个 50MB 的 PDF,你要:

• 存哪里?(不能放 Postgres——数据库不该存文件)
• 怎么让前端能下载?
• 怎么避免重复上传?

答案是对象存储(S3 / OSS)。

第三个场景:你做 RAG,用户的 query 要先转向量,然后在文档库里找最相似的 10 段——Postgres 装个 pgvector 扩展就能干。

这一篇讲数据层的三个关键扩展:

技术	解决什么问题	类比
Redis	热数据加速、重复计算省成本	GPU 显存 vs 磁盘
S3 / MinIO	海量文件存储	网盘 / OSS
pgvector	向量相似度检索	FAISS,但是在数据库里

---

1. 核心概念(最少必要理论)

1.1 缓存是什么

缓存(Cache) = 把昂贵计算的结果存起来,下次同样请求直接返回。

LLM 应用的缓存场景:

场景	命中率	节省
完全相同的问题(Exact Match)	5-20%	90% 成本 + 99% 延迟
语义相同的问题(Semantic Cache)	20-40%	70% 成本 + 90% 延迟
Embedding 结果	高	90% 延迟(Embedding API 慢)
用户 session	高	90% 数据库查询

缓存的代价:

• 数据不一致:数据库改了,缓存还是旧的
• 内存成本:Redis 比 Postgres 贵 10 倍(内存 vs 磁盘)
• 复杂度:多了缓存层,代码和运维都复杂

经验:不要过早优化。先上线,看哪慢,再加缓存。

1.2 Redis 基础

Redis 是内存数据库,所有数据放内存,所以极快(单实例 10 万 QPS)。

数据结构(5 种最常用):

类型	用途	例子
String	缓存 JSON、计数器	SET token_count_123 500
List	队列、最新 N 条	LPUSH recent_queries "..."
Hash	对象字段	HSET user:123 name "sanbu" credits 100
Set	去重、标签	SADD online_users 123 456
ZSet(Sorted Set)	排行榜、Top N	ZADD leaderboard 100 player1

关键特性:

• TTL(Time To Live):每个 key 可以设过期时间,到期自动删除
• 原子操作:INCR 是原子的,高并发计数不冲突
• 持久化(可选):RDB / AOF,断电数据不丢(但有性能损失)

1.3 缓存模式(Cache Patterns)

Cache-Aside(旁路缓存,最常用):

1. 应用查 Redis → 命中?返回
2. 没命中 → 查数据库
3. 把结果写入 Redis(带 TTL)
4. 返回

特点:简单,但写数据时要主动删缓存(不然脏数据)。

Write-Through(写穿透):

应用写 → 同时写 DB 和缓存

特点:强一致,但写慢。

Write-Behind(写回):

应用写 → 只写缓存 → 异步刷到 DB

特点:写极快,但缓存挂了数据可能丢(高风险)。

本系列用 Cache-Aside,适合大多数场景。

1.4 缓存三大问题

问题 1:缓存穿透(Cache Penetration)

症状:大量请求查一个不存在的 key(比如 user_id=-1),每次都打到数据库。

原因:缓存只存"查到的结果",查不到的不缓存。

解决:

• 缓存空结果(SET user:-1 "" EX 60),TTL 短一点(60 秒)
• 用 布隆过滤器(Bloom Filter)预过滤不存在的 key

问题 2:缓存击穿(Cache Breakdown)

症状:某个热 key 突然过期,瞬间 1000 个请求同时打到数据库。

解决:

• 热 key 不设过期时间,主动更新
• 加互斥锁:第一个请求查 DB,其他请求等

# 简化版互斥锁
def get_with_lock(key):
    val = redis.get(key)
    if val:
        return val
    lock = redis.set(f"lock:{key}", "1", nx=True, ex=10)
    if lock:
        val = db.query(key)
        redis.set(key, val, ex=3600)
        redis.delete(f"lock:{key}")
        return val
    else:
        time.sleep(0.1)
        return get_with_lock(key)  # 重试

问题 3:缓存雪崩(Cache Avalanche)

症状:大量 key 同时过期,全部请求打 DB。

解决:

• TTL 加随机扰动(3600 + random(0, 300))
• 多级缓存(本地缓存 + Redis)
• 限流降级

1.5 对象存储(S3 / OSS)

为什么不用文件系统?

• 文件系统难扩展(单机容量上限)
• 没法多机器访问
• 没有版本控制、生命周期管理

对象存储特点:

• 海量(EB 级)、便宜、高可用
• HTTP API(PUT /bucket/key 上传,GET /bucket/key 下载)
• 不像文件系统有目录,只有 bucket + key(key 可以带 / 模拟目录)

主流选择:

服务	备注
AWS S3	行业标准,几乎所有 SDK 都兼容
Cloudflare R2	兼容 S3 API,零出口流量费,便宜
阿里云 OSS、腾讯云 COS	国内厂商,兼容 S3
MinIO	自建 S3,本地开发用

关键操作:

• 上传(PUT)
• 下载(GET)
• 预签名 URL(Presigned URL):让前端拿临时 token 直接上传/下载,不用走你的服务器(省带宽)

1.6 向量数据库

为什么 Postgres 不够用?

Postgres 擅长精确匹配(WHERE id = 123)。但 LLM 场景要的是相似度——"找跟这段文本语义最接近的 10 段"。

向量化后,每段文本是一个 768/1536 维浮点向量。相似度 = 余弦距离或 L2 距离。

暴力检索:O(N),100 万段文本要算 100 万次距离。

近似最近邻(ANN):用 HNSW、IVF 等算法,牺牲一点精度换 100-1000 倍速度。

主流向量库:

技术	特点	什么时候用
pgvector	Postgres 扩展	已有 Postgres,中小规模(千万向量级)
Qdrant	Rust 写的,快	中大规模,独立向量服务
Milvus	国产,分布式	大规模(亿级向量)
Chroma	Python 友好	原型、小项目
Pinecone	托管 SaaS	不想运维

本系列用 pgvector,因为:已有 Postgres,不用多一个服务,够用。

---

2. 实战:给应用加缓存 + 文件 + 向量

2.1 起 Redis 和 MinIO

升级 docker-compose.yml:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: myuser
      POSTGRES_PASSWORD: mypassword
      POSTGRES_DB: myapp
    ports: ["5432:5432"]
    volumes: [pgdata:/var/lib/postgresql/data]

  redis:
    image: redis:7
    ports: ["6379:6379"]
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

  minio:
    image: minio/minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ":9001"
    volumes: [miniodata:/data]

volumes:
  pgdata:
  miniodata:

启动:

docker compose up -d
docker compose ps

• Redis:localhost:6379
• MinIO API:localhost:9000
• MinIO 控制台:http://localhost:9001,账号 minioadmin / 密码 minioadmin

2.2 给 Postgres 装 pgvector

docker compose exec postgres psql -U myuser -d myapp -c "CREATE EXTENSION IF NOT EXISTS vector;"

2.3 Redis 缓存实现

装包:

pip install redis boto3

新建 cache.py:

import json
import redis
from typing import Optional, Any

redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

def cache_get(key: str) -> Optional[Any]:
    val = redis_client.get(key)
    if val is None:
        return None
    return json.loads(val)

def cache_set(key: str, value: Any, ttl: int = 3600):
    redis_client.setex(key, ttl, json.dumps(value, default=str))

def make_cache_key(*args) -> str:
    """根据参数生成稳定的缓存 key"""
    import hashlib
    raw = "|".join(str(a) for a in args)
    return hashlib.sha256(raw.encode()).hexdigest()

改 summarize 接口加缓存:

from cache import cache_get, cache_set, make_cache_key

@app.post("/api/summarize", response_model=SummarizeResponse)
async def summarize(req: SummarizeRequest, session: Session = Depends(get_session)):
    # 1. 查缓存
    cache_key = make_cache_key("summarize", req.text, req.max_length)
    cached = cache_get(cache_key)
    if cached:
        # 命中缓存,记一条数据库记录(用于统计),不调 LLM
        return SummarizeResponse(**cached)

    # 2. 没命中,调 LLM
    resp = await client.chat.completions.create(...)
    result = SummarizeResponse(
        summary=resp.choices[0].message.content,
        tokens_used=resp.usage.total_tokens,
    )

    # 3. 写缓存(TTL 1 小时)
    cache_set(cache_key, result.model_dump(), ttl=3600)

    # 4. 写数据库记录
    session.add(Query(...))
    session.commit()

    return result

测试:

# 第一次调用,1-2 秒,命中缓存标记为 false
curl -X POST .../api/summarize -d '{"text":"FastAPI 是..."}'

# 第二次同样请求,< 10ms,直接从 Redis 返回
curl -X POST .../api/summarize -d '{"text":"FastAPI 是..."}'

进 Redis CLI 看看:

docker compose exec redis redis-cli
> KEYS *
1) "a3b4c5..."   # 你的缓存 key
> TTL a3b4c5...
(integer) 3598   # 还有 3598 秒过期

2.4 S3 / MinIO 文件上传

新建 storage.py:

import boto3
from botocore.client import Config
from datetime import timedelta

# MinIO 兼容 S3 API,用同样的 boto3
s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:9000",
    aws_access_key_id="minioadmin",
    aws_secret_access_key="minioadmin",
    config=Config(signature_version="s3v4"),
    region_name="us-east-1",
)

BUCKET = "myapp-uploads"

def ensure_bucket():
    try:
        s3.create_bucket(Bucket=BUCKET)
    except s3.exceptions.BucketAlreadyOwnedByYou:
        pass

def upload_file(key: str, data: bytes, content_type: str = "application/octet-stream"):
    s3.put_object(Bucket=BUCKET, Key=key, Body=data, ContentType=content_type)

def download_file(key: str) -> bytes:
    return s3.get_object(Bucket=BUCKET, Key=key)["Body"].read()

def presigned_upload_url(key: str, expires: int = 3600) -> str:
    """生成预签名上传 URL,前端直接 PUT 上传,不走后端"""
    return s3.generate_presigned_url(
        "put_object",
        Params={"Bucket": BUCKET, "Key": key},
        ExpiresIn=expires,
    )

def presigned_download_url(key: str, expires: int = 3600) -> str:
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": BUCKET, "Key": key},
        ExpiresIn=expires,
    )

FastAPI 接口:

from fastapi import UploadFile, File

@app.post("/api/files/upload")
async def upload(file: UploadFile = File(...), session: Session = Depends(get_session)):
    data = await file.read()
    key = f"uploads/{datetime.utcnow().strftime('%Y/%m/%d')}/{uuid4()}.{file.filename.split('.')[-1]}"
    upload_file(key, data, file.content_type)
    return {"key": key, "size": len(data)}

@app.get("/api/files/presign")
def presign_upload(filename: str):
    key = f"uploads/{filename}"
    url = presigned_upload_url(key)
    return {"url": url, "key": key}

前端直传流程(伪代码):

// 1. 前端向后端要预签名 URL
const { url, key } = await fetch('/api/files/presign?filename=doc.pdf').then(r => r.json())

// 2. 前端直接 PUT 到 S3,不走后端
await fetch(url, { method: 'PUT', body: file })

// 3. 前端告诉后端:文件在 key 这个位置
await fetch('/api/files/confirm', { method: 'POST', body: JSON.stringify({ key }) })

好处:大文件不走后端,省带宽 + 省服务器 CPU。

2.5 用 pgvector 做最小 RAG

步骤:

1. 文档切块
2. 每块用 embedding API 转向量
3. 向量存 pgvector 表
4. 查询时:query → 向量 → 余弦检索 top-K

加模型:

from pgvector.sqlmodel import Vector

class Document(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    source: str  # 来源(文件名、URL)
    chunk: str   # 文本块
    embedding: List[float] = Field(sa_column=Column(Vector(1536)))  # OpenAI text-embedding-3-small
    created_at: datetime = Field(default_factory=datetime.utcnow)

建索引(IVFFlat,ANN 加速):

CREATE INDEX idx_docs_embedding ON documents
USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

Embedding 函数:

async def get_embedding(text: str) -> list[float]:
    resp = await client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return resp.data[0].embedding

RAG 接口:

@app.post("/api/rag/ask")
async def rag_ask(req: RagRequest, session: Session = Depends(get_session)):
    # 1. query 转 embedding
    q_emb = await get_embedding(req.question)

    # 2. 余弦相似度 top-5
    from sqlalchemy import text as sa_text
    sql = sa_text("""
        SELECT id, source, chunk, embedding <=> :emb AS distance
        FROM documents
        ORDER BY embedding <=> :emb
        LIMIT 5
    """)
    results = session.exec(sql, params={"emb": q_emb}).all()

    # 3. 拼 prompt 喂 LLM
    context = "\n\n".join(r.chunk for r in results)
    resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"根据以下资料回答问题:\n{context}"},
            {"role": "user", "content": req.question},
        ],
    )
    return {"answer": resp.choices[0].message.content, "sources": [r.source for r in results]}

<=> 是 pgvector 的余弦距离运算符。

---

3. 进阶:生产级改造

3.1 缓存防穿透

def cached_lookup(key: str, db_lookup_fn, ttl_hit=3600, ttl_miss=60):
    val = cache_get(key)
    if val is not None:
        return val if val != "__NULL__" else None

    # 查 DB
    result = db_lookup_fn()
    if result is None:
        cache_set(key, "__NULL__", ttl=ttl_miss)  # 缓存空结果,短 TTL
    else:
        cache_set(key, result, ttl=ttl_hit)
    return result

3.2 Semantic Cache(语义缓存)

精确匹配命中率低("FastAPI 是什么" vs "FastAPI 是啥"算两个 key)。

方案:把 query 转向量,在 Redis 里查相似向量(GPT4ALL、RedisVL 等工具)。

async def semantic_cache_lookup(query: str, threshold: float = 0.95):
    q_emb = await get_embedding(query)
    # 用 Redis 的 VECTORSIMILARITY 命令查最相似的 key
    # 详见 Redis Search 模块
    ...

命中率能从 5% 提到 30%+。

3.3 S3 生命周期

把旧文件归档到便宜存储:

aws s3api put-bucket-lifecycle-configuration --bucket mybucket --lifecycle-configuration '{
  "Rules": [
    {
      "ID": "ArchiveOldFiles",
      "Status": "Enabled",
      "Filter": {"Prefix": "uploads/"},
      "Transitions": [{"Days": 30, "StorageClass": "GLACIER"}]
    }
  ]
}'

GLACIER 比 S3 Standard 便宜 80%,但取回要几分钟到几小时——适合冷数据。

3.4 pgvector 性能

• 向量维度:1536 维,每行占 6KB。100 万行约 6GB。
• lists 参数:ivfflat 的 lists 越大,精度越高但建索引慢。经验值:sqrt(N)(N = 行数)。
• HNSW 索引(pgvector 0.5+):比 IVFFlat 快,但占内存大。

CREATE INDEX idx_docs_embedding_hnsw ON documents
USING hnsw (embedding vector_cosine_ops);

3.5 监控缓存命中率

生产环境必须监控:

from prometheus_client import Counter

cache_hits = Counter("cache_hits_total", "Cache hits")
cache_misses = Counter("cache_misses_total", "Cache misses")

# 在 cache_get 里埋点
if val is not None:
    cache_hits.inc()
else:
    cache_misses.inc()

经验:

• 命中率 < 20%:缓存设计有问题,考虑 Semantic Cache
• 命中率 20-60%:正常
• 命中率 > 80%:警惕——可能 TTL 设太长,数据脏了

详见 06. 可观测性。

---

4. 常见踩坑

坑 1:缓存和数据库不一致

# ❌ 先改 DB 后删缓存,删缓存失败 → 脏数据
session.update(user)
redis.delete(key)  # 失败怎么办?

# ✅ 删缓存失败要有重试 + 告警
try:
    redis.delete(key)
except Exception:
    # 入消息队列,异步重试
    enqueue_cache_invalidation(key)

坑 2:大 value 撑爆 Redis

# ❌ 一个 key 存 10MB 的列表
redis.set("big_list", json.dumps(huge_list))

# ✅ 切片或用 Stream
for i, chunk in enumerate(chunks(huge_list, 100)):
    redis.rpush(f"big_list:{i}", *chunk)

经验:单个 value 控制在 10KB 以内,大数据用 Hash 或 Stream 分片。

坑 3:缓存 key 设计不当

# ❌ key 太长(原始文本)→ 内存浪费
key = f"summary:{req.text}"

# ✅ hash 后用摘要
key = f"summary:{sha256(req.text.encode()).hexdigest()}"

坑 4:把 S3 当数据库用

S3 延迟高(几十到几百毫秒),不能做高频查询的存储。元数据存 Postgres,文件存 S3,两表通过 key 关联。

坑 5:pgvector 索引漏建

100 万向量没建索引,查询要几秒。建了 IVFFlat 索引后,< 50ms。

验证:

EXPLAIN ANALYZE SELECT * FROM documents ORDER BY embedding <=> '[...]'::vector LIMIT 5;

看到 "Index Scan using idx_docs_embedding" 才是命中。

---

5. 检查清单

• 我能用 Docker 起 Redis、MinIO,客户端能连
• 我会写 Cache-Aside 模式的代码(查缓存 → 查 DB → 写缓存)
• 我能解释缓存穿透、击穿、雪崩分别是什么,怎么防
• 我会用 boto3 上传 / 下载 S3 文件
• 我会生成预签名 URL,让前端直传
• 我能解释为什么预签名 URL 比走后端中转好
• 我会用 pgvector 建表、插向量、做 top-K 检索
• 我知道 ivfflat 和 hnsw 的区别
• 我会给缓存加 TTL + 随机扰动防雪崩
• 我知道要监控缓存命中率

---

6. 下一步

数据层到这就完整了——Postgres 存业务数据、Redis 缓存、S3 存文件、pgvector 做 RAG。

下一站回到应用层:

• 04. Web 服务开发:JWT 认证、RBAC 权限、流式 SSE、限流

把认证、权限、缓存、流式整合起来,就是一个完整的 LLM 后端了。

进阶学习资源

• Redis 官方教程:https://redis.io/docs/
• Redis Best Practices(Redis Labs):https://redis.io/learn/howtos/solutions
• AWS S3 文档:https://docs.aws.amazon.com/s3/
• pgvector GitHub:https://github.com/pgvector/pgvector
• 《Designing Data-Intensive Applications》第 1、3 章:Martin Kleppmann,讲缓存、存储引擎