数据分析 Agent · Agent Atlas

数据分析 Agent 的风险不在「SQL 写错」，而在「写对了但读错、算错、把结果发给了不该看的人」。这个教程做一个只读、可校验、能出图的分析助手。

目标

bash

python -m data_agent "上个月各渠道的退款率是多少？按退款率从高到低排"

输出：

text

渠道        订单数   退款数   退款率
抖音         1,240     86     6.94%
天猫         2,310    102     4.42%
京东         1,870     61     3.26%

洞察：抖音退款率最高，主要来自「尺码不符」；建议优先检查该渠道的商品详情页。
图表：out/refund_rate.png

架构

text

问题
 → Schema 卡片（表、字段、含义、样例）
 → Schema linking（选相关表/字段）
 → 生成 SQL + 解释 + 图表类型
 → 静态校验（只读、单语句、限行）
 → 只读执行（超时、租户过滤）
 → 结果校验（空、异常值、行数）
 → 生成洞察 + 图表

示例数据库

sql

CREATE TABLE orders (
  id TEXT PRIMARY KEY,
  channel TEXT NOT NULL,
  user_id TEXT NOT NULL,
  amount REAL NOT NULL,
  status TEXT NOT NULL,       -- paid / shipped / completed / cancelled
  created_at TEXT NOT NULL
);
CREATE TABLE refunds (
  id TEXT PRIMARY KEY,
  order_id TEXT NOT NULL,
  reason TEXT,
  amount REAL NOT NULL,
  created_at TEXT NOT NULL
);

准备 1 万条 mock 数据即可。真实项目换成 PostgreSQL、ClickHouse 或数仓，思路不变。

Step 1：Schema 卡片

不要把整个 DDL 丢给模型。给每张表一个「卡片」：字段、类型、业务含义、枚举值、样例。

python

def schema_card(conn) -> str:
    lines = []
    for (table,) in conn.execute("SELECT name FROM sqlite_master WHERE type='table'"):
        cols = conn.execute(f"PRAGMA table_info({table})").fetchall()
        lines.append(f"表 {table}")
        for c in cols:
            lines.append(f"  - {c[1]} {c[2]}")
        sample = conn.execute(f"SELECT * FROM {table} LIMIT 2").fetchall()
        lines.append(f"  样例：{sample}")
    return "\n".join(lines)

更好的卡片还会包含：

字段的业务定义（status 的枚举含义）；
主键/外键；
时间字段的时区；
哪些字段含 PII；
常用指标的口径（退款率 = 退款订单数 / 支付订单数）。

Step 2：生成 SQL

python

import json

SQL_SCHEMA = {
    "type": "object",
    "properties": {
        "sql": {"type": "string"},
        "explanation": {"type": "string"},
        "chart": {"type": "string", "enum": ["bar", "line", "table"]},
        "x": {"type": ["string", "null"]},
        "y": {"type": ["string", "null"]},
    },
    "required": ["sql", "explanation", "chart"],
}

PROMPT = """你是数据分析师。根据 Schema 与问题生成 SQLite 查询。
规则：
1. 只生成一条 SELECT 或 WITH 查询。
2. 不要生成 INSERT/UPDATE/DELETE/DDL。
3. 时间范围要明确；「上个月」按当前日期 {today} 计算。
4. 金额用 SUM，比率注意分母为 0。
5. 结果列名用中文别名，便于展示。
6. 不确定口径时，在 explanation 里说明假设。

Schema：
{schema}

问题：{question}
输出 JSON：sql、explanation、chart、x、y。"""


def generate_sql(question: str, schema: str) -> dict:
    return llm.complete_json(
        system="只输出合法 JSON。",
        user=PROMPT.format(today=date.today().isoformat(), schema=schema, question=question),
        schema=SQL_SCHEMA,
    )

Step 3：静态校验

python

import re

FORBIDDEN = re.compile(
    r"\b(insert|update|delete|drop|alter|create|attach|detach|pragma|vacuum|replace|truncate)\b",
    re.IGNORECASE,
)


def validate_sql(sql: str, max_rows: int = 500) -> str:
    s = sql.strip().rstrip(";")
    if ";" in s:
        raise ValueError("只允许单条语句")
    if not re.match(r"^(select|with)\b", s, re.IGNORECASE):
        raise ValueError("只允许 SELECT/WITH")
    if FORBIDDEN.search(s):
        raise ValueError("包含禁止的关键字")
    if not re.search(r"\blimit\b", s, re.IGNORECASE):
        s = f"SELECT * FROM ({s}) LIMIT {max_rows}"
    return s

这仍然不是完整防御。生产必须：

数据库使用只读账号；
连接串 mode=ro；
行级权限/租户过滤；
查询超时；
结果行数与字节上限；
禁止访问系统表。

Step 4：只读执行

python

import sqlite3
import time


def run_sql(db_path: str, sql: str, timeout_s: float = 5.0, max_rows: int = 500):
    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    started = time.monotonic()

    def progress():
        return 1 if time.monotonic() - started > timeout_s else 0

    conn.set_progress_handler(progress, 1000)
    try:
        cur = conn.execute(sql)
        rows = cur.fetchmany(max_rows)
        return {
            "columns": [d[0] for d in cur.description],
            "rows": [dict(r) for r in rows],
            "truncated": len(rows) == max_rows,
        }
    finally:
        conn.close()

mode=ro 从连接层阻止写入；set_progress_handler 防止笛卡尔积跑死；fetchmany 防止一次拉回百万行。

Step 5：失败修复

SQL 报错时，把错误和 Schema 一起喂回去，只修复一次：

python

def ask(question: str, db_path: str) -> dict:
    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    schema = schema_card(conn)
    conn.close()

    plan = generate_sql(question, schema)
    try:
        sql = validate_sql(plan["sql"])
        result = run_sql(db_path, sql)
    except Exception as e:
        repair = generate_sql(
            f"{question}\n\n上一次 SQL 失败：{plan['sql']}\n错误：{e}\n请修正。",
            schema,
        )
        sql = validate_sql(repair["sql"])
        result = run_sql(db_path, sql)
        plan = repair

    if not result["rows"]:
        return {"answer": "查询没有返回数据，请确认时间范围或筛选条件。", "sql": sql, "result": result}

    insight = summarize(question, result)
    return {"answer": insight, "sql": sql, "result": result, "plan": plan}

只修复一次。两次还错，说明问题或 Schema 有问题，应该让人看。

Step 6：结果校验

代码能检查的不要交给模型：

python

def validate_result(result: dict) -> list[str]:
    problems = []
    if not result["rows"]:
        problems.append("空结果")
    if result["truncated"]:
        problems.append("结果被截断，可能不是全量")
    for row in result["rows"][:20]:
        for k, v in row.items():
            if isinstance(v, float) and (v != v or abs(v) > 1e12):
                problems.append(f"{k} 数值异常：{v}")
    return problems

指标类问题还要检查：

分母是否为 0；
比率是否在 0–1；
时间范围是否覆盖完整月份；
退款数是否大于订单数（口径错误）。

Step 7：洞察与图表

python

def summarize(question: str, result: dict) -> str:
    table = json.dumps(result["rows"][:20], ensure_ascii=False)
    return llm.complete_json(
        system="""你是数据分析师。只根据查询结果写三条洞察。
每条必须引用具体数字；不要编造原因；不确定时写「需要进一步验证」。
输出 JSON：{"insights": ["...", "...", "..."]}""",
        user=f"问题：{question}\n结果：{table}",
        schema={"type": "object", "properties": {"insights": {"type": "array", "items": {"type": "string"}}}, "required": ["insights"]},
    )["insights"]

图表用 matplotlib：

python

import matplotlib.pyplot as plt

def chart(result: dict, x: str, y: str, out: str = "out/chart.png") -> str:
    xs = [str(r[x]) for r in result["rows"]]
    ys = [float(r[y]) for r in result["rows"]]
    plt.figure(figsize=(8, 4.5))
    plt.bar(xs, ys, color="#0e6b5d")
    plt.xticks(rotation=30, ha="right")
    plt.tight_layout()
    plt.savefig(out, dpi=150)
    return out

图表类型由模型建议，但最终由代码校验：x/y 必须存在于结果列中，y 必须是数值。

Step 8：评估

构造 30 个问题，每条有：

标准 SQL 或标准结果；
期望的列；
允许的误差。

指标：

指标	说明
Execution Accuracy	结果集与标准结果一致
Column Accuracy	列名/顺序是否符合预期
Empty Rate	空结果比例
Repair Rate	需要修复的比例
Insight Faithfulness	洞察是否只使用结果中的数字
P95 延迟	生成 + 执行

结果集比较要处理浮点误差、行顺序、列别名。不要只比较 SQL 字符串——同一个结果可以有很多写法。

安全清单

只读账号 + mode=ro
只允许 SELECT/WITH
单语句、禁止注释绕过
行数、字节、超时限制
租户/用户过滤在 SQL 层
结果中的 PII 脱敏
不把数据库连接串/密钥放进上下文
高风险导出走审批

让模型直接连生产库

即使提示词写「只读」，模型也可能生成写语句。权限必须在数据库层，不在提示词里。

常见失败

口径不一致

「退款率」按订单数还是金额？必须在 Schema 卡片里写清楚，否则数字对不上。

时间边界错

「上个月」按自然月还是滚动 30 天？让模型显式写出假设。

分母为 0

比率出现 NaN/Inf。SQL 里用 NULLIF，结果校验兜底。

结果太大

一次拉回 10 万行，上下文爆炸。强制 LIMIT + 聚合。

洞察编原因

数字对，但「因为尺码问题」是编的。洞察只能引用结果，原因需要额外证据。

验收清单

Schema 卡片包含业务含义与口径
SQL 只读、单语句、限行、超时
执行失败只修复一次
结果有数值/空值/截断校验
洞察只引用结果中的数字
图表 x/y 经过校验
30 条评估集，报告执行准确率
PII 与租户隔离

小结

数据分析 Agent 的核心是「口径 + 只读 + 校验」，不是 SQL 生成。
Schema 卡片决定模型能否理解业务。
权限必须在数据库层，不能靠提示词。
洞察只能引用查询结果，原因需要额外证据。

下一篇是一组可以直接抄的提示词模板。

#目标

#架构

#示例数据库

#Step 1：Schema 卡片

#Step 2：生成 SQL

#Step 3：静态校验

#Step 4：只读执行

#Step 5：失败修复

#Step 6：结果校验

#Step 7：洞察与图表

#Step 8：评估

#安全清单

#常见失败

#验收清单

#小结

相关阅读

目标

架构

示例数据库

Step 1：Schema 卡片

Step 2：生成 SQL

Step 3：静态校验

Step 4：只读执行

Step 5：失败修复

Step 6：结果校验

Step 7：洞察与图表

Step 8：评估

安全清单

常见失败

验收清单

小结