Mastra DynamoDB 存储适配器深度解析单表设计、TTL 过期、域存储架构与分页 API 演进【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra本文围绕 Mastra 官方mastra/dynamodb存储适配器的核心能力与关键演进展开涵盖单表Single-Table ElectroDB 设计、可配置 TTL 数据过期、基于getStore()的域存储架构、listMessages分页 API 迁移、并发安全与评分持久化等主题并结合仓库源码与配置说明帮助读者掌握如何将 Mastra 的 Agent、Memory、Workflow 数据可靠地落到 Amazon DynamoDB 上。读完本文你将能够独立完成 DynamoDB 表的创建、TTL 策略配置、存储实例接入 Memory 全链路并理解该适配器历次版本迭代背后的设计决策。一、适配器定位与整体设计mastra/dynamodb是 Mastra 官方的 DynamoDB 存储适配器位于仓库 stores/dynamodb 目录。其核心定位在 README.md 中一句话即可概括用「单表设计 ElectroDB」把 Mastra 的全部持久化数据放进 Amazon DynamoDB同时支持可配置索引、TTL 过期、AWS 凭证、以及表初始化校验。从 package.json 可以看到该包的关键技术栈aws-sdk/client-dynamodb/aws-sdk/lib-dynamodb当前要求^3.1095.0AWS SDK v3 的底层客户端与 Document Clientelectrodb当前要求^3.9.1类型安全的 DynamoDB 数据访问层负责把 TypeScript 实体模型映射为 DynamoDB 的pk/sk键与索引查询mastra/core作为 peerDependency要求1.61.0-0 2.0.0-0运行环境要求 Node.js22.13.0。适配器的历史起点记录在 CHANGELOG.md 的 0.10.0 版本它引入了一个「使用 ElectroDB 的单表设计存储连接器」能力清单包括单表存储、类型安全访问、AWS 凭证/区域/端点支持、兼容 DynamoDB Local 本地开发以及 Thread、Message、Trace、Eval、Workflow 等实体操作特别适合 serverless 环境。二、安装与基础使用安装命令来自 README.mdnpm install mastra/dynamodb基础用法是把它作为Memory的存储后端再搭配一个向量库如 Pinecone实现语义召回import { Memory } from mastra/memory; import { DynamoDBStore } from mastra/dynamodb; import { PineconeVector } from mastra/pinecone; // 初始化 DynamoDB 存储 const storage new DynamoDBStore({ name: dynamodb, config: { region: us-east-1, tableName: mastra-single-table, // 你的 DynamoDB 表名 }, }); // 初始化向量库如需语义召回 const vector new PineconeVector({ id: dynamodb-pinecone, apiKey: process.env.PINECONE_API_KEY, }); // Memory 由存储 可选向量库组合而成 const memory new Memory({ storage, vector, options: { lastMessages: 10, semanticRecall: true, }, });从 src/storage/index.ts 的源码可以看出DynamoDBStore在构造时会校验config.tableName非空、字符合法、长度 3–255并支持两种配置形态传入预配置的DynamoDBDocumentClient适合需要自定义中间件、重试策略的场景传入 AWS 配置region/endpoint/credentials由适配器内部创建客户端region默认us-east-1。此外init()通过DescribeTableCommand校验外部托管的表是否可访问ResourceNotFoundException时判定表不存在且初始化 Promise 会被缓存避免并发调用重复校验失败时会重置以便重试。close()负责销毁底层客户端。三、单表结构与二级索引GSI搭建mastra/dynamodb采用单表设计所有实体共用同一张表靠pk/sk前缀区分。具体建表结构记录在 TABLE_SETUP.md主键分区键pkString、排序键skStringGSI1索引名gsi1分区键gsi1pk、排序键gsi1sk投影ALL。用于常见查找模式threadEntity按resourceId查询byResourcemessageEntity按threadId查询byThreadtraceEntity按name查询byNameevalEntity按agent_name查询byAgent。GSI2索引名gsi2分区键gsi2pk、排序键gsi2sk投影ALL。用于traceEntity按scope查询byScopeworkflowSnapshotEntity按run_id查询byRunId。仓库给出了三种建表方式CloudFormation 模板要点PAY_PER_REQUEST计费、PITR 与 SSE 开启Resources: MastraSingleTable: Type: AWS::DynamoDB::Table Properties: TableName: mastra-single-table BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: pk AttributeType: S - AttributeName: sk AttributeType: S - AttributeName: gsi1pk AttributeType: S - AttributeName: gsi1sk AttributeType: S - AttributeName: gsi2pk AttributeType: S - AttributeName: gsi2sk AttributeType: S KeySchema: - AttributeName: pk KeyType: HASH - AttributeName: sk KeyType: RANGE GlobalSecondaryIndexes: - IndexName: gsi1 KeySchema: - AttributeName: gsi1pk KeyType: HASH - AttributeName: gsi1sk KeyType: RANGE Projection: ProjectionType: ALL - IndexName: gsi2 KeySchema: - AttributeName: gsi2pk KeyType: HASH - AttributeName: gsi2sk KeyType: RANGE Projection: ProjectionType: ALL PointInTimeRecoverySpecification: PointInTimeRecoveryEnabled: true SSESpecification: SSEEnabled: trueAWS CDK 方式注意pk/sk字符串类型并分别addGlobalSecondaryIndex添加gsi1与gsi2。CDK 的projectionType默认为ALL适合灵活查询但也有相应成本生产环境可按需收敛投影列。本地开发可直接运行官方镜像docker run -p 8000:8000 amazon/dynamodb-local随后在配置中指定endpoint: http://localhost:8000指向本地实例其余代码无需改动。四、TTL 自动过期按实体配置数据生命周期DynamoDB 的 TTL 机制是控制数据保留成本的关键手段。CHANGELOG 的 1.0.0 版本为DynamoDBStore增加了可配置 TTL支持支持 7 类实体thread、message、resource、trace、eval、workflow_snapshot、score。示例配置如下import { DynamoDBStore, type DynamoDBStoreConfig } from mastra/dynamodb; const config: DynamoDBStoreConfig { id: my-store, tableName: mastra-table, region: us-east-1, ttl: { message: { enabled: true, defaultTtlSeconds: 86400 }, // 1 天 trace: { enabled: true, defaultTtlSeconds: 604800 }, // 7 天 workflow_snapshot: { enabled: true, defaultTtlSeconds: 2592000 }, // 30 天 }, }; const store new DynamoDBStore({ name: dynamodb, config });从 src/storage/index.ts 中的类型定义可以看到 TTL 配置的完整字段enabled是否对该实体启用 TTLattributeName写入 TTL 的属性名必须与 DynamoDB 表上开启的 TTL 属性名一致默认ttldefaultTtlSeconds从创建/更新时间起算的过期秒数例如30 * 24 * 60 * 60表示 30 天。关键前置条件必须在表上开启TABLE_SETUP.md 给出了三种方式# AWS CLI对已存在的表启用 TTL aws dynamodb update-time-to-live \ --table-name mastra-single-table \ --time-to-live-specification Enabledtrue, AttributeNamettlCloudFormation 中加TimeToLiveSpecification: { AttributeName: ttl, Enabled: true }CDK 中设置timeToLiveAttribute: ttl。需要特别注意的是源码注释与文档均强调DynamoDB TTL 是后台进程过期项最长会在过期后 48 小时内才被实际删除到期之前数据仍可被查询到。因此 TTL 适合做数据最终清理不能当作精确的硬删除语义。仓库中对应实现集中在 src/storage/ttl.ts并导出calculateTtl、getTtlAttributeName、isTtlEnabled、getTtlProps等工具函数另有 ttl.test.ts 覆盖其行为。五、域存储架构getStore()与可组合存储Mastra 在 1.0.0 系列版本中对存储架构做了两次大的重构这两次重构直接影响了DynamoDBStore的公开 API引入StorageDomain基类PR #11249每个域memory、workflows、scores、observability、agents都继承StorageDomain实现init()与dangerouslyClearAll()域既可以接受外部数据库客户端也可以根据配置内部创建。改为getStore()模式PR #11361移除MastraStorage基类上的透传方法改为按域取存储// Before const thread await storage.getThreadById({ threadId }); await storage.persistWorkflowSnapshot({ workflowName, runId, snapshot }); await storage.createSpan(span); // After const memory await storage.getStore(memory); const thread await memory?.getThreadById({ threadId }); const workflows await storage.getStore(workflows); await workflows?.persistWorkflowSnapshot({ workflowName, runId, snapshot }); const observability await storage.getStore(observability); await observability?.createSpan(span);在 src/storage/index.ts 的构造函数中可以看到DynamoDBStore内部初始化了 4 个域存储workflows、memory、scores、backgroundTasks并向外导出了BackgroundTasksStorageDynamoDB、MemoryStorageDynamoDB、ScoresStorageDynamoDB、WorkflowStorageDynamoDB四个域类供MastraCompositeStore组合使用。对应实现位于 src/storage/domains 目录。与此同时MastraStorage更名为MastraCompositeStore旧名保留为废弃别名支持把不同数据库适配器按域组合在一起例如 PostgreSQL 负责 memory/workflows、LibSQL 负责其他域。这也解释了为什么新代码推荐用storage.getStore(memory)来探测某个域是否可用。六、消息分页 API 演进从getMessages到listMessagesCHANGELOG 中最重要的破坏性变更集中在 1.0.0Mastra 统一了存储层的读取 APIDynamoDBStore同步跟进。6.1 移除getMessages()统一为listMessages()// Before const messages await storage.getMessages({ threadId: thread-1 }); // After const result await storage.listMessages({ threadId: thread-1, page: 0, perPage: 50, }); const messages result.messages; // 消息数组 console.log(result.total); // 总数 console.log(result.hasMore); // 是否还有下一页listMessages()默认按createdAt升序最旧在前要改为最新在前const result await storage.listMessages({ threadId: thread-1, orderBy: { field: createdAt, direction: DESC }, });客户端 SDK 侧对应改名为client.listThreadMessages()原getThreadMessages类型StorageGetMessagesArg也由StorageListMessagesInput取代。6.2 分页参数统一为page/perPage1.0.0 还把所有分页 API 从offset/limit迁移到page0 起始/perPage// Before await memory.listThreadsByResourceId({ resourceId: user-123, offset: 20, limit: 10 }); // After await memory.listThreadsByResourceId({ resourceId: user-123, page: 2, perPage: 10 });同样适用于memory.listMessages()与storage.listWorkflowRuns()。迁移时page Math.floor(offset / limit)。同时所有存储实现都新增了负数page、非法perPage负数、0、false的校验。6.3perPage: false一次取回全部移除废弃的getMessagesPaginated()后listMessages()与评分查询listScoresBySpan、listScoresByRunId、listScoresByExecutionId支持perPage: false表示不分页全量拉取HTTP 查询参数也接受字符串?perPagefalse。此外listMessages()的threadId校验变得更严格空字符串或纯空白会直接抛错而非返回空结果。6.4 游标式分页startExclusive/endExclusive为了支持聊天应用在持续写入场景下的无缝隙分页filter.dateRange增加了startExclusive与endExclusive布尔选项PR #11479。用法以本页最后一条消息的时间戳作为游标配合endExclusive: true拉取下一页// 第一页 const page1 await memory.recall({ threadId: thread-123, perPage: 10, orderBy: { field: createdAt, direction: DESC }, }); // 基于游标的下一页 const oldestMessage page1.messages[page1.messages.length - 1]; const page2 await memory.recall({ threadId: thread-123, perPage: 10, orderBy: { field: createdAt, direction: DESC }, filter: { dateRange: { end: oldestMessage.createdAt, endExclusive: true, // 排除游标消息本身 }, }, });相比 offset 分页游标方式在会话期间有新消息写入时不会出现跳页或重复。6.5 DynamoDB 上的分页正确性修复DynamoDB 的Query/Scan天然按 1MB 分页因此适配器在 1.1.3 版本集中修复了一批listMessages的分页问题这在选择该存储时需要特别留意对应测试位于 src/storage/domains/memory/list-messages.test.ts大线程不再被静默截断total与hasMore元数据正确include上下文如跳到线程深处的某条消息并获取前后文能跨 DynamoDB 物理页正确取回当includewithNextMessages/withPreviousMessages已经覆盖全部消息时hasMore正确置为false当include上下文落在resourceId/dateRange过滤范围之外时不再导致hasMore提前为false只有匹配过滤条件的消息才计入返回总数。七、查询能力增强多线程、元数据过滤与线程列表7.1 多线程查询listMessages的threadId支持字符串数组string | string[]可一次跨多个线程查询消息内部_getIncludedMessages改为按消息 ID 反查所属线程消息 ID 全局唯一。这对聚合多个会话上下文类场景很有用。7.2 精确元数据过滤1.2.0 引入了对消息历史的精确元数据过滤支持字符串、有限数字、布尔值与null多字段之间为 AND 语义const messages await memory.recall({ threadId: thread-1, filter: { metadata: { status: done, priority: high, }, }, });7.3listThreads灵活的线程筛选新增的listThreads可按resourceId、metadata或两者组合过滤metadata 键值对为 AND 逻辑全部过滤参数可选支持分页与排序// 列出全部线程 const allThreads await memory.listThreads({}); // 按 resourceId 过滤 const userThreads await memory.listThreads({ filter: { resourceId: user-123 }, }); // 组合过滤 分页 排序 const filteredThreads await memory.listThreads({ filter: { resourceId: user-123, metadata: { priority: high, status: open }, }, orderBy: { field: updatedAt, direction: DESC }, page: 0, perPage: 20, });此外 1.0.7 起getThreadById尊重可选的resourceId——当线程不属于该资源时返回null避免资源越界读取const thread await memory.getThreadById({ threadId: my-thread-id, resourceId: my-user-id, }); // 若该线程不属于 my-user-id返回 null八、错误语义、并发安全与数据完整性8.1 读取错误不再被静默吞掉1.3.0 是一个重要的行为变更listThreads、listMessages、listMessagesByResourceId、listMessagesById这些分页读操作此前在数据库故障时如表被锁、连接断开会捕获错误、打日志并返回空结果{ threads: [], total: 0, hasMore: false }导致 Agent 在短暂故障时把读取失败误判为没有历史而覆盖真实状态。现在这些方法会把故障以MastraError重新抛出校验类USER错误与真实的空结果保持原样。直接调用这些读取方法而非经由 Agent时建议包一层 try/catchtry { const { threads } await storage.listThreads({ resourceId }); // ...使用 threads } catch (error) { // 这是真实的存储故障决定重试、上报或降级 // 空线程列表不再藏在这里它只代表没有线程 }同时 1.0.0 之后所有存储的错误 ID 统一为MASTRA_STORAGE_{STORE}_{OPERATION}_{STATUS}模式createStorageErrorIdDynamoDB 侧可从构造、初始化、表校验、关闭等路径的错误详情中定位问题。8.2 线程标题不再被覆盖1.3.0 修复了线程标题被覆盖的竞态updateThread之前要求同时传入title与metadata导致只改 metadata 的调用方必须先把线程读回来再回写 title当标题生成完成正好发生在读与写之间时新生成的标题会被旧值覆盖。现在title与metadata互相独立可选省略哪个字段就保持哪列不动。存储适配器同时声明了支持部分线程更新的能力让新版mastra/memory能保留已有标题并对旧版存储保持向后兼容混合版本部署依然可用。DynamoDB 侧还修复了空标题被写成Thread id占位符而阻止标题自动生成的问题1.0.5对应 issue #15998。8.3 并发 resume 的原子声明409 语义针对同一挂起工作流被并发恢复导致下游步骤重复执行的问题1.3.1 引入了原子化修复resume 在真正执行前会先原子地认领该 run只有一方能够继续失败的调用方抛出WORKFLOW_RESUME_ALREADY_CLAIMED且不会执行任何步骤。工作流状态更新新增可选的expectedStatus守卫只有存储中的 run 处于预期状态时才应用状态变更HTTP 层的 resume 冲突返回409 Conflict。这对 DynamoDB 存储意味着其工作流域必须提供足够强的原子更新语义来支撑该特性。8.4 后台任务的原子状态更新1.3.2 要求后台任务状态更新使用原子条件写compare-and-set确保 dispatch 过程中的取消操作不会被覆盖由于 Cloudflare KV 与 ClickHouse 无法提供 CAS 语义后台任务存储不再由这两者暴露而 DynamoDB 侧由BackgroundTasksStorageDynamoDB位于 src/storage/domains/background-tasks实现。1.0.7 还开始跟踪后台任务的suspendedAt与suspendPayload字段。8.5 版本协同与 peerDependency 修复1.2.2 修复了一个容易被忽略的安装期问题9 个存储适配器声明的mastra/corepeer 范围过低实际却导入了mastra/core/storage的storageMessageMatchesMetadataFiltercore 1.53.0 才导出导致安装成功、首次 import 时报错SyntaxError: The requested module mastra/core/storage does not provide an export named storageMessageMatchesMetadataFilter修复后这些适配器统一声明1.53.0-0 2.0.0-0让 npm/pnpm 在安装阶段就暴露冲突而不是让项目在首次导入时才崩溃。当前 package.json 中该包要求的 core 范围为1.61.0-0 2.0.0-0。九、评分持久化批次溯源与多租户隔离CHANGELOG 1.1.2 为评分Scores域增加了两类能力批次溯源字段——saveScore/scoreTrace支持顶层batchId、datasetId、datasetItemId便于把一次基线评分归组并回连到对应的数据集条目await scoreTrace({ storage, scorer, target: { traceId }, batchId: baseline-batch-1, datasetId, datasetItemId, });多租户字段——organizationId与projectId用于租户隔离listScoresBy*系列方法接受filters按组织与项目过滤await storage.saveScore({ ...score, organizationId: org-a, projectId: proj-1 }); const result await storage.listScoresByScorerId({ scorerId, filters: { organizationId: org-a, projectId: proj-1 }, });注意语义区别projectId标识项目作用域resourceId继续表示 Agent 记忆资源。早期版本还修复了 DynamoDB 下内置评分器如 hallucination-scorer分数静默丢失1.0.0-beta.10以及saveScore未持久化 ID 导致getScoreById取不到1.0.0-beta.4等问题。DynamoDB 评分域实现位于 src/storage/domains/scores。十、运维建议与进一步阅读表结构先行由于DynamoDBStore.init()只校验表是否存在DescribeTableCommand并不自动建表务必先用 CloudFormation/CDK 按 TABLE_SETUP.md 建好pk/sk与两个 GSI否则初始化会抛出表不存在或不可访问。TTL 两处配置代码里config.ttl与 DynamoDB 表上的 TTL 属性名必须一致默认ttl且注意 48 小时删除延迟。版本匹配使用前确认mastra/core版本在 peer 范围内避免安装通过但导入崩溃。升级关注破坏性变更从旧版本升级时重点检查getMessages/getMessagesPaginated/getThreadsByResourceId等旧 API 是否已迁移到listMessages/listMessagesById/listThreadsByResourceId以及offset/limit是否已改为page/perPage。本地联调可用amazon/dynamodb-local容器配合endpoint参数快速验证无需真实 AWS 资源。仓库内可继续深入阅读的实现与测试文件存储主类与 TTL 配置类型src/storage/index.ts表结构文档TABLE_SETUP.mdTTL 实现与测试src/storage/ttl.ts、src/storage/ttl.test.ts域存储目录src/storage/domainsmemory/workflows/scores/background-tasks 四个域消息分页测试src/storage/domains/memory/list-messages.test.ts实体定义src/entitiesthread、message、resource、trace、eval、score、workflow-snapshot、background-task完整版本历史CHANGELOG.md综上mastra/dynamodb通过单表 两级 GSI 支撑了 Memory、Workflow、Scores 与后台任务等全部核心域TTL 与disableInit让它可以适配从开发到生产的完整生命周期getStore()域架构与统一的listMessages分页语义则让它在多存储适配器之间保持一致的编程模型。理解这些设计后你就能更自信地把 Mastra 的 AI 应用持久层放到 DynamoDB 上。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考