複合儲存

MastraCompositeStore 可以組合來自不同 Provider 的儲存域。當你需要為不同用途使用不同資料庫時，可使用此功能。例如，使用 LibSQL 儲存 memory，並使用 PostgreSQL 儲存 Workflow。

安裝

MastraCompositeStore 已包含在 @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

你亦需要安裝要組合的儲存 Provider：

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

儲存域

Mastra 將儲存空間劃分為多個域，每個域處理特定類型的資料。每個域可由不同的儲存 adapter 支援，而各儲存套件均會匯出域 class。

Domain	說明
memory	持久儲存 Agent 對話。儲存 thread（對話 session）、訊息、resource（使用者身分）及 working memory（跨對話持續保留的情境）。
workflows	Workflow 執行狀態。當 Workflow 因等待人工輸入、外部事件或排程恢復而暫停時，其狀態會持久儲存於此，以便伺服器重新啟動後恢復。
scores	Mastra Evals 系統的評估結果。分數及指標會持久儲存於此，以便分析及比較不同時間的結果。
observability	包括 Trace 及 span 的遙測資料。Agent 互動、Tool 調用及 LLM 請求會產生 span，並彙集成 Trace，以供除錯及效能分析。
agents	已儲存 Agent 的 Agent 設定。毋須部署程式碼，亦可在執行階段定義及更新 Agent。
datasets	實驗執行所用的評估 dataset。儲存 dataset 定義、結構描述及版本化項目。
experiments	與 dataset 及 target 連結的實驗執行及逐項實驗結果。

備註

MastraCompositeStore 接受以上所有域 key，但各套件的儲存 adapter 支援程度有所不同。你可以為每個域混合使用 adapter，但只限於該 adapter 已實作並匯出的域。例如，memory: new MemoryLibSQL(...) 及 workflows: new WorkflowsPG(...) 均有效，因為兩個套件都有匯出這些域 class。

使用方法

基本組合

直接從各儲存套件匯入域 class，然後加以組合：

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

配合預設儲存使用

使用 default 指定後備儲存，然後覆寫特定域：

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

混合後端

使用各儲存套件的域 class，將不同域路由至不同後端。以下範例把 memory 及 Workflow 狀態儲存在 MongoDB，並將 observability 路由至 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,
      }),
    },
  }),
})

停用域

將域設為 false 即可停用。已停用的域不會後備使用 default，因此不會持久儲存該域的資料：

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

選項

id:

string

此儲存 instance 的唯一識別碼。

default?:

MastraCompositeStore

預設儲存 adapter。未在 domains 明確指定的域，會使用此儲存的域作為後備。

editor?:

MastraCompositeStore

Editor 所有域的儲存 adapter，包括 Agent、prompt block、scorer、MCP client 及 server、Workspace 與 Skill。其優先次序高於預設儲存，但低於明確的域覆寫。

disableInit?:

boolean

設為 true 時會停用自動初始化。你必須明確調用 init()。

domains?:

object

個別域覆寫。每個域可來自不同的儲存 adapter。其優先次序高於 editor 及 default 儲存。將域設為 false 可完全停用；已停用的域不會後備使用 editor 或 default。

object

memory?:

MemoryStorage

thread、訊息及 resource 的儲存空間。

workflows?:

WorkflowsStorage

Workflow 快照的儲存空間。

scores?:

ScoresStorage

評估分數的儲存空間。

observability?:

ObservabilityStorage

Trace 及 span 的儲存空間。

agents?:

AgentsStorage

已儲存 Agent 設定的儲存空間。

datasets?:

DatasetsStorage

dataset metadata、dataset 項目及 dataset 版本的儲存空間。

experiments?:

ExperimentsStorage

實驗執行及逐項實驗結果的儲存空間。

初始化

MastraCompositeStore 會獨立初始化每個已設定的域。傳入 Mastra class 時，系統會自動調用 init()：

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

如直接使用儲存，請明確調用 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: '...' })

關閉連線

close() 會釋放用來建立複合儲存的各個 store 連線，包括 default 及 editor store，以及任何擁有本身 client 的域。即使同一 store 支援多個域，也只會關閉一次。傳入 Mastra class 時，shutdown() 會調用 close()：

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

如果你建立 store 的唯一用途是提供某個域，便無法透過複合儲存存取該 store。請保留其參照並自行關閉：

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

使用情境

為不同工作負載使用獨立資料庫

開發時使用本機資料庫，同時把生產環境資料保留在受管理服務中：

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

observability 專用儲存

在生產環境中，observability 資料可迅速令通用資料庫不勝負荷。單次 Agent 互動可產生數百個 span，而高流量應用程式每日可產生數千個 Trace。

建議在生產環境使用 ClickHouse 處理 observability，因為它針對大量且寫入密集的分析工作負載作了最佳化。使用複合儲存將 observability 路由至 ClickHouse，同時把其他資料保留在主要資料庫：

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

備註

ObservabilityStorageClickhouseVNext 是目前的 observability 域實作。系統亦會匯出舊版 ObservabilityStorageClickhouse class，並繼續支援尚未遷移的項目。詳情請參閱 ClickHouse 儲存參考。

適用於多副本 cluster 的複寫 ClickHouse

如使用包含多個副本的自行管理 ClickHouse cluster，請設定 replication，讓 Mastra 產生 ReplicatedMergeTree 引擎，並在其 DDL 套用 ON CLUSTER：

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

請勿在 ClickHouse Cloud 設定 replication。Cloud 會在伺服器端將 MergeTree 改寫為 SharedMergeTree。如需完整設定結構及操作注意事項，請參閱 ClickHouse 儲存參考。

資訊

使用不支援 observability 的儲存 Provider（例如 Convex、DynamoDB 或 Cloudflare）時，亦必須採用此方法。完整的受支援 Provider 清單請參閱 MastraStorageExporter 文件。