跳到主要内容

Mastra — 架构与原理

30 秒导读: Mastra 是一个 TypeScript 的 AI 应用框架。它给你两种"跑 LLM"的方式——agent(模型自己决定调哪个工具、迭代到给出答案)和 workflow(你画一张步骤图,框架按图执行,还能中途暂停等人批准)。它最有意思的设计是:记忆不是一个黑盒服务,而是一串"处理器",在 prompt 送进模型前后各插一脚,把历史消息、语义召回结果、工作记忆塞进上下文。

1. 这是什么(零基础也能懂)​

一句话定义: Mastra 是一套用 TypeScript 写、面向生产的框架,帮你构建"会推理、会用工具、有记忆"的 AI agent 和多步骤 workflow。

解决什么问题 / 给谁用: 假设你是一个 TS 工程师,想做一个客服 agent——它要能查数据库、调外部 API、记住用户上次说过什么、必要时停下来等人工审批。如果从零手写,你得自己处理:把对话历史拼进 prompt、解析模型想调哪个工具、执行后把结果再喂回去、循环直到模型给出最终答案、把这一切流式吐给前端。Mastra 把这些都封装好了。

它能做什么(功能):

  • Agent:自主循环——模型推理 → 决定调工具 → 执行 → 把结果喂回 → 再推理,直到给出最终文本或命中停止条件。
  • Workflow:图式编排——用 .then() / .branch() / .parallel() 串步骤,支持 suspend/resume(人在环、长时间等待)。
  • 记忆:对话历史、跨会话语义召回(向量检索)、工作记忆(结构化的"用户档案"模板)、观察式记忆(后台 agent 压缩超长历史)。
  • 工具 / MCP:createTool 定义工具;也能作为 MCP server 把工具暴露给别的系统。
  • 模型路由:一个接口接 40+ 提供商(OpenAI / Anthropic / Gemini …)。

用起来什么样: 一段最小的真实风格调用——

import { Agent } from '@mastra/core/agent';

// 示意,非源码:展示 Mastra agent 的最小用法
const agent = new Agent({
  name: 'support',
  instructions: '你是客服助手,回答要简洁。', // 系统提示
  model: 'openai/gpt-4o',                      // 模型路由用字符串 ID
  tools: { lookupOrder },                       // 一组 createTool 定义的工具
});

// 流式调用,带记忆(thread = 一次会话,resource = 一个用户)
const stream = await agent.stream('我上周的订单到哪了?', {
  memory: { thread: 't1', resource: 'user-42' },
});

for await (const chunk of stream.fullStream) {
  // 文本增量、工具调用、工具结果…都从这里流出来
}

一句话直觉 / 类比: 把 agent 想成"一个会一直追问自己'我还需要做什么'的人",每轮它可以选择"说话(出文本)"或"动手(调工具)";Mastra 就是那个反复把它推回桌前、并随时把它需要的资料(记忆)摆好的助理。

2. 顶层全景(它大概怎么转)​

Mastra 是一个庞大的 monorepo(packages/、stores/、deployers/ …),但价值的核心集中在 packages/core 的少数几个子系统。本节先建立"大盘"。

2.1 一张顶层图:一次 agent.stream() 怎么流动​

下面这张图从上到下是一次 agent 调用的生命周期。关键直觉:记忆不是单独一步,而是"输入处理器"在送进模型前往消息列表里塞东西。

  用户消息
      │
      ▼
┌─────────────────────────────────────────────┐
│  Agent.stream() / generate()                  │  packages/core/src/agent/agent.ts
│  (合并默认选项、解析模型、做权限校验)          │
└───────────────┬───────────────────────────────┘
                │ 调 #execute(...)
                ▼
┌─────────────────────────────────────────────┐
│  ① 输入处理器链 (input processors)             │  prompt 组装前
│     - MessageHistory  载入最近 N 条历史        │  ← 记忆在这里进场
│     - SemanticRecall  向量召回相关旧消息        │
│     - WorkingMemory   把"用户档案"塞成系统消息   │
│  全部往同一个 MessageList 里 add               │
└───────────────┬───────────────────────────────┘
                │ 组好的消息 + 工具 + 模型
                ▼
┌─────────────────────────────────────────────┐
│  ② loop(): 流式 agentic 循环                   │  packages/core/src/loop/loop.ts
│     模型出 token / 工具调用 ──┐                 │
│        ▲                     │ 调工具          │
│        └── 把工具结果喂回 ◀──┘                 │  循环直到 finish
└───────────────┬───────────────────────────────┘
                │ 完成
                ▼
┌─────────────────────────────────────────────┐
│  ③ 输出处理器链 + 落库                          │  prompt 之后
│     - SemanticRecall  给新消息做 embedding 存库 │  ← 记忆在这里落盘
│     - 保存消息到 storage                       │
└───────────────┬───────────────────────────────┘
                ▼
           MastraModelOutput (流式输出对象)

怎么读这张图: 从上往下是时间顺序;记忆出现在两端——进模型前(召回、注入)和出模型后(embedding、落库)。中间 ② 才是大家说的"agentic loop"。

2.2 部件一句话职责​

部件干什么在哪个文件
Agent对外门面:stream/generate,合并配置、解析模型、跑处理器、起循环packages/core/src/agent/agent.ts
loop()真正的流式 agentic 循环(包成一个内部 workflow 跑)packages/core/src/loop/loop.ts
MessageList统一的消息容器,跨 AI SDK v4/v5/v6 格式互转,记录每条消息来源(user/memory/response)packages/core/src/agent/message-list/message-list.ts
MessageHistory / SemanticRecall / WorkingMemory三个记忆处理器,作为 input/output processor 插进管线packages/core/src/processors/memory/
Memory(抽象类)记忆门面:把上面三个处理器按配置装配出来packages/core/src/memory/memory.ts
ObservationalMemory超长对话压缩:Observer + Reflector 两个后台 agentpackages/memory/src/processors/observational-memory/
Workflow / createStep图式编排:then/branch/parallel/dowhile/foreach + suspend/resumepackages/core/src/workflows/workflow.ts
Tool / createTool工具定义(输入/输出 schema + execute)packages/core/src/tools/tool.ts

2.3 主线走一遍(高层,不进代码)​

  1. 你调 agent.stream(messages, { memory: { thread, resource } })。
  2. Agent 合并默认选项、解析出具体模型(字符串 ID → 真实 LanguageModel)、做权限校验,然后进 #execute。
  3. 输入处理器依次往一个 MessageList 里加东西:历史消息、语义召回的相关旧消息、工作记忆系统提示。
  4. 组好的消息 + 工具 + 模型交给 loop(),它跑流式循环:模型出文本/工具调用 → 执行工具 → 结果喂回 → 再推理,直到 finish。
  5. 输出处理器给新产生的消息做 embedding 并落库,供下次语义召回。
  6. 返回 MastraModelOutput,你用 for await ... of stream.fullStream 消费,或 await .text。

3. 阅读地图(建议顺序)​

这是个大项目,按下面顺序读,由浅入深:

  1. 01-agent-loop.md — 先搞懂 agent 的中央循环:stream/generate 怎么把消息、记忆、工具、模型串起来,loop() 怎么流式迭代。这是主线,先读。
  2. 02-memory.md — 再看记忆三件套(消息历史 / 语义召回 / 工作记忆)如何作为处理器插进 prompt 组装。理解了 §1 的"输入处理器"位置才好读这章。
  3. 03-observational-memory.md — 进阶:观察式记忆用两个后台 agent 把超长对话压成"观察",解决上下文窗口爆炸。
  4. 04-workflows.md — 另一条主线:当你要确定性编排而非让模型自主决定时,用图式 workflow + suspend/resume。

4. 边界与定位(诚实)​

  • TypeScript only。 它的卖点就是 TS 一等公民;没有 Python SDK。
  • 强依赖外部 storage / vector。 记忆、suspend/resume、语义召回都要你接一个 store(LibSQL / Postgres / Redis …,见 stores/)。没有 storage,记忆和 streamUntilIdle 这类特性会退化。
  • 版本兼容是个真问题。 代码里大量处理 AI SDK v4/v5/v6 的差异(MessageList 的 adapters、stream() vs streamLegacy())——这是框架"夹在"快速演进的上游 SDK 中间的代价。
  • ee/ 目录是企业许可。 部分鉴权特性(auth/ee)在企业许可下,非纯 Apache-2.0。

5. 代码地图(导航索引)​

主题文件关键符号
Agent 门面 / 入口packages/core/src/agent/agent.tsAgent、stream、generate、#execute
流式 agentic 循环packages/core/src/loop/loop.tsloop、workflowLoopStream
消息容器packages/core/src/agent/message-list/message-list.tsMessageList、AIV4Adapter/AIV5Adapter/AIV6Adapter(message-list.ts:12)
记忆装配packages/core/src/memory/memory.tsMemory(抽象)、getInputProcessors 一带逻辑
记忆处理器packages/core/src/processors/memory/MessageHistory、SemanticRecall、WorkingMemory
观察式记忆packages/memory/src/processors/observational-memory/observational-memory.tsObservationalMemory、observe、reflect
Workflow 构建器packages/core/src/workflows/workflow.tscreateStep、then、branch、parallel、dowhile、foreach
工具定义packages/core/src/tools/tool.tsTool、createTool