第 21 章 · 实战三：数据问答 BI Agent
本章目标：综合运用工具、Workflow 与 RAG 所学，构建一个"自然语言查数出图"的 BI Agent——Text-to-SQL 生成、只读安全护栏、图表产出与防幻觉重试，一条流水线跑通。
21.1 需求分析与架构设计
业务同学最怕两件事：等数据、写 SQL。这个 BI Agent 让他们直接用中文提问：
"上个月华南区的销售额是多少？环比怎么样？画个柱状图。"
Agent 返回：SQL 执行结果 + 图表 + 一句人话结论。架构如下：
text
用户问题（自然语言）
│
▼
┌────────────────────┐
│ Step1 generateSql │ Text-to-SQL Agent
│ （注入表结构 DDL） │ 只生成 SELECT
└───────┬────────────┘
▼
┌────────────────────┐ SQL 非法 / 报错
│ Step2 validate │──────────┐
│ （只读护栏校验） │ │ 回传错误重试一次
└───────┬────────────┘◀─────────┘
▼
┌────────────────────┐ 空结果 → 引导改问法
│ Step3 executeQuery │
│ （libsql 只读执行）│
└───────┬────────────┘
▼
┌────────────────────┐
│ Step4 visualize │ chart 工具产出 ECharts 配置
└───────┬────────────┘
▼
┌────────────────────┐
│ Step5 summarize │ 结论解读 Agent（人话总结）
└────────────────────┘与第 19 章客服 Agent 的区别：这条链路路径完全可预测，所以用 Workflow 编排而非单个大 Agent——每个环节 schema 强约束，SQL 错了能定点重试，不会整链报废。
21.2 Schema 注入：让模型知道表长什么样
Text-to-SQL 的准确率取决于模型对表结构的了解。两种注入方式：
typescript
// src/mastra/bi/schema.ts —— 把核心表的 DDL 作为常量维护
export const SALES_DDL = `
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
region TEXT NOT NULL, -- 大区：华东/华南/华北/西南
amount REAL NOT NULL, -- 订单金额（元）
status TEXT NOT NULL, -- paid/refunded/pending
created_at TEXT NOT NULL -- ISO8601，如 '2026-08-01T10:00:00Z'
);
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT NOT NULL, -- 品类：数码/家居/服饰
price REAL NOT NULL
);
`;小库（几张表）直接把 DDL 拼进 instructions；表多了几十张就改走 RAG——把每张表的 DDL 和字段说明做成文档入库，检索相关表再注入（复用第 12 章方案）。本项目只有两张表，选前者：
typescript
// src/mastra/agents/sql-agent.ts —— Text-to-SQL 专家
import { Agent } from '@mastra/core/agent';
import { SALES_DDL } from '../bi/schema';
export const sqlAgent = new Agent({
id: 'sql-generator',
name: 'SqlGenerator',
model: 'openai/gpt-5-mini', // Text-to-SQL 是成熟任务，小模型足够
instructions: `
你是 SQL 专家，根据用户问题生成 SQLite 查询语句。
表结构如下：
${SALES_DDL}
规则：
1. 只输出一条 SELECT 语句，禁止任何修改数据的语句；
2. 时间过滤用 created_at 字符串比较（ISO8601 可直接比较）；
3. 必须带 LIMIT，默认 LIMIT 100；
4. 只输出 SQL 本身，不要解释、不要 markdown 代码块标记。
`,
});21.3 queryDatabase 工具：只读是底线
BI 场景的第一红线是绝不能改生产数据。防线有三层：数据库用只读账号（应用层兜底）、语句白名单校验、强制 LIMIT：
typescript
// src/mastra/tools/query-database.ts
import { createTool } from '@mastra/core/tools';
import { createClient } from '@libsql/client';
import { z } from 'zod';
// 生产环境此 URL 应指向只读副本账号，即使校验被绕过也无权限写
const db = createClient({ url: process.env.BI_DB_URL! });
// 白名单：只允许 SELECT；黑名单关键词双保险
const FORBIDDEN = /\b(drop|delete|update|insert|alter|attach|pragma)\b/i;
export const queryDatabaseTool = createTool({
id: 'query-database',
description: '在销售数据库上执行只读 SELECT 查询并返回行数据',
inputSchema: z.object({
sql: z.string().describe('要执行的 SQLite SELECT 语句'),
}),
outputSchema: z.object({
rows: z.array(z.record(z.unknown())),
rowCount: z.number(),
}),
execute: async ({ inputData }) => {
const { sql } = inputData;
const trimmed = sql.trim();
// 护栏一：必须以 SELECT 开头（含 CTE 时以 WITH 开头也放行）
if (!/^(select|with)\b/i.test(trimmed)) {
throw new Error('拒绝执行：仅允许 SELECT/WITH 开头的只读查询');
}
// 护栏二：黑名单关键词
if (FORBIDDEN.test(trimmed)) {
throw new Error('拒绝执行：包含被禁止的关键词');
}
// 护栏三：无 LIMIT 则强制追加，防止全表扫描拖垮库
const safeSql = /\blimit\b/i.test(trimmed) ? trimmed : `${trimmed.replace(/;$/, '')} LIMIT 100`;
try {
const rs = await db.execute(safeSql);
return { rows: rs.rows as Record<string, unknown>[], rowCount: rs.rows.length };
} catch (err) {
// 把数据库原始报错抛给上游——Workflow 会用它触发重试
throw new Error(`SQL 执行失败：${(err as Error).message}`);
}
},
});21.4 chart 工具：把行数据变成图
前端渲染场景下返回 ECharts 配置 JSON 最灵活；需要落文件时也可服务端渲染成 PNG。这里实现配置生成：
typescript
// src/mastra/tools/chart-tool.ts
import { createTool } from '@mastra/core/tools';
import { z } from 'zod';
export const buildChartTool = createTool({
id: 'build-chart',
description: '把查询结果转成 ECharts 柱状图/折线图/饼图配置',
inputSchema: z.object({
title: z.string().describe('图表标题'),
type: z.enum(['bar', 'line', 'pie']).describe('图表类型'),
xAxisData: z.array(z.string()).describe('X 轴类目（或饼图的名称序列）'),
seriesName: z.string().describe('数据系列名'),
values: z.array(z.number()).describe('数值序列'),
}),
outputSchema: z.object({
option: z.record(z.unknown()), // 直接可喂给 echarts.setOption 的对象
}),
execute: async ({ inputData }) => {
const { title, type, xAxisData, seriesName, values } = inputData;
if (type === 'pie') {
// 饼图的数据形态不同：[{name, value}] 数组
const data = xAxisData.map((name, i) => ({ name, value: values[i] }));
return {
option: {
title: { text: title },
tooltip: { trigger: 'item' },
series: [{ type: 'pie', radius: '60%', data }],
},
};
}
return {
option: {
title: { text: title },
tooltip: { trigger: 'axis' },
xAxis: { type: 'category', data: xAxisData },
yAxis: { type: 'value' },
series: [{ name: seriesName, type, data: values }],
},
};
},
});21.5 Workflow 编排：五个步骤 + 失败重试
现在把所有环节串成 Workflow。注意 generateSqlStep 内部带一次重试：SQL 报错时把错误信息回传给 Agent 修正，这是治理 LLM 幻觉最朴素有效的手段：
typescript
// src/mastra/workflows/bi-workflow.ts（节选核心三步）
import { createStep, createWorkflow } from '@mastra/core/workflows';
import { z } from 'zod';
import { sqlAgent } from '../agents/sql-agent';
import { queryDatabaseTool } from '../tools/query-database';
// Step1+2 合并视图：生成 SQL（内部含一次纠错重试）
const generateSqlStep = createStep({
id: 'generate-sql',
inputSchema: z.object({ question: z.string() }),
outputSchema: z.object({ sql: z.string(), attempts: z.number() }),
execute: async ({ inputData, mastra }) => {
let lastError = '';
for (let attempt = 1; attempt <= 2; attempt++) {
const prompt = attempt === 1
? `用户问题：${inputData.question}`
// 重试时附带上次报错，让模型针对性修正
: `用户问题：${inputData.question}\n\n上次生成的 SQL 报错：${lastError}\n请修正后重新输出。`;
const res = await mastra!.getAgent('sql-generator').generate([{ role: 'user', content: prompt }]);
const sql = res.text.trim();
try {
// 用轻量执行做预检（LIMIT 1），语法/字段错误在此暴露
await queryDatabaseTool.execute!({
context: { inputData: { sql: sql.replace(/\blimit\s+\d+\b/i, 'LIMIT 1') } },
tracingContext: undefined as never,
} as never);
return { sql, attempts: attempt };
} catch (e) {
if (attempt === 2) throw e; // 两次都失败才真正抛出
lastError = (e as Error).message;
}
}
throw new Error('unreachable');
},
});
// Step3：正式执行查询
const executeStep = createStep({
id: 'execute-query',
inputSchema: z.object({ sql: z.string(), attempts: z.number() }),
outputSchema: z.object({
rows: z.array(z.record(z.unknown())),
isEmpty: z.boolean(),
sql: z.string(),
}),
execute: async ({ inputData }) => {
if (!/\blimit\b/i.test(inputData.sql)) inputData.sql += ' LIMIT 100';
const result = await queryDatabaseTool.execute!({
context: { inputData: { sql: inputData.sql } },
tracingContext: undefined as never,
} as never);
return {
rows: result.rows,
isEmpty: result.rows.length === 0, // 空结果标记，下游引导用户换问法
sql: inputData.sql,
};
},
});visualizeStep 调用 buildChartTool 把行数据转成图表配置，summarizeStep 用一个小模型把数字翻译成人话（"上月华南区销售额 ¥48.6 万，环比 +12%"），空结果时输出引导话术而不是硬编一句"没有数据"。组装方式与第 7 章一致：
typescript
// 同文件末尾：五步串联
export const biWorkflow = createWorkflow({
id: 'bi-workflow',
inputSchema: z.object({ question: z.string() }),
outputSchema: z.object({
summary: z.string(),
chartOption: z.record(z.unknown()),
sql: z.string(),
}),
})
.then(generateSqlStep)
.then(executeStep)
.then(visualizeStep)
.then(summarizeStep)
.commit();本章小结
- 路径可预测的 BI 流程用 Workflow 编排，比单一大 Agent 可控得多；
- Schema 注入是 Text-to-SQL 准确率的基础：小库拼 DDL 进 instructions，大库走 RAG 检索；
- 只读三层防线：只读账号兜底 + SELECT/WITH 白名单 + 黑名单关键词 + 强制 LIMIT；
- 失败回传重试：SQL 报错信息喂回模型修正一次，是治幻觉性价比最高的手段；
- 空结果不硬编，引导用户改问法才是好的产品体验。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. BI 流程为什么选择 Workflow 编排而不是一个大 Agent 自主决策？
2. queryDatabase 工具的只读防护中，哪一层是"最后兜底"？
3. SQL 执行报错后，系统如何处理以提高最终成功率？
4. 查询返回 0 行数据时，体验良好的做法是什么？
🛠️ 动手实践
- 给
orders表增加一张关联的order_items明细表，更新 DDL 注入后测试"客单价最高的三个品类"这类多表 JOIN 问题是否还能答对。 - 把
queryDatabaseTool改造为同时支持"查询超时自动降级到只读副本"的重试逻辑，并用日志记录每次降级事件。 - 在前端用 ECharts 渲染
biWorkflow返回的chartOption，并为饼图模式补充"占比低于 2% 自动合并为其他"的逻辑。
下一章：返回课程导学