Agent03：状态管理
理解 Agent 的状态管理能力，包括短期上下文、长期记忆、任务进度和中间产物，让 Agent 能够处理长时任务和多轮对话。
字数 1558
阅读时长 ≈ 5 分钟
2026-7-15 2026-7-27 状态管理是 Agent 的核心能力之一，它让 Agent 能够在多轮对话和长时任务中保持信息，避免重复计算和遗忘关键信息。理解状态管理的机制，才能构建能够处理复杂场景的 Agent 系统。
状态管理的基本原理
什么是状态管理
状态管理是指 Agent 在执行任务过程中，对信息进行存储、更新和检索的能力：
| 状态类型 | 说明 | 示例 |
|---|
| 短期上下文 | 当前对话的信息 | 用户的当前问题、历史对话 |
| 长期记忆 | 跨对话的持久化信息 | 用户偏好、历史任务记录 |
| 任务进度 | 当前任务的执行状态 | 已完成步骤、待执行步骤 |
| 中间产物 | 任务执行中的临时结果 | API 返回数据、计算结果 |
状态管理的必要性
| 问题 | 说明 | 后果 |
|---|
| 遗忘上下文 | 不记得之前的对话内容 | 重复提问、回答不一致 |
| 重复计算 | 相同问题重复处理 | 效率低下、成本增加 |
| 任务中断 | 无法恢复中断的任务 | 用户体验差 |
| 信息丢失 | 中间结果未保存 | 需要重新执行 |
状态管理的挑战
| 挑战 | 说明 | 应对策略 |
|---|
| 上下文窗口限制 | 模型上下文窗口有限 | 上下文压缩、摘要 |
| 存储成本 | 大量状态需要存储 | 分层存储、过期清理 |
| 检索效率 | 需要快速检索相关状态 | 索引、缓存 |
| 一致性 | 多轮对话中状态保持一致 | 版本控制、冲突检测 |
短期上下文管理
上下文的类型
| 类型 | 说明 | 示例 |
|---|
| 用户输入 | 用户的问题和指令 | ”帮我订一张机票” |
| 系统提示 | 系统给模型的指令 | ”你是一个旅行助手” |
| 工具调用 | 工具调用的记录 | 调用了航班查询 API |
| 工具结果 | 工具返回的结果 | 航班列表数据 |
| 对话历史 | 之前的对话内容 | 多轮对话记录 |
上下文管理策略
class ContextManager:
def __init__(self, max_tokens=8192):
self.max_tokens = max_tokens
self.context = []
def add_message(self, role, content):
"""添加消息到上下文"""
self.context.append({"role": role, "content": content})
self.trim_context()
def trim_context(self):
"""修剪上下文，确保不超过 token 限制"""
total_tokens = sum(len(msg["content"]) // 4 for msg in self.context)
while total_tokens > self.max_tokens and len(self.context) > 1:
# 移除最早的非系统消息
removed = self.context.pop(1)
total_tokens -= len(removed["content"]) // 4
def get_context(self):
"""获取当前上下文"""
return self.context
def clear(self):
"""清空上下文"""
# 保留系统提示
system_prompt = self.context[0] if self.context else None
self.context = []
if system_prompt:
self.context.append(system_prompt)
上下文压缩技术
| 技术 | 说明 | 适用场景 |
|---|
| 摘要压缩 | 生成对话摘要 | 长对话 |
| 关键信息提取 | 提取关键信息 | 需要保留特定信息 |
| 分层保留 | 保留最近的详细信息，早期的摘要 | 多轮对话 |
| 选择性保留 | 根据相关性保留信息 | 需要过滤无关内容 |
长期记忆
长期记忆的类型
| 类型 | 说明 | 示例 |
|---|
| 用户画像 | 用户的基本信息和偏好 | 姓名、偏好的航空公司 |
| 历史记录 | 用户的历史对话和任务 | 之前订过的机票 |
| 知识图谱 | 领域知识和实体关系 | 城市、机场、航班信息 |
| 经验规则 | 从经验中学习的规则 | 用户喜欢靠窗座位 |
长期记忆的存储
class LongTermMemory:
def __init__(self, storage_backend):
self.storage = storage_backend
def save_user_profile(self, user_id, profile):
"""保存用户画像"""
self.storage.set(f"user:{user_id}:profile", profile)
def get_user_profile(self, user_id):
"""获取用户画像"""
return self.storage.get(f"user:{user_id}:profile")
def save_task_history(self, user_id, task):
"""保存任务历史"""
history = self.storage.get(f"user:{user_id}:tasks") or []
history.append(task)
self.storage.set(f"user:{user_id}:tasks", history[-50:]) # 只保留最近50个
def get_task_history(self, user_id, limit=10):
"""获取任务历史"""
history = self.storage.get(f"user:{user_id}:tasks") or []
return history[-limit:]
def save_knowledge(self, key, knowledge):
"""保存知识"""
self.storage.set(f"knowledge:{key}", knowledge)
def get_knowledge(self, key):
"""获取知识"""
return self.storage.get(f"knowledge:{key}")
长期记忆的检索
| 方法 | 说明 | 适用场景 |
|---|
| 关键词检索 | 根据关键词查找记忆 | 精确匹配 |
| 向量检索 | 根据语义相似度查找 | 模糊匹配 |
| 时间检索 | 根据时间查找 | 近期任务 |
| 类型检索 | 根据类型查找 | 用户画像、任务历史 |
任务进度管理
任务状态定义
| 状态 | 说明 | 示例 |
|---|
| pending | 任务待执行 | 用户提交了任务但还没开始 |
| in_progress | 任务执行中 | 正在查询航班 |
| paused | 任务暂停 | 等待用户确认 |
| completed | 任务完成 | 机票已预订成功 |
| failed | 任务失败 | 支付失败 |
| cancelled | 任务取消 | 用户取消了任务 |
任务进度管理实现
class TaskProgressManager:
def __init__(self):
self.tasks = {}
def create_task(self, user_id, goal, steps):
"""创建任务"""
task_id = f"{user_id}_{datetime.now().timestamp()}"
task = {
"task_id": task_id,
"user_id": user_id,
"goal": goal,
"steps": steps,
"current_step": 1,
"status": "pending",
"created_at": datetime.now().isoformat(),
"updated_at": datetime.now().isoformat(),
"results": {}
}
self.tasks[task_id] = task
return task_id
def update_task(self, task_id, updates):
"""更新任务状态"""
if task_id in self.tasks:
self.tasks[task_id].update(updates)
self.tasks[task_id]["updated_at"] = datetime.now().isoformat()
return True
return False
def complete_step(self, task_id, step_number, result):
"""完成一个步骤"""
if task_id not in self.tasks:
return False
task = self.tasks[task_id]
task["results"][step_number] = result
task["current_step"] = step_number + 1
if task["current_step"] > len(task["steps"]):
task["status"] = "completed"
task["updated_at"] = datetime.now().isoformat()
return True
def get_task(self, task_id):
"""获取任务信息"""
return self.tasks.get(task_id)
def get_user_tasks(self, user_id, status=None):
"""获取用户的任务列表"""
user_tasks = [t for t in self.tasks.values() if t["user_id"] == user_id]
if status:
user_tasks = [t for t in user_tasks if t["status"] == status]
return sorted(user_tasks, key=lambda x: x["updated_at"], reverse=True)
任务进度可视化
| 方式 | 说明 | 示例 |
|---|
| 进度条 | 显示任务完成比例 | 50% |
| 步骤列表 | 显示每个步骤的状态 | 步骤1 ✓ 步骤2 ○ 步骤3 ○ |
| 时间线 | 显示任务执行时间线 | 时间轴 |
| 状态卡片 | 显示任务当前状态 | 执行中、已完成 |
中间产物管理
中间产物的类型
| 类型 | 说明 | 示例 |
|---|
| API 响应 | 外部 API 返回的数据 | 航班列表 |
| 计算结果 | 计算产生的结果 | 价格汇总 |
| 文档内容 | 生成或处理的文档 | 周报文档 |
| 图片数据 | 生成或处理的图片 | 图表图片 |
中间产物存储策略
class IntermediateProductManager:
def __init__(self, cache_backend, storage_backend):
self.cache = cache_backend # 短期缓存
self.storage = storage_backend # 长期存储
def cache_result(self, key, result, ttl=300):
"""缓存中间结果"""
self.cache.set(key, result, ttl)
def get_cached_result(self, key):
"""获取缓存结果"""
return self.cache.get(key)
def persist_result(self, task_id, step, result):
"""持久化中间结果"""
self.storage.set(f"task:{task_id}:step:{step}:result", result)
def get_persisted_result(self, task_id, step):
"""获取持久化结果"""
return self.storage.get(f"task:{task_id}:step:{step}:result")
def cleanup(self, task_id):
"""清理任务的中间产物"""
# 清理缓存
self.cache.delete_pattern(f"task:{task_id}:*")
# 清理持久化存储（可选）
# self.storage.delete_pattern(f"task:{task_id}:*")
中间产物复用
| 策略 | 说明 | 适用场景 |
|---|
| 缓存复用 | 相同请求直接使用缓存 | 重复查询 |
| 结果共享 | 多个任务共享结果 | 公共数据 |
| 增量更新 | 只更新变化的部分 | 数据更新 |
| 版本控制 | 保留历史版本 | 需要回溯 |
状态管理的常见问题
问题1：上下文超限
表现：对话历史超过模型上下文窗口限制
解决方案：
问题2：记忆冲突
表现：不同对话中的信息产生冲突
解决方案：
问题3：检索效率低
表现：从长期记忆中检索信息太慢
解决方案：
问题4：状态不一致
表现：任务状态与实际执行情况不一致
解决方案：
- 添加状态同步机制
- 实现状态校验
- 使用事务保证一致性
状态管理的最佳实践
状态管理架构
┌─────────────────────────────────────────────────────────┐
│ 状态管理层 │
├─────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 短期上下文 │ │ 长期记忆 │ │ 任务进度 │ │
│ │ Context │ │ Long-term │ │ Task │ │
│ │ Manager │ │ Memory │ │ Progress │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
└─────────┼────────────────┼────────────────┼─────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────┐
│ 存储层 │
├─────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 内存缓存 │ │ Redis │ │ 数据库 │ │
│ │ (短期) │ │ (中期) │ │ (长期) │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────┘
状态管理流程
1. 状态初始化
- 创建任务状态
- 加载用户记忆
- 初始化上下文
2. 状态更新
- 添加对话内容到上下文
- 保存用户偏好到长期记忆
- 更新任务进度
- 存储中间产物
3. 状态检索
- 获取当前上下文
- 查询用户历史记录
- 读取任务进度
- 恢复中间产物
4. 状态清理
- 压缩过期上下文
- 删除过期缓存
- 清理完成任务的中间产物
评估指标
| 指标 | 定义 | 目标值 |
|---|
| 上下文命中率 | 从上下文中找到所需信息的比例 | > 90% |
| 记忆检索延迟 | 从长期记忆中检索信息的时间 | < 100ms |
| 状态一致性 | 状态与实际情况一致的比例 | > 99% |
| 存储效率 | 有效存储占总存储的比例 | > 80% |
项目判断清单
- 需要多轮对话 → 实现短期上下文管理
- 需要记住用户偏好 → 实现长期记忆
- 需要处理长时任务 → 实现任务进度管理
- 需要复用计算结果 → 实现中间产物管理
- 上下文超限 → 实现上下文压缩
- 检索效率低 → 使用索引和缓存
- 状态不一致 → 添加状态校验
- 需要持久化 → 使用数据库存储
