多 Agent 客服的短期记忆改造：用 Redis + Checkpointer 管理会话状态-北京尧图网络

前言mall-customer的短期记忆原本是一个手写的 Redis 封装把对话消息按会话 ID 存进 Redis list读的时候取最近 N 条。封装本身是通的但它和 Agent 图之间是断开的——消息存下来了模型看不到。用户第一轮说我要退款第二轮说给它升级优先级客服会把第二句当成一件新的事处理要么答非所问要么再建一张新工单。Agent State 本来就是用来记录整个图执行过程中的状态的其中就包括消息——它完全可以替代原来那套短期记忆只需要把这份状态持久化下来。框架已经提供了现成的做法——checkpointer编译图时传入它每个节点执行后的状态都会自动持久化下一轮自动恢复存进去的状态就是下一轮要用的。所以这次改造没有重写短期记忆只是把自建的 Redis 封装删掉状态存取交给 checkpointer。过程中还清掉了状态模型里的两个冗余字段并解决了暴露出来的两个问题。一、旧短期记忆的问题memory/short_term.py短期记忆就围绕三个核心操作async def add_message(self, session_id: str, role: str, content: str) - None: message {role: role, content: content, timestamp: datetime.now().isoformat()} key fsmartcs:short_term:{session_id} await r.rpush(key, json.dumps(message, ensure_asciiFalse)) # 追加一条消息 await r.ltrim(key, -self.max_turns, -1) # 只保留最近 20 条 await r.expire(key, self.ttl_seconds) # 30 分钟过期Redis 里确实存放着最近的对话/api/history也能读回来。但它和图是断开的具体有两个问题•存了但没用上。/api/chat调用graph.ainvoke()时初始 state 只带本轮这一条新消息messages: [HumanMessage(contentrequest.message)]。Redis 里存着 20 条历史图里只读得到 1 条——历史从未注入状态多轮追问缺少上下文。•两套状态各自维护。短期记忆里存的是消息文本图的内部还有intent路由给哪个 Agent、compliance_passed合规是否通过这类编排状态。这些字段既不落 Redis、也不进下一轮进程一重启就丢失。二、Checkpointer 如何持久化会话状态checkpointer 存到哪、怎么存graph.compile(checkpointer...)之后每次执行都会以thread_id为键把每个 superstep 之后的完整状态快照写入存储下一轮用同一个thread_id调用ainvoke()时框架先恢复上次的状态再把这一轮的输入按 reducermessages字段上是add_messages合并进去任何时刻都可以用graph.aget_state()读出当前状态。Agent State 检查点读写流程按 thread_id 读取 → 本轮执行 → 每个 superstep 写回 Redis三、换成 Redis Checkpointer依赖与前置条件•requirements.txt增加一行langgraph-checkpoint-redis0.5.0本次接入用的 0.5.2• 这个 checkpointer 不是普通的 KV 存储它用 RedisJSON 保存状态、用 RediSearch 建索引所以需要Redis Stack 或 Redis 8。本地 compose 用的本来就是redis/redis-stack-server镜像不用额外改•REDIS_URL环境变量补上密码。初始化与降级新的memory/short_term_memory.py里初始化都放在setup()方法中# Agent State 检查点TTL 30 分钟读取/写入均刷新过期时间CHECKPOINT_TTL {default_ttl: 30, refresh_on_read: True}class ShortTermMemory: async def setup(self) - None: 初始化 checkpointer优先 Redis失败降级进程内 MemorySaver from langgraph.checkpoint.memory import MemorySaver from langgraph.checkpoint.redis.aio import AsyncRedisSaver stack AsyncExitStack() try: checkpointer await stack.enter_async_context( AsyncRedisSaver.from_conn_string(self._redis_url, ttlCHECKPOINT_TTL) ) await checkpointer.asetup() self._stack, self._checkpointer stack, checkpointer except Exception as e: print(f[ShortTermMemory] Redis checkpointer 初始化失败降级为内存 MemorySaver: {e}) await stack.aclose() self._stack, self._checkpointer None, MemorySaver()几个实现细节•ttl控制的是会话的生命周期default_ttl的单位是分钟refresh_on_read: True表示读取时也续期——会话持续使用就一直存活30 分钟没有读写才过期•from_conn_string()是异步上下文管理器这里用AsyncExitStack手动进入好处是进出成对连接建立失败时已申请的资源会被回滚掉然后降级为进程内的MemorySaver服务正常启动。降级只保证可用性——内存版重启即失效多进程之间也不共享当前版本里MemorySaver是InMemorySaver的兼容别名•asetup()负责建立 RediSearch 索引和 JSON 文档结构启动时执行一次幂等。接线thread_id、/api/chat 与 /api/history在api/main.py的 lifespan 里先初始化短期记忆再把 checkpointer 传给图asynccontextmanagerasync def lifespan(app: FastAPI): global graph # 短期记忆Redis 持久化 Agent StateRedis 不可用时自动降级内存 await short_term_memory.setup() graph create_supervisor_graph( working_memoryworking_memory, long_term_memorylong_term_memory, checkpointershort_term_memory.checkpointer, ) ... yield await short_term_memory.aclose()create_supervisor_graph的参数里原来传的是short_term_memory对象现在传checkpointer——图不再依赖具体的存储实现def create_supervisor_graph(..., checkpointer: Any | None None, enable_checkpointing: bool True): ... if checkpointer is None and enable_checkpointing: from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() compiled graph.compile(checkpointercheckpointer) return compiled/api/chat里会话 ID 直接当作thread_id。这一轮的初始 state 只包含新消息和业务字段历史对话由框架恢复session_id request.session_id or str(uuid.uuid4())config {configurable: {thread_id: session_id}}initial_state { messages: [HumanMessage(contentrequest.message)], user_id: request.user_id, session_id: session_id, ...}result await graph.ainvoke(initial_state, configconfig)在/api/history里通过读取图的状态快照获取对话历史async def get_history(self, graph: Any, session_id: str, last_n: int 20) - list[dict]: 读取会话对话历史 — 从 LangGraph Agent Statecheckpointer获取 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage try: snapshot await graph.aget_state({configurable: {thread_id: session_id}}) except Exception: snapshot None messages (snapshot.values or {}).get(messages, []) if snapshot else [] role_map {HumanMessage: user, AIMessage: assistant, SystemMessage: system} history [ {role: role_map.get(type(msg), system), content: msg.text} for msg in messages ] return history[-last_n:]只保留最近 20 条TTL 控制的是会话何时过期不管状态里累积了多少消息——如果只有 TTL持续时间长的会话检查点里的消息会一直累积。旧实现用ltrim把列表裁到 20 条新方案需要补上同样的条数限制。做法是在图的入口加一个trim_history节点紧跟本轮输入合并之后执行消息数超过窗口的部分裁掉——用RemoveMessage把最旧的几条标记移除而不是让 reducer 静默丢弃经add_messages缩减之后再做路由。这样后续节点Supervisor 路由、子 Agent、合规读到的就是同一个窗口checkpointer 落盘的也是有界状态# 短期记忆滑动窗口Agent State 中最多保留最近 20 条消息约 10 轮对话# 超出部分由 trim_history 节点用 RemoveMessage 移除防止状态无限增长MESSAGE_WINDOW_SIZE 20def trim_history(state: AgentState) - dict[str, Any]: 滑动窗口把 Agent State 的消息裁剪到最近 MESSAGE_WINDOW_SIZE 条作为图的入口节点执行紧接着本轮的输入合并之后——保证后续所有节点的 LLM 只看得到窗口内的消息checkpointer 持久化的也是有界状态。通过 RemoveMessage 走 add_messages reducer 裁剪而不是直接覆盖列表。 messages state.get(messages, []) overflow len(messages) - MESSAGE_WINDOW_SIZE if overflow 0: return {} removals [RemoveMessage(idm.id) for m in messages[:overflow] if m.id] return {messages: removals}接线时把入口从supervisor_route改为trim_historygraph.add_node(trim_history, trim_history) # 滑动窗口裁剪超窗口的历史消息...graph.set_entry_point(trim_history)graph.add_edge(trim_history, supervisor_route)和ltrim的差别在于ltrim是物理删除被裁掉的消息无法找回RemoveMessage是往状态里提交一次删除指令由add_messagesreducer 处理语义上仍是一次状态更新——和第四节合规撤回是同一套机制。窗口大小 20 条约 10 轮对话沿用旧实现的默认值。四、状态模型调整从 sub_results 到 messages接入 checkpointer 之前AgentState里还有sub_results和final_response两个字段节选class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] user_id: str session_id: str intent: str sub_results: dict[str, Any] # 各子 Agent 的结果按 Agent 名分桶 compliance_passed: bool final_response: str # 合成节点拼出来的最终回复 current_agent: str retry_count: intsub_results和final_response的问题在于• 子 Agent 处理完不是往消息流里写回答而是把结果写进sub_results[knowledge_rag]这样的字典这些内容不是消息也不会被恢复成对话上下文•final_response是合成节点拼好之后再写回状态的和 messages 里最后追加的那条 AIMessage 是同一份内容存了两遍。既然会话状态都放在messages里这两个字段就是多余的。改造后子 Agent 的结果直接作为AIMessage写入messages——它既是要给用户的回复也是下一轮对话历史的组成部分return { **state, messages: [AIMessage(contentresult)],}合成节点不再做拼接末尾的AIMessage就是最终回复。改造过程中处理了两个边界情况均来自实际运行其一合规不通过时子 Agent 的结果要撤回。该结果已经写入messages如果合规不通过只是追加一句转人工话术违规内容会留在会话历史中下一轮仍会作为上下文送入模型。因此需要用RemoveMessage将其移除if not state.get(compliance_passed, True): final_response ( 抱歉您的请求涉及敏感内容已转交人工客服处理。工单编号已自动生成请留意后续通知。 ) updates: list[BaseMessage] [] if isinstance(last, AIMessage) and last.id: updates.append(RemoveMessage(idlast.id)) # 撤回子 Agent 的结果 updates.append(AIMessage(contentfinal_response)) return {**state, messages: updates}RemoveMessage配合add_messagesreducer按消息 id 移除已写入的消息——状态统一到 messages 之后撤回才能这样实现。其二合规通过时不再追加第二份最终回复。合成节点最初沿用旧逻辑无条件把final_response追加进 messages导致子 Agent 的回复与合成节点追加的回复同时存在会话历史中出现两条内容相同的 AI 消息。修复方式是在追加前判断末尾是否为子 Agent 已写入的回复if isinstance(last, AIMessage) and bool(last.content): return state # 子 Agent 已写入回复直接收尾配套地合规节点的审查对象也从sub_results拼接串改成了messages末尾的AIMessage末尾没有子 Agent 的回复时比如直达合规审查直接跳过last messages[-1] if messages else Noneif not isinstance(last, AIMessage) or not last.content: return {**state, compliance_passed: True}compliance_result await self.full_check(last.content)改造后的整张图业务咨询办理 / 投诉闲聊寒暄每个 superstep 读写用户消息trim_history裁剪到 20 条supervisor_route意图路由knowledge_ragticket_handlerchitchatcompliance_checksynthesize末尾 AIMessage即最终回复Redis Checkpointerid: session_id五、多轮工单记忆的存储与消费状态改造完成、checkpointer 接入之后多轮场景的测试显示上下文问题并没有完全解决。测试会话中依次发送三条消息• 第一条“我上周买的超声波香薰机坏了我要申请退款” → 建单成功TK-20261011-B35059• 第二条“太慢了给它升级为紧急优先级” → 又建了一张新单TK-20261011-78F6D1• 第三条“查一下它的进度” → 再建一张新单TK-20261011-80BF76。三条消息三张工单。检查点里已存有完整历史为什么升级优先级和查进度的请求仍在新建工单原因在ticket_handler的分析逻辑分析请求时只有最后一条消息进入了 LLM 的 promptlast_message messages[-1].contentticket_info await self.analyze_request(last_message) # 历史从未传入修复分两层•把历史注入分析 prompt。取最近 10 条对话格式化成文本和当前消息一起交给 LLM# 关键修复把对话历史一并交给分析 LLM否则它只看得到最新一条消息# 无法得知用户指代的已有工单记忆存了却用不上history_text self._format_history(messages[:-1])ticket_info await self.analyze_request(last_message, history_text)同时在系统 prompt 里把规则写死指代、查询、处理已存在工单的消息“那个工单”“它”“刚才的单子”“查一下进度”“给它升级优先级”action必须是query或update并从对话历史里提取工单号此时严禁新建工单。•正则兜底LLM 可能漏带ticket_id因此在历史消息中由新到旧扫描用正则提取最近出现过的工单号作为兜底TICKET_RE re.compile(rTK-\d{8}-[A-Fa-f0-9]{6})def _extract_ticket_id(self, messages: list[Any]) - str | None: 从对话历史由新到旧中提取最近出现的已有工单号。 for m in reversed(messages): content m.content if isinstance(m.content, str) else str(m.content) match TICKET_RE.search(content) if match: return match.group(0) return None# 双保险LLM 没带 ticket_id 时直接从对话历史里提取ticket_id ticket_info.get(ticket_id) or self._extract_ticket_id(messages)判定逻辑变成createupdate / queryupdatequery用户消息analyze_request携带最近 10 条历史action创建新工单ticket_id 取自 LLM缺失时正则兜底提取update_priority更新优先级query_ticket查询状态六、补充一个 chitchat 节点改造验证过程中发现一类不必要的开销用户只发送一句你好消息也会被路由到knowledge_rag多出一次无意义的知识库检索。解决办法是加一个chitchat节点专门处理打招呼、感谢、告别这类非业务消息不检索知识库直接生成回复。行为约束写在系统 prompt 中CHITCHAT_SYSTEM_PROMPT 你是一个友好的电商客服负责处理用户的寒暄与闲聊。...回答规则1. 简短友好符合客服场景不超过3句话2. 不检索、不编造任何业务信息商品价格、订单状态等一律不回答3. 不做任何业务承诺退款、赔付、优惠等4. 用户若提及具体业务问题礼貌引导其重新描述由知识库回答改动共三处Supervisor 的路由 prompt 增加chitchat选项和路由规则闲聊 → chitchat不要为此检索知识库图上注册chitchat节点和另外两个子 Agent 一样汇聚到合规检查route_map增加对应映射。七、验证验证环境为本地 docker compose 启动的中间件含redis-stack-servermall-customer用python -m api.main直接启动测试用例直接调用 Python 服务的接口不经过 Java 后端测的就是短期记忆本身。以下均为实际运行结果模型回复每次生成略有不同。多轮上下文与历史读取新建一个会话依次发送两轮[轮1] user: 你好我叫小明记住我的名字[轮1] assistant: 你好小明很高兴认识你有什么需要帮忙的比如查订单、退换货或商品咨询都可以告诉我。[轮2] user: 我叫什么名字[轮2] assistant: 你叫小明呀我记住啦。有什么电商相关的问题需要我帮忙吗 ![](http://cdn.zhipoai.cn/4c441306.jpg) 门户端智能客服两轮对话第二轮正确回溯了第一轮告知的姓名第二轮回答正确引用了第一轮的信息。随后读 /api/history返回 4 条消息角色顺序为 user / assistant / user / assistant与对话一致last_n2 时只返回最近两条。 ### 跨进程重启的状态恢复会话建立后终止 mall-customer 进程并重启再读 /api/history4 条消息完整返回。重启后继续追问 plaintext user: 我刚才说我叫什么名字来着assistant: 你刚才说你叫小明呀我记着呢。有什么需要帮忙的吗 ![](http://cdn.zhipoai.cn/6c7c6028.jpg) 重启 mall-customer 进程后完整历史恢复新回复正确引用重启前告知的姓名 ### 多轮工单修复前后同一个场景建单 → 要求升级优先级 → 查进度在修复前后分别验证 * • 修复前三条消息建了三张工单TK-20261011-B35059 / TK-20261011-78F6D1 / TK-20261011-80BF76升级优先级和查进度均被当作新诉求 * • 修复后第一条消息建单 TK-20261011-6C4751中等优先级第二条太慢了给它升级为紧急优先级返回工单已更新成功工单号: TK-20261011-6C4751 ⚡ 优先级: 紧急第三条查一下它的进度直接查回同一张单更新时间正是升级优先级操作的时间。全程只存在一张工单。 ![](http://cdn.zhipoai.cn/9c7702ae.jpg) 修复后同一条工单贯穿建单、升级优先级与进度查询 ### 合规撤回与消息去重这两条路径涉及不适合展示给用户的内容无法通过对话直接验证因此用 FakeListChatModel 编写临时冒烟脚本按图的三条出口分别断言 * • 合规通过历史里只有一条 AI 回复没有重复追加 * • 合规不通过子 Agent 的违规结果被 RemoveMessage 撤回messages 里查不到只留转人工话术 * • 直达合规追加兜底话术抱歉暂时无法处理您的请求请稍后重试。 * • 三轮累积对话第 1 轮的回复仍在、第 2 轮违规内容没有泄漏、第 3 轮的兜底话术正常messages 长度与轮次严格对应。以上断言全部通过。 ### Redis 里存了什么一个会话在 Redis 里的实际键会话 ID test-stm-3ffa40b6 plaintext checkpoint:test-stm-3ffa40b6:__empty__:{checkpoint_id}checkpoint_latest:test-stm-3ffa40b6:__empty__checkpoint_write:test-stm-3ffa40b6:__empty__:{checkpoint_id}:{task_id}:{序号}write_keys_zset:test-stm-3ffa40b6:__empty__:{checkpoint_id}•checkpoint:是每个 superstep 的状态快照•checkpoint_write:是各节点执行的写入记录channel 更新•checkpoint_latest:指向该会话最新的检查点•write_keys_zset:记录某次检查点关联的写入键交给 saver 统一管理 TTL• 中间那段__empty__是空命名空间checkpoint_ns在存储层的占位子图场景才会取别的值• 取样时 TTL 剩余 1797 秒——正好是default_ttl: 30分钟且会随读写刷新。窗口部分验证会话连续进行了 15 轮对话状态中的消息数在第 10 轮达到 20 后不再增长每轮入口先裁到 20 条再处理本轮轮末最多到 2120 条窗口本轮回复下一轮入口再次裁剪到第 15 轮时最早的几轮对话已滑出窗口。向一个会话直接写入 22 条消息再触发一轮最早 2 条被RemoveMessage移除从第 3 条开始保留。降级路径同样验证过把REDIS_URL指向一个不存在的端口启动服务打印降级日志后正常启动checkpointer 类型从AsyncRedisSaver变为InMemorySaver[ShortTermMemory] Redis checkpointer 初始化失败降级为内存 MemorySaver: ...Redis 不可用时 InMemorySaverRedis 可用时 AsyncRedisSaver小结这次改造去掉了手写的状态存取交给框架来做• 短期记忆不再由业务代码维护消息列表AsyncRedisSaver以thread_id为键持久化整份会话状态TTL 30 分钟失效trim_history节点把状态裁剪到最近 20 条Redis 不可用时自动降级进程内存储•AgentState里只留messages一条主线子 Agent 的结果直接作为消息存入短期记忆合规撤回用RemoveMessage合成节点不再重复追加• 子 Agent 消费历史是另一件事工单 Agent 现在分析请求时会带上最近 10 条对话并用正则从历史里兜底提取工单号给它升级优先级能落到那张已有的工单上。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

多 Agent 客服的短期记忆改造：用 Redis + Checkpointer 管理会话状态

相关推荐

FreeRTos 正点原子笔记（2）：解读 FreeRTOSConfig . h；函数名前缀 v x u 等；port 代表移植层，平台适配层；configTOTAL_HEAP_SIZE 的内存管理

[ZYNQ Linux] VCU使用

从小白到程序员：大模型Agent开发入门指南，抓住2026年蓝海机遇！

更多资讯

AI测试技术解析：从用例生成到企业落地实践

Jev SDK：Python开发者友好的TypeSafe AI API封装

电商推荐系统冷启动解决方案：Java混合模型实践

Jev不是模型，而是TypeSafe AI的类型安全API SDK

LLM数据管线中单元测试为何失灵？三层运行时防线帮你兜底

大模型知识基础：从参数到部署的完整认知框架

今日更新

PLFM_RADAR：从被动告警到主动感知的微服务健康监控系统

SPI隔离设计实战：MAX14431BAEE+与STM32F407工业级应用

数字隔离器如何根治SPI通信地环路干扰：MAX14431与TM4C129实战

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？

相关推荐

FreeRTos 正点原子 笔记（2）：解读 FreeRTOSConfig . h；函数名前缀 v x u 等；port 代表移植层，平台适配层；configTOTAL_HEAP_SIZE 的内存管理

[ZYNQ Linux] VCU使用

从小白到程序员：大模型Agent开发入门指南，抓住2026年蓝海机遇！

更多资讯

AI测试技术解析：从用例生成到企业落地实践

Jev SDK：Python开发者友好的TypeSafe AI API封装

电商推荐系统冷启动解决方案：Java混合模型实践

Jev不是模型，而是TypeSafe AI的类型安全API SDK

LLM数据管线中单元测试为何失灵？三层运行时防线帮你兜底

大模型知识基础：从参数到部署的完整认知框架

今日更新

PLFM_RADAR：从被动告警到主动感知的微服务健康监控系统

SPI隔离设计实战：MAX14431BAEE+与STM32F407工业级应用

数字隔离器如何根治SPI通信地环路干扰：MAX14431与TM4C129实战

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？

FreeRTos 正点原子笔记（2）：解读 FreeRTOSConfig . h；函数名前缀 v x u 等；port 代表移植层，平台适配层；configTOTAL_HEAP_SIZE 的内存管理