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

# DurableAgent

`DurableAgent` 會以持久執行與可恢復 stream 包裝現有的 [`Agent`](https://mastra.zisheng.pro/zh-TW/reference/agents/agent)。它會執行 Agent 迴圈，讓 client 中斷連線後仍可重新連線而不漏掉 event，並透過 [PubSub](https://mastra.zisheng.pro/zh-TW/docs/server/pubsub) 串流這些 event。若 run 必須比單一 request 存續更久，或要在連線中斷後繼續，請使用此類別。

使用 [`createDurableAgent`](#createdurableagentoptions) factory 建立執行個體；若要在內建 Workflow 引擎上進行傳送後不等待結果的執行，請使用 [`createEventedAgent`](#createeventedagentoptions)。若要由 Inngest 支援執行，請使用 `@mastra/inngest` 的 [`createInngestAgent`](https://mastra.zisheng.pro/zh-TW/reference/agents/inngest-agent)。

## 使用範例

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

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

const durableAgent = createDurableAgent({ agent })

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
})
```

串流回應並讀取結果。run 使用完畢後，`cleanup` 函式會取消 PubSub 訂閱：

```typescript
const { output, runId, cleanup } = await durableAgent.stream('Hello!')

const text = await output.text

cleanup()
```

### 使用 `durable` 設定旗標

在 `AgentConfig` 上設定 `durable: true`；將 Agent 附加至 `Mastra` 執行個體時，系統便會自動以 `createDurableAgent` 包裝。若要轉送 `cache`、`pubsub`、`maxSteps` 或 `cleanupTimeoutMs` 等進階選項，請使用物件。

```typescript
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'

const myAgent = new Agent({
  id: 'my-agent',
  name: 'My Agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
  durable: true, // or: { maxSteps: 10, cleanupTimeoutMs: 60_000 }
})

export const mastra = new Mastra({
  agents: { myAgent },
})
```

`mastra.getAgent('myAgent')` 會回傳包裝後的 `DurableAgent`。獨立 Agent（已建構但未在 `Mastra` 執行個體上註冊）不會成為持久 Agent。包裝會在註冊時套用。

## `createDurableAgent(options)`

以持久執行與可恢復 stream 包裝 `Agent`。這是建立 `DurableAgent` 的建議方式。

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

const durableAgent = createDurableAgent({ agent })
```

回傳： `DurableAgent`

### 參數

**agent** (`Agent`): 要以持久執行能力包裝的 Agent。Agent 方法會委派給此 Agent。

**id** (`string`): ID 覆寫值。 (Default: `agent.id`)

**name** (`string`): 名稱覆寫值。 (Default: `agent.name`)

**cache** (`MastraServerCache | false`): 儲存 stream event 的快取，可啟用可恢復 stream。若省略，Agent 會繼承 Mastra 執行個體的快取，或使用 InMemoryServerCache。設為 false 可停用快取，stream 將無法恢復。

**pubsub** (`PubSub`): 用於串流 event 的 PubSub 執行個體。 (Default: `EventEmitterPubSub`)

**maxSteps** (`number`): Agent 迴圈的最大步驟數。

## `createEventedAgent(options)`

在內建 Workflow 引擎上，以傳送後不等待結果的持久執行方式包裝 `Agent`。和 `createDurableAgent` 一樣，它會回傳可供串流的結果；但底層 Workflow 會以非阻塞方式（透過 `startAsync`）執行，而不是先執行至完成再連接 stream。若希望 run 獨立於呼叫端繼續進行，請使用此函式。它不接受 `id` 或 `name` 覆寫值。

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

const eventedAgent = createEventedAgent({ agent })
```

回傳：`EventedAgent`（`DurableAgent` 的 subclass）

### 參數

**agent** (`Agent`): 要以 event 驅動持久執行能力包裝的 Agent。

**cache** (`MastraServerCache | false`): 儲存 stream event 的快取，可啟用可恢復 stream。若省略，Agent 會繼承 Mastra 執行個體的快取，或使用 InMemoryServerCache。設為 false 可停用快取。

**pubsub** (`PubSub`): 用於串流 event 的 PubSub 執行個體。 (Default: `EventEmitterPubSub`)

**maxSteps** (`number`): Agent 迴圈的最大步驟數。

## constructor 參數

`DurableAgent` 類別接受與 `createDurableAgent` 相同的選項，另加 `cleanupTimeoutMs`。除非需要建立 subclass，否則建議使用 factory。

**agent** (`Agent`): 要以持久執行能力包裝的 Agent。

**id** (`string`): ID 覆寫值。 (Default: `agent.id`)

**name** (`string`): 名稱覆寫值。 (Default: `agent.name`)

**cache** (`MastraServerCache | false`): 儲存 stream event 的快取。若省略，會繼承 Mastra 執行個體的快取或使用 InMemoryServerCache。設為 false 可停用快取。

**pubsub** (`PubSub`): 用於串流 event 的 PubSub 執行個體。 (Default: `EventEmitterPubSub`)

**maxSteps** (`number`): Agent 迴圈的最大步驟數。

**cleanupTimeoutMs** (`number`): stream 完成或發生錯誤後，自動清理 registry 項目前的寬限時間（毫秒）。設為 0 可停用自動清理，並要求手動呼叫 cleanup()。自動清理不會在暫停 event 時觸發。 (Default: `30000`)

## 方法

### 執行

#### `stream(messages, options?)`

使用持久執行串流回應。立即回傳結果；隨 run 進行，結果的 `output` 會產生 event。

```typescript
const { output, runId, cleanup } = await durableAgent.stream('Hello!', {
  onChunk: chunk => console.log(chunk),
  onFinish: result => console.log('done', result),
})

const text = await output.text
cleanup()
```

回傳： [`Promise<DurableAgentStreamResult>`](#durableagentstreamresult)

#### `resume(runId, resumeData, options?)`

恢復已暫停的 run，例如 Tool 核准後。請傳入原始 stream 的 `runId`，以及 run 等待的資料。若 registry 中沒有該 run 的項目，便會擲回錯誤。

```typescript
const { output, cleanup } = await durableAgent.resume(runId, {
  approved: true,
})

await output.text
cleanup()
```

回傳： [`Promise<DurableAgentStreamResult>`](#durableagentstreamresult)

#### `observe(runId, options?)`

重新連線至現有 run；先重播快取 event，再傳送即時 event。網路中斷後請使用此方法。傳入 `offset` 可從已知位置開始重播。

```typescript
const { output, cleanup } = await durableAgent.observe(runId, {
  offset: 0,
  onChunk: chunk => console.log(chunk),
})

await output.text
```

`observe()` 預設會無限期等待 event。若執行 run 的處理程序意外停止，run 會停止產生 event，但絕不會發出完成 event，因此受觀察的 stream 會永遠等待。傳入 `idleTimeoutMs` 可限制等待時間：靜默達到指定毫秒數後，stream 便會結束。系統會先執行選用的 `isAlive` 檢查。run 仍在處理中時（例如長時間執行的 Tool 呼叫，或 run 暫停並等待人工輸入），請回傳 `true` 以繼續等待。回傳 `false` 或省略 `isAlive`，stream 會以錯誤結束。`isAlive` 暫時擲回錯誤會視為「仍在運作」，因此短暫的檢查失敗不會結束即時 stream。

```typescript
const { output } = await durableAgent.observe(runId, {
  idleTimeoutMs: 30_000,
  isAlive: () => runHeartbeat.isFresh(runId),
})
```

因閒置逾時而結束 run 時，會執行與 run 發生錯誤時相同的清理作業（請參閱下方警告），因此會釋放快取狀態，而非保留。兩個選項皆需明確啟用。省略即可使用原本的無限期等待行為。

回傳： `Promise<DurableAgentStreamResult>`

> **警告:** `observe()` 回傳的 `cleanup()` 會刪除 run 的 registry 項目與快取 event。只有在 run 使用完畢後才能呼叫。若 run 已暫停且您打算稍後恢復，請勿呼叫 `cleanup()`；請讓自動清理計時器在 run 完成或發生錯誤後處理。自動清理不會在暫停 event 時觸發。

#### `prepare(messages, options?)`

準備 run 以供持久執行，但不會啟動。此方法會在內部 registry 註冊 run，並回傳序列化的 Workflow 輸入。需要控制 Workflow 的觸發時機與方式時，請使用此方法。

```typescript
const { runId, messageId, workflowInput, threadId, resourceId } = await durableAgent.prepare(
  'Summarize the document',
  {
    memory: { threadId: 'thread-1', resourceId: 'user-1' },
  },
)
```

回傳：

```typescript
interface PrepareResult {
  runId: string
  messageId: string
  workflowInput: any
  registryEntry: object
  threadId?: string
  resourceId?: string
}
```

### 復原

#### `recoverActiveRuns(options?)`

探索此 Agent 卡在 `running` 狀態的 run，並從最後一個持久化快照重新驅動。最多復原 `options.limit` 個 run（預設：100），並回傳復原摘要。

```typescript
const result = await durableAgent.recoverActiveRuns()
// { recovered: [{ runId, status }], succeeded: 2, failed: 0 }
```

傳入 `runId` 可復原單一已知 run：

```typescript
await durableAgent.recoverActiveRuns({ runId: 'run-abc-123' })
```

回傳：

```typescript
interface DurableAgentRecoverActiveRunsResult {
  recovered: Array<{ runId: string; status: 'success' | 'failed'; error?: Error }>
  succeeded: number
  failed: number
}
```

**options.runId** (`string`): 依 ID 復原特定 run。設定此值時會忽略探索篩選條件。

**options.limit** (`number`): 要探索的作用中 run 數量上限。預設為 100。

**options.createdBefore** (`Date`): 只復原在此日期之前建立的 run。

#### `recover(runId, options?)`

依 ID 復原單一 run。回傳結構與 `stream()` 相同的可串流結果。需要即時觀察復原 stream 時，請使用此方法。

```typescript
const { output, cleanup } = await durableAgent.recover('run-abc-123', {
  onChunk: chunk => console.log(chunk),
  onError: ({ error }) => console.error(error),
})

await output.text
cleanup()
```

回傳： [`Promise<DurableAgentStreamResult>`](#durableagentstreamresult)

## stream 選項

`stream()` 接受 `DurableAgentStreamOptions` 物件。它支援下列 Agent 執行選項，以及生命週期 callback。

**runId** (`string`): 此 run 的唯一識別碼。稍後可搭配 resume() 或 observe() 使用。

**instructions** (`AgentExecutionOptions['instructions']`): 覆寫此 run 的 Agent 預設指示。接受靜態字串，或 Agent 支援的相同動態 instructions 值。

**context** (`ModelMessage[]`): 要提供給 Agent 的其他 context 訊息。

**memory** (`object`): 用於保存及擷取對話的記憶體設定。

**requestContext** (`RequestContext`): 帶有此 run 動態設定與狀態的 request context。

**maxSteps** (`number`): 此 stream 最多可執行的步驟數。

**toolsets** (`object`): 此 run 可用的其他 Tool set。

**clientTools** (`object`): 執行期間可用的 client 端 Tool。

**toolChoice** (`'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }`): Tool 選擇策略。

**activeTools** (`string[]`): 限制只執行 Agent Tool 中指定名稱的子集。

**modelSettings** (`object`): 模型專屬設定，例如 temperature。序列化快照跨越處理程序邊界前，會移除帶有憑證的 header（Authorization、X-Api-Key 等）。

**stopWhen** (`AgentExecutionOptions['stopWhen']`): 可提早結束 Agent 迴圈的 predicate 或組合。closure 會存放在處理程序內 run registry；跨處理程序恢復時會降級為只使用 maxSteps。

**system** (`string | string[]`): 附加在 Agent 指示之後、使用者訊息之前的其他 system message。

**requireToolApproval** (`boolean | ((args: { toolName: string; args: unknown; requestContext: RequestContext; workspace?: string }) => boolean | Promise<boolean>)`): 要求核准 Tool 呼叫。傳入 true 或 false 可控管全部或完全不控管；也可傳入函式，為每次呼叫設定原則。函式形式的原則存放在處理程序內 run registry；跨處理程序恢復時會退回使用 true shadow。

**autoResumeSuspendedTools** (`boolean`): 自動恢復已暫停的 Tool，而不是等待外部 resume() 呼叫。

**toolCallConcurrency** (`number`): 可同時執行的 Tool 呼叫數量上限。

**includeRawChunks** (`boolean`): 在 stream 輸出中包含原始 Provider chunk。

**maxProcessorRetries** (`number`): 每次生成時 Processor 的重試次數上限。

**structuredOutput** (`object`): 結構化輸出設定。

**untilIdle** (`boolean | { maxIdleMs?: number }`): 設定後，stream 會在背景任務接續執行期間保持開啟，直到 Agent 閒置為止。傳入 true 可使用預設 5 分鐘閒置逾時，或傳入 { maxIdleMs } 自訂。等同已棄用的 streamUntilIdle() 方法。resume() 也支援此選項。

**disableBackgroundTasks** (`boolean`): 停用此 run 的背景任務派送。可在背景執行的 Tool 會改為 inline 執行。

**tracingOptions** (`AgentExecutionOptions['tracingOptions']`): 轉送至 Agent 與模型 span 的 Tracing metadata、標籤、Trace ID、父 span ID 與 requestContextKeys。可完整序列化為 JSON。

**actor** (`AgentExecutionOptions['actor']`): 轉送至 FGA 檢查與 Tool 執行的單次呼叫 actor signal。

**transform** (`AgentExecutionOptions['transform']`): 每次叫用的 Tool payload 轉換原則。transformToolPayload closure 存放在處理程序內 run registry；只有 JSON-safe 的 targets shadow 會序列化。

**prepareStep** (`AgentExecutionOptions['prepareStep']`): 每個反覆運算開始時，以 PrepareStepProcessor 叫用的步驟準備 hook。僅限 closure，並儲存在處理程序內 run registry。跨處理程序恢復時會遺失此 hook。

**isTaskComplete** (`AgentExecutionOptions['isTaskComplete']`): 單次呼叫完成原則。Scorer 執行個體與 onComplete 存放在處理程序內 run registry；JSON-safe primitive（strategy、timeout、parallel、suppressFeedback、scorerNames）會序列化，以供跨處理程序可觀測性使用。

**delegation** (`AgentExecutionOptions['delegation']`): 子 Agent 委派 hook（onDelegationStart、onDelegationComplete、messageFilter）。準備時會將 callback 寫入子 Agent Tool wrapper。跨處理程序恢復時會遺失 callback。

**versions** (`object`): 子 Agent 委派的版本覆寫值。

**abortSignal** (`AbortSignal`): 外部中止 signal。會轉送至持久 run 的內部 AbortController，因此任一來源都能取消 run。跨處理程序恢復無法復原 signal；若需要在恢復後中止，請將新的 signal 傳給 resume()。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): 每個串流 chunk 都會呼叫。

**onStepFinish** (`(result: AgentStepFinishEventData) => void | Promise<void>`): Agent 迴圈中的步驟完成時呼叫。

**onFinish** (`(result: AgentFinishEventData) => void | Promise<void>`): run 完成時呼叫。

**onError** (`(error: Error) => void | Promise<void>`): run 發生錯誤時呼叫。

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): run 暫停時呼叫，例如等待 Tool 核准。

**onAbort** (`AgentExecutionOptions['onAbort']`): run 透過 abortSignal 或 result.abort() 中止時呼叫。

**onIterationComplete** (`AgentExecutionOptions['onIterationComplete']`): 每次 Agent 迴圈反覆運算後呼叫，並提供最新的 messageList、finishReason 與 isFinal 旗標。在持久 Agent 上僅供觀察：回傳 continue: false 或 feedback 不會影響迴圈。

`resume()` 與 `observe()` 接受相同的生命週期 callback（`onChunk`、`onStepFinish`、`onFinish`、`onError`、`onSuspended`）。`observe()` 也接受 `offset`，可控制重播起點。

## DurableAgentStreamResult

由 `stream()`、`resume()`、`observe()` 與 `recover()` 回傳的物件。

```typescript
interface DurableAgentStreamResult<OUTPUT = undefined> {
  output: MastraModelOutput<OUTPUT>
  readonly fullStream: ReadableStream<any>
  runId: string
  threadId?: string
  resourceId?: string
  cleanup: () => void
  abort: () => void
}
```

**output** (`MastraModelOutput`): 串流輸出。await output.text 可取得完整文字，或取用 output.fullStream。

**fullStream** (`ReadableStream`): 完整 event stream，會委派給 output.fullStream。

**runId** (`string`): 唯一的 run ID。將其傳給 resume() 或 observe() 可重新連線。

**threadId** (`string`): 使用記憶體時的 thread ID。

**resourceId** (`string`): 使用記憶體時的 resource ID。

**cleanup** (`() => void`): 取消 PubSub 訂閱，並清除 run 的 registry 項目。run 使用完畢後請呼叫此函式。

**abort** (`() => void`): 透過切換內部 AbortController 中止 run。會在持久 LLM 執行步驟內顯示為 AbortError，並觸發 onAbort callback。run 完成後也可安全呼叫；此時不會執行任何作業。

## 相關內容

- [`createInngestAgent()`](https://mastra.zisheng.pro/zh-TW/reference/agents/inngest-agent)
- [Agent 類別](https://mastra.zisheng.pro/zh-TW/reference/agents/agent)
- [PubSub](https://mastra.zisheng.pro/zh-TW/docs/server/pubsub)
- [`.getMemory()`](https://mastra.zisheng.pro/zh-TW/reference/agents/getMemory)