持久 Agent

加入版本： @mastra/core@1.45.0

注意

持久 Agent 目前处于 beta 阶段。API 可能在未来版本中发生变化。

持久 Agent 会封装常规 Agent，使 Agent 循环在 Workflow 中运行。事件通过 PubSub 流动，因此客户端可以断开并重新连接，而不会错过 chunk。运行状态会持久化，因此可在进程重启后继续存在。

何时使用持久 Agent

存在以下任一情况时，请使用持久 Agent：

• 客户端可能在流式传输中途断开并重新连接（移动设备、网络不稳定、长时间运行的调用）。
• Agent 循环可能比单个 HTTP 请求存续更久（后台研究、多步骤 Tool 使用）。
• 需要 observe/reconnect API，让第二个客户端接续第一个客户端启动的 stream。
• 希望使用由 Inngest 提供支持的执行，获得步骤 memoization、重试和监控。

对于客户端会保持连接、生命周期较短且限定于请求的调用，使用常规 Agent 的 stream() 或 generate() 更简单。

快速入门

使用 @mastra/core/agent/durable 中的 createDurableAgent() 封装现有 Agent：

src/mastra/agents/researcher.ts

import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'researcher',
  name: 'Researcher',
  instructions: 'You research topics thoroughly.',
  model: 'openai/gpt-5.6-sol',
})

export const durableResearcher = createDurableAgent({ agent })

向 Mastra 注册持久 Agent，然后调用 stream()：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { durableResearcher } from './agents/researcher'

const mastra = new Mastra({
  agents: { durableResearcher },
})

const { output, runId, cleanup } = await durableResearcher.stream(
  'Research quantum computing advances in 2025',
)

for await (const chunk of output.fullStream) {
  // Process each chunk as it arrives
}

// Release PubSub subscriptions and clear the run from the registry.
// If you skip this, an automatic cleanup timer fires after the stream ends.
cleanup()

返回的 runId 用于标识本次执行。将其传给 observe()，即可从其他客户端重新连接。有关完整配置和方法 API，请参阅 DurableAgent Reference。

工作原理

持久 Agent 在常规 Agent 之上添加三层能力：

1. Workflow 执行：stream() 将消息和选项序列化为 Workflow 输入，然后在持久 Workflow 内触发 Agent 循环。Workflow 运行与 Agent.stream() 相同的循环，但每个步骤都可以 memoize 和 replay。

2. PubSub 流式传输：循环运行时，chunk 会发布到以运行 ID 为键的 PubSub topic。调用方订阅该 topic，并将 chunk 传入 ReadableStream。如果调用方断开并重新连接，cache 会 replay 错过的 chunk。

3. Cache 层：可选的 cache（默认为内存；生产环境可使用 Redis 或其他后端）存储已发布事件，使稍后订阅的客户端可以接续。

执行变体

Mastra 提供三个用于创建持久 Agent 的 factory 函数。它们的区别在于 Workflow 的执行方式：

Factory	包	最适合
createDurableAgent()	@mastra/core	本地开发和单进程 Server。可以直接等待返回的 stream。
createEventedAgent()	@mastra/core	后台执行。Workflow 启动时不会阻塞，通过 PubSub 消费 chunk。
createInngestAgent()	@mastra/inngest	生产部署。Inngest 提供步骤 memoization、重试和监控 dashboard。

三者都返回一个对象，可以像常规 Agent 一样向 Mastra 注册。createDurableAgent() 和 createEventedAgent() 返回扩展 Agent 的类实例。createInngestAgent() 返回由 Proxy 支持的对象，将 Agent 方法转发给底层 Agent。

使用 createDurableAgent() 在进程内运行

封装 Agent 并调用 stream()。在同一进程中会返回 DurableAgentStreamResult。不需要外部基础设施，因此这是最快的入门方式：

src/mastra/agents/durable.ts

import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'helper',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5.6-sol',
})

export const durableHelper = createDurableAgent({ agent })

使用 createEventedAgent() 触发后即不再等待

Workflow 会在后台启动，不会阻塞调用方。仍会通过 PubSub 接收 chunk，因此 stream() 会返回可供消费的结果。触发运行的 HTTP handler 无需等待 Workflow 完成：

src/mastra/agents/evented.ts

import { Agent } from '@mastra/core/agent'
import { createEventedAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'writer',
  instructions: 'You write articles.',
  model: 'openai/gpt-5.6-sol',
})

export const eventedWriter = createEventedAgent({ agent })

使用 createInngestAgent() 通过 Inngest 运行

在 Inngest platform 上运行 Workflow。每次 Tool 调用都会成为可 memoize 的步骤，Inngest 可以单独重试这些步骤，并提供用于监控运行的 dashboard：

src/mastra/agents/inngest.ts

import { Agent } from '@mastra/core/agent'
import { createInngestAgent } from '@mastra/inngest'
import { Inngest } from 'inngest'

const inngest = new Inngest({ id: 'my-app' })

const agent = new Agent({
  id: 'analyst',
  instructions: 'You analyze data.',
  model: 'openai/gpt-5.6-sol',
})

export const inngestAnalyst = createInngestAgent({ agent, inngest })

有关完整 API（包括 PubSub 和 cache 配置等 Inngest 特定选项），请参阅 createInngestAgent() Reference。

可恢复的 stream

持久 Agent 通过 PubSub 和事件 cache 支持可恢复的 stream。客户端在流式传输中途断开时，cache 会继续存储事件。同一客户端可以使用 runId 调用 observe() 重新连接：

src/server/reconnect.ts

const { output, cleanup } = await durableResearcher.observe(runId)

for await (const chunk of output.fullStream) {
  // Chunks from the run, including any missed while disconnected
}

cleanup()

createDurableAgent() 和 createEventedAgent() 默认使用内存 cache，因此可恢复 stream 仅适用于单个进程。生产环境中，请提供持久化 cache 后端（例如 Redis），使缓存事件在进程重启后继续存在：

src/mastra/agents/durable-with-cache.ts

import { createDurableAgent } from '@mastra/core/agent/durable'
import { RedisServerCache } from '@mastra/redis'
import Redis from 'ioredis'

const cache = new RedisServerCache({ client: new Redis('redis://localhost:6379') })

export const durableAgent = createDurableAgent({
  agent,
  cache,
})

createInngestAgent() 默认不启用 cache。传入 cache 选项，或向配置了 serverCache 的 Mastra 实例注册 Agent，即可启用可恢复 stream。

通过后台任务进行流式传输

持久 Agent 支持与常规 Agent 相同的 untilIdle 选项。设置 untilIdle 后，stream() 会在后台任务继续执行期间保持连接打开，直到 Agent 空闲：

src/mastra/run.ts

const { output, cleanup } = await durableAgent.stream('Research and summarize the topic', {
  untilIdle: true,
  memory: { thread: 'thread-1', resource: 'user-1' },
})

for await (const chunk of output.fullStream) {
  // Chunks from the initial turn AND any follow-up turns triggered by
  // background task completions
}

cleanup()

传入 { maxIdleMs } 可自定义空闲超时（默认为 5 分钟）：

await durableAgent.stream('Research topic', {
  untilIdle: { maxIdleMs: 30_000 },
  memory: { thread: 'thread-1', resource: 'user-1' },
})

有关完整后台任务指南（包括配置、subagent 和暂停/恢复），请参阅后台任务。

清理

每次 stream() 和 observe() 调用都会返回 cleanup 函数。调用它会取消 PubSub 订阅，并从内部 Registry 移除运行。如果忘记调用，stream 结束后会触发自动 timer；但自行调用 cleanup() 可以立即释放资源。

Tool 审批

持久 Agent 支持 Tool 审批（human-in-the-loop）。当 Tool 调用需要审批时，Workflow 会暂停、发出 onSuspended 回调，并等待调用方使用 resume() 恢复：

src/mastra/run.ts

const { output, runId, cleanup } = await durableAgent.stream('Delete the old records', {
  requireToolApproval: true,
  onSuspended: ({ toolCallId, toolName, args }) => {
    // Notify the user and ask for approval
  },
})

批准后恢复暂停的运行：

await durableAgent.resume(runId, { approved: true })

崩溃恢复

如果 Server 进程在持久 Agent 运行期间崩溃，该运行会在 Storage 中保持 running 状态，且不会自动重试。下次 Server 启动时，可以重新驱动这些孤立运行，使其从停止的位置继续。

自动恢复

在 Mastra 配置中将 recovery.durableAgents 设为 'auto'。Deployer 会在启动时调用 recoverAllDurableAgents()，紧接着重新启动活跃的 Workflow 运行：

src/mastra/index.ts

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
  storage: new PostgresStore({ connectionString: process.env.DATABASE_URL! }),
  recovery: { durableAgents: 'auto' },
})

启动时，系统会发现每个注册的持久 Agent 中停留在 running 状态的运行，并从最后一个持久化快照重新驱动。

注意

恢复会从最后一个快照重新运行 Agent 循环，因此会重新发出 LLM 调用（产生实际成本），并重新执行 Tool 调用。启用自动恢复前，请确保 Tool 具有幂等性。

手动恢复

如果需要更精细的控制，例如通过 leader election 限制恢复，或定时运行恢复，请直接调用相应方法：

// Recover all durable agents
const result = await mastra.recoverAllDurableAgents()
console.log(`Recovered ${result.recovered} runs (${result.succeeded} ok, ${result.failed} failed)`)

// Recover a specific agent
const agentResult = await durableAgent.recoverActiveRuns()

// Recover a single known run
await durableAgent.recoverActiveRuns({ runId: 'run-abc-123' })

多实例部署

Mastra 尚未提供分布式 lease 或 lock。在多副本部署中，每个以 recovery.durableAgents: 'auto' 启动的副本都会竞相恢复相同的运行。目前，请通过自己的 leader election 限制恢复，或只从单个副本运行恢复。

相关内容

• DurableAgent Reference
• createInngestAgent() Reference
• 后台任务
• Inngest 部署指南
• Agent 概述
• Worker 概述