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

# `createInngestAgent()`

`createInngestAgent()` 為現有的 [`Agent`](https://mastra.zisheng.pro/zh-HK/reference/agents/agent) 套用由 [Inngest](https://www.inngest.com/docs) 驅動的持久執行包裝。它與 [`createDurableAgent()`](https://mastra.zisheng.pro/zh-HK/reference/agents/durable-agent) 一樣，透過 [PubSub](https://mastra.zisheng.pro/zh-HK/docs/server/pubsub) 串流傳送事件，並支援可恢復串流；但 Agent 迴圈會在 Inngest 的執行引擎上執行，而非在處理程序內執行。當一次執行必須不受處理程序重新啟動影響，或需要在分散式環境中執行時，請使用此函式。

如需處理程序內的持久執行，請使用 [`createDurableAgent()`](https://mastra.zisheng.pro/zh-HK/reference/agents/durable-agent)。如需在內置工作流程引擎上執行觸發後毋須等待結果的工作，請使用 [`createEventedAgent()`](https://mastra.zisheng.pro/zh-HK/reference/agents/durable-agent)。

## 使用範例

設定 Inngest 用戶端、包裝 Agent、向 Mastra 註冊，並公開 Inngest 服務端點：

```typescript
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'
import { createInngestAgent, serve as inngestServe } from '@mastra/inngest'
import { Inngest } from 'inngest'

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

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

const durableAgent = createInngestAgent({ agent, inngest })

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
  server: {
    apiRoutes: [
      {
        path: '/inngest/api',
        method: 'ALL',
        createHandler: async ({ mastra }) => inngestServe({ mastra, inngest }),
      },
    ],
  },
})
```

串流傳送回應並讀取結果：

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

const text = await output.text

cleanup()
```

## `createInngestAgent(options)`

使用由 Inngest 驅動的持久執行及可恢復串流來包裝 `Agent`。

```typescript
import { createInngestAgent } from '@mastra/inngest'

const durableAgent = createInngestAgent({ agent, inngest })
```

傳回： [`InngestAgent`](#inngestagent-interface)

### 參數

**agent** (`Agent`): 要使用 Inngest 持久執行包裝的 Agent。InngestAgent 未有實作的方法（例如 listTools() 和 getMemory()）會透過 Proxy 委派給此 Agent。

**inngest** (`Inngest`): Inngest 用戶端實例，用於傳送工作流程事件；在 SDK v4 中亦用於發佈即時串流事件。

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

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

**pubsub** (`PubSub`): 用於串流傳送事件的 PubSub 實例。預設的 InngestPubSub 使用 Inngest Realtime，可跨處理程序運作。 (Default: `InngestPubSub`)

**cache** (`MastraServerCache`): 用於儲存串流事件的快取，可啟用可恢復串流。提供此項後，PubSub 會自動以 CachingPubSub 包裝。如省略，Agent 會繼承 Mastra 實例的快取。

**mastra** (`Mastra`): 用於可觀測性的 Mastra 實例。向 Mastra 註冊 Agent 時會自動設定。

## `InngestAgent` 介面

`createInngestAgent()` 傳回的物件，提供以下持久執行方法。任何未有明確定義的屬性或方法（例如 `listTools()` 和 `getMemory()`）都會透過 Proxy 轉送至底層 Agent。

### 屬性

**id** (`string`): Agent ID。

**name** (`string`): Agent 名稱。

**agent** (`Agent`): 底層 Mastra Agent。

**inngest** (`Inngest`): Inngest 用戶端。

**cache** (`MastraServerCache | undefined`): 如已啟用可恢復串流，則為解析後的快取實例。

**pubsub** (`PubSub`): 用於串流傳送事件的 PubSub 實例。

## 方法

### 執行

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

使用 Inngest 的持久執行引擎串流傳送回應。建立 PubSub 訂閱後，工作流程會透過 Inngest 事件觸發。

```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<InngestAgentStreamResult>`](#inngestagentstreamresult)

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

恢復已暫停的 Inngest 執行，例如在 Tool 獲批後。此方法會從儲存空間載入工作流程快照、找出已暫停的步驟，並向 Inngest 傳送恢復事件。

```typescript
const { output, cleanup } = await durableAgent.resume(
  runId,
  {
    approved: true,
  },
  { threadId: 'thread-1', resourceId: 'user-1' },
)

await output.text
cleanup()
```

第三個引數除了接受生命週期回調函式外，亦接受 `threadId` 和 `resourceId`：

**threadId** (`string`): 要與已恢復執行關聯的對話串 ID。

**resourceId** (`string`): 要與已恢復執行關聯的資源 ID。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): 每個串流區塊傳送時呼叫。

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

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

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

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): 執行暫停時呼叫。

傳回： [`Promise<InngestAgentStreamResult>`](#inngestagentstreamresult)

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

在 Inngest 的持久執行引擎上執行回應，並以單一 `FullOutput` 解析。如執行暫停，`generate()` 會以 `finishReason: 'suspended'` 解析。`runId` 選項並非必要；如省略，`generate()` 會建立執行 ID，並在 `result.runId` 中傳回。請使用 [`resumeGenerate()`](#resumegeneraterunid-resumedata-options) 繼續執行。如呼叫者需要暫停回調函式，請使用帶有 `onSuspended` 的 [`stream()`](#streammessages-options)。

```typescript
const result = await durableAgent.generate('Delete the old records', {
  requireToolApproval: true,
})

result.runId // Generated automatically
result.finishReason // 'suspended' when approval is required
```

傳回： `Promise<FullOutput<TOutput>>`

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

恢復已暫停的 `generate()` 執行，並以單一 `FullOutput` 解析。

```typescript
if (!result.runId) {
  throw new Error('Run ID is missing')
}

const resumedResult = await durableAgent.resumeGenerate(result.runId, { approved: true })
```

傳回： `Promise<FullOutput<TOutput>>`

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

重新連接現有執行，先重播快取事件，再傳送即時事件。網絡中斷後可使用此方法。傳入 `offset` 可從已知位置開始重播。

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

await output.text
```

`observe()` 的結果不包括 `threadId` 或 `resourceId`。

傳回： `Promise<Omit<InngestAgentStreamResult, 'threadId' | 'resourceId'>>`

> **注意:** `observe()` 傳回的 `cleanup()` 會刪除該次執行的登錄項目及快取事件。請只在完成該次執行後才呼叫。如執行已暫停，而你打算稍後恢復，請勿呼叫 `cleanup()`。

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

準備一次持久執行，但不會觸發執行。傳回已序列化的工作流程輸入，可用於手動觸發 Inngest 工作流程事件。

```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
  threadId?: string
  resourceId?: string
}
```

### 內省

#### `isInngestAgent(obj)`

檢查物件是否為 `InngestAgent` 的型別守衛。

```typescript
import { isInngestAgent } from '@mastra/inngest'

if (isInngestAgent(agent)) {
  // agent is InngestAgent
}
```

傳回： `boolean`

## 串流選項

`stream()` 接受 `InngestAgentStreamOptions` 物件。除了生命週期回調函式外，它亦支援與 [`DurableAgent.stream()`](https://mastra.zisheng.pro/zh-HK/reference/agents/durable-agent) 相同的 Agent 執行選項。

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

**instructions** (`AgentExecutionOptions['instructions']`): 覆寫 Agent 對此次執行的預設指示。

**context** (`ModelMessage[]`): 提供給 Agent 的額外上下文訊息。

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

**requestContext** (`RequestContext`): 帶有此次執行之動態設定及狀態的請求上下文。

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

**toolsets** (`object`): 此次執行可用的額外 Tool 集合。

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

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

**modelSettings** (`object`): 模型特定設定，例如 temperature。

**requireToolApproval** (`boolean`): 要求批准所有 Tool 呼叫；執行會暫停，直至恢復為止。

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

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

**includeRawChunks** (`boolean`): 在串流輸出中包括原始 Provider 區塊。

**maxProcessorRetries** (`number`): 每次生成中處理器重試次數上限。

**untilIdle** (`boolean | { maxIdleMs?: number }`): 設定後，串流會在背景工作延續期間保持開啟，直至 Agent 閒置為止。傳入 true 可使用預設的 5 分鐘閒置逾時，或傳入 { maxIdleMs } 自訂。

**onChunk** (`(chunk: ChunkType) => void | Promise<void>`): 每個串流區塊傳送時呼叫。

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

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

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

**onSuspended** (`(data: AgentSuspendedEventData) => void | Promise<void>`): 執行暫停時呼叫，例如等候 Tool 批准。

`observe()` 接受生命週期回調函式（`onChunk`、`onStepFinish`、`onFinish`、`onError`、`onSuspended`），以及用於控制重播起點的 `offset`。

## `InngestAgentStreamResult`

`stream()` 和 `resume()` 傳回的物件。`observe()` 方法傳回相同結構，但不包括 `threadId` 和 `resourceId`。

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

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

**fullStream** (`ReadableStream`): 完整事件串流，委派至 output.fullStream。

**runId** (`string`): 唯一執行 ID。將其傳給 resume() 或 observe() 以重新連接。

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

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

**cleanup** (`() => void`): 取消訂閱 PubSub，並清除該次執行的登錄項目。完成該次執行後請呼叫此函式。

## 提供 Inngest 函式

`@mastra/inngest` 套件提供 `serve()` 和 `createServe()`，讓你在 HTTP 框架中註冊 Inngest 工作流程函式。

### `serve(options)`

使用 Hono（預設框架）提供 Mastra 工作流程。此函式會從 Mastra 收集所有由 Inngest 支援的工作流程，並將其註冊為 Inngest 函式。

```typescript
import { serve } from '@mastra/inngest'

app.use('/inngest/api', async c => {
  return serve({ mastra, inngest })(c)
})
```

### `createServe(adapter)`

這個工廠函式接受任何 Inngest 服務適配器（`inngest/express`、`inngest/fastify`、`inngest/next` 等），並傳回該框架所用的服務函式。

```typescript
import { createServe } from '@mastra/inngest'
import { serve } from 'inngest/express'

const serveExpress = createServe(serve)
app.use('/inngest/api', serveExpress({ mastra, inngest }))
```

```typescript
import { createServe } from '@mastra/inngest'
import { serve } from 'inngest/next'

const serveNext = createServe(serve)
export const { GET, POST, PUT } = serveNext({ mastra, inngest })
```

### 服務選項

**mastra** (`Mastra`): 包含已註冊 Agent 及 Workflow 的 Mastra 實例。

**inngest** (`Inngest`): Inngest 用戶端實例。

**functions** (`InngestFunction.Like[]`): 與 Mastra Workflow 一併提供的額外 Inngest 函式。

**registerOptions** (`RegisterOptions`): 傳給 Inngest 註冊處理常式的選項。

## 相關內容

- [DurableAgent 參考](https://mastra.zisheng.pro/zh-HK/reference/agents/durable-agent)
- [Agent 類別](https://mastra.zisheng.pro/zh-HK/reference/agents/agent)
- [Inngest 部署指南](https://mastra.zisheng.pro/zh-HK/guides/deployment/inngest)
- [PubSub](https://mastra.zisheng.pro/zh-HK/docs/server/pubsub)