第 29 章 实战项目三：数据分析助手（Text2SQL）

29.1 需求与场景

数据分析助手面向企业内部分析师、运营与产品人员，以自然语言查询关系型数据库。用户输入中文业务问题（如"上个月华东地区订单金额 Top10 的品类是哪些"），系统自动生成 SQL、执行、返回表格结果，并以自然语言解释业务含义与给出可视化建议。流程以单图线性循环实现，结构简单、易调试，展示"简单任务不要过度工程化"的架构原则。

典型使用路径：分析师在 BI 聊天框输入问题；系统加载相关表结构（含字段注释与示例行）；LLM 生成 SQL；经过安全校验层确认是只读查询后在只读账号上执行；成功则将结果集转为自然语言回答并给出图表建议，失败则将数据库错误信息回喂 LLM 自动修正，最多 3 轮；超过上限则给出友好错误并建议改写问题或联系数据团队。

功能需求：数据库元数据加载（表结构、列名、注释、示例数据、主外键关系）；SQL 生成（结合问题与 schema 提示，结构化输出）；SQL 校验与安全检查（禁止 DROP/DELETE/UPDATE/INSERT/ALTER/TRUNCATE/CREATE 等写操作）；只读连接执行 SQL（带 statement_timeout）；错误自纠正（SQL 报错返回 LLM 修正，最多 3 次）；结果可视化建议（按数据特征推荐折线/柱状/饼图/表格/KPI 卡片）；自然语言回答；查询历史记录写入 Store 供后续相似问题召回。

非功能需求：单查询端到端 P95 < 8 秒（不含慢查询触发超时报错的路径）；强制只读，禁止写操作与跨库访问；单次返回行数 ≤ 1000；错误自纠正成功率 ≥ 80%；每一次问题、SQL、执行结果、token 消耗均可审计；支持同时连接多个数据库实例（analytics/marketing/finance）通过 configurable 切换。

为什么选择单图而非多 Agent：Text2SQL 任务线性明确，schema 加载 → SQL 生成 → 校验 → 执行 → 解释是强顺序，不存在并行路径与角色分工；多 Agent 只会引入额外的通信开销、状态同步复杂度与调试成本。这一选择体现 LangGraph 设计理念——图的节点粒度由任务本身决定，不必为了"Agent"概念过度工程化。对于复杂场景（如跨库联合分析、多步推理、自动可视化），可在单图基础上新增节点扩展，而无需一开始就构建多 Agent 体系。

29.2 架构设计

┌────────┐  HTTP/SSE  ┌─────────────┐   ┌───────────────────────────────────────────────┐
│ Client │ ─────────▶ │  FastAPI    │   │  Text2SQL Graph (单 StateGraph)                │
│ BI/IM  │ ◀───────── │  /ask/stream│   │                                               │
└────────┘            └─────────────┘   │  START → load_schema → generate_sql           │
                                        │                         │                     │
                                        │                         ▼                     │
                                        │                    validate_sql                │
                                        │                      │       │                │
                                        │                   不通过    通过                │
                                        │                      │       │                │
                                        │                      ▼       ▼                │
                                        │                 error_cor? execute_sql         │
                                        │                      │       │                │
                                        │             attempts<3?│   成功/失败            │
                                        │                ┌Yes   │   │   │               │
                                        │                │      ▼   ▼   │               │
                                        │                └──▶generate_sql (loop)        │
                                        │                │No       │                    │
                                        │                ▼         ▼                    │
                                        │           friendly_err explain_result ─▶ END  │
                                        └──────────────────┬────────────────────────────┘
                                                           │
                                            ┌──────────────┼──────────────┐
                                            ▼              ▼              ▼
                                      ┌──────────┐  ┌──────────┐  ┌───────────┐
                                      │Memory/PG │  │ Readonly │  │ Postgres  │
                                      │Saver     │  │ PG DB    │  │ Store     │
                                      └──────────┘  └──────────┘  └───────────┘

技术选型：单 StateGraph + 条件路由循环；结构化输出以 Pydantic 模型 SQLQuery 校验 LLM 输出；不引入多 Agent 编排，节点线性串联，便于调试与性能优化；MemorySaver 即可满足单次查询 checkpoint 需求（查询在秒级完成，无需跨断点持久化）；查询历史写入 PostgresStore（namespace=("sql_history", user_id)）供相似问题召回与运营审计；模型选择 gpt-4o 以获得更好的 SQL 生成准确率，简单查询可降级 gpt-4o-mini。

条件路由循环机制说明：error_correction 节点只做计数器自增，不直接连回 generate_sql，而是通过 should_revise 条件边判断 retry/fail 两个分支，这样做的好处是循环边界显式化，每一轮都会经过 validate_sql 与 execute_sql，不会绕过安全校验；attempts 字段也会被 checkpoint 保存，即使在循环中中断（例如 execute_sql 超时触发 interrupt 等待用户确认"是否继续"），恢复后也能准确知道当前轮次而不会重置计数。对于希望让用户在长 SQL 执行前确认的场景，可以在 validate_sql 和 execute_sql 之间增加 await_user_confirm 节点调用 interrupt，用户确认后再 resume 到 execute_sql。

数据库连接池配置要点：psycopg 的 ConnectionPool 应在应用启动时初始化一次并复用，避免每个请求新建连接池；pool_size 按 worker 数 × 每 worker 峰值并发设置，典型值为 10-20；只读账号的 statement_timeout 通过连接串 options 传递让数据库层强制超时，这是安全兜底，应用层 asyncio.wait_for 不能替代，因为 Python 侧超时无法中断数据库侧正在执行的查询。

项目目录结构：

text2sql/
├── app/
│   ├── main.py              # FastAPI + SSE
│   ├── graph.py             # Text2SQL 图构建
│   ├── state.py
│   ├── nodes/
│   │   ├── schema_loader.py
│   │   ├── sql_gen.py
│   │   ├── validator.py
│   │   ├── executor.py
│   │   ├── corrector.py
│   │   └── explainer.py
│   ├── tools/
│   │   └── db.py            # 只读连接池与 schema 读取
│   └── utils/
│       ├── security.py      # SQL 安全校验
│       └── prompts.py
├── tests/
│   ├── test_security.py
│   ├── test_error_correction.py
│   ├── test_golden_set.py
│   └── test_concurrency.py
├── docker-compose.yml
├── requirements.txt
└── .env

依赖安装命令：

pip install "langgraph>=1.2,<1.3" langchain langchain-openai \
    psycopg[binary,pool] fastapi uvicorn sqlparse pydantic pyyaml

Schema 检索策略：当数据库表数量较少（< 15 张）时直接将全部 DDL 塞入 schema_info，模型准确率最高；当表数量较大（15-100 张）时，先对问题做关键词提取，再从 schema_embeddings 表（预先存储表/列注释的 embedding）中召回 Top-K 相关表（默认 K=8），仅将相关表 DDL 注入 prompt；当表数量超大（> 100 张）时采用分阶段策略，先让 LLM 根据表名与列名粗选候选表，再注入候选表详细 DDL 生成最终 SQL。schema_embeddings 的更新由 DDL 迁移脚本触发，保持与数据库结构一致。

29.3 State 与节点设计

# state.py
from typing import Optional, List, Dict, Any
from typing_extensions import TypedDict

class SQLState(TypedDict):
    question: str
    db_name: str
    schema_info: str               # DDL + 注释 + 示例行 + JOIN 提示
    generated_sql: str
    validation_error: str
    query_result: List[Dict[str, Any]]
    answer: str
    chart_suggestion: str         # line/bar/pie/table/kpi
    attempts: int
    last_db_error: str

节点职责：

• load_schema：首次调用时从数据库 INFORMATION_SCHEMA 读取表结构、字段注释、每表 3 行脱敏示例数据与主外键关系，组装为 schema_info；写入 Store namespace=("schema", db_name) 做缓存（TTL 1 小时）。
• generate_sql：LLM 根据 question + schema_info 输出 SQLQuery（含 reasoning 推理过程与 sql 字段），通过 Pydantic 校验结构。
• validate_sql：正则 + sqlparse 词法分析检查危险关键字、强制只读前缀、校验是否查询了授权表白名单、强制或追加 LIMIT 1000。
• execute_sql：使用只读数据库账号执行，连接串设置 statement_timeout=10s；异常捕获并将数据库错误信息回填 last_db_error；成功则 query_result 为 list[dict]，列名取自 cursor.description。
• error_correction：attempts += 1，记录轮次。
• should_retry：attempts < 3 且存在 validation_error 或 last_db_error 时回到 generate_sql，同时把错误信息作为额外 user message 注入；否则返回友好错误。
• explain_result：将 query_result 转为自然语言回答，根据字段类型与数据分布给出图表建议；若失败超限直接输出友好错误。

Pydantic 输出模型与图构建：

# graph.py
from pydantic import BaseModel, Field
class SQLQuery(BaseModel):
    reasoning: str = Field(description="生成 SQL 的推理过程，说明用到了哪些表与 JOIN 关系")
    sql: str = Field(description="最终可执行的只读 SQL 语句")

from langgraph.graph import StateGraph, START, END
from .state import SQLState
from .nodes.schema_loader import load_schema
from .nodes.sql_gen import generate_sql
from .nodes.validator import validate_sql
from .nodes.executor import execute_sql
from .nodes.corrector import error_correction, should_retry
from .nodes.explainer import explain_result

builder = StateGraph(SQLState)
builder.add_node("load_schema", load_schema)
builder.add_node("generate_sql", generate_sql)
builder.add_node("validate_sql", validate_sql)
builder.add_node("execute_sql", execute_sql)
builder.add_node("error_correction", error_correction)
builder.add_node("explain_result", explain_result)

builder.add_edge(START, "load_schema")
builder.add_edge("load_schema", "generate_sql")
builder.add_edge("generate_sql", "validate_sql")
builder.add_conditional_edges("validate_sql",
    lambda s: "execute_sql" if not s.get("validation_error") else "error_correction",
    {"execute_sql":"execute_sql","error_correction":"error_correction"})
builder.add_conditional_edges("execute_sql",
    lambda s: "explain_result" if not s.get("last_db_error") else "error_correction",
    {"explain_result":"explain_result","error_correction":"error_correction"})
builder.add_conditional_edges("error_correction", should_retry,
    {"retry":"generate_sql","fail":"explain_result"})
builder.add_edge("explain_result", END)

29.4 Schema 提示、安全与错误自纠正

Schema 提示工程是 Text2SQL 准确率的关键。load_schema 组装四部分信息：CREATE TABLE DDL（含列类型、NOT NULL、默认值）、COMMENT ON 字段注释、每表 3 行示例数据（对金额、姓名等敏感字段做脱敏替换）、常见 JOIN 关系提示（外键与常用业务关联）。当库表数量超过 30 张时，根据问题关键词对表/列注释做 embedding 向量检索，仅将相关表纳入 schema_info，避免 prompt 过长导致模型忽略重要信息。

示例 schema 片段：

-- orders 订单表
CREATE TABLE orders (
  id BIGSERIAL PRIMARY KEY,
  user_id BIGINT NOT NULL,
  amount NUMERIC(12,2) NOT NULL,
  status VARCHAR(32) NOT NULL,
  region VARCHAR(32),
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
COMMENT ON COLUMN orders.status IS '订单状态: created/paid/shipped/done/refunded';
COMMENT ON COLUMN orders.region IS '地区: 华东/华北/华南/西南/其他';
-- 示例行:
-- (1001, 8801, 299.00, 'done', '华东', '2026-01-10 10:00:00+08')
-- (1002, 8802, 129.00, 'shipped', '华北', '2026-01-11 14:20:00+08')
-- 相关 JOIN: orders.user_id = users.id; orders.id = payments.order_id

安全机制采用五层防线。第一层数据库层：使用独立只读账号，收回 INSERT/UPDATE/DELETE/TRUNCATE 等权限，按 schema 级别授权；连接串通过 options 参数设置 statement_timeout=10s 防止慢查询拖垮数据库。第二层 SQL 层：validate_sql 使用正则与词法分析组合校验，关键字白名单仅允许 SELECT/WITH/SHOW/EXPLAIN 开头；拦截 DROP/DELETE/UPDATE/INSERT/ALTER/TRUNCATE/CREATE/GRANT/REVOKE 等关键字；校验 FROM 后的表名必须在授权白名单内；若 SQL 未带 LIMIT 则自动追加 LIMIT 1000。第三层结果层：在 execute_sql 中使用 cur.fetchmany(1000) 截断返回行数。第四层 schema 屏蔽：password、id_card、phone、email 等敏感列在 schema 加载阶段直接从 DDL 中删除列名并在示例数据中替换为 ***，不暴露给 LLM。第五层审计层：将每次 question、generated_sql、executed_sql、query_result 行数、延迟、attempts 写入结构化日志与 LangSmith，便于事后回溯。

SQL 校验函数实现：

# utils/security.py
import re
FORBIDDEN = re.compile(
    r"\b(DROP|DELETE|UPDATE|INSERT|ALTER|TRUNCATE|CREATE|GRANT|REVOKE|VACUUM|COPY)\b",
    re.IGNORECASE)
READONLY_PREFIX = re.compile(r"^\s*(SELECT|WITH|SHOW|EXPLAIN)\b", re.IGNORECASE)
FROM_TABLE = re.compile(r"\bFROM\s+([a-zA-Z_][a-zA-Z0-9_]*)", re.IGNORECASE)
JOIN_TABLE = re.compile(r"\bJOIN\s+([a-zA-Z_][a-zA-Z0-9_]*)", re.IGNORECASE)
SENSITIVE_COLS = {"password","passwd","id_card","idcard","phone","mobile","email"}

def sanitize_schema(ddl: str) -> str:
    """在 schema 文本中屏蔽敏感列。"""
    for col in SENSITIVE_COLS:
        ddl = re.sub(rf"\b{col}\b\s+[A-Z]+", f"{col}_masked VARCHAR(32)", ddl, flags=re.I)
    return ddl

def validate_sql(sql: str, allowed_tables: set) -> tuple[str, str]:
    """返回 (possibly_rewritten_sql, error_message)，error_message 为空表示通过。"""
    if not READONLY_PREFIX.match(sql):
        return sql, "SQL 必须以 SELECT/WITH/SHOW/EXPLAIN 开头"
    if FORBIDDEN.search(sql):
        return sql, "SQL 包含写操作或 DDL 关键字，已被安全校验拦截"
    # 表名白名单
    refs = {m.group(1).lower() for m in FROM_TABLE.finditer(sql)} | \
           {m.group(1).lower() for m in JOIN_TABLE.finditer(sql)}
    for t in refs:
        if t not in allowed_tables:
            return sql, f"查询了未授权的表: {t}"
    # 自动追加 LIMIT
    if "limit" not in sql.lower():
        sql = sql.rstrip(";").rstrip() + " LIMIT 1000;"
    return sql, ""

数据库连接与执行：

# tools/db.py
from psycopg_pool import ConnectionPool

READONLY_URL = ("postgresql://readonly:ro_pwd@db:5432/analytics"
                "?options=-c%20statement_timeout%3D10000")
read_pool = ConnectionPool(READONLY_URL, max_size=10)
ALLOWED_TABLES = {"orders","users","products","sessions","payments","categories"}

def execute_readonly(sql: str):
    with read_pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(sql)
            cols = [c.name for c in cur.description]
            rows = cur.fetchmany(1000)
            return [dict(zip(cols, r)) for r in rows]

def load_schema_text() -> str:
    """从数据库读取 DDL、注释、示例行并拼接（实现略）。"""
    ...

generate_sql 节点在重试轮次会附加错误信息作为额外 user message，给模型足够的修复线索：

# nodes/sql_gen.py
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from ..utils.prompts import load_prompt
from ..graph import SQLQuery

async def generate_sql(state: SQLState):
    llm = init_chat_model("openai:gpt-4o").with_structured_output(SQLQuery)
    messages = [
        SystemMessage(content=load_prompt("sql_gen").format(
            schema_info=state["schema_info"])),
        HumanMessage(content=state["question"])]
    if state.get("attempts",0) > 0 and state.get("last_db_error"):
        messages.append(HumanMessage(
            content=f"上一次执行 SQL 报错：{state['last_db_error']}。请结合 schema 修正后重新生成。"))
    if state.get("validation_error"):
        messages.append(HumanMessage(
            content=f"上一次 SQL 未通过安全校验：{state['validation_error']}。请修正后重新生成。"))
    q: SQLQuery = await llm.ainvoke(messages)
    return {"generated_sql": q.sql,
            "validation_error": "",
            "last_db_error": ""}

错误自纠正机制将数据库错误原文（如 Postgres 返回的 ERROR: column "amout" does not exist LINE 1: ... 或 operator does not exist: character varying = integer）直接回填给 LLM。列名拼写错误、类型不匹配、遗漏 GROUP BY、JOIN 条件缺失等常见问题通常 1-2 轮即可修复；涉及业务逻辑误解或 schema 缺失的问题可能 3 轮都无法修复，此时应友好失败并提示用户换种问法或联系数据团队，而不是无限循环。

# nodes/corrector.py
async def error_correction(state: SQLState):
    return {"attempts": state.get("attempts",0) + 1}

def should_retry(state):
    if state.get("attempts",0) < 3 and (
            state.get("validation_error") or state.get("last_db_error")):
        return "retry"
    return "fail"

# nodes/executor.py
import asyncio
async def execute_sql(state: SQLState):
    from ..tools.db import execute_readonly
    try:
        rows = await asyncio.to_thread(execute_readonly, state["generated_sql"])
        return {"query_result": rows, "last_db_error": ""}
    except Exception as e:
        return {"last_db_error": str(e)[:2000], "query_result": []}

explain_result 节点除自然语言回答外，还给出可视化建议。图表类型推荐规则：结果包含时间序列字段（date/day/month）+ 数值字段 → line（折线）；包含分类字段（category/region）+ 数值 → bar（柱状）；分类字段 ≤ 5 且数值合计为占比 → pie（饼图）；结果集行列都较小且数值需精确对比 → table；结果为单个数值（如"本月 GMV"）→ kpi；其他 → none。explain 节点让 LLM 结合上述规则与数据内容决策，输出结构化 AnswerOut。

Few-shot 示例注入：为提升 SQL 生成准确率，schema_info 末尾可附带 3-5 组用户问题 → 正确 SQL 的 Few-shot 示例，示例从历史审计库中按问题相似度召回。示例选择使用 embedding 相似度（pgvector），距离 < 0.3 的示例才注入，避免低相关示例误导模型。

完整工具模块示例：

# tools/db.py
from psycopg_pool import ConnectionPool
from psycopg.rows import dict_row
import psycopg

READONLY_URL = ("postgresql://readonly:ro_pwd@db:5432/analytics"
                "?options=-c%20statement_timeout%3D10000")
read_pool = ConnectionPool(READONLY_URL, max_size=10, open=False)
read_pool.open()
ALLOWED_TABLES = {"orders","users","products","sessions","payments","categories"}
SENSITIVE_COLS = {"password","passwd","id_card","idcard","phone","mobile","email"}

def execute_readonly(sql: str):
    with read_pool.connection() as conn:
        with conn.cursor(row_factory=dict_row) as cur:
            cur.execute(sql)
            rows = cur.fetchmany(1000)
            return list(rows)

def load_schema_text(db_name: str = "analytics") -> str:
    """加载指定库的 DDL、注释、示例行、外键关系，拼成 schema_info。"""
    with read_pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
              SELECT table_name, column_name, data_type, is_nullable,
                     col_description((table_schema||'.'||table_name)::regclass, ordinal_position) as comment
              FROM information_schema.columns
              WHERE table_schema='public'
              ORDER BY table_name, ordinal_position""")
            cols = cur.fetchall()
            # 按表分组，屏蔽敏感列，拼 DDL...（省略具体字符串拼接）
            parts = []
            current_table = None
            for table_name, column_name, data_type, is_nullable, comment in cols:
                if column_name.lower() in SENSITIVE_COLS:
                    continue
                if table_name != current_table:
                    if current_table is not None:
                        parts.append(");")
                        # 取 3 行示例并脱敏（略）
                    parts.append(f"-- {table_name}\nCREATE TABLE {table_name} (")
                    current_table = table_name
                else:
                    parts.append(",")
                line = f"  {column_name} {data_type}"
                if is_nullable == "NO":
                    line += " NOT NULL"
                if comment:
                    line += f"  -- {comment}"
                parts.append(line)
            parts.append(");")
            # 外键与示例行（略）
            return "\n".join(parts)

上述 load_schema_text 以块拼接方式构造 DDL 字符串，对敏感列直接跳过不输出，确保模型无法看到这些列名。

# nodes/explainer.py
import json
from langchain.chat_models import init_chat_model
from pydantic import BaseModel

class AnswerOut(BaseModel):
    answer: str
    chart_suggestion: str   # line/bar/pie/table/kpi/none

async def explain_result(state: SQLState):
    llm = init_chat_model("openai:gpt-4o").with_structured_output(AnswerOut)
    if state.get("attempts",0) >= 3 and (
            state.get("last_db_error") or state.get("validation_error")):
        return {"answer":"多次尝试未能生成可执行 SQL，建议换一种问法或联系数据团队协助。",
                "chart_suggestion":"none"}
    out: AnswerOut = await llm.ainvoke([
        {"role":"system","content":"根据 SQL 查询结果，用简洁中文回答用户问题，并根据数据字段（时间/分类/数值）推荐最合适的可视化类型。"},
        {"role":"user","content":f"问题：{state['question']}\nSQL：{state['generated_sql']}\n结果前 20 行：{json.dumps(state['query_result'][:20], ensure_ascii=False, default=str)}"}])
    return {"answer": out.answer, "chart_suggestion": out.chart_suggestion}

29.5 部署与测试

部署采用 FastAPI + 异步执行 + 可选 PostgresSaver（用于审计查询历史）。数据库连接池通过 psycopg_pool 管理，避免每次查询新建连接。流式输出以 messages 模式推送自然语言回答，custom 事件报告"正在加载 schema / 正在生成 SQL / 正在执行查询 / 正在生成解释"等进度。

# main.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver
from .graph import builder
import json, uuid

app = FastAPI()
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

class AskIn(BaseModel):
    question: str
    db_name: str = "analytics"

@app.post("/ask/stream")
async def ask(inp: AskIn):
    run_id = str(uuid.uuid4())
    async def gen():
        config = {"configurable":{"thread_id":f"sql:{run_id}"},
                  "recursion_limit": 20}
        async for mode, payload in graph.astream(
            {"question": inp.question,
             "db_name": inp.db_name,
             "attempts": 0,
             "validation_error": "",
             "last_db_error": ""},
            config=config,
            stream_mode=["messages","custom"]):
            yield f"data: {json.dumps({'mode':mode,'preview':str(payload)[:300]}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

扩展方向：增加 visualization 节点，将 query_result 送入 matplotlib 或 pyecharts 直接生成图表图片（base64 或上传对象存储返回 URL）；增加 dashboard 生成能力，连续多轮查询后自动汇总为看板配置 JSON；接入 Query Plan 解释节点，对复杂 SQL 执行 EXPLAIN 并以自然语言提示成本问题（如全表扫描、缺少索引、JOIN 顺序不佳）；接入 Query Library，将高频自然语言问题缓存为 SQL 模板（hash(question, schema_version) → sql），命中缓存时直接执行而不再调用 LLM，降低成本与延迟；增加权限节点，根据用户角色限制可访问的表与列（如财务人员才能访问 payments 表），在校验层动态调整 allowed_tables。

测试方案：第一层安全单测，维护 50+ 条包含写操作、跨表访问、注入攻击的 SQL 样例，断言 validate_sql 全部拦截，对合法 SELECT 样例自动追加 LIMIT 1000。第二层错误自纠正集成测试，故意构造列名错误、类型错误、JOIN 缺失三类典型错误 SQL，断言 1-2 轮内修正成功率 ≥ 80%，超过 3 轮触发友好错误。第三层 Golden Dataset 端到端测试，准备 100 条标注问题-SQL-期望结果特征三元组，mocking 数据库返回固定数据集，断言 generated_sql 与期望在语义上等价（可使用 SQL 解析树对比或执行结果对比）、answer 包含关键数据、chart_suggestion 合理。第四层并发压测，以 50 QPS 持续 5 分钟，监控 P95 延迟、数据库连接池使用率、失败率，目标失败率 < 1%。第五层失败注入测试，以 mock 数据库抛异常、LLM 返回畸形 JSON、schema 为空等场景验证 fallback 与 attempts 熔断。第六层 SQL 注入安全测试，使用常见注入 payload（'; DROP TABLE orders;--、1 OR 1=1、UNION SELECT password FROM users）发起请求，断言全部被 validate_sql 拦截且数据库端只读账号无权限执行写操作。

Prompt 优化经验：schema_info 中示例数据对准确率影响显著，务必提供真实脱敏样例；字段注释必须包含枚举值与业务含义（如 status 字段的可选值）；JOIN 关系提示可显著减少多表连接错误；将问题中的单位（元/万元、日/月）显式提示给模型，避免单位错算；在 prompt 中强制要求"日期范围默认近 30 天，除非用户指定"，防止遗漏时间条件导致全表扫描。

load_schema 节点实现要点：节点首次进入时检查 Store 缓存，命中则直接返回 schema_info 字符串；未命中则调用 INFORMATION_SCHEMA 查询（columns 表的 column_name、data_type、is_nullable、column_default、character_maximum_length，再查 pg_description 取 comment），对每张表取 3 行示例行（SELECT * FROM table LIMIT 3）并对敏感列值脱敏，同时查 information_schema.table_constraints/key_column_usage 取外键关系生成 JOIN 提示。schema 缓存 TTL 1 小时，数据库 DDL 变更时通过管理接口主动失效。

流式输出设计：SSE 流的事件顺序为 schema_loading → sql_generating → sql_validating → sql_executing → explaining → result/done；每个阶段开始与结束都发一次 custom 事件，前端展示进度条与当前状态；最后发送 sql_preview 事件将最终生成并执行的 SQL 发给前端展示，便于分析师复核与复用；result 事件携带 query_result（前 20 行预览）、answer 文本与 chart_suggestion；前端可展示"查看完整 SQL"与"在 BI 中打开"按钮。

权限与多数据库支持：state.db_name 字段标识本次查询的目标数据库，工具层按 db_name 路由到不同连接池（analytics、marketing、finance 等），每个数据库使用独立的只读账号与 ALLOWED_TABLES 集合；load_schema 根据 db_name 加载对应 schema_embeddings；configurable 中可传入 user_role，validate_sql 按角色动态收窄可访问列（例如普通运营看不到 payments 表的 amount 字段），实现行/列级权限。

FastAPI 主程序完整端点示意：

# main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
from langgraph.checkpoint.memory import MemorySaver
from .graph import builder
from .tools.db import read_pool
import json, uuid, asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时打开连接池
    graph = builder.compile(checkpointer=MemorySaver())
    app.state.graph = graph
    yield
    read_pool.close()

app = FastAPI(lifespan=lifespan)

class AskIn(BaseModel):
    question: str
    db_name: str = "analytics"

@app.post("/ask/stream")
async def ask(inp: AskIn):
    run_id = str(uuid.uuid4())
    async def gen():
        config = {"configurable":{"thread_id":f"sql:{run_id}"},
                  "recursion_limit": 20}
        async for mode, payload in app.state.graph.astream(
            {"question": inp.question, "db_name": inp.db_name,
             "attempts": 0, "validation_error":"", "last_db_error":""},
            config=config, stream_mode=["messages","custom"]):
            yield f"data: {json.dumps({'mode':mode,'preview':str(payload)[:500]}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

@app.get("/healthz")
async def healthz():
    # 简单探活：取一次连接
    try:
        with read_pool.connection() as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT 1")
        return {"ok": True}
    except Exception as e:
        raise HTTPException(500, str(e))

查询历史与相似问题召回：每次 explain_result 成功后，将 question、generated_sql、answer 摘要、时间戳写入 PostgresStore namespace=("sql_history", user_id)；新问题进入 generate_sql 之前，增加一个 recall_similar 节点（可选），用 embedding 检索历史问题，若相似度 > 0.9 直接返回历史 SQL 与结果，跳过 LLM 调用，显著降低高频问题成本与延迟。

SQL 正确性评估：上线前构建 Golden Dataset 时，使用"执行结果等价"而非"SQL 文本相同"评估——两条 SQL 只要对同一数据库返回相同的结果集（列名、行数、数据一致），就视为等价。这种评估方法避免把语义等价但写法不同的 SQL 判为错误，更贴近业务需求。评估脚本对每条标注问题执行生成 SQL 与标注 SQL，将结果集排序后做 hash 对比。

动态 LIMIT 与分页：对于确实需要大结果集的分析任务（如导出明细），不要通过 Text2SQL 直接返回全量数据，而是检测到无 LIMIT 时除了自动追加 LIMIT 1000 外，还在回答中提示"结果已限制为前 1000 行，如需全量导出请使用报表功能"；并提供"导出 CSV"子图分支，该分支使用专门的高权限只读账号与更长 statement_timeout（如 60s），且必须走 HITL 审批，防止拖垮数据库。

数据库连接泄漏防护：所有数据库操作都使用 with 上下文管理器确保连接归还；连接池配置 min_size/max_size 与 timeout，避免流量突增时无限等待连接；在 FastAPI shutdown 事件中显式调用 pool.close()；监控 db_connections_in_use 指标，超过 max_size 80% 时告警。

Text2SQL 准确率提升 Checklist：字段注释必须填写业务含义与枚举值；数字字段注明单位；日期字段注明时区；常见 JOIN 关系在 schema 注释中列出；提供 3-5 条 Few-shot 示例；prompt 中明确要求"不要猜测不存在的列，若字段缺失请在 SQL 注释中说明"；错误自纠正时把数据库错误原文传给模型；对复杂问题（含同比环比、分组 Top N、窗口函数）单独收集 few-shot 示例。

结果解释的质量：explain_result 不应简单复述 SQL 或表格数据，而应总结趋势、极值、异常点与业务含义（如"华东地区本月订单金额环比增长 23%，主要由电子产品品类拉动"）；图表建议给出字段映射（x 轴、y 轴、颜色维度），前端可直接根据建议渲染 ECharts 图表，减少二次配置。

SQL 审计与人工复核：对 WHERE 条件包含敏感范围（如金额 > 100 万、涉及用户明细）的查询，explain_result 之前可增加 await_human_review 节点，interrupt 等待数据分析师审核后再返回结果，防止数据泄露或误报；审计日志中记录所有生成 SQL 与执行时长，便于 DBA 发现并优化慢查询、推荐缺失索引。

Text2SQL 与 BI 工具集成：生成的 SQL 与结果可通过标准接口推送到 Superset、Metabase 等 BI 工具新建图表；explain_result 输出 chart_suggestion 的同时输出 chart_spec（Vega-Lite 或 ECharts option JSON），前端直接渲染；用户点击"保存到看板"时将 chart_spec 与 SQL 持久化为看板组件，实现从问答到看板的闭环。

Schema 演进处理：当数据库表结构变更（新增列、重命名列、删除列）时，schema 缓存必须立即失效；可通过 DDL 触发器或迁移脚本中的 webhook 调用系统的 /schemas/refresh 接口；加载新 schema 后，建议自动跑一遍 Golden Dataset 验证准确率无回退，避免因 schema 变化导致大量 SQL 生成错误。

对话式追问支持：单图单次查询模式的基础上，可追加 followup 节点处理用户对结果的追问（如"按省份再细分一下""只看华东地区"），此时将上一轮 generated_sql、query_result 与追问一并送入 generate_sql 做"基于上一次查询的增量修改"，无需重新加载 schema；这要求 MemorySaver 升级为 PostgresSaver 以跨轮次保留 state，thread_id 改为会话维度而非单次 run 维度。

错误信息脱敏：返回给用户的错误信息不能直接暴露数据库表结构、IP 地址、堆栈信息；execute_sql 捕获到异常后，将原始错误存入 last_db_error 供 LLM 修正用，面向用户的 answer 中只出现"查询执行失败，请换一种问法"等友好提示，具体错误详情写入日志与审计库供排障。

Text2SQL 场景边界：涉及多步推理（先跑 A 查询再根据结果决定 B 查询）、跨库联合分析、图表自动化生成的复杂任务，单图线性架构可能不再适用，应升级为多 Agent 或子图协作架构；但在生产初期保持单图是更好的选择——架构复杂度应与实际准确率数据匹配，避免为罕见场景提前付出工程成本。

模型路由策略：简单查询（从 question_length、关键词、是否单表判断）使用 gpt-4o-mini，复杂查询（多表 JOIN、窗口函数、嵌套子查询）使用 gpt-4o；在 generate_sql 前增加 route_model 节点做轻量分类，分类本身用小模型或规则完成，成本可忽略；观察线上数据后动态调整简单/复杂阈值以平衡成本与准确率。