Mastra TS框架构建AI Agent

可视化演示

展示Mastra TypeScript框架的Agent架构、执行时间线与技术栈分层，包含工具组合与工作流。

框架概览——即开即用的AI后厨

假设你要开一家餐厅，最理想的状态是租下一间“拎包入住”的后厨：灶台、冰箱、传菜梯、洗碗机全部就位，你只需要招一个主厨，把菜谱写清楚，就能开门营业。Mastra 就是 AI 应用世界的这间后厨——它把 Agent 开发中常用的底座能力预装好了，让熟悉 TypeScript 的开发者能快速把想法变成可运行的 AI 应用。

在 Mastra 里，你的 AI 应用由一个或多个 Agent 组成。Agent 就像餐厅里的主厨，负责理解客人需求、指挥备菜、最终出餐。主厨需要工具，比如下单系统、菜品库存查询——这对应 Mastra 的 Tools；主厨需要记住熟客口味——对应 Memory；主厨还要按固定流程走菜——对应 Workflow；遇到不会的菜则翻菜谱——对应 RAG。这五个模块被 Mastra 打包成统一的框架，正是它给出的解决之道。

更关键的是：Mastra 是 TypeScript 原生的。类型安全不是锦上添花，而是在你写错配置、传错参数时，在编译期就把你拦住。比如你在 new Agent({ model, tools, memory }) 里少传了一个工具，IDE 会立刻标红；而传统 Python 项目往往要运行后才会报错。对于已经在用 TS/JS 技术栈的团队，Mastra 的学习成本被降到很低，不需要为了 AI 额外引入一整套 Python 数据科学体系。

你可以把 Mastra 想象成一间“精装后厨”：设备齐全、动线清晰，你要做的就是写清楚菜单（Agent 指令），然后点亮这家餐厅。这正是它与其他“从零拼装”的原生方案最大的不同——不是能不能做，而是能不能做到又快又稳。

架构设计——后厨的三层动线

后厨不是一间大平层，而是分层的：前厅负责接待与传达客人需求，后厨负责烹饪出品，库房负责供货。Mastra 的架构也分了清晰的层次：最上层是“应用层”，由你自己写的代码构成；中间是“核心引擎层”，负责调度、路由、记忆读写；最底层是“Provider 层”，负责对接各个大模型服务商。

应用层就是你写的 Agent 定义、Workflow 配置等业务代码，这也是你唯一需要关心的部分。中间层，核心引擎会读取你的 Agent 配置，决定何时调用模型、何时执行工具、何时读写记忆，并把结果拼装成回复。最底层的 Provider 层把 OpenAI、Anthropic、Google 等厂商的 API 统一成一套接口。换模型时，只要改一行配置，而不是把业务代码重写一遍。

三层之间的解耦带来直接好处：业务代码不依赖任何一家模型厂商。今天用 GPT-4o，明天想换 Claude，只需要改 Provider 配置。同时，核心引擎提供大量可扩展的钩子（hooks），允许你插入自定义 Memory 存储（文件、Redis、数据库）、自定义检索器，像给后厨加装多层货架一样灵活。

三层布局让整个后厨保持运转流畅：前厅只管点餐，后厨只管做菜，仓库只管供货，互不干扰。等你要把餐厅开成连锁店（扩展多个 Agent），也不需要推翻重建——只要在对应层增加配置即可。这种职责边界，正是架构设计的核心价值。

核心API——主厨的任务清单

主厨开工前需要一张“任务清单”：掌勺做什么菜、用什么锅、放什么调料。Mastra 给了你同样的清单——Agent 对象、Tool 对象、Workflow 对象。学会这几个对象，你就掌握了整套 API 的 80%。

最常用的是 new Agent()：传入 instructions（系统提示词）、model（LLM 选择）、tools（工具注册表）、memory（记忆存储）。然后调用 agent.generate() 或 agent.stream()。Generate 是“一次性出餐”：等全部结果生成完再给你；Stream 是“边做边传菜”：可以一个 token 一个 token 地流式返回，让用户看到“正在写”的过程。

Tools 的注册也很直观：你定义函数的参数 schema（可以用 zod 做类型校验），Mastra 会自动将这个 schema 发给 LLM，让模型在合适时机“请求调用工具”；执行完成后，结果再回到 Agent 的流程中。所有这些都是声明式的——你描述数据结构，框架替你完成与模型的中间协商。

这几样东西，就是主厨开工前的全部准备。你给 Agent 配好大脑、厨具和记事本，它就能独立应对一整桌复杂的客人需求。越是复杂的 AI 应用，越需要这种结构化配置来避免混乱——否则十几个 prompt 散落在各处，最终无人能维护。

工作流编排——传菜动线

一家高效的餐厅，传菜动线是精心设计的：凉菜先出，热菜按灶台能力排队，甜品最后。Mastra 的 Workflow 就是把 AI 业务拆成一连串步骤（Step），并规定它们之间的先后顺序与分支。每个 Step 可以是调用一个 LLM、执行一个工具、或跑一段逻辑。

Workflow 的核心概念是 Step 和 Route。Step 是工作流中的一个节点；Route 决定下一步去哪个 Step。比如“判断用户意图”这个 Step，如果结果 travel，就进入查询航班步骤；否则进入推荐酒店步骤。这是典型的条件路由——像后厨根据订单类型决定走哪条加工线。

步骤之间通过状态（state）传递数据。前一步的输出会成为后一步的输入。因为你可以在 Steps 中做异步调用（比如等外部 API 返回），Mastra 会像传菜一样自动等待每一步完成，然后根据路由决定下一步。相对于“一段 prompt 写完所有事”，Workflow 让流程天然可观测、可调试、可回放。

这一设计让“出餐流程”不再是一团浆糊。以前你可能把好几十个决策全塞进一个 prompt；现在你把它拆成明确的传菜路线，每一个环节都能单独测试、单独上线——这正是工程化 AI 应用的关键：把不可解释的“黑盒对话”变成可审计的“流水线”。

记忆管理——主厨的记事本

好的主厨会记得熟客忌口：张先生不吃香菜、李小姐喜欢微辣。AI Agent 也一样，如果每次对话都“失忆”，体验会像跟一个陌生服务员打交道。Mastra 内置了记忆系统，帮你把“见过的人”和“说过的话”随时带在身边。

Mastra 的记忆分为短期记忆、长期记忆和工作记忆。短期记忆保存当前会话的上下文窗口，通常用于维持对话连贯性；长期记忆把重要信息（用户偏好、历史事实）写入向量数据库，即使重启服务也不会丢；工作记忆则用于 Agent 执行任务时临时保留中间结果，相当于主厨桌上的“出餐备忘卡”。

实现上，Mastra 的长期记忆通常对接向量存储（如 PostgreSQL + pgvector、Pinecone 等），通过嵌入（embedding）把文本向量化后保存。当新对话到来，系统会从向量库中检索与该用户相关的历史片段，把它们塞进上下文，让 Agent“想起来”。这个过程对业务代码几乎透明——你只需要配置一个 Memory 存储，剩下的由框架代劳。

有了记事本，AI 应用就不再是“每次从零开始”的服务员，而是一个能记住老客人喜好的熟手主厨。用户在跟你的 Agent 打交道时，也会明显感到“这个人好像记得我之前说过的话”——这是 AI 应用体验升级的重要分水岭。

工具调用与RAG——厨具与菜谱库

再厉害的主厨也不能徒手变出食材，他需要厨具来切菜、搅拌、控温。AI Agent 也一样：如果你要做一个可以查实时天气、下订单的助手，就必须让它在合适时使用外部 API——这就是 Mastra 的工具调用能力。

在 Mastra 里，你可以用 tool 函数把一段代码包装成一个“可被 AI 调用的工具”。它需要声明名称、描述、输入参数 schema。当用户问“帮我查一下天气”，LLM 看到工具描述后，会生成一次工具调用，Mastra 在本地执行该函数，再把结果继续交给 LLM 生成最终回复。这个循环是 Agent 理解并操作系统世界的基础。

RAG（检索增强生成）则是“菜谱库”：你有很多外部文档（产品手册、FAQ、内部知识库），它们太大，不能全部塞进 prompt。Mastra 先对文档做切分并向量化存入向量库；用户提问时，它把问题转为向量，检索最相似 K 个片段，将这些片段作为上下文交给 LLM 生成答案。这相当于主厨做一道不熟悉的菜时，先翻到菜谱对应页再动手。

工具让 Agent 有了“手”，RAG 让 Agent 有了“知识库”。两者结合时，一个 Agent 既能查实时数据，又能基于私有文档回答，已经逼近一个“有经验的老员工”的水平。Mastra 把这两条能力做成内置接口，你不用自己去拼接 OpenAI 的 function calling 和向量查询，整个流程在框架里自动完成。

对比LangGraph——两套厨房的选择

后厨的装修方案不止一种。LangGraph 是 Python 生态里成熟的后厨装修队，而 Mastra 是 TypeScript 生态里的新式精装方案。这两套方案各有拥趸。关键是你自己的后厨（团队技术栈）用哪种更方便。

比较几个维度。语言生态：LangGraph 以 Python 为核心，有非常丰富的 Python ML 生态；Mastra 原生 TypeScript，能直接复用 npm 生态、与前端代码共享类型。内置能力：Mastra 把 Memory、RAG、Workflow 等打包在框架内，开箱即用；LangGraph 更偏“编排引擎”哲学，记忆、向量库往往要通过 LangChain / LangSmith 等组合出来。开发体验：TS 有编译期类型检查，重构时能更早发现错误；Python 上手门槛更低，适合以研究探索为核心的团队。

选型建议是：如果你的团队已经深度依赖 TypeScript / JavaScript（Node 服务端、全栈一体化），Mastra 可以把 Agent 开发接入现有代码库，几乎零额外学习成本；如果你的团队主要是 Python 数据工程出身，并且已经用 LangChain 做过大量开发，那 LangGraph 的生态延续性更合适。两者都是从“节点式编排”起步的架构，不存在绝对优劣。

所以无论选 Mastra 还是 LangGraph，本质上不是“谁更好”，而是“哪个更像为你这支团队量身定做的后厨”。看别人家的菜谱评价只是参考，合不合口味只有自己试过才知道。选完就扎进去做，比反复纠结“站队”更有价值。

部署上线——开业大吉

后厨装修好、试菜完成，最终还是要把餐厅开起来营业。Agent 应用也一样，写完流程、接好工具，下一步就是部署上线。Mastra 的部署方式和普通 Node.js 服务没有本质区别：打包、启动、对外暴露 API——因为它本质就是一个服务端应用。

Mastra 可以本地通过 npx mastra dev 或 Docker 运行，也可以部署到各类云平台。由于框架底层就是 TS/Node，你可以将它接入现有的 Node 部署管道（PM2、Docker Compose、Kubernetes）或 Serverless（如 Vercel、Netlify、AWS Lambda）。部署时要注意 Worker 的启动时长、内存，以及对 OpenAI 等外部 API 的网络出站规则。

上线之后更重要的是监控与降级策略。Agent 依赖外部大模型的稳定性，一旦模型服务抖动，你的应用会直接表现为“变慢”或“报错”。所以你需要：记录每次调用的日志（延迟、token 数、错误码）、对重要流程加超时重试、在模型不可用时给用户一个兜底回复（比如“稍后再试”）。这就是把 Agent 当系统而不是当脚本对待的成熟心态。

开业只是开始，稳定的经营才是目标。用 Mastra 快速上线只是第一步，真正的功力体现在后续的监控、调优、降级设计上。把每一道菜（请求）都做得稳定，这家 AI 餐厅才能长久红火，客人（用户）才会愿意再来。

AI 应用场景

Python 演示

"""用 Python 模拟 Mastra 风格 Agent 的最小运行时，演示记忆、工具与工作流的协作。""" import json  class Tool:    def __init__(self, name, handler, description=""):        self.name = name        self.handler = handler        self.description = description     def run(self, args):        return self.handler(**args)  class Memory:    def __init__(self, limit=6):        self.messages = []        self.limit = limit     def add(self, role, content):        self.messages.append({"role": role, "content": content})        self.messages = self.messages[-self.limit:]     def context(self):        return "\n".join(m["content"] for m in self.messages)  class WorkflowStep:    def __init__(self, name, handler):        self.name = name        self.handler = handler  class MastraLikeAgent:    def __init__(self, instructions, memory=None):        self.instructions = instructions        self.memory = memory or Memory()        self.tools = {}        self.steps = []        self.routes = {}     def register_tool(self, tool):        self.tools[tool.name] = tool     def add_step(self, step, routes=None):        self.steps.append(step)        if routes:            self.routes[step.name] = routes     def run_workflow(self, initial_data):        state = dict(initial_data)        current = self.steps[0] if self.steps else None        while current:            out = current.handler(state)            state[current.name] = out            route_map = self.routes.get(current.name, {})            next_name = route_map.get(out, None)            current = next(                (s for s in self.steps if s.name == next_name),                None            )        return state     def chat(self, user_input):        self.memory.add("user", user_input)        context = self.memory.context()        tool_names = "、".join(self.tools.keys())        prompt = (            f"{self.instructions}\n用户说：{user_input}\n"            f"上下文：{context}\n可用工具：{tool_names}"        )        reply = f"[Agent 模拟回复] 已接收“{user_input}”，将结合上下文与工具结果生成答复。"        self.memory.add("assistant", reply)        return reply  def main():    agent = MastraLikeAgent(instructions="你是一个擅长调用工具的客服助手。")    agent.register_tool(Tool("查询订单", lambda order_id: {"status": "已发货"}, "按订单号查询状态"))    agent.register_tool(Tool("修改备注", lambda order_id, note: {"ok": True}, "修改订单备注"))     def first_step(state):        state["intent"] = "query"        return "query"     def second_step(state):        ret = agent.tools["查询订单"].run({"order_id": "A123"})        state["order_status"] = ret["status"]        return "done"     agent.add_step(WorkflowStep("parse", first_step), routes={"query": "fetch"})    agent.add_step(WorkflowStep("fetch", second_step), routes={"done": None})     state = agent.run_workflow({"order_id": "A123"})    print(json.dumps(state, ensure_ascii=False, indent=2))    print(agent.chat("我的订单到哪了？"))  if __name__ == "__main__":    main()

1. class MastraLikeAgent:模拟 Mastra 的 Agent 核心类，同时持有工具、记忆和工作流。
2. def register_tool(self, tool):把工具挂到 Agent 上，相当于把厨具挂上工具墙。
3. def add_step(self, step, routes=None):把一个工作流步骤加入执行序列，并声明它的路由表。
4. def run_workflow(self, initial_data):模拟 Mastra 工作流引擎，循环执行直到没有下一步。
5. out = current.handler(state)调用当前步骤的处理函数，得到输出结果。
6. next_name = route_map.get(out, None)根据步骤输出查路由表，决定下一步走哪个节点。
7. current = next((s for s in self.steps if s.name == next_name), None)找到下一个 Step 对象，找不到就结束。
8. self.memory.add("user", user_input)把用户问题写入短期记忆，形成上下文。
9. agent.register_tool(Tool("查询订单", lambda order_id: {"status": "已发货"}, "按订单号查询状态"))注册一个带描述的工具，LLM 可据此决定何时调用。
10. agent.add_step(WorkflowStep("parse", first_step), routes={"query": "fetch"})声明路由：如果第一步返回 query，则进入 fetch 步骤。
11. state = agent.run_workflow({"order_id": "A123"})启动工作流并传入初始状态。

常见误区

面试题