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

Mastra是什么

Mastra是一个用TypeScript编写的Agent开发框架。与ai-sdk、langgraph这类底层框架不同——它们更关注单次tool call的自动响应和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 === "a-dangerous-email@host.com") {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

有一点需要特别注意：不要同时使用requireApproval和suspend两种机制。因为它们底层共用同一套行为逻辑，都是通过resumeData恢复挂起的流。如果同时开启，会先经过requireApproval的校验，检查resumeData的approved属性，如果不为true，就会直接以Tool call was not approved by the user作为tool call的结果——即使之前已经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。目前可以简单理解为：resourceId = userId，threadId = sessionId，runId = 每轮会话的id。

发起流式消息（开始会话或多轮对话）：

// 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

AgentController是Mastra预设的Agent运行时，集成了自动会话管理、消息队列、会话fork等功能，还封装了UI更新、工具审批等一系列API。不过这个API目前还处于beta阶段，文档不准，体验极差，未来可期吧。

resource,thread和run

resource和thread属于记忆层，thread必定归属某个resource。run属于执行层，每次调用stream时，都会从thread继承过往消息并开启新对话，本轮完全生成完毕后run会被删除。在tool审批或suspense时，run会挂起，runid不变。

thread不是session

thread内部存储的是完整的消息——即完整的tool call + result/text（一次step的完整输出），流式消息片段不会加入thread。Mastra并不默认限制从thread产生run，可以同时从thread发起对话，然后按时间顺序将新消息加入thread，这就可能导致会话混乱，比如两个tab页同时对话。

sendMessage/sendSignal

sendMessage可以在一定程度上解决上述的stream竞态问题。

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

ifActive：

{beha vior?: "deliver" | "persist" | "discard"}

deliver（默认）——作为当前run的下一条消息；persist——消息加入thread，当前run无法感知；discard——丢弃。

ifIdle：

{beha vior?: "persist" | "discard" | "wake"}

wake（默认）——唤起新run，其余同上。以上机制在单个后端实例时生效。

多实例的竞态问题

多实例场景下的竞态问题，官方推荐使用@mastra/redis-streams的RedisStreamsPubSub，在new Mastra时传入pubsub属性来解决。

前端展示

Mastra是基于ai-sdk构建的，所以Next + ai-sdk/react + ai-elements的组合非常适合展示Mastra。不过Mastra并不是以peerDependencies方式引入ai-sdk的，所以不能直接拿来用。参考官方文档搭好基础设施，然后做如下优化。

解决类型错误

api/chat/route.ts POST函数：

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

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

增加approval和suspend能力

目标很明确：无论流式传输还是历史消息，都能以相同的方式向用户展示挂起状态并恢复会话。但两种情况下消息结构不同。在approval场景下，流式传输期间toolcall和data-tool-call-approval之间有一条tool-approval-request消息，而历史消息中缺少这条消息。suspense则始终在toolcall后跟随data-tool-call-suspended消息，流式传输和历史消息是一致的。恢复会话的方案如下：

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要求的格式即可}})})

以上代码仅作示例，实际使用时还需要考虑这个挂起是否已经处理过。