Storage

Storage API 已标准化，所有方法均使用一致的分页和命名模式。

数据库迁移

请通过常规迁移流程运行这些 SQL 迁移，例如 Prisma Migrate、Drizzle Kit 或 DBA 审核流程。

重命名 scorer 表列

mastra_scorers 中的 runtimeContext 列已重命名为 requestContext。

哪些用户需要此迁移

仅当你将 @mastra/pg 或 @mastra/libsql 用于 Evals/评分，并且 runtimeContext 列中已有数据时才需要。

不迁移会出现什么问题

现有评分记录中的请求上下文数据将无法访问。

部署 v1（初始化时会添加新的 requestContext 列）后，请复制数据并删除旧列：

UPDATE mastra_scorers SET "requestContext" = "runtimeContext" WHERE "runtimeContext" IS NOT NULL;
ALTER TABLE mastra_scorers DROP COLUMN "runtimeContext";

重复 span 迁移

如果从旧版 Mastra 升级，mastra_spans 表中可能存在重复的 (traceId, spanId) 条目。v1 为这些列添加了唯一约束以确保数据完整性，但存在重复项时无法添加该约束。

哪些用户需要此迁移

仅当你拥有 Mastra v1 之前版本的现有 span 数据，并遇到重复键冲突或约束创建失败错误时才需要。

不迁移会出现什么问题

Storage 初始化可能在尝试添加唯一约束时失败，或者可能出现“duplicate key value violates unique constraint”等错误。

方案 1：使用 CLI（推荐）

运行迁移命令，自动对 span 去重并添加约束：

npx mastra migrate

CLI 会打包项目并连接到已配置的 Storage，然后运行迁移。移除重复项时，它会保留最完整的记录（根据 endTime 和属性判断）。

方案 2：手动执行 SQL（PostgreSQL）

如果希望手动运行迁移：

-- Remove duplicates, keeping the most complete record
DELETE FROM mastra_spans a USING mastra_spans b
WHERE a.ctid < b.ctid
  AND a."traceId" = b."traceId"
  AND a."spanId" = b."spanId";

-- Add the unique constraint
ALTER TABLE mastra_spans ADD CONSTRAINT mastra_spans_trace_span_unique UNIQUE ("traceId", "spanId");

方案 3：手动迁移（其他数据库）

对于 ClickHouse、LibSQL、MongoDB 或 MSSQL，请使用编程式 API：

const storage = mastra.getStorage()
const observabilityStore = await storage.getStore('observability')

// Check if migration is needed
const status = await observabilityStore?.checkSpansMigrationStatus()
console.log(status)

// Run the migration
const result = await observabilityStore?.migrateSpans()
console.log(result)

JSON 列（TEXT → JSONB）

仅适用于 PostgreSQL。 mastra_threads 中的 metadata 列和 mastra_workflow_snapshot 中的 snapshot 列已从 TEXT 改为 JSONB。

推荐

迁移到 JSONB 后可使用 PostgreSQL 原生 JSON 运算符和 GIN 索引，提高 JSON 字段的查询性能。

ALTER TABLE mastra_threads
ALTER COLUMN metadata TYPE jsonb
USING metadata::jsonb;

ALTER TABLE mastra_workflow_snapshot
ALTER COLUMN snapshot TYPE jsonb
USING snapshot::jsonb;

新增

MastraCompositeStore 中的 Storage 组合

MastraCompositeStore 现在可以组合来自不同适配器的 Storage 领域。需要针对不同用途使用不同数据库时可使用它，例如用 PostgreSQL 存储 Memory 和 Workflow，同时为 observability 使用专用数据库。

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

// Compose domains from different stores
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 }),
    },
  }),
})

更多详情请参阅 Storage 组合参考。

已变更

MastraStorage 重命名为 MastraCompositeStore

MastraStorage 类已重命名为 MastraCompositeStore，以更准确地体现其作为组合 Storage 实现的作用：将不同领域路由到不同的底层 Store。这样可以避免与通用的“Mastra Storage”概念（Mastra 实例上的 storage 属性）混淆。

为保持向后兼容，旧名称 MastraStorage 仍作为已弃用别名提供，但会在未来版本中移除。

迁移时，请更新导入和实例化代码：

- import { MastraStorage } from "@mastra/core/storage";
+ import { MastraCompositeStore } from "@mastra/core/storage";
  import { MemoryLibSQL } from "@mastra/libsql";
  import { WorkflowsPG } from "@mastra/pg";

  export const mastra = new Mastra({
-   storage: new MastraStorage({
+   storage: new MastraCompositeStore({
      id: "composite",
      domains: {
        memory: new MemoryLibSQL({ url: "file:./memory.db" }),
        workflows: new WorkflowsPG({ connectionString: process.env.DATABASE_URL }),
      },
    }),
  });

备注

如果直接使用单一 Store 实现（例如 PostgresStore 或 LibSQLStore），请保留现有配置。此变更仅影响显式使用 MastraStorage 进行组合存储的代码。

Storage 实例必须提供 id 属性

Storage 实例现在必须提供 id 属性。此唯一标识符用于在 Mastra 中跟踪和管理 Storage 实例。应用中的每个 Storage 实例都应使用描述清晰且唯一的 id 字符串。

迁移时，请在 Storage 构造函数中添加 id 字段。

  const storage = new PostgresStore({
+   id: 'main-postgres-store',
    connectionString: process.env.POSTGRES_CONNECTION_STRING,
    schemaName: 'public',
  });

  const upstashStore = new UpstashStore({
+   id: 'upstash-cache-store',
    url: process.env.UPSTASH_REDIS_REST_URL,
    token: process.env.UPSTASH_REDIS_REST_TOKEN,
  });

分页从 offset/limit 改为 page/perPage

所有分页 API 现在都使用 page 和 perPage，而不再使用 offset 和 limit，以匹配基于页的 Web 分页。

迁移时，请将所有分页参数从 offset/limit 更新为 page/perPage。请注意，page 从 0 开始计数。

  memoryStore.listMessages({
    threadId: 'thread-123',
-   offset: 0,
-   limit: 20,
+   page: 0,
+   perPage: 20,
  });

getMessagesPaginated 改为 listMessages

getMessagesPaginated() 方法已由 listMessages() 替代。新方法支持使用 perPage: false 获取全部记录而不进行分页。此变更与 list* 命名约定保持一致，并提高了获取全部记录时的灵活性。

迁移时，请重命名方法并更新分页参数。现在可以使用 perPage: false 获取全部记录。

+ const memoryStore = await storage.getStore('memory');
+
  // Paginated
- const result = await storage.getMessagesPaginated({
+ const result = await memoryStore?.listMessages({
    threadId: 'thread-123',
-   offset: 0,
-   limit: 20,
+   page: 0,
+   perPage: 20,
  });

  // Fetch all records (no pagination limit)
+ const allMessages = await memoryStore?.listMessages({
+   threadId: 'thread-123',
+   page: 0,
+   perPage: false,
+ });

Codemod

你可以使用 Mastra 的 codemod CLI 自动更新代码：

npx @mastra/codemod@latest v1/storage-get-messages-paginated .

通过 getStore() 访问特定领域的 Storage

Storage 操作现在通过特定领域的 Store 访问，而不再直接在 Storage 实例上访问。

领域包括：

• memory — 线程、消息和资源
• workflows — Workflow 快照
• scores — 评估分数
• observability — Trace 和 span
• agents — 存储的 Agent 数据

迁移时，请使用领域名称调用 getStore()，然后在返回的 Store 上调用方法。

  const storage = mastra.getStorage();

  // Memory operations (threads, messages, resources)
- const thread = await storage.getThread({ threadId: '123' });
- await storage.saveThread({ thread });
+ const memoryStore = await storage.getStore('memory');
+ const thread = await memoryStore?.getThreadById({ threadId: '123' });
+ await memoryStore?.saveThread({ thread });

  // Workflow operations (snapshots)
- const snapshot = await storage.loadWorkflowSnapshot({ runId, workflowName });
- await storage.persistWorkflowSnapshot({ runId, workflowName, snapshot });
+ const workflowStore = await storage.getStore('workflows');
+ const snapshot = await workflowStore?.loadWorkflowSnapshot({ runId, workflowName });
+ await workflowStore?.persistWorkflowSnapshot({ runId, workflowName, snapshot });

  // Observability operations (traces, spans)
- const traces = await storage.listTraces({ page: 0, perPage: 20 });
+ const observabilityStore = await storage.getStore('observability');
+ const traces = await observabilityStore?.listTraces({ page: 0, perPage: 20 });

  // Score operations (evaluations)
- const scores = await storage.listScoresByScorerId({ scorerId: 'helpfulness' });
+ const scoresStore = await storage.getStore('scores');
+ const scores = await scoresStore?.listScoresByScorerId({ scorerId: 'helpfulness' });

getThreadsByResourceId 改为 listThreads

getThreadsByResourceId() 方法已由 listThreads() 替代。新方法增加了分页支持，并支持按 resourceId、metadata 或两者组合进行筛选。

important

旧的 getThreadsByResourceId() 会返回所有匹配的线程而不进行分页。新的 listThreads() 需要分页参数。若要保留获取全部线程的旧行为，请使用 perPage: false。

迁移时，请使用 Memory Store 和新的 listThreads() 方法，并提供分页参数和可选的筛选对象。

- const threads = await storage.getThreadsByResourceId({
-   resourceId: 'res-123',
- });
+ const memoryStore = await storage.getStore('memory');
+
+ // Paginated (recommended for large datasets)
+ const result = await memoryStore?.listThreads({
+   filter: { resourceId: 'res-123' },
+   page: 0,
+   perPage: 20,
+ });
+ const threads = result?.threads;
+
+ // Or fetch all threads like before (use perPage: false)
+ const allResult = await memoryStore?.listThreads({
+   filter: { resourceId: 'res-123' },
+   perPage: false,
+ });
+ const allThreads = allResult?.threads;

新方法还支持：

• 列出所有线程（省略 filter）
• 仅按 metadata 筛选
• 组合使用 resourceId 和 metadata 筛选条件

// List all threads
await memoryStore?.listThreads({ page: 0, perPage: 20 })

// Filter by metadata only
await memoryStore?.listThreads({
  filter: { metadata: { status: 'active' } },
  page: 0,
  perPage: 20,
})

// Combined filter
await memoryStore?.listThreads({
  filter: {
    resourceId: 'user-123',
    metadata: { category: 'support' },
  },
  page: 0,
  perPage: 20,
})

Codemod

你可以使用 Mastra 的 codemod CLI 自动更新代码：

npx @mastra/codemod@latest v1/storage-list-threads-by-resource-to-list-threads .

getWorkflowRuns 改为 listWorkflowRuns

getWorkflowRuns() 方法已重命名为 listWorkflowRuns()。此变更遵循 list* 方法返回集合的约定。

迁移时，请使用 Workflow Store，重命名方法调用并更新分页参数。

- const runs = await storage.getWorkflowRuns({
+ const workflowStore = await storage.getStore('workflows');
+ const runs = await workflowStore?.listWorkflowRuns({
    fromDate,
    toDate,
+   page: 0,
+   perPage: 20,
  });

Codemod

你可以使用 Mastra 的 codemod CLI 自动更新代码：

npx @mastra/codemod@latest v1/storage-list-workflow-runs .

getMessagesById 改为 listMessagesById

getMessagesById() 方法已重命名为 listMessagesById()。此变更遵循 list* 方法返回集合的约定。

迁移时，请使用 Memory Store 并重命名方法调用。

+ const memoryStore = await storage.getStore('memory');
- const result = await storage.getMessagesById({
+ const result = await memoryStore?.listMessagesById({
    messageIds: ['msg-1', 'msg-2'],
  });

Codemod

你可以使用 Mastra 的 codemod CLI 自动更新代码：

npx @mastra/codemod@latest v1/storage-list-messages-by-id .

Storage getMessages 和 saveMessages 签名

getMessages() 和 saveMessages() 方法的签名及返回类型已变更。格式重载已移除，这些方法现在始终使用 MastraDBMessage。移除格式变体后，API 得到简化。

迁移时，请使用 Memory Store，移除格式参数，并更新代码以使用统一的返回类型。

+ const memoryStore = await storage.getStore('memory');
+
  // Always returns { messages: MastraDBMessage[] }
- const v1Messages = await storage.getMessages({ threadId, format: 'v1' });
- const v2Messages = await storage.getMessages({ threadId, format: 'v2' });
+ const result = await memoryStore?.getMessages({ threadId });
+ const messages = result?.messages; // MastraDBMessage[]

  // SaveMessages always uses MastraDBMessage
- await storage.saveMessages({ messages: v1Messages, format: 'v1' });
- await storage.saveMessages({ messages: v2Messages, format: 'v2' });
+ const saveResult = await memoryStore?.saveMessages({ messages: mastraDBMessages });
+ const saved = saveResult?.messages; // MastraDBMessage[]

Vector Store API 从位置参数改为命名参数

所有 Vector Store 方法现在都使用参数对象，而不再使用位置参数。这样可以在调用处明确各值的用途，也让方法签名无需依赖参数顺序即可演进。

迁移时，请更新所有 Vector Store 方法调用以使用参数对象。

- await vectorDB.createIndex(indexName, 3, 'cosine');
+ await vectorDB.createIndex({
+   indexName: indexName,
+   dimension: 3,
+   metric: 'cosine',
+ });

- await vectorDB.upsert(indexName, [[1, 2, 3]], [{ test: 'data' }]);
+ await vectorDB.upsert({
+   indexName: indexName,
+   vectors: [[1, 2, 3]],
+   metadata: [{ test: 'data' }],
+ });

- await vectorDB.query(indexName, [1, 2, 3], 5);
+ await vectorDB.query({
+   indexName: indexName,
+   queryVector: [1, 2, 3],
+   topK: 5,
+ });

Vector Store 方法重命名

updateIndexById 和 deleteIndexById 方法已分别重命名为 updateVector 和 deleteVector。新名称明确表示这些方法操作的是 Vector。

迁移时，请重命名方法并传入参数对象。

- await vectorDB.updateIndexById(indexName, id, update);
- await vectorDB.deleteIndexById(indexName, id);
+ await vectorDB.updateVector({ indexName, id, update });
+ await vectorDB.deleteVector({ indexName, id });

PGVector 构造函数从连接字符串改为对象

PGVector 构造函数现在需要对象参数，而不再接收连接字符串。此变更让所有 Storage 适配器的 API 更加一致。

迁移时，请将连接字符串作为对象属性传入。

- const pgVector = new PgVector(process.env.POSTGRES_CONNECTION_STRING!);
+ const pgVector = new PgVector({
+   connectionString: process.env.POSTGRES_CONNECTION_STRING,
+ });

Codemod

你可以使用 Mastra 的 codemod CLI 自动更新代码：

npx @mastra/codemod@latest v1/vector-pg-constructor .

PGVector defineIndex 改为 buildIndex

defineIndex() 方法已移除，请改用 buildIndex()。新名称明确表示该方法会构建索引。

迁移时，请重命名方法并传入参数对象。

- await vectorDB.defineIndex(indexName, 'cosine', { type: 'flat' });
+ await vectorDB.buildIndex({
+   indexName: indexName,
+   metric: 'cosine',
+   indexConfig: { type: 'flat' },
+ });

PostgresStore：schema 改为 schemaName

PostgresStore 构造函数中的 schema 参数已重命名为 schemaName。新名称明确表示该值是数据库 Schema 的名称。

迁移时，请重命名该参数。

  const pgStore = new PostgresStore({
    connectionString: process.env.POSTGRES_CONNECTION_STRING,
-   schema: customSchema,
+   schemaName: customSchema,
  });

Codemod

你可以使用 Mastra 的 codemod CLI 自动更新代码：

npx @mastra/codemod@latest v1/storage-postgres-schema-name .

评分 Storage 方法改为 listScoresBy* 模式

评分 Storage API 已重命名为遵循 listScoresBy* 模式。此变更与整体 API 命名约定保持一致。

迁移时，请将方法名从 getScores 更新为 listScoresByScorerId 及相关变体。

- const scores = await storage.getScores({ scorerName: 'helpfulness-scorer' });
+ const scores = await storage.listScoresByScorerId({
+   scorerId: 'helpfulness-scorer',
+ });
+ // Also available: listScoresByRunId, listScoresByEntityId, listScoresBySpan

已移除

非分页 Storage 函数

非分页 Storage 函数已移除，请改用分页版本。所有 list 操作现在都使用分页，但可以通过 perPage: false 获取全部记录。此变更确保 API 一致，并防止意外加载大型数据集。

迁移时，请通过领域 Store 使用分页方法。若要获取全部记录，请使用 perPage: false。

- // Non-paginated direct access
- const messages = await storage.getMessages({ threadId });

+ // Use paginated methods via domain stores
+ const memoryStore = await storage.getStore('memory');
+ const result = await memoryStore?.listMessages({ threadId, page: 0, perPage: 20 });
+ // Or fetch all
+ const allMessages = await memoryStore?.listMessages({
+   threadId,
+   page: 0,
+   perPage: false,
+ });

getTraces 和 getTracesPaginated

Storage 中已移除 getTraces() 和 getTracesPaginated() 方法。请使用 observability 包访问 Trace，而不是 core Storage。

迁移时，请改用 observability Storage 方法。

- const traces = await storage.getTraces({ traceId: 'trace-123' });
- const paginated = await storage.getTracesPaginated({ page: 0, perPage: 20 });

+ // Use observability API for traces
+ import { initObservability } from '@mastra/observability';
+ const observability = initObservability({ config: { ... } });
+ // Access traces through observability API

Evals 测试工具函数

@internal/test-utils 中已移除 Evals 领域测试工具函数。此变更反映了旧版 Evals 功能的移除。

迁移时，请直接使用 Storage API 进行测试，不要使用专用的 Evals 测试工具函数。

- import { createEvalsTests } from '@internal/test-utils/domains/evals';
- createEvalsTests({ storage });

+ // Use storage APIs directly for testing

MSSQL Storage 中的 TABLE_EVALS

MSSQL Storage 实现中已移除 TABLE_EVALS 表。此变更反映了旧版 Evals 功能的移除。

如果将 MSSQL Storage 用于 Evals，请迁移到其他 Storage 适配器，或移除 Evals 功能。