Text-to-SQL 与数据分析 Agent（语义层 / 指标口径 / SQL 安全）

自然语言问题
  ├── 意图识别（query / compare / drill-down / forecast）
  ├── 实体抽取（时间范围 / 维度 / 指标名）
  ▼
语义层
  ├── Schema Linking：表/字段候选 + 打分 + 消歧
  ├── 指标口径库：GMV = sum(paid_amount) where status='paid'
  ├── 业务词典：别名 / 同义词 / 黑话
  ▼
生成层
  ├── SQL 生成（few-shot + schema 裁剪 + 结构化输出）
  ├── 可选：pandas / Polars 后置计算
  ▼
信任层
  ├── SQL AST 校验（只读、表白名单、行数限制）
  ├── Dry-run + 超时 + cost 估算
  ├── 错误恢复（自修复 / 追问）
  ├── 高风险 HITL 审批（大表扫描 / 跨库 / 敏感字段）
  ▼
执行层（只读账号）
  ▼
解释与可视化
  ├── 查询解释（为什么查这些表、这样 join）
  ├── 图表推荐（折线 / 柱状 / 漏斗 / 表格 / KPI 卡）
  ├── 洞察摘要（环比、同比、异常点）
  ▼
前端流式渲染（与 ai-frontend-streaming-ux 联动）

# metrics/gmv.yaml
name: GMV
aliases: [销售额, 成交金额, 交易额]
definition: |
  支付成功的订单金额总和（不含退款）
sql: |
  SUM(CASE WHEN status='paid' THEN paid_amount ELSE 0 END)
filters:
  - status: paid
dimensions_allowed: [date, region, channel, category]
owner: data-team
updated_at: 2026-03-01

def nl2sql(question: str, ctx: DataCtx) -> SQLResult:
    intent = classify_intent(question)            # query/compare/trend
    tables = schema_link(question, ctx, top_k=8)
    metrics = metric_lookup(question)             # 命中口径直出模板
    prompt = build_prompt(question, tables, metrics, few_shots=ctx.few_shots)
    sql = llm_structured(prompt, schema=SQLPlan)  # JSON 输出
    return sql

"为回答'上周华东 GMV'，我查询了 orders 表与 regions 表，
通过 orders.region_id = regions.id 关联，
过滤 status='paid'（口径：GMV）
过滤 created_at in 2026-04-29 ~ 2026-05-05，
按 region='华东' 汇总。"

def recommend_chart(cols, rows):
    if len(rows) == 1 and len(cols) == 1: return "kpi"
    if has_time(cols) and len(numeric(cols)) >= 1: return "line"
    if len(categorical(cols)) == 1 and len(numeric(cols)) == 1:
        return "bar" if len(rows) <= 20 else "table"
    if is_funnel_pattern(cols, rows): return "funnel"
    return "table"

type DataAgentEvent =
 | { type: "plan"; intent: string; tables: string[] }
 | { type: "sql"; sql: string; explain: string; warning?: string }
 | { type: "approval_needed"; reason: string; est_rows: number }
 | { type: "rows"; preview: any[]; total: number }
 | { type: "chart"; spec: EChartsOption; insight: string }

CREATE ROLE ai_reader LOGIN PASSWORD '...' CONNECTION LIMIT 20;
GRANT USAGE ON SCHEMA public TO ai_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_reader;
REVOKE SELECT ON sensitive_users FROM ai_reader;   -- 黑名单
ALTER ROLE ai_reader SET statement_timeout = '5s';
ALTER ROLE ai_reader SET default_transaction_read_only = on;

import sqlglot
def validate_sql(sql: str, allowed_tables: set[str]) -> SQL:
    tree = sqlglot.parse_one(sql, dialect="postgres")
    if not isinstance(tree, sqlglot.exp.Select):
        raise Unsafe("只允许 SELECT")
    # 表白名单
    for t in tree.find_all(sqlglot.exp.Table):
        if t.name not in allowed_tables:
            raise Unsafe(f"表 {t.name} 不在白名单")
    # 拒绝危险函数
    for fn in tree.find_all(sqlglot.exp.Anonymous):
        if fn.this.lower() in DANGEROUS_FUNCS:
            raise Unsafe(f"禁用函数 {fn.this}")
    # 强制 LIMIT
    if not tree.args.get("limit"):
        tree = tree.limit(10000)
    return tree.sql(dialect="postgres")

# 召回
bm25_hits   = bm25(question, over="table+col names+descriptions")
vec_hits    = vec_search(question, top_k=30)
candidates  = rrf_merge(bm25_hits, vec_hits, k=20)
 
# 精排（可选 cross-encoder）
ranked      = cross_encoder_rerank(question, candidates, top_k=10)
 
# 消歧
if has_ambiguity(ranked):
    clarify  = ask_clarifying_question(ranked)   # "你说的用户是 customer 还是 employee？"

Table orders (电商订单表):
  id BIGINT PK
  user_id BIGINT FK -> users.id
  status TEXT  -- 'paid'/'unpaid'/'refunded'
  paid_amount NUMERIC  -- 实付金额（含税，不含运费）
  created_at TIMESTAMP
 
Relation: orders.user_id = users.id

name: GMV
def: 支付成功订单金额（不含退款、不含运费）
sql_expr: SUM(CASE WHEN status='paid' THEN paid_amount END)
cautions:
  - 不要用 total_amount（含未支付）
  - 退款走单独表 refunds，月末对齐时要单独扣

name: D1_Retention
def: |
  Day0 活跃用户在 Day1 再次活跃的比例
sql_template: |
  WITH d0 AS (
    SELECT DISTINCT user_id FROM events
    WHERE date = :date AND event_name = 'login'
  ),
  d1 AS (
    SELECT DISTINCT user_id FROM events
    WHERE date = :date + 1 AND event_name = 'login'
  )
  SELECT COUNT(d1.user_id)::float / NULLIF(COUNT(d0.user_id), 0) AS retention
  FROM d0 LEFT JOIN d1 USING(user_id)
cautions:
  - "活跃"定义必须和业务对齐：login / any_event / pay 三种口径差异很大

name: Checkout_Conversion
def: "加购 → 支付"转化
steps: [add_to_cart, start_checkout, paid]
window: 7d
cautions:
  - 窗口外的成交不算
  - 同一用户多次 funnel 取最后一次

name: LTV_90d
def: 用户首次付费后 90 天内累计付费金额
cautions:
  - 只算 paid 状态
  - 退款从 LTV 扣除
  - 新老用户要分开看