学习Agent开发8 高级Agent框架Mastra基础用法

作者：袖梨 2026-07-29

什么是Mastra

Mastra是一个由ts编写的Agent开发框架ai-sdk langgraph 是低级的框架因为它们关注的单次toolcall的自动响应以及ai会话本身而Mastra则封装了ai-sdk的能力提供了上下文管理、会话持久化、rag、skill、长期记忆等能力

该库采用 Apache License 2.0，商业使用较为宽松，只有使用云服务时才需要付费

Mastra如何管理上下文

Mastra提出了名为 Observational Memory 的全新上下文管理方法

先设定token上限(默认30k)，主会话每次达到该上限时，就调用另一个小模型(gpt-5-mini)生成日志总结；日志到达上限后，再次对日志进行总结

系统还提供名为recall的tool，可对当前会话及过往其他会话进行精确的消息擦护照，从而确保模型始终拥有充裕上下文

构造Agent

import { createOpenAI } from "@ai-sdk/openai";import { Agent } from "@mastra/core/agent";import { createTool } from "@mastra/core/tools";import { createSkill } from "@mastra/core/skills";import { Memory } from "@mastra/memory";import { PostgresStore } from "@mastra/pg";import { z } from "zod";const API_KEY = process.env.API_KEY!;const BASE_URL = process.env.BASE_URL!;/** 主会话模型 */const CHAT_MODEL_ID = process.env.CHAT_MODEL_ID!;/** OM模型 */const OM_MODEL_ID = process.env.OM_MODEL_ID!;const POSTGRES_URL = process.env.POSTGRES_URL!;// 模型providerconst provider = createOpenAI({apiKey: API_KEY,baseURL: BASE_URL,});// 数据存储层当前采用postgresexport const store = new PostgresStore({connectionString: POSTGRES_URL,id: "mastra-demo",});// 消息持久化和上下文管理层const memory = new Memory({storage: store,options: {// om配置observationalMemory: {model: provider.languageModel(OM_MODEL_ID),// om模型的总结范围仅限当前会话scope: "thread",// 主会话的检索范围仅限当前会话retrieval: { scope: "thread" },},// 生成标题generateTitle: {model: provider.languageModel(OM_MODEL_ID),instructions: "根据用户第一条消息生成一个简短的会话标题",}},});// 需要审批的toolexport const sendEmail = createTool({id: "sendEmail",description: "发邮件",inputSchema: z.object({body: z.string(),subject: z.string(),to: z.string(),}),outputSchema: z.object({status: z.union([z.literal("success"), z.literal("failed")]),reason: z.string().optional(),}),// 预审批可以传函数// 拒绝时无法传入理由requireApproval: true,// 执行过程中可以中断进行二次确认suspendSchema: z.object({body: z.string(),subject: z.string(),to: z.string(),}),// 最后一次恢复得到的值resumeSchema: z.object({approved: z.boolean(),reason: z.string().optional(),}),// 如果抛出异常其消息会被作为tool-call-errorexecute: async (input, context) => {// input可以信任但resumeData不可信任const resumeData = context.agent?.resumeData;if (resumeData) {if (!resumeData.approved) {return { status: "failed" as const, reason: resumeData.reason };}} else if (input.to === "[email protected]") {return await context.agent?.suspend(input);}/** 发送邮件 */return { status: "success" as const };},});const testSkill = createSkill({name: "xx",description: "xx",instructions: "skill-body",})export const agent = new Agent({id: "demo",name: "demo",// 系统提示词允许异步函数返回stringinstructions: "系统提示词",memory,model: provider.languageModel(CHAT_MODEL_ID),tools: { sendEmail },skills: [testSkill],});

关于requireApproval和suspend

不要同时使用两种机制

恢复挂起的流依赖resumeData，而二者底层执行的是同一套行为逻辑

二者同时开启时会先经过requireApproval的校验检查resumeData的approved属性不为true就直接以Tool call was not approved by the user作为toolcall的结果即使之前已经approval过了

关于Subagent

不会抛异常，是因为subagent在调用需要resume的tool后便结束执行；此前已经生成的text会作为result直接交回主会话

使用Agent

要开展多轮对话，必须先把Agent注册到Mastra实例中

export const mastra = new Mastra({agents: { demo: agent },storage: store,})

agent对话涉及三个id resourceId threadId runId

对应关系目前可先这样理解：每轮会话的id是runId，sessionId是threadId，userId是resoueceId

发送流式消息(开始会话或多轮对话)

// stream.id就是runIdconst stream = await agent.stream('xx', {memory: { thread: threadId, resource: resourceId },})for await (const c of stream.fullStream){/** */}

从挂起的流恢复

// 通过予审批(拒绝也可以)// approved是一个新的流const approved = agent.approveToolCall({ runId, toolCallId })// resume一个挂起的toolconst resumed = agent.resumeStream(resumeData, { runId, toolCallId })

停止生成

// 停止所有指定thread的runagent.abortThreadStream({resourceId,threadId})// 停止指定runagent.abortRunStream(runId)

sendMessage/sendSignal

agent.sendMessage用于向thread中加入用户消息可以基于当前会话fork新thread 可以等待当前run生成完毕适用于一个thread多人参与的场合(比如群聊Agent)

agent.sendSignal它与sendMessage相似，但插入的是“系统消息”而非“用户消息”，因此呈现给模型的方式不同

这两个函数自身不会返回流，使用前需要先订阅流

const subscription = await agent.subscribeToThread({ resourceId, threadId })for await (const c of subscription.stream) {}subscription.unsubscribe()

resume同上

AgentController

自动会话管理、消息队列、会话fork等均由mastra预设的Agent运行时AgentController提供；ui更新、工具审批等一系列api也被它封装

这个api目前仍处于beta阶段，文档不够准确，体验很差，但未来可期

resource,thread和run

记忆层由resource和thread构成，每个thread都必定归属于一个resouce

run处于执行层。每次调用stream都会继承thread中的历史消息并启动新对话；本轮内容全部生成后，run会被删除

在tool 审批/suspense时 run挂起 runid不变

thread不是session

thread内存储的是完整的消息即完整的toolcall+result/text(一次step的完整输出) 流式消息片段不会加入thread

会话可能产生混乱，例如两个tab页同时对话：原因是mastra默认允许同时从thread发起对话，不限制由thread产生run，随后再按时间顺序把新消息写入thread

sendMessage/sendSignal

上述stream竞态问题，可由sendMessage在一定程度上处理

agent.sendMessage('Compare that with the previous option.', {resourceId,threadId,// thread存在活跃/挂起run的行为ifActive: {},// thread不存在活跃run的行为(即 thread空闲)ifIdle: {},})

ifAvcive

{behavior?: "deliver" | "persist" | "discard"}

persist-消息加入thread但当前run无法感知；discard-丢弃；deliver(默认)-放入当前run的下一条消息中

ifIdle

{behavior?: "persist" | "discard" | "wake"}

wake(默认)-唤起新run其余同上

上述机制仅在单个后端实例中生效

多实例环境中的竞态问题

官方提供的解决方案是使用@mastra/redis-streams的RedisStreamsPubSub

在new Mastra时传入pubsub属性，以便处理竞态问题

前端展示

next+ai-sdk/react+ai-elements很适合用于展示mastra，因为mastra构建在ai-sdk之上

无法直接使用的原因，是mastra引入ai-sdk时没有采用peerDependencies方式

可以参考官方文档搭建基础设施，并完成以下优化

处理类型错误

api/chat/route.ts POST函数

const stream = await handleChatStream({// 增加versionversion: "v6",});

不要改GET的内容 v5的sdk messages是正确的版本

增加approval和suspend能力

目标是让流式传输和历史消息都能采用相同方式，向用户显示挂起状态并恢复会话

这两种场景采用的消息结构并不相同

approval：历史消息里没有tool-approval-request消息；流式传输期间，该消息位于toolcall和data-tool-call-approval之间

流式传输与历史消息在suspense上表现一致：data-tool-call-suspended消息始终紧跟在toolcall之后

此处采用以下方案恢复会话

const { sendMessage } = useChat({/* xx */})const handleResume = (runId: string, resumeData: Record<string, unknown>) =>sendMessage(undefined, { body: { runId, resumeData } })

body会被传给后端并用于恢复会话。前文已经说过 approval和suspended本质是一套机制可以这样写这种写法还抹平了流式传输和历史会话消息结构的差异

前端需要进行如下修改

messages.map(message=>{message.parts?.map((part, i)=>{if (part.type === 'data-tool-call-approval') {const data = part.data as ToolCallApprovalDatareturn (<><Button onClick={() => handleResume(data.runId, { approved: false })}>拒绝</Button><Button onClick={() => handleResume(data.runId, { approved: true })}>同意</Button></>)}if (part.type === 'data-tool-call-suspended') {const data = part.data as ToolCallSuspendedData// 同上 resumeData按照tool要求的格式即可}})})