Mastra 开发（13）：TS 版数据分析 Agent

上一篇把 Agent 长进了 Next.js：一个仓库、一次部署、类型从 schema 穿到浏览器。零件至此备齐，收官实战把它们拼成一台完整的机器：数据分析 Agent。它的需求规格书不在本系列手里——《AI Agent 开发》系列（Python 加 LangGraph）第 30 篇收官时做过同一个项目（端到端数据分析实战），本篇刻意同题再做一遍：同款需求、同款护栏、同款审批，TS 栈给出镜像答案，两篇互为对照，这也是全博客跨系列对照的压轴。

这是《TypeScript Agent 开发（Mastra）》系列第 13 篇。全文按"需求与架构 → SQL 工具层 → Agent 与结构化报告 → 偏好记忆 → 写操作审批 → Web 界面 → 两栈终对照"组织，每一步标注双出处（本系列第 N 篇 / Agent 系列第 N 篇），API 口径沿用系列已核实的 2026 年 10 月 Mastra v1 文档，最后照例留练习。

1、需求与架构

需求与 Agent 系列第 30 篇一字不差，四句话：问得自然（自然语言直接查库）、答得扎实（结论必须带 SQL 与数据行）、聊得下去（"再按渠道拆一下"这类追问成立）、改得安全（写操作必须人工审批）。数据也同款：一个 SQLite 销售库，orders（订单）与 regions（区域）两张表。TS 版把这四条需求映射到 Mastra 零件上，组装图如下：

解读：六个组件全部来自前 12 篇，没有一个新 API。组件级溯源如下，方便与 Python 版对位阅读：

一处刻意的选型差异要先说清：Python 版把 SQL 工具放进独立的 MCP Server，经 MCPAdapter 接入；TS 版收回进程内 createTool 直接定义——本应用是单体，工具没有跨进程复用的诉求，进程内定义换来零 IPC 开销和全程类型检查。哪天要供多个服务共享，再按第 6 篇的 MCPClient 包装成远端工具，护栏函数原样搬走。

2、SQL 工具层

工具层是护栏的主战场：防线与 Agent 系列第 30 篇同款四层——连接层物理只读、语句层白名单、关键字黑名单、结果层 LIMIT 强制；工具写法沿用本系列第 3 篇口径（字段清单见 createTool 参考）。

2.1、只读护栏

护栏是一个纯函数，返回判别联合——通过给出改写后的 SQL，拦截给出错误码：

type Guard = { ok: true; sql: string } | { ok: false; code: string }

export function guardSql(input: string, limit: number): Guard {
  const s = input.trim().replace(/;+\s*$/, '')
  if (s.includes(';')) return { ok: false, code: 'MULTI_STATEMENT' }
  if (!/^\s*(select|with)\b/i.test(s)) return { ok: false, code: 'NOT_SELECT' }
  if (/\b(insert|update|delete|drop|alter|create)\b/i.test(s))
    return { ok: false, code: 'FORBIDDEN_KEYWORD' }
  const m = s.match(/\blimit\s+(\d+)\s*$/i)
  if (m && Number(m[1]) > limit) return { ok: false, code: 'LIMIT_TOO_LARGE' }
  return { ok: true, sql: m ? s : `select * from (${s}) limit ${limit}` }
}

解读：先剥掉尾部分号再检测多语句；白名单只放行 SELECT 与 WITH 开头（CTE 查询合法）；黑名单拦写关键字；最后查 LIMIT——已有 LIMIT 且超过上限就拒绝，没有则包一层子查询注入上限，而不是往尾部追加，因为原语句可能带着 ORDER BY，尾部追加会产出非法 SQL。参数上限归 Zod、语句规则归护栏，正是第 3 篇"schema 是硬约束、护栏是业务规则"的分工。

注意：护栏是纵深防御的一层，不是全部。只读连接拦"真写"，正则拦"大部分坏读"；模型生成大表全扫的慢查询，护栏管不了，要靠驱动超时与限流兜底。

2.2、查询工具与结构工具

查询工具把护栏、只读连接与结构化输出装配到一起：

import { createTool } from '@mastra/core/tools'
import { z } from 'zod'
import Database from 'better-sqlite3'

const db = new Database('sales.db', { readonly: true })

export const queryOutput = z.discriminatedUnion('ok', [
  z.object({
    ok: z.literal(true),
    sql: z.string(), rowCount: z.number().int(),
    rows: z.array(z.record(z.string(), z.unknown())),
  }),
  z.object({ ok: z.literal(false), code: z.string() }),
])

export const sqlQueryTool = createTool({
  id: 'sql-query',
  description: '对销售库执行只读 SELECT 查询并返回行数据',
  inputSchema: z.object({
    sql: z.string().min(1),
    limit: z.number().int().min(1).max(200).default(50),
  }),
  outputSchema: queryOutput,
  execute: async ({ context }) => {
    const g = guardSql(context.sql, context.limit)
    if (!g.ok) return g                    // 拦截结果原样回传，模型自我修正
    const rows = db.prepare(g.sql).all() as Record<string, unknown>[]
    return { ok: true, sql: g.sql, rowCount: rows.length, rows }
  },
})

解读：readonly: true 是物理只读，即使护栏被绕过，驱动也会拒绝写语句——与 Python 版 file:sales.db?mode=ro 同一道闸。outputSchema 用第 3 篇第 6 节的判别联合：护栏拦截属于可预期失败，返回 ok: false 而不是抛异常，模型读到错误码会在下一轮改写 SQL 重来——坏语句是 agent loop 内的自动纠偏信号。queryOutput 单独导出，第 6 节的前端卡片靠它推导 props 类型。

表结构说明更简单：一个 db-schema 工具，入参空对象，返回两表 DDL 常量文档。Python 版拆成 get_schema 与 describe_table 两个 MCP 工具，TS 版收成一个文档工具，模型一次读完结构，省一轮往返。

3、Agent 与结构化报告

报告契约用一个 Zod schema 声明（本系列第 3 篇；官方口径见结构化输出文档），字段与 Python 版 AnalysisReport 一一对应，仅命名换成 camelCase：

import { z } from 'zod'

export const reportSchema = z.object({
  question: z.string(),   // 用户的问题
  answer: z.string(),     // 结论，必须由数据支撑
  sqlUsed: z.string(),    // 支撑结论的 SQL
  rows: z.array(z.record(z.string(), z.unknown())), // 关键数据行
  caveats: z.string(),    // 口径与局限说明
})

解读：sqlUsed 与 rows 是"答得扎实"的落点——报告要通过校验，模型就必须把真实执行的 SQL 与关键行填进去，结构本身逼着结论有出处。Agent 的组装与调用如下：

import { Agent } from '@mastra/core/agent'
import { sqlQueryTool, schemaTool } from '../tools/sql.ts'
import { reportSchema } from '../domain/report.ts'
import { memory } from './memory.ts'

export const dataAgent = new Agent({
  id: 'data-analyst',
  name: '数据分析 Agent',
  model: 'anthropic/claude-sonnet-5-5',
  instructions: '你是销售数据分析助手。先用 db-schema 了解表结构，'
    + '再用 sql-query 查询；结论必须来自查询结果，并遵循用户的偏好便签。',
  tools: { sqlQueryTool, schemaTool },
  memory,                   // 第 4 节展开
})

解读：调用侧 dataAgent.generate(问题, { output: reportSchema, memory: { resource, thread } }) 直接拿到类型完整的报告对象，与第 4 篇实战里 Agent 进 step 的写法同源。对照 Python 版的 response_format=AnalysisReport（Agent 系列第 30 篇）：机制等价，差别在产物——Pydantic 校验通过后类型止步于 Python 进程，Zod 的类型会一路穿到第 6 节的 React 组件。

4、跨会话偏好记忆

"聊得下去"有两层：会话内多轮靠 thread 隔离的历史，跨会话偏好靠 resource 沉淀——用户上周说过"金额都用万为单位"，本周新开会话也该记得。本系列第 5 篇的三件套正好盖住（口径见 Memory 文档）：

import { Memory } from '@mastra/core/memory'

export const memory = new Memory({
  options: {
    lastMessages: 20,
    workingMemory: {
      enabled: true,
      template: `# 用户分析偏好
## 口径
- 默认时间范围：
- 常看区域：
- 常看维度：
## 输出
- 数字格式：[千分位/万为单位]
- 结论风格：[一句话/分点]`,
    },
  },
})

解读：工作记忆是一块随每轮注入 system message 的便签，scope 按 v1 默认取 resource——同一用户的便签跨所有 thread 共享：thread A 里说"以后金额都用万为单位"，开 thread B 问"上月销售额"，报告的 answer 风格立即生效。对照 Python 版的三件手写（Agent 系列第 23、30 篇）——Store 按 namespace 存偏好、手写 save_preference 工具、dynamic_prompt 钩子注入——Mastra 一个配置项全包：updateWorkingMemory 工具框架自带，注入格式框架拼好，第 3 节 instructions 里那句"遵循用户的偏好便签"就是全部约定。

5、写操作审批流

"改得安全"对应数据修正：把某笔订单金额改对、删一条脏数据。这类操作在 TS 版走 Workflow 而非 Agent 工具——判断标准回到第 4 篇那句话："下一步谁说了算"——审批闸门是确定性流程，先提议、再人工批、后执行，顺序写死。Python 侧的同位件是 HumanInTheLoopMiddleware 配 interrupt（Agent 系列第 11、30 篇），本系列第 4 篇的 suspend/resume 是镜像原语。

5.1、审批步骤

流程两步：先生成修正方案，再过人工闸门。方案生成复用第 3 节的 Agent——它挂了 db-schema，知道表结构：

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'
import { dataAgent } from '../agents/data-agent.ts'

const fixPlanSchema = z.object({
  sql: z.string(),      // update/delete 修正语句
  reason: z.string(),
})

const planStep = createStep({
  id: 'plan-fix',
  inputSchema: z.object({ instruction: z.string() }),
  outputSchema: fixPlanSchema,
  execute: async ({ inputData }) => {
    const { object } = await dataAgent.generate(
      `为数据修正需求生成一条 SQL：${inputData.instruction}`,
      { output: fixPlanSchema },
    )
    return object
  },
})

同一个 Agent 换一个 output schema 就干新活。审批步骤是第 4 篇第 5 节的标准形态：

const approvalStep = createStep({
  id: 'fix-approval',
  inputSchema: fixPlanSchema,
  outputSchema: z.object({ applied: z.boolean() }),
  execute: async ({ inputData, resumeData, suspend, bail }) => {
    if (!resumeData) {
      // 首轮挂起：suspend 载荷就是给审批人的申请单
      await suspend({ sql: inputData.sql, reason: inputData.reason })
      return { applied: false }        // 类型占位，不会执行到
    }
    if (!resumeData.approved) await bail()   // 拒绝：整条流程 failed
    if (!/\bwhere\b/i.test(inputData.sql))
      return { applied: false }        // 写语句无 WHERE 一律不执行
    executeFix(inputData.sql)          // 执行并写审计日志（略）
    return { applied: true }
  },
})

解读：首轮 resumeData 为空即挂起，suspend 载荷就是申请单；恢复后按 approved 分流，拒绝走 bail() 直接终结流程。写语句不带 WHERE 不执行、执行必留痕，与 Python 版 execute_fix 的规矩同款。

5.2、组装与恢复

组装成链，一次完整的挂起与恢复如下：

import { createWorkflow } from '@mastra/core/workflows'

export const fixWorkflow = createWorkflow({
  id: 'fix-workflow',
  inputSchema: z.object({ instruction: z.string() }),
  outputSchema: z.object({ applied: z.boolean() }),
})
  .then(planStep)
  .then(approvalStep)
  .commit()

const run = await fixWorkflow.createRun()
const result = await run.start({
  inputData: { instruction: '把订单 1024 的金额改为 199' },
})

if (result.status === 'suspended') {
  // 申请单在 suspendPayload；隔多久审批都行，状态在 storage 里
  const final = await run.resume({ resumeData: { approved: true } })
}

解读：挂起的是运行记录不是进程——HTTP 场景把 runId 与 suspendPayload 返回给前端，审批接口按 runId 重建运行句柄再 resume，隔天恢复也不丢（第 4 篇已验证跨进程恢复）。对照 Python 版把 interrupt 映射成 pending_approval 再用 Command(resume=...) 恢复：语义同构；差别是那边审批有 approve/edit/reject 三态，这边 approved 布尔加 bail 两个通道，窄而直白。

6、Next.js 聊天界面

外壳沿用第 12 篇的全套接线（Next.js 集成指南）：useChat 配 DefaultChatTransport 指向 /api/chat，route handler 用 handleChatStream 出流式响应（version: 'v7' 别漏），只把 agentId 换成 data-analyst。本节的增量是两张卡片，也是 TS 版对 Python 版的降维优势所在。

6.1、查询结果卡片

聊天流里工具调用的渲染按第 12 篇的 parts 模型：part.type 为 tool-sqlQueryTool 的分片交给卡片组件，三态渲染加 z.infer 推导 props：

import type { ToolUIPart } from 'ai'
import type { z } from 'zod'
import type { queryOutput } from '@/mastra/tools/sql'

type QueryResult = z.infer<typeof queryOutput>

export function QueryCard({ part }: { part: ToolUIPart }) {
  if (part.state === 'input-available') return <p>查询执行中……</p>
  if (part.state === 'output-error') return <p>查询失败</p>
  const out = part.output as QueryResult
  if (!out.ok) return <p className="warn">已拦截：{out.code}</p>
  return (
    <div className="query-card">
      <code>{out.sql}</code>
      <table>
        <tbody>
          {out.rows.slice(0, 5).map((row, i) => (
            <tr key={i}>
              {Object.values(row).map((v, j) => <td key={j}>{String(v)}</td>)}
            </tr>
          ))}
        </tbody>
      </table>
      <footer>共 {out.rowCount} 行，预览前 5 行</footer>
    </div>
  )
}

解读：part.output 的结构就是第 2 节那份 queryOutput——卡片 props 的类型由后端 schema 推导，护栏拦截的 ok: false 分支在组件里有专门呈现，改 schema 一处、卡片编译期立刻标红；import type 编译期擦除，服务端模块不会进浏览器包。Python 版同样有结构化输出，但卡片 props 类型要照 Pydantic 模型手抄一遍再对齐，这里只有一份事实。

6.2、审批卡片

审批流的界面入口独立于聊天：用户点"修正数据"提交需求，/api/fix 返回 { runId, payload }，前端渲染审批卡片：

'use client'

export function ApprovalCard({ runId, payload }: {
  runId: string
  payload: { sql: string; reason: string }
}) {
  const decide = async (approved: boolean) => {
    await fetch('/api/fix/resume', {
      method: 'POST',
      body: JSON.stringify({ runId, approved }),
    })
  }
  return (
    <div className="approval-card">
      <p>待审批的写操作：</p>
      <code>{payload.sql}</code>
      <p>{payload.reason}</p>
      <button onClick={() => decide(true)}>批准</button>
      <button onClick={() => decide(false)}>拒绝</button>
    </div>
  )
}

解读：按钮到恢复的整条链路在同一个仓库里完成——/api/fix/resume 按 runId 恢复第 5 节挂起的运行，结果回流聊天页。Python 版要做同样的事，得在 FastAPI 之外再立一个前端工程对接口径；TS 版只是同一个 app/ 目录下多两个文件，这就是第 12 篇"全栈一体"落进实战的样子。

7、Python 与 TS 终对照

两个系列在同一份需求上会师，把全部差异收进一张表：

代码量口径：Python 收官篇片段累计约 200 行（第 30 篇），不含前端工程；本篇后端约 220 行加两张卡片约 60 行，行数基本打平。差别在性质——TS 版每一行都在类型检查器视线里，Python 版的护栏正则与报告字段写错，多数要跑到运行时才露馅。

结论不玄：选栈不选信仰。主场在数据科学——pandas 清洗、notebook 协作、模型实验——Python 版顺理成章；Agent 要嵌进 Web 产品交付用户，前后端一体的 TS 版迭代速度压倒一切。两个系列 44 篇在此收拢成这条判断，第 14 篇会把决策线画全。

8、总结

回顾全文：需求四句话与 Agent 系列第 30 篇同款，六个组件全部来自本系列前 12 篇；SQL 工具用 Zod 参数约束加四层护栏，拦截结果以判别联合回传模型自动纠偏；报告契约强制结论带 SQL 与数据行；偏好记忆一个 workingMemory 配置盖住 Python 版三件手写；写操作走 suspend/resume 审批闸门；前端用 useChat 加两张 outputSchema 驱动的卡片收尾。

留一个练习：给 dataAgent 加一个 fact-check 工具——inputSchema 收 { claim, sql }，execute 里复用第 2 节的 guardSql 与 queryOutput 重跑查询，返回 { supported: boolean, evidence: string }，再在 instructions 里要求成稿前对每条结论自查。进阶做法是把逐条自查换成随行打分：按第 7 篇挂 createFaithfulnessScorer（sampling 取 0.2），分数进 Studio 对照看，对应 Agent 系列第 27 篇的评测与可观测。对于要把分析能力交付给业务同学的团队，建议就用这套一体结构起步：本地 LibSQL 加默认配置先跑通闭环，再按第 11、12 篇的部署章法上生产。下一篇是系列收官：何时选 TS 栈——把 13 篇攒下的证据收成一张决策表。