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

认识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说明

不会有异常被抛出：需要resume的tool一旦由subagent调用，执行便立即终止，此前生成的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

作为mastra预设的Agent运行时，AgentController包含自动会话管理、消息队列和会话fork等能力，并封装ui更新、工具审批等一系列api。

未来仍值得期待，但现阶段这个api处在beta阶段，不仅文档不准，体验也极差。

resource,thread和run

每个thread都必定属于一个resouce，而记忆层包含的正是thread和resource。

run属于执行层 每次调用stream都会从thread继承过往消息 并开启新的对话 本轮完全生成完毕后 run删除

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

thread不是session

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

从thread产生run时，mastra默认不作限制，因此可以同时由thread发起对话，再按时间顺序把新消息加入thread；这可能造成会话混乱，例如两个tab页同时进行对话。

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

discard-丢弃；persist-加入thread，但当前run无法感知这条消息；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并未通过peerDependencies方式引入ai-sdk，因此无法直接使用。

可先参考官方文档搭建基础设施，再进行以下优化。

处理类型错误

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

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