Stockage composite

MastraCompositeStore permet de composer des domaines de stockage provenant de différents Providers. Utilisez-le lorsque vous avez besoin de bases de données différentes selon les usages. Par exemple, utilisez LibSQL pour la mémoire et PostgreSQL pour les Workflows.

Installation

MastraCompositeStore est inclus dans @mastra/core :

npm

npm install @mastra/core@latest

pnpm

pnpm add @mastra/core@latest

Yarn

yarn add @mastra/core@latest

Bun

bun add @mastra/core@latest

Vous devez également installer les Providers de stockage que vous souhaitez combiner :

npm

npm install @mastra/pg@latest @mastra/libsql@latest @mastra/mongodb@latest

pnpm

pnpm add @mastra/pg@latest @mastra/libsql@latest @mastra/mongodb@latest

Yarn

yarn add @mastra/pg@latest @mastra/libsql@latest @mastra/mongodb@latest

Bun

bun add @mastra/pg@latest @mastra/libsql@latest @mastra/mongodb@latest

Domaines de stockage

Mastra organise le stockage en domaines, chacun prenant en charge un type de données précis. Chaque domaine peut reposer sur un adaptateur de stockage différent, et les classes de domaine sont exportées depuis chaque package de stockage.

Domaine	Description
memory	Persistance des conversations des Agents. Stocke les fils de discussion (sessions de conversation), les messages, les ressources (identités des utilisateurs) et la mémoire de travail (contexte persistant entre les conversations).
workflows	État d'exécution des Workflows. Lorsque les Workflows sont suspendus dans l'attente d'une intervention humaine, d'événements externes ou d'une reprise planifiée, leur état est conservé ici afin de permettre leur reprise après le redémarrage du serveur.
scores	Résultats d'évaluation du système d'Evals de Mastra. Les scores et les métriques sont conservés ici pour être analysés et comparés dans le temps.
observability	Données de télémétrie, notamment les Traces et les spans. Les interactions des Agents, les appels de Tool et les requêtes aux LLM génèrent des spans regroupés en Traces pour le débogage et l'analyse des performances.
agents	Configurations des Agents stockés. Permet de définir et de mettre à jour des Agents à l'exécution sans déployer de code.
datasets	Jeux de données d'évaluation utilisés pour les exécutions d'expériences. Stocke les définitions, les schémas et les éléments versionnés des jeux de données.
experiments	Exécutions d'expériences et résultats par élément associés aux jeux de données et aux cibles.

remarque

MastraCompositeStore accepte toutes les clés de domaine ci-dessus, mais la prise en charge par les adaptateurs de stockage varie selon le package. Vous pouvez combiner les adaptateurs par domaine, mais uniquement pour les domaines qu'ils implémentent et exportent. Par exemple, l'association de memory: new MemoryLibSQL(...) et de workflows: new WorkflowsPG(...) est valide, car les deux packages exportent ces classes de domaine.

Utilisation

Composition de base

Importez directement les classes de domaine depuis chaque package de stockage, puis composez-les :

src/mastra/index.ts

import { MastraCompositeStore } from '@mastra/core/storage'
import { WorkflowsPG, ScoresPG } from '@mastra/pg'
import { MemoryLibSQL } from '@mastra/libsql'
import { Mastra } from '@mastra/core'

export const mastra = new Mastra({
  storage: new MastraCompositeStore({
    id: 'composite',
    domains: {
      memory: new MemoryLibSQL({ url: 'file:./local.db' }),
      workflows: new WorkflowsPG({ connectionString: process.env.DATABASE_URL }),
      scores: new ScoresPG({ connectionString: process.env.DATABASE_URL }),
    },
  }),
})

Avec un stockage par défaut

Utilisez default pour définir un stockage de repli, puis remplacez certains domaines :

src/mastra/index.ts

import { MastraCompositeStore } from '@mastra/core/storage'
import { PostgresStore } from '@mastra/pg'
import { MemoryLibSQL } from '@mastra/libsql'
import { Mastra } from '@mastra/core'

const pgStore = new PostgresStore({
  id: 'pg',
  connectionString: process.env.DATABASE_URL,
})

export const mastra = new Mastra({
  storage: new MastraCompositeStore({
    id: 'composite',
    default: pgStore,
    domains: {
      memory: new MemoryLibSQL({ url: 'file:./local.db' }),
    },
  }),
})

Backends mixtes

Utilisez les classes de domaine de chaque package de stockage pour acheminer différents domaines vers différents backends. L'exemple suivant stocke la mémoire et l'état des Workflows dans MongoDB, puis achemine l'Observability vers ClickHouse :

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { MastraCompositeStore } from '@mastra/core/storage'
import { ObservabilityStorageClickhouse } from '@mastra/clickhouse'
import { MemoryStorageMongoDB, WorkflowsStorageMongoDB } from '@mastra/mongodb'

export const mastra = new Mastra({
  storage: new MastraCompositeStore({
    id: 'composite',
    domains: {
      memory: new MemoryStorageMongoDB({
        uri: process.env.MONGODB_URI,
        dbName: 'mastra_memory',
      }),
      workflows: new WorkflowsStorageMongoDB({
        uri: process.env.MONGODB_URI,
        dbName: 'mastra_workflows',
      }),
      observability: new ObservabilityStorageClickhouse({
        url: process.env.CLICKHOUSE_URL,
        username: process.env.CLICKHOUSE_USERNAME,
        password: process.env.CLICKHOUSE_PASSWORD,
      }),
    },
  }),
})

Désactivation d'un domaine

Définissez un domaine sur false pour le désactiver. Un domaine désactivé n'utilise pas default comme solution de repli ; ses données ne sont donc pas conservées :

src/mastra/index.ts

import { MastraCompositeStore } from '@mastra/core/storage'
import { PostgresStore } from '@mastra/pg'
import { Mastra } from '@mastra/core'

const pgStore = new PostgresStore({
  id: 'pg',
  connectionString: process.env.DATABASE_URL,
})

export const mastra = new Mastra({
  storage: new MastraCompositeStore({
    id: 'composite',
    default: pgStore,
    domains: {
      // don't persist traces and spans
      observability: false,
    },
  }),
})

Options

id:

string

Identifiant unique de cette instance de stockage.

default?:

MastraCompositeStore

Adaptateur de stockage par défaut. Les domaines non explicitement définis dans domains utiliseront les domaines de ce stockage comme solutions de repli.

editor?:

MastraCompositeStore

Adaptateur de stockage pour les domaines appartenant à l'Editor, notamment les Agents, les blocs de prompt, les Scorers, les clients et serveurs MCP, les Workspaces et les Skills. Prioritaire sur le stockage par défaut, mais pas sur les remplacements explicites de domaines.

disableInit?:

boolean

Lorsque la valeur est true, l'initialisation automatique est désactivée. Vous devez appeler init() explicitement.

domains?:

object

Remplacements propres à chaque domaine. Chaque domaine peut provenir d'un adaptateur de stockage différent. Ils sont prioritaires sur les stockages editor et default. Définissez un domaine sur false pour le désactiver entièrement ; un domaine désactivé n'utilise ni editor ni default comme solution de repli.

object

memory?:

MemoryStorage

Stockage des fils de discussion, des messages et des ressources.

workflows?:

WorkflowsStorage

Stockage des snapshots de Workflow.

scores?:

ScoresStorage

Stockage des scores d'évaluation.

observability?:

ObservabilityStorage

Stockage des Traces et des spans.

agents?:

AgentsStorage

Stockage des configurations des Agents stockés.

datasets?:

DatasetsStorage

Stockage des métadonnées, des éléments et des versions des jeux de données.

experiments?:

ExperimentsStorage

Stockage des exécutions d'expériences et de leurs résultats par élément.

Initialisation

MastraCompositeStore initialise séparément chaque domaine configuré. Lorsqu'il est transmis à la classe Mastra, init() est appelé automatiquement :

src/mastra/index.ts

import { MastraCompositeStore } from '@mastra/core/storage'
import { MemoryPG, WorkflowsPG, ScoresPG } from '@mastra/pg'
import { Mastra } from '@mastra/core'

const storage = new MastraCompositeStore({
  id: 'composite',
  domains: {
    memory: new MemoryPG({ connectionString: process.env.DATABASE_URL }),
    workflows: new WorkflowsPG({ connectionString: process.env.DATABASE_URL }),
    scores: new ScoresPG({ connectionString: process.env.DATABASE_URL }),
  },
})

export const mastra = new Mastra({
  storage, // init() called automatically
})

Si vous utilisez directement le stockage, appelez explicitement init() :

import { MastraCompositeStore } from '@mastra/core/storage'
import { MemoryPG } from '@mastra/pg'

const storage = new MastraCompositeStore({
  id: 'composite',
  domains: {
    memory: new MemoryPG({ connectionString: process.env.DATABASE_URL }),
  },
})

await storage.init()

// Access domain-specific stores via getStore()
const memoryStore = await storage.getStore('memory')
const thread = await memoryStore?.getThreadById({ threadId: '...' })

Fermeture des connexions

close() libère les connexions des stockages à partir desquels le composite a été construit : les stockages default et editor, ainsi que tout domaine possédant son propre client. Chaque stockage est fermé une seule fois, même s'il prend en charge plusieurs domaines. Lorsque le composite est transmis à la classe Mastra, close() est appelé par shutdown() :

src/mastra/index.ts

import { MastraCompositeStore } from '@mastra/core/storage'
import { PostgresStore } from '@mastra/pg'
import { Mastra } from '@mastra/core'

const pgStore = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
})

export const mastra = new Mastra({
  storage: new MastraCompositeStore({ id: 'composite', default: pgStore }),
})

process.on('SIGTERM', async () => {
  // Releases the Postgres pool, so the process can exit
  await mastra.shutdown()
})

Un stockage que vous construisez uniquement pour fournir un domaine n'est pas accessible via le composite. Conservez une référence vers celui-ci et fermez-le vous-même :

src/mastra/index.ts

import { MastraCompositeStore } from '@mastra/core/storage'
import { ClickhouseStore } from '@mastra/clickhouse'
import { PostgresStore } from '@mastra/pg'
import { Mastra } from '@mastra/core'

const pgStore = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
})

const clickhouseStore = new ClickhouseStore({
  id: 'clickhouse-storage',
  url: process.env.CLICKHOUSE_URL,
  username: process.env.CLICKHOUSE_USERNAME,
  password: process.env.CLICKHOUSE_PASSWORD,
})

export const mastra = new Mastra({
  storage: new MastraCompositeStore({
    id: 'composite',
    default: pgStore,
    domains: { observability: clickhouseStore.stores?.observability },
  }),
})

process.on('SIGTERM', async () => {
  await mastra.shutdown()
  await clickhouseStore.close()
})

Cas d'utilisation

Bases de données distinctes selon les charges de travail

Utilisez une base de données locale pour le développement tout en conservant les données de production dans un service géré :

import { MastraCompositeStore } from '@mastra/core/storage'
import { MemoryPG, WorkflowsPG, ScoresPG } from '@mastra/pg'
import { MemoryLibSQL } from '@mastra/libsql'

const storage = new MastraCompositeStore({
  id: 'composite',
  domains: {
    // Use local SQLite for development, PostgreSQL for production
    memory:
      process.env.NODE_ENV === 'development'
        ? new MemoryLibSQL({ url: 'file:./dev.db' })
        : new MemoryPG({ connectionString: process.env.DATABASE_URL }),
    workflows: new WorkflowsPG({ connectionString: process.env.DATABASE_URL }),
    scores: new ScoresPG({ connectionString: process.env.DATABASE_URL }),
  },
})

Stockage spécialisé pour l'Observability

Les données d'Observability peuvent rapidement submerger les bases de données généralistes en production. Une seule interaction d'Agent peut générer des centaines de spans, et les applications à fort trafic peuvent produire des milliers de Traces par jour.

ClickHouse est recommandé pour l'Observability en production, car il est optimisé pour les charges analytiques à volume élevé et à nombreuses écritures. Utilisez le stockage composite pour acheminer l'Observability vers ClickHouse tout en conservant les autres données dans votre base principale :

import { MastraCompositeStore } from '@mastra/core/storage'
import { MemoryPG, WorkflowsPG, ScoresPG } from '@mastra/pg'
import { ObservabilityStorageClickhouseVNext } from '@mastra/clickhouse'

const storage = new MastraCompositeStore({
  id: 'composite',
  domains: {
    memory: new MemoryPG({ connectionString: process.env.DATABASE_URL }),
    workflows: new WorkflowsPG({ connectionString: process.env.DATABASE_URL }),
    scores: new ScoresPG({ connectionString: process.env.DATABASE_URL }),
    observability: new ObservabilityStorageClickhouseVNext({
      url: process.env.CLICKHOUSE_URL,
      username: process.env.CLICKHOUSE_USERNAME,
      password: process.env.CLICKHOUSE_PASSWORD,
    }),
  },
})

remarque

ObservabilityStorageClickhouseVNext est l'implémentation actuelle du domaine d'Observability. L'ancienne classe ObservabilityStorageClickhouse est également exportée et reste prise en charge pour les projets qui n'ont pas migré. Consultez la référence du stockage ClickHouse pour plus de détails.

ClickHouse répliqué pour les clusters à plusieurs réplicas

Pour les clusters ClickHouse autogérés comportant plusieurs réplicas, définissez replication afin que Mastra produise des moteurs ReplicatedMergeTree et applique ON CLUSTER à son DDL :

import { MastraCompositeStore } from '@mastra/core/storage'
import { MemoryPG, WorkflowsPG, ScoresPG } from '@mastra/pg'
import { ObservabilityStorageClickhouseVNext } from '@mastra/clickhouse'

const storage = new MastraCompositeStore({
  id: 'composite',
  domains: {
    memory: new MemoryPG({ connectionString: process.env.DATABASE_URL }),
    workflows: new WorkflowsPG({ connectionString: process.env.DATABASE_URL }),
    scores: new ScoresPG({ connectionString: process.env.DATABASE_URL }),
    observability: new ObservabilityStorageClickhouseVNext({
      url: process.env.CLICKHOUSE_URL,
      username: process.env.CLICKHOUSE_USERNAME,
      password: process.env.CLICKHOUSE_PASSWORD,
      replication: {
        cluster: 'production_cluster',
        // Optional (defaults shown):
        // zookeeperPath: '/clickhouse/tables/{shard}/{database}/{table}',
        // replicaName: '{replica}',
      },
    }),
  },
})

Ne définissez pas replication sur ClickHouse Cloud. Cloud réécrit MergeTree en SharedMergeTree côté serveur. Consultez la référence du stockage ClickHouse pour découvrir la forme complète de la configuration et les remarques destinées aux opérateurs.

info

Cette approche est également nécessaire lorsque vous utilisez des Providers de stockage qui ne prennent pas en charge l'Observability, tels que Convex, DynamoDB ou Cloudflare. Consultez la documentation de MastraStorageExporter pour obtenir la liste complète des Providers pris en charge.