AI Agent 真正上线跑起来和 demo 完全是两回事。本地搭个例子感受不到压力等用户量上来、调用量上来最先扛不住的不一定是模型接口而是 Agent 编排服务本身。我这两年一直在做 Agent 落地从 LangChain 起步用 FastAPI 起接口中间踩过无数性能相关的坑。这里面用 Redis 做缓存层是收益最高、见效最快的一步也是最容易被一开始就埋头写 Prompt 的人忽视的一步。这篇文章不聊“缓存是什么”直接聊 AI Agent 和 Redis 怎么组合。我先拆解 Agent 请求链路里真正慢的环节再给出缓存设计的核心思路接着放一套可以直接抄的 FastAPI LangChain 集成 Redis 的实操代码最后把上线后容易踩的坑也一并列出来。1. AI Agent 请求链路里的性能瓶颈到底在哪1.1 一次 Agent 请求背后从用户提问到最终回复先看一次普通 Agent 请求要经历什么。用户发来一句话服务先把历史会话拉出来组装成上下文然后送入大模型。模型不是直接给答案它可能先返回一个工具调用意图服务执行工具查库、调 API把结果再次拼接进 Prompt再调用一次模型甚至循环好几轮才会输出最终结果。这意味着一次用户请求背后可能是 3 到 5 次 LLM 调用外加若干次工具调用。每轮 LLM 调用的延迟是几百毫秒到几秒费用按 token 计费越长的上下文越贵。如果 100 个用户同时提问而你的 Agent 服务没有缓存等于同一个问题可能被重复计算几百次、重复调用几百次。这既是性能问题也是账单问题。我见过一个很典型的场景客服类 Agent 上线后后台统计显示同样的“退款流程是什么”这个问题一天被问了 800 多次。模型每次都要重新理解、重新组织答案单次成本不高但乘上 800 次、再乘上 30 天就是一笔不小的开销。而且用户的等待时间也长因为每次都要完整走一遍模型调用。1.2 哪里能省缓存的三个发力点第一个是重复的 LLM 调用。用户反复问同一类问题比如商品咨询里“你们的配送时间是多久”这个问题的上下文和答案完全可以缓存。缓存命中后直接从 Redis 返回几百毫秒变成几毫秒成本也趋近于零。第二个是工具调用的结果。很多工具返回的数据在一定时间内是不变的比如新闻列表、天气数据、价格信息。Agent 里如果频繁调用天气 API同一个城市同一个小时内的数据每次调用都用缓存返回能省掉大量第三方 API 的调用配额和网络延迟。第三个是 Agent 状态和中间产物。多轮对话里的计划、上下文、中间推理结果没必要每次都重新生成。比如 Agent 已经规划好的三步执行方案如果用户只是追问其中一步的细节Plan 本身可以原样复用。这三个点对应 Redis 的三类用法纯 KV 缓存、带 TTL 的数据缓存、结构化的会话存储。理解了这三件事后面的方案就有方向了。1.3 为什么是 Redis不是本地缓存或者直接查数据库有人会说我做个 Python dict 缓存不也行吗单机演示没问题但 Agent 服务不可能永远单实例。部署两台实例后dict 就各存各的同一个请求落到不同实例命中率直接对半砍而且没法做全局限流、没法共享热数据。用数据库存缓存就更不成立了Agent 这层的缓存要求是微秒到毫秒级返回数据库一次查询的延迟就把缓存的优势抹掉了。Redis 的优势是共享、快、结构灵活。它支持字符串、哈希、列表、有序集合等类型意味着它能干的事不只是缓存放值ZSet 可以用来做 Token 用量排行和滑动窗口限流Hash 可以存 Agent 的多步状态String 配合过期时间可以当分布式锁。这些都是用内存缓存或者数据库替代不了的。打个比方本地缓存像家里冰箱只能自己用数据库像仓库拿东西要跑一趟Redis 像小区里的公共储物柜大家都从里面取速度跟开自家冰箱差不多内容却是共享的。2. 动手前先想清楚Agent 场景下缓存什么、怎么设计 Key2.1 四类值得缓存的数据我按自己的项目经验把 Agent 场景下的缓存对象分成四类LLM 响应、工具结果、会话状态、限流计数。LLM 响应缓存指的是完全相同的 Prompt 和模型参数下模型输出直接复用。这是省钱最直接的方式。工具结果缓存适合第三方 API 返回的稳定数据比如天气、汇率、股票行情TTL 从几十秒到几十分钟不等。会话状态缓存用于存多轮对话的中间数据比如当前意图识别结果、已填好的表单字段、Agent 的 Plan这类数据用 Hash 存最合适。限流计数用于控制单个用户或单个 API Key 的请求频率防止 Agent 服务被高频调用打垮。需要注意LLM 缓存不是所有场景都适合。如果是开放式的创意写作、头脑风暴缓存的意义不大因为用户每次需求都不同。但如果你的 Agent 是客服、问答、数据分析解释、固定流程咨询这类业务的重复度非常高缓存收益就很明显。2.2 Key 设计像给快递写门牌号一样缓存 Key 设计不好后面全是坑。我见过最典型的错误是把整个 Prompt 塞进 Key结果就是缓存几乎永远不命中因为同样的语义不同的说法字面完全不一样比如“天气怎么样”和“今天天气如何”模型认为是一个意思字符串却完全不同。正确做法是给 Key 加命名空间加语义哈希。命名空间用来区分数据类型比如 llm:、tool:、session:、rate:。语义部分用固定模型名称加上规范化后的输入内容做 SHA256 哈希。这样同一个语义的请求即使标点符号有区别也能命中同一个缓存。我一般还会把模型版本号写进 Key模型升级了旧缓存自动失效不会返回过期答案。一个常见问题是规范化到什么程度我的经验是不要做太复杂的 NLP 预处理简单去空格、转小写、去标点就够了。太重的预处理会拖慢缓存查询本身捡了芝麻丢西瓜。2.3 TTL 怎么定不是越长越好TTL 的选择要按数据类型来。LLM 响应缓存我一般设 1 到 24 小时具体看业务对时效性的要求工具结果缓存按工具的稳定性来设价格类信息 60 秒天气预报 30 分钟新闻列表 5 分钟会话状态缓存通常是用户会话周期最长不超过一天限流计数窗口本身就是 TTL通常 60 秒。还有一个容易忽略的点Redis 的过期策略是惰性删除加定期删除。大量 Key 同时过期时Redis 会有短暂的 CPU 波动所以给 TTL 加一个随机扰动是好习惯。比如基础 TTL 是 3600 秒实际设置成 3600 加 0 到 300 的随机值避免缓存雪崩式的集体失效。3. 实操落地FastAPI LangChain Agent 接入 Redis 缓存3.1 环境准备与 Redis 连接池本地开发用 Docker 起一个 Redis 最省事docker run -d --name redis-cache -p 6379:6379 redis:7-alpine生产环境建议至少一主一从加哨兵或者用托管版。这里提醒一个关键配置maxmemory-policy要设置成allkeys-lru否则缓存写满后 Redis 会拒绝写入进而把所有请求压到模型和数据库上造成连锁故障。Python 侧用redis-py客户端安装命令是pip install redis这里插一句我用过的教训很多人会在 FastAPI 里每次请求都 new 一个 Redis 客户端这个做法高并发下必踩坑。redis-py内部自带连接池但前提是你得让 Redis 客户端是全局单例。我一般把连接池单独初始化核心参数是max_connections按服务预估并发量配置成 50 到 200import redis pool redis.ConnectionPool( hostlocalhost, port6379, db0, max_connections100, decode_responsesTrue, ) r redis.Redis(connection_poolpool)decode_responsesTrue会让键值对自动返回字符串而不是字节串省掉很多手动 decode 的麻烦。连接数的设置不是越大越好过大的连接池会撑爆 Redis 的client-output-buffer-limit反而引发异常。3.2 LLM 响应缓存直接把账单打三折核心思路很简单调用模型之前先根据 Prompt 内容查缓存命中就直接返回没命中才调模型把结果写回缓存。我封装了一个带缓存的方法import hashlib import json def get_llm_response_with_cache(prompt: str, model: str gpt-4o, ttl: int 3600) - str: norm_prompt .join(prompt.strip().lower().split()) cache_key fllm:{model}: hashlib.sha256(norm_prompt.encode()).hexdigest() cached r.get(cache_key) if cached is not None: return cached response call_llm(prompt, model) # 这里可以是 LangChain 的 LLM 调用 r.set(cache_key, response, exttl) return response注意几个细节。第一call_llm必须是你真实的模型调用入口LangChain 的话就是llm.invoke(prompt)或model.invoke([...])。第二缓存命中时返回的是字符串如果你的下游需要结构化的数据比如 JSON建议在写缓存时直接存 JSON 序列化的结果读取时再反序列化不要存模型输出的原始文本然后再解析一遍。第三如果你的 Agent 用了不同 temperature 和 max_tokens这些参数也必须写进 Key比如llm:{model}:{temperature}:{key}。实测下来加了这一层之后重复问题的 P95 延迟从 1.8 秒降到 20 毫秒以内效果立竿见影。3.3 工具结果缓存别让第三方 API 把你拖垮工具调用是 Agent 里最容易超时的环节。外部 API 不稳定响应慢还常常有限流。我把每个工具的调用都包了一层缓存按工具名称和参数哈希做 KeyTOOL_CACHE_TTL { search_news: 300, get_weather: 1800, query_orders: 60, } def call_tool_with_cache(tool_name: str, params: dict) - dict: key_payload json.dumps(params, sort_keysTrue, ensure_asciiFalse) cache_key ftool:{tool_name}: hashlib.sha256(key_payload.encode()).hexdigest() cached r.get(cache_key) if cached is not None: return json.loads(cached) result execute_tool(tool_name, params) ttl TOOL_CACHE_TTL.get(tool_name, 600) r.set(cache_key, json.dumps(result, ensure_asciiFalse), exttl) return result这里sort_keysTrue很关键它保证参数顺序不同但内容相同的请求能命中同一个缓存。比如{city: 北京, date: 2025-06-01}和{date: 2025-06-01, city: 北京}JSON 字符串排序后一致哈希也一致才能命中。有一种业务场景要注意用户订单查询这类数据不同用户之间永远不该互相共享缓存。所以 Key 里必须带上用户标识比如tool:query_orders:{user_id}:{hash}。这表明工具结果缓存必须区分“全局共享数据”和“用户私有数据”千万不要一刀切。3.4 Agent 状态与会话数据缓存多轮对话的 Agent 需要维护状态。Airflow 的 Plan、当前执行到第几步、已经收集到的表单信息、用户的心智模型这些都不适合放在内存里因为服务重启就丢了也不适合全量塞进数据库因为每次读写太重。用 Redis Hash 正合适。def save_agent_state(session_id: str, state: dict): key fsession:{session_id} r.hset(key, mappingstate) r.expire(key, 86400) def get_agent_state(session_id: str) - dict: key fsession:{session_id} return r.hgetall(key)sessions:{session_id}这个 Key 下的每个 Field 对应一个状态字段比如step、plan、intent。更新单字段用hset不会影响其他字段天然支持并发修改。配合 TTL 86400 秒用户一天内即使关掉页面重新进来也能恢复上下文。但如果你的状态结构很深比如嵌套 JSONHash 就不好使了。要么拍平成扁平结构存储要么干脆用 String 存序列化后的 JSON。我个人经验是Agent 状态尽量设计得扁平这不仅是缓存的需要对后续排查问题也有好处。3.5 用 Redis 做分布式限流INCR 加 EXPIRE除了缓存Redis 在 Agent 服务里第二个大用途是限流。Agent 服务比普通接口更脆弱因为一次请求会触发多次模型调用并发一高成本和下游压力都成倍增加。我用一个简单的滑动窗口计数器def is_rate_limited(user_id: str, limit: int 10, window: int 60) - bool: key frate:{user_id} count r.incr(key) if count 1: r.expire(key, window) return count limit第一次请求时 INCR 返回 1顺势设置过期时间 60 秒。窗口内第 11 次请求就会被拦截。这个方案在高并发下不是 100% 精确的滑动窗口但作为服务保护足够用。需要更严格的限流可以用 Redis 的 ZSet 做真正的时间戳滑动窗口代价是每个 Key 存储开销更大。限流触发后直接返回 429 状态码前端可以提示“稍后再试”不要让它继续打到模型上。4. 高并发下的缓存安全穿透、击穿、雪崩与调优4.1 三大经典问题在 Agent 场景的变体缓存穿透是指大量请求查询一个注定不存在的数据。比如用户问“这个优惠券代码有效吗”Agent 去查订单系统发现不存在如果每次都去查数据库缓存形同虚设。解决办法是空值缓存查询结果为 None 时也写一个空标记进 RedisTTL 设短一点比如 60 秒下次同样的查询直接走缓存。缓存击穿是指某个热点 Key 突然过期同时大量请求涌进来全部穿透到下游。AI Agent 场景里的热点 Key 经常是“最新系统公告”“热门问题标准答案”一旦过期所有用户请求同时打到模型上模型接口直接超时。解决方法后面细说用分布式锁保证单线程回源。缓存雪崩是指大量 Key 同时过期或者 Redis 本身挂了。前者我前面提过加 TTL 随机扰动后者要靠 Redis 高可用架构解决同时全链路还要有降级方案。我见过一个项目把 Redis 当成了“永远在线”的组件没做降级一次 Redis 节点切换直接把服务打挂了。正确做法是代码里捕获 Redis 异常后降级为直接调用虽然慢但服务不挂。4.2 分布式锁防击穿SET NX EX 的正确用法这里重点说一下缓存的击穿。传统做法里有个“缓存过期就让第一个线程去回源”的思路但在 Agent 场景要注意回源一次 LLM 调用可能要 2 秒如果有 10 个线程同时回源就是白白浪费 20 秒的模型调用。用 Redis 分布式锁很容易实现“单飞回源”def get_llm_with_lock(key: str, ttl: int): cached r.get(key) if cached is not None: return cached lock_key flock:{key} # SET key value NX EX 30只有 Key 不存在时才能设置成功30 秒自动释放 acquired r.set(lock_key, 1, nxTrue, ex30) if not acquired: # 别人正在回源等待后重试 time.sleep(0.1) return r.get(key) try: # 二次检查防止拿到锁之前已经被别的线程写入 cached r.get(key) if cached is not None: return cached response call_llm_from_key(key) r.set(key, response, exttl) return response finally: r.delete(lock_key)这里有一个细节锁必须带过期时间否则持有锁的线程崩溃锁就变成死锁了。其次是二次检查拿到锁之后不要立刻调模型先再查一遍缓存因为可能有其他线程在你等待锁的时候已经完成了回源。这个“先查、再锁、又查、再回源”的套路是防止重复调用的关键。4.3 连接池、序列化和 Redis 本身调优高并发下连接池参数是要调的。除了max_connections我会关注socket_connect_timeout和socket_timeout默认值往往太长一旦 Redis 卡住时请求会堆积。我通常设置连接超时 5 秒读写超时 2 秒宁可让个别请求报错重试也不要让线程卡死。序列化方面如果缓存对象是大量 JSON用json.dumps就够。但如果你缓存的是高吞吐的中间结果比如 embedding 向量、大量结构化数据可以考虑msgpack它比 JSON 体积小、序列化速度快在高频读写的场景有明显优势。注意一点一旦用了 msgpack所有读写缓存的地方都要统一用混用会导致反序列化错误。我踩过这个坑上线后出现一堆unpack requires a buffer of 4 bytes的报错定位了大半天。Redis 本身的调优最重要的参数是maxmemory和maxmemory-policy。缓存服务的maxmemory-policy建议allkeys-lru。如果你担心热门 Key 永不淘汰也要知道 LRU 策略下一旦内存不够冷门 Key 会先被淘汰热门的反而留下。这个策略和缓存场景是匹配的。5. 上线之后踩过的坑排查实录与速查表5.1 高频报错timeout 与连接耗尽我在项目上线后遇到最多的问题就是redis command timed out。这个报错的直接原因是命令执行超过了socket_timeout但背后的原因要分几种来看。第一种是连接池被占满。出现这个报错时我先用redis-cli info clients看连接数再用redis-cli client list看活跃连接。如果堆满了说明某段时间的并发请求超过了连接池上限或者有请求拿连接后没有及时归还。代码里要检查 Redis 操作是否都在with上下文或 try/finally 中完成避免异常后连接泄漏。第二种是 Redis 服务本身慢。redis-cli info stats里看instantaneous_ops_per_sec和latest_fork_usec如果命令总数不高但响应慢可能是 Redis 所在的服务器内存 swap 了或者持久化策略写盘太频繁占用了主线程时间。生产环境开 AOF 时要注意appendfsync参数everysec是比较折中的选择。第三种是网络问题。比如 Redis 和你的 Agent 服务不在同一个可用区跨可用区的延迟通常比本区高 1 到 3 毫秒。对一般接口来说区别不大但 Agent 里高频缓存访问会放大这个延迟。5.2 缓存不生效、缓存命中率低的排查加了缓存之后发现命中率不到 10%这种情况我见过多次。第一个要查的是 Key 是否一致。我把 LLM 缓存和工具缓存的 Key 生成规则单独抽成一个函数方便统一测试。调接口的时候把真实 Key 打出来直接去 Redis 里get一下看当初写入的 Key 和读取的 Key 是不是同一个。很多时候问题出在规范化方式不同比如一个地方去标点另一个地方没去。第二个要查的是序列化方式。写入时存的是 JSON 字符串读取时却用 msgpack 反序列化或者反过来都会直接抛异常但有些代码吞掉了异常导致缓存命中失败。排查时可以用type命令看 Redis 里存的到底是什么类型用get看原始值长什么样。第三个原因更隐蔽TTL 设置太短。比如消息列表 TTL 设了 30 秒用户访问一次后 30 秒内没有第二次访问缓存就过期了下一次请求又重新回源。数据表明命中率低不一定是 Key 设计问题可能是 TTL 短于用户的真实访问间隔。把 TTL 根据访问日志重新统计后再调整。5.3 序列化和主从一致性主从架构下还有一个容易出现的问题从节点读到了过期数据。某些长时间没有写入的 Key 在主节点已经过期但从节点可能还没清理如果客户端把读请求打到从节点上就返回了旧数据。Redis 从节点默认会执行主节点同步来的删除命令一般来说问题不大但如果你手动改过配置或者网络分区导致同步延迟就需要注意。我的建议是对一致性要求高的缓存比如订单状态、支付结果在主节点上读写或者直接不缓存对一致性要求低的比如新闻列表、天气信息从节点读完全没问题。这需要在业务层就明确区分而不是把所有缓存一视同仁。5.4 常见问题速查表问题现象优先排查点常用解决手段缓存全部不命中Key 生成规则、规范化逻辑统一 Key 生成函数打印真实 Key 对比高并发下 Timeout连接池大小、Redis 主线程阻塞调大连接池、检查 AOF 策略、加连接超时启动后内存暴涨maxmemory未设置设置maxmemory和allkeys-lru读到旧数据主从同步延迟、TTL 过长高一致性数据走主节点缩短 TTL缓存穿透导致下游压力大空结果是否有缓存加空值缓存TTL 60 秒左右热点 Key 过期打爆模型是否有分布式锁用 SET NX EX 单飞回源反序列化报错写入和读取序列化方式是否一致统一为 JSON 或统一为 msgpack这张表不算完整但覆盖了我踩过的大部分坑。真正上线前建议把表里的场景都写成测试用例压测时多跑几轮比到时候看报警再排查要省力得多。6. 一些实战中的心得体会最后分享一点个人的体会。缓存不是加得越多越好我对 Agent 里每一处缓存的收益都做了统计再决定放不放。有些地方放了缓存反而麻烦比如涉及用户私人数据的工具结果缓存了要处理权限隔离、数据更新、作废机制复杂度提升一大截。最开始我先只缓存了两类全网共享的 LLM 响应和工具结果这两类收益最大复杂度最低。还有一点缓存的监控一定要做起来。我在 Redis 的INFO上定时采集命中率、内存占用、慢查询日志再配合代码里给每条缓存读写打上埋点。上线第一个月我根据命中率数据发现某个业务线的缓存命中率波动很大顺着排查发现是 TTL 设置和用户访问频次不匹配调整之后成本下降了 40%。最后再给一个实用的小技巧在本地开发时同一个服务可能同时连着开发和生产两套 Redis为了不把开发请求写进生产缓存我在所有的 Key 前都加了一个环境前缀比如dev:llm:和prod:llm:切换环境时只需要改一个前缀变量。加上这个前缀之后开发调试时再也不用担心污染线上数据缓存也避免开发环境用旧缓存的尴尬。这个习惯帮我省了非常多的麻烦建议你从第一天就养成。
AI Agent性能优化实战：用Redis缓存解决高并发瓶颈
AI Agent 真正上线跑起来和 demo 完全是两回事。本地搭个例子感受不到压力等用户量上来、调用量上来最先扛不住的不一定是模型接口而是 Agent 编排服务本身。我这两年一直在做 Agent 落地从 LangChain 起步用 FastAPI 起接口中间踩过无数性能相关的坑。这里面用 Redis 做缓存层是收益最高、见效最快的一步也是最容易被一开始就埋头写 Prompt 的人忽视的一步。这篇文章不聊“缓存是什么”直接聊 AI Agent 和 Redis 怎么组合。我先拆解 Agent 请求链路里真正慢的环节再给出缓存设计的核心思路接着放一套可以直接抄的 FastAPI LangChain 集成 Redis 的实操代码最后把上线后容易踩的坑也一并列出来。1. AI Agent 请求链路里的性能瓶颈到底在哪1.1 一次 Agent 请求背后从用户提问到最终回复先看一次普通 Agent 请求要经历什么。用户发来一句话服务先把历史会话拉出来组装成上下文然后送入大模型。模型不是直接给答案它可能先返回一个工具调用意图服务执行工具查库、调 API把结果再次拼接进 Prompt再调用一次模型甚至循环好几轮才会输出最终结果。这意味着一次用户请求背后可能是 3 到 5 次 LLM 调用外加若干次工具调用。每轮 LLM 调用的延迟是几百毫秒到几秒费用按 token 计费越长的上下文越贵。如果 100 个用户同时提问而你的 Agent 服务没有缓存等于同一个问题可能被重复计算几百次、重复调用几百次。这既是性能问题也是账单问题。我见过一个很典型的场景客服类 Agent 上线后后台统计显示同样的“退款流程是什么”这个问题一天被问了 800 多次。模型每次都要重新理解、重新组织答案单次成本不高但乘上 800 次、再乘上 30 天就是一笔不小的开销。而且用户的等待时间也长因为每次都要完整走一遍模型调用。1.2 哪里能省缓存的三个发力点第一个是重复的 LLM 调用。用户反复问同一类问题比如商品咨询里“你们的配送时间是多久”这个问题的上下文和答案完全可以缓存。缓存命中后直接从 Redis 返回几百毫秒变成几毫秒成本也趋近于零。第二个是工具调用的结果。很多工具返回的数据在一定时间内是不变的比如新闻列表、天气数据、价格信息。Agent 里如果频繁调用天气 API同一个城市同一个小时内的数据每次调用都用缓存返回能省掉大量第三方 API 的调用配额和网络延迟。第三个是 Agent 状态和中间产物。多轮对话里的计划、上下文、中间推理结果没必要每次都重新生成。比如 Agent 已经规划好的三步执行方案如果用户只是追问其中一步的细节Plan 本身可以原样复用。这三个点对应 Redis 的三类用法纯 KV 缓存、带 TTL 的数据缓存、结构化的会话存储。理解了这三件事后面的方案就有方向了。1.3 为什么是 Redis不是本地缓存或者直接查数据库有人会说我做个 Python dict 缓存不也行吗单机演示没问题但 Agent 服务不可能永远单实例。部署两台实例后dict 就各存各的同一个请求落到不同实例命中率直接对半砍而且没法做全局限流、没法共享热数据。用数据库存缓存就更不成立了Agent 这层的缓存要求是微秒到毫秒级返回数据库一次查询的延迟就把缓存的优势抹掉了。Redis 的优势是共享、快、结构灵活。它支持字符串、哈希、列表、有序集合等类型意味着它能干的事不只是缓存放值ZSet 可以用来做 Token 用量排行和滑动窗口限流Hash 可以存 Agent 的多步状态String 配合过期时间可以当分布式锁。这些都是用内存缓存或者数据库替代不了的。打个比方本地缓存像家里冰箱只能自己用数据库像仓库拿东西要跑一趟Redis 像小区里的公共储物柜大家都从里面取速度跟开自家冰箱差不多内容却是共享的。2. 动手前先想清楚Agent 场景下缓存什么、怎么设计 Key2.1 四类值得缓存的数据我按自己的项目经验把 Agent 场景下的缓存对象分成四类LLM 响应、工具结果、会话状态、限流计数。LLM 响应缓存指的是完全相同的 Prompt 和模型参数下模型输出直接复用。这是省钱最直接的方式。工具结果缓存适合第三方 API 返回的稳定数据比如天气、汇率、股票行情TTL 从几十秒到几十分钟不等。会话状态缓存用于存多轮对话的中间数据比如当前意图识别结果、已填好的表单字段、Agent 的 Plan这类数据用 Hash 存最合适。限流计数用于控制单个用户或单个 API Key 的请求频率防止 Agent 服务被高频调用打垮。需要注意LLM 缓存不是所有场景都适合。如果是开放式的创意写作、头脑风暴缓存的意义不大因为用户每次需求都不同。但如果你的 Agent 是客服、问答、数据分析解释、固定流程咨询这类业务的重复度非常高缓存收益就很明显。2.2 Key 设计像给快递写门牌号一样缓存 Key 设计不好后面全是坑。我见过最典型的错误是把整个 Prompt 塞进 Key结果就是缓存几乎永远不命中因为同样的语义不同的说法字面完全不一样比如“天气怎么样”和“今天天气如何”模型认为是一个意思字符串却完全不同。正确做法是给 Key 加命名空间加语义哈希。命名空间用来区分数据类型比如 llm:、tool:、session:、rate:。语义部分用固定模型名称加上规范化后的输入内容做 SHA256 哈希。这样同一个语义的请求即使标点符号有区别也能命中同一个缓存。我一般还会把模型版本号写进 Key模型升级了旧缓存自动失效不会返回过期答案。一个常见问题是规范化到什么程度我的经验是不要做太复杂的 NLP 预处理简单去空格、转小写、去标点就够了。太重的预处理会拖慢缓存查询本身捡了芝麻丢西瓜。2.3 TTL 怎么定不是越长越好TTL 的选择要按数据类型来。LLM 响应缓存我一般设 1 到 24 小时具体看业务对时效性的要求工具结果缓存按工具的稳定性来设价格类信息 60 秒天气预报 30 分钟新闻列表 5 分钟会话状态缓存通常是用户会话周期最长不超过一天限流计数窗口本身就是 TTL通常 60 秒。还有一个容易忽略的点Redis 的过期策略是惰性删除加定期删除。大量 Key 同时过期时Redis 会有短暂的 CPU 波动所以给 TTL 加一个随机扰动是好习惯。比如基础 TTL 是 3600 秒实际设置成 3600 加 0 到 300 的随机值避免缓存雪崩式的集体失效。3. 实操落地FastAPI LangChain Agent 接入 Redis 缓存3.1 环境准备与 Redis 连接池本地开发用 Docker 起一个 Redis 最省事docker run -d --name redis-cache -p 6379:6379 redis:7-alpine生产环境建议至少一主一从加哨兵或者用托管版。这里提醒一个关键配置maxmemory-policy要设置成allkeys-lru否则缓存写满后 Redis 会拒绝写入进而把所有请求压到模型和数据库上造成连锁故障。Python 侧用redis-py客户端安装命令是pip install redis这里插一句我用过的教训很多人会在 FastAPI 里每次请求都 new 一个 Redis 客户端这个做法高并发下必踩坑。redis-py内部自带连接池但前提是你得让 Redis 客户端是全局单例。我一般把连接池单独初始化核心参数是max_connections按服务预估并发量配置成 50 到 200import redis pool redis.ConnectionPool( hostlocalhost, port6379, db0, max_connections100, decode_responsesTrue, ) r redis.Redis(connection_poolpool)decode_responsesTrue会让键值对自动返回字符串而不是字节串省掉很多手动 decode 的麻烦。连接数的设置不是越大越好过大的连接池会撑爆 Redis 的client-output-buffer-limit反而引发异常。3.2 LLM 响应缓存直接把账单打三折核心思路很简单调用模型之前先根据 Prompt 内容查缓存命中就直接返回没命中才调模型把结果写回缓存。我封装了一个带缓存的方法import hashlib import json def get_llm_response_with_cache(prompt: str, model: str gpt-4o, ttl: int 3600) - str: norm_prompt .join(prompt.strip().lower().split()) cache_key fllm:{model}: hashlib.sha256(norm_prompt.encode()).hexdigest() cached r.get(cache_key) if cached is not None: return cached response call_llm(prompt, model) # 这里可以是 LangChain 的 LLM 调用 r.set(cache_key, response, exttl) return response注意几个细节。第一call_llm必须是你真实的模型调用入口LangChain 的话就是llm.invoke(prompt)或model.invoke([...])。第二缓存命中时返回的是字符串如果你的下游需要结构化的数据比如 JSON建议在写缓存时直接存 JSON 序列化的结果读取时再反序列化不要存模型输出的原始文本然后再解析一遍。第三如果你的 Agent 用了不同 temperature 和 max_tokens这些参数也必须写进 Key比如llm:{model}:{temperature}:{key}。实测下来加了这一层之后重复问题的 P95 延迟从 1.8 秒降到 20 毫秒以内效果立竿见影。3.3 工具结果缓存别让第三方 API 把你拖垮工具调用是 Agent 里最容易超时的环节。外部 API 不稳定响应慢还常常有限流。我把每个工具的调用都包了一层缓存按工具名称和参数哈希做 KeyTOOL_CACHE_TTL { search_news: 300, get_weather: 1800, query_orders: 60, } def call_tool_with_cache(tool_name: str, params: dict) - dict: key_payload json.dumps(params, sort_keysTrue, ensure_asciiFalse) cache_key ftool:{tool_name}: hashlib.sha256(key_payload.encode()).hexdigest() cached r.get(cache_key) if cached is not None: return json.loads(cached) result execute_tool(tool_name, params) ttl TOOL_CACHE_TTL.get(tool_name, 600) r.set(cache_key, json.dumps(result, ensure_asciiFalse), exttl) return result这里sort_keysTrue很关键它保证参数顺序不同但内容相同的请求能命中同一个缓存。比如{city: 北京, date: 2025-06-01}和{date: 2025-06-01, city: 北京}JSON 字符串排序后一致哈希也一致才能命中。有一种业务场景要注意用户订单查询这类数据不同用户之间永远不该互相共享缓存。所以 Key 里必须带上用户标识比如tool:query_orders:{user_id}:{hash}。这表明工具结果缓存必须区分“全局共享数据”和“用户私有数据”千万不要一刀切。3.4 Agent 状态与会话数据缓存多轮对话的 Agent 需要维护状态。Airflow 的 Plan、当前执行到第几步、已经收集到的表单信息、用户的心智模型这些都不适合放在内存里因为服务重启就丢了也不适合全量塞进数据库因为每次读写太重。用 Redis Hash 正合适。def save_agent_state(session_id: str, state: dict): key fsession:{session_id} r.hset(key, mappingstate) r.expire(key, 86400) def get_agent_state(session_id: str) - dict: key fsession:{session_id} return r.hgetall(key)sessions:{session_id}这个 Key 下的每个 Field 对应一个状态字段比如step、plan、intent。更新单字段用hset不会影响其他字段天然支持并发修改。配合 TTL 86400 秒用户一天内即使关掉页面重新进来也能恢复上下文。但如果你的状态结构很深比如嵌套 JSONHash 就不好使了。要么拍平成扁平结构存储要么干脆用 String 存序列化后的 JSON。我个人经验是Agent 状态尽量设计得扁平这不仅是缓存的需要对后续排查问题也有好处。3.5 用 Redis 做分布式限流INCR 加 EXPIRE除了缓存Redis 在 Agent 服务里第二个大用途是限流。Agent 服务比普通接口更脆弱因为一次请求会触发多次模型调用并发一高成本和下游压力都成倍增加。我用一个简单的滑动窗口计数器def is_rate_limited(user_id: str, limit: int 10, window: int 60) - bool: key frate:{user_id} count r.incr(key) if count 1: r.expire(key, window) return count limit第一次请求时 INCR 返回 1顺势设置过期时间 60 秒。窗口内第 11 次请求就会被拦截。这个方案在高并发下不是 100% 精确的滑动窗口但作为服务保护足够用。需要更严格的限流可以用 Redis 的 ZSet 做真正的时间戳滑动窗口代价是每个 Key 存储开销更大。限流触发后直接返回 429 状态码前端可以提示“稍后再试”不要让它继续打到模型上。4. 高并发下的缓存安全穿透、击穿、雪崩与调优4.1 三大经典问题在 Agent 场景的变体缓存穿透是指大量请求查询一个注定不存在的数据。比如用户问“这个优惠券代码有效吗”Agent 去查订单系统发现不存在如果每次都去查数据库缓存形同虚设。解决办法是空值缓存查询结果为 None 时也写一个空标记进 RedisTTL 设短一点比如 60 秒下次同样的查询直接走缓存。缓存击穿是指某个热点 Key 突然过期同时大量请求涌进来全部穿透到下游。AI Agent 场景里的热点 Key 经常是“最新系统公告”“热门问题标准答案”一旦过期所有用户请求同时打到模型上模型接口直接超时。解决方法后面细说用分布式锁保证单线程回源。缓存雪崩是指大量 Key 同时过期或者 Redis 本身挂了。前者我前面提过加 TTL 随机扰动后者要靠 Redis 高可用架构解决同时全链路还要有降级方案。我见过一个项目把 Redis 当成了“永远在线”的组件没做降级一次 Redis 节点切换直接把服务打挂了。正确做法是代码里捕获 Redis 异常后降级为直接调用虽然慢但服务不挂。4.2 分布式锁防击穿SET NX EX 的正确用法这里重点说一下缓存的击穿。传统做法里有个“缓存过期就让第一个线程去回源”的思路但在 Agent 场景要注意回源一次 LLM 调用可能要 2 秒如果有 10 个线程同时回源就是白白浪费 20 秒的模型调用。用 Redis 分布式锁很容易实现“单飞回源”def get_llm_with_lock(key: str, ttl: int): cached r.get(key) if cached is not None: return cached lock_key flock:{key} # SET key value NX EX 30只有 Key 不存在时才能设置成功30 秒自动释放 acquired r.set(lock_key, 1, nxTrue, ex30) if not acquired: # 别人正在回源等待后重试 time.sleep(0.1) return r.get(key) try: # 二次检查防止拿到锁之前已经被别的线程写入 cached r.get(key) if cached is not None: return cached response call_llm_from_key(key) r.set(key, response, exttl) return response finally: r.delete(lock_key)这里有一个细节锁必须带过期时间否则持有锁的线程崩溃锁就变成死锁了。其次是二次检查拿到锁之后不要立刻调模型先再查一遍缓存因为可能有其他线程在你等待锁的时候已经完成了回源。这个“先查、再锁、又查、再回源”的套路是防止重复调用的关键。4.3 连接池、序列化和 Redis 本身调优高并发下连接池参数是要调的。除了max_connections我会关注socket_connect_timeout和socket_timeout默认值往往太长一旦 Redis 卡住时请求会堆积。我通常设置连接超时 5 秒读写超时 2 秒宁可让个别请求报错重试也不要让线程卡死。序列化方面如果缓存对象是大量 JSON用json.dumps就够。但如果你缓存的是高吞吐的中间结果比如 embedding 向量、大量结构化数据可以考虑msgpack它比 JSON 体积小、序列化速度快在高频读写的场景有明显优势。注意一点一旦用了 msgpack所有读写缓存的地方都要统一用混用会导致反序列化错误。我踩过这个坑上线后出现一堆unpack requires a buffer of 4 bytes的报错定位了大半天。Redis 本身的调优最重要的参数是maxmemory和maxmemory-policy。缓存服务的maxmemory-policy建议allkeys-lru。如果你担心热门 Key 永不淘汰也要知道 LRU 策略下一旦内存不够冷门 Key 会先被淘汰热门的反而留下。这个策略和缓存场景是匹配的。5. 上线之后踩过的坑排查实录与速查表5.1 高频报错timeout 与连接耗尽我在项目上线后遇到最多的问题就是redis command timed out。这个报错的直接原因是命令执行超过了socket_timeout但背后的原因要分几种来看。第一种是连接池被占满。出现这个报错时我先用redis-cli info clients看连接数再用redis-cli client list看活跃连接。如果堆满了说明某段时间的并发请求超过了连接池上限或者有请求拿连接后没有及时归还。代码里要检查 Redis 操作是否都在with上下文或 try/finally 中完成避免异常后连接泄漏。第二种是 Redis 服务本身慢。redis-cli info stats里看instantaneous_ops_per_sec和latest_fork_usec如果命令总数不高但响应慢可能是 Redis 所在的服务器内存 swap 了或者持久化策略写盘太频繁占用了主线程时间。生产环境开 AOF 时要注意appendfsync参数everysec是比较折中的选择。第三种是网络问题。比如 Redis 和你的 Agent 服务不在同一个可用区跨可用区的延迟通常比本区高 1 到 3 毫秒。对一般接口来说区别不大但 Agent 里高频缓存访问会放大这个延迟。5.2 缓存不生效、缓存命中率低的排查加了缓存之后发现命中率不到 10%这种情况我见过多次。第一个要查的是 Key 是否一致。我把 LLM 缓存和工具缓存的 Key 生成规则单独抽成一个函数方便统一测试。调接口的时候把真实 Key 打出来直接去 Redis 里get一下看当初写入的 Key 和读取的 Key 是不是同一个。很多时候问题出在规范化方式不同比如一个地方去标点另一个地方没去。第二个要查的是序列化方式。写入时存的是 JSON 字符串读取时却用 msgpack 反序列化或者反过来都会直接抛异常但有些代码吞掉了异常导致缓存命中失败。排查时可以用type命令看 Redis 里存的到底是什么类型用get看原始值长什么样。第三个原因更隐蔽TTL 设置太短。比如消息列表 TTL 设了 30 秒用户访问一次后 30 秒内没有第二次访问缓存就过期了下一次请求又重新回源。数据表明命中率低不一定是 Key 设计问题可能是 TTL 短于用户的真实访问间隔。把 TTL 根据访问日志重新统计后再调整。5.3 序列化和主从一致性主从架构下还有一个容易出现的问题从节点读到了过期数据。某些长时间没有写入的 Key 在主节点已经过期但从节点可能还没清理如果客户端把读请求打到从节点上就返回了旧数据。Redis 从节点默认会执行主节点同步来的删除命令一般来说问题不大但如果你手动改过配置或者网络分区导致同步延迟就需要注意。我的建议是对一致性要求高的缓存比如订单状态、支付结果在主节点上读写或者直接不缓存对一致性要求低的比如新闻列表、天气信息从节点读完全没问题。这需要在业务层就明确区分而不是把所有缓存一视同仁。5.4 常见问题速查表问题现象优先排查点常用解决手段缓存全部不命中Key 生成规则、规范化逻辑统一 Key 生成函数打印真实 Key 对比高并发下 Timeout连接池大小、Redis 主线程阻塞调大连接池、检查 AOF 策略、加连接超时启动后内存暴涨maxmemory未设置设置maxmemory和allkeys-lru读到旧数据主从同步延迟、TTL 过长高一致性数据走主节点缩短 TTL缓存穿透导致下游压力大空结果是否有缓存加空值缓存TTL 60 秒左右热点 Key 过期打爆模型是否有分布式锁用 SET NX EX 单飞回源反序列化报错写入和读取序列化方式是否一致统一为 JSON 或统一为 msgpack这张表不算完整但覆盖了我踩过的大部分坑。真正上线前建议把表里的场景都写成测试用例压测时多跑几轮比到时候看报警再排查要省力得多。6. 一些实战中的心得体会最后分享一点个人的体会。缓存不是加得越多越好我对 Agent 里每一处缓存的收益都做了统计再决定放不放。有些地方放了缓存反而麻烦比如涉及用户私人数据的工具结果缓存了要处理权限隔离、数据更新、作废机制复杂度提升一大截。最开始我先只缓存了两类全网共享的 LLM 响应和工具结果这两类收益最大复杂度最低。还有一点缓存的监控一定要做起来。我在 Redis 的INFO上定时采集命中率、内存占用、慢查询日志再配合代码里给每条缓存读写打上埋点。上线第一个月我根据命中率数据发现某个业务线的缓存命中率波动很大顺着排查发现是 TTL 设置和用户访问频次不匹配调整之后成本下降了 40%。最后再给一个实用的小技巧在本地开发时同一个服务可能同时连着开发和生产两套 Redis为了不把开发请求写进生产缓存我在所有的 Key 前都加了一个环境前缀比如dev:llm:和prod:llm:切换环境时只需要改一个前缀变量。加上这个前缀之后开发调试时再也不用担心污染线上数据缓存也避免开发环境用旧缓存的尴尬。这个习惯帮我省了非常多的麻烦建议你从第一天就养成。