Agent任务恢复：Task、Session、Action三层状态管理实战

1. 从一次线上事故说起对话恢复了任务却丢了去年我在做一个智能客服 Agent 项目时遇到过一个让我印象非常深刻的线上问题。用户投诉说他在跟客服机器人聊到一半时切出去接了个电话回来继续对话机器人倒是很热情地接着聊但之前正在处理的退款申请流程却完全消失了——机器人像失忆了一样重新问他请问您遇到了什么问题。用户当时就炸了觉得这个系统看着聪明实际是个傻子。这个问题的根因就是标题里说的那件事Agent 能恢复对话不等于能恢复任务。当时我们的实现是把对话历史messages直接存下来用户回来就把 messages 重新喂给模型模型自然能接着聊。但退款申请这个任务的状态——走到哪一步了、收集了哪些信息、下一步该调哪个工具——这些信息根本没有被独立管理它们只是隐含在对话文本里。一旦对话被截断、被压缩、或者模型注意力跑偏任务状态就丢了。后来我把这套系统重构了一遍核心思路就是把Task任务、Session会话、Action动作这三层彻底拆开管理。这篇文章就把我踩过的坑、想明白的道理、以及最终落地的方案完整讲一遍。如果你正在做 Agent 相关的开发尤其是涉及多轮任务、工具调用、长对话的场景这篇内容应该能帮你少走不少弯路。先说结论Session 管的是人和 Agent 的交互上下文Task 管的是一件要完成的事的状态机Action 管的是Task 推进过程中的一次具体执行。三者生命周期不同、存储结构不同、恢复策略也不同。把它们混在一起短期能跑长期必崩。2. 为什么对话历史撑不起一个真正的任务2.1 对话历史本质上是无结构的文本流大多数人做 Agent 的第一版都是把对话历史当成唯一的状态载体。用户说一句模型回一句全部 append 到一个 list 里下次请求整个 list 丢给模型。这个方案在 demo 阶段非常好用因为简单、直观、不需要额外设计。但它的本质问题是对话历史是一段无结构的自然语言文本流而任务状态是一个有结构的、离散的状态机。这两者根本不是一回事。任务状态需要回答的问题是当前处于哪个阶段还缺哪些必填信息下一步该做什么这些是精确的、可判定的而对话历史里这些信息是隐含的、模糊的、需要模型重新推理才能提取的。我举个具体的例子。一个订机票的 Agent任务状态可能是这样的阶段收集中已收集出发地北京目的地上海日期下周三缺失舱位偏好、乘机人信息下一步询问舱位偏好这些信息如果只存在对话历史里模型每次都要从头读一遍所有对话重新推理哦用户已经说了出发地目的地还差舱位。对话短的时候没问题对话一长模型就可能漏掉早期信息或者把已经确认的信息又反问一遍。这就是为什么很多 Agent 聊到十几轮之后就开始犯糊涂。2.2 上下文窗口不是无限的压缩就会丢信息第二个致命问题是上下文窗口。不管模型支持多长的上下文实际工程中你都不可能无限往里塞。成本是一方面更重要的是长上下文里模型的注意力会稀释关键信息淹没在大量寒暄和无关内容里。所以实际系统一定会做上下文压缩或者截断。常见做法有几种滑动窗口只保留最近 N 轮、对早期对话做摘要、按重要性筛选保留。这些做法对聊天是够用的但对任务是灾难性的——因为你无法保证压缩算法知道哪些信息对任务推进是关键的。我见过一个团队的做法是保留最近 10 轮 早期摘要结果用户在第 3 轮说的收货地址到第 15 轮要下单时摘要里没提最近 10 轮里也没有Agent 就又问了一遍。用户觉得这系统怎么这么笨。其实不是模型笨是状态管理没做好。2.3 多任务并发时对话历史会互相污染还有一个更隐蔽的坑一个 Session 里可能同时存在多个 Task。比如用户先问帮我查下明天天气Agent 回答了接着用户说顺便帮我订张去上海的票这时候 Session 里就有两个任务了。如果状态全塞在对话历史里模型很容易把两个任务的上下文搅在一起比如把明天这个时间误用到订票任务上。正确的做法是每个 Task 有自己独立的状态对象Session 只负责把用户的输入路由到对应的 Task。这样即使两个任务在同一个会话里交替推进状态也不会串。我个人的经验是只要你的 Agent 涉及多步骤才能完成的一件事就应该立刻把 Task 状态独立出来不要等到出问题再重构。重构成本远高于一开始就设计好。3. Task、Session、Action 三层到底各管什么3.1 Session交互的容器不是任务的载体Session 的职责很纯粹标识这是同一个人在同一段时间内的一系列交互。它管的东西包括用户身份、会话创建时间、对话消息列表、当前活跃的 Task 列表、以及一些会话级别的偏好设置比如语言、时区。Session 的生命周期通常比较长可能跨越几小时甚至几天用户今天聊一半明天回来继续。但 Session 本身不应该承载任务逻辑它更像一个信封里面装着消息和任务引用。一个典型的 Session 结构大概长这样{ session_id: sess_abc123, user_id: user_456, created_at: 2025-01-15T10:00:00Z, messages: [ {role: user, content: 帮我订张去上海的票}, {role: assistant, content: 好的请问哪天出发} ], active_tasks: [task_001, task_002], context: {language: zh-CN, timezone: Asia/Shanghai} }注意messages和active_tasks是分开的。messages 是给人看的对话流active_tasks 是给系统用的任务索引。这个分离非常关键。3.2 Task一件要完成的事本质是个状态机Task 是三层里最重要的。它描述的是一件要完成的事核心是一个状态机。一个 Task 至少要包含任务类型、当前状态、已收集的槽位slot、缺失的槽位、历史执行记录、以及失败重试信息。我习惯把 Task 设计成显式的状态机状态之间的转移由明确的规则或模型决策驱动。比如订票任务状态含义可转移到的状态INIT刚创建还没开始COLLECTINGCOLLECTING正在收集信息CONFIRMING / FAILEDCONFIRMING等待用户确认EXECUTING / COLLECTINGEXECUTING正在调用工具执行DONE / FAILEDDONE完成-FAILED失败COLLECTING重试有了这个状态机恢复任务就变得非常简单读一下当前状态和槽位就知道该继续做什么完全不需要重新读一遍对话历史去推理。{ task_id: task_001, type: flight_booking, state: COLLECTING, slots: { origin: 北京, destination: 上海, date: 2025-01-22 }, missing_slots: [cabin_class, passenger_info], created_at: 2025-01-15T10:00:05Z, updated_at: 2025-01-15T10:02:30Z }3.3 Action任务推进中的一次具体执行Action 是粒度最细的一层指的是Task 推进过程中的一次具体动作。它可能是调用一个工具、执行一次数据库查询、发一封邮件、或者仅仅是向用户问一个问题。为什么要单独管理 Action因为 Action 是可能失败、可能重试、可能产生副作用的。比如调用支付接口这个 Action如果超时了你不知道到底扣没扣款这时候就需要 Action 级别的幂等控制和状态记录。如果 Action 只是隐含在对话里你根本没法做这些。Action 通常包含动作类型、输入参数、执行结果、执行时间、重试次数、幂等键。{ action_id: act_789, task_id: task_001, type: call_tool, tool: flight_search, params: {origin: 北京, destination: 上海, date: 2025-01-22}, status: SUCCESS, result: {flights: [...]}, idempotency_key: task_001_search_1, executed_at: 2025-01-15T10:03:00Z }三层的关系可以这样理解Session 是舞台Task 是剧本Action 是演员的每一个动作。舞台可以一直开着剧本可以有好几本同时演每个动作都要被记录清楚。4. 恢复任务时三层各自怎么恢复4.1 Session 恢复重建对话上下文Session 的恢复相对简单主要是把消息列表和活跃任务列表加载回来。但这里有个细节不要把所有历史消息都塞回模型。我的做法是保留最近若干轮原始消息加上一个任务状态摘要作为系统提示。这个摘要不是对话摘要而是从 Task 状态生成的、结构化的当前任务描述。比如当前有一个订票任务进行中已确认出发地北京、目的地上海、日期1月22日还缺舱位和乘机人信息。 这样模型一上来就知道该干什么不需要从对话里猜。4.2 Task 恢复读状态机不读对话Task 的恢复是核心。因为状态是显式存储的恢复时只需要加载 Task 对象 → 读取当前 state 和 slots → 根据状态机决定下一步。整个过程完全不需要模型参与推理是确定性的。这一点非常重要。确定性意味着可靠。如果每次恢复都要靠模型重新理解对话那模型一抖动任务就乱了。而读状态机是纯代码逻辑不会出错。我通常会在 Task 恢复后生成一个下一步动作建议然后交给模型去执行。比如状态是 COLLECTING 且缺失 cabin_class那下一步就是向用户询问舱位偏好。模型只负责把这个意图表达成自然语言不负责决定做什么。4.3 Action 恢复幂等与补偿Action 的恢复最容易被忽略但恰恰是生产环境最要命的。设想一个场景Agent 正在执行提交订单这个 Action请求发出去了但响应还没回来服务重启了。恢复后你怎么知道这个订单到底提交成功没有解决方案是幂等键 状态查询。每个 Action 执行前先写一条 PENDING 记录带上幂等键执行成功后更新为 SUCCESS失败更新为 FAILED。恢复时如果发现有 PENDING 的 Action就去查询下游系统的真实状态如果下游支持或者根据幂等键重试下游保证幂等。def recover_pending_actions(task_id): pending db.query_actions(task_id, statusPENDING) for action in pending: # 先查下游真实状态 real_status query_downstream(action.idempotency_key) if real_status SUCCESS: db.update_action(action.id, statusSUCCESS) elif real_status NOT_FOUND: # 安全重试幂等键保证不会重复 retry_action(action) else: # 状态不明标记为需人工介入 db.update_action(action.id, statusUNKNOWN)这里有个血泪教训任何有副作用的 Action 都必须有幂等键哪怕你觉得这个接口不会重复调用。线上环境什么诡异情况都有网络抖动、超时重试、服务重启没有幂等保护迟早出事。5. 落地时的几个关键设计决策5.1 状态存哪里内存、Redis 还是数据库这是第一个要拍板的问题。我的建议是分场景纯 demo / 单机内存字典就够了简单直接。生产环境 / 多实例必须用外部存储。Redis 适合存 Session 和活跃 Task读写频繁、可容忍少量丢失数据库适合存 Task 和 Action 的持久化记录需要事务和审计。我实际项目里的组合是Session 和 Task 的当前状态放 Redis带 TTLTask 和 Action 的完整历史落库。这样既保证了读取速度又保证了可追溯。5.2 状态更新用乐观锁还是悲观锁多个请求可能同时更新同一个 Task比如用户快速连发两条消息。这时候必须加锁否则状态会写乱。我推荐乐观锁给 Task 加一个 version 字段更新时检查 version 是否变化变了就重试。def update_task(task_id, expected_version, new_state): result db.execute( UPDATE tasks SET state?, versionversion1 WHERE id? AND version?, (new_state, task_id, expected_version) ) if result.rowcount 0: raise ConcurrentModificationError(任务已被其他请求修改请重试)乐观锁的好处是不会长时间持锁适合 Agent 这种读多写少、偶尔并发的场景。5.3 什么时候创建新 Task什么时候复用这是个容易纠结的问题。我的判断标准是如果用户的新意图和现有 Task 是同一件事的延续就复用如果是新的一件事就新建。比如用户说订去上海的票创建 Task A接着说要头等舱这是补充信息复用 Task A然后说再帮我订张去北京的这是新任务创建 Task B。判断是不是同一件事可以先用规则比如是否在补充槽位规则搞不定的再让模型判断。5.4 对话消息和任务状态的一致性怎么保证两者分开存储后可能出现不一致消息写成功了但状态没更新或者反过来。我的做法是以 Task 状态为准消息只是展示层。也就是说即使消息丢了只要 Task 状态在任务就能继续反过来消息在但状态丢了那才是真麻烦。所以写入顺序上我会先更新 Task 状态这是关键路径再写消息记录这是展示路径。如果第二步失败最多是对话记录不全不影响任务推进。6. 那些年我踩过的坑和对应的解法6.1 坑一把槽位信息只存在对话里最开始我图省事槽位信息不单独存全靠模型从对话里提取。结果就是前面说的对话一长就丢信息。后来改成显式槽位存储模型每次只负责从最新一句话里提取槽位提取完就合并到 Task 的 slots 里再也不依赖历史对话。这个改动带来的收益是巨大的上下文可以放心压缩了因为关键信息都在 Task 里对话历史压缩掉也不影响任务。6.2 坑二Action 重试导致重复下单有一次测试环境网络抖动导致创建订单的 Action 超时系统自动重试结果创建了两个订单。幸好是测试环境。后来我给所有写操作都加了幂等键重试前先查状态问题就解决了。这里的关键认知是超时不等于失败。超时只是不知道结果可能是成功也可能是失败。所以超时后的正确动作是查询真实状态而不是直接重试。6.3 坑三Session 过期把进行中的 Task 也删了早期我把 Task 存在 Session 对象里Session 一过期Task 跟着没了。用户第二天回来任务全丢。后来把 Task 独立存储Session 过期只影响对话上下文Task 该在还在。这个坑的本质是生命周期耦合。Session 的生命周期和 Task 的生命周期本来就不一样硬绑在一起必然出问题。分开之后Session 可以短命Task 可以长命各得其所。6.4 坑四多任务并发时状态串了前面提过一个 Session 里多个 Task 并发时如果状态管理不清晰很容易串。我的解法是每次处理用户输入时先做意图路由判断这句话是给哪个 Task 的。路由对了再更新对应 Task 的状态。路由这一步可以用规则 模型结合规则处理明显的比如取消对应最近的 Task模型处理模糊的。7. 一套可以直接抄的最小实现骨架说了这么多原理最后给一套可以直接上手的最小实现。我用 Python 伪代码写重点是结构具体存储你换成自己的即可。class Session: def __init__(self, session_id, user_id): self.session_id session_id self.user_id user_id self.messages [] self.active_task_ids [] self.context {} class Task: def __init__(self, task_id, task_type): self.task_id task_id self.type task_type self.state INIT self.slots {} self.missing_slots [] self.version 0 self.actions [] def next_step(self): # 根据状态机决定下一步纯确定性逻辑 if self.state COLLECTING and self.missing_slots: return {type: ask_user, slot: self.missing_slots[0]} if self.state COLLECTING and not self.missing_slots: self.state CONFIRMING return {type: confirm, slots: self.slots} if self.state CONFIRMING: return {type: execute} return {type: done} class Action: def __init__(self, action_id, task_id, action_type, params): self.action_id action_id self.task_id task_id self.type action_type self.params params self.status PENDING self.idempotency_key f{task_id}_{action_type}_{hash(params)} self.retries 0 def handle_user_input(session, user_text): # 1. 路由判断这句话属于哪个 Task task route_to_task(session, user_text) if task is None: task create_new_task(session, user_text) # 2. 提取槽位并合并 extracted extract_slots(user_text, task.type) task.slots.update(extracted) task.missing_slots compute_missing(task) # 3. 推进状态机 step task.next_step() # 4. 执行动作 if step[type] ask_user: reply generate_question(step[slot]) elif step[type] execute: action Action(new_id(), task.task_id, execute_task, task.slots) execute_with_idempotency(action) reply 已为您处理完成 # 5. 记录消息 session.messages.append({role: user, content: user_text}) session.messages.append({role: assistant, content: reply}) # 6. 持久化 save_task(task) save_session(session) return reply这套骨架的核心就是路由 → 提取 → 推进 → 执行 → 记录每一步职责清晰状态显式存储。你可以在此基础上加自己的工具调用、多任务路由、失败重试逻辑。8. 关于这套设计我最后想说的几点体会做了几个 Agent 项目之后我越来越觉得状态管理才是 Agent 工程的核心难点模型能力反而是相对好解决的部分。模型再强如果状态管理一团糟用户体验照样崩。反过来状态管理做扎实了哪怕模型一般系统也能稳定可用。Task、Session、Action 这三层分开本质上是一种关注点分离的思想。Session 关注交互Task 关注目标Action 关注执行。每一层只做自己该做的事层与层之间通过清晰的接口通信。这个原则说起来简单但真正落地时很多人会忍不住把逻辑往对话历史里塞因为那样看起来简单。我的建议是忍住这个冲动该建的结构一定要建。还有一个体会是关于确定性的。Agent 系统里能用确定性代码解决的部分就不要交给模型。状态机转移、槽位校验、幂等控制这些都是确定性的交给代码只有自然语言理解、意图判断、回复生成这些真正需要智能的部分才交给模型。这样系统的可靠性会高一个数量级。最后分享一个小技巧给每个 Task 加一个人类可读的状态描述字段。这个字段不参与逻辑纯粹用于调试和展示。当线上出问题时你打开这个字段一看就知道任务卡在哪了排查效率极高。我现在的项目里这个字段救了我无数次。这套方案不是银弹但它帮我解决了一个真实且普遍的问题。如果你也在做 Agent希望这些经验能让你少踩几个坑。