> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# 持久 Agent

**加入版本：** `@mastra/core@1.45.0`

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

持久 Agent 会封装常规 [`Agent`](https://mastra.zisheng.pro/docs/agents/overview)，使 Agent 循环在 Workflow 中运行。事件通过 [PubSub](https://mastra.zisheng.pro/docs/server/pubsub) 流动，因此客户端可以断开并重新连接，而不会错过 chunk。运行状态会持久化，因此可在进程重启后继续存在。

## 何时使用持久 Agent

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

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

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

## 快速入门

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

```typescript
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()`：

```typescript
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](https://mastra.zisheng.pro/reference/agents/durable-agent)。

## 工作原理

持久 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`。不需要外部基础设施，因此这是最快的入门方式：

```typescript
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 完成：

```typescript
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](https://www.inngest.com/docs) platform 上运行 Workflow。每次 Tool 调用都会成为可 memoize 的步骤，Inngest 可以单独重试这些步骤，并提供用于监控运行的 dashboard：

```typescript
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](https://mastra.zisheng.pro/reference/agents/inngest-agent)。

## 可恢复的 stream

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

```typescript
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），使缓存事件在进程重启后继续存在：

```typescript
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`](https://mastra.zisheng.pro/reference/streaming/agents/stream) 选项。设置 `untilIdle` 后，`stream()` 会在后台任务继续执行期间保持连接打开，直到 Agent 空闲：

```typescript
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 分钟）：

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

有关完整后台任务指南（包括配置、subagent 和暂停/恢复），请参阅[后台任务](https://mastra.zisheng.pro/docs/long-running-agents/background-tasks)。

## 清理

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

## Tool 审批

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

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

批准后恢复暂停的运行：

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

## 崩溃恢复

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

### 自动恢复

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

```typescript
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 限制恢复，或定时运行恢复，请直接调用相应方法：

```typescript
// 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](https://mastra.zisheng.pro/reference/agents/durable-agent)
- [`createInngestAgent()` Reference](https://mastra.zisheng.pro/reference/agents/inngest-agent)
- [后台任务](https://mastra.zisheng.pro/docs/long-running-agents/background-tasks)
- [Inngest 部署指南](https://mastra.zisheng.pro/guides/deployment/inngest)
- [Agent 概述](https://mastra.zisheng.pro/docs/agents/overview)
- [Worker 概述](https://mastra.zisheng.pro/docs/deployment/workers)