AI智能体在处理连续对话时，不能把每次请求当成孤立输入。否则第二句“那靠窗的还有吗”会被模型理解成全新问题，丢失前一轮的航线、日期和人数。Agent需要在多轮交互中记录用户意图、已确认参数、中间推理结果和任务进度。状态管理解决的就是这些信息如何存储、何时写入、怎样跨请求读取。内存方案适合单实例低并发，Redis方案适合分布式部署。
接下来从状态建模开始，分别实现内存字典和Redis两种存储，并讨论过期清理与故障恢复。
一、多轮对话状态到底要存什么
状态不是只保存聊天记录。聊天记录只是原始输入，Agent还需要从这些输入中拆出可执行的信息。通常一个会话状态至少包含四类字段。第一类是对话历史，用于给大模型提供上下文，但要限制条数，避免超过上下文窗口。第二类是意图和槽位，意图表示用户当前想做的事，如订机票、查物流；槽位是已经提取到的具体参数，例如出发城市、到达城市、日期、舱位。第三类是流程状态，记录当前步骤、已经完成哪些节点、下一步需要追问什么。第四类是元数据，包括会话ID、用户ID、创建时间、最近活跃时间和模型版本。
用结构化对象代替纯文本历史，可以让状态恢复更稳定。例如下面这个Python数据类把关键信息分开存储：
from dataclasses import dataclass, field
from typing import Any, Dict, List
@dataclass
class AgentState:
session_id: str
messages: List[Dict[str, str]] = field(default_factory=list)
current_intent: str = ""
slots: Dict[str, Any] = field(default_factory=dict)
stage: str = "idle"
created_at: float = 0.0
last_active_at: float = 0.0
messages 只保留必要轮次，比如最近10到20条。slots 中不要直接放身份证号或银行卡号，只保存脱敏后的摘要或临时引用。stage 可以用来表示当前工作流节点，比如 waiting_for_departure_city。这样当新请求进来时，加载状态后先检查 stage，就知道上一轮停在哪里，该继续问什么。
状态内容的粒度直接影响恢复效果。如果只存聊天记录，每次都要重新解析全部历史，既慢又容易出错。如果存得太细，状态对象会膨胀，写回和序列化成本上升。一般建议保存可解释的中间变量，而不是保存模型的完整推理张量。
二、内存方案：字典加锁的单机实现
内存方案的核心是一个进程内的映射结构，用会话ID作为键，状态对象作为值。Python里用字典最简单，但因为Web框架通常会并发处理多个请求，需要对字典读写加锁，避免同一会话被同时修改时出现竞态。
下面的类封装了获取、写入和删除操作，并引入 threading.Lock 保护共享字典：
import threading
from typing import Dict, Optional
class MemoryStateStore:
def __init__(self):
self._states: Dict[str, AgentState] = {}
self._lock = threading.Lock()
def get(self, session_id: str) -> Optional[AgentState]:
with self._lock:
return self._states.get(session_id)
def set(self, session_id: str, state: AgentState) -> None:
with self._lock:
self._states[session_id] = state
def delete(self, session_id: str) -> None:
with self._lock:
self._states.pop(session_id, None)
使用起来也很直接：请求进入时先根据会话ID读取状态，如果不存在就创建一个新的 AgentState；业务处理完成后调用 set 写回。内存读取没有网络往返，单次操作通常是微秒级，很适合本地调试和单实例部署。
但它有三个明显短板。第一，进程重启后所有会话状态消失，用户刷新页面后可能丢失整个任务进度。第二，如果服务部署了多个副本，用户第一次请求落在A实例，第二次被负载均衡分到B实例，B实例的内存里没有这份状态，对话就会断裂。第三，长期运行的进程会积累大量无用户会话，必须有清理机制，否则内存持续增长。
三、Redis方案：跨实例共享与持久化
Redis把状态从进程内存搬到独立存储层。多个Agent实例通过同一个Redis服务读写会话状态，天然解决负载均衡场景下的上下文丢失问题。常见的做法是以 agent:session:{session_id} 作为键，值保存JSON字符串。JSON格式可读性好，也方便跨语言协作。
下面是一个基于 redis-py 的Redis状态存储实现：
import json
import redis
from typing import Optional
class RedisStateStore:
def __init__(self, host: str = "127.0.0.1", port: int = 6379,
db: int = 0, ttl: int = 1800):
self.client = redis.Redis(host=host, port=port, db=db,
decode_responses=True)
self.ttl = ttl
def _key(self, session_id: str) -> str:
return f"agent:session:{session_id}"
def get(self, session_id: str) -> Optional[dict]:
raw = self.client.get(self._key(session_id))
if raw is None:
return None
return json.loads(raw)
def set(self, session_id: str, state: dict) -> None:
raw = json.dumps(state, ensure_ascii=False)
self.client.setex(self._key(session_id), self.ttl, raw)
def delete(self, session_id: str) -> None:
self.client.delete(self._key(session_id))
这里用 setex 在写入时同时设置过期时间，避免状态永久堆积。get 返回字典而不是自定义对象，是因为跨服务边界时对象类型需要反序列化成本。如果你希望拿到对象，可以在读取后手动构造 AgentState，或使用 pydantic 模型做校验。
除了整体JSON字符串，也可以用Redis的Hash结构存储。Hash允许按字段读取和修改，例如单独更新 slots 而不重写整个messages，还能配合 HINCRBY 做计数。但Hash不支持直接设置整体TTL，需要额外EXPIRE，并且复杂嵌套结构仍需序列化。对于大多数Agent应用，JSON字符串加合适TTL已经足够，开发成本更低。
Redis方案同样不是万能的。网络延迟比内存高，单次读写从微秒级上升到亚毫秒级；Redis服务本身需要维护、监控和备份。并发写同一会话时，两个请求可能互相覆盖，必要时可以加分布式锁或使用带版本号的CAS更新。短会话或强一致的简单场景，不必强行引入Redis。
四、过期清理与异常恢复
会话状态不能无限保留。用户停止交互后，状态还占用存储空间，既浪费资源也增加隐私风险。Redis通过 setex 的TTL参数可以自动删除，建议根据业务类型设置30分钟到24小时不等。客服类Agent可以设30分钟，预订流程较长的Agent可以设2小时，并配合前端心跳或用户操作更新 last_active_at 来续期。
内存方案没有自动过期机制，需要自己实现。可以在 AgentState 中记录 last_active_at，再启动一个后台线程定期扫描字典，删除超过阈值未活跃的会话。扫描频率不需要太高，比如每5分钟一次。还要注意字典很大时不要一次性遍历造成阻塞，可以分批处理或使用有序结构存储到期时间。
异常恢复的关键在于避免写坏状态。Agent处理一个请求可能同时调用模型和多个工具，如果中途报错，不能把半成品状态覆盖原来的完好状态。稳妥做法是先处理，再写入，失败时不更新；更复杂的场景可以保存 previous_state 字段，允许前端或系统请求回滚。如果状态已经写入Redis但响应发送失败，用户重试时可能重复执行某一步，这时需要给关键操作增加幂等键。
最后提醒一点，Redis默认不加密，不建议在公网直接暴露。生产环境应设置密码、限制访问来源、开启TLS，并且不要在多轮对话状态里保存明文敏感信息。必要字段脱敏后再写入，能显著降低数据泄露后的影响范围。