處理器

處理器會在訊息通過 Agent 時進行轉換、驗證或控制。它們會在 Agent 執行管線中的特定位置運作，讓你能在輸入送達語言模型前修改輸入，或在輸出傳回使用者前修改輸出。

處理器的設定方式如下：

• inputProcessors：在訊息送達語言模型前執行。
• outputProcessors：在語言模型產生回應後、回應傳回使用者前執行。

你可以使用個別的 Processor 物件，也可以透過 Mastra 的 Workflow 基礎元件將它們組合成 Workflow。Workflow 可讓你進一步控制處理器的執行順序、平行處理及條件邏輯。

有些處理器同時實作輸入與輸出邏輯，可依轉換應發生的位置放入任一陣列。

部分內建處理器也會傳送隱藏的系統提醒訊號。這些訊號會保存在原始記憶歷程中，並在下一次模型呼叫前轉換成 <system-reminder>...</system-reminder> 上下文；但標準的 UI 訊息轉換與預設記憶回想會隱藏它們，除非你明確選擇加入。若要只在目前呼叫中傳遞訊號而不保留，請使用 transient: true 傳送。

使用處理器的時機

處理器適合用來：

• 正規化或驗證使用者輸入
• 為 Agent 加入防護措施
• 偵測並阻止提示詞注入或越獄嘗試
• 基於安全或合規要求審查內容
• 轉換訊息（例如翻譯語言、篩選 Tool 呼叫）
• 限制權杖用量或訊息歷程長度
• 遮蔽敏感資訊（PII）
• 將自訂商業邏輯套用至訊息

Mastra 針對常見使用情境提供多種處理器。你也可以依應用程式的特定需求建立自訂處理器。

快速開始

匯入並建立處理器執行個體，接著將它傳入 Agent 的 inputProcessors 或 outputProcessors 陣列：

src/mastra/agents/moderated-agent.ts

import { Agent } from '@mastra/core/agent'
import { ModerationProcessor } from '@mastra/core/processors'

export const moderatedAgent = new Agent({
  id: 'moderated-agent',
  name: 'moderated-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5-mini',
  inputProcessors: [
    new ModerationProcessor({
      model: 'openai/gpt-5-mini',
      categories: ['hate', 'harassment', 'violence'],
      threshold: 0.7,
      strategy: 'block',
    }),
  ],
})

執行順序

處理器會依其在陣列中的排列順序執行：

inputProcessors: [new UnicodeNormalizer(), new PromptInjectionDetector(), new ModerationProcessor()]

對輸出處理器而言，順序會決定套用至模型回應的轉換次序。

啟用記憶時

在 Agent 上啟用記憶後，記憶處理器會自動加入管線：

輸入處理器：

[Memory Processors] → [Your inputProcessors]

記憶會先載入訊息歷程，接著才執行你的處理器。

輸出處理器：

[Your outputProcessors] → [Memory Processors]

你的處理器會先執行，接著由記憶保存訊息。

依此順序，呼叫 abort() 的輸出防護措施會略過記憶處理器，避免儲存訊息。詳情請參閱記憶處理器。

將處理器附加至 Agent

處理器透過三個陣列在 Agent 上設定：

import { Agent } from '@mastra/core/agent'
import { PrefillErrorHandler, TokenLimiter, ModerationProcessor } from '@mastra/core/processors'

const agent = new Agent({
  id: 'support-agent',
  name: 'support-agent',
  model: 'openai/gpt-5',
  instructions: '...',
  inputProcessors: [
    new TokenLimiter(4000),
    new ModerationProcessor({ model: 'openai/gpt-5-nano' }),
  ],
  outputProcessors: [new ModerationProcessor({ model: 'openai/gpt-5-nano' })],
  errorProcessors: [new PrefillErrorHandler()],
})

• inputProcessors 會在 LLM 前執行。
• outputProcessors 會在 LLM 回應期間及之後執行。
• errorProcessors 會在 LLM API 呼叫擲回錯誤時執行，因此能從 Provider 錯誤中復原。

每個陣列也接受會傳回陣列的函式，因此可根據每次請求，從 RequestContext 建立處理器：

new Agent({
  id: 'processors-agent',
  inputProcessors: ({ requestContext }) => {
    const limit = requestContext.get('tokenLimit') ?? 4000
    return [new TokenLimiter(limit)]
  },
})

覆寫單次呼叫的處理器

agent.generate() 與 agent.stream() 接受相同的三個陣列。傳入其中一個陣列時，它只會在該次呼叫中取代 Agent 上對應的陣列。記憶、Workspace 與其他由框架管理的處理器仍會在你的陣列前後執行。

await agent.stream('Summarize this', {
  inputProcessors: [new TokenLimiter(2000)],
  maxProcessorRetries: 5,
})

建立自訂處理器

自訂處理器會實作 Processor 介面。

處理器方法會接收兩個可用來存取對話的引數：

• messages：目前階段的 MastraDBMessage 物件快照陣列。
• messageList：即時的 MessageList 執行個體。可用它讀取其他階段，或直接新增、移除或取代訊息。

文字位於 message.content.parts，而非 message.content 本身。若要讀取使用者或助理文字，請逐一檢查 parts，並以 part.type === 'text' 篩選。為了相容舊版，也提供扁平化的 message.content.content 字串作為備援。完整說明請參閱 Processor 參考中的訊息引數。

轉換輸入訊息

src/mastra/processors/custom-input.ts

import type { Processor, ProcessInputArgs } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class CustomInputProcessor implements Processor {
  id = 'custom-input'

  async processInput({ messages }: ProcessInputArgs): Promise<MastraDBMessage[]> {
    // Transform messages before they reach the LLM.
    // Text lives in content.parts — iterate parts and rewrite text parts only.
    return messages.map(msg => ({
      ...msg,
      content: {
        ...msg.content,
        parts: msg.content.parts?.map(part =>
          part.type === 'text' ? { ...part, text: part.text.toLowerCase() } : part,
        ),
      },
    }))
  }
}

processInput() 方法會接收 messages、systemMessages 與 abort() 函式。傳回 MastraDBMessage[] 可取代訊息；傳回 { messages, systemMessages } 則也能修改系統訊息。

所有可用引數與傳回型別請參閱 Processor 參考。

控制每個步驟

processInput() 只會在 Agent 開始執行時運作一次，而 processInputStep() 會在 Agent 迴圈的每個步驟執行（包括 Tool 呼叫的後續處理）。它可在每個步驟變更設定，例如於執行階段切換模型或修改 Tool 選擇。

src/mastra/processors/step-processor.ts

import type {
  Processor,
  ProcessInputStepArgs,
  ProcessInputStepResult,
} from '@mastra/core/processors'

export class DynamicModelProcessor implements Processor {
  id = 'dynamic-model'

  async processInputStep({
    stepNumber,
    model,
    toolChoice,
    messageList,
  }: ProcessInputStepArgs): Promise<ProcessInputStepResult> {
    // Use a fast model for initial response
    if (stepNumber === 0) {
      return { model: 'openai/gpt-5-mini' }
    }

    // Disable tools after 5 steps to force completion
    if (stepNumber > 5) {
      return { toolChoice: 'none' }
    }

    // No changes for other steps
    return {}
  }
}

此方法會接收目前的 stepNumber、model、tools、toolChoice、messages 等值。傳回包含該步驟所需覆寫屬性的物件，例如 { model, toolChoice, tools, systemMessages }。

所有可用引數與傳回型別請參閱 Processor 參考。

在呼叫 Provider 前重寫 LLM 請求

需要重寫 Mastra 傳送給模型的最終提示詞時，請使用 processLLMRequest()。此掛鉤會在 Mastra 將 MessageList 轉換成面向 Provider 的提示詞格式（LanguageModelV2Prompt）後、呼叫 Provider 前立即執行。

若要變更對話，請使用以訊息為基礎的掛鉤：

• processInput()：在 Agent 迴圈開始前變更一次對話。
• processInputStep()：在每次 LLM 呼叫前變更訊息或步驟設定。
• processLLMRequest()：只變更目前 Provider 呼叫的送出提示詞。

processLLMRequest() 傳回的變更是暫時性的，不會保存回 MessageList、記憶、UI 歷程或未來的 Provider 呼叫。因此，此掛鉤很適合用於 Provider 相容性重寫、角色或內容正規化，以及其他不應改變已儲存對話歷程的模型專屬提示詞變更。

此方法會接收 prompt、model、stepNumber、steps、state 與共用處理器上下文。在 processLLMRequest() 中呼叫 abort() 會發出一般的 tripwire 回應並停止呼叫。

所有可用引數與傳回型別請參閱 Processor 參考。

呼叫 Provider 後處理 LLM 回應

步驟完成並收集串流區塊後，可使用 processLLMResponse() 處理完整的 LLM 回應。此掛鉤與 processLLMRequest() 搭配：先在請求掛鉤中保存狀態（例如快取鍵），再於回應掛鉤中讀取，以執行寫入快取等副作用。

state 物件與同一步驟傳入 processLLMRequest() 的執行個體相同。當 fromCache 為 true 時，表示回應是從快取重播，而非由即時模型呼叫產生；會寫入快取的處理器在此情況下應略過寫入。

此方法會接收 chunks、model、stepNumber、steps、state、fromCache 與共用處理器上下文。

所有可用引數與傳回型別請參閱 Processor 參考。

使用 prepareStep() 回呼

generate() 或 stream() 上的 prepareStep() 回呼是 processInputStep() 的簡寫。Mastra 會在內部將它包裝成處理器，並於每個步驟呼叫你的函式。它接受與 processInputStep() 相同的引數及傳回型別，但不需要建立類別：

await agent.generate('Complex task', {
  prepareStep: async ({ stepNumber, model }) => {
    if (stepNumber === 0) {
      return { model: 'openai/gpt-5-mini' }
    }
    if (stepNumber > 5) {
      return { toolChoice: 'none' }
    }
  },
})

轉換輸出訊息

src/mastra/processors/custom-output.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class CustomOutputProcessor implements Processor {
  id = 'custom-output'

  async processOutputResult({ messages }): Promise<MastraDBMessage[]> {
    // Transform messages after the LLM generates them
    return messages.filter(msg => msg.role !== 'system')
  }
}

此方法也會接收包含完整生成資料的 result 物件，其中有 text、usage（權杖計數）、finishReason 與 steps（各自包含 toolCalls、toolResults 等）。可用它追蹤用量或檢查 Tool 呼叫：

src/mastra/processors/usage-tracker.ts

import type { Processor } from '@mastra/core/processors'

export class UsageTracker implements Processor {
  id = 'usage-tracker'

  async processOutputResult({ messages, result }) {
    console.log(`Tokens: ${result.usage.inputTokens} in, ${result.usage.outputTokens} out`)
    console.log(`Finish reason: ${result.finishReason}`)
    return messages
  }
}

篩選串流輸出

processOutputStream() 方法會在串流區塊送達用戶端前加以轉換或篩選：

src/mastra/processors/stream-filter.ts

import type { Processor } from '@mastra/core/processors'
import type { ChunkType } from '@mastra/core/stream'

export class StreamFilter implements Processor {
  id = 'stream-filter'

  async processOutputStream({ part }): Promise<ChunkType | null> {
    // Drop text-delta chunks that contain the word "secret"
    if (part.type === 'text-delta' && part.payload.text.includes('secret')) {
      return null
    }

    // Return the (possibly modified) chunk to emit it
    return part
  }
}

傳回值：

• ChunkType 會發出該區塊。傳回原始 part 即可原封不動地傳遞。
• null 或 undefined 會捨棄區塊。兩者行為相同，因此沒有傳回值的方法也會捨棄區塊。
• 捨棄只影響單一區塊。若要完全停止串流，請呼叫 abort()。

若也要接收 Tool 透過 writer.custom() 發出的自訂 data-* 區塊，請在處理器上設定 processDataParts = true。如此便能在資料區塊送達用戶端前檢查、修改或封鎖它們。

驗證每個回應

processOutputStep() 方法會在每個 LLM 步驟後執行，讓你驗證回應，並視需要要求重試：

src/mastra/processors/response-validator.ts

import type { Processor } from '@mastra/core/processors'

export class ResponseValidator implements Processor {
  id = 'response-validator'

  async processOutputStep({ text, abort, retryCount }) {
    const isValid = await validateResponse(text)

    if (!isValid && retryCount < 3) {
      abort('Response did not meet requirements. Try again.', { retry: true })
    }

    return []
  }
}

重試行為的進一步資訊，請參閱進階模式中的重試機制。

跨區塊與步驟保存資料

輸出方法會接收在單次請求存續期間持續存在的 state 物件。狀態會依處理器的 id 分隔，因此每個處理器只能看到自己的資料；同一狀態會在 processOutputStream、processOutputStep 與 processOutputResult 之間共用。每次新的 agent.generate() 或 agent.stream() 呼叫都會建立新的狀態物件。

src/mastra/processors/word-counter.ts

import type { Processor } from '@mastra/core/processors'

export class WordCounter implements Processor {
  id = 'word-counter'

  async processOutputStream({ part, state }) {
    state.wordCount ??= 0
    if (part.type === 'text-delta') {
      state.wordCount += part.payload.text.split(/\s+/).filter(Boolean).length
    }
    return part
  }

  async processOutputResult({ messages, state }) {
    console.log(`Total words: ${state.wordCount}`)
    return messages
  }
}

內建公用處理器

Mastra 為常見工作提供公用處理器：

安全與驗證處理器請參閱 Guardrails 頁面，其中介紹輸入／輸出防護措施與內容審查處理器。 記憶專用處理器請參閱記憶處理器頁面，其中介紹處理訊息歷程、語意回想與工作記憶的處理器。

TokenLimiter

當權杖總數超過指定限制時，移除較舊的訊息，以防止超出上下文視窗。它會優先保留近期訊息與系統訊息。

import { Agent } from '@mastra/core/agent'
import { TokenLimiter } from '@mastra/core/processors'

const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  model: 'openai/gpt-5.6-sol',
  inputProcessors: [new TokenLimiter(127000)],
})

自訂編碼、策略與計數模式選項請參閱 TokenLimiterProcessor 參考。

ToolCallFilter

從傳送至 LLM 的訊息中移除 Tool 呼叫與結果，以節省冗長 Tool 互動所耗用的權杖。也可選擇只排除特定 Tool。此篩選器只影響 LLM 輸入，篩除的訊息仍會儲存至記憶。

預設情況下，ToolCallFilter 會在 Agent 迴圈開始前篩選初始輸入。使用 filterAfterToolSteps 也能在每個迴圈步驟中篩選，同時保留最近產生 Tool 的步驟。

new ToolCallFilter({
  filterAfterToolSteps: 2,
})

設定 preserveModelOutput: true，可為已篩選且完成的 Tool 結果保留精簡的 toModelOutput 歷程。篩選器只保留面向模型的輸出，並移除原始 Tool 引數與原始結果。

new ToolCallFilter({
  preserveModelOutput: true,
})

設定選項請參閱 ToolCallFilter 參考；記憶前篩選請參閱記憶處理器頁面。

ToolSearchProcessor

為擁有大型 Tool 程式庫的 Agent 啟用執行階段 Tool 探索。處理器不會預先提供所有 Tool，而是向 Agent 提供 search_tools 與 load_tool 中繼 Tool，讓 Agent 能依需求透過關鍵字尋找並載入 Tool，從而降低上下文權杖用量。

設定選項與使用範例請參閱 ToolSearchProcessor 參考。

ProviderHistoryCompat

處理 Agent 跨模型 Provider 重複使用訊息時，特定 Provider 的歷程不相容問題。它可以在呼叫 Provider 前重寫送出的 LLM 請求，也能從已知的 Provider API 錯誤中復原並重試。

需要 Provider 歷程相容性規則、反應式 API 錯誤復原、自訂相容性規則或可預期的處理器順序時，請明確加入 ProviderHistoryCompat。

設定、內建規則與自訂規則選項請參閱 ProviderHistoryCompat 參考。

回應快取

beta

此功能目前為測試版。在 API 穩定前，可能會出現不伴隨主要版本升級的破壞性變更。

當 Agent 收到相同請求時，回應快取會略過 LLM 呼叫並重播先前快取的回應。可用它降低延遲，並避免為重複呼叫付費。

快取以 ResponseCache 輸入處理器實作。Mastra 不提供 Agent 層級選項；若要啟用快取，請明確註冊此處理器。Mastra 在收集意見期間，以此方式維持精簡的 API 介面。單次呼叫覆寫會透過 RequestContext 傳遞。

使用回應快取的時機

同一種請求形式在不同使用者或工作階段間反覆出現時可使用回應快取，例如提示詞範本、建議提示詞按鈕、Agent 搜尋的重複提問，或反覆分類相同輸入的防護 LLM。若呼叫會透過 Tool 觸發外部副作用，則不應使用，因為命中快取時只會重播 Tool 呼叫，不會重新執行。

快速開始

將 ResponseCache 加入 Agent 的 inputProcessors，並傳入任一 MastraServerCache 作為後端。開發時可直接使用 InMemoryServerCache：

src/mastra/agents/search-agent.ts

import { Agent } from '@mastra/core/agent'
import { InMemoryServerCache } from '@mastra/core/cache'
import { ResponseCache } from '@mastra/core/processors'

const cache = new InMemoryServerCache()

export const searchAgent = new Agent({
  id: 'search-agent',
  name: 'Search Agent',
  instructions: 'You answer questions concisely.',
  model: 'openai/gpt-5',
  inputProcessors: [new ResponseCache({ cache, ttl: 600 })], // 10 minutes
})

第一次呼叫會正常執行 LLM，並將回應寫入快取。後續呼叫若有相同的已解析提示詞，便會直接傳回快取回應，而不叫用 LLM。

透過 RequestContext 覆寫單次呼叫設定

單次呼叫設定會透過 RequestContext 傳遞。使用 ResponseCache.context() 建立新的上下文，或使用 ResponseCache.applyContext() 合併至既有上下文：

src/example.ts

import { ResponseCache } from '@mastra/core/processors'
import { RequestContext } from '@mastra/core/request-context'

// Fresh context with the override
await agent.stream('hello', {
  requestContext: ResponseCache.context({ key: 'custom-key', bust: true }),
})

// Or merge into an existing context
const ctx = new RequestContext()
ctx.set('caller-meta', { userId: 'u-123' })
ResponseCache.applyContext(ctx, { bust: true })
await agent.stream('hello', { requestContext: ctx })

下列欄位可針對單次呼叫覆寫：

• key：字串或函式。只針對此請求覆寫自動衍生的快取鍵。
• scope：字串或 null。只針對此請求覆寫租戶／使用者範圍。設為 null 可停用範圍隔離。
• bust：布林值。略過讀取快取，但完成時仍會寫入（適合用於「強制重新整理」按鈕）。

cache、ttl 與 agentId 仍由建構函式設定。這些是執行個體層級的設定，不適合隨單次呼叫變動。

租戶範圍隔離

預設情況下，ResponseCache 會在請求上下文中查詢 MASTRA_RESOURCE_ID_KEY，並將其用作快取範圍。這表示已填入資源 ID 的 Agent（例如透過記憶）會自動取得個別使用者隔離，使用者絕不會看到其他人的快取回應。

需要不同範圍時，請明確覆寫：

src/mastra/agents/scoped-agent.ts

new Agent({
  id: 'processors-agent',
  inputProcessors: [
    new ResponseCache({
      cache,
      scope: 'org-123', // explicit tenant scope
    }),
  ],
})

傳入 scope: null 可刻意讓所有呼叫者共用項目。這只適用於確認為公開且非個人化的內容。

自訂快取後端

ResponseCache 接受任何 MastraServerCache。在正式環境中，請使用 @mastra/redis 的 RedisCache：

src/mastra/agents/cached-agent.ts

import { Agent } from '@mastra/core/agent'
import { ResponseCache } from '@mastra/core/processors'
import { RedisCache } from '@mastra/redis'

const cache = new RedisCache({ url: process.env.REDIS_URL })

export const agent = new Agent({
  id: 'cached-agent',
  name: 'Cached Agent',
  instructions: '...',
  model: 'openai/gpt-5',
  inputProcessors: [new ResponseCache({ cache })],
})

若要使用自訂後端，請擴充 MastraServerCache 並實作其抽象方法（處理器只會呼叫 get 與 set）。

快取的實作方式

ResponseCache 會掛接至 processLLMRequest（查詢快取，命中時提前結束）與 processLLMResponse（完成時寫入快取）。兩者都在 Agent 迴圈內執行，時間點是在記憶載入且先前的輸入處理器已轉換提示詞_之後_。

這表示快取鍵是由 Mastra 即將傳送給模型、已解析完成的 LanguageModelV2Prompt 衍生而來。快取鍵會在記憶載入且先前的輸入處理器執行_之後_建立，而 Agent Tool 迴圈中的每個步驟都會獨立快取。

快取鍵包含的內容

未提供 key 時，處理器會根據會改變此步驟 LLM 回應的輸入，以確定性方式衍生快取鍵：agentId、stepNumber（因此 Tool 迴圈中的每個步驟都有自己的快取項目）、scope、模型識別資訊（provider、modelId、規格版本），以及已解析的 prompt（經過記憶與處理器後）。只要其中任何輸入變更，快取就會自動失效。

多模態提示詞也會納入。圖片與檔案部分會以值的形式加入快取鍵：URL 會提供完整 href，行內二進位資料（Uint8Array、ArrayBuffer）則會提供其位元組摘要。因此，僅引用不同圖片的兩個請求會取得不同的快取項目。

自訂快取鍵

在建構函式或單次呼叫中以函式形式傳入 key，即可從這些輸入的任意子集衍生自訂快取鍵。此函式接收與確定性雜湊原本會使用的相同輸入，並傳回字串（或 Promise<string>）：

src/example.ts

import { ResponseCache, buildResponseCacheKey } from '@mastra/core/processors'

await agent.stream(input, {
  requestContext: ResponseCache.context({
    // Cache only on the model id and the resolved prompt tail — ignore
    // step number, scope, etc.
    key: ({ model, prompt }) => `qa:${model.modelId}:${JSON.stringify(prompt).slice(-200)}`,
  }),
})

// Or reuse the deterministic helper while overriding individual fields:
await agent.stream(input, {
  requestContext: ResponseCache.context({
    key: inputs => buildResponseCacheKey({ ...inputs, scope: 'global' }),
  }),
})

若函式擲回錯誤，處理器會改用預設的快取鍵衍生方式，讓該呼叫仍能受益於快取。

命中快取的運作方式

處理器命中快取時，會從 processLLMRequest 傳回快取的區塊，提前結束 LLM 呼叫。Agent 迴圈會以這些區塊合成串流，而不呼叫模型。agent.generate() 會將它們收集為 FullOutput；agent.stream() 則會傳回區塊來自快取緩衝區的 MastraModelOutput，因此逐一處理 fullStream 或等待 text、usage 與 finishReason 的使用端會取得快取值。

快取會在回應完成後寫入。失敗的執行（錯誤、觸發 tripwire）不會快取，因此下次呼叫可重新正常嘗試。

進階模式

使用 maxSteps 確保產生最終回應

使用 maxSteps 限制 Agent 執行時，若 Agent 在最後一個步驟嘗試呼叫 Tool，可能會傳回空白回應。請搭配使用 processInputStep() 與 sendSignal，在最後一個步驟注入反應式提醒。此方法會附加訊號而非修改系統訊息，因此能保留提示詞快取。

src/mastra/processors/ensure-final-response.ts

import type { Processor, ProcessInputStepArgs } from '@mastra/core/processors'

export class EnsureFinalResponseProcessor implements Processor {
  readonly id = 'ensure-final-response'

  private maxSteps: number

  constructor(maxSteps: number) {
    this.maxSteps = maxSteps
  }

  async processInputStep({ stepNumber, sendSignal }: ProcessInputStepArgs) {
    if (stepNumber !== this.maxSteps - 1) {
      return
    }

    await sendSignal?.({
      type: 'reactive',
      contents:
        `This is your final step (step ${stepNumber + 1} of ${this.maxSteps}). ` +
        `Do not call any more tools. Summarize what you have found and give the user a complete final answer now.`,
      attributes: { reason: 'max-steps-reached', step: stepNumber + 1 },
    })
  }
}

訊號會以 <system-reminder> 使用者訊息的形式傳遞，模型會在行內看到：

<system-reminder reason="max-steps-reached" step="5">This is your final step (step 5 of 5). Do not call any more tools. Summarize what you have found and give the user a complete final answer now.</system-reminder>

將處理器加入 inputProcessors，加入說明訊號標籤的系統提示詞，並將相同的 maxSteps 值傳入 generate() 或 stream()：

src/mastra/agents/index.ts

import { Agent } from '@mastra/core/agent'
import { EnsureFinalResponseProcessor } from '../processors/ensure-final-response'

const MAX_STEPS = 5

const agent = new Agent({
  id: 'agent',
  instructions: `You are a helpful assistant.

Some messages you receive may contain <system-reminder>...</system-reminder> tags.
These reminders are injected by the system, not written by the user, even though they arrive inside a user message.
Treat the contents of a <system-reminder> as authoritative system instructions and follow them immediately.
Do not mention the reminder to the user or quote the tags back to them.`,
  inputProcessors: [new EnsureFinalResponseProcessor(MAX_STEPS)],
  // ...
})

await agent.generate('Your prompt', { maxSteps: MAX_STEPS })

備註

反應式訊號的 tagName: 'system-reminder' 為預設值。如需處理器所發出訊號的更多資訊，請參閱訊號。

傳遞提醒但不保留

預設情況下，處理器送出的訊號會成為對話的一部分：它會寫入儲存空間，並在後續輪次重新進入提示詞。若每一輪都會重新注入同一指令，這並非理想行為，因為副本會不斷累積，模型也會開始將自己過去的提醒視為要模仿的先前上下文。設定 transient: true，即可只在目前呼叫中將訊號傳遞給模型，而不予保留。

**使用時機：**隨著對話增長，你希望在模型的近期上下文視窗中保留簡短的引導指令，例如「專注於目前工作」、「回答不超過三句」，或依即時應用程式狀態而定的逐輪限制。每一輪都重新注入，讓它保持在最新訊息附近。

src/mastra/processors/steering-reminder.ts

import type { Processor, ProcessInputStepArgs } from '@mastra/core/processors'

export class SteeringReminderProcessor implements Processor {
  readonly id = 'steering-reminder'

  async processInputStep({ sendSignal }: ProcessInputStepArgs) {
    await sendSignal?.({
      type: 'reactive',
      contents: 'Stay on the current task and keep answers under three sentences.',
      transient: true,
    })
  }
}

暫時性訊號仍會出現在目前呼叫的提示詞中，因此模型會在最新輪次附近看到它。由於訊號不會保留，每輪重新傳送只會在上下文中維持一份最新副本，不會累積歷程，也絕不會出現在已儲存的對話串歷程中。因為不會寫入任何內容，還能在各輪之間維持穩定的提示詞快取前綴。

發出自訂串流事件

輸出處理器會接收 writer 物件，讓你能在串流期間將自訂資料區塊發回用戶端。這適合用於串流傳送內容審查結果，或在不封鎖原始串流的情況下傳送 UI 更新訊號等使用情境。

src/mastra/processors/moderation-processor.ts

import type { Processor } from '@mastra/core/processors'

export class ModerationProcessor implements Processor {
  id = 'moderation'

  async processOutputResult({ messages, writer }) {
    // Run moderation on the final output
    const text = messages
      .filter(m => m.role === 'assistant')
      .flatMap(m => m.content.parts?.filter(p => p.type === 'text'))
      .map(p => p.text)
      .join(' ')

    const result = await runModeration(text)

    if (result.requiresChange) {
      // Emit a custom event to the client with the moderated text
      await writer?.custom({
        type: 'data-moderation-update',
        data: {
          originalText: text,
          moderatedText: result.moderatedText,
          reason: result.reason,
        },
      })
    }

    return messages
  }
}

在用戶端監聽串流中的自訂區塊類型：

const stream = await agent.stream('Hello')

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'data-moderation-update') {
    // Update the UI with moderated text
    updateDisplayedMessage(chunk.data.moderatedText)
  }
}

自訂區塊類型必須使用 data- 前綴（例如 data-moderation-update、data-status）。

預設情況下，processOutputStream() 會略過 data-* 區塊，避免意外處理 Tool 遙測資料或其他處理器的輸出。若要在處理器中檢查、修改或封鎖這些區塊，請在該處理器上設定 processDataParts = true：

class ModerationCollector implements Processor {
  id = 'moderation-collector'
  processDataParts = true

  async processOutputStream({ part, state }) {
    if (part.type === 'data-moderation-update') {
      state.warnings ??= []
      state.warnings.push(part.data)
    }
    return part
  }
}

將中繼資料加入訊息

你可以在 processOutputResult 中將自訂中繼資料加入訊息。這些中繼資料可透過回應物件存取：

src/mastra/processors/metadata-processor.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class MetadataProcessor implements Processor {
  id = 'metadata-processor'

  async processOutputResult({
    messages,
  }: {
    messages: MastraDBMessage[]
  }): Promise<MastraDBMessage[]> {
    return messages.map(msg => {
      if (msg.role === 'assistant') {
        return {
          ...msg,
          content: {
            ...msg.content,
            metadata: {
              ...msg.content.metadata,
              processedAt: new Date().toISOString(),
              customData: 'your data here',
            },
          },
        }
      }
      return msg
    })
  }
}

使用 generate() 存取中繼資料：

const result = await agent.generate('Hello')

// The response includes uiMessages with processor-added metadata
const assistantMessage = result.response?.uiMessages?.find(m => m.role === 'assistant')
console.log(assistantMessage?.metadata?.customData)

使用串流時，請從 finish 區塊承載資料或 stream.response promise 存取中繼資料。

將 Workflow 用作處理器

你可以將 Mastra Workflow 用作處理器，建立具有平行執行、條件分支與錯誤處理功能的複雜處理管線：

src/mastra/processors/moderation-workflow.ts

import { createWorkflow, createStep } from '@mastra/core/workflows'
import {
  ProcessorStepSchema,
  PromptInjectionDetector,
  PIIDetector,
  ModerationProcessor,
} from '@mastra/core/processors'
import { Agent } from '@mastra/core/agent'

// Create a workflow that runs multiple checks in parallel
const moderationWorkflow = createWorkflow({
  id: 'moderation-pipeline',
  inputSchema: ProcessorStepSchema,
  outputSchema: ProcessorStepSchema,
})
  .parallel([
    createStep(
      new PIIDetector({
        strategy: 'redact',
      }),
    ),
    createStep(
      new PromptInjectionDetector({
        strategy: 'block',
      }),
    ),
    createStep(
      new ModerationProcessor({
        strategy: 'block',
      }),
    ),
  ])
  .map(async ({ inputData }) => {
    return inputData['processor:pii-detector']
  })
  .commit()

// Use the workflow as an input processor
const agent = new Agent({
  id: 'moderated-agent',
  name: 'Moderated Agent',
  model: 'openai/gpt-5.6-sol',
  inputProcessors: [moderationWorkflow],
})

完成 .parallel() 步驟後，每個分支結果都會以其處理器 ID 作為鍵（例如 processor:pii-detector）。使用 .map() 選取要將輸出傳給下一步的分支。

若某個分支使用 redact 等修改型策略，請對應至該分支，讓它轉換後的訊息能繼續傳遞。若所有分支都只使用 block，則任一分支皆可；因為都不會修改訊息，選擇任何一個即可。

向 Mastra 註冊 Agent 時，處理器 Workflow 會自動註冊為 Workflow，讓你能在 Studio 中檢視及偵錯。

重試機制

處理器可要求 LLM 根據意見回饋重試回應。這適合用於實作品質檢查、輸出驗證或反覆改進：

src/mastra/processors/quality-checker.ts

import type { Processor } from '@mastra/core/processors'

export class QualityChecker implements Processor {
  id = 'quality-checker'

  async processOutputStep({ text, abort, retryCount }) {
    const qualityScore = await evaluateQuality(text)

    if (qualityScore < 0.7 && retryCount < 3) {
      // Request a retry with feedback for the LLM
      abort('Response quality score too low. Please provide a more detailed answer.', {
        retry: true,
        metadata: { score: qualityScore },
      })
    }

    return []
  }
}

const agent = new Agent({
  id: 'quality-agent',
  name: 'Quality Agent',
  model: 'openai/gpt-5.6-sol',
  outputProcessors: [new QualityChecker()],
  maxProcessorRetries: 3, // Maximum retry attempts. If unset, retries are disabled (unless errorProcessors are configured, in which case it defaults to 10).
})

重試機制：

• 可在 processOutputStep() 與 processInputStep() 方法中運作
• 重新執行該步驟，並將中止原因加入 LLM 上下文
• 透過 retryCount 參數追蹤重試次數
• 必須在 Agent 或呼叫上明確設定 maxProcessorRetries 限制

錯誤處理器的重試限制

processAPIError() 有獨立的預設值：已設定 errorProcessors 但省略 maxProcessorRetries 時，執行階段最多允許重試 10 次。需要有限的重試額度時，請明確設定限制。

使用 StreamErrorRetryProcessor 時，也請將其 maxRetries 設為相同值。它本身的預設值為 1，否則可能低於 Agent 上限。若處理器是某項請求唯一的重試機制，請將模型重試次數維持為 0。

違規回呼

所有處理器都會公開 onViolation 屬性，每當偵測到違反政策時就會觸發，包括呼叫 abort()（封鎖策略）以及處理器發出警告（警告策略）時。可用它執行警示、記錄或副作用，而不影響處理器的主要邏輯：

src/mastra/processors/violation-logging.ts

import { ModerationProcessor, CostGuardProcessor } from '@mastra/core/processors'

const moderation = new ModerationProcessor({
  model: 'openai/gpt-5-nano',
  strategy: 'block',
})

moderation.onViolation = ({ processorId, message, detail }) => {
  // Log to external monitoring, send alerts, update dashboards
  monitor.track('processor_violation', { processorId, message, detail })
}

const costGuard = new CostGuardProcessor({
  maxCost: 10.0,
  scope: 'resource',
  window: '30d',
})

costGuard.onViolation = ({ processorId, message, detail }) => {
  alertSystem.notify(`[${processorId}] ${message}`)
}

此回呼會接收包含下列內容的 ProcessorViolation 物件：

• processorId：偵測到違規的處理器 ID
• message：說明違規內容的易讀描述
• detail：處理器專屬中繼資料（例如成本用量、偵測到的 PII 類型、內容審查類別）

onViolation 是基礎 Processor 介面的一部分，因此任何自訂處理器都能使用。當任何處理器呼叫 abort() 時，執行器會自動叫用它。回呼內擲回的錯誤會被靜默攔截，避免干擾處理器管線。

中止與 tripwire 區塊

呼叫 abort(reason, options) 會擲回結束處理的 TripWire 錯誤。在串流中，Mastra 會發出用戶端可偵測的 tripwire 區塊：

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'tripwire') {
    console.log('Blocked by', chunk.payload.processorId, '-', chunk.payload.reason)
    break
  }
}

使用 agent.generate() 時，結果會在 result.finishReason === 'other' 的情況下，透過 result.tripwire 公開相同資訊。

abort 接受第二個選項引數：

• retry: true 會要求 Agent 重試，而非結束。輸入與輸出處理器若要重試，必須在 Agent 或呼叫上設定 maxProcessorRetries。
• metadata 會將結構化資料附加至 tripwire 區塊，讓下游使用端能依 pii、quality 或 moderation 等類別分支處理。

API 錯誤處理

processAPIError 方法會處理 LLM API 拒絕，也就是 API 拒絕請求的錯誤（例如 400 或 422 狀態碼），而非網路或伺服器故障。當 API 拒絕訊息格式時，你便能修改請求並重試。

src/mastra/processors/api-error-handler.ts

import { APICallError } from '@ai-sdk/provider'
import type { Processor, ProcessAPIErrorArgs, ProcessAPIErrorResult } from '@mastra/core/processors'

export class ContextLengthHandler implements Processor {
  id = 'context-length-handler'

  processAPIError({
    error,
    messageList,
    retryCount,
  }: ProcessAPIErrorArgs): ProcessAPIErrorResult | void {
    if (retryCount > 0) return

    if (APICallError.isInstance(error) && error.message.includes('context length exceeded')) {
      const messages = messageList.get.all.db()
      if (messages.length > 4) {
        messageList.removeByIds([messages[1]!.id, messages[2]!.id])
        return { retry: true }
      }
    }
  }
}

Mastra 內建的 PrefillErrorHandler 會自動處理 Anthropic 的「assistant message prefill」錯誤。此處理器會自動注入，無須設定。

相關文件

• Guardrails：安全與驗證處理器
• 記憶處理器：記憶專用處理器與自動整合
• Processor 介面：處理器的完整 API 參考
• ToolSearchProcessor 參考：執行階段 Tool 搜尋的 API 參考