title

Reference: Observational Memory | Memory

description

API reference for Observational Memory in Mastra — a three-tier memory system that uses Observer and Reflector agents to maintain long-term memory across conversations.

packages

@mastra/core

@mastra/memory

Observational Memory

Added in: @mastra/memory@1.1.0

Observational Memory (OM) is Mastra's memory system for long-context agentic memory. Two background agents — an Observer that watches conversations and creates observations, and a Reflector that restructures observations by combining related items, reflecting on overarching patterns, and condensing where possible — maintain an observation log that replaces raw message history as it grows.

Usage

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

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

Configuration

The observationalMemory option accepts true, a configuration object, or false. Setting true enables OM with google/gemini-2.5-flash as the default model. When passing a config object, a model must be explicitly set — either at the top level, or on observation.model and/or reflection.model.

Observer input is multimodal-aware. OM keeps text placeholders like [Image #1: screenshot.png] in the transcript it builds for the Observer, and also sends the underlying image parts when possible. This applies to both single-thread observation and batched multi-thread observation. Non-image files appear as placeholders only.

OM performs thresholding with fast local token estimation. Text uses tokenx, and image-like inputs use provider-aware heuristics plus deterministic fallbacks when metadata is incomplete.

Token estimate metadata cache

OM persists token payload estimates so repeated counting can reuse prior token estimation work.

Part-level cache: part.providerMetadata.mastra.
String-content fallback cache: message-level metadata when no parts exist.
Cache entries are ignored and recomputed if cache version/tokenizer source doesn't match.
Per-message and per-conversation overhead are always recomputed at runtime and aren't cached.
data-* and reasoning parts are skipped and don't receive cache entries.

Extractor API

Extractor defines a value that OM should extract during observation or reflection. Built-in OM values such as current-task, suggested-response, and thread-title use the same extractor pipeline as custom values.

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

const memory = new Memory({
  options: {
    observationalMemory: {
      model: '__GATEWAY_OPENAI_MODEL_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(),
            }),
          }),
        ],
      },
    },
  },
})

<PropertiesTable content={[ { name: 'name', type: 'string', description: 'Human-readable extractor name. OM slugifies this value into the extractor slug. Names must be unique after slug generation.', }, { name: 'slug', type: 'string', description: 'Generated stable identifier for persisted values and XML tags. Slugs use lowercase letters, numbers, and hyphens. Built-in slugs and reserved XML tags cannot be used by custom extractors.', }, { name: 'instructions', type: 'string | (context) => string', description: 'Instructions for what to extract and when to update the value. Use a function to derive instructions from runtime context.', }, { name: 'schema', type: 'ZodType | (context) => ZodType | undefined', description: 'Optional Zod schema for structured extraction. When provided, OM runs a follow-up structured output call after the main OM operation. When omitted, the extractor is an inline string extractor emitted in the Observer or Reflector response. Use a function to derive the schema from runtime context.', isOptional: true, }, { name: 'includePreviousExtraction', type: 'boolean', description: 'Controls whether the previous extraction is shown to the extractor on future OM runs. Set to false for values that should only come from the current OM run.', isOptional: true, defaultValue: 'true', }, { name: 'onExtracted', type: '(context) => T | void | Promise<T | void>', description: 'Optional hook called after a custom extractor returns a value and before metadata is persisted. Returning a value replaces the extracted value. Throwing records an extraction failure.', isOptional: true, }, ]} />

Extraction behavior

Extracted values are stored in thread OM metadata under om.extracted.
Built-in extractor values are also mirrored to the compatibility metadata fields currentTask, suggestedResponse, and threadTitle.
thread-title updates the thread title only when observation.threadTitle is enabled.
observation.extract runs during observation. reflection.extract runs during reflection.
Schema-backed extractors add a follow-up structured output request.
Schema-less extractors are inline string extractors emitted directly in the Observer or Reflector output.
Dynamic extractor functions receive runtime context, including source, threadId, resourceId, mainAgent, memory, and requestContext when available.
WorkingMemoryExtractor uses the normal extractor pipeline to update working memory through the active Memory instance. It uses structured extraction when working memory has a JSON schema and skips OM metadata persistence, so the working memory payload isn't duplicated under OM extracted metadata.
observationalMemory.observation.manageWorkingMemory adds WorkingMemoryExtractor, defaults workingMemory.agentManaged to false, and defaults workingMemory.useStateSignals to true when working memory is enabled.
Extraction failures are reported in OM marker data and do not discard other successful extracted values.

Examples

Working memory updates

Use observationalMemory.observation.manageWorkingMemory when OM should update working memory.

import { Memory } from '@mastra/memory'

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

Set workingMemory.agentManaged: true if the main agent should still receive working memory tool and instruction injection.

Resource scope with custom thresholds (experimental)

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

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

Shared token budget

When shareTokenBudget is enabled, the total budget is observation.messageTokens + reflection.observationTokens (100k in this example). If observations only use 30k tokens, messages can expand to use up to 70k. If messages are short, observations have more room before triggering reflection.

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

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

Custom model

By passing a model in the config, you can use any model from Mastra's model router.

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

export const agent = new Agent({
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: '__GATEWAY_OPENAI_MODEL__',
  memory: new Memory({
    options: {
      observationalMemory: {
        // highlight-next-line
        model: '__GATEWAY_OPENAI_MODEL_MINI__',
      },
    },
  }),
})

Different models per agent

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

export const agent = new Agent({
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: '__GATEWAY_OPENAI_MODEL__',
  memory: new Memory({
    options: {
      observationalMemory: {
        // highlight-start
        observation: {
          model: 'google/gemini-2.5-flash',
        },
        reflection: {
          model: '__GATEWAY_OPENAI_MODEL_MINI__',
        },
        // highlight-end
      },
    },
  }),
})

Custom instructions

Customize what the Observer and Reflector focus on by providing custom instructions:

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

export const agent = new Agent({
  name: 'health-assistant',
  instructions: 'You are a health and wellness assistant.',
  model: '__GATEWAY_OPENAI_MODEL__',
  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.',
        },
      },
    },
  }),
})

Async buffering

Async buffering is enabled by default. It pre-computes observations in the background as the conversation grows — when the messageTokens threshold is reached, buffered observations activate instantly with no blocking LLM call.

The lifecycle is: buffer → activate → remove messages → repeat. Background Observer calls run at bufferTokens intervals, each producing a chunk of observations. At threshold, chunks activate: observations move into the log, raw messages are removed from context. The blockAfter threshold forces a synchronous fallback if buffering can't keep up.

Default settings:

observation.bufferTokens: 0.2 — buffer every 20% of messageTokens (e.g. every ~6k tokens with a 30k threshold)
observation.bufferActivation: 0.8 — on activation, remove enough messages to keep only 20% of the threshold remaining
Buffered observations include continuation hints (suggestedResponse, currentTask) that survive activation to maintain conversational continuity
reflection.bufferActivation: 0.5 — start background reflection at 50% of observation threshold

To customize:

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

export const agent = new Agent({
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: '__GATEWAY_OPENAI_MODEL_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,
        },
      },
    },
  }),
})

To disable async buffering entirely:

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

Setting bufferTokens: false disables both observation and reflection async buffering. Observations and reflections will run synchronously when their thresholds are reached.

:::note

Async buffering isn't supported with scope: 'resource' and is automatically disabled in resource scope.

:::

Streaming data parts

Observational Memory emits typed data parts during agent execution that clients can use for real-time UI feedback. These are streamed alongside the agent's response.

`data-om-status`

Emitted once per agent loop step, before model generation. Provides a snapshot of the current memory state, including token usage for both context windows and the state of any async buffered content.

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 is the size of the observations that were sent to the Reflector. buffered.reflection.observationTokens is the compressed result — the size of what will replace those observations when the reflection activates. A client can use these two values to show a compression ratio.

Clients can derive percentages and post-activation estimates from the raw values:

// 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`

Emitted when the Observer or Reflector agent begins processing.

`data-om-observation-end`

Emitted when observation or reflection completes successfully.

<PropertiesTable content={[ { name: 'cycleId', type: 'string', description: 'Matches the corresponding start marker.' }, { name: 'operationType', type: "'observation' | 'reflection'", description: 'Type of operation that completed.' }, { name: 'completedAt', type: 'string', description: 'ISO timestamp when processing completed.' }, { name: 'durationMs', type: 'number', description: 'Duration in milliseconds.' }, { name: 'tokensObserved', type: 'number', description: 'Message tokens (input) that were processed.' }, { name: 'observationTokens', type: 'number', description: 'Resulting observation tokens (output) after the Observer compressed them.', }, { name: 'observations', type: 'string', description: 'The generated observations text.', isOptional: true }, { name: 'currentTask', type: 'string', description: 'Current task extracted by the Observer.', isOptional: true }, { name: 'suggestedResponse', type: 'string', description: 'Suggested response extracted by the Observer.', isOptional: true, }, { name: 'extractedValues', type: 'Record<string, unknown>', description: 'Values extracted during this OM operation, keyed by extractor slug.', isOptional: true, }, { name: 'extractionFailures', type: 'Array<{ slug: string; error: string }>', description: 'Extractor failures from this OM operation. Successful extractor values are still included.', isOptional: true, }, { name: 'recordId', type: 'string', description: 'The OM record ID.' }, { name: 'threadId', type: 'string', description: "This thread's ID." }, ]} />

`data-om-observation-failed`

Emitted when observation or reflection fails. The system falls back to synchronous processing.

`data-om-buffering-start`

Emitted when async buffering begins in the background. Buffering pre-computes observations or reflections before the main threshold is reached.

`data-om-buffering-end`

Emitted when async buffering completes. The content is stored but not yet activated in the main context.

<PropertiesTable content={[ { name: 'cycleId', type: 'string', description: 'Matches the corresponding buffering-start marker.' }, { name: 'operationType', type: "'observation' | 'reflection'", description: 'Type of operation that was buffered.' }, { name: 'completedAt', type: 'string', description: 'ISO timestamp when buffering completed.' }, { name: 'durationMs', type: 'number', description: 'Duration in milliseconds.' }, { name: 'tokensBuffered', type: 'number', description: 'Message tokens (input) that were buffered.' }, { name: 'bufferedTokens', type: 'number', description: 'Observation tokens (output) after the Observer compressed them.', }, { name: 'observations', type: 'string', description: 'The buffered content.', isOptional: true }, { name: 'extractedValues', type: 'Record<string, unknown>', description: 'Values extracted during this buffered OM operation, keyed by extractor slug.', isOptional: true, }, { name: 'extractionFailures', type: 'Array<{ slug: string; error: string }>', description: 'Extractor failures from this buffered OM operation. Successful extractor values are still included.', isOptional: true, }, { name: 'recordId', type: 'string', description: 'The OM record ID.' }, { name: 'threadId', type: 'string', description: "This thread's ID." }, ]} />

`data-om-buffering-failed`

Emitted when async buffering fails. The system falls back to synchronous processing when the threshold is reached.

`data-om-activation`

Emitted when buffered observations or reflections are activated (moved into the active context window). This is an instant operation — no LLM call is involved.

Standalone usage

Most users should use the Memory class above. Using ObservationalMemory directly is mainly useful for benchmarking, experimentation, or when you need to control processor ordering with other processors (like guardrails).

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

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

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

export const agent = new Agent({
  name: 'my-agent',
  instructions: 'You are a helpful assistant.',
  model: '__GATEWAY_OPENAI_MODEL_MINI__',
  inputProcessors: [om],
  outputProcessors: [om],
})

Standalone config

The standalone ObservationalMemory class accepts all the same options as the observationalMemory config object above, plus the following:

<PropertiesTable content={[ { name: 'storage', type: 'MemoryStorage', description: 'Storage adapter for persisting observations. Must be a MemoryStorage instance (from MastraStorage.stores.memory).', isOptional: false, }, { name: 'onDebugEvent', type: '(event: ObservationDebugEvent) => void', description: 'Debug callback for observation events. Called whenever observation-related events occur. Useful for debugging and understanding the observation flow.', isOptional: true, }, { name: 'obscureThreadIds', type: 'boolean', description: 'When enabled, thread IDs are hashed before being included in observation context. This prevents the LLM from recognizing patterns in thread identifiers. Automatically enabled when using resource scope through the Memory class.', isOptional: true, defaultValue: 'false', }, ]} />

Recall tool

When retrieval is set (any truthy value), a recall tool is registered so the agent can page through raw messages behind observation group ranges. By default (scope 'resource'), the tool supports listing threads (mode: "threads"), browsing other threads (threadId), and cross-thread search. With retrieval: { vector: true }, semantic search is available (mode: "search"). Set scope: 'thread' to restrict the tool to the current thread only. The tool is automatically added to the agent's tool list — no manual registration is needed.

Parameters

Returns (messages mode)

Returns (threads mode)

Returns (search mode)

ModelByInputTokens

ModelByInputTokens selects a model based on the input token count. It chooses the model for the smallest threshold that covers the actual input size.

Constructor

new ModelByInputTokens(config)

Where config is an object with upTo keys that map token thresholds (numbers) to model targets.

Example

import { ModelByInputTokens } from '@mastra/memory'

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

Behavior

Thresholds are sorted internally, so the order in the config object doesn't matter.
inputTokens ≤ smallest threshold → uses that threshold's model
inputTokens > largest threshold → resolve() throws an error. If this happens during an OM Observer or Reflector run, OM aborts via TripWire, so callers receive an empty text result or streamed tripwire instead of a normal assistant response.
OM computes the input token count for the Observer or Reflector call and resolves the matching model tier directly

Methods

<PropertiesTable content={[ { name: 'resolve', type: '(inputTokens: number) => MastraModelConfig', description: 'Returns the model for the given input token count. Throws if inputTokens exceeds the largest configured threshold. When this happens during an OM run, callers receive a TripWire/empty-text outcome instead of a normal assistant response.', }, { name: 'getThresholds', type: '() => number[]', description: 'Returns the configured thresholds in ascending order. Useful for introspection.', }, ]} />

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Observational Memory

Usage

Configuration

Token estimate metadata cache

Extractor API

Extraction behavior

Examples

Working memory updates

Resource scope with custom thresholds (experimental)

Shared token budget

Custom model

Different models per agent

Custom instructions

Async buffering

Streaming data parts

`data-om-status`

`data-om-observation-start`

`data-om-observation-end`

`data-om-observation-failed`

`data-om-buffering-start`

`data-om-buffering-end`

`data-om-buffering-failed`

`data-om-activation`

Standalone usage

Standalone config

Recall tool

Parameters

Returns (messages mode)

Returns (threads mode)

Returns (search mode)

ModelByInputTokens

Constructor

Example

Behavior

Methods

Related

Uh oh!

FilesExpand file tree

observational-memory.mdx

Latest commit

History

observational-memory.mdx

File metadata and controls

Observational Memory

Usage

Configuration

Token estimate metadata cache

Extractor API

Extraction behavior

Examples

Working memory updates

Resource scope with custom thresholds (experimental)

Shared token budget

Custom model

Different models per agent

Custom instructions

Async buffering

Streaming data parts

data-om-status

data-om-observation-start

data-om-observation-end

data-om-observation-failed

data-om-buffering-start

data-om-buffering-end

data-om-buffering-failed

data-om-activation

Standalone usage

Standalone config

Recall tool

Parameters

Returns (messages mode)

Returns (threads mode)

Returns (search mode)

ModelByInputTokens

Constructor

Example

Behavior

Methods

Related

`data-om-status`

`data-om-observation-start`

`data-om-observation-end`

`data-om-observation-failed`

`data-om-buffering-start`

`data-om-buffering-end`

`data-om-buffering-failed`

`data-om-activation`