欢迎关注专栏CSDNAI 开发技术上一篇Redis 核心数据结构速查与终端演练下一篇系统架构与数据库理论基础PostgreSQL 与 pgvector大模型 Agent 开发实战Redis 缓存与锁的高阶应用大模型 Agent 开发实战Redis 缓存与锁的高阶应用1. 业务背景 / 核心问题本地验证思路2. 场景一LLM 响应请求缓存 (Response Caching)3. 场景二Agent 对话记忆滑动窗口 (Chat History)4. 场景三分布式互斥锁 (Distributed Lock)5. 总结大模型 Agent 开发实战Redis 缓存与锁的高阶应用关键词Agent开发, LLM 缓存, 记忆滑动窗口, 分布式互斥锁, Python阅读耗时约 15 分钟1. 业务背景 / 核心问题在大语言模型LLM与 Agent 应用中Redis 常用于响应缓存、短期会话状态和跨进程协调。它适合保存可过期、可重建的数据但通常不应取代关系型数据库中的最终业务记录。本文聚焦三种常见工程场景并提供可独立理解的 Python 示例。示例假设 Redis 客户端启用了decode_responsesTrue因此读取结果直接是字符串。本地验证思路执行 Python 脚本后终端会输出对应的测试结果涵盖了 LLM 缓存命中、历史记录滑动窗口截断以及分布式任务锁的防重复触发等验证可以在测试环境使用SCAN 0 MATCH llm_cache:* COUNT 100和SCAN 0 MATCH chat_history:* COUNT 100检查示例键。不要在大型生产实例中使用KEYS *因为它会一次性遍历键空间。2. 场景一LLM 响应请求缓存 (Response Caching)业务痛点当模型、系统提示词和采样参数完全相同时重复请求可能产生额外 Token 成本和网络延迟。核心解决机制对所有影响输出的参数进行稳定序列化再用 SHA-256 生成缓存键通过 Redis 的过期写入限制缓存生命周期。核心逻辑代码importhashlibimportjsondefget_llm_response(self,prompt:str,model:str,temperature:float0.0,ttl_seconds:int3600,)-str:cache_inputjson.dumps({prompt:prompt,model:model,temperature:temperature,},ensure_asciiFalse,sort_keysTrue,separators(,,:),)digesthashlib.sha256(cache_input.encode(utf-8)).hexdigest()cache_keyfllm_cache:{digest}# 尝试命中缓存cached_resself.r.get(cache_key)ifcached_resisnotNone:returnf[Cache Hit]{cached_res}# 模拟耗时的 LLM API 网络调用responsefSimulated LLM response for:{prompt}# 写入缓存并设置过期时间self.r.setex(cache_key,ttl_seconds,response)returnf[API Call]{response}Python 内置的hash()默认会在不同进程间随机化不能用作跨进程稳定缓存键。缓存键还必须纳入模型版本、系统提示词、工具配置和采样参数等真正影响结果的输入。对于非确定性生成或包含敏感信息的请求还要进一步评估是否应该缓存。命令行验证思路通过GET查询为空时触发网络调用再通过SETEX指定秒数缓存结果。3. 场景二Agent 对话记忆滑动窗口 (Chat History)业务痛点LLM 模型本身是无状态的必须把前置上下文一并发送。但上下文 Token 有上限且可能增加调用成本因此必须对早期的历史内容进行截断。核心解决机制利用 Redis List通过RPUSH追加新消息再用LTRIM保留最近若干条记录。通过事务管道把追加、截断和设置过期时间作为一个连续操作提交。核心逻辑代码defadd_chat_memory(self,session_id:str,role:str,content:str,max_history:int10,ttl_seconds:int86400,):keyfchat_history:{session_id}messagejson.dumps({role:role,content:content},ensure_asciiFalse)withself.r.pipeline(transactionTrue)aspipe:pipe.rpush(key,message)pipe.ltrim(key,-max_history,-1)pipe.expire(key,ttl_seconds)pipe.execute()defget_chat_memory(self,session_id:str)-list:keyfchat_history:{session_id}# LRANGE 0 -1 获取整个列表全部记录返回给模型raw_historyself.r.lrange(key,0,-1)return[json.loads(msg)formsginraw_history]这里限制的是消息条数并不等于严格限制模型 Token。生产应用通常还需要在调用模型前按实际 tokenizer 计算预算并对超出窗口的历史做摘要或淘汰。命令行验证思路推送两条记录后执行LTRIM进行截断修剪。4. 场景三分布式互斥锁 (Distributed Lock)业务痛点当同一任务可能被多个 Agent 实例同时触发时需要减少重复执行与资源冲突。核心解决机制使用SET NX EX竞争带过期时间的锁并为每个持有者生成唯一随机值。释放锁时必须原子地“先比对持有者再删除”避免旧任务误删后来者重新获得的锁。核心逻辑代码fromcontextlibimportcontextmanagerimportsecrets RELEASE_LOCK_SCRIPT if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) end return 0 contextmanagerdefacquire_lock(self,lock_name:str,lock_timeout:int30):lock_keyfagent_lock:{lock_name}lock_valuesecrets.token_hex(16)acquiredbool(self.r.set(lock_key,lock_value,nxTrue,exlock_timeout))try:yieldacquiredfinally:ifacquired:self.r.eval(RELEASE_LOCK_SCRIPT,1,lock_key,lock_value)命令行验证思路在第一个锁仍有效时再次竞争同名锁第二次通常会失败。测试还应覆盖“任务执行超过锁超时时间”的情况长任务需要续租、栅栏令牌或更成熟的协调方案不能仅靠固定 TTL 断言绝对互斥。Redis 官方文档也要求使用唯一随机值并通过比较后删除来安全释放锁Distributed Locks with Redis。5. 总结关系型数据库通常负责最终业务记录Redis 更适合承担可过期缓存、短期状态和跨进程协调。本文三个示例展示了常见起点但上线前仍需补充容量评估、敏感数据保护、故障恢复、可观测性和并发测试。