持久 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()，便可從另一個客戶端重新連線。請參閱 DurableAgent 參考資料，了解完整設定及方法 API。

運作方式

持久 Agent 會在一般 Agent 之上加入三個層次：

1. Workflow 執行：stream() 會將訊息及選項序列化為 Workflow 輸入，然後在持久 Workflow 內觸發 Agent 循環。Workflow 運行的循環與 Agent.stream() 相同，但每個步驟均可記憶化及重播。

2. PubSub 串流：循環運行時，區塊會發佈到以運行 ID 為索引鍵的 PubSub 主題。呼叫者會訂閱此主題，並將區塊傳送到 ReadableStream。如果呼叫者中斷後重新連線，遺漏的區塊會從快取重播。

3. 快取層：可選用的快取（預設為記憶體內快取；正式環境則可使用 Redis 或其他後端）會儲存已發佈的事件，讓較遲訂閱的客戶端補回先前的內容。

執行方式

Mastra 提供三個建立持久 Agent 的工廠函式，分別採用不同的 Workflow 執行方式：

工廠函式	套件	最適合
createDurableAgent()	@mastra/core	本機開發及單一程序伺服器。你會取得可直接等待的串流。
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 })

請參閱 createInngestAgent() 參考資料，了解完整 API，包括 PubSub 及快取設定等 Inngest 專用選項。

可恢復的串流

持久 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'。部署程式會在啟動時呼叫 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 具備冪等性。

手動復原

如果你需要更精細的控制，例如只在領袖選舉通過後進行復原，或按排程執行，可直接呼叫以下方法：

// 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 目前尚未提供分散式租約或鎖定機制。在多副本部署中，每個以 recovery.durableAgents: 'auto' 啟動的副本都會競相復原相同運行。現階段，你可透過自己的領袖選舉限制復原操作，或只從單一副本執行復原。

相關內容

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