TypeScript 开发者的 Agent 框架

2026 年,Agent 框架的生态以 Python 为主流,但大量全栈开发者的技术栈是 TypeScript。Mastra 正是为这群人打造的——一个 TypeScript 原生的 Agent 框架,不是 Python 框架的移植,而是从零开始为 TypeScript 生态设计的。

设计理念

Mastra 的核心理念:

  1. TS Native:利用 TypeScript 的类型系统,而非 Python 的动态特性
  2. Edge Ready:支持 Vercel Edge Functions、Cloudflare Workers 等边缘计算平台
  3. Full-Stack:前端、后端、CLI 工具链一体化
  4. React Integration:与 React/Next.js 生态无缝集成

快速上手

安装与初始化

# 创建新项目
npx create-mastra@latest my-agent-app

# 项目结构
my-agent-app/
├── src/
│   ├── agents/
│   │   ├── research-agent.ts
│   │   └── writing-agent.ts
│   ├── workflows/
│   │   └── content-pipeline.ts
│   ├── tools/
│   │   ├── search.ts
│   │   └── database.ts
│   ├── Mastra.config.ts
│   └── index.ts
├── mastra/
│   └── playground/      # 可视化调试界面
├── package.json
└── tsconfig.json

定义 Agent

import { Mastra } from '@mastra/core';
import { createAgent } from '@mastra/core/agent';
import { openai } from '@ai-sdk/openai';
import { z } from 'zod';

// 使用 Zod 定义结构化输出
const researchOutputSchema = z.object({
  topic: z.string(),
  keyFindings: z.array(z.string()),
  confidence: z.number().min(0).max(1),
  sources: z.array(z.object({
    url: z.string().url(),
    title: z.string(),
    reliability: z.enum(['high', 'medium', 'low'])
  })),
  recommendation: z.enum(['proceed', 'caution', 'avoid'])
});

// 创建 Agent
export const researchAgent = createAgent({
  name: 'research-agent',
  description: '深度研究分析师',
  model: openai('gpt-4o'),
  instructions: `你是一名专业研究分析师。
    - 始终提供来源链接
    - 区分事实和推测
    - 给出明确的置信度评分
    - 保守地给出建议`,
  outputSchema: researchOutputSchema,
  tools: {
    webSearch: searchTool,
    knowledgeBase: kbTool,
  },
  memory: {
    type: 'semantic',
    config: {
      embeddings: openai.embedding('text-embedding-3-large'),
      vectorStore: 'pgvector',
    },
  },
  // TypeScript 完整类型推导
  // output 类型自动推断为 z.infer<typeof researchOutputSchema>
});

工具定义

import { createTool } from '@mastra/core/tools';
import { z } from 'zod';

// 类型安全的工具定义
export const searchTool = createTool({
  id: 'web-search',
  description: '搜索互联网获取最新信息',
  inputSchema: z.object({
    query: z.string().describe('搜索查询'),
    maxResults: z.number().default(10).describe('最大返回数'),
    timeRange: z.enum(['day', 'week', 'month', 'year']).optional()
      .describe('时间范围'),
  }),
  outputSchema: z.object({
    results: z.array(z.object({
      title: z.string(),
      url: z.string().url(),
      snippet: z.string(),
      publishedAt: z.string().datetime().optional(),
    })),
    totalResults: z.number(),
  }),
  execute: async ({ input, context }) => {
    // input 类型自动推断:{ query: string, maxResults: number, ... }
    const response = await fetch(
      `https://api.search.com/v1/search?q=${encodeURIComponent(input.query)}&limit=${input.maxResults}`
    );
    const data = await response.json();
    
    return {
      results: data.items.map((item: any) => ({
        title: item.title,
        url: item.link,
        snippet: item.snippet,
        publishedAt: item.publishedAt,
      })),
      totalResults: data.totalResults,
    };
  },
});

// 数据库工具
export const dbTool = createTool({
  id: 'database-query',
  description: '查询内部数据库',
  inputSchema: z.object({
    sql: z.string().describe('SQL 查询语句(只读)'),
    params: z.array(z.union([z.string(), z.number()])).optional(),
  }),
  outputSchema: z.object({
    rows: z.array(z.record(z.unknown())),
    rowCount: z.number(),
  }),
  execute: async ({ input, context }) => {
    const client = await context.db.connect();
    try {
      // SQL 注入防护
      if (!/^SELECT/i.test(input.sql)) {
        throw new Error('仅允许 SELECT 查询');
      }
      const result = await client.query(input.sql, input.params || []);
      return {
        rows: result.rows,
        rowCount: result.rowCount,
      };
    } finally {
      client.release();
    }
  },
});

Workflow 引擎

Mastra 的 Workflow 引擎支持状态机和工作流编排:

import { createWorkflow, step, branch, parallel } from '@mastra/core/workflow';
import { z } from 'zod';

const contentPipeline = createWorkflow({
  id: 'content-pipeline',
  inputSchema: z.object({
    topic: z.string(),
    targetAudience: z.string(),
    wordCount: z.number().default(2000),
  }),
  outputSchema: z.object({
    research: researchOutputSchema,
    outline: z.object({
      sections: z.array(z.object({
        title: z.string(),
        keyPoints: z.array(z.string()),
      })),
    }),
    article: z.string(),
    review: z.object({
      score: z.number(),
      feedback: z.string(),
      approved: z.boolean(),
    }),
  }),
  steps: [
    // 步骤 1:研究
    step({
      id: 'research',
      execute: async ({ input, mastra }) => {
        const result = await researchAgent.execute({
          messages: [{ role: 'user', content: `研究主题: ${input.topic}` }],
        });
        return { research: result.output };
      },
    }),
    
    // 步骤 2:生成大纲 + 并行获取数据
    parallel([
      step({
        id: 'outline',
        execute: async ({ input, mastra, prevStep }) => {
          const result = await outlineAgent.execute({
            messages: [{
              role: 'user',
              content: `基于以下研究生成文章大纲: ${JSON.stringify(prevStep.research)}`
            }],
          });
          return { outline: result.output };
        },
      }),
      step({
        id: 'data-collection',
        execute: async ({ input, mastra }) => {
          const data = await collectData(input.topic);
          return { supplementaryData: data };
        },
      }),
    ]),
    
    // 步骤 3:条件分支
    branch({
      id: 'review-check',
      condition: ({ prevSteps }) => prevSteps.outline.sections.length >= 3,
      ifTrue: [
        step({
          id: 'write',
          execute: async ({ input, prevSteps }) => {
            const result = await writingAgent.execute({
              messages: [{
                role: 'user',
                content: `基于大纲撰写文章: ${JSON.stringify(prevSteps.outline)}`
              }],
            });
            return { article: result.output };
          },
        }),
      ],
      ifFalse: [
        step({
          id: 'revise-outline',
          execute: async ({ prevSteps }) => {
            // 大纲不够详细,重新生成
            return { needsRevision: true };
          },
        }),
      ],
    }),
    
    // 步骤 4:审核
    step({
      id: 'review',
      execute: async ({ prevSteps }) => {
        const result = await reviewAgent.execute({
          messages: [{
            role: 'user',
            content: `审核文章: ${prevSteps.article}`
          }],
        });
        return { review: result.output };
      },
    }),
  ],
});

React 集成

Mastra 提供了 React hooks,使 Agent 可以直接在前端使用:

import { useAgent, useWorkflow } from '@mastra/react';
import { researchAgent } from './agents/research-agent';
import { contentPipeline } from './workflows/content-pipeline';

function ResearchApp() {
  const { messages, send, isLoading, error } = useAgent(researchAgent);
  const { start, status, data } = useWorkflow(contentPipeline);

  return (
    <div>
      {/* Agent 聊天界面 */}
      <div className="chat">
        {messages.map((msg) => (
          <div key={msg.id} className={msg.role}>
            {msg.content}
          </div>
        ))}
      </div>
      
      <button onClick={() => send('分析 2026 年 AGI 行业')}>
        研究
      </button>
      
      {/* Workflow 状态 */}
      <div className="workflow-status">
        <p>状态: {status}</p>
        {data && (
          <div>
            <h3>研究发现:</h3>
            <ul>
              {data.research?.keyFindings.map((finding, i) => (
                <li key={i}>{finding}</li>
              ))}
            </ul>
          </div>
        )}
      </div>
      
      <button onClick={() => start({
        topic: 'AGI 行业分析',
        targetAudience: '投资者',
        wordCount: 3000,
      })}>
        启动内容管道
      </button>
    </div>
  );
}

Edge 部署

// Vercel Edge Function
export const config = {
  runtime: 'edge',
};

export default async function handler(req: Request) {
  const mastra = new Mastra({
    agents: { research: researchAgent },
    storage: { type: 'upstash-redis', url: process.env.UPSTASH_URL },
  });
  
  const { messages } = await req.json();
  const result = await mastra.agents.research.stream(messages);
  
  return new Response(result.toReadableStream());
}

// Cloudflare Worker
export default {
  async fetch(request: Request, env: Env) {
    const mastra = new Mastra({
      agents: { research: researchAgent },
      storage: { 
        type: 'd1', 
        binding: env.DB 
      },
    });
    
    return mastra.handleRequest(request);
  },
};

性能基准

指标MastraLangChain.jsVercel AI SDK
Cold Start150ms400ms80ms
简单调用0.9s1.3s0.8s
结构化输出1.1s1.6s1.0s
Bundle 大小85KB320KB45KB
Edge 支持
Workflow

框架对比

特性MastraLangChain.jsVercel AI SDK
TypeScript 原生移植
Agent 定义类型安全动态类型安全
工具系统Zod SchemaJSON SchemaZod Schema
Workflow内置
记忆系统语义记忆基础
RAG内置内置需手动
React 集成官方 Hooks官方 Hooks
Edge 部署原生不支持原生
可视化调试Playground

适用场景

最适合

  1. Next.js / React 全栈团队:无缝集成现有技术栈
  2. Edge 部署需求:需要低延迟的边缘计算场景
  3. 类型安全要求高:TypeScript 严格模式团队
  4. SaaS 产品:需要前后端统一的 Agent 体验

不太适合

  1. 数据科学团队:Python 生态更适合数据处理
  2. 复杂多 Agent:多 Agent 协作能力不如 LangGraph
  3. 本地模型优先:对 Ollama 等本地模型支持不如 Python 框架
  4. 研究型项目:生态和社区不如 Python 框架丰富

生态数据

  • GitHub Stars:12.8k
  • npm 周下载量:8.5 万
  • Discord 成员:3200+
  • 企业用户:Vercel、Linear、Cal.com 等
  • 插件/工具:85+

总结

Mastra 在 2026 年确立了"TypeScript Agent 框架首选"的地位。它不是 LangChain 的 JavaScript 翻译版,而是一个从 TypeScript 生态出发、为全栈开发者设计的框架。

对于 TypeScript 技术栈的团队,Mastra 几乎是无需犹豫的选择。类型安全的 Agent 定义、内置 Workflow 引擎、React 集成、Edge 部署支持——这些都是 Python 框架无法提供的独特价值。

不过需要承认的是,Mastra 的生态规模和社区活跃度与 Python 阵营的 LangChain 还有差距。如果你需要在 Python 和 TypeScript 之间做选择,核心问题是:你的团队更熟悉哪个语言生态?Agent 的功能需求是否严重依赖 Python 数据科学生态?回答了这两个问题,选型就有了答案。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞,认知差异创造无限可能。