1. 从一次线上故障说起为什么我开始思考Agent的记忆存储去年我们团队上线了一个面向企业内部的智能客服Agent。这个Agent的核心能力是处理复杂的、多轮次的业务咨询比如员工报销政策查询、IT设备申领流程跟进等。在最初的架构里我们采用了最“直觉”的方案将对话的上下文也就是Agent的“短期记忆”直接序列化成JSON字符串然后随着每一次用户请求完整地来回传递。简单来说就是把整个对话历史都塞进了每次API调用的请求体或响应体里。这个方案在开发和测试阶段运行得相当顺畅。直到某一天运营部门发起了一场全员政策宣贯活动咨询量瞬间激增。我们监控到核心的对话处理接口响应时间从平均200ms飙升到了2秒以上错误率也开始攀升。紧急排查后问题直指那个不断膨胀的“上下文包袱”。我们很快定位到几个核心痛点请求体膨胀一个典型的报销流程咨询可能涉及10轮以上的问答。每轮对话都包含用户query、Agent的思考过程、工具调用结果和最终回复。几轮下来这个上下文JSON的体积轻松超过20KB。当这个“包袱”在客户端、网关、后端服务之间反复传输时网络I/O和序列化/反序列化的开销变得不可忽视。令牌数Token的浪费与成本我们的Agent后端接入了大语言模型LLMAPI。LLM的计费通常基于输入和输出的总令牌数。每次请求我们都需要把完整的对话历史作为“系统提示词”或“上下文”的一部分喂给LLM。这意味着对话轮次越多我们为重复传输的、早已固定的历史信息所支付的费用就越高。这是一笔随着对话长度线性增长的、纯属浪费的成本。状态管理的混乱当用户同时从多个设备如PC网页和手机App发起会话时或者当服务因部署需要重启时维护对话状态的一致性成了噩梦。内存中的上下文在服务重启后会丢失而依赖客户端传递状态又不可靠。这次故障让我们彻底反思将短期记忆上下文作为请求/响应的载荷Payload来传递是一个在架构上存在根本缺陷的方案。它混淆了“数据”和“状态”的边界将本应由服务端持久化管理的会话状态退化成了客户端需要关心的传输细节。这之后我们开始系统性地寻找解决方案而Redis以其独特的优势成为了我们架构改造的核心支柱。2. 拆解Agent的“记忆”什么是短期记忆为什么它如此关键在深入技术选型之前我们必须先厘清概念。在一个多轮对话Agent中“记忆”通常被分为长期记忆Long-term Memory和短期记忆Short-term Memory。长期记忆更像是一个知识库或用户档案。它存储的是跨会话的、相对稳定的信息例如用户的偏好、历史订单摘要、产品知识等。这类数据通常存储在关系型数据库如MySQL、PostgreSQL或向量数据库如Milvus、Pinecone中访问频率较低但要求结构化和可关联查询。短期记忆特指当前会话的上下文Context。它是一系列按时间顺序排列的交互记录构成了本次对话的“短期工作记忆”。这是LLM理解当前query、保持对话连贯性的唯一依据。短期记忆的具体内容通常是一个消息Message列表每个消息可能包含以下字段{ “role”: “user”, // 或 “assistant”, “system”, “tool” “content”: “我想查询一下北京飞往上海的航班”, “timestamp”: 1715167890, // 可能还有工具调用、函数执行结果等元数据 }短期记忆的核心价值与挑战在于它是对话连贯性的生命线没有它Agent就是“金鱼”每一轮对话都是独立的无法引用上文体验支离破碎。它体积增长快但生命周期短一个会话的短期记忆从对话开始到结束可能几分钟到几小时会不断增长但一旦会话结束其价值就急剧下降可以被安全地清理。它需要极低的读写延迟每次用户发言Agent都需要即时读取历史上下文生成响应后又要立即写入新的记录。这个过程的延迟必须远低于用户可感知的等待时间通常要求100ms。它需要高并发访问一个活跃的Agent服务可能同时处理成千上万个会话每个会话的读写操作都要求高度隔离且高效。理解了这些特性我们就能明白短期记忆的存储方案必须满足几个刚性指标超高速读写、支持高并发、具备TTL生存时间自动过期能力、以及能够方便地存储半结构化的会话数据。而Redis几乎是为这个场景量身定做的。3. Redis vs. 传统方案为什么数据库和纯内存缓存都不够好在决定使用Redis之前我们当然评估过其他方案。让我们来逐一分析为什么它们不适合作为Agent短期记忆的主存储。方案一关系型数据库如MySQL这似乎是存储数据的“正统”选择。我们可以设计一张conversation_messages表用session_id和turn_index来组织数据。劣势读写延迟过高即使对session_id建立索引每次读取一个完整会话的历史比如最近的20条记录仍然涉及磁盘I/O和复杂的查询组装。在对话高峰期这很容易成为瓶颈。写入开销大每一轮对话都需要执行一次INSERT操作。对于高频交互的Agent这会给数据库带来巨大的写入压力。清理成本高过期会话的清理需要定时任务执行DELETE可能产生表碎片影响性能。不适合非结构化数据对话消息中的content字段可能很长且结构多变用TEXT类型存储不利于快速解析。方案二服务进程的纯内存缓存如Pythondict这是最简单的方案将上下文存储在服务进程的内存字典里以session_id为键。劣势无状态服务的灾难现代云原生架构下服务通常是多实例、无状态的。用户的下一次请求可能被负载均衡到另一个实例导致上下文丢失。内存限制与扩容所有会话数据都堆在单个实例内存中容量有限且无法通过水平扩展来增加总内存容量。服务重启数据丢失任何部署更新或故障重启都会导致所有进行中的会话中断。方案三外部集中式缓存/数据库如Memcached、MongoDBMemcached它简单快速但数据结构单一主要支持字符串缺乏Redis丰富的哈希Hash、列表List、有序集合ZSet等结构。更重要的是它不支持数据持久化虽然短期记忆不需要强持久化但服务重启导致缓存雪崩也是个风险而且集群功能相对较弱。MongoDB作为文档数据库它存储JSON数据非常自然。但其读写性能尤其是写入在同等资源下通常低于Redis且更侧重于海量数据存储和复杂查询对于我们这个超高频、简单Key-Value访问的场景有点“杀鸡用牛刀”运维成本也更高。Redis的胜出点性能王者基于内存的操作读写速度在亚毫秒级完全满足Agent对延迟的苛刻要求。丰富的数据结构Hash结构可以完美存储一个会话的所有元信息创建时间、用户ID等和消息列表List或ZSet可以按顺序存储消息。这比将所有数据序列化成一个大字符串如用Memcached要高效和直观得多。内置TTL可以为每个会话的Key设置过期时间例如24小时。会话结束后数据自动被清理无需额外开发维护清理任务。高可用与持久化通过Redis Sentinel或Cluster模式可以实现高可用。虽然短期记忆对数据丢失不敏感但可选的RDB/AOF持久化机制能在服务重启时避免所有会话“归零”提供更好的用户体验。原子操作与Lua脚本支持对数据结构的原子性操作甚至可以通过Lua脚本实现复杂的“读取-修改-写入”逻辑保证在并发环境下的数据一致性。注意这里并不是说Redis是万能的。对于Agent的长期记忆知识库向量数据库依然是不可替代的选择。Redis的核心定位是解决短期、高频、临时的状态存储问题。4. 实战架构如何用Redis设计一个高效的Agent短期记忆模块理论说完了我们来点实际的。下面我将分享我们目前在生产环境中使用的架构设计包含关键的数据结构定义和核心操作逻辑。4.1 数据结构设计我们为每个会话Session在Redis中主要使用两种数据结构1. 会话元信息哈希HashKey:agent:session:{session_id}Value (Hash Field):user_id: 用户标识created_at: 会话创建时间戳updated_at: 最后活动时间戳ttl: 预设的过期时间秒metadata: 其他自定义元数据如渠道来源、语言等的JSON字符串这个Hash存储了会话的静态和描述性信息。2. 消息列表ListKey:agent:session:{session_id}:messagesValue: 一个列表每个元素是一条消息的JSON字符串。 例如[“{‘role’:’user’, ‘content’:’…’, ‘ts’:…}”, “{‘role’:’assistant’, …}”, …]我们选择List而不是ZSet因为对话消息天然是顺序追加的List的RPUSH右端推入和LRANGE范围读取操作效率极高且能完美保持顺序。ZSet更适合需要按分数如时间戳排序和范围查询的场景对于我们简单的追加读取来说稍显冗余。4.2 核心操作流程与代码示例Python假设我们使用redis-py客户端。初始化一个会话import redis import json import time redis_client redis.Redis(host‘localhost’, port6379, decode_responsesTrue) def create_session(user_id, session_id, initial_system_promptNone, ttl3600): “””创建新会话并可选地添加系统提示词””” session_key f“agent:session:{session_id}” msg_key f“{session_key}:messages” # 使用pipeline减少网络往返 pipe redis_client.pipeline() # 1. 设置会话元信息 pipe.hset(session_key, mapping{ ‘user_id’: user_id, ‘created_at’: time.time(), ‘updated_at’: time.time(), ‘ttl’: ttl }) # 设置这个Hash键的过期时间 pipe.expire(session_key, ttl) # 2. 初始化消息列表并设置同样的过期时间 if initial_system_prompt: system_msg json.dumps({‘role’: ‘system’, ‘content’: initial_system_prompt, ‘ts’: time.time()}) pipe.rpush(msg_key, system_msg) pipe.expire(msg_key, ttl) # 执行所有命令 pipe.execute() return session_id处理一轮用户输入并获取完整上下文 这是最核心的流程。注意我们需要在每次读写后更新会话的“最后活动时间”并刷新TTL滑动过期。def process_user_message(session_id, user_input): session_key f“agent:session:{session_id}” msg_key f“{session_key}:messages” # 1. 准备新消息 user_msg json.dumps({‘role’: ‘user’, ‘content’: user_input, ‘ts’: time.time()}) # 2. 使用事务MULTI/EXEC或Pipeline保证原子性 pipe redis_client.pipeline() # 2.1 将用户消息追加到列表 pipe.rpush(msg_key, user_msg) # 2.2 获取最近N条历史消息例如最近20条避免上下文过长 # 注意LLEN获取长度LRANGE获取内容。这里我们先获取在管道中执行。 # 更严谨的做法是用Lua脚本但为清晰起见这里分两步。 current_length redis_client.llen(msg_key) start_index max(0, current_length - 20) # 假设我们只保留最近20条作为上下文 pipe.lrange(msg_key, start_index, -1) # 获取从start_index到末尾的所有消息 # 2.3 更新会话的最后活动时间并刷新过期时间(TTL) pipe.hset(session_key, ‘updated_at’, time.time()) pipe.expire(session_key, 3600) pipe.expire(msg_key, 3600) # 执行管道命令 pipe.execute() # 3. 解析历史消息 # pipe.execute()返回一个结果列表对应每个命令的结果。 # rpush的结果是列表的新长度我们不需要。 # lrange的结果是消息JSON字符串列表。 history_msgs_json pipe.execute()[1] # 注意索引根据命令顺序调整 history_messages [json.loads(msg) for msg in history_msgs_json] # 4. 此时history_messages就是包含最新用户输入的完整短期记忆。 # 你可以将其传递给LLM进行推理生成助手回复。 # … (调用LLM的逻辑) … # 5. 假设LLM返回了助手回复 assistant_response assistant_msg json.dumps({‘role’: ‘assistant’, ‘content’: assistant_response, ‘ts’: time.time()}) # 6. 将助手回复也写入记忆 redis_client.rpush(msg_key, assistant_msg) # 注意这里可以再次刷新TTL或者依赖下一次用户交互来刷新。 return assistant_response4.3 高级模式上下文窗口管理与摘要生成直接存储所有原始消息会遇到LLM的上下文长度限制。一个成熟的方案需要实现上下文窗口管理。当消息列表超过一定长度如Token数或条数时不能简单粗暴地截断最早的几条那可能会丢失关键信息。我们的策略是滚动摘要Rolling Summary。在Redis中为每个会话额外维护一个summary字段存储在会话的Hash里。当原始消息列表过长时触发一个摘要生成过程将最早的一部分消息例如前10轮连同当前的摘要初始为空一起发送给LLM要求其生成一个浓缩的段落来概括之前对话的核心事实和决定。用这个新的摘要更新Redis中的summary字段并从消息列表中物理删除那些已被摘要的消息。后续的对话上下文就由“最新摘要 未被摘要的近期原始消息”共同构成。这样我们既控制了上下文长度又保留了对话的长期脉络。这个摘要生成过程可以异步执行避免阻塞主对话流程。5. 避坑指南Redis实战中的性能、一致性与运维陷阱将短期记忆迁移到Redis带来了巨大的收益但也引入了新的复杂性。下面是我在实战中踩过的坑和总结的经验。5.1 连接管理与性能调优连接池是必须的不要为每个请求创建新的Redis连接。使用连接池redis.ConnectionPool是保证高性能、低资源消耗的基础。根据你的并发量合理设置max_connections。Pipeline批量操作如上文示例所示将多个相关命令如写入消息、读取历史、更新TTL放入一个Pipeline中执行能大幅减少网络往返延迟RTT。这在处理一轮对话时效果显著。避免大Key虽然一个会话的消息列表会增长但要警惕单个Key过大如超过10KB。过大的Key在序列化、网络传输和Redis自身内存分配中都会影响性能。这就是为什么我们需要实施上文提到的“上下文窗口管理”和“摘要”机制定期修剪列表长度。内存监控与告警Redis是内存数据库必须严密监控内存使用率。设置告警阈值如80%并提前规划好内存淘汰策略maxmemory-policy。对于Agent记忆这种场景allkeys-lru最近最少使用或volatile-lru在设置了过期时间的键中LRU是比较合适的选择。5.2 数据一致性与并发安全原子性操作像“读取-修改-写入”这样的复合操作在并发环境下可能出错。例如两个并发的请求同时读取了旧的消息列表然后分别追加了自己的消息会导致其中一条消息丢失。最优雅的解决方案是使用Lua脚本。Redis保证一个Lua脚本在执行时是原子的。append_message_script “”” local msg_key KEYS[1] local session_key KEYS[2] local new_msg ARGV[1] local new_ttl ARGV[2] -- 追加消息 redis.call(‘RPUSH’, msg_key, new_msg) -- 获取最新的N条消息例如从倒数第20条开始 local recent_msgs redis.call(‘LRANGE’, msg_key, -20, -1) -- 更新活动时间 redis.call(‘HSET’, session_key, ‘updated_at’, ARGV[3]) -- 刷新过期时间 redis.call(‘EXPIRE’, msg_key, new_ttl) redis.call(‘EXPIRE’, session_key, new_ttl) return recent_msgs “”” append_message_lua redis_client.register_script(append_message_script) # 调用脚本 recent_msgs append_message_lua(keys[msg_key, session_key], args[new_message_json, 3600, time.time()])会话过期与雪崩如果大量会话在同一时间点创建并设置了相同的固定TTL如1小时它们可能会在同一时间点大量过期导致Redis内存释放压力骤增。解决方案是为TTL添加一个随机抖动例如base_ttl random.randint(0, 300)让过期时间分散开。5.3 高可用与灾难恢复单点故障绝对不要使用单机Redis。生产环境必须部署Redis Sentinel主从哨兵或Redis Cluster集群模式。Sentinel模式提供了自动故障转移适合数据量不是特别巨大的场景。Cluster模式则提供了数据分片和线性扩展能力。持久化权衡短期记忆可以容忍少量丢失用户大不了从头开始对话。因此可以优先考虑性能将appendfsync配置为everysec甚至no并主要依赖RDB快照进行周期性持久化。这样能在性能和可靠性间取得平衡。备份与恢复策略定期备份RDB文件到对象存储如S3。制定恢复演练流程。虽然数据不重要但快速恢复服务的能力很重要。5.4 监控与可观测性你需要监控的关键指标包括Redis服务器级内存使用率、连接数、每秒操作数OPS、命中率、网络输入/输出流量。应用级会话读写平均延迟、P99延迟、会话创建成功率。业务级活跃会话数、平均会话时长、因记忆丢失导致的会话异常终止率。将这些指标接入你的APM如PrometheusGrafana和日志系统才能做到心中有数快速定位问题是出在Redis、网络还是应用代码。6. 超越基础当Agent遇到复杂会话与分布式扩展基本的读写模式解决了大部分问题但随着业务复杂化我们遇到了更高级的场景。场景一支持话题分支与回溯有些复杂的客服或游戏Agent对话可能存在分支比如用户问“回到上一个问题”。简单的列表无法高效处理。我们的优化是引入ZSet有序集合。用时间戳作为分数score消息ID或内容作为成员member。这样我们可以ZADD按时间戳插入消息。ZRANGEBYSCORE获取某个时间范围内的消息实现“回溯到10分钟前的对话状态”。结合ZREMRANGEBYSCORE清理旧消息。 这比用List模拟要高效和直观得多但代价是存储开销稍大。场景二超大规模下的分片与隔离当你的用户量达到千万级即使有Cluster管理数十亿个会话Key也可能有压力。我们采取了逻辑分片在session_id的生成规则中融入用户ID哈希值。根据哈希值的前几位将会话路由到不同的Redis Cluster分片甚至是不同的物理Redis集群例如按业务线或地理区域隔离。在应用层维护一个路由表或使用一致性哈希库。 这样做不仅分散了压力也实现了故障隔离一个集群的问题不会影响所有用户。场景三与向量数据库的协同在高级的Agent中短期记忆Redis和长期记忆向量数据库需要联动。例如当用户说“还记得我上次提过的关于XX的需求吗”Agent需要从短期记忆中感知到这是一个“回忆”指令。然后它需要从Redis中取出当前会话的user_id和关键实体信息。利用这些信息作为查询条件去向量数据库中检索该用户历史上相关的长期记忆片段。最后将检索到的长期记忆片段与Redis中的短期记忆上下文拼接一起送给LLM生成回复。 Redis在这里扮演了状态枢纽和元数据索引的角色桥接了即时对话和长期知识。从那次痛苦的线上故障到今天稳定支撑日均百万级对话的架构把Agent的短期记忆剥离出来放入Redis是我们做的最正确的技术决策之一。它不仅仅是一个存储组件的更换更是一种架构思想的转变让专业的数据结构去做专业的事让状态管理回归服务端让通信链路轻装上阵。这个过程没有银弹你需要根据自己Agent的对话复杂度、流量规模和对一致性的要求仔细设计Redis的数据结构、过期策略和访问模式。但无论如何相比于把上下文在网络上“搬来搬去”拥有一个高速、可靠、中心化的记忆存储无疑是构建健壮、可扩展的多轮对话Agent的基石。
基于Redis构建Agent短期记忆：解决多轮对话上下文存储与性能挑战
1. 从一次线上故障说起为什么我开始思考Agent的记忆存储去年我们团队上线了一个面向企业内部的智能客服Agent。这个Agent的核心能力是处理复杂的、多轮次的业务咨询比如员工报销政策查询、IT设备申领流程跟进等。在最初的架构里我们采用了最“直觉”的方案将对话的上下文也就是Agent的“短期记忆”直接序列化成JSON字符串然后随着每一次用户请求完整地来回传递。简单来说就是把整个对话历史都塞进了每次API调用的请求体或响应体里。这个方案在开发和测试阶段运行得相当顺畅。直到某一天运营部门发起了一场全员政策宣贯活动咨询量瞬间激增。我们监控到核心的对话处理接口响应时间从平均200ms飙升到了2秒以上错误率也开始攀升。紧急排查后问题直指那个不断膨胀的“上下文包袱”。我们很快定位到几个核心痛点请求体膨胀一个典型的报销流程咨询可能涉及10轮以上的问答。每轮对话都包含用户query、Agent的思考过程、工具调用结果和最终回复。几轮下来这个上下文JSON的体积轻松超过20KB。当这个“包袱”在客户端、网关、后端服务之间反复传输时网络I/O和序列化/反序列化的开销变得不可忽视。令牌数Token的浪费与成本我们的Agent后端接入了大语言模型LLMAPI。LLM的计费通常基于输入和输出的总令牌数。每次请求我们都需要把完整的对话历史作为“系统提示词”或“上下文”的一部分喂给LLM。这意味着对话轮次越多我们为重复传输的、早已固定的历史信息所支付的费用就越高。这是一笔随着对话长度线性增长的、纯属浪费的成本。状态管理的混乱当用户同时从多个设备如PC网页和手机App发起会话时或者当服务因部署需要重启时维护对话状态的一致性成了噩梦。内存中的上下文在服务重启后会丢失而依赖客户端传递状态又不可靠。这次故障让我们彻底反思将短期记忆上下文作为请求/响应的载荷Payload来传递是一个在架构上存在根本缺陷的方案。它混淆了“数据”和“状态”的边界将本应由服务端持久化管理的会话状态退化成了客户端需要关心的传输细节。这之后我们开始系统性地寻找解决方案而Redis以其独特的优势成为了我们架构改造的核心支柱。2. 拆解Agent的“记忆”什么是短期记忆为什么它如此关键在深入技术选型之前我们必须先厘清概念。在一个多轮对话Agent中“记忆”通常被分为长期记忆Long-term Memory和短期记忆Short-term Memory。长期记忆更像是一个知识库或用户档案。它存储的是跨会话的、相对稳定的信息例如用户的偏好、历史订单摘要、产品知识等。这类数据通常存储在关系型数据库如MySQL、PostgreSQL或向量数据库如Milvus、Pinecone中访问频率较低但要求结构化和可关联查询。短期记忆特指当前会话的上下文Context。它是一系列按时间顺序排列的交互记录构成了本次对话的“短期工作记忆”。这是LLM理解当前query、保持对话连贯性的唯一依据。短期记忆的具体内容通常是一个消息Message列表每个消息可能包含以下字段{ “role”: “user”, // 或 “assistant”, “system”, “tool” “content”: “我想查询一下北京飞往上海的航班”, “timestamp”: 1715167890, // 可能还有工具调用、函数执行结果等元数据 }短期记忆的核心价值与挑战在于它是对话连贯性的生命线没有它Agent就是“金鱼”每一轮对话都是独立的无法引用上文体验支离破碎。它体积增长快但生命周期短一个会话的短期记忆从对话开始到结束可能几分钟到几小时会不断增长但一旦会话结束其价值就急剧下降可以被安全地清理。它需要极低的读写延迟每次用户发言Agent都需要即时读取历史上下文生成响应后又要立即写入新的记录。这个过程的延迟必须远低于用户可感知的等待时间通常要求100ms。它需要高并发访问一个活跃的Agent服务可能同时处理成千上万个会话每个会话的读写操作都要求高度隔离且高效。理解了这些特性我们就能明白短期记忆的存储方案必须满足几个刚性指标超高速读写、支持高并发、具备TTL生存时间自动过期能力、以及能够方便地存储半结构化的会话数据。而Redis几乎是为这个场景量身定做的。3. Redis vs. 传统方案为什么数据库和纯内存缓存都不够好在决定使用Redis之前我们当然评估过其他方案。让我们来逐一分析为什么它们不适合作为Agent短期记忆的主存储。方案一关系型数据库如MySQL这似乎是存储数据的“正统”选择。我们可以设计一张conversation_messages表用session_id和turn_index来组织数据。劣势读写延迟过高即使对session_id建立索引每次读取一个完整会话的历史比如最近的20条记录仍然涉及磁盘I/O和复杂的查询组装。在对话高峰期这很容易成为瓶颈。写入开销大每一轮对话都需要执行一次INSERT操作。对于高频交互的Agent这会给数据库带来巨大的写入压力。清理成本高过期会话的清理需要定时任务执行DELETE可能产生表碎片影响性能。不适合非结构化数据对话消息中的content字段可能很长且结构多变用TEXT类型存储不利于快速解析。方案二服务进程的纯内存缓存如Pythondict这是最简单的方案将上下文存储在服务进程的内存字典里以session_id为键。劣势无状态服务的灾难现代云原生架构下服务通常是多实例、无状态的。用户的下一次请求可能被负载均衡到另一个实例导致上下文丢失。内存限制与扩容所有会话数据都堆在单个实例内存中容量有限且无法通过水平扩展来增加总内存容量。服务重启数据丢失任何部署更新或故障重启都会导致所有进行中的会话中断。方案三外部集中式缓存/数据库如Memcached、MongoDBMemcached它简单快速但数据结构单一主要支持字符串缺乏Redis丰富的哈希Hash、列表List、有序集合ZSet等结构。更重要的是它不支持数据持久化虽然短期记忆不需要强持久化但服务重启导致缓存雪崩也是个风险而且集群功能相对较弱。MongoDB作为文档数据库它存储JSON数据非常自然。但其读写性能尤其是写入在同等资源下通常低于Redis且更侧重于海量数据存储和复杂查询对于我们这个超高频、简单Key-Value访问的场景有点“杀鸡用牛刀”运维成本也更高。Redis的胜出点性能王者基于内存的操作读写速度在亚毫秒级完全满足Agent对延迟的苛刻要求。丰富的数据结构Hash结构可以完美存储一个会话的所有元信息创建时间、用户ID等和消息列表List或ZSet可以按顺序存储消息。这比将所有数据序列化成一个大字符串如用Memcached要高效和直观得多。内置TTL可以为每个会话的Key设置过期时间例如24小时。会话结束后数据自动被清理无需额外开发维护清理任务。高可用与持久化通过Redis Sentinel或Cluster模式可以实现高可用。虽然短期记忆对数据丢失不敏感但可选的RDB/AOF持久化机制能在服务重启时避免所有会话“归零”提供更好的用户体验。原子操作与Lua脚本支持对数据结构的原子性操作甚至可以通过Lua脚本实现复杂的“读取-修改-写入”逻辑保证在并发环境下的数据一致性。注意这里并不是说Redis是万能的。对于Agent的长期记忆知识库向量数据库依然是不可替代的选择。Redis的核心定位是解决短期、高频、临时的状态存储问题。4. 实战架构如何用Redis设计一个高效的Agent短期记忆模块理论说完了我们来点实际的。下面我将分享我们目前在生产环境中使用的架构设计包含关键的数据结构定义和核心操作逻辑。4.1 数据结构设计我们为每个会话Session在Redis中主要使用两种数据结构1. 会话元信息哈希HashKey:agent:session:{session_id}Value (Hash Field):user_id: 用户标识created_at: 会话创建时间戳updated_at: 最后活动时间戳ttl: 预设的过期时间秒metadata: 其他自定义元数据如渠道来源、语言等的JSON字符串这个Hash存储了会话的静态和描述性信息。2. 消息列表ListKey:agent:session:{session_id}:messagesValue: 一个列表每个元素是一条消息的JSON字符串。 例如[“{‘role’:’user’, ‘content’:’…’, ‘ts’:…}”, “{‘role’:’assistant’, …}”, …]我们选择List而不是ZSet因为对话消息天然是顺序追加的List的RPUSH右端推入和LRANGE范围读取操作效率极高且能完美保持顺序。ZSet更适合需要按分数如时间戳排序和范围查询的场景对于我们简单的追加读取来说稍显冗余。4.2 核心操作流程与代码示例Python假设我们使用redis-py客户端。初始化一个会话import redis import json import time redis_client redis.Redis(host‘localhost’, port6379, decode_responsesTrue) def create_session(user_id, session_id, initial_system_promptNone, ttl3600): “””创建新会话并可选地添加系统提示词””” session_key f“agent:session:{session_id}” msg_key f“{session_key}:messages” # 使用pipeline减少网络往返 pipe redis_client.pipeline() # 1. 设置会话元信息 pipe.hset(session_key, mapping{ ‘user_id’: user_id, ‘created_at’: time.time(), ‘updated_at’: time.time(), ‘ttl’: ttl }) # 设置这个Hash键的过期时间 pipe.expire(session_key, ttl) # 2. 初始化消息列表并设置同样的过期时间 if initial_system_prompt: system_msg json.dumps({‘role’: ‘system’, ‘content’: initial_system_prompt, ‘ts’: time.time()}) pipe.rpush(msg_key, system_msg) pipe.expire(msg_key, ttl) # 执行所有命令 pipe.execute() return session_id处理一轮用户输入并获取完整上下文 这是最核心的流程。注意我们需要在每次读写后更新会话的“最后活动时间”并刷新TTL滑动过期。def process_user_message(session_id, user_input): session_key f“agent:session:{session_id}” msg_key f“{session_key}:messages” # 1. 准备新消息 user_msg json.dumps({‘role’: ‘user’, ‘content’: user_input, ‘ts’: time.time()}) # 2. 使用事务MULTI/EXEC或Pipeline保证原子性 pipe redis_client.pipeline() # 2.1 将用户消息追加到列表 pipe.rpush(msg_key, user_msg) # 2.2 获取最近N条历史消息例如最近20条避免上下文过长 # 注意LLEN获取长度LRANGE获取内容。这里我们先获取在管道中执行。 # 更严谨的做法是用Lua脚本但为清晰起见这里分两步。 current_length redis_client.llen(msg_key) start_index max(0, current_length - 20) # 假设我们只保留最近20条作为上下文 pipe.lrange(msg_key, start_index, -1) # 获取从start_index到末尾的所有消息 # 2.3 更新会话的最后活动时间并刷新过期时间(TTL) pipe.hset(session_key, ‘updated_at’, time.time()) pipe.expire(session_key, 3600) pipe.expire(msg_key, 3600) # 执行管道命令 pipe.execute() # 3. 解析历史消息 # pipe.execute()返回一个结果列表对应每个命令的结果。 # rpush的结果是列表的新长度我们不需要。 # lrange的结果是消息JSON字符串列表。 history_msgs_json pipe.execute()[1] # 注意索引根据命令顺序调整 history_messages [json.loads(msg) for msg in history_msgs_json] # 4. 此时history_messages就是包含最新用户输入的完整短期记忆。 # 你可以将其传递给LLM进行推理生成助手回复。 # … (调用LLM的逻辑) … # 5. 假设LLM返回了助手回复 assistant_response assistant_msg json.dumps({‘role’: ‘assistant’, ‘content’: assistant_response, ‘ts’: time.time()}) # 6. 将助手回复也写入记忆 redis_client.rpush(msg_key, assistant_msg) # 注意这里可以再次刷新TTL或者依赖下一次用户交互来刷新。 return assistant_response4.3 高级模式上下文窗口管理与摘要生成直接存储所有原始消息会遇到LLM的上下文长度限制。一个成熟的方案需要实现上下文窗口管理。当消息列表超过一定长度如Token数或条数时不能简单粗暴地截断最早的几条那可能会丢失关键信息。我们的策略是滚动摘要Rolling Summary。在Redis中为每个会话额外维护一个summary字段存储在会话的Hash里。当原始消息列表过长时触发一个摘要生成过程将最早的一部分消息例如前10轮连同当前的摘要初始为空一起发送给LLM要求其生成一个浓缩的段落来概括之前对话的核心事实和决定。用这个新的摘要更新Redis中的summary字段并从消息列表中物理删除那些已被摘要的消息。后续的对话上下文就由“最新摘要 未被摘要的近期原始消息”共同构成。这样我们既控制了上下文长度又保留了对话的长期脉络。这个摘要生成过程可以异步执行避免阻塞主对话流程。5. 避坑指南Redis实战中的性能、一致性与运维陷阱将短期记忆迁移到Redis带来了巨大的收益但也引入了新的复杂性。下面是我在实战中踩过的坑和总结的经验。5.1 连接管理与性能调优连接池是必须的不要为每个请求创建新的Redis连接。使用连接池redis.ConnectionPool是保证高性能、低资源消耗的基础。根据你的并发量合理设置max_connections。Pipeline批量操作如上文示例所示将多个相关命令如写入消息、读取历史、更新TTL放入一个Pipeline中执行能大幅减少网络往返延迟RTT。这在处理一轮对话时效果显著。避免大Key虽然一个会话的消息列表会增长但要警惕单个Key过大如超过10KB。过大的Key在序列化、网络传输和Redis自身内存分配中都会影响性能。这就是为什么我们需要实施上文提到的“上下文窗口管理”和“摘要”机制定期修剪列表长度。内存监控与告警Redis是内存数据库必须严密监控内存使用率。设置告警阈值如80%并提前规划好内存淘汰策略maxmemory-policy。对于Agent记忆这种场景allkeys-lru最近最少使用或volatile-lru在设置了过期时间的键中LRU是比较合适的选择。5.2 数据一致性与并发安全原子性操作像“读取-修改-写入”这样的复合操作在并发环境下可能出错。例如两个并发的请求同时读取了旧的消息列表然后分别追加了自己的消息会导致其中一条消息丢失。最优雅的解决方案是使用Lua脚本。Redis保证一个Lua脚本在执行时是原子的。append_message_script “”” local msg_key KEYS[1] local session_key KEYS[2] local new_msg ARGV[1] local new_ttl ARGV[2] -- 追加消息 redis.call(‘RPUSH’, msg_key, new_msg) -- 获取最新的N条消息例如从倒数第20条开始 local recent_msgs redis.call(‘LRANGE’, msg_key, -20, -1) -- 更新活动时间 redis.call(‘HSET’, session_key, ‘updated_at’, ARGV[3]) -- 刷新过期时间 redis.call(‘EXPIRE’, msg_key, new_ttl) redis.call(‘EXPIRE’, session_key, new_ttl) return recent_msgs “”” append_message_lua redis_client.register_script(append_message_script) # 调用脚本 recent_msgs append_message_lua(keys[msg_key, session_key], args[new_message_json, 3600, time.time()])会话过期与雪崩如果大量会话在同一时间点创建并设置了相同的固定TTL如1小时它们可能会在同一时间点大量过期导致Redis内存释放压力骤增。解决方案是为TTL添加一个随机抖动例如base_ttl random.randint(0, 300)让过期时间分散开。5.3 高可用与灾难恢复单点故障绝对不要使用单机Redis。生产环境必须部署Redis Sentinel主从哨兵或Redis Cluster集群模式。Sentinel模式提供了自动故障转移适合数据量不是特别巨大的场景。Cluster模式则提供了数据分片和线性扩展能力。持久化权衡短期记忆可以容忍少量丢失用户大不了从头开始对话。因此可以优先考虑性能将appendfsync配置为everysec甚至no并主要依赖RDB快照进行周期性持久化。这样能在性能和可靠性间取得平衡。备份与恢复策略定期备份RDB文件到对象存储如S3。制定恢复演练流程。虽然数据不重要但快速恢复服务的能力很重要。5.4 监控与可观测性你需要监控的关键指标包括Redis服务器级内存使用率、连接数、每秒操作数OPS、命中率、网络输入/输出流量。应用级会话读写平均延迟、P99延迟、会话创建成功率。业务级活跃会话数、平均会话时长、因记忆丢失导致的会话异常终止率。将这些指标接入你的APM如PrometheusGrafana和日志系统才能做到心中有数快速定位问题是出在Redis、网络还是应用代码。6. 超越基础当Agent遇到复杂会话与分布式扩展基本的读写模式解决了大部分问题但随着业务复杂化我们遇到了更高级的场景。场景一支持话题分支与回溯有些复杂的客服或游戏Agent对话可能存在分支比如用户问“回到上一个问题”。简单的列表无法高效处理。我们的优化是引入ZSet有序集合。用时间戳作为分数score消息ID或内容作为成员member。这样我们可以ZADD按时间戳插入消息。ZRANGEBYSCORE获取某个时间范围内的消息实现“回溯到10分钟前的对话状态”。结合ZREMRANGEBYSCORE清理旧消息。 这比用List模拟要高效和直观得多但代价是存储开销稍大。场景二超大规模下的分片与隔离当你的用户量达到千万级即使有Cluster管理数十亿个会话Key也可能有压力。我们采取了逻辑分片在session_id的生成规则中融入用户ID哈希值。根据哈希值的前几位将会话路由到不同的Redis Cluster分片甚至是不同的物理Redis集群例如按业务线或地理区域隔离。在应用层维护一个路由表或使用一致性哈希库。 这样做不仅分散了压力也实现了故障隔离一个集群的问题不会影响所有用户。场景三与向量数据库的协同在高级的Agent中短期记忆Redis和长期记忆向量数据库需要联动。例如当用户说“还记得我上次提过的关于XX的需求吗”Agent需要从短期记忆中感知到这是一个“回忆”指令。然后它需要从Redis中取出当前会话的user_id和关键实体信息。利用这些信息作为查询条件去向量数据库中检索该用户历史上相关的长期记忆片段。最后将检索到的长期记忆片段与Redis中的短期记忆上下文拼接一起送给LLM生成回复。 Redis在这里扮演了状态枢纽和元数据索引的角色桥接了即时对话和长期知识。从那次痛苦的线上故障到今天稳定支撑日均百万级对话的架构把Agent的短期记忆剥离出来放入Redis是我们做的最正确的技术决策之一。它不仅仅是一个存储组件的更换更是一种架构思想的转变让专业的数据结构去做专业的事让状态管理回归服务端让通信链路轻装上阵。这个过程没有银弹你需要根据自己Agent的对话复杂度、流量规模和对一致性的要求仔细设计Redis的数据结构、过期策略和访问模式。但无论如何相比于把上下文在网络上“搬来搬去”拥有一个高速、可靠、中心化的记忆存储无疑是构建健壮、可扩展的多轮对话Agent的基石。