Observational Memory

新增於： @mastra/memory@1.1.0

Observational Memory（OM）是 Mastra 用於長上下文 Agent 記憶的記憶系統。Observer 會監察對話並建立觀察結果；Reflector 則透過合併相關項目及濃縮整體模式，重新整理這些觀察結果。兩者共同維護一份觀察記錄，並在記錄增長時取代原始訊息歷史。

用法

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5-mini',
  memory: new Memory({
    options: {
      observationalMemory: true,
    },
  }),
})

設定

observationalMemory 選項接受 true、設定物件或 false。設為 true 會啟用 OM，並以 google/gemini-2.5-flash 作為預設模型。傳入設定物件時，請在頂層設定 model，或在 observation.model 及／或 reflection.model 設定；如省略所有模型欄位，OM 會回退至 google/gemini-2.5-flash。

Observer 輸入支援多模態。OM 會在為 Observer 建立的文字記錄中保留 [Image #1: screenshot.png] 一類文字佔位符，並在可行情況下同時傳送底層圖片部分。這同時適用於單一 thread 觀察及批次多 thread 觀察。非圖片檔案只會顯示為佔位符。

OM 會使用快速的本機 token 估算來判斷閾值。文字使用 tokenx；類圖片輸入則使用能識別 Provider 的啟發式方法，並在 metadata 不完整時採用確定性的回退方式。

enabled?:

boolean

= true

啟用或停用 Observational Memory。如設定物件省略此項，預設為 true。只有 enabled: false 會明確停用此功能。

model?:

string | LanguageModel | DynamicModel | ModelByInputTokens | ModelWithRetries[]

= 'google/gemini-2.5-flash'

Observer 及 Reflector Agent 共用的模型，可一次過為兩者設定模型。不可與 observation.model 或 reflection.model 同時使用；如兩者皆有設定，系統會擲回錯誤。如這個欄位及 observation.model/reflection.model 均省略，OM 會回退至 google/gemini-2.5-flash。使用 "default" 可明確指定預設模型（google/gemini-2.5-flash）。

scope?:

'resource' | 'thread'

= 'thread'

觀察結果的記憶範圍。'thread' 按 thread 分別保存觀察結果。'resource'（實驗功能）會在某項資源的所有 thread 之間共享觀察結果，從而啟用跨對話記憶。

activateAfterIdle?:

number | string | false | "auto"

在閒置後強制啟用已緩衝觀察結果前的等待時間，即使尚未達到 observation.messageTokens 亦會啟用。接受如 300_000 的毫秒數值、如 "5m" 或 "1hr" 的時長字串、代表可感知 Provider 的 prompt 快取 TTL 的 "auto"，或以 false 停用繼承的觀察閒置啟用。反思不會繼承此設定；使用 reflection.activateAfterIdle 可選擇讓反思採用閒置啟用。

activateOnProviderChange?:

boolean

= false

當執行者的 Provider 或模型變更時，強制啟用已緩衝的觀察結果。反思不會繼承此設定；使用 reflection.activateOnProviderChange 可選擇讓反思在 Provider 變更時啟用。

shareTokenBudget?:

boolean

= false

讓訊息與觀察結果共享 token 預算。啟用後，總預算為 observation.messageTokens + reflection.observationTokens。觀察結果較少時，訊息可使用更多空間，反之亦然。這種彈性分配能盡量善用上下文。shareTokenBudget 尚未支援非同步緩衝。使用此選項時，必須設定 observation: { bufferTokens: false }（此為暫時限制）。

temporalMarkers?:

boolean

= false

如 thread 中上一則訊息至少早 10 分鐘，便在新的使用者訊息前插入時間間隔提醒標記。標記會持久保存於記憶中，並以 inline 提醒事件發出，讓用戶端可採用特殊方式顯示；Observer 亦會看到標記，以便按事件發生時間定位觀察結果。

retrieval?:

boolean | { vector?: boolean; scope?: 'thread' | 'resource'; instructions?: string }

= false

讓 Agent 查閱觀察結果背後的原始訊息記錄。觀察群組會保留指向原始訊息的持久指標，並註冊 recall Tool，讓 Agent 可以瀏覽這些訊息。true 預設啟用跨 thread 瀏覽。{ vector: true } 亦會使用 Memory 的向量儲存及 embedder 啟用語意搜尋。{ scope: 'thread' } 將 recall Tool 限制於目前 thread。預設範圍為 'resource'。{ instructions: '...' } 會在 Mastra 內置的檢索指示後附加應用程式專用的 recall 指引。

hooks?:

ObserveHooks

每個觀察／反思週期都會觸發的生命週期 hook，包括手動 observe()/reflect() API、由 turn 驅動的同步觀察，以及發出後不等待結果的非同步緩衝。callback 會收到 threadId/resourceId/trigger 呼叫上下文（'manual' | 'turn-sync' | 'async-buffer'）；結束 hook（onObservationEnd/onReflectionEnd）亦會收到 OM 模型呼叫的 token usage 及 providerMetadata，讓應用程式毋須以 middleware 包裝 Observer／Reflector 模型，也可計算 OM 模型開支（AI Gateway 等 Provider 會在此報告每次呼叫的成本）。非同步緩衝週期即使失敗亦不會擲回錯誤，而會透過結束 hook 的 error 欄位報告。這些 hook 擲回的錯誤會被捕捉及記錄，絕不會令週期失敗。

observation?:

ObservationalMemoryObservationConfig

觀察步驟的設定，用於控制 Observer Agent 的執行時機及行為。

ObservationalMemoryObservationConfig

model?:

string | LanguageModel | DynamicModel | ModelByInputTokens | ModelWithRetries[]

Observer Agent 使用的模型。如亦提供頂層 model，則不可設定此項。如這個欄位及頂層 model 均未設定，便回退至 reflection.model。

instruction?:

string

附加至 Observer system prompt 的自訂指示。可用於自訂 Observer 的關注重點，例如特定領域的偏好或優先次序。

threadTitle?:

boolean

設為 true 時，Observer 會建議簡短的 thread 標題，並在對話主題出現實質變化時更新標題。此功能須選擇啟用，預設為停用。

extract?:

Extractor[]

觀察後要擷取的自訂值。系統會要求 Observer 在輸出內 inline 提供無 schema extractor；有 schema 的 extractor 則會執行後續結構化輸出呼叫，並儲存於 thread OM metadata。

manageWorkingMemory?:

boolean

讓 Observer 透過 OM 擷取管理工作記憶。此設定會加入 WorkingMemoryExtractor、將 workingMemory.agentManaged 預設為 false，並將 workingMemory.useStateSignals 預設為 true。請參閱更新工作記憶。

observeAttachments?:

'auto' | boolean | string[]

控制哪些圖片／檔案附件會連同其佔位符文字行轉送至 Observer 模型。true（預設）會轉送所有附件；false 會捨棄所有附件，但仍顯示佔位符。'auto' 使用 Provider 功能登記資料來決定：Observer 模型支援多模態輸入時轉送附件，否則捨棄；如沒有該模型的功能資料，亦會轉送。陣列是不區分大小寫的 mimeType 允許清單，支援完全匹配（'application/pdf'）、萬用字元子類型（'image/*'），以及代表所有類型的 '*'。當 Observer 模型只支援文字（例如部分 DeepSeek endpoint），而主要 Agent 使用多模態模型時，此選項相當實用。Tool 結果附件亦按相同規則篩選。

messageTokens?:

number

觸發觀察的未觀察訊息 token 數量。未觀察訊息的 token 超過此閾值時，系統會呼叫 Observer Agent。文字會在本機使用 tokenx 估算；可行情況下，圖片部分會使用可感知模型的啟發式方法計算，圖片 metadata 不完整時則採用確定性的回退方式。如上載內容已正規化為檔案，類圖片的 file 部分亦以相同方式計算。

maxTokensPerBatch?:

number

在 resource 範圍觀察多個 thread 時，每個批次的 token 上限。thread 會按此大小分批並行處理。數值愈低，並行程度愈高，但 API 呼叫亦愈多。

modelSettings?:

ObservationalMemoryModelSettings

Observer Agent 的模型設定。maxOutputTokens: 100_000 預設值只會在選用預設模型時套用（未設定模型、使用 "default"，或使用 ModelByInputTokens 選擇器）。自訂模型沒有 maxOutputTokens 預設值。

ObservationalMemoryModelSettings

temperature?:

number

生成時的 temperature。數值愈低，輸出愈一致。

maxOutputTokens?:

number

輸出 token 上限。設定較高的數值可避免觀察結果被截斷。100000 預設值只會在選用預設模型時套用；自訂模型沒有預設值。

providerOptions?:

ProviderOptions

傳送至 Observer Agent 的 Provider 專用選項，例如 Google thinking 設定。

bufferTokens?:

number | false

背景觀察緩衝的執行頻率。0 至 1 之間的值是 messageTokens 的比例：0.25 表示每達閾值的 25% 便緩衝一次（預設 30k 時為 7.5k token）。1 或以上的值是絕對 token 數量：5000 表示每 5k token 緩衝一次。緩衝的觀察結果會一直儲存，直至達到 messageTokens 閾值，屆時無需阻塞式 LLM 呼叫即可立即啟用。計算結果必須小於 messageTokens。設為 false 可停用所有非同步緩衝（包括觀察及反思）。

bufferOnIdle?:

boolean

Agent turn 結束並進入閒置狀態時，在背景執行觀察緩衝。這與控制步驟執行期間非同步緩衝的 bufferTokens 不同。設為 true，即可緩衝短暫閒置的 turn，而毋須等待下一個 turn 或達到 messageTokens 閾值。

bufferActivation?:

number

啟用已緩衝觀察結果時，要清除多少訊息視窗。0 至 1 之間的值是要移除的 messageTokens 比例：0.8 會移除約 80% 的訊息記錄並保留約 20%（預設 30k 時為 6k token）。1000 或以上的值是要保留的 token 數量：4000 會在啟用後保留約 4k 訊息 token。請留意方向相反：比例愈高，移除的記錄愈多；token 數量愈高，保留的內容愈多。

activateAfterIdle?:

number | string | false | "auto"

閒置後強制啟用已緩衝觀察結果前的等待時間。接受毫秒值、時長字串、代表可感知 Provider 的 prompt 快取 TTL 的 "auto"，或 false。如未設定，觀察會使用頂層 activateAfterIdle 值。設為 false 可停用觀察的頂層閒置設定。目前只在使用獨立 ObservationalMemory 類別時套用；new Memory(...) 只會套用頂層 activateAfterIdle。

activateOnProviderChange?:

boolean

當執行者的 Provider 或模型變更時，強制啟用已緩衝的觀察結果。如未設定，觀察會使用頂層 activateOnProviderChange 值。目前只在使用獨立 ObservationalMemory 類別時套用；new Memory(...) 只會套用頂層 activateOnProviderChange。

blockAfter?:

number

當背景緩衝未能趕上時，強制執行同步（阻塞式）觀察的安全機制。1 至小於 100 的值是 messageTokens 的倍數：1.2 會在閾值的 120% 強制執行阻塞式觀察（預設 30k 時為 36k token）。100 或以上的值是絕對 token 數量，而且必須大於 messageTokens。在 messageTokens 與 blockAfter 之間，只會執行非同步緩衝及啟用；緩衝啟用仍會保留最少的剩餘上下文（1000 token 或保留下限，取較小者）。只在設定了 bufferTokens 時適用。啟用非同步緩衝時，預設為 1.2。

previousObserverTokens?:

number | false

Observer 先前觀察結果上下文的可選 token 預算。設為數值時，傳送至 Observer Agent 的觀察結果會從尾端截取以符合預算，同時保留最新觀察結果，並盡可能保留已標示的 🔴 項目。如有已緩衝的反思等候處理，截斷前會自動以反思摘要取代已反思的觀察行。設為 0 可完全省略先前觀察結果，設為 false 則明確停用截斷。

reflection?:

ObservationalMemoryReflectionConfig

反思步驟的設定，用於控制 Reflector Agent 的執行時機及行為。

ObservationalMemoryReflectionConfig

model?:

string | LanguageModel | DynamicModel | ModelByInputTokens | ModelWithRetries[]

Reflector Agent 使用的模型。如亦提供頂層 model，則不可設定此項。如這個欄位及頂層 model 均未設定，便回退至 observation.model。

instruction?:

string

附加至 Reflector system prompt 的自訂指示。可用於自訂 Reflector 整合觀察結果的方式，例如優先處理某些類型的資料。

extract?:

Extractor[]

反思後要擷取的自訂值。系統會要求 Reflector 在輸出內 inline 提供無 schema extractor；有 schema 的 extractor 則會執行後續結構化輸出呼叫，並儲存於 thread OM metadata。

observationTokens?:

number

觸發反思的觀察結果 token 數量。觀察結果的 token 超過此閾值時，系統會呼叫 Reflector Agent 加以濃縮。

modelSettings?:

ObservationalMemoryModelSettings

Reflector Agent 的模型設定。maxOutputTokens: 100_000 預設值只會在選用預設模型時套用（未設定模型、使用 "default"，或使用 ModelByInputTokens 選擇器）。自訂模型沒有 maxOutputTokens 預設值。

ObservationalMemoryModelSettings

temperature?:

number

生成時的 temperature。數值愈低，輸出愈一致。

maxOutputTokens?:

number

輸出 token 上限。設定較高的數值可避免觀察結果被截斷。100000 預設值只會在選用預設模型時套用；自訂模型沒有預設值。

providerOptions?:

ProviderOptions

傳送至 Reflector Agent 的 Provider 專用選項，例如 Google thinking 設定。

bufferActivation?:

number

以 observationTokens 的比例（0 至 1）指定何時開始背景反思：0.5 表示觀察結果達到閾值的 50% 時開始背景反思（預設 40k 時為 20k token）。達到完整閾值後，已緩衝的反思會取代其涵蓋的觀察結果，同時保留該範圍後新增的任何觀察結果。

activateAfterIdle?:

number | string | false | "auto"

閒置後強制啟用已緩衝反思前的等待時間。接受毫秒值、時長字串、代表可感知 Provider 的 prompt 快取 TTL 的 "auto"，或 false。反思不會繼承頂層 activateAfterIdle；請明確設定此項，選擇讓反思採用閒置啟用。目前只在使用獨立 ObservationalMemory 類別時套用；透過 new Memory(...) 使用時，此設定沒有效果。

activateOnProviderChange?:

boolean

當執行者的 Provider 或模型變更時，強制啟用已緩衝的反思。反思不會繼承頂層 activateOnProviderChange；請明確設定此項，選擇讓反思在 Provider 變更時啟用。目前只在使用獨立 ObservationalMemory 類別時套用；透過 new Memory(...) 使用時，此設定沒有效果。

blockAfter?:

number

當背景反思未能趕上時，強制執行同步（阻塞式）反思的安全機制。1 至小於 100 的值是 observationTokens 的倍數：1.2 會在閾值的 120% 強制執行阻塞式反思（預設 40k 時為 48k token）。100 或以上的值是絕對 token 數量，而且必須大於 observationTokens。在 observationTokens 與 blockAfter 之間，只會執行非同步緩衝及啟用。只在設定了 bufferActivation 時適用。啟用非同步反思時，預設為 1.2。

Token 估算 metadata 快取

OM 會持久保存 token payload 估算，讓重複計算可重用先前的 token 估算結果。

• 部分層級快取：part.providerMetadata.mastra。
• 字串內容回退快取：沒有任何部分時使用訊息層級 metadata。
• 如快取版本或 tokenizer 來源不相符，系統會忽略快取項目並重新計算。
• 每則訊息及每段對話的額外開銷一律在執行階段重新計算，不會快取。
• 系統會略過 data-* 及 reasoning 部分，亦不會為它們建立快取項目。

Extractor API

Extractor 定義 OM 應在觀察或反思期間擷取的值。current-task、suggested-response 及 thread-title 等內置 OM 值，會使用與自訂值相同的 extractor pipeline。

src/mastra/agents/agent.ts

import { Memory, Extractor } from '@mastra/memory'
import { z } from 'zod'

const memory = new Memory({
  options: {
    observationalMemory: {
      model: 'openai/gpt-5-mini',
      observation: {
        extract: [
          new Extractor({
            name: 'User profile',
            instructions: 'Extract stable user profile facts that should be remembered.',
            schema: z.object({
              name: z.string().optional(),
              timezone: z.string().optional(),
            }),
          }),
        ],
      },
    },
  },
})

name:

string

方便閱讀的 extractor 名稱。OM 會將此值轉換為 extractor slug。產生 slug 後，各名稱必須保持獨一無二。

slug:

string

由 name 衍生的唯讀屬性，並非 constructor 選項。這是為持久保存值及 XML tag 產生的穩定識別碼。slug 使用小寫英文字母、數字及連字號。自訂 extractor 不可使用內置 slug 及保留的 XML tag。

instructions:

string | (context) => string

指定要擷取甚麼以及何時更新值的指示。可使用函式從 runtime 上下文衍生指示。

schema?:

ZodType<T> | (context) => ZodType<T> | undefined

用於結構化擷取的可選 Zod schema。提供後，OM 會在主要 OM 操作後執行後續結構化輸出呼叫。省略時，extractor 是在 Observer 或 Reflector 回應中發出的 inline 字串 extractor。可使用函式從 runtime 上下文衍生 schema。

includePreviousExtraction?:

boolean

= true

控制日後執行 OM 時，是否向 extractor 顯示先前的擷取結果。如值只應來自目前的 OM 執行，請設為 false。

metadataKeyPath?:

string | false

= 'extracted.<slug>'

以句點分隔的 OM metadata 路徑，用於持久保存擷取值。設為 false 可完全略過 OM metadata 持久保存。

onExtracted?:

(context) => T | void | Promise<T | void>

在自訂 extractor 傳回值後、持久保存 metadata 前呼叫的可選 hook。傳回值會取代擷取值；擲回錯誤則會記錄擷取失敗。

擷取行為

• 擷取值會儲存在 thread OM metadata 的 om.extracted 下。
• 內置 extractor 值亦會鏡像至相容性 metadata 欄位 currentTask、suggestedResponse 及 threadTitle。
• 只有啟用 observation.threadTitle 時，thread-title 才會更新 thread 標題。
• observation.extract 在觀察期間執行；reflection.extract 在反思期間執行。
• 有 schema 的 extractor 會加入後續結構化輸出請求。
• 無 schema 的 extractor 是直接在 Observer 或 Reflector 輸出中發出的 inline 字串 extractor。
• 動態 extractor 函式會收到 runtime 上下文，包括可用的 source、threadId、resourceId、mainAgent、memory 及 requestContext。
• WorkingMemoryExtractor 使用一般 extractor pipeline，透過使用中的 Memory instance 更新工作記憶。工作記憶有 JSON schema 時，它會使用結構化擷取，並略過 OM metadata 持久保存，避免工作記憶 payload 在 OM 擷取 metadata 下重複出現。
• observationalMemory.observation.manageWorkingMemory 會加入 WorkingMemoryExtractor，並將 workingMemory.agentManaged 預設為 false。啟用工作記憶時，它會將 workingMemory.useStateSignals 預設為 true。
• 擷取失敗會在 OM marker data 中報告，而且不會捨棄其他成功擷取的值。

範例

更新工作記憶

如 OM 應更新工作記憶，請使用 observationalMemory.observation.manageWorkingMemory。

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'

const memory = new Memory({
  options: {
    workingMemory: {
      enabled: true,
    },
    observationalMemory: {
      enabled: true,
      observation: {
        manageWorkingMemory: true,
      },
    },
  },
})

如主要 Agent 仍應接收工作記憶 Tool 及指令注入，請設定 workingMemory.agentManaged: true。

使用自訂閾值的 resource scope（實驗性）

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5-mini',
  memory: new Memory({
    options: {
      observationalMemory: {
        model: 'google/gemini-2.5-flash',
        scope: 'resource',
        observation: {
          messageTokens: 20_000,
        },
        reflection: {
          observationTokens: 60_000,
        },
      },
    },
  }),
})

共用 token 預算

啟用 shareTokenBudget 後，總預算為 observation.messageTokens + reflection.observationTokens（本例為 100k）。如觀察結果只使用 30k tokens，訊息最多可擴展至使用 70k。如訊息較短，觀察結果在觸發反思前會有更多空間。

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5-mini',
  memory: new Memory({
    options: {
      observationalMemory: {
        shareTokenBudget: true,
        observation: {
          messageTokens: 20_000,
          bufferTokens: false, // required when using shareTokenBudget (temporary limitation)
        },
        reflection: {
          observationTokens: 80_000,
        },
      },
    },
  }),
})

自訂模型

在設定中傳入 model，即可使用 Mastra model router 中的任何模型。

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    options: {
      observationalMemory: {
        model: 'openai/gpt-5-mini',
      },
    },
  }),
})

為每個 Agent 使用不同模型

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    options: {
      observationalMemory: {
        observation: {
          model: 'google/gemini-2.5-flash',
        },
        reflection: {
          model: 'openai/gpt-5-mini',
        },
      },
    },
  }),
})

自訂指令

提供自訂指令，以自訂 Observer 及 Reflector 的關注重點：

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'health-assistant',
  name: 'health-assistant',
  instructions: 'You are a health and wellness assistant.',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    options: {
      observationalMemory: {
        model: 'google/gemini-2.5-flash',
        observation: {
          // Focus observations on health-related preferences and goals
          instruction:
            'Prioritize capturing user health goals, dietary restrictions, exercise preferences, and medical considerations. Avoid capturing general chit-chat.',
        },
        reflection: {
          // Guide reflection to consolidate health patterns
          instruction:
            'When consolidating, group related health information together. Preserve specific metrics, dates, and medical details.',
        },
      },
    },
  }),
})

非同步緩衝處理

非同步緩衝處理預設為啟用。隨着對話內容增加，它會在背景預先計算觀察結果：達到 messageTokens 閾值時，已緩衝的觀察結果會立即啟用，毋須等待會造成阻塞的 LLM 呼叫。

其生命週期依循緩衝 → 啟用 → 移除訊息 → 重複。背景 Observer 呼叫會按 bufferTokens 間距執行，每次產生一組觀察結果。達到閾值時，這些內容便會啟用：觀察結果移至日誌，而原始訊息則從上下文移除。如果緩衝處理未能跟上，blockAfter 閾值會強制改用同步後備處理。

預設設定：

• observation.bufferTokens: 0.2：每累積相當於 messageTokens 20% 的內容便進行緩衝（例如閾值為 30k 時，每約 ~6k tokens 一次）
• observation.bufferActivation: 0.8：啟用時移除足夠的訊息，使餘下內容只佔閾值的 20%
• 已緩衝的觀察結果包含延續提示（suggestedResponse、currentTask），這些提示會在啟用後保留，以維持對話連貫性
• reflection.bufferActivation: 0.5：觀察結果達到閾值的 50% 時，在背景開始反思

如要自訂：

src/mastra/agents/agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5-mini',
  memory: new Memory({
    options: {
      observationalMemory: {
        model: 'google/gemini-2.5-flash',
        observation: {
          messageTokens: 30_000,
          // Buffer every 5k tokens (runs in background)
          bufferTokens: 5_000,
          // Activate to retain 30% of threshold
          bufferActivation: 0.7,
          // Force synchronous observation at 1.5x threshold
          blockAfter: 1.5,
        },
        reflection: {
          observationTokens: 60_000,
          // Start background reflection at 50% of threshold
          bufferActivation: 0.5,
          // Force synchronous reflection at 1.2x threshold
          blockAfter: 1.2,
        },
      },
    },
  }),
})

如要完全停用非同步緩衝處理：

observationalMemory: {
  model: "google/gemini-2.5-flash",
  observation: {
    bufferTokens: false,
  },
}

設定 bufferTokens: false 會同時停用觀察與反思的非同步緩衝處理。達到各自的閾值時，觀察與反思會同步執行。

備註

scope: 'resource' 不支援非同步緩衝處理，因此在 resource scope 下會自動停用。

串流 data parts

Agent 執行期間，Observational Memory 會發出具類型的 data parts，讓 client 用於即時 UI 回饋。這些資料會連同 Agent 的回應以串流方式傳送。

讀取 extractor 結果

兩種完成事件的 data payload 均包含 extractor 輸出。extractor 欄位如下：

interface DataOmObservationEndPart {
  type: 'data-om-observation-end'
  data: {
    /** Whether the completed work was an observation or reflection */
    operationType: 'observation' | 'reflection'
    /** Values extracted during this OM operation, keyed by extractor slug */
    extractedValues?: Record<string, unknown>
    /** Extractor failures from this OM operation. Successful extractor values are still included */
    extractionFailures?: Array<{ slug: string; error: string }>
    // ...other fields documented in the tables below
  }
}

兩個 extractor 欄位均為選填。完成事件可包含值、失敗資料、兩者兼有，亦可兩者皆無。data-om-observation-end 報告同步完成；data-om-buffering-end 則報告已完成的背景工作，其緩衝內容仍有待啟用，但 extractor metadata 已經持久保存。DataOmBufferingEndPart 包含相同的 extractor 欄位，而兩種類型均由 @mastra/memory/processors 匯出。consumer 範例請參閱從串流讀取擷取值。

data-om-status

每個 Agent 迴圈步驟會在模型生成前發出一次。它提供目前記憶狀態的快照，包括兩個上下文視窗的 token 使用量，以及任何非同步緩衝內容的狀態。

interface DataOmStatusPart {
  type: 'data-om-status'
  data: {
    windows: {
      active: {
        /** Unobserved message tokens and the threshold that triggers observation */
        messages: { tokens: number; threshold: number }
        /** Observation tokens and the threshold that triggers reflection */
        observations: { tokens: number; threshold: number }
      }
      buffered: {
        observations: {
          /** Number of buffered chunks staged for activation */
          chunks: number
          /** Total message tokens across all buffered chunks */
          messageTokens: number
          /** Projected message tokens that would be removed if activation happened now (based on bufferActivation ratio and chunk boundaries) */
          projectedMessageRemoval: number
          /** Observation tokens that will be added on activation */
          observationTokens: number
          /** idle: no buffering in progress. running: background observer is working. complete: chunks are ready for activation. */
          status: 'idle' | 'running' | 'complete'
        }
        reflection: {
          /** Observation tokens that were fed into the reflector (pre-compression size) */
          inputObservationTokens: number
          /** Observation tokens the reflection will produce on activation (post-compression size) */
          observationTokens: number
          /** idle: no reflection buffered. running: background reflector is working. complete: reflection is ready for activation. */
          status: 'idle' | 'running' | 'complete'
        }
      }
    }
    recordId: string
    threadId: string
    stepNumber: number
    /** Increments each time the Reflector creates a new generation */
    generationCount: number
  }
}

buffered.reflection.inputObservationTokens 是傳送至 Reflector 的觀察結果大小。buffered.reflection.observationTokens 是壓縮後的結果，即反思啟用時用來取代這些觀察結果的內容大小。client 可利用這兩個值顯示壓縮比例。

client 可從原始值計算百分比及啟用後的估算值：

// Message window usage %
const msgPercent = status.windows.active.messages.tokens / status.windows.active.messages.threshold

// Observation window usage %
const obsPercent =
  status.windows.active.observations.tokens / status.windows.active.observations.threshold

// Projected message tokens after buffered observations activate
// Uses projectedMessageRemoval which accounts for bufferActivation ratio and chunk boundaries
const postActivation =
  status.windows.active.messages.tokens -
  status.windows.buffered.observations.projectedMessageRemoval

// Reflection compression ratio (when buffered reflection exists)
const { inputObservationTokens, observationTokens } = status.windows.buffered.reflection
if (inputObservationTokens > 0) {
  const compressionRatio = observationTokens / inputObservationTokens
}

data-om-observation-start

Observer 或 Reflector Agent 開始處理時發出。

cycleId:

string

此週期的唯一 ID，供 start/end/failed 標記共用。

operationType:

'observation' | 'reflection'

此操作是觀察還是反思。

startedAt:

string

開始處理時的 ISO 時間戳記。

tokensToObserve:

number

此批次正在處理的訊息 tokens（輸入）。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

threadIds:

string[]

此批次的所有 thread ID（用於 resource scope）。

config:

ObservationMarkerConfig

進行觀察時 messageTokens、observationTokens 及 scope 的設定快照。

data-om-observation-end

觀察或反思成功完成時發出。

cycleId:

string

與對應的 start 標記相符。

operationType:

'observation' | 'reflection'

已完成的操作類型。

completedAt:

string

處理完成時的 ISO 時間戳記。

durationMs:

number

持續時間（毫秒）。

tokensObserved:

number

已處理的訊息 tokens（輸入）。

observationTokens:

number

經 Observer 壓縮後所得的觀察結果 tokens（輸出）。

observations?:

string

產生的觀察結果文字。

currentTask?:

string

Observer 擷取的目前任務。

suggestedResponse?:

string

Observer 擷取的建議回應。

extractedValues?:

Record<string, unknown>

此 OM 操作期間擷取的值，以 extractor slug 為 key。

extractionFailures?:

Array<{ slug: string; error: string }>

此 OM 操作中的 extractor 失敗資料。成功擷取的 extractor 值仍會包含在內。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

data-om-observation-failed

觀察或反思失敗時發出。系統會改用同步處理作為後備方案。

cycleId:

string

與對應的 start 標記相符。

operationType:

'observation' | 'reflection'

失敗的操作類型。

failedAt:

string

發生失敗時的 ISO 時間戳記。

durationMs:

number

發生失敗前的持續時間（毫秒）。

tokensAttempted:

number

曾嘗試處理的訊息 tokens（輸入）。

error:

string

錯誤訊息。

observations?:

string

任何可供顯示的部分內容。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

data-om-buffering-start

非同步緩衝處理在背景開始時發出。緩衝處理會在達到主要閾值前預先計算觀察結果或反思。

cycleId:

string

此緩衝週期的唯一 ID。

operationType:

'observation' | 'reflection'

正在緩衝的操作類型。

startedAt:

string

開始緩衝時的 ISO 時間戳記。

tokensToBuffer:

number

此週期正在緩衝的訊息 tokens（輸入）。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

threadIds:

string[]

正在緩衝的所有 thread ID（用於 resource scope）。

config:

ObservationMarkerConfig

進行緩衝時的設定快照。

data-om-buffering-end

非同步緩衝處理完成時發出。內容已儲存，但尚未在主要上下文中啟用。

cycleId:

string

與對應的 buffering-start 標記相符。

operationType:

'observation' | 'reflection'

已緩衝的操作類型。

completedAt:

string

緩衝完成時的 ISO 時間戳記。

durationMs:

number

持續時間（毫秒）。

tokensBuffered:

number

已緩衝的訊息 tokens（輸入）。

bufferedTokens:

number

經 Observer 壓縮後的觀察結果 tokens（輸出）。

observations?:

string

已緩衝的內容。

extractedValues?:

Record<string, unknown>

此已緩衝 OM 操作期間擷取的值，以 extractor slug 為 key。

extractionFailures?:

Array<{ slug: string; error: string }>

此已緩衝 OM 操作中的 extractor 失敗資料。成功擷取的 extractor 值仍會包含在內。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

data-om-buffering-failed

非同步緩衝處理失敗時發出。達到閾值時，系統會改用同步處理作為後備方案。

cycleId:

string

與對應的 buffering-start 標記相符。

operationType:

'observation' | 'reflection'

失敗的操作類型。

failedAt:

string

發生失敗時的 ISO 時間戳記。

durationMs:

number

發生失敗前的持續時間（毫秒）。

tokensAttempted:

number

曾嘗試緩衝的訊息 tokens（輸入）。

error:

string

錯誤訊息。

observations?:

string

任何部分內容。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

data-om-activation

已緩衝的觀察結果或反思啟用（移至作用中的上下文視窗）時發出。這項操作會即時完成，不涉及 LLM 呼叫。

cycleId:

string

此啟用事件的唯一 ID。

operationType:

'observation' | 'reflection'

已啟用的內容類型。

activatedAt:

string

啟用時的 ISO 時間戳記。

chunksActivated:

number

已啟用的緩衝內容組數。

tokensActivated:

number

來自已啟用內容組的訊息 tokens（輸入）。啟用觀察結果時，這些 tokens 會從訊息視窗移除；啟用反思時，則表示被壓縮的觀察結果 tokens。

observationTokens:

number

啟用後所得的觀察結果 tokens。

messagesActivated:

number

透過啟用而被觀察的訊息數目。

generationCount:

number

目前的反思世代計數。

observations?:

string

已啟用的觀察結果文字。

triggeredBy?:

'threshold' | 'ttl' | 'provider_change'

啟用是因超過閾值、activateAfterIdle 到期，還是模型/provider 變更而觸發。

lastActivityAt?:

number

用於 TTL 檢查的最後一個 assistant 訊息部分之 Unix 毫秒時間戳記。

ttlExpiredMs?:

number

觸發啟用時，超出 activateAfterIdle 的時長。

previousModel?:

string

觸發啟用的上一個 assistant 模型識別符（例如 openai/gpt-4o）。

currentModel?:

string

觸發啟用的目前 actor 模型識別符。

recordId:

string

OM 記錄 ID。

threadId:

string

此 thread 的 ID。

config:

ObservationMarkerConfig

啟用時的設定快照。

data-om-thread-update

Observer 更新 thread 標題時發出。只會在啟用 observation.threadTitle 時發出。

cycleId:

string

此觀察週期的唯一 ID，與觀察標記共用。

threadId:

string

已更新的 thread ID。

oldTitle?:

string

先前的 thread 標題。如 thread 沒有標題，則為 undefined。

newTitle:

string

新的 thread 標題。

timestamp:

string

此更新發生的時間。

獨立使用

大部分使用者都應使用上文的 Memory 類別。直接使用 ObservationalMemory 主要適用於效能基準測試、實驗，或需要控制它與其他 processor（例如 guardrails）之間的排序時。

ObservationalMemory 類別是核心引擎；如要將它附加至 Agent，請用 ObservationalMemoryProcessor 將它包裝起來，而該 processor 需要一個 Memory 實例來載入及持久保存訊息。請注意，storage adapter 將 stores.memory 的型別定義為可選，因此需要使用非空斷言（或在執行階段檢查）：

src/mastra/agents/agent.ts

import { ObservationalMemory, ObservationalMemoryProcessor } from '@mastra/memory/processors'
import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'
import { LibSQLStore } from '@mastra/libsql'

const storage = new LibSQLStore({
  id: 'my-storage',
  url: 'file:./memory.db',
})

const memory = new Memory({ storage })

const om = new ObservationalMemory({
  storage: storage.stores.memory!,
  memory,
  model: 'google/gemini-2.5-flash',
  scope: 'resource',
  observation: {
    messageTokens: 20_000,
  },
  reflection: {
    observationTokens: 60_000,
  },
})

const omProcessor = new ObservationalMemoryProcessor(om, memory)

export const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5-mini',
  inputProcessors: [omProcessor],
  outputProcessors: [omProcessor],
})

獨立設定

獨立的 ObservationalMemory 類別接受上文 observationalMemory 設定物件的所有相同選項，以及以下選項：

storage:

MemoryStorage

用於持久保存觀察結果的 storage adapter。必須是 MemoryStorage 實例（來自 MastraStorage.stores.memory）。

onDebugEvent?:

(event: ObservationDebugEvent) => void

觀察事件的除錯回呼。每當發生觀察相關事件時便會呼叫，有助除錯及了解觀察流程。

obscureThreadIds?:

boolean

= false

啟用後，thread ID 會先經過雜湊處理，再加入觀察情境。這可防止 LLM 識別 thread 識別碼中的模式。透過 Memory 類別使用 resource scope 時會自動啟用。

Recall tool

設定 retrieval（任何 truthy 值）後，系統會註冊一個 recall tool，讓 Agent 可以逐頁瀏覽觀察群組範圍背後的原始訊息。預設情況下（scope 為 'resource'），此 tool 支援列出 thread（mode: "threads"）、瀏覽其他 thread（threadId），以及跨 thread 搜尋。使用 retrieval: { vector: true } 時，可進行語意搜尋（mode: "search"）。設定 scope: 'thread'，可將此 tool 限制為只存取目前的 thread。此 tool 會自動加入 Agent 的 tool 清單。

Mastra 亦會將能感知 scope 的使用指示注入 Agent 的情境。對於使用 vector: true 的 resource scope，這些指示涵蓋如何在 search、threads 與 messages 之間選擇路徑，包括當搜尋結果不合適時，改為探索 thread。若沒有 vector: true，指示只會涵蓋瀏覽 threads 與 messages，因此不會引導 Agent 使用尚未設定的搜尋模式。即使尚未存在任何觀察群組，resource scope 的指示亦會注入，讓 Agent 從第一則訊息起便可瀏覽其他 thread。使用 retrieval: { instructions: '...' }，可在內置指示之後加入應用程式專用指引。

參數

mode?:

'messages' | 'threads' | 'search'

= 'messages'

要擷取的內容。"messages"（預設）逐頁瀏覽訊息記錄。"threads" 列出目前使用者的所有 thread。"search" 按語意相似度在所有 thread 中尋找訊息（需要 vector store 及 embedder）。

query?:

string

mode: "search" 的搜尋查詢。在目前使用者的所有 thread 中尋找與此文字語意相似的訊息。

cursor?:

string

用作 recall 查詢定位點的訊息 ID。從觀察群組範圍擷取起始或結束 ID（例如從 _range: \startId:endId\_ 使用 startId 或 endId）。如果直接傳入範圍字串，此 tool 會傳回提示，說明如何擷取正確的 ID。當同時省略 cursor 與 threadId，並使用 mode: "messages" 時，此 tool 會從 anchor 所設定的位置開始瀏覽目前的 thread。

threadId?:

string

按 ID 瀏覽另一個 thread，或傳入 "current" 以使用使用中的 thread。請先使用 mode: "threads" 尋找 thread ID。如果提供此參數但沒有提供 cursor，便會從 thread 開頭開始讀取。

anchor?:

'start' | 'end'

= 'start'

對於 mode: "messages"，如沒有 cursor，便從 thread 開頭（最舊優先）或結尾（最新優先）開始分頁。

page?:

number

= 1

分頁偏移量。對於訊息：正值從 cursor 向前分頁，負值向後分頁。對於 thread：頁碼（從 0 開始）。訊息模式會將 0 視為 1。

limit?:

number

= 20

每頁傳回的項目數目上限。

detail?:

'low' | 'high'

= 'low'

控制每個訊息部分顯示多少內容。'low' 顯示截短的文字及附有位置索引（[p0]、[p1]）的 tool 名稱。'high' 顯示包括 tool 引數及結果在內的完整內容，每次呼叫最多顯示一個部分，並附有繼續提示。

partType?:

'text' | 'tool-call' | 'tool-result' | 'reasoning' | 'image' | 'file'

篩選結果，只包括此類型的訊息部分。只適用於 mode: "messages"。

toolName?:

string

篩選結果，只包括符合此 tool 名稱的 tool-call 及 tool-result 部分。只適用於 mode: "messages"。

partIndex?:

number

按位置索引擷取單一訊息部分的完整詳情。當低詳情 recall 在 [p1] 顯示值得留意的部分時使用此參數——以 partIndex: 1 再次呼叫，即可查看完整內容，而毋須載入每個部分。

before?:

string

只適用於 mode: "threads"。篩選在此日期之前建立的 thread。接受 ISO 8601 格式（例如 "2026-03-15"、"2026-03-10T00:00:00Z"）。

after?:

string

只適用於 mode: "threads"。篩選在此日期之後建立的 thread。接受 ISO 8601 格式（例如 "2026-03-01"、"2026-03-10T00:00:00Z"）。

傳回值（messages 模式）

messages:

string

已格式化的訊息內容。格式取決於 detail 層級。

count:

number

此頁的訊息數目。

cursor:

string

此查詢所用的 cursor 訊息 ID。

page:

number

傳回的頁碼。

limit:

number

此查詢所用的 limit。

detail:

'low' | 'high'

此查詢所用的 detail 層級。

hasNextPage:

boolean

此頁之後是否還有更多訊息。

hasPrevPage:

boolean

此頁之前是否還有更多訊息。

truncated?:

boolean

當輸出受 token 預算上限限制時，此欄位會存在並為 true。Agent 可透過分頁或使用 partIndex 存取餘下內容。

tokenOffset?:

number

當 truncated 為 true 時，被刪減的約略 token 數目。

傳回值（threads 模式）

threads:

string

已格式化的 thread 清單。每個 thread 都會顯示其標題、ID 及日期。目前的 thread 會以 ← current 標示。

count:

number

傳回的 thread 數目。

page:

number

傳回的頁碼。

hasMore:

boolean

下一頁是否還有更多 thread。

傳回值（search 模式）

results:

string

按 thread 分組的已格式化搜尋結果。每項結果都會顯示 thread 標題、thread ID、相關度分數、訊息預覽，以及用於瀏覽該 thread 的 cursor ID。

count:

number

找到的相符訊息數目。

ModelByInputTokens

ModelByInputTokens 根據輸入 token 數目選擇模型。它會選擇能涵蓋實際輸入大小的最小閾值所對應的模型。

建構函式

new ModelByInputTokens(config)

其中 config 是一個含有 upTo 鍵的物件，這些鍵會將 token 閾值（數字）對應至目標模型。

範例

import { ModelByInputTokens } from '@mastra/memory'

const selector = new ModelByInputTokens({
  upTo: {
    10_000: 'google/gemini-2.5-flash', // Fast for small inputs
    40_000: 'openai/gpt-5-mini', // Stronger for medium inputs
    1_000_000: 'openai/gpt-5.6-sol', // Most capable for large inputs
  },
})

行為

• 閾值會在內部排序，因此設定物件中的次序並不重要。
• inputTokens ≤ smallest threshold → 使用該閾值的模型
• inputTokens > largest threshold → resolve() 會拋出錯誤。如果在 OM Observer 或 Reflector 執行期間發生此情況，OM 會透過 TripWire 中止，因此呼叫者會收到空白的 text 結果或串流的 tripwire，而非正常的 assistant 回應。
• OM 會計算 Observer 或 Reflector 呼叫的輸入 token 數目，並直接解析相符的模型層級

方法

resolve:

(inputTokens: number) => MastraModelConfig

傳回指定輸入 token 數目所對應的模型。如果 inputTokens 超出所設定的最大閾值，便會拋出錯誤。在 OM 執行期間發生此情況時，呼叫者會收到 TripWire／空白文字結果，而非正常的 assistant 回應。

getThresholds:

() => number[]

按升序傳回已設定的閾值，有助進行內省。

相關內容

• Observational Memory
• Memory 概覽
• Memory 類別
• Memory Processors
• Processors