欢迎关注专栏：CSDN：AI 开发技术

• 上一篇：Redis 核心数据结构速查与终端演练
• 下一篇：系统架构与数据库理论基础：PostgreSQL 与 pgvector

大模型 Agent 开发实战：Redis 缓存与锁的高阶应用

关键词：Agent开发, LLM 缓存, 记忆滑动窗口, 分布式互斥锁, Python
阅读耗时：约 15 分钟

1. 业务背景 / 核心问题

在大语言模型（LLM）与 Agent 应用中，Redis 常用于响应缓存、短期会话状态和跨进程协调。它适合保存可过期、可重建的数据，但通常不应取代关系型数据库中的最终业务记录。

本文聚焦三种常见工程场景，并提供可独立理解的 Python 示例。示例假设 Redis 客户端启用了decode_responses=True，因此读取结果直接是字符串。

本地验证思路

执行 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:float=0.0,ttl_seconds:int=3600,)->str:cache_input=json.dumps({"prompt":prompt,"model":model,"temperature":temperature,},ensure_ascii=False,sort_keys=True,separators=(",",":"),)digest=hashlib.sha256(cache_input.encode("utf-8")).hexdigest()cache_key=f"llm_cache:{digest}"# 尝试命中缓存cached_res=self.r.get(cache_key)ifcached_resisnotNone:returnf"[Cache Hit]{cached_res}"# 模拟耗时的 LLM API 网络调用response=f"Simulated 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:int=10,ttl_seconds:int=86400,):key=f"chat_history:{session_id}"message=json.dumps({"role":role,"content":content},ensure_ascii=False)withself.r.pipeline(transaction=True)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:key=f"chat_history:{session_id}"# LRANGE 0 -1 获取整个列表全部记录返回给模型raw_history=self.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:int=30):lock_key=f"agent_lock:{lock_name}"lock_value=secrets.token_hex(16)acquired=bool(self.r.set(lock_key,lock_value,nx=True,ex=lock_timeout))try:yieldacquiredfinally:ifacquired:self.r.eval(RELEASE_LOCK_SCRIPT,1,lock_key,lock_value)

命令行验证思路：
在第一个锁仍有效时再次竞争同名锁，第二次通常会失败。测试还应覆盖“任务执行超过锁超时时间”的情况；长任务需要续租、栅栏令牌或更成熟的协调方案，不能仅靠固定 TTL 断言绝对互斥。Redis 官方文档也要求使用唯一随机值并通过比较后删除来安全释放锁：Distributed Locks with Redis。

5. 总结

关系型数据库通常负责最终业务记录，Redis 更适合承担可过期缓存、短期状态和跨进程协调。本文三个示例展示了常见起点，但上线前仍需补充容量评估、敏感数据保护、故障恢复、可观测性和并发测试。