重复问题不花冤枉钱：Redis 缓存怎么加

ScreenShot_2026-08-03_094456_019.png

「从零到 AI 应用工程师」专栏 · 第 6 篇

大模型按 token 计费。同一个问题被问十次，你就付十次——如果每次都老老实实打到模型。

更现实的场景：演示时反复点同一句、前端重试、用户复制粘贴只多打了两个空格……这些都该命中缓存。

今天目标：用 Redis 给「规范化后的相同问题」做答案缓存，并明确三件事：

命中时跳过模型调用；
没命中或 Redis 挂了，主链路还能用；
无论是否命中，历史照样落库。

一、缓存在链路里的位置

POST /chat
  → 校验
  → 查 Redis（规范化后的问题 → Key）
       ├─ 命中：用缓存答案，from_cache=true，仍写历史
       └─ 未命中：调模型 → 写入 Redis（带 TTL）→ 写历史

缓存是加速层，不是真相层。真相在 PostgreSQL 历史表里。

二、Key、TTL、规范化——三件套

1. 输入先规范化

否则 "什么是缓存" 和 "什么是缓存 " 会变成两个 Key，命中率惨不忍睹。

def normalize(message: str) -> str:
    return " ".join(message.split())

2. Key 要有前缀 + 哈希

import hashlib

def cache_key(message: str) -> str:
    digest = hashlib.sha256(normalize(message).encode("utf-8")).hexdigest()
    return f"chat:answer:{digest}"

前缀 chat:answer: 方便以后按业务清理；哈希避免 Key 过长、也避免特殊字符。

3. TTL 分档

类型	建议 TTL	原因
正常答案	30 分钟（1800s）	常见重复提问窗口
空结果 / 临时失败占位	5 分钟（300s）	避免把短期故障「缓存」成长期不可用

CACHE_OK_TTL = 1800
CACHE_FAIL_TTL = 300
EMPTY_MARK = "__EMPTY__"
ERROR_MARK = "__ERROR__"

三、代码：可失败的加速层

1. Redis 读写都要吞掉故障

# clients/cache_client.py
import logging
import redis

logger = logging.getLogger(__name__)
redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)


def get_cache(key: str) -> str | None:
    try:
        return redis_client.get(key)
    except Exception:
        logger.warning("cache read failed", exc_info=True)
        return None  # 降级：当作未命中


def set_cache(key: str, value: str, ttl: int) -> None:
    try:
        redis_client.setex(key, ttl, value)
    except Exception:
        logger.warning("cache write failed", exc_info=True)
        # 写失败不影响主链路

Redis 挂了 → 你的服务变慢变贵一点，但不该直接 500。这是缓存设计的底线。

2. Service 里串起来

# services/chat_service.py（核心逻辑示意）
from clients import llm_client, cache_client
from clients.cache_client import cache_key, CACHE_OK_TTL, CACHE_FAIL_TTL, EMPTY_MARK, ERROR_MARK
from repositories import chat_repository as repo
from core.response import BusinessError

async def reply(db, user_id: str, session_id: str, message: str) -> dict:
    key = cache_key(message)
    cached = cache_client.get_cache(key)

    if cached == EMPTY_MARK:
        # 仍建议记一条失败/空结果历史，按你的产品决定
        raise BusinessError("模型未返回有效内容", status_code=503)
    if cached == ERROR_MARK:
        raise BusinessError("模型服务暂时不可用", status_code=503)

    if cached:
        row = repo.create_processing(db, user_id, session_id, message)
        repo.mark_completed(db, row, cached)
        return {"answer": cached, "from_cache": True, "message_id": row.id}

    row = repo.create_processing(db, user_id, session_id, message)
    try:
        answer = await llm_client.generate(message)
        if not answer or not answer.strip():
            cache_client.set_cache(key, EMPTY_MARK, CACHE_FAIL_TTL)
            repo.mark_failed(db, row)
            raise BusinessError("模型未返回有效内容", status_code=503)

        cache_client.set_cache(key, answer, CACHE_OK_TTL)
        repo.mark_completed(db, row, answer)
        return {"answer": answer, "from_cache": False, "message_id": row.id}
    except BusinessError:
        raise
    except Exception:
        cache_client.set_cache(key, ERROR_MARK, CACHE_FAIL_TTL)
        repo.mark_failed(db, row)
        raise

注意两处：

响应里带 from_cache，方便你验收和统计命中率；
命中缓存也走「写历史」——演示、审计、用户侧会话列表才完整。

四、验收：第二次必须更快、更便宜

1. 第一次：未命中

curl -X POST http://127.0.0.1:8000/chat \
  -H "Authorization: Bearer dev-token" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "cache_demo",
    "session_id": "session_cache_01",
    "message": "什么是缓存失效"
  }'

预期：from_cache: false，耗时接近模型调用。

2. 第二次：故意多空格

curl -X POST http://127.0.0.1:8000/chat \
  -H "Authorization: Bearer dev-token" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "cache_demo",
    "session_id": "session_cache_01",
    "message": "什么是   缓存失效"
  }'

预期：from_cache: true，明显更快。

3. 历史应有两条

curl "http://127.0.0.1:8000/history?user_id=cache_demo&page=1&page_size=10" \
  -H "Authorization: Bearer dev-token"

预期：两条记录，内容相近——说明缓存省的是模型钱，不是历史。

五、进阶提醒（知道即可）

Key 只含问题可能不够。
若答案依赖用户权限、模型版本、系统 Prompt、知识库版本，要把这些维度编进 Key，否则会串答。本专栏演示场景问题共享答案，先保持简单。
并发同时 miss。
同一瞬间多个请求都没打到缓存，会一起打模型。流量大时再上单飞（single-flight）或短锁；早期可先接受。
敏感内容进缓存。
有权限隔离、TTL、删除策略；不要把不该共享的答案做成全局 Key。
失败不要永久缓存。
短 TTL 占位即可，否则一次故障被放大成「半小时都不能用」。

六、带走这三条

好的缓存 = Key 设计 + TTL + 失效策略 + 故障降级，缺一不可。
命中只跳过昂贵计算，不跳过历史与审计。
输入规范化既是数据质量，也是命中率工程。

下一篇把「本机装 Postgres、本机装 Redis、本机起 uvicorn」收掉：用 Docker Compose 一条命令把应用、数据库、缓存一起拉起来。这是你从「能跑」走向「能交付」的关键一步。

这是专栏第 6 篇。重复问题开始不花冤枉钱了。两天一更，下篇见。