拓冰网站定制 
1. AI Agent 接入 Redis 缓存从零搭建到扛住并发的实战拆解AI Agent 这个方向最近一年有多热不用我多说。但真正上手搭过 Agent 的人都知道一个能跑起来的 Demo 和一个能扛住真实流量的系统之间差的不只是模型能力更多是工程层面的东西。其中 Redis 缓存就是最容易被忽视、却又最能决定 Agent 响应速度和并发上限的一环。我自己从去年开始陆续搭过几个 AI Agent 项目有基于 Python 的也有用 Rust 重写的踩过的坑从连接池打满到缓存雪崩从序列化格式选错到分布式锁死锁基本都经历了一遍。这篇就把 AI Agent 和 Redis 缓存结合这件事从架构设计到实操落地再到线上问题排查完整拆一遍。不管你是刚接触 AI Agent 搭建的新手还是已经在做线上缓存治理的老手应该都能从里面找到能直接抄作业的东西。核心就一句话AI Agent 的瓶颈往往不在模型推理而在数据层的读写效率Redis 缓存是解决这个问题的第一把刀但用不好它自己就会变成新的瓶颈。2. 为什么 AI Agent 必须配 Redis 缓存2.1 Agent 的请求链路到底慢在哪先搞清楚一个 AI Agent 处理一次用户请求时间都花在哪了。典型链路是这样的接收用户输入做意图识别调用工具或检索知识库拼装 Prompt请求大模型拿到结果后做后处理最后返回。这里面大模型推理通常占大头几百毫秒到几秒不等但除了模型本身还有大量重复性的数据操作在拖后腿。比如多轮对话场景每一轮都要把历史消息读出来拼进上下文。如果每次都从数据库或者对象存储里拉光是 IO 就够呛。再比如工具调用Agent 经常需要查用户信息、查配置、查知识库片段这些数据变化频率低但读取频率极高。还有会话状态、限流计数、幂等标记这些都是典型的缓存场景。我实测过一个对话 Agent在没加缓存的情况下单次请求里数据库相关操作累计耗时能到 300 到 500 毫秒加了 Redis 之后压到 20 毫秒以内。这个差距在低并发时感知不明显一旦 QPS 上去就是能不能扛住的问题。2.2 Redis 在 Agent 架构里的四个核心角色很多人一提 Redis 就只想到缓存其实在 AI Agent 体系里它至少承担四个角色每个角色的配置策略都不一样。第一个是会话上下文缓存。多轮对话的历史消息、用户偏好、当前任务状态这些数据读写频繁、生命周期跟会话绑定非常适合放 Redis。用 Hash 结构存一个会话一个 key字段存消息列表和元数据设置合理的过期时间。第二个是工具调用结果缓存。Agent 调用的很多工具是幂等的比如查天气、查汇率、查知识库同样的输入短时间内结果不会变。这类结果缓存起来能大幅减少外部 API 调用既省钱又提速。第三个是限流与配额计数。Agent 服务通常要对用户或 API Key 做限流Redis 的原子递增和过期机制天然适合做计数器。用 INCR 加 EXPIRE或者用 Lua 脚本保证原子性。第四个是分布式锁与幂等控制。多个 Agent 实例同时处理同一个任务时需要互斥。比如同一个用户同时发了两条消息要保证不会重复触发某个副作用操作这时候分布式锁就派上用场了。2.3 不接缓存会踩的三个坑不接缓存最直接的后果就是数据库压力大。Agent 的读放大特别严重一次用户请求可能触发十几次数据读取数据库连接池很容易被打满然后就是经典的redis command timed out或者数据库连接超时。第二个坑是响应时间不稳定。数据库的 P99 延迟远高于 Redis一旦有慢查询或者锁竞争Agent 的响应时间就会抖动用户体验很差。而 Redis 是内存操作延迟稳定在亚毫秒级。第三个坑是成本。大模型调用本身按 token 计费如果因为上下文拼装慢导致重试或者因为工具调用重复执行成本会悄悄涨上去。缓存能砍掉大量重复计算和重复调用。3. Redis 缓存方案选型与核心配置3.1 数据结构怎么选别拿 String 存一切Redis 的数据类型选对了性能和可维护性差一大截。我见过太多项目所有东西都用 String 存 JSON能用是能用但浪费内存也浪费操作效率。会话上下文用Hash最合适。一个会话一个 key消息列表、用户信息、任务状态作为不同 field。好处是可以单独更新某个 field不用整体读写。比如只更新任务状态时用 HSET 就行不用把整个会话读出来改完再写回去。工具调用结果缓存用String存序列化后的结果就行key 设计成tool:{tool_name}:{input_hash}简单直接。如果结果比较大可以考虑压缩后再存。限流计数用String配合 INCR 和 EXPIRE或者直接用 Redis 的滑动窗口方案。计数场景不需要复杂结构原子操作才是关键。排行榜或者需要排序的场景用Sorted Set。比如 Agent 的任务优先级队列用 ZADD 加分数ZRANGE 按优先级取任务。去重场景用Set或者Bitmap。比如判断某个用户今天是否已经触发过某个操作用 Set 存用户 ID或者用 Bitmap 更省内存。3.2 序列化格式JSON 还是 MessagePack序列化格式直接影响存储体积和序列化开销。JSON 可读性好、调试方便但体积大、解析慢。MessagePack 体积小、速度快但可读性差。Protobuf 更极致但需要定义 schema改起来麻烦。我的建议是开发调试阶段用 JSON线上对性能敏感的场景换 MessagePack。Python 里用msgpack库Rust 里用rmp-serde切换成本不高。实测下来 MessagePack 比 JSON 体积小 30% 到 50%序列化速度快 2 到 3 倍。如果数据里有大量重复的字段名可以考虑把字段名映射成短码再序列化进一步压缩体积。不过这会牺牲可读性看团队取舍。3.3 过期策略与内存淘汰过期时间设置是个技术活。设太短缓存频繁失效等于没缓存设太长数据陈旧还可能把内存撑爆。会话上下文一般设 30 分钟到 2 小时跟业务场景有关。工具调用结果看数据变化频率天气类设 10 分钟汇率类设 1 分钟知识库类可以设几小时甚至一天。内存淘汰策略要选对。Agent 场景下我一般用allkeys-lru让 Redis 在内存满时淘汰最近最少使用的 key。如果缓存的数据重要性有区分可以用volatile-lru只淘汰设置了过期时间的 key把永久数据保护起来。注意千万不要用noeviction内存满了之后写操作直接报错Agent 会大面积失败。这个坑我踩过线上报警响成一片。3.4 连接池配置别让连接成为瓶颈连接池配置不好高并发下会出现连接等待甚至超时。Python 的 redis-py 用ConnectionPoolRust 的 redis-rs 用ConnectionManager或者连接池。关键参数是最大连接数。设太小请求排队设太大Redis 服务端压力大。经验值是最大连接数设为预期 QPS 的 1.5 到 2 倍但不要超过 Redis 的maxclients配置。比如预期 1000 QPS每个请求平均占用连接 5 毫秒那并发连接数大概是 5但考虑到突发流量设 50 到 100 比较稳妥。还要设置合理的超时时间。连接超时和读写超时都要设一般 1 到 3 秒。超时太长会拖垮整个请求链路太短会误杀正常请求。4. AI Agent 缓存实操从搭建到跑通4.1 环境准备与 Redis 安装先搞定 Redis。Linux 上用包管理器装最省事macOS 用 HomebrewWindows 建议用 Docker。生产环境强烈建议用 Docker 或者直接上云服务别自己编译。# macOS 安装 brew install redis brew services start redis # Docker 安装单机 docker run -d --name redis -p 6379:6379 redis:7-alpine # Docker 安装主从 docker run -d --name redis-master -p 6379:6379 redis:7-alpine docker run -d --name redis-slave -p 6380:6379 redis:7-alpine redis-server --slaveof host.docker.internal 6379装完用redis-cli连上去测一下ping返回PONG就通了。可视化客户端可以用 RedisInsight 或者 Another Redis Desktop Manager调试的时候看 key 很方便。4.2 Python 侧接入示例Python 是 AI Agent 最常用的语言redis-py 是标配。下面是一个带连接池和序列化的封装示例。import redis import msgpack import hashlib from typing import Any, Optional class AgentCache: def __init__(self, hostlocalhost, port6379, db0, max_connections100): self.pool redis.ConnectionPool( hosthost, portport, dbdb, max_connectionsmax_connections, socket_timeout3, socket_connect_timeout2, decode_responsesFalse ) self.client redis.Redis(connection_poolself.pool) def _serialize(self, value: Any) - bytes: return msgpack.packb(value, use_bin_typeTrue) def _deserialize(self, data: bytes) - Any: return msgpack.unpackb(data, rawFalse) def get_session(self, session_id: str) - Optional[dict]: key fagent:session:{session_id} data self.client.hgetall(key) if not data: return None return {k.decode(): self._deserialize(v) for k, v in data.items()} def set_session_field(self, session_id: str, field: str, value: Any, ttl: int 3600): key fagent:session:{session_id} pipe self.client.pipeline() pipe.hset(key, field, self._serialize(value)) pipe.expire(key, ttl) pipe.execute() def cache_tool_result(self, tool_name: str, input_data: Any, result: Any, ttl: int 600): input_hash hashlib.md5(str(input_data).encode()).hexdigest() key fagent:tool:{tool_name}:{input_hash} self.client.setex(key, ttl, self._serialize(result)) def get_tool_result(self, tool_name: str, input_data: Any) - Optional[Any]: input_hash hashlib.md5(str(input_data).encode()).hexdigest() key fagent:tool:{tool_name}:{input_hash} data self.client.get(key) return self._deserialize(data) if data else None这个封装里几个点值得说。用 pipeline 把 HSET 和 EXPIRE 打包发送减少网络往返。序列化用 MessagePack体积和速度都优于 JSON。key 命名用冒号分隔方便用SCAN按前缀遍历。4.3 Rust 侧接入示例如果 Agent 用 Rust 写redis-rs 是主流选择。Rust 的异步生态配合连接池性能很能打。use redis::AsyncCommands; use redis::aio::ConnectionManager; use serde::{Serialize, Deserialize}; #[derive(Serialize, Deserialize)] struct SessionData { messages: VecString, state: String, } pub struct AgentCache { conn: ConnectionManager, } impl AgentCache { pub async fn new(redis_url: str) - redis::RedisResultSelf { let client redis::Client::open(redis_url)?; let conn ConnectionManager::new(client).await?; Ok(Self { conn }) } pub async fn set_session(mut self, session_id: str, data: SessionData, ttl: u64) - redis::RedisResult() { let key format!(agent:session:{}, session_id); let value serde_json::to_string(data).unwrap(); let _: () self.conn.set_ex(key, value, ttl).await?; Ok(()) } pub async fn get_session(mut self, session_id: str) - redis::RedisResultOptionSessionData { let key format!(agent:session:{}, session_id); let value: OptionString self.conn.get(key).await?; Ok(value.and_then(|v| serde_json::from_str(v).ok())) } }Rust 这边用ConnectionManager而不是普通连接因为它自带重连和连接复用适合长时间运行的服务。序列化用 serde_json如果追求极致性能可以换 rmp-serde。4.4 分布式锁的正确用法Agent 处理任务时经常需要互斥比如同一个任务不能被两个实例同时执行。Redis 分布式锁是常见方案但坑很多。import uuid import time class RedisLock: def __init__(self, client, key: str, ttl: int 30): self.client client self.key fagent:lock:{key} self.ttl ttl self.token str(uuid.uuid4()) def acquire(self, timeout: int 10) - bool: end time.time() timeout while time.time() end: if self.client.set(self.key, self.token, nxTrue, exself.ttl): return True time.sleep(0.05) return False def release(self): lua if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end self.client.eval(lua, 1, self.key, self.token)关键点有三个。第一value 必须用唯一 token释放时校验防止误删别人的锁。第二释放锁必须用 Lua 脚本保证原子性先 get 再 del 会有竞态。第三一定要设过期时间防止持锁进程崩溃后锁永远不释放。提示如果业务对锁的可靠性要求极高考虑 Redlock 算法或者直接用 etcd、ZooKeeper。Redis 单机锁在主从切换时可能丢锁这个风险要评估。5. 高并发场景下的缓存治理与问题排查5.1 缓存穿透、击穿、雪崩的应对这三个词面试常问但真正在 Agent 场景里踩过才知道疼。缓存穿透是查一个不存在的数据缓存和数据库都没有每次请求都打到数据库。Agent 场景里常见于查不存在的用户或会话。解决办法是缓存空值设一个较短的过期时间比如 60 秒。或者用布隆过滤器提前拦截。缓存击穿是某个热点 key 过期瞬间大量请求同时打到数据库。Agent 里常见于热门工具的结果缓存。解决办法是热点 key 不过期或者用互斥锁保证只有一个请求去回源其他请求等待。缓存雪崩是大量 key 同时过期数据库瞬间压力暴增。解决办法是过期时间加随机抖动比如基础 600 秒加 0 到 120 秒的随机值避免同时失效。import random def set_with_jitter(client, key, value, base_ttl600): ttl base_ttl random.randint(0, 120) client.setex(key, ttl, value)5.2 连接超时与命令超时排查redis command timed out这个报错我见过太多次。原因通常有几个连接池不够用、Redis 服务端阻塞、网络抖动、慢命令。排查步骤是这样的。先用redis-cli --latency看服务端延迟正常应该在亚毫秒级。再用INFO clients看连接数如果connected_clients接近maxclients说明连接不够。用SLOWLOG GET 10看有没有慢命令比如KEYS *、大 key 的HGETALL。如果是连接池问题调大max_connections。如果是慢命令优化命令或者拆分大 key。如果是网络问题检查 Agent 服务和 Redis 之间的网络质量。5.3 大 key 与热 key 治理大 key 是 Redis 性能杀手。一个 key 存了几 MB 甚至几十 MB读写都会阻塞其他请求。Agent 场景里常见于把整个会话历史塞进一个 key或者工具返回的大结果直接缓存。治理方法是拆分。会话历史按时间分片比如每 50 条消息一个 key。大结果压缩后再存或者只存摘要详细内容放对象存储。热 key 是访问频率极高的 key容易把单个 Redis 实例的 CPU 打满。治理方法是本地缓存加多级缓存或者把热 key 复制多份分散到不同实例。用redis-cli --bigkeys可以扫描大 key用redis-cli --hotkeys需要开启 LFU 淘汰策略才能用。5.4 常见问题速查表问题现象可能原因排查方法解决方案command timed out连接池不足INFO clients调大 max_connections响应时间抖动慢命令阻塞SLOWLOG GET优化命令拆分大 key内存持续增长未设过期或泄漏INFO memory检查 TTL加淘汰策略缓存命中率低TTL 太短或 key 设计差INFO stats调整 TTL优化 key 结构主从数据不一致复制延迟INFO replication检查网络考虑读写分离策略锁不释放进程崩溃或未设 TTL检查锁 key必须设过期时间6. 几个我踩过的坑和实操心得第一个坑是序列化格式不兼容。早期用 JSON 存后来换 MessagePack结果老数据读不出来线上直接报错。教训是切换序列化格式一定要做版本兼容或者干脆换 key 前缀让老数据自然过期。第二个坑是过期时间设成固定值。所有 key 都是 3600 秒结果整点批量失效数据库瞬间被打爆。后来加了随机抖动才解决。这个坑很隐蔽低并发时完全看不出来。第三个坑是分布式锁没设 TTL。有一次进程被 kill锁没释放后续所有任务都卡住。加了 TTL 之后虽然解决了但又要考虑业务执行时间超过 TTL 的情况需要加续期机制。第四个坑是用 KEYS 命令遍历。开发时图方便用KEYS agent:*测试环境没事线上数据量一大直接把 Redis 阻塞了好几秒。后来全部换成SCAN渐进式遍历。第五个坑是连接池没设超时。Redis 网络抖动时请求一直挂着不返回把 Agent 的线程池占满。加了 socket_timeout 之后快速失败快速重试整体可用性反而更高。实操心得方面我建议 Agent 的缓存 key 一定要有统一的命名规范比如agent:{模块}:{业务}:{ID}方便排查和批量清理。监控一定要上命中率、延迟、连接数、内存使用率这几个指标要盯着。还有缓存逻辑一定要有降级方案Redis 挂了不能让整个 Agent 挂掉可以降级到直接查数据库或者返回默认值。最后分享一个小技巧Agent 的工具调用结果缓存可以在 key 里带上模型版本或者 Prompt 版本。这样 Prompt 改了之后旧缓存自动失效不用手动清理。这个细节能省很多事。 AI Agent 接入 Redis 缓存实战：高并发架构设计与性能优化
1. AI Agent 接入 Redis 缓存从零搭建到扛住并发的实战拆解AI Agent 这个方向最近一年有多热不用我多说。但真正上手搭过 Agent 的人都知道一个能跑起来的 Demo 和一个能扛住真实流量的系统之间差的不只是模型能力更多是工程层面的东西。其中 Redis 缓存就是最容易被忽视、却又最能决定 Agent 响应速度和并发上限的一环。我自己从去年开始陆续搭过几个 AI Agent 项目有基于 Python 的也有用 Rust 重写的踩过的坑从连接池打满到缓存雪崩从序列化格式选错到分布式锁死锁基本都经历了一遍。这篇就把 AI Agent 和 Redis 缓存结合这件事从架构设计到实操落地再到线上问题排查完整拆一遍。不管你是刚接触 AI Agent 搭建的新手还是已经在做线上缓存治理的老手应该都能从里面找到能直接抄作业的东西。核心就一句话AI Agent 的瓶颈往往不在模型推理而在数据层的读写效率Redis 缓存是解决这个问题的第一把刀但用不好它自己就会变成新的瓶颈。2. 为什么 AI Agent 必须配 Redis 缓存2.1 Agent 的请求链路到底慢在哪先搞清楚一个 AI Agent 处理一次用户请求时间都花在哪了。典型链路是这样的接收用户输入做意图识别调用工具或检索知识库拼装 Prompt请求大模型拿到结果后做后处理最后返回。这里面大模型推理通常占大头几百毫秒到几秒不等但除了模型本身还有大量重复性的数据操作在拖后腿。比如多轮对话场景每一轮都要把历史消息读出来拼进上下文。如果每次都从数据库或者对象存储里拉光是 IO 就够呛。再比如工具调用Agent 经常需要查用户信息、查配置、查知识库片段这些数据变化频率低但读取频率极高。还有会话状态、限流计数、幂等标记这些都是典型的缓存场景。我实测过一个对话 Agent在没加缓存的情况下单次请求里数据库相关操作累计耗时能到 300 到 500 毫秒加了 Redis 之后压到 20 毫秒以内。这个差距在低并发时感知不明显一旦 QPS 上去就是能不能扛住的问题。2.2 Redis 在 Agent 架构里的四个核心角色很多人一提 Redis 就只想到缓存其实在 AI Agent 体系里它至少承担四个角色每个角色的配置策略都不一样。第一个是会话上下文缓存。多轮对话的历史消息、用户偏好、当前任务状态这些数据读写频繁、生命周期跟会话绑定非常适合放 Redis。用 Hash 结构存一个会话一个 key字段存消息列表和元数据设置合理的过期时间。第二个是工具调用结果缓存。Agent 调用的很多工具是幂等的比如查天气、查汇率、查知识库同样的输入短时间内结果不会变。这类结果缓存起来能大幅减少外部 API 调用既省钱又提速。第三个是限流与配额计数。Agent 服务通常要对用户或 API Key 做限流Redis 的原子递增和过期机制天然适合做计数器。用 INCR 加 EXPIRE或者用 Lua 脚本保证原子性。第四个是分布式锁与幂等控制。多个 Agent 实例同时处理同一个任务时需要互斥。比如同一个用户同时发了两条消息要保证不会重复触发某个副作用操作这时候分布式锁就派上用场了。2.3 不接缓存会踩的三个坑不接缓存最直接的后果就是数据库压力大。Agent 的读放大特别严重一次用户请求可能触发十几次数据读取数据库连接池很容易被打满然后就是经典的redis command timed out或者数据库连接超时。第二个坑是响应时间不稳定。数据库的 P99 延迟远高于 Redis一旦有慢查询或者锁竞争Agent 的响应时间就会抖动用户体验很差。而 Redis 是内存操作延迟稳定在亚毫秒级。第三个坑是成本。大模型调用本身按 token 计费如果因为上下文拼装慢导致重试或者因为工具调用重复执行成本会悄悄涨上去。缓存能砍掉大量重复计算和重复调用。3. Redis 缓存方案选型与核心配置3.1 数据结构怎么选别拿 String 存一切Redis 的数据类型选对了性能和可维护性差一大截。我见过太多项目所有东西都用 String 存 JSON能用是能用但浪费内存也浪费操作效率。会话上下文用Hash最合适。一个会话一个 key消息列表、用户信息、任务状态作为不同 field。好处是可以单独更新某个 field不用整体读写。比如只更新任务状态时用 HSET 就行不用把整个会话读出来改完再写回去。工具调用结果缓存用String存序列化后的结果就行key 设计成tool:{tool_name}:{input_hash}简单直接。如果结果比较大可以考虑压缩后再存。限流计数用String配合 INCR 和 EXPIRE或者直接用 Redis 的滑动窗口方案。计数场景不需要复杂结构原子操作才是关键。排行榜或者需要排序的场景用Sorted Set。比如 Agent 的任务优先级队列用 ZADD 加分数ZRANGE 按优先级取任务。去重场景用Set或者Bitmap。比如判断某个用户今天是否已经触发过某个操作用 Set 存用户 ID或者用 Bitmap 更省内存。3.2 序列化格式JSON 还是 MessagePack序列化格式直接影响存储体积和序列化开销。JSON 可读性好、调试方便但体积大、解析慢。MessagePack 体积小、速度快但可读性差。Protobuf 更极致但需要定义 schema改起来麻烦。我的建议是开发调试阶段用 JSON线上对性能敏感的场景换 MessagePack。Python 里用msgpack库Rust 里用rmp-serde切换成本不高。实测下来 MessagePack 比 JSON 体积小 30% 到 50%序列化速度快 2 到 3 倍。如果数据里有大量重复的字段名可以考虑把字段名映射成短码再序列化进一步压缩体积。不过这会牺牲可读性看团队取舍。3.3 过期策略与内存淘汰过期时间设置是个技术活。设太短缓存频繁失效等于没缓存设太长数据陈旧还可能把内存撑爆。会话上下文一般设 30 分钟到 2 小时跟业务场景有关。工具调用结果看数据变化频率天气类设 10 分钟汇率类设 1 分钟知识库类可以设几小时甚至一天。内存淘汰策略要选对。Agent 场景下我一般用allkeys-lru让 Redis 在内存满时淘汰最近最少使用的 key。如果缓存的数据重要性有区分可以用volatile-lru只淘汰设置了过期时间的 key把永久数据保护起来。注意千万不要用noeviction内存满了之后写操作直接报错Agent 会大面积失败。这个坑我踩过线上报警响成一片。3.4 连接池配置别让连接成为瓶颈连接池配置不好高并发下会出现连接等待甚至超时。Python 的 redis-py 用ConnectionPoolRust 的 redis-rs 用ConnectionManager或者连接池。关键参数是最大连接数。设太小请求排队设太大Redis 服务端压力大。经验值是最大连接数设为预期 QPS 的 1.5 到 2 倍但不要超过 Redis 的maxclients配置。比如预期 1000 QPS每个请求平均占用连接 5 毫秒那并发连接数大概是 5但考虑到突发流量设 50 到 100 比较稳妥。还要设置合理的超时时间。连接超时和读写超时都要设一般 1 到 3 秒。超时太长会拖垮整个请求链路太短会误杀正常请求。4. AI Agent 缓存实操从搭建到跑通4.1 环境准备与 Redis 安装先搞定 Redis。Linux 上用包管理器装最省事macOS 用 HomebrewWindows 建议用 Docker。生产环境强烈建议用 Docker 或者直接上云服务别自己编译。# macOS 安装 brew install redis brew services start redis # Docker 安装单机 docker run -d --name redis -p 6379:6379 redis:7-alpine # Docker 安装主从 docker run -d --name redis-master -p 6379:6379 redis:7-alpine docker run -d --name redis-slave -p 6380:6379 redis:7-alpine redis-server --slaveof host.docker.internal 6379装完用redis-cli连上去测一下ping返回PONG就通了。可视化客户端可以用 RedisInsight 或者 Another Redis Desktop Manager调试的时候看 key 很方便。4.2 Python 侧接入示例Python 是 AI Agent 最常用的语言redis-py 是标配。下面是一个带连接池和序列化的封装示例。import redis import msgpack import hashlib from typing import Any, Optional class AgentCache: def __init__(self, hostlocalhost, port6379, db0, max_connections100): self.pool redis.ConnectionPool( hosthost, portport, dbdb, max_connectionsmax_connections, socket_timeout3, socket_connect_timeout2, decode_responsesFalse ) self.client redis.Redis(connection_poolself.pool) def _serialize(self, value: Any) - bytes: return msgpack.packb(value, use_bin_typeTrue) def _deserialize(self, data: bytes) - Any: return msgpack.unpackb(data, rawFalse) def get_session(self, session_id: str) - Optional[dict]: key fagent:session:{session_id} data self.client.hgetall(key) if not data: return None return {k.decode(): self._deserialize(v) for k, v in data.items()} def set_session_field(self, session_id: str, field: str, value: Any, ttl: int 3600): key fagent:session:{session_id} pipe self.client.pipeline() pipe.hset(key, field, self._serialize(value)) pipe.expire(key, ttl) pipe.execute() def cache_tool_result(self, tool_name: str, input_data: Any, result: Any, ttl: int 600): input_hash hashlib.md5(str(input_data).encode()).hexdigest() key fagent:tool:{tool_name}:{input_hash} self.client.setex(key, ttl, self._serialize(result)) def get_tool_result(self, tool_name: str, input_data: Any) - Optional[Any]: input_hash hashlib.md5(str(input_data).encode()).hexdigest() key fagent:tool:{tool_name}:{input_hash} data self.client.get(key) return self._deserialize(data) if data else None这个封装里几个点值得说。用 pipeline 把 HSET 和 EXPIRE 打包发送减少网络往返。序列化用 MessagePack体积和速度都优于 JSON。key 命名用冒号分隔方便用SCAN按前缀遍历。4.3 Rust 侧接入示例如果 Agent 用 Rust 写redis-rs 是主流选择。Rust 的异步生态配合连接池性能很能打。use redis::AsyncCommands; use redis::aio::ConnectionManager; use serde::{Serialize, Deserialize}; #[derive(Serialize, Deserialize)] struct SessionData { messages: VecString, state: String, } pub struct AgentCache { conn: ConnectionManager, } impl AgentCache { pub async fn new(redis_url: str) - redis::RedisResultSelf { let client redis::Client::open(redis_url)?; let conn ConnectionManager::new(client).await?; Ok(Self { conn }) } pub async fn set_session(mut self, session_id: str, data: SessionData, ttl: u64) - redis::RedisResult() { let key format!(agent:session:{}, session_id); let value serde_json::to_string(data).unwrap(); let _: () self.conn.set_ex(key, value, ttl).await?; Ok(()) } pub async fn get_session(mut self, session_id: str) - redis::RedisResultOptionSessionData { let key format!(agent:session:{}, session_id); let value: OptionString self.conn.get(key).await?; Ok(value.and_then(|v| serde_json::from_str(v).ok())) } }Rust 这边用ConnectionManager而不是普通连接因为它自带重连和连接复用适合长时间运行的服务。序列化用 serde_json如果追求极致性能可以换 rmp-serde。4.4 分布式锁的正确用法Agent 处理任务时经常需要互斥比如同一个任务不能被两个实例同时执行。Redis 分布式锁是常见方案但坑很多。import uuid import time class RedisLock: def __init__(self, client, key: str, ttl: int 30): self.client client self.key fagent:lock:{key} self.ttl ttl self.token str(uuid.uuid4()) def acquire(self, timeout: int 10) - bool: end time.time() timeout while time.time() end: if self.client.set(self.key, self.token, nxTrue, exself.ttl): return True time.sleep(0.05) return False def release(self): lua if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end self.client.eval(lua, 1, self.key, self.token)关键点有三个。第一value 必须用唯一 token释放时校验防止误删别人的锁。第二释放锁必须用 Lua 脚本保证原子性先 get 再 del 会有竞态。第三一定要设过期时间防止持锁进程崩溃后锁永远不释放。提示如果业务对锁的可靠性要求极高考虑 Redlock 算法或者直接用 etcd、ZooKeeper。Redis 单机锁在主从切换时可能丢锁这个风险要评估。5. 高并发场景下的缓存治理与问题排查5.1 缓存穿透、击穿、雪崩的应对这三个词面试常问但真正在 Agent 场景里踩过才知道疼。缓存穿透是查一个不存在的数据缓存和数据库都没有每次请求都打到数据库。Agent 场景里常见于查不存在的用户或会话。解决办法是缓存空值设一个较短的过期时间比如 60 秒。或者用布隆过滤器提前拦截。缓存击穿是某个热点 key 过期瞬间大量请求同时打到数据库。Agent 里常见于热门工具的结果缓存。解决办法是热点 key 不过期或者用互斥锁保证只有一个请求去回源其他请求等待。缓存雪崩是大量 key 同时过期数据库瞬间压力暴增。解决办法是过期时间加随机抖动比如基础 600 秒加 0 到 120 秒的随机值避免同时失效。import random def set_with_jitter(client, key, value, base_ttl600): ttl base_ttl random.randint(0, 120) client.setex(key, ttl, value)5.2 连接超时与命令超时排查redis command timed out这个报错我见过太多次。原因通常有几个连接池不够用、Redis 服务端阻塞、网络抖动、慢命令。排查步骤是这样的。先用redis-cli --latency看服务端延迟正常应该在亚毫秒级。再用INFO clients看连接数如果connected_clients接近maxclients说明连接不够。用SLOWLOG GET 10看有没有慢命令比如KEYS *、大 key 的HGETALL。如果是连接池问题调大max_connections。如果是慢命令优化命令或者拆分大 key。如果是网络问题检查 Agent 服务和 Redis 之间的网络质量。5.3 大 key 与热 key 治理大 key 是 Redis 性能杀手。一个 key 存了几 MB 甚至几十 MB读写都会阻塞其他请求。Agent 场景里常见于把整个会话历史塞进一个 key或者工具返回的大结果直接缓存。治理方法是拆分。会话历史按时间分片比如每 50 条消息一个 key。大结果压缩后再存或者只存摘要详细内容放对象存储。热 key 是访问频率极高的 key容易把单个 Redis 实例的 CPU 打满。治理方法是本地缓存加多级缓存或者把热 key 复制多份分散到不同实例。用redis-cli --bigkeys可以扫描大 key用redis-cli --hotkeys需要开启 LFU 淘汰策略才能用。5.4 常见问题速查表问题现象可能原因排查方法解决方案command timed out连接池不足INFO clients调大 max_connections响应时间抖动慢命令阻塞SLOWLOG GET优化命令拆分大 key内存持续增长未设过期或泄漏INFO memory检查 TTL加淘汰策略缓存命中率低TTL 太短或 key 设计差INFO stats调整 TTL优化 key 结构主从数据不一致复制延迟INFO replication检查网络考虑读写分离策略锁不释放进程崩溃或未设 TTL检查锁 key必须设过期时间6. 几个我踩过的坑和实操心得第一个坑是序列化格式不兼容。早期用 JSON 存后来换 MessagePack结果老数据读不出来线上直接报错。教训是切换序列化格式一定要做版本兼容或者干脆换 key 前缀让老数据自然过期。第二个坑是过期时间设成固定值。所有 key 都是 3600 秒结果整点批量失效数据库瞬间被打爆。后来加了随机抖动才解决。这个坑很隐蔽低并发时完全看不出来。第三个坑是分布式锁没设 TTL。有一次进程被 kill锁没释放后续所有任务都卡住。加了 TTL 之后虽然解决了但又要考虑业务执行时间超过 TTL 的情况需要加续期机制。第四个坑是用 KEYS 命令遍历。开发时图方便用KEYS agent:*测试环境没事线上数据量一大直接把 Redis 阻塞了好几秒。后来全部换成SCAN渐进式遍历。第五个坑是连接池没设超时。Redis 网络抖动时请求一直挂着不返回把 Agent 的线程池占满。加了 socket_timeout 之后快速失败快速重试整体可用性反而更高。实操心得方面我建议 Agent 的缓存 key 一定要有统一的命名规范比如agent:{模块}:{业务}:{ID}方便排查和批量清理。监控一定要上命中率、延迟、连接数、内存使用率这几个指标要盯着。还有缓存逻辑一定要有降级方案Redis 挂了不能让整个 Agent 挂掉可以降级到直接查数据库或者返回默认值。最后分享一个小技巧Agent 的工具调用结果缓存可以在 key 里带上模型版本或者 Prompt 版本。这样 Prompt 改了之后旧缓存自动失效不用手动清理。这个细节能省很多事。