Agent状态管理

要点

Agent 状态是某一时刻 Agent 所有运行时数据的快照——它比「记忆」的覆盖面更广，包括会话状态、任务状态和工具状态
记忆系统解决的是「Agent 知道什么」，状态管理解决的是「Agent 现在在做什么、做到哪一步、中间结果是什么」
状态按生命周期分成三类：请求级（单次 HTTP 请求内）、会话级（一次对话内）、任务级（一个多步任务的完整执行过程）
内存存储速度快但 Worker 无状态，请求结束后数据丢失；KV 适合中等规模的键值状态；D1 适合需要查询和事务的复杂状态
状态持久化的核心问题是「什么时候存」——每步都存开销大，只在关键节点存又可能在崩溃时丢失更多进度
状态恢复不只是把数据读回来，还需要处理过期、版本冲突和部分完成的脏状态
多个请求同时操作同一个 Agent 会话时，需要乐观锁或状态版本号来防止并发写入互相覆盖

1. 状态和记忆的区别

上一篇讲了 Agent 的记忆系统——短期记忆、工作记忆、长期记忆，解决的是「Agent 知道什么」的问题。这篇要讲的状态管理，解决的是另一个维度的问题：「Agent 现在在做什么、做到哪一步、中间产出了什么」。

先看一段前面 Agent Loop 的代码：

// src/agent/loop.ts
export async function runAgent(userInput: string, env: Env) {
  const messages: Message[] = [{ role: 'user', content: userInput }]
 
  for (let i = 0; i < 10; i++) {
    const response = await callLLM(messages)
 
    if (response.toolCalls.length === 0) {
      return response.text
    }
 
    messages.push({ role: 'assistant', content: response.text, toolCalls: response.toolCalls })
    for (const toolCall of response.toolCalls) {
      const result = await executeTool(toolCall)
      messages.push({ role: 'tool', toolCallId: toolCall.id, content: JSON.stringify(result) })
    }
  }
}

这个函数执行期间，messages 数组、循环变量 i、每个工具调用的中间结果——这些都是状态。它们只活在内存里，请求结束就消失。

对于一个 5 秒内完成的任务，这没有问题。但当你需要：

一个任务执行了 3 步之后 Worker 被回收了，重新请求时能从第 3 步继续
用户中途关闭页面，回来时看到任务还在跑（或能恢复到断点）
同一个用户在不同设备上发起请求，看到的任务进度是一致的

内存里的临时变量就撑不住了。需要把状态存到外部存储里，这就是状态管理要做的事情。

2. 三种状态类型

Agent 运行过程中的状态，按职责可以分成三类。

会话状态（Session State）

一次用户会话的生命周期数据——会话 ID、用户身份、创建时间、上次活跃时间、会话是否结束。

// src/agent/state/types.ts
export type SessionState = {
  sessionId: string
  userId: string
  status: 'active' | 'paused' | 'completed' | 'expired'
  createdAt: string
  lastActiveAt: string
  messageCount: number
}

会话状态回答的问题是：这个用户当前有没有一个正在进行的会话？会话是什么时候开始的？还在活跃吗？

任务状态（Task State）

一个具体任务的执行进度——任务 ID、所属会话、当前步骤、每步的输入输出、整体状态。

// src/agent/state/types.ts
export type TaskState = {
  taskId: string
  sessionId: string
  goal: string
  status: 'planned' | 'running' | 'waiting_tool' | 'completed' | 'failed'
  currentStep: number
  totalSteps: number
  steps: StepResult[]
  createdAt: string
  updatedAt: string
}
 
export type StepResult = {
  stepIndex: number
  action: string
  input: unknown
  output: unknown
  status: 'pending' | 'running' | 'success' | 'failed'
  error?: string
  startedAt?: string
  completedAt?: string
}

任务状态回答的问题是：这个任务执行到第几步了？每一步的结果是什么？失败了能从哪一步恢复？

工具状态（Tool State）

工具调用过程中产生的临时数据——API 请求的中间结果、分页游标、速率限制计数器、重试次数。

// src/agent/state/types.ts
export type ToolState = {
  toolName: string
  taskId: string
  status: 'idle' | 'calling' | 'success' | 'retrying' | 'failed'
  retryCount: number
  maxRetries: number
  lastResult?: unknown
  lastError?: string
  rateLimitResetAt?: string
}

工具状态回答的问题是：这个工具上一次调用结果是什么？需不需要重试？有没有被限速？

三者的关系：

会话状态（一个用户会话）
  └── 任务状态（会话中的一个任务）
        ├── 工具状态（任务中某个工具的调用记录）
        ├── 工具状态
        └── 工具状态

一个会话可以包含多个任务（用户先让 Agent 查航班，又让 Agent 发邮件），一个任务可以调用多个工具。

3. 内存存储：最快但也最短暂

最直接的方案是把状态放在内存里。前面几篇的 messages 数组就是一种内存状态。稍微正式一点，用一个 Map 来管理：

// src/agent/state/memory-store.ts
import type { SessionState, TaskState } from './types'
 
export class MemoryStateStore {
  private sessions = new Map<string, SessionState>()
  private tasks = new Map<string, TaskState>()
 
  async getSession(sessionId: string): Promise<SessionState | null> {
    return this.sessions.get(sessionId) ?? null
  }
 
  async saveSession(session: SessionState): Promise<void> {
    this.sessions.set(session.sessionId, session)
  }
 
  async getTask(taskId: string): Promise<TaskState | null> {
    return this.tasks.get(taskId) ?? null
  }
 
  async saveTask(task: TaskState): Promise<void> {
    this.tasks.set(task.taskId, task)
  }
 
  async getTasksBySession(sessionId: string): Promise<TaskState[]> {
    return [...this.tasks.values()].filter((t) => t.sessionId === sessionId)
  }
}

内存存储的优点是快，读写都在进程内，没有网络开销。缺点是 Cloudflare Workers 无状态——请求处理完后实例可能被回收，内存里的数据就没了。即使在实例存活期间，后续请求也不一定落到同一个实例上。

内存存储适合两种场景：

单次请求内的临时状态：一个 Agent Loop 执行期间的 messages 数组，请求结束就不需要了
本地开发和测试：在 wrangler dev 里跑，实例不会在请求之间被回收

对于需要跨请求保持的状态，必须引入外部存储。

4. KV 存储：适合中等规模的键值状态

Cloudflare KV 是一个分布式键值存储，读写有网络开销，但数据持久化，不依赖 Worker 实例的生命周期。

设计 key 的结构

KV 只有一个扁平的 key-value 空间，key 的设计直接决定了查询效率。按 类型:标识符 的格式组织：

// src/agent/state/kv-keys.ts
export const keys = {
  session: (sessionId: string) => `agent:session:${sessionId}`,
  task: (taskId: string) => `agent:task:${taskId}`,
  sessionTasks: (sessionId: string) => `agent:session:${sessionId}:tasks`,
  toolState: (taskId: string, toolName: string) => `agent:tool:${taskId}:${toolName}`,
}

实现 KV 状态存储

// src/agent/state/kv-store.ts
import type { SessionState, TaskState, ToolState } from './types'
import { keys } from './kv-keys'
 
export class KVStateStore {
  constructor(private kv: KVNamespace) {}
 
  // --- 会话状态 ---
 
  async getSession(sessionId: string): Promise<SessionState | null> {
    const data = await this.kv.get(keys.session(sessionId))
    return data ? JSON.parse(data) : null
  }
 
  async saveSession(session: SessionState): Promise<void> {
    await this.kv.put(keys.session(session.sessionId), JSON.stringify(session), {
      // 会话 24 小时无活动自动过期
      expirationTtl: 60 * 60 * 24,
    })
  }
 
  // --- 任务状态 ---
 
  async getTask(taskId: string): Promise<TaskState | null> {
    const data = await this.kv.get(keys.task(taskId))
    return data ? JSON.parse(data) : null
  }
 
  async saveTask(task: TaskState): Promise<void> {
    // 同时维护会话的任务索引
    await this.kv.put(keys.task(task.taskId), JSON.stringify(task))
 
    const indexKey = keys.sessionTasks(task.sessionId)
    const indexData = await this.kv.get(indexKey)
    const taskIds: string[] = indexData ? JSON.parse(indexData) : []
    if (!taskIds.includes(task.taskId)) {
      taskIds.push(task.taskId)
      await this.kv.put(indexKey, JSON.stringify(taskIds))
    }
  }
 
  async getTasksBySession(sessionId: string): Promise<TaskState[]> {
    const indexData = await this.kv.get(keys.sessionTasks(sessionId))
    if (!indexData) return []
    const taskIds: string[] = JSON.parse(indexData)
 
    const tasks: TaskState[] = []
    for (const taskId of taskIds) {
      const task = await this.getTask(taskId)
      if (task) tasks.push(task)
    }
    return tasks
  }
 
  // --- 工具状态 ---
 
  async getToolState(taskId: string, toolName: string): Promise<ToolState | null> {
    const data = await this.kv.get(keys.toolState(taskId, toolName))
    return data ? JSON.parse(data) : null
  }
 
  async saveToolState(state: ToolState, taskId: string): Promise<void> {
    await this.kv.put(keys.toolState(taskId, state.toolName), JSON.stringify(state))
  }
}

KV 适合状态数据量不大、按 key 直接读写的场景。但它有两个限制：

不支持复杂查询：不能按条件筛选（比如「查找所有失败的任务」），需要维护额外的索引
没有事务：保存任务和更新索引是两步操作，中间失败了会出现不一致

对于需要事务和查询能力的状态，D1 更合适。

5. D1 存储：需要查询和事务时的选择

当状态数据之间有复杂的关联关系，或者需要按条件查询时，关系型数据库比 KV 更合适。

表结构

-- migrations/0001_create_agent_state.sql
CREATE TABLE agent_sessions (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'active',
  message_count INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL,
  last_active_at TEXT NOT NULL
);
 
CREATE TABLE agent_tasks (
  id TEXT PRIMARY KEY,
  session_id TEXT NOT NULL REFERENCES agent_sessions(id),
  goal TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'planned',
  current_step INTEGER NOT NULL DEFAULT 0,
  total_steps INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL
);
 
CREATE TABLE agent_task_steps (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  task_id TEXT NOT NULL REFERENCES agent_tasks(id),
  step_index INTEGER NOT NULL,
  action TEXT NOT NULL,
  input TEXT,
  output TEXT,
  status TEXT NOT NULL DEFAULT 'pending',
  error TEXT,
  started_at TEXT,
  completed_at TEXT
);
 
CREATE TABLE agent_tool_states (
  task_id TEXT NOT NULL REFERENCES agent_tasks(id),
  tool_name TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'idle',
  retry_count INTEGER NOT NULL DEFAULT 0,
  max_retries INTEGER NOT NULL DEFAULT 3,
  last_result TEXT,
  last_error TEXT,
  rate_limit_reset_at TEXT,
  PRIMARY KEY (task_id, tool_name)
);

D1 状态存储实现

// src/agent/state/d1-store.ts
import type { SessionState, TaskState, StepResult, ToolState } from './types'
import type { DB } from 'db/types'
 
export class D1StateStore {
  constructor(private db: DB) {}
 
  async getSession(sessionId: string): Promise<SessionState | null> {
    const row = await this.db
      .select()
      .from(agentSessions)
      .where(eq(agentSessions.id, sessionId))
      .get()
    return row ?? null
  }
 
  async saveSession(session: SessionState): Promise<void> {
    await this.db
      .insert(agentSessions)
      .values({
        id: session.sessionId,
        userId: session.userId,
        status: session.status,
        messageCount: session.messageCount,
        createdAt: session.createdAt,
        lastActiveAt: session.lastActiveAt,
      })
      .onConflictDoUpdate({
        target: agentSessions.id,
        set: {
          status: session.status,
          messageCount: session.messageCount,
          lastActiveAt: session.lastActiveAt,
        },
      })
  }
 
  async saveTaskWithSteps(task: TaskState): Promise<void> {
    // 事务：任务和步骤一起写入，要么都成功要么都回滚
    await this.db.batch([
      this.db
        .insert(agentTasks)
        .values({
          id: task.taskId,
          sessionId: task.sessionId,
          goal: task.goal,
          status: task.status,
          currentStep: task.currentStep,
          totalSteps: task.totalSteps,
          createdAt: task.createdAt,
          updatedAt: task.updatedAt,
        })
        .onConflictDoUpdate({
          target: agentTasks.id,
          set: {
            status: task.status,
            currentStep: task.currentStep,
            updatedAt: task.updatedAt,
          },
        }),
      ...task.steps.map((step) =>
        this.db
          .insert(agentTaskSteps)
          .values({
            taskId: task.taskId,
            stepIndex: step.stepIndex,
            action: step.action,
            input: JSON.stringify(step.input),
            output: JSON.stringify(step.output),
            status: step.status,
            error: step.error,
            startedAt: step.startedAt,
            completedAt: step.completedAt,
          })
          .onConflictDoNothing(),
      ),
    ])
  }
 
  async getFailedTasks(sessionId: string): Promise<TaskState[]> {
    const rows = await this.db
      .select()
      .from(agentTasks)
      .where(and(eq(agentTasks.sessionId, sessionId), eq(agentTasks.status, 'failed')))
      .all()
    return rows
  }
}

D1 相比 KV 的改进：

事务：db.batch() 把多个写操作包成一个原子操作
查询能力：能按条件筛选（失败的任务、某个用户的所有会话）
关联关系：任务和步骤之间有外键约束，数据完整性由数据库保证

代价是写入延迟比 KV 高。对于高频的状态更新（比如工具调用计数器），KV 更合适；对于需要持久化和查询的核心状态（任务进度、执行历史），D1 更合适。

6. 什么时候存：持久化时机

状态持久化的核心问题不是「存在哪里」，而是「什么时候存」。每步都存开销大、延迟高；只在最后存，中间崩溃了进度全丢。

关键节点持久化

在 Agent 执行的关键节点保存状态，而不是每个操作都存：

// src/agent/state/checkpoint.ts
import type { TaskState } from './types'
import type { KVStateStore } from './kv-store'
 
export class StateCheckpoint {
  private dirty = false
 
  constructor(
    private store: KVStateStore,
    private task: TaskState,
  ) {}
 
  markDirty() {
    this.dirty = true
  }
 
  getTask(): TaskState {
    return this.task
  }
 
  updateStep(stepIndex: number, update: Partial<StepResult>) {
    const step = this.task.steps[stepIndex]
    this.task.steps[stepIndex] = { ...step, ...update }
    this.dirty = true
  }
 
  // 在关键节点调用，而不是每步都调用
  async flush() {
    if (!this.dirty) return
    this.task.updatedAt = new Date().toISOString()
    await this.store.saveTask(this.task)
    this.dirty = false
  }
}

在 Agent Loop 里使用 checkpoint：

// src/agent/loop-with-state.ts
export async function runAgentWithState(
  userInput: string,
  sessionId: string,
  store: KVStateStore,
) {
  // 检查是否有未完成的任务可以恢复
  const existingTask = await findResumableTask(store, sessionId)
  const task = existingTask ?? createNewTask(sessionId, userInput)
  const checkpoint = new StateCheckpoint(store, task)
 
  // 任务创建时保存一次
  if (!existingTask) {
    await checkpoint.flush()
  }
 
  for (let i = task.currentStep; i < task.totalSteps; i++) {
    checkpoint.updateStep(i, { status: 'running', startedAt: new Date().toISOString() })
 
    try {
      const result = await executeStep(task.steps[i])
      checkpoint.updateStep(i, {
        status: 'success',
        output: result,
        completedAt: new Date().toISOString(),
      })
    } catch (err) {
      checkpoint.updateStep(i, {
        status: 'failed',
        error: err instanceof Error ? err.message : String(err),
      })
      // 失败时立即持久化，确保错误状态不丢
      await checkpoint.flush()
      throw err
    }
 
    checkpoint.markDirty()
 
    // 每 3 步持久化一次，平衡性能和安全性
    if (i % 3 === 0) {
      await checkpoint.flush()
    }
  }
 
  // 任务完成时最终持久化
  checkpoint.getTask().status = 'completed'
  await checkpoint.flush()
}

持久化时机的选择：

时机	适用场景	开销
每步都存	步骤耗时长（分钟级）、崩溃代价大	高
每 N 步存	步骤多但每步很快、允许丢失少量进度	中
关键节点存	任务完成后、失败时、工具调用超限时	低
最后才存	一次性短任务，不需要中断恢复	最低

7. 状态恢复：不只是把数据读回来

Worker 被回收后，新的请求带着同样的 sessionId 过来，需要从存储里恢复状态继续执行。恢复比保存更复杂，因为要处理几种脏状态。

过期状态

会话已经超过 TTL 被清理了，但客户端还拿着旧的 sessionId：

// src/agent/state/recovery.ts
export async function loadOrRejectSession(
  sessionId: string,
  store: KVStateStore,
): Promise<{ session: SessionState; tasks: TaskState[] } | { error: string }> {
  const session = await store.getSession(sessionId)
  if (!session) {
    return { error: 'SESSION_EXPIRED' }
  }
 
  // 会话超过 1 小时没有活动，视为过期
  const lastActive = new Date(session.lastActiveAt).getTime()
  if (Date.now() - lastActive > 60 * 60 * 1000) {
    return { error: 'SESSION_EXPIRED' }
  }
 
  const tasks = await store.getTasksBySession(sessionId)
  return { session, tasks }
}

部分完成的任务

任务跑了一半中断了，currentStep 停在中间，有些步骤既没有 success 也没有 failed：

// src/agent/state/recovery.ts
export function findResumableStep(task: TaskState): number | null {
  // 找到第一个未完成的步骤
  const idx = task.steps.findIndex(
    (s) => s.status === 'pending' || s.status === 'running',
  )
  if (idx === -1) return null
 
  // 对于 running 状态的步骤（崩溃时正在执行），需要回退到 pending
  // 因为它可能已经产生了副作用，也可能没有，保守地重新执行
  if (task.steps[idx].status === 'running') {
    task.steps[idx].status = 'pending'
    task.steps[idx].startedAt = undefined
  }
 
  return idx
}

这里有一个需要注意的点：running 状态的步骤可能已经执行了一部分工具调用。比如工具是「发邮件」，Worker 在发邮件的过程中崩溃了——邮件可能已经发出去了，也可能没有。对于这种幂等性不确定的操作，有两种处理方式：

保守重试：把 running 回退到 pending，重新执行。如果工具是幂等的（比如「查询航班」），重试没有副作用
标记跳过：让用户决定是重试还是跳过。适合非幂等操作（比如「发邮件」「扣款」）

版本号冲突

两个请求几乎同时读到同一个状态版本，各自修改后写回，后写入的会覆盖先写入的。这在 Agent 场景下不太常见（一个会话通常只有一个活跃的 Agent 在执行），但如果有前端轮询更新状态的情况，就可能触发。

加一个 version 字段做乐观锁：

// src/agent/state/versioned-store.ts
export type VersionedTaskState = TaskState & {
  version: number
}
 
export async function saveTaskWithVersion(
  kv: KVNamespace,
  task: VersionedTaskState,
): Promise<{ success: boolean; currentVersion?: number }> {
  const key = `agent:task:${task.taskId}`
  const existing = await kv.get(key)
 
  if (existing) {
    const parsed = JSON.parse(existing) as VersionedTaskState
    if (parsed.version !== task.version) {
      // 版本号不匹配，说明有其他请求已经修改了状态
      return { success: false, currentVersion: parsed.version }
    }
  }
 
  const nextVersion = task.version + 1
  await kv.put(key, JSON.stringify({ ...task, version: nextVersion }))
  return { success: true }
}

写入时检查版本号，不一致就拒绝，让调用方重新读取最新状态再重试。

8. 并发状态管理

一个用户不太会同时跑两个 Agent 任务，但以下场景会产生并发：

前端页面和后台 Worker 同时更新同一个任务的状态
用户在一个设备上发消息，同时在另一个设备上查看进度
定时任务（比如 Cron Trigger）和用户请求同时操作会话状态

用 KV 的 conditional put 做简单互斥

Cloudflare KV 支持 metadata 和条件写入，可以用来做简单的乐观锁：

// src/agent/state/concurrent.ts
export async function acquireTaskLock(
  kv: KVNamespace,
  taskId: string,
  holderId: string,
  ttlSeconds = 30,
): Promise<boolean> {
  const lockKey = `agent:lock:${taskId}`
 
  // 尝试写入锁，只在该 key 不存在时成功
  const result = await kv.put(lockKey, holderId, {
    expirationTtl: ttlSeconds,
    metadata: { holder: holderId },
  })
 
  // 检查当前锁的持有者
  const current = await kv.getWithMetadata<{ holder: string }>(lockKey)
  if (current.metadata?.holder === holderId) {
    return true
  }
 
  // 锁被其他 holder 持有
  return false
}
 
export async function releaseTaskLock(
  kv: KVNamespace,
  taskId: string,
  holderId: string,
): Promise<void> {
  const lockKey = `agent:lock:${taskId}`
  const current = await kv.getWithMetadata<{ holder: string }>(lockKey)
 
  // 只释放自己持有的锁
  if (current.metadata?.holder === holderId) {
    await kv.delete(lockKey)
  }
}

锁设置了 TTL，即使 Worker 崩溃没有主动释放，锁也会自动过期。TTL 的值要大于单个步骤的最大执行时间。

在 Agent Loop 中使用锁

// src/agent/loop-with-lock.ts
export async function runAgentWithLock(
  userInput: string,
  sessionId: string,
  env: Env,
) {
  const holderId = crypto.randomUUID()
  const store = new KVStateStore(env.KV)
 
  // 获取或创建任务
  const task = await getOrCreateTask(store, sessionId, userInput)
 
  // 尝试获取锁
  const locked = await acquireTaskLock(env.KV, task.taskId, holderId, 60)
  if (!locked) {
    // 有其他实例正在执行这个任务
    return { status: 'conflict', message: '任务正在执行中，请稍后再试' }
  }
 
  try {
    // 执行 Agent Loop
    const result = await executeAgentLoop(task, store)
    return { status: 'completed', result }
  } finally {
    // 无论成功还是失败，都要释放锁
    await releaseTaskLock(env.KV, task.taskId, holderId)
  }
}

finally 块确保锁一定会被释放。如果 Worker 在 finally 之前就崩溃了，锁会在 60 秒后自动过期。

总结

回顾一下这篇的要点：

Agent 状态包括会话状态、任务状态和工具状态三类，分别管理不同生命周期的数据
内存存储适合单次请求内的临时状态，KV 适合跨请求的键值状态，D1 适合需要查询和事务的复杂状态
持久化时机需要在性能和安全性之间权衡——每步存开销大，最后存风险高，关键节点存是比较折中的选择
状态恢复需要处理三种脏状态：过期会话、部分完成的任务、版本号冲突
并发操作用乐观锁或分布式锁做互斥，锁要设 TTL 防止死锁

状态管理和上一篇的记忆系统有交叉但侧重不同：记忆系统管的是「Agent 知道什么」，状态管理管的是「Agent 正在做什么」。两者配合起来，Agent 才能在崩溃后恢复进度、在跨设备间保持一致、在长时间任务中不丢失中间结果。

下一篇来看 Agent 的错误处理与恢复机制——当 LLM 返回异常、工具调用超时、状态恢复失败时，怎样让 Agent 有策略地降级而不是直接崩溃。