Observational Memory

追加バージョン： @mastra/memory@1.1.0

Observational Memory（OM）は、長いコンテキストを扱う Agent Memory のための Mastra の Memory システムです。Observer は会話を監視して観察結果を作成します。Reflector は、関連項目を組み合わせ、全体的なパターンを要約して観察結果を再構成します。この2つが連携して観察ログを維持し、ログが増えるにつれて生のメッセージ履歴を置き換えます。

使用方法

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 を設定すると、google/gemini-2.5-flash をデフォルトモデルとして OM が有効になります。設定オブジェクトを渡す場合は、トップレベルの model、または observation.model と reflection.model の一方もしくは両方を設定します。すべてのモデルフィールドを省略すると、OM は google/gemini-2.5-flash にフォールバックします。

Observer の入力はマルチモーダルに対応しています。OM は Observer 用に作成するトランスクリプトに [Image #1: screenshot.png] のようなテキストプレースホルダーを残し、可能な場合は元の画像パートも送信します。これは単一スレッドの観察と複数スレッドのバッチ観察の両方に適用されます。画像以外のファイルはプレースホルダーとしてのみ表示されます。

OM は高速なローカルトークン推定を使ってしきい値を判定します。テキストには tokenx を使用し、画像形式の入力には Provider を考慮したヒューリスティクスと、メタデータが不完全な場合の決定論的フォールバックを使用します。

enabled?:

boolean

= true

Observational Memory を有効または無効にします。設定オブジェクトで省略した場合のデフォルトは true です。明示的に無効にするのは enabled: false のみです。

model?:

string | LanguageModel | DynamicModel | ModelByInputTokens | ModelWithRetries[]

= 'google/gemini-2.5-flash'

Observer Agent と Reflector Agent の両方に使用するモデル。両方のモデルを一度に設定します。observation.model または reflection.model とは併用できず、両方を設定するとエラーがスローされます。これと observation.model／reflection.model をすべて省略すると、OM は google/gemini-2.5-flash にフォールバックします。デフォルトモデル（google/gemini-2.5-flash）を明示的に使用するには "default" を指定します。

scope?:

'resource' | 'thread'

= 'thread'

観察結果の Memory スコープ。'thread' は観察結果をスレッドごとに保持します。'resource'（試験的）は、あるリソースの全スレッドで観察結果を共有し、会話をまたぐ Memory を可能にします。

activateAfterIdle?:

number | string | false | "auto"

非アクティブになった後、observation.messageTokens に達する前でも、バッファーされた観察結果を強制的に有効化するまでの時間。300_000 のようなミリ秒の数値、"5m" や "1hr" のような期間文字列、Provider を考慮したプロンプトキャッシュ TTL を使う "auto"、または継承した観察のアイドル時有効化を無効にする false を指定できます。Reflection はこの設定を継承しません。Reflection でもアイドル時有効化を使うには reflection.activateAfterIdle を指定します。

activateOnProviderChange?:

boolean

= false

実行側の Provider またはモデルが変わったとき、バッファーされた観察結果を強制的に有効化します。Reflection はこの設定を継承しません。Reflection でも Provider 変更時の有効化を使うには reflection.activateOnProviderChange を指定します。

shareTokenBudget?:

boolean

= false

メッセージと観察結果でトークン予算を共有します。有効にすると、総予算は observation.messageTokens + reflection.observationTokens になります。観察結果が少ないときはメッセージがより多くの領域を使用でき、その逆も可能です。柔軟な割り当てによりコンテキストを最大限活用できます。shareTokenBudget はまだ非同期バッファリングと互換性がありません。このオプションを使用する場合は observation: { bufferTokens: false } を設定する必要があります（これは一時的な制限です）。

temporalMarkers?:

boolean

= false

スレッド内の前のメッセージから10分以上経過している場合、新しいユーザーメッセージの前に時間差を知らせるマーカーを挿入します。マーカーは Memory に永続化され、クライアントが特別に表示できるインラインリマインダーイベントとして出力されます。また、出来事が発生した時点を観察結果に結び付けられるよう Observer にも表示されます。

retrieval?:

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

= false

Agent が観察結果の基になった生のメッセージ履歴を検索できるようにします。観察グループは元のメッセージへの永続的なポインターを保持し、Agent が参照できるよう recall Tool が登録されます。true はデフォルトでスレッド横断の参照を有効にします。{ vector: true } は Memory のベクトルストアと Embedder を使うセマンティック検索も有効にします。{ scope: 'thread' } は recall Tool を現在のスレッドだけに制限します。デフォルトのスコープは 'resource' です。{ instructions: '...' } は Mastra 組み込みの取得指示の後に、アプリケーション固有の recall ガイダンスを追加します。

hooks?:

ObserveHooks

観察／Reflection の各サイクル（手動の observe()／reflect() API、ターン駆動の同期観察、fire-and-forget の非同期バッファリング）で実行されるライフサイクルフック。コールバックは threadId／resourceId／trigger の呼び出しコンテキスト（'manual' | 'turn-sync' | 'async-buffer'）を受け取ります。終了フック（onObservationEnd／onReflectionEnd）は、さらに OM モデル呼び出しのトークン usage と providerMetadata（AI Gateway などの Provider が呼び出しごとのコストを報告する場所）を受け取るため、アプリは Observer／Reflector モデルをミドルウェアでラップせずに OM モデルの費用を集計できます。非同期バッファリングのサイクルが失敗してもスローされず、終了フックの error フィールドで報告されます。これらのフックがスローしたエラーは捕捉されてログに記録され、サイクルを失敗させることはありません。

observation?:

ObservationalMemoryObservationConfig

観察ステップの設定。Observer Agent を実行するタイミングと動作を制御します。

ObservationalMemoryObservationConfig

model?:

string | LanguageModel | DynamicModel | ModelByInputTokens | ModelWithRetries[]

Observer Agent のモデル。トップレベルの model も指定されている場合は設定できません。どちらも未設定の場合は reflection.model にフォールバックします。

instruction?:

string

Observer のシステムプロンプトに追加するカスタム指示。ドメイン固有の設定や優先事項など、Observer が注目する内容を調整できます。

threadTitle?:

boolean

true の場合、Observer が短いスレッドタイトルを提案し、会話の話題が大きく変わるとタイトルを更新します。オプトイン機能で、デフォルトは無効です。

extract?:

Extractor[]

観察後に抽出するカスタム値。スキーマなし Extractor は Observer の出力内で要求されます。スキーマ付き Extractor は後続の構造化出力呼び出しとして実行され、スレッドの OM メタデータに保存されます。

manageWorkingMemory?:

boolean

OM の抽出を通じて Observer にワーキングメモリを管理させます。WorkingMemoryExtractor を追加し、workingMemory.agentManaged のデフォルトを false、workingMemory.useStateSignals のデフォルトを true にします。ワーキングメモリの更新を参照してください。

observeAttachments?:

'auto' | boolean | string[]

プレースホルダーのテキスト行とともに Observer モデルへ転送する画像／ファイル添付を制御します。true（デフォルト）はすべて転送し、false はプレースホルダーを残して添付をすべて除外します。'auto' は Provider の機能レジストリを参照し、Observer モデルがマルチモーダル入力に対応する場合、または機能データがない場合に転送します。配列は大文字小文字を区別しない mimeType の許可リストで、完全一致（'application/pdf'）、ワイルドカードのサブタイプ（'image/*'）、すべてを表す '*' に対応します。Tool 結果の添付にも同じ規則が適用されます。

messageTokens?:

number

観察を開始する未観察メッセージのトークン数。未観察メッセージがこのしきい値を超えると Observer Agent が呼び出されます。テキストは tokenx でローカル推定します。画像パートには可能な限りモデルを考慮したヒューリスティクスを使い、メタデータが不完全な場合は決定論的にフォールバックします。画像形式の file パートも同様にカウントします。

maxTokensPerBatch?:

number

リソーススコープで複数スレッドを観察するときの、バッチあたりの最大トークン数。スレッドはこのサイズのバッチに分割され、並列処理されます。値を小さくすると並列度と API 呼び出し回数が増えます。

modelSettings?:

ObservationalMemoryModelSettings

Observer Agent のモデル設定。maxOutputTokens: 100_000 のデフォルトは、デフォルトのモデル選択（model 未設定、"default"、または ModelByInputTokens セレクター）でのみ適用されます。カスタムモデルには maxOutputTokens のデフォルトがありません。

ObservationalMemoryModelSettings

temperature?:

number

生成時の Temperature。値を低くすると出力の一貫性が高まります。

maxOutputTokens?:

number

最大出力トークン数。観察結果の切り捨てを防ぐには大きな値を設定します。デフォルトの 100000 はデフォルトのモデル選択時だけ適用され、カスタムモデルには適用されません。

providerOptions?:

ProviderOptions

Google の thinking 設定など、Observer Agent に渡す Provider 固有のオプション。

bufferTokens?:

number | false

バックグラウンドで観察をバッファリングする頻度。0～1 は messageTokens に対する割合で、0.25 ならしきい値の 25% ごとにバッファリングします。1 以上は絶対トークン数で、5000 なら 5,000 トークンごとにバッファリングします。観察結果は messageTokens のしきい値に達するまで保存され、その後 LLM 呼び出しをブロックせず即座に有効化されます。messageTokens 未満になる必要があります。観察と Reflection の非同期バッファリングをすべて無効にするには false を設定します。

bufferOnIdle?:

boolean

Agent のターンが終了してアイドルになったとき、バックグラウンドで観察をバッファリングします。ステップ中の非同期バッファリングを制御する bufferTokens とは別の設定です。次のターンや messageTokens のしきい値を待たずに短いアイドルターンをバッファリングするには true を設定します。

bufferActivation?:

number

バッファーされた観察結果の有効化時に消去するメッセージウィンドウの量。0～1 は削除する messageTokens の割合で、0.8 は履歴の約 80% を削除します。1000 以上は保持するトークン数で、4000 は有効化後に約 4k を保持します。割合は大きいほど多く削除し、トークン数は大きいほど多く保持します。

activateAfterIdle?:

number | string | false | "auto"

非アクティブになってから、バッファーされた観察結果を強制的に有効化するまでの時間。ミリ秒、期間文字列、Provider を考慮したプロンプトキャッシュ 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% で強制します。100 以上は絶対トークン数で、messageTokens より大きい必要があります。messageTokens から blockAfter までは非同期バッファリングと有効化だけが実行されます。bufferTokens 設定時のみ有効で、非同期バッファリング有効時のデフォルトは 1.2 です。

previousObserverTokens?:

number | false

Observer の過去の観察結果コンテキストに対する任意のトークン予算。数値を設定すると、最新の観察結果と可能な限り 🔴 の項目を保持しつつ、予算内に収まるよう末尾側を残して切り詰めます。バッファーされた Reflection が保留中の場合、Reflection 済みの行は切り詰め前に要約へ置き換えられます。過去の観察結果をすべて省略するには 0、切り詰めを明示的に無効にするには false を設定します。

reflection?:

ObservationalMemoryReflectionConfig

Reflection ステップの設定。Reflector Agent を実行するタイミングと動作を制御します。

ObservationalMemoryReflectionConfig

model?:

string | LanguageModel | DynamicModel | ModelByInputTokens | ModelWithRetries[]

Reflector Agent のモデル。トップレベルの model も指定されている場合は設定できません。どちらも未設定の場合は observation.model にフォールバックします。

instruction?:

string

Reflector のシステムプロンプトに追加するカスタム指示。特定の情報を優先するなど、観察結果の統合方法を調整できます。

extract?:

Extractor[]

Reflection 後に抽出するカスタム値。スキーマなし Extractor は Reflector の出力内で要求されます。スキーマ付き Extractor は後続の構造化出力呼び出しとして実行され、スレッドの OM メタデータに保存されます。

observationTokens?:

number

Reflection を開始する観察結果のトークン数。観察結果がこのしきい値を超えると Reflector Agent が呼び出され、内容を要約します。

modelSettings?:

ObservationalMemoryModelSettings

Reflector Agent のモデル設定。maxOutputTokens: 100_000 のデフォルトは、デフォルトのモデル選択（model 未設定、"default"、または ModelByInputTokens セレクター）でのみ適用されます。カスタムモデルには maxOutputTokens のデフォルトがありません。

ObservationalMemoryModelSettings

temperature?:

number

生成時の Temperature。値を低くすると出力の一貫性が高まります。

maxOutputTokens?:

number

最大出力トークン数。観察結果の切り捨てを防ぐには大きな値を設定します。デフォルトの 100000 はデフォルトのモデル選択時だけ適用され、カスタムモデルには適用されません。

providerOptions?:

ProviderOptions

Google の thinking 設定など、Reflector Agent に渡す Provider 固有のオプション。

bufferActivation?:

number

バックグラウンド Reflection を開始するタイミングを observationTokens に対する割合（0～1）で指定します。0.5 は観察結果がしきい値の 50% に達すると開始します。しきい値全体に達すると、バッファーされた Reflection が対象範囲の観察結果を置き換え、その後に追加された新しい観察結果は保持されます。

activateAfterIdle?:

number | string | false | "auto"

非アクティブになってから、バッファーされた Reflection を強制的に有効化するまでの時間。ミリ秒、期間文字列、Provider を考慮したプロンプトキャッシュ TTL を使う "auto"、または false を指定できます。Reflection はトップレベルの activateAfterIdle を継承しないため、明示的な設定が必要です。現在、単独の ObservationalMemory クラスでのみ適用され、new Memory(...) では効果がありません。

activateOnProviderChange?:

boolean

実行側の Provider またはモデルが変わったとき、バッファーされた Reflection を強制的に有効化します。Reflection はトップレベルの activateOnProviderChange を継承しないため、明示的な設定が必要です。現在、単独の ObservationalMemory クラスでのみ適用され、new Memory(...) では効果がありません。

blockAfter?:

number

バックグラウンド Reflection が追いつかない場合に、同期（ブロッキング）Reflection を強制する安全策。1 以上 100 未満は observationTokens の倍率で、1.2 はしきい値の 120% で強制します。100 以上は絶対トークン数で、observationTokens より大きい必要があります。observationTokens から blockAfter までは非同期バッファリングと有効化だけが実行されます。bufferActivation 設定時のみ有効で、非同期 Reflection 有効時のデフォルトは 1.2 です。

トークン推定メタデータキャッシュ

OM はトークンペイロードの推定値を永続化し、繰り返しカウントするときに以前の推定結果を再利用できるようにします。

• パート単位のキャッシュ：part.providerMetadata.mastra。
• 文字列コンテンツのフォールバックキャッシュ：パートが存在しない場合のメッセージ単位メタデータ。
• キャッシュのバージョンまたは Tokenizer のソースが一致しない場合、キャッシュ項目は無視され再計算されます。
• メッセージ単位と会話単位のオーバーヘッドは実行時に常に再計算され、キャッシュされません。
• data-* と reasoning のパートはスキップされ、キャッシュ項目も作成されません。

Extractor API

Extractor は、観察または Reflection の実行中に OM が抽出する値を定義します。current-task、suggested-response、thread-title などの OM 組み込み値にも、カスタム値と同じ Extractor パイプラインが使用されます。

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 から派生する読み取り専用プロパティで、コンストラクターオプションではありません。永続化する値と XML タグ用の安定した識別子です。slug には小文字、数字、ハイフンを使用します。組み込み slug と予約済み XML タグはカスタム Extractor で使用できません。

instructions:

string | (context) => string

抽出する内容と値を更新するタイミングの指示。実行時コンテキストから指示を生成するには関数を使用します。

schema?:

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

構造化抽出に使用する任意の Zod スキーマ。指定すると、OM の主処理後に構造化出力呼び出しを実行します。省略すると、Observer または Reflector の応答に含まれるインライン文字列 Extractor になります。実行時コンテキストからスキーマを生成するには関数を使用します。

includePreviousExtraction?:

boolean

= true

以後の OM 実行時に、以前の抽出結果を Extractor に表示するかを制御します。現在の OM 実行だけから取得する値には false を設定します。

metadataKeyPath?:

string | false

= 'extracted.<slug>'

抽出値の永続化に使用する、ドット区切りの OM メタデータパス。OM メタデータへの永続化をすべて省略するには false を設定します。

onExtracted?:

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

カスタム Extractor が値を返してからメタデータを永続化するまでの間に呼び出される任意のフック。値を返すと抽出値を置き換え、スローすると抽出失敗として記録されます。

抽出の動作

• 抽出された値は、スレッドの OM メタデータ内の om.extracted に保存されます。
• 組み込み Extractor の値は、互換性のため currentTask、suggestedResponse、threadTitle の各メタデータフィールドにも反映されます。
• thread-title がスレッドタイトルを更新するのは、observation.threadTitle が有効な場合だけです。
• observation.extract は観察中に、reflection.extract は Reflection 中に実行されます。
• スキーマ付き Extractor は、後続の構造化出力リクエストを追加します。
• スキーマなし Extractor は、Observer または Reflector の出力に直接含まれるインライン文字列 Extractor です。
• 動的 Extractor 関数は、利用可能な場合に source、threadId、resourceId、mainAgent、memory、requestContext を含む実行時コンテキストを受け取ります。
• WorkingMemoryExtractor は通常の Extractor パイプラインを使い、アクティブな Memory インスタンスを介してワーキングメモリを更新します。ワーキングメモリに JSON Schema がある場合は構造化抽出を使用し、OM メタデータへの永続化を省略するため、ワーキングメモリのペイロードが OM の抽出済みメタデータに重複して保存されることはありません。
• observationalMemory.observation.manageWorkingMemory は WorkingMemoryExtractor を追加し、workingMemory.agentManaged のデフォルトを false にします。ワーキングメモリが有効な場合、workingMemory.useStateSignals のデフォルトは true になります。
• 抽出の失敗は OM マーカーデータで報告され、正常に抽出された他の値は破棄されません。

使用例

ワーキングメモリの更新

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 を設定します。

カスタムしきい値を使うリソーススコープ（試験的）

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,
        },
      },
    },
  }),
})

共有トークン予算

shareTokenBudget を有効にすると、総予算は observation.messageTokens + reflection.observationTokens（この例では 100k）になります。観察結果が 30k トークンしか使わない場合、メッセージは最大 70k まで拡張できます。メッセージが短い場合は、Reflection が開始されるまで観察結果により多くの余裕ができます。

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 のモデルルーターにある任意のモデルを使用できます。

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 トークンごと）
• observation.bufferActivation: 0.8：有効化時に、しきい値の 20% だけが残るようメッセージを削除します
• バッファーされた観察結果には、有効化後も保持されて会話の連続性を維持する継続ヒント（suggestedResponse、currentTask）が含まれます
• reflection.bufferActivation: 0.5：観察結果のしきい値の 50% でバックグラウンド Reflection を開始します

カスタマイズするには、次のように設定します。

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 を設定すると、観察と Reflection の両方で非同期バッファリングが無効になります。観察と Reflection は、それぞれのしきい値に達したときに同期実行されます。

注記

非同期バッファリングは scope: 'resource' ではサポートされず、リソーススコープでは自動的に無効になります。

ストリーミングデータパート

Observational Memory は Agent の実行中に型付きデータパートを出力します。クライアントはこれをリアルタイムの UI フィードバックに使用できます。データパートは Agent の応答とともにストリーミングされます。

Extractor の結果を読み取る

どちらの完了イベントも、data ペイロードに 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 のメタデータはすでに永続化されているものの、バッファーされた内容がまだ有効化を待っているバックグラウンド処理の完了を報告します。DataOmBufferingEndPart にも同じ Extractor フィールドがあり、どちらの型も @mastra/memory/processors からエクスポートされます。利用側の例は、ストリームから抽出値を読み取るを参照してください。

data-om-status

モデル生成の前に、Agent のループステップごとに1回出力されます。両方のコンテキストウィンドウのトークン使用量や、非同期でバッファーされた内容の状態を含む、現在の Memory 状態のスナップショットを提供します。

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 は圧縮後の結果、つまり Reflection が有効になったときに元の観察結果を置き換える内容のサイズです。クライアントはこの2つの値を使って圧縮率を表示できます。

クライアントは生の値から割合と有効化後の推定値を算出できます。

// 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 Agent または Reflector Agent が処理を開始したときに出力されます。

cycleId:

string

このサイクルの一意な ID。start／end／failed マーカー間で共有されます。

operationType:

'observation' | 'reflection'

観察処理か Reflection 処理かを示します。

startedAt:

string

処理を開始した ISO タイムスタンプ。

tokensToObserve:

number

このバッチで処理するメッセージトークン（入力）。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

threadIds:

string[]

このバッチに含まれるすべてのスレッド ID（リソーススコープの場合）。

config:

ObservationMarkerConfig

観察時点の messageTokens、observationTokens、scope のスナップショット。

data-om-observation-end

観察または Reflection が正常に完了したときに出力されます。

cycleId:

string

対応する start マーカーと一致します。

operationType:

'observation' | 'reflection'

完了した処理の種類。

completedAt:

string

処理が完了した ISO タイムスタンプ。

durationMs:

number

所要時間（ミリ秒）。

tokensObserved:

number

処理されたメッセージトークン（入力）。

observationTokens:

number

Observer による圧縮後の観察結果トークン（出力）。

observations?:

string

生成された観察結果のテキスト。

currentTask?:

string

Observer が抽出した現在のタスク。

suggestedResponse?:

string

Observer が抽出した推奨応答。

extractedValues?:

Record<string, unknown>

この OM 処理で抽出された値。Extractor の slug をキーとします。

extractionFailures?:

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

この OM 処理で発生した Extractor の失敗。正常に抽出された値は引き続き含まれます。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

data-om-observation-failed

観察または Reflection が失敗したときに出力されます。システムは同期処理にフォールバックします。

cycleId:

string

対応する start マーカーと一致します。

operationType:

'observation' | 'reflection'

失敗した処理の種類。

failedAt:

string

失敗が発生した ISO タイムスタンプ。

durationMs:

number

失敗するまでの時間（ミリ秒）。

tokensAttempted:

number

処理を試みたメッセージトークン（入力）。

error:

string

エラーメッセージ。

observations?:

string

表示可能な部分的な内容。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

data-om-buffering-start

バックグラウンドで非同期バッファリングが始まったときに出力されます。バッファリングは、主しきい値に達する前に観察結果または Reflection を事前計算します。

cycleId:

string

このバッファリングサイクルの一意な ID。

operationType:

'observation' | 'reflection'

バッファリング対象の処理の種類。

startedAt:

string

バッファリングを開始した ISO タイムスタンプ。

tokensToBuffer:

number

このサイクルでバッファリングするメッセージトークン（入力）。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

threadIds:

string[]

バッファリング対象のすべてのスレッド ID（リソーススコープの場合）。

config:

ObservationMarkerConfig

バッファリング時点の設定のスナップショット。

data-om-buffering-end

非同期バッファリングが完了したときに出力されます。内容は保存されていますが、メインコンテキストではまだ有効になっていません。

cycleId:

string

対応する buffering-start マーカーと一致します。

operationType:

'observation' | 'reflection'

バッファリングされた処理の種類。

completedAt:

string

バッファリングが完了した ISO タイムスタンプ。

durationMs:

number

所要時間（ミリ秒）。

tokensBuffered:

number

バッファリングされたメッセージトークン（入力）。

bufferedTokens:

number

Observer による圧縮後の観察結果トークン（出力）。

observations?:

string

バッファーされた内容。

extractedValues?:

Record<string, unknown>

このバッファー OM 処理で抽出された値。Extractor の slug をキーとします。

extractionFailures?:

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

このバッファー OM 処理で発生した Extractor の失敗。正常に抽出された値は引き続き含まれます。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

data-om-buffering-failed

非同期バッファリングが失敗したときに出力されます。しきい値に達すると、システムは同期処理にフォールバックします。

cycleId:

string

対応する buffering-start マーカーと一致します。

operationType:

'observation' | 'reflection'

失敗した処理の種類。

failedAt:

string

失敗が発生した ISO タイムスタンプ。

durationMs:

number

失敗するまでの時間（ミリ秒）。

tokensAttempted:

number

バッファリングを試みたメッセージトークン（入力）。

error:

string

エラーメッセージ。

observations?:

string

部分的な内容。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

data-om-activation

バッファーされた観察結果または Reflection が有効化され、アクティブなコンテキストウィンドウへ移動したときに出力されます。これは即時処理であり、LLM 呼び出しは発生しません。

cycleId:

string

この有効化イベントの一意な ID。

operationType:

'observation' | 'reflection'

有効化された内容の種類。

activatedAt:

string

有効化が発生した ISO タイムスタンプ。

chunksActivated:

number

有効化されたバッファーチャンク数。

tokensActivated:

number

有効化されたチャンクのメッセージトークン（入力）。観察の有効化ではメッセージウィンドウから削除され、Reflection の有効化では圧縮された観察結果トークンを示します。

observationTokens:

number

有効化後の観察結果トークン。

messagesActivated:

number

有効化によって観察されたメッセージ数。

generationCount:

number

現在の Reflection 生成回数。

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

有効化の契機となった現在の実行側モデル識別子。

recordId:

string

OM レコードの ID。

threadId:

string

このスレッドの ID。

config:

ObservationMarkerConfig

有効化時点の設定のスナップショット。

data-om-thread-update

Observer がスレッドタイトルを更新したときに出力されます。observation.threadTitle が有効な場合にのみ出力されます。

cycleId:

string

この観察サイクルの一意な ID。観察マーカーと共有されます。

threadId:

string

更新されたスレッドの ID。

oldTitle?:

string

以前のスレッドタイトル。スレッドにタイトルがなかった場合は undefined です。

newTitle:

string

新しいスレッドタイトル。

timestamp:

string

この更新が発生した時刻。

単独での使用

ほとんどの場合は、前述の Memory クラスを使用してください。ObservationalMemory を直接使用する方法は、主にベンチマーク、実験、または他の Processor（ガードレールなど）との実行順序を制御する必要がある場合に役立ちます。

ObservationalMemory クラスがエンジンです。Agent に組み込むには ObservationalMemoryProcessor でラップします。この Processor には、メッセージの読み込みと永続化に使用する Memory インスタンスが必要です。ストレージアダプターでは stores.memory が任意として型付けされているため、非 null アサーション（または実行時チェック）が必要です。

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

観察結果を永続化するストレージアダプター。MastraStorage.stores.memory の MemoryStorage インスタンスである必要があります。

onDebugEvent?:

(event: ObservationDebugEvent) => void

観察イベント用のデバッグコールバック。観察関連のイベントが発生するたびに呼び出され、デバッグや観察フローの把握に役立ちます。

obscureThreadIds?:

boolean

= false

有効にすると、スレッド ID は観察コンテキストに含める前にハッシュ化されます。これにより、LLM がスレッド識別子のパターンを認識することを防ぎます。Memory クラスでリソーススコープを使用すると自動的に有効になります。

Recall Tool

retrieval に truthy な値を設定すると recall Tool が登録され、Agent は観察グループ範囲の基になった生のメッセージをページ単位で参照できます。デフォルト（スコープは 'resource'）では、スレッドの一覧表示（mode: "threads"）、別スレッドの参照（threadId）、スレッド横断検索に対応します。retrieval: { vector: true } を指定すると、セマンティック検索（mode: "search"）も利用できます。Tool を現在のスレッドだけに制限するには scope: 'thread' を設定します。Tool は Agent の Tool リストへ自動的に追加されます。

Mastra は、スコープを考慮した使用方法の指示も Agent のコンテキストへ注入します。リソーススコープで vector: true を指定した場合は、検索結果が不適切なときのスレッド探索へのフォールバックを含め、search、threads、messages 間のルーティングを扱います。vector: true を指定しない場合、指示は threads と messages の参照だけを扱うため、未設定の検索モードへ Agent を誘導しません。リソーススコープの指示は観察グループがまだ存在しない段階でも注入されるため、Agent は最初のメッセージから他のスレッドを参照できます。組み込み指示の後にアプリケーション固有のガイダンスを追加するには、retrieval: { instructions: '...' } を使用します。

パラメーター

mode?:

'messages' | 'threads' | 'search'

= 'messages'

取得対象。"messages"（デフォルト）はメッセージ履歴、"threads" は現在のユーザーの全スレッドをページ単位で取得します。"search" は全スレッドから意味的に類似するメッセージを検索します（ベクトルストアと Embedder が必要）。

query?:

string

mode: "search" の検索クエリ。現在のユーザーの全スレッドから、このテキストと意味的に類似するメッセージを検索します。

cursor?:

string

recall クエリの基点となるメッセージ ID。観察グループの範囲から開始または終了 ID を取り出します（例：_range: \startId:endId\_ から startId または endId を使用）。範囲文字列を直接渡すと、正しい ID の取り出し方を示すヒントが返されます。mode: "messages" で cursor と threadId を省略すると、anchor で設定した位置から現在のスレッドを参照します。

threadId?:

string

ID を指定して別のスレッドを参照します。アクティブなスレッドには "current" を渡します。まず mode: "threads" でスレッド ID を取得してください。cursor なしで指定すると、スレッドの先頭から読み取ります。

anchor?:

'start' | 'end'

= 'start'

cursor を指定しない mode: "messages" で、スレッドの先頭（古い順）または末尾（新しい順）のどちらからページングするかを指定します。

page?:

number

= 1

ページネーションのオフセット。メッセージでは正の値で cursor から前方、負の値で後方へ移動します。スレッドでは 0 始まりのページ番号です。メッセージの場合、0 は 1 として扱われます。

limit?:

number

= 20

1ページあたりに返す項目の最大数。

detail?:

'low' | 'high'

= 'low'

メッセージパートごとに表示する内容量を制御します。'low' は切り詰めたテキストと Tool 名を位置インデックス（[p0]、[p1]）付きで表示します。'high' は Tool の引数と結果を含む完全な内容を表示し、1回の呼び出しにつき1パートに制限して続きのヒントを示します。

partType?:

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

この種類のメッセージパートだけを結果に含めます。mode: "messages" にのみ適用されます。

toolName?:

string

この Tool 名に一致する tool-call と tool-result パートだけを結果に含めます。mode: "messages" にのみ適用されます。

partIndex?:

number

位置インデックスを指定し、1つのメッセージパートを完全な詳細度で取得します。低詳細度の recall で [p1] に必要なパートが見つかった場合、partIndex: 1 で再度呼び出すと、全パートを読み込まずに完全な内容を確認できます。

before?:

string

mode: "threads" 専用。この日時より前に作成されたスレッドだけに絞り込みます。ISO 8601 形式を使用できます（例："2026-03-15"、"2026-03-10T00:00:00Z"）。

after?:

string

mode: "threads" 専用。この日時より後に作成されたスレッドだけに絞り込みます。ISO 8601 形式を使用できます（例："2026-03-01"、"2026-03-10T00:00:00Z"）。

戻り値（messages モード）

messages:

string

整形済みのメッセージ内容。形式は detail レベルによって異なります。

count:

number

このページのメッセージ数。

cursor:

string

このクエリで使用した cursor メッセージ ID。

page:

number

返されたページ番号。

limit:

number

このクエリで使用した上限値。

detail:

'low' | 'high'

このクエリで使用した詳細度。

hasNextPage:

boolean

このページより後にメッセージが存在するかどうか。

hasPrevPage:

boolean

このページより前にメッセージが存在するかどうか。

truncated?:

boolean

トークン予算によって出力が制限された場合に存在し、true になります。Agent はページネーションまたは partIndex を使って残りの内容にアクセスできます。

tokenOffset?:

number

truncated が true の場合に切り詰められた概算トークン数。

戻り値（threads モード）

threads:

string

整形済みのスレッド一覧。各スレッドのタイトル、ID、日付を表示します。現在のスレッドには ← current が付きます。

count:

number

返されたスレッド数。

page:

number

返されたページ番号。

hasMore:

boolean

次のページにスレッドが存在するかどうか。

戻り値（search モード）

results:

string

スレッドごとにまとめた整形済み検索結果。各結果にはスレッドタイトル、スレッド ID、関連度スコア、メッセージのプレビュー、そのスレッドを参照するための cursor ID が表示されます。

count:

number

見つかった一致メッセージ数。

ModelByInputTokens

ModelByInputTokens は入力トークン数に基づいてモデルを選択します。実際の入力サイズを収められる最小のしきい値に対応するモデルを選びます。

コンストラクター

new ModelByInputTokens(config)

config は、トークンのしきい値（数値）を対象モデルに対応付ける upTo キーを持つオブジェクトです。

使用例

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 によって中止されるため、呼び出し元は通常の Assistant 応答ではなく、空の text 結果またはストリーミングされた tripwire を受け取ります。
• OM は Observer または Reflector 呼び出しの入力トークン数を計算し、一致するモデル階層を直接解決します

メソッド

resolve:

(inputTokens: number) => MastraModelConfig

指定した入力トークン数に対応するモデルを返します。inputTokens が設定済みの最大しきい値を超えるとスローします。OM の実行中に発生した場合、呼び出し元は通常の Assistant 応答ではなく TripWire／空テキストの結果を受け取ります。

getThresholds:

() => number[]

設定されたしきい値を昇順で返します。内部状態の確認に役立ちます。

関連項目

• Observational Memory
• Memory の概要
• Memory クラス
• Memory Processor
• Processor