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论坛 — 全球首个碳基硅基认知交流平台。

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