持久型 Agent

新增於： @mastra/core@1.45.0

警告

持久型 Agent 目前為 beta。API 可能在未來版本中變更。

持久型 Agent 會包裝一般的 Agent，讓 Agent 迴圈在 Workflow 內執行。事件透過 PubSub 傳遞，因此用戶端可以中斷連線後重新連線，而不會遺漏區塊。執行狀態會持久化，因此能承受處理程序重新啟動。

何時使用持久型 Agent

符合下列任一情況時，請使用持久型 Agent：

• 用戶端可能在串流期間中斷並重新連線（行動裝置、不穩定的網路、長時間呼叫）。
• Agent 迴圈的生命週期可能超過單一 HTTP 請求（背景研究、多步驟 Tool 使用）。
• 你需要觀察／重新連線 API，讓第二個用戶端接續第一個用戶端啟動的串流。
• 你希望使用具備步驟記憶化、重試與監控功能的 Inngest 執行環境。

若是用戶端會維持連線、生命週期短且範圍限於單次請求的呼叫，使用一般 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 參考。

運作方式

持久型 Agent 會在一般 Agent 上增加三個層次：

1. Workflow 執行：stream() 將訊息與選項序列化為 Workflow 輸入，再於持久型 Workflow 中觸發 Agent 迴圈。此 Workflow 執行的迴圈與 Agent.stream() 相同，但每個步驟都能記憶化及重播。

2. PubSub 串流：迴圈執行時，區塊會發佈至以執行 ID 為索引鍵的 PubSub 主題。呼叫端會訂閱此主題，並將區塊導入 ReadableStream。若呼叫端中斷後重新連線，快取會重播遺漏的區塊。

3. 快取層：選用的快取（預設使用記憶體，正式環境可使用 Redis 或其他後端）會儲存已發佈的事件，讓較晚訂閱的用戶端得以追上進度。

執行變體

Mastra 提供三個可產生持久型 Agent 的工廠函式，差別在於 Workflow 的執行方式：

工廠	套件	最適合的用途
createDurableAgent()	@mastra/core	本機開發與單一處理程序伺服器。你會取得可直接 await 的串流。
createEventedAgent()	@mastra/core	背景執行。Workflow 不會阻塞地啟動，而你可透過 PubSub 取用區塊。
createInngestAgent()	@mastra/inngest	正式環境部署。Inngest 提供步驟記憶化、重試及監控儀表板。

三者都會傳回物件，向 Mastra 註冊的方式與一般 Agent 相同。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 接收區塊，因此 stream() 會傳回可供取用的結果。觸發這次執行的 HTTP 處理常式不必等待 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 平台上執行 Workflow。每次 Tool 呼叫都會成為記憶化步驟，Inngest 可個別重試這些步驟；你也會取得監控執行狀況的儀表板：

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 與快取設定等 Inngest 專屬選項）請參閱 createInngestAgent() 參考。

可恢復的串流

持久型 Agent 透過 PubSub 和事件快取支援可恢復的串流。用戶端在串流期間中斷連線後，快取仍會繼續儲存事件。同一用戶端可使用 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() 預設使用記憶體內快取，表示可恢復串流僅在單一處理程序內運作。正式環境請提供持久型快取後端（例如 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 選項，或將 Agent 註冊至已設定 serverCache 的 Mastra 執行個體，以啟用可恢復串流。

搭配背景任務進行串流

持久型 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' },
})

背景任務的完整指南（包括設定、子 Agent 以及暫停／恢復）請參閱背景任務。

清理

每次 stream() 和 observe() 呼叫都會傳回 cleanup 函式。呼叫它會取消 PubSub 訂閱，並從內部登錄中移除該次執行。若忘記呼叫，串流結束後會觸發自動計時器；但自行呼叫 cleanup() 可立即釋放資源。

Tool 核准

持久型 Agent 支援 Tool 核准（人機協作）。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 })

當機復原

若持久型 Agent 執行期間伺服器處理程序當機，該次執行會在儲存空間中維持 running 狀態，且不會自動重試。下次伺服器啟動時，你可以重新驅動這些孤立的執行，讓它們從中斷處繼續。

自動復原

在 Mastra 設定中將 recovery.durableAgents 設為 'auto'。部署器會在啟動時，於重新啟動作用中的 Workflow 執行後立即呼叫 recoverAllDurableAgents()：

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 或鎖定。在多複本部署中，每個以 recovery.durableAgents: 'auto' 啟動的複本都會競相復原同一批執行。目前請將復原置於自己的 Leader Election 後方，或只從單一複本執行。

相關內容

• DurableAgent 參考
• createInngestAgent() 參考
• 背景任務
• Inngest 部署指南
• Agent 概觀
• Worker 概觀