AI Agent 工程实践教程 · 第 09 章

从业务问题、LangGraph 状态流转和 SQL 生成链路理解 Text-to-SQL。

返回系列目录

第九天_Text-to-SQL业务介绍

第九天：Text-to-SQL 业务介绍与系统设计

在真实项目里，业务人员经常会提出数据问题：

查询最近 7 天未支付订单。
统计昨天各地区的订单数量。
查看本月退款金额最高的产品。

这些问题本身并不复杂，但如果每出现一个新问题，都要走一遍产品、前端、后端、测试和上线流程，数据查询就会变得很慢。

今天学习的 Text-to-SQL，就是为了解决这件事。

Text-to-SQL：把用户的自然语言问题转换成 SQL，查询数据库后，再把结果返回给用户。

订单查询只是今天贯穿使用的一个例子。今天真正要理解的主体是：Text-to-SQL 为什么需要、怎么工作、能带来什么价值。

1. 传统数据查询方式有什么问题

假设运营同学提出一个新需求：

帮我看一下最近 7 天未支付的订单有哪些。

传统开发方式通常是这样的：

业务提出需求
  ↓
产品梳理查询条件和页面原型
  ↓
前端开发查询页面、筛选项和表格
  ↓
后端开发查询接口
  ↓
开发人员编写 SQL
  ↓
前后端联调、测试
  ↓
上线后，业务人员才能看到数据

这个流程适合高频、固定、必须长期使用的核心页面，例如订单管理页、商品管理页和学生管理页。

但它不适合大量临时、变化快、组合多的数据问题。

1.1 一个新问题，往往就要重新开发一次

今天业务要查：

最近 7 天未支付订单。

下周又会问：

最近 30 天，线上产品的未支付订单有多少？

再过两天还可能问：

昨天支付成功但之后申请退款的订单有哪些？

这些问题都和订单有关，但筛选条件、统计维度和展示方式不同。

如果每一个问题都开发一个页面，会遇到几个问题：

本质上，传统方式是：

先把可能的问题预测出来，再提前把查询能力做成页面。

但业务问题往往无法被提前穷尽。

2. 为什么需要 Text-to-SQL

数据库本来就保存着业务数据，也具备很强的筛选、统计、分组和排序能力。

问题在于：数据库听不懂自然语言，它只认识 SQL。

例如，业务人员说：

查询最近 7 天未支付订单。

数据库不能直接执行这句话，而是需要一段结构化查询语句，例如：

SELECT order_no, student_name, order_amount, created_at
FROM order_payment
WHERE status IN ('INIT', 'PROCESSING')
  AND created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
ORDER BY created_at DESC;

传统情况下，这段 SQL 要由开发人员理解需求后编写。

Text-to-SQL 把中间的翻译工作交给 AI：

业务人员提出自然语言问题
  ↓
Text-to-SQL 理解问题含义
  ↓
生成数据库可执行的 SQL
  ↓
查询数据库
  ↓
返回结果

有了 Text-to-SQL 后，业务人员不需要学习 SQL，也不需要为每个临时查询需求等待一个新页面。

这并不表示以后不需要后台页面。

更准确的分工是：

Text-to-SQL 的目标不是替代所有系统页面，而是让“问数据”这件事更直接。

3. Text-to-SQL 的核心能力

Text-to-SQL 做的不是简单的关键词替换，而是完成一条从“人话”到“数据结果”的链路。

用户输入：

统计昨天各地区的订单数量。

系统需要理解：

最后，系统生成 SQL、执行查询，并把结果组织成用户看得懂的内容。

因此可以把 Text-to-SQL 概括为：

自然语言问题
  + 数据库 Schema
  + 业务规则
  ↓
SQL
  ↓
查询结果
  ↓
自然语言回答或图表

这里的三个输入缺一不可：

模型即使很会写 SQL，如果不知道项目中的表和字段，也无法生成正确查询。

4. Text-to-SQL 的基础工作流程

下面用“查询最近 7 天未支付订单”贯穿整个流程。

4.1 第一步：用户输入自然语言问题

用户不需要知道表名和字段名，只需要描述自己想看的数据：

查询最近 7 天未支付订单。

用户说的是业务语言：

• 最近 7 天
• 未支付
• 订单

这些词还不能直接交给数据库执行。

4.2 第二步：理解用户意图

系统需要从问题里提取查询意图。

对于“未支付”，系统还需要知道它在当前项目中的业务定义。

在订单业务中，支付订单的状态可能包括：

“未支付”不能只简单理解为 status != 'SUCCESS'。是否把支付失败和支付超时算作“未支付”，要由业务口径决定。

这一步说明：Text-to-SQL 不只是识别词语，还要把词语映射成业务规则。

4.3 第三步：读取数据库 Schema 和业务资料

模型并不知道项目里有什么表、字段和状态值。因此系统需要给它提供必要上下文。

对于订单查询示例，模型至少需要知道：

表：order_payment，支付订单表

关键字段：
- order_no：支付订单号
- student_name：学生姓名
- product_id：购买的产品 ID
- order_amount：订单支付金额
- status：支付订单状态
- pay_success_time：支付成功时间
- created_at：支付订单创建时间

同时还需要业务说明：

- SUCCESS 表示支付成功。
- INIT 和 PROCESSING 表示尚未支付完成。
- 订单创建时间使用 created_at。
- 支付成功时间使用 pay_success_time。

这类“表、字段、字段含义、状态说明、表关联关系”的资料，统称为 Schema 和业务上下文。

没有这些上下文，模型可能会猜出不存在的表名或字段名。

4.4 第四步：生成 SQL

模型结合用户问题、Schema 和业务规则后，生成 SQL。

例如：

SELECT
    order_no,
    student_name,
    order_amount,
    status,
    created_at
FROM order_payment
WHERE status IN ('INIT', 'PROCESSING')
  AND created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
ORDER BY created_at DESC
LIMIT 50;

这里不要求业务人员理解每一行 SQL，但系统必须确保它的含义和问题一致：

• 查的是支付订单表；
• 筛选的是尚未完成支付的状态；
• 时间使用订单创建时间；
• 明细结果限制最多 50 条。

4.5 第五步：校验并执行 SQL

模型生成 SQL 后，不能不加检查就直接执行。

系统至少要做这些保护：

校验通过后，系统才使用只读数据库账号执行 SQL。

模型生成 SQL
  ↓
程序校验 SQL 是否安全、是否符合权限
  ↓
只读数据库执行查询

模型负责理解和生成，程序负责控制边界。

4.6 第六步：返回查询结果

数据库返回的通常是结构化记录：

订单号 | 学生 | 金额 | 状态 | 创建时间

系统可以按问题类型选择合适的展示方式：

例如对刚才的问题，最终可以直接回答：

最近 7 天共有 18 笔尚未完成支付的订单。
已按创建时间从新到旧展示前 50 条明细。

用户看到的是结果，不需要阅读 SQL。

5. 先理解指标（Metric）：用户到底想查什么

后面会频繁出现“指标”这个词。

指标不是一个普通字段名，也不是数据库里现成的一列数据。它是对一类业务事实的统一定义：统计什么对象、按什么条件筛选、用什么方式计算。

例如，支付订单表中的 order_amount 只是一个字段；而“支付成功订单金额”才是一个指标。

它的完整含义是：

统计对象：支付订单
筛选条件：支付状态为 SUCCESS
计算方式：对 order_amount 求和
时间口径：通常按 pay_success_time 统计

所以，指标可以理解成：

把原始业务数据转换为可稳定复用的业务数字的规则。

5.1 为什么 Text-to-SQL 必须先识别指标

用户通常不会说字段名，而会说业务语言：

昨天交易额是多少？
最近 7 天订单完成率怎么样？
本月客单价比上月增长了吗？

这三句话里都没有直接写出表名、字段名和计算公式。

系统需要先识别：用户想看的到底是哪个指标，再决定查什么表、使用什么状态、怎样聚合。

如果跳过指标识别，直接让模型生成 SQL，常见结果是：

SQL 即使能成功执行，得到的也可能不是用户真正想要的数字。

5.2 基础业务指标示例

下面用订单业务说明几个最基础的指标。

以“支付成功订单金额”为例，它不是简单的：

SUM(order_amount)

还包含两个关键限制：

只统计支付成功订单
按支付成功时间归属到对应日期

这就是为什么指标必须沉淀成业务定义，而不能只在 SQL 里临时拼接。

5.3 分析指标：在基础指标上继续计算

有些指标可以直接从一张表汇总得到；另一些指标是在基础指标之上继续计算出来的分析指标。

例如，用户问：

本月客单价是多少？

系统不能只查 order_amount 的平均值就结束。

如果业务定义的客单价是“每位付费学生的平均支付金额”，它应当是：

本月支付成功订单金额
÷
本月支付成功学生手机号去重数

如果业务定义的是“平均每笔支付金额”，分母又会变成支付成功订单数。两个结果都可能有意义，但不是同一个指标。

5.4 一个完整查询，不只有指标

指标回答的是“算什么”；一个完整的数据查询通常还要补齐其他信息。

例如：

查询最近 7 天线上课程按产品分组的支付成功订单金额，按金额倒序取前 10 名。

可以拆成：

这比“请生成一条 SQL”更接近系统真正需要理解的查询任务。

5.5 指标识别为后续模块提供什么

当系统已经识别出指标，后续模块才有明确方向：

指标：支付成功订单金额
  ↓
知道需要 order_payment.order_amount
  ↓
知道需要 status = SUCCESS
  ↓
知道时间通常使用 pay_success_time
  ↓
再结合时间、筛选、分组等条件生成 SQL

因此，Text-to-SQL 的查询理解并不是直接从一句话跳到 SQL，而是先形成一个结构化查询描述：

指标 + 时间范围 + 筛选条件 + 维度/分组 + 排序/限制 + 展示形式

下一节介绍的 Router、澄清和 Schema 检索，都是为了让这份结构化查询描述足够正确、完整和安全。

5.6 指标不清晰时，先补指标还是先问用户

这里要区分两类问题：指标口径缺失 和 本次查询参数缺失。

因此，下面这句话：

昨天交易额是多少？

如果指标语义层已经规定：

交易额 = 支付成功订单的 order_amount 之和，按 pay_success_time 统计

系统就不应该再问“按创建时间还是支付成功时间”。它应直接把指标口径补齐，再根据“昨天”生成查询。

真正需要澄清的是这种情况：

昨天订单量是多少？

如果系统中同时维护了“创建订单量”和“支付成功订单量”，但“订单量”没有默认指向其中一个，才应问用户：

你想看昨天创建的订单数，还是昨天支付成功的订单数？

这背后有一个很重要的设计原则：

能被指标口径统一解决的问题，不要重复追问每一个用户；只有指标无法唯一确定，或本次查询条件确实缺失时，才向用户澄清。

这样既能保证统计结果一致，也能避免 Text-to-SQL 变成一个不停追问的系统。

6. 企业级 Text-to-SQL：为什么不能直接生成 SQL

上面的流程描述了 Text-to-SQL 的主链路，但真实系统不能把每一句用户输入都直接交给模型生成 SQL。

更可靠的设计应当是：先判断问题能不能查、信息够不够、需要哪些上下文，再决定是否生成 SQL。

用户问题
  ↓
Router：这是不是数据查询？
  ├─ 普通问答 -> 交给普通问答能力
  ├─ 操作请求 -> 拒绝直接执行，引导到业务页面
  ├─ 无法判断 -> 只澄清“想查数据还是想了解业务规则”
  └─ 是
       ↓
RAG 检索：候选指标定义 + 候选表目录 + 表关联关系
       ↓
大模型做查询规划：选择真实指标、需要的表和关联路径
  ├─ 指标无法唯一确定 -> 澄清或补充指标定义
  ├─ 查询条件缺失 -> 澄清或使用安全默认值
  └─ 查询计划完整
       ↓
取得已选表的详细 Schema、已选指标口径和关联路径
  ↓
生成 SQL
  ↓
SQL 安全校验、权限改写和执行
  ↓
结果解释；失败时有限次数修复或明确反馈

这不是把流程故意做复杂，而是为了避免模型在错误的问题、错误的上下文或错误的权限范围内生成一条“看起来很合理”的 SQL。

6.1 Router：先判断是不是数据查询

用户在同一个对话框里可能问任何事情：

你好。
帮我解释一下退款流程。
查一下最近 7 天未支付订单。
把这些订单全部取消。

只有第三句是可以进入 Text-to-SQL 的只读数据查询。

如果没有 Router，系统很容易出现两个问题：

Router 可以由规则、轻量模型，或两者组合完成。它的任务不是生成 SQL，而是输出一个小而稳定的分类结果：

例如：

“最近 7 天未支付订单有多少？” -> DATA_QUERY
“退款为什么失败？” -> GENERAL_CHAT
“把昨天超时订单改成成功” -> DATA_OPERATION

通常会优先使用轻量模型或规则做 Router，因为这一步只需要分类，不需要昂贵模型进行复杂推理。它能降低成本和响应时间，也把高风险的操作请求提前挡在 SQL 生成之前。

Router 的澄清只处理“请求类型无法判断”的极少数情况，不负责确认指标、时间范围、表和字段。这些事情进入 Text-to-SQL 查询链路后，再由查询规划模块处理。

6.2 RAG 检索：先提供候选资料，不直接决定查询答案

Router 确认是 DATA_QUERY 后，系统先从两类知识库中检索候选资料：指标库和 Schema 目录。

例如用户问：

线上课程退款额是多少？

RAG 可以召回：

候选指标：退款成功金额
指标定义：SUCCESS 退款订单的 refund_amount 求和，按 refund_success_time 统计
核心表：order_refund

候选表资料：
- order_refund：退款订单
- order_payment：支付订单
- order_product：产品及 teaching_mode

关联关系：
order_refund.payment_order_no -> order_payment.order_no
order_payment.product_id -> order_product.id

这里的 RAG 结果只是候选资料，不是最终答案。它负责把“可能相关的信息”找回来，避免模型凭空猜测；真正选择哪个指标、哪些表、哪条关联路径，仍由下一步的查询规划模型根据用户问题决定。

对于表较少的系统，Schema 目录可以直接作为轻量上下文提供给检索或规划模型；对于表很多的系统，可以先做向量检索、关键词检索或规则召回，再把少量候选表交给模型。

6.3 查询规划与要素检查：不是每个数据问题都能立刻查

RAG 返回候选指标和候选表后，查询规划模型仍然不能马上生成 SQL。它需要结合用户问题，从候选资料中选择真实对应的指标、表和关联路径，并判断查询信息是否足够明确。

查询规划阶段最好输出结构化结果，例如：

selected_metrics：退款成功金额
selected_tables：order_refund、order_payment、order_product
join_path：order_refund -> order_payment -> order_product
time_range：本月
filters：teaching_mode = ONLINE，refund_status = SUCCESS
group_by：无
sort：无
needs_clarification：false

如果 needs_clarification 为 true，系统先向用户提问，不进入 SQL 生成阶段。

6.3.1 查询计划必须是可校验的契约

查询规划不能只返回一段自然语言，例如：

我认为应该查询退款表、支付表和产品表。

这段话人能看懂，程序却很难可靠地继续处理。

更好的做法是让模型按固定 JSON Schema 输出查询计划，程序再校验它：

{
  "status": "READY",
  "selected_metric": "退款成功金额",
  "selected_tables": ["order_refund", "order_payment", "order_product"],
  "join_path": [
    "order_refund.payment_order_no = order_payment.order_no",
    "order_payment.product_id = order_product.id"
  ],
  "time_range": {
    "type": "MONTH_TO_DATE"
  },
  "filters": [
    {"field": "order_product.teaching_mode", "operator": "=", "value": "ONLINE"}
  ],
  "group_by": [],
  "order_by": [],
  "limit": 1,
  "result_columns": [
    {"alias": "refund_amount", "role": "MEASURE", "unit": "元"}
  ],
  "needs_clarification": false
}

程序至少要校验：

SQL 生成模型收到这份已校验计划后，必须使用约定别名，例如：

SELECT SUM(r.refund_amount) AS refund_amount
...

这样执行后的结果处理模块知道 refund_amount 是一个金额指标，而不需要从任意 SQL 的列名里猜测语义。

这份查询计划就是模块之间的契约：规划模型负责选择什么，SQL 模型负责如何查询，程序负责校验和执行，结果模块负责按约定展示。没有这个契约，后续每一步都会依赖模型的自由文本，系统很难稳定。

一次查询通常由下面几个要素组成：

不是每个问题都必须包含全部要素。例如“当前未支付订单有多少”不一定需要时间范围；但“哪个地区订单最多”通常需要明确时间范围，也需要明确“订单”是创建订单、支付成功订单还是全部订单。

下面几个问题就存在关键歧义：

如果没有这个模块，模型会在“指标未定义”和“查询参数缺失”之间混为一谈：要么自行猜测口径，要么对本可直接执行的问题反复追问。SQL 可能可以执行，但回答的是另一个问题，这比直接报错更危险。

6.4 澄清对话：让系统问最少、最关键的问题

发现信息不完整后，系统不应把全部字段都抛给用户选择，也不应连续问很多问题。

更好的做法是：每次只追问一个对结果影响最大的缺口。

例如：

用户：昨天订单量是多少？

系统：你想统计昨天创建的订单数，还是昨天支付成功的订单数？

用户回答后，系统把答案保存在本次会话上下文中，再继续补齐后续条件。

澄清模块解决的是“自然语言天然不完整”的问题。它的目标不是让用户填一张复杂表单，而是通过简短对话把模糊问题变成可执行的查询任务。

设计上还需要设置边界：

• 连续澄清超过限定次数仍无法明确时，给出可选问题示例或转人工；
• 有安全默认值的项目可以自动补齐，例如明细默认最多返回 50 条；
• 指标口径缺失时，不能由模型擅自补齐；应补充指标定义，或让用户在多个候选指标中确认。

6.4.1 多轮追问：保存查询上下文，不只是保存聊天记录

Text-to-SQL 常见的使用方式不是一次提问后结束，而是基于上一轮结果继续追问：

用户：最近 7 天每天的交易额趋势怎么样？
系统：返回按天的交易额趋势。

用户：那只看线上课程。
用户：那退款额呢？

如果系统只保存原始聊天文本，模型需要每次重新猜测上一轮选了什么指标、哪些表、时间范围是什么，很容易漏掉条件或错误继承条件。

因此，每轮成功查询后，程序应保存结构化 QueryContext：

selected_metric：支付成功订单金额
selected_tables：order_payment、order_product
time_range：最近 7 天
filters：teaching_mode = ONLINE
group_by：按天
order_by：日期升序
result_columns：pay_date、paid_amount

下一轮追问到来后，查询规划模块先判断它与上一轮的关系。

这里有一个原则：

可以继承用户没有要求改变的查询条件，但不能直接复用上一轮 SQL。

指标、表和关联关系一旦变化，就必须重新规划、重新校验并重新生成 SQL。这样既能让追问自然，又不会把上一轮错误的表关系或权限范围带到下一轮。

6.5 Schema 检索与选表：先让模型知道“有哪些表”，再决定“用哪些表”

真实数据库可能有几十上百张表、成千上万个字段。

这不代表模型一开始就什么都不知道。更合理的做法是，把所有表的简要介绍和表关联关系整理成一份轻量级的表目录，交给“选表”模块使用。

例如，表目录不需要包含每个表的全部字段，而可以先提供：

同时提供表之间的关联关系：

order_refund.payment_order_no -> order_payment.order_no
order_payment.product_id -> order_product.id
order_payment.prepay_order_no -> prepay_order.order_no

这份表目录既是 6.2 中 Schema RAG 的资料来源，也是查询规划模型选表时使用的候选上下文。它的作用是：让模型在理解问题后，先选择可能需要的表，而不是让它凭记忆猜表名。

6.5.1 指标也要告诉模型核心来源表

指标定义不只包含计算公式，也应包含它通常依赖的核心表和字段。

指标可以直接确定“核心事实表”，但不一定能决定所有表。

例如“退款成功金额”本身只需要 order_refund；如果用户继续要求“线上课程的退款成功金额”，就还需要通过支付订单和产品表补上“线上课程”这个维度：

指标：退款成功金额
  核心表：order_refund

筛选条件：线上课程
  需要补充关系：
  order_refund -> order_payment -> order_product

6.5.2 两阶段：先选表，再生成 SQL

因此，推荐把 Schema 使用拆成两个阶段：

第一阶段：选表
用户问题 + RAG 召回的候选指标 + 候选表目录 + 表关联关系
  ↓
模型输出：需要哪些表、为什么需要、表如何关联

第二阶段：生成 SQL
用户问题 + 已选表的完整字段说明 + 已选关联关系 + 指标口径
  ↓
模型输出：SQL

对于“最近 7 天未支付订单有哪些”：

选表结果：
- order_payment
- 原因：包含订单状态、创建时间和订单明细

生成 SQL 时提供：
- order_payment 的完整字段说明
- 未支付状态的业务定义
- 最近 7 天的时间条件

对于“线上课程退款额是多少”：

选表结果：
- order_refund：提供退款成功金额和退款成功时间
- order_payment：把退款订单关联回原支付订单
- order_product：提供产品的 teaching_mode

关联路径：
order_refund.payment_order_no
  -> order_payment.order_no
  -> order_product.id（通过 order_payment.product_id）

第二个问题需要三张表，不是因为模型要“多找资料”，而是因为业务条件分散在不同表中。

6.5.3 为什么不把所有完整字段都交给 SQL 生成模型

如果直接把所有表、所有字段和所有规则放进 SQL 生成提示词，会造成：

表目录可以相对简短，并覆盖所有表；SQL 生成阶段只接收选中的少量表的详细 Schema。

对于表不多的小型系统，可以直接把完整表目录交给选表模型。对于表很多的大型系统，可以先用关键词、向量检索或规则从目录中召回候选表，再让模型在候选表中做最终选择。

这就是 Schema 检索模块的价值：不是让模型拥有一份越长越好的数据库说明，而是在正确时机把正确的表、关系和字段交给正确的模型。

6.6 多数据源的第一种落地方式：统一进入数据仓库

微服务系统中，订单、支付、用户、商品等业务通常各自拥有数据库。经营分析如果每次都直接跨这些业务库查询，关联复杂、性能不可控，也容易影响线上服务。

第一个、也是最常见的落地方式是：

把需要分析的业务数据同步到统一数据仓库，Text-to-SQL 只查询数据仓库。

订单库 ─┐
支付库 ─┼─ 数据同步/ETL/CDC ─> 数据仓库
用户库 ─┤                         ↓
商品库 ─┘                   Text-to-SQL 查询

业务库仍然负责线上交易和服务；数据仓库负责跨业务域的统计和分析。这样用户问：

上个月购买线上课程的用户中，有多少人在 7 天内申请退款？

Text-to-SQL 不需要同时连订单库、支付库、退款库和产品库，而是直接查询数仓中已经整理好的分析表、宽表或聚合表。

6.6.1 数据仓库中应该准备什么

数据仓库不是简单复制所有业务表。它需要把经营分析常用的业务键和关系提前整理好。

例如，可以沉淀一张订单支付退款分析宽表：

这样“线上课程退款额”“按地区统计交易额”“支付后 7 天退款率”等跨域问题，都可以在统一分析表或主题表中完成，而不需要让模型临时设计跨库 Join。

6.6.2 Text-to-SQL 如何使用数据仓库 Schema

在这个方案里，Schema 目录和指标库都围绕数据仓库维护：

数据源：analytics_dw
表：dwd_order_payment_refund
说明：订单、支付、退款和产品维度的经营分析宽表
适用指标：交易量、交易额、退款量、退款额、退款率
可用维度：日期、产品、授课方式、地区

RAG 检索到指标和 Schema 后，查询规划模型只需要选择：

使用哪张数仓表
使用哪个指标口径
按什么时间、维度和筛选条件统计

例如：

问题：最近 30 天线上课程按产品的退款额 Top 10

查询规划：
- 数据源：analytics_dw
- 表：dwd_order_payment_refund
- 指标：退款成功金额
- 筛选：授课方式 = ONLINE，退款状态 = SUCCESS
- 时间：最近 30 天，按退款成功日期过滤
- 分组：产品
- 排序：退款额倒序
- 限制：前 10 名

SQL 生成模型随后只拿到这张数仓表的详细字段、指标口径和查询计划生成 SQL。

6.6.3 为什么优先让 Text-to-SQL 查询数据仓库

6.6.4 这对查询规划意味着什么

采用统一数据仓库后，查询规划不再需要为每次问题选择跨库执行策略，而是固定为：

数据源：analytics_dw
  ↓
RAG 召回：数仓表目录、指标定义和维度说明
  ↓
大模型选择：分析表、指标、时间、筛选、分组和排序
  ↓
生成面向数据仓库的 SQL

这能让学生先把 Text-to-SQL 的核心流程做稳：指标识别、Schema 检索、查询规划、SQL 安全和结果处理。其他跨源方案后面再作为架构扩展讨论。

6.7 指标语义层：让“交易额”只有一个可追溯定义

表结构说明只能告诉模型字段在哪里，不能保证模型知道业务指标是什么意思。

例如“交易额”在不同团队口中，可能是：

• 创建订单金额；
• 支付成功金额；
• 支付成功金额减退款成功金额；
• 某个产品的标价乘以订单数。

因此需要把常用指标沉淀成可复用的语义资料：

这层资料有两个好处：

1. 业务人员、开发人员和模型使用同一套指标语言。
2. 指标口径变化时，修改定义即可，不需要在每一个提示词里手工替换。

没有语义层时，同一个问题在不同日期、不同模型或不同开发人员手里，可能得到不同 SQL 和不同结果。

6.8 SQL Guard 与执行计划检查：SQL 能执行，不代表 SQL 可以执行

模型生成 SQL 后，需要由程序中的 SQL Guard 做独立检查。这里不是让大模型自己审核自己生成的 SQL，而是让程序和数据库共同决定这条 SQL 能否执行。

第一层是静态检查和权限检查：

这些规则必须由程序确定性执行。例如：程序解析 SQL 后发现不是单条 SELECT，就直接拒绝；发现用户没有查看手机号的权限，就拒绝该字段或改为脱敏结果。大模型不能拥有放行权。

6.8.1 EXPLAIN：执行前检查数据库准备怎样查询

SQL 语法正确、表字段也都合法，仍然可能非常慢。

例如，模型生成：

SELECT order_no, student_name, order_amount, created_at
FROM order_payment
WHERE status IN ('INIT', 'PROCESSING')
  AND created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
ORDER BY created_at DESC
LIMIT 50;

这条 SQL 看起来没有问题，但系统仍然需要关心：

数据库会使用哪个索引？
预计要扫描多少行？
是否会创建临时表？
是否要做额外排序？
多表关联时，关联字段是否有索引？

程序可以先执行：

EXPLAIN SELECT ...

EXPLAIN 不会真正查询出业务结果，而是让 MySQL 返回优化器计划使用的访问方式。程序可以读取其中的关键信息：

注意：是否走索引不是一个绝对的是非题。

小表只有几十行时，全表扫描通常完全可以接受；但百万级订单表扫描几十万行，再做多表关联和排序，就可能拖慢业务库。系统应该根据表规模、预估扫描行数、查询超时和当前负载设置风险阈值，而不是机械规定“没有索引就一律拒绝”。

6.8.2 从真实订单查询理解索引设计

当前项目的 order_payment 已有这些索引：

uk_order_no(order_no)
idx_student_phone(student_phone)
idx_prepay_order_no(prepay_order_no)
idx_status(status)

对于刚才“最近 7 天未支付订单”的查询，条件同时使用了：

status
created_at
ORDER BY created_at DESC

现有的 idx_status 可能帮助先筛选订单状态，但 created_at 目前没有对应索引。随着订单数据增长，数据库可能仍要在符合状态的记录中扫描和排序。

如果这类查询是高频需求，开发人员或 DBA 应基于真实 EXPLAIN 结果、数据量和线上负载评估是否增加组合索引，例如：

(status, created_at)

组合索引不是看到两个筛选字段就立刻创建。还要考虑：

• 这个查询是否高频；
• status 的区分度是否足够；
• 查询是否还需要按其他字段分组或排序；
• 新索引会增加写入成本和存储空间；
• 线上真实执行计划是否确实改善。

索引设计是开发和数据库治理的长期工作，不应该由 Text-to-SQL 在每次请求时自动创建索引。Text-to-SQL 需要做的是：发现高风险计划后拒绝、降级或提示，并把高频慢查询记录下来，交给后续优化。

6.8.3 程序如何决定放行、降级或拒绝

执行前可以形成下面的程序决策链：

模型生成 SQL
  ↓
程序解析 SQL：只读、单语句、表字段白名单、权限范围
  ↓
程序补充 LIMIT、数据范围和脱敏规则
  ↓
程序执行 EXPLAIN，读取索引和预估扫描行数
  ├─ 风险可接受 -> 使用只读账号执行查询
  ├─ 风险偏高 -> 缩小时间范围、要求增加筛选条件，或提示用户改查汇总
  └─ 风险过高 -> 拒绝执行，并记录慢查询风险

运行时还需要数据库和连接层的兜底：

这里的职责边界非常清晰：

这里有一个关键设计原则：

模型可以提出查询方案，但程序必须拥有最终执行权。

即使前面做过 Router 和 Schema 检索，SQL Guard 仍然不能省略。因为模型输出始终是不可信的外部输入。

6.9 查询结果后处理：原始数据不等于用户答案

SQL 执行完成后，数据库返回的只是原始结果集，例如：

pay_date    | paid_amount
2026-07-01  | 12800.00
2026-07-02  | 15600.00
2026-07-03  | 9800.00

这份结果对数据库来说已经正确，但对业务人员来说仍然不够友好：

• 不知道 paid_amount 是交易额、退款额还是其他金额；
• 不容易一眼看出每天的趋势；
• 不知道数据是否为空、是否有异常波动；
• 不知道应该看表格、图表还是一句结论；
• 不能要求业务人员自己解释聚合结果。

因此，Text-to-SQL 在执行 SQL 后还需要一个结果后处理模块。它的任务不是重新计算或篡改数据，而是把已经得到的可信结果转换成用户容易理解和使用的分析结果。

数据库原始结果
  ↓
程序识别列名、数据类型、行数和查询计划
  ↓
生成标准化结果数据与图表配置
  ↓
模型严格基于结果生成自然语言总结
  ↓
前端展示：指标卡、表格、图表和文字结论

6.9.1 先理解结果结构，而不是直接把行数据交给模型

程序需要先把数据库结果整理成带有语义的结构化数据。

例如，查询计划已经说明：

指标：支付成功订单金额
时间范围：最近 7 天
分组维度：按天

结合 SQL 返回列，程序可以形成：

{
  "metric": "支付成功订单金额",
  "unit": "元",
  "time_range": "最近 7 天",
  "dimensions": ["pay_date"],
  "measures": ["paid_amount"],
  "rows": [
    {"pay_date": "2026-07-01", "paid_amount": 12800.00},
    {"pay_date": "2026-07-02", "paid_amount": 15600.00}
  ]
}

这一步通常由程序完成，因为列类型转换、金额格式、日期格式、空值处理和字段脱敏都应该是确定性的。

如果结果列与查询计划不一致，例如计划要查“交易额”，结果却没有金额字段，系统不应该直接让模型编写总结，而应将其标记为异常结果并进入修复或失败处理。

6.9.2 图表推荐：根据数据形态选择，不是随便画图

图表的作用是帮助用户更快发现趋势、差异和结构；它不是给结果套一个好看的外壳。

程序可以根据查询计划和结果结构生成推荐图表类型：

例如：

问题：最近 7 天每天的交易额趋势
结果结构：日期 + 交易额
推荐：折线图

问题：本月各产品交易额 Top 10
结果结构：产品名称 + 交易额 + 排名
推荐：横向柱状图，同时展示排名表

问题：查询最近 7 天未支付订单有哪些
结果结构：订单号、学生、金额、状态、创建时间
推荐：表格，不建议画图

图表推荐可以使用规则优先、模型辅助的方式：

• 程序根据维度类型、指标数量、行数和查询意图先给出安全的默认图表；
• 模型可以在规则允许的范围内推荐更适合业务表达的展示方式；
• 前端最终只渲染受支持的图表配置，不能让模型输出任意脚本或图表代码。

这能避免把几十个分类硬塞进饼图，或把本该查看明细的订单数据错误地画成趋势图。

6.9.3 自然语言总结：模型负责解释，不负责编造

结果后处理模块还需要把数据交给模型生成面向业务人员的回答。

此时模型收到的内容应当是：

用户问题
已确认的指标口径
时间范围和筛选条件
标准化后的查询结果
图表推荐或关键统计值

模型可以生成：

最近 7 天交易额整体呈上升趋势。
7 月 2 日交易额最高，为 15,600 元；7 月 3 日回落至 9,800 元。

但提示词必须明确约束：

只能依据给定结果回答。
不能补造数据库中没有的原因、比例或业务结论。
结果为空时，明确说明没有查到符合条件的数据。
金额、数量和日期必须与结果一致。

模型擅长把结构化数字组织成人话，但它不应该替代数据计算。同比、环比、Top N、异常值等需要计算的结论，应优先由程序或经过确认的指标公式计算后，再交给模型解释。

6.9.4 面向用户的最终返回结构

前端最终不应该只接收一段模型文本，而应接收可组合展示的结构化结果：

query_summary：用户问题和已确认口径
metric_cards：核心指标卡，例如交易额、订单量
table_data：明细或分组结果表
chart_config：受支持的图表类型、维度和指标数据
natural_language_answer：基于结果的文字总结
warnings：数据为空、结果被截断、时间范围较大等提醒

这样同一份查询结果可以在不同终端以不同方式展示：

• 管理后台显示表格和图表；
• 对话框优先显示简短结论和关键指标；
• 导出功能使用标准化表格数据；
• 审计页面保留查询计划、SQL 和执行摘要。

结果后处理的价值在于：数据库负责返回事实，程序负责保证结构和展示安全，模型负责帮助用户理解数据。三者分工清楚，才能让 Text-to-SQL 从“能查到数据”走到“能帮助业务分析数据”。

6.10 执行失败与结果为空：不要无限重试

即使上下文完整，SQL 仍可能失败：字段名拼错、数据库版本差异、权限不足、Schema 刚变更，都会导致执行异常。

合理的处理方式是：

执行 SQL
  ├─ 成功且有结果 -> 解释结果
  ├─ 成功但无结果 -> 如实说明未查到符合条件的数据
  └─ 失败 -> 把安全的错误信息交给模型修正一次或两次

重试次数必须受限。无限重试只会不断消耗模型与数据库资源，还可能让用户一直等待。

如果多次修复后仍失败，系统应当明确反馈，例如：

当前无法确定“订单量”的统计口径。请说明需要统计创建订单数还是支付成功订单数。

比起生成一条错误 SQL 后假装成功，清楚地告诉用户缺少什么信息更可靠。

6.11 评测与观测：系统上线后如何知道它是否可靠

Text-to-SQL 不能只用“能跑出一条 SQL”来判断好坏。

应该为高频和高风险问题准备评测集，例如：

• 单表明细查询；
• 聚合、分组、排序查询；
• 多表关联查询；
• 模糊时间表达；
• 指标口径歧义；
• 无权限和敏感字段请求；
• 非数据查询和数据修改请求。

每次执行还应该记录：用户问题、Router 分类、澄清次数、检索到的 Schema、生成 SQL、校验结果、执行耗时、错误类型和最终回答。

这些记录用于发现三类问题：模型理解错了、业务口径资料不完整、数据库结构发生了变化。没有评测和观测，系统很难长期稳定迭代。

7. 用一个订单查询例子串起全流程

再看一遍完整过程：

用户：查询最近 7 天未支付订单
  ↓
系统理解：订单明细 + 最近 7 天 + 未支付状态
  ↓
系统读取：order_payment 表、status 字段、created_at 字段和状态字典
  ↓
模型生成：只查询未完成支付订单的 SQL
  ↓
系统校验：只读、表字段合法、结果数量受限、用户有订单查看权限
  ↓
数据库执行：返回订单明细
  ↓
前端展示：表格、统计数量和自然语言总结

订单只是示例。把查询对象换成学生、课程、考试、作业、退款、直播数据，Text-to-SQL 的基本过程仍然一样。

变化的只是：

• 要提供给模型的表和字段；
• 业务指标的定义；
• 不同角色允许查询的数据范围。

8. Text-to-SQL 常见应用场景

Text-to-SQL 的价值，在于让业务人员提出以前没有预设过的问题。

这里最重要的不是“系统预置了多少个问题”，而是业务人员可以自己组合时间、范围、维度和指标：

昨天
  + 线上课程
  + 支付成功
  + 按产品分组
  + 统计交易额

传统页面通常很难提前覆盖所有组合；Text-to-SQL 正好补足了这种探索式查询能力。

9. Text-to-SQL 并不是“模型随便写 SQL”

一个可靠的 Text-to-SQL 系统，至少要处理三类问题。

9.1 Schema 问题

模型需要知道真实数据库结构。

如果项目中表叫 order_payment，模型却猜成 orders；字段叫 order_amount，模型却猜成 amount，SQL 就无法执行。

因此，系统需要持续维护：

• 表说明；
• 字段说明；
• 主外键或业务关联关系；
• 枚举状态和状态含义；
• 常用指标口径。

9.2 业务口径问题

模型生成的 SQL 语法正确，不代表查询结果符合业务含义。

例如：

交易额

至少可能有三种不同理解：

在当前订单业务里，若没有额外说明，“交易额”建议使用支付成功订单的 order_amount 之和，并按 pay_success_time 过滤时间。

9.3 安全与权限问题

数据库里不仅有统计数据，也可能有手机号、姓名、订单号等敏感信息。

因此，Text-to-SQL 不能因为用户问了一句“把所有学生手机号给我”就直接返回数据。

系统需要同时判断：

这个人有没有权限看这些表？
这个人有没有权限看这些字段？
这条 SQL 是否只是查询？
这次结果是否应该脱敏或限制条数？

Text-to-SQL 的目标是降低查询门槛，不是绕过系统权限。

10. Text-to-SQL 带来的价值

Text-to-SQL 最终改变的是数据访问方式：

过去：先开发一个查询页面，再查询数据
现在：先提出一个数据问题，再即时得到查询结果

它带来的价值包括：

当然，Text-to-SQL 不应该替代所有固定报表和核心管理页面。

高频、标准化、操作型的工作仍然适合专门页面；Text-to-SQL 更适合回答那些“现在突然想知道”的数据问题。

11. 本节总结

今天先建立了 Text-to-SQL 的基本认识：

传统方式：业务提需求 -> 开发页面和接口 -> 查询数据

Text-to-SQL：业务提问题 -> AI 理解问题 -> 生成 SQL -> 查询数据 -> 返回结果

一个可用于真实业务的 Text-to-SQL 系统，完整链路应当是：

用户输入自然语言问题
  ↓
Router 判断：数据查询、普通问答还是操作请求
  ↓
RAG 召回候选指标定义、候选表目录和关联关系
  ↓
查询规划：选择真实指标、需要的表和关联路径
  ↓
查询要素检查：补齐时间、筛选、维度、分组和排序；必要时澄清
  ↓
跨业务域数据统一同步到数据仓库，Text-to-SQL 只查询分析库
  ↓
取得已选数仓表的详细 Schema、已选指标口径和关联路径
  ↓
生成 SQL
  ↓
SQL Guard 校验只读性、权限、敏感字段，并通过 EXPLAIN 检查执行计划风险
  ↓
只读数据库执行；失败时有限次数修复
  ↓
结果后处理：标准化数据、选择表格或图表、生成可信文字总结
  ↓
返回指标卡、表格、图表和自然语言回答

最后记住六件事：

1. Text-to-SQL 的价值，是让业务人员用自然语言直接访问数据。
2. 不是所有输入都该生成 SQL，Router 要先把普通问答和操作请求拦在查询链路外。
3. 指标决定“算什么”；完整查询还要补齐时间、筛选、维度、分组和排序。
4. SQL 执行结果还要经过结构化后处理和结果解释，模型只能基于真实结果总结，不能编造分析结论。
5. 跨业务域的分析数据优先统一进入数据仓库，Text-to-SQL 不直接对多个业务库做跨库 Join。
6. Text-to-SQL 必须经过程序安全校验、执行计划检查、权限控制、评测与观测，不能让模型直接拥有数据库的全部能力。

第九天_LangGraph基础与Text-to-SQL流程

第九天：LangGraph 基础与流程编排

前面已经学过 Chain 和 Agent。

Chain 适合把固定步骤串起来。

Agent 适合让模型自己判断要不要调用工具。

但是很多真实 AI 业务不是一条简单直线，也不能完全交给模型自由决定。

例如：

用户问题
  ↓
判断请求类型
  ↓
理解问题
  ↓
调用模型或工具
  ↓
根据结果决定下一步
  ↓
必要时重试、暂停、人工确认、保存状态
  ↓
返回最终结果

这种流程需要程序清楚地管理每一步。

这一节学习的 LangGraph，就是用来组织这种复杂 AI 流程的。

1. 为什么需要 LangGraph

如果流程很简单，用 Chain 就够了。

比如：

Prompt -> Model -> Parser

或者：

用户问题 -> 检索资料 -> 拼接 Prompt -> 模型回答

这种流程的特点是：

每一步都固定。
第一步做完，一定进入第二步。
第二步做完，一定进入第三步。

但真实项目里，经常会遇到这些情况：

有些输入要走 A 流程，有些输入要走 B 流程；
某一步失败后要重试；
某一步结果不安全，要拒绝继续执行；
某一步需要人工确认；
流程跑到一半，需要保存状态；
后面要回看中间状态，排查到底哪一步错了。

这时流程就不再是一条直线，而是一张图。

开始
  ↓
判断类型
  ├─ 普通问答 -> 直接回答
  ├─ 工具任务 -> 调用工具 -> 解释结果
  └─ 高风险任务 -> 人工确认 -> 继续或拒绝

LangGraph 的作用就是：

把复杂 AI 流程拆成节点、边、状态和分支，让程序知道每一步应该怎么继续。

2. Chain、Agent 和 LangGraph 的区别

可以先这样记：

换一种更直白的说法：

Chain：程序规定好固定顺序。
Agent：模型决定下一步要不要调用工具。
LangGraph：程序把流程图设计好，节点之间按条件流转。

这三者不是互相替代。

在真实项目里，它们经常一起使用。

例如：

LangGraph 负责整个流程。
某个节点里面可以调用 Chain。
某个节点里面也可以调用模型或工具。

3. LangGraph 核心概念

学习 LangGraph 时，先记住 5 个词。

3.1 State：流程运行时的数据

State 就是这张图运行时手里拿着的数据。

例如：

class DemoState(TypedDict, total=False):
    question: str
    cleaned_question: str
    answer: str

可以理解成：

question：用户原始问题
cleaned_question：清理后的问题
answer：最终回答

节点可以读取 State，也可以返回新字段写回 State。

3.2 Node：一个处理步骤

Node 就是流程里的一个处理步骤。

例如：

清理问题
提取关键词
生成回答
校验结果
调用工具

节点函数一般长这样：

def clean_question_node(state: DemoState) -> DemoState:
    return {
        "cleaned_question": state["question"].strip()
    }

返回的字典会合并回 State。

3.3 Edge：下一步去哪里

Edge 用来连接节点。

graph_builder.add_edge("clean_question", "answer")

意思是：

clean_question 执行完，下一步进入 answer。

3.4 Conditional Edge：根据结果走分支

有些流程不是固定下一步。

例如：

分数 >= 60 -> pass
分数 < 60 -> retry

这时就用条件边。

graph_builder.add_conditional_edges(
    "grade",
    choose_next_node,
    {
        "pass": "pass_node",
        "retry": "retry_node",
    },
)

3.5 Checkpoint：保存流程状态

State 是流程正在使用的数据。

Checkpoint 是把某一刻的 State 保存成进度记录。

可以这样理解：

State：现在手里的草稿纸。
Checkpoint：点了一下保存，留下来的存档。

有了 checkpoint，后面才能做这些事情：

保存同一段对话的状态；
从暂停点继续执行；
回看历史 State；
从中间某一步分叉重跑。

4. LangGraph 基础代码怎么写

最小代码通常按 6 步写。

1. 定义 State
2. 写节点函数
3. 创建 StateGraph
4. 注册节点
5. 连接边
6. compile 后运行

完整示例：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class DemoState(TypedDict, total=False):
    question: str
    cleaned_question: str
    answer: str


def clean_question_node(state: DemoState) -> DemoState:
    return {
        "cleaned_question": state["question"].strip()
    }


def answer_node(state: DemoState) -> DemoState:
    return {
        "answer": f"收到问题：{state['cleaned_question']}"
    }


graph_builder = StateGraph(DemoState)
graph_builder.add_node("clean_question", clean_question_node)
graph_builder.add_node("answer", answer_node)

graph_builder.add_edge(START, "clean_question")
graph_builder.add_edge("clean_question", "answer")
graph_builder.add_edge("answer", END)

graph = graph_builder.compile()

result = graph.invoke({
    "question": "  什么是 LangGraph？  "
})

print(result)

运行后最终 State 里会有：

question
cleaned_question
answer

对应 demo：

uv run python test/langgraph_demo/01_create_graph_demo.py

5. 常见流程控制

这一节只掌握常见写法。

不要一开始就背所有 API。

5.1 条件分支

条件分支适合这种场景：

通过 -> 结束
不通过 -> 重试

对应 demo：

uv run python test/langgraph_demo/04_conditional_edge_demo.py

5.2 循环重试

循环本质上就是条件边回到前面的节点。

生成草稿
  ↓
检查质量
  ├─ 合格 -> 结束
  └─ 不合格 -> 回到生成草稿

对应 demo：

uv run python test/langgraph_demo/05_loop_retry_demo.py

5.3 多节点汇总

一个节点后面可以同时跑多个节点。

例如：

输入主题
  ├─ 总结优点
  ├─ 总结风险
  └─ 总结建议
       ↓
    汇总结果

如果多个节点都写同一个字段，要告诉 LangGraph 怎么合并。

对应 demo：

uv run python test/langgraph_demo/06_parallel_summary_demo.py

5.4 Checkpoint 保存状态

使用 checkpoint 需要两步。

第一，compile 时传入 checkpointer：

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

第二，运行时传入 thread_id：

config = {
    "configurable": {
        "thread_id": "student-1001"
    }
}

result = graph.invoke(input_state, config=config)

thread_id 可以理解成这段流程或这段对话的编号。

取当前最新 checkpoint：

saved_checkpoint = graph.get_state(config)
print(saved_checkpoint.values)

这里要注意：

saved_checkpoint 不是普通 dict。
saved_checkpoint.values 才是保存下来的 State。

对应 demo：

uv run python test/langgraph_demo/07_checkpoint_state_demo.py
uv run python test/langgraph_demo/10_multiple_checkpoint_records_demo.py

5.5 stream 观察每一步输出

invoke 会等整张图跑完，再返回最终 State。

stream 可以在图运行过程中，看到每一步输出。

默认 stream 输出每个节点本次写入的内容。

for event in graph.stream(input_state):
    print(event)

这等价于常用的 updates 模式。

例如输出：

{'clean_question': {'cleaned_question': 'LangGraph 的 stream 有什么用？'}}
{'extract_keyword': {'keyword': 'LangGraph'}}
{'answer': {'answer': '你问的是 LangGraph。'}}

它的意思是：

clean_question 节点这一步写入了 cleaned_question；
extract_keyword 节点这一步写入了 keyword；
answer 节点这一步写入了 answer。

所以默认 stream 不是每次都返回完整 State。

它更像是在告诉你：

这一小步刚刚更新了什么。

如果想每一步都看到完整 State：

for state in graph.stream(input_state, stream_mode="values"):
    print(state)

values 模式会输出每一步执行后的完整 State。

例如：

{'question': '  LangGraph 的 stream 有什么用？  '}
{'question': '  LangGraph 的 stream 有什么用？  ', 'cleaned_question': 'LangGraph 的 stream 有什么用？'}
{'question': '  LangGraph 的 stream 有什么用？  ', 'cleaned_question': 'LangGraph 的 stream 有什么用？', 'keyword': 'LangGraph'}
{'question': '  LangGraph 的 stream 有什么用？  ', 'cleaned_question': 'LangGraph 的 stream 有什么用？', 'keyword': 'LangGraph', 'answer': '你问的是 LangGraph。'}

课堂里先掌握两个就够：

当前 LangGraph 还支持这些模式：

如果只是课堂演示：

先用 updates 看节点更新；
再用 values 看完整 State；
其他模式知道名字即可。

对应 demo：

uv run python test/langgraph_demo/11_stream_output_demo.py

6. 真实业务里常用的进阶能力

基础节点和边解决的是：

流程怎么走。

下面这些能力解决的是：

流程怎么在真实项目里稳定、安全、可追踪地运行。

6.1 Store：长期记忆

checkpointer 保存的是某个 thread_id 里的流程状态。

store 保存的是跨 thread_id 都能读取的长期数据。

可以这样记：

checkpointer：保存当前对话或当前流程的状态。
store：保存长期记忆或长期业务数据。

典型写法：

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = graph_builder.compile(store=store)

写入记忆时，需要 namespace。

namespace = ("student-1001", "memories")

runtime.store.put(
    namespace,
    "example_language_preference",
    {
        "memory": "用户希望以后尽量用 Java 举例。"
    },
)

这里有三个东西要分清楚：

namespace：记忆放在哪个分组。
key：这个分组里的哪一条记忆。
value：真正保存的记忆内容。

在上面的代码里：

("student-1001", "memories") 是 Python 里的 tuple。

tuple 可以先理解成：

用小括号包起来的一组值。

它里面有两个值：

第一个值：student-1001
第二个值：memories

在 Store 里，这个 tuple 表示分组路径。

可以理解成：

student-1001 / memories

namespace 可以多级：

("tenant-001", "student-1001", "memories")
("tenant-001", "student-1001", "preferences")
("tenant-001", "student-1001", "text_to_sql", "query_preferences")

可以理解成：

tenant-001 / student-1001 / memories
tenant-001 / student-1001 / preferences
tenant-001 / student-1001 / text_to_sql / query_preferences

tuple 和 list 有点像，都能放多个值。

简单区别是：

LangGraph Store 的 namespace 要求用 tuple。

所以这里写：

namespace = ("student-1001", "memories")

不要写成：

namespace = ["student-1001", "memories"]

读取记忆：

memories = runtime.store.search(namespace, limit=10)

这里的 memories 是查出来的结果列表。

这行代码可以翻译成：

去 student-1001 / memories 这个分组下面，
最多查询 10 条记忆，
然后把查出来的结果放到 memories 变量里。

所以：

namespace 是“去哪里找”；
memories 是“找出来的结果”。

注意：

InMemoryStore 只存在当前 Python 进程内。
程序停止后，数据就没有了。

对应 demo：

uv run python test/langgraph_demo/12_store_long_term_memory_demo.py

6.2 Runtime / context：运行时上下文

State 里放流程正在加工的数据。

context 里放这次运行时的外部条件。

例如：

节点可以读取 runtime.context：

def select_schema_node(
    state: QueryState,
    runtime: Runtime[RequestContext],
) -> QueryState:
    user_id = runtime.context.user_id
    allowed_tables = runtime.context.allowed_tables

这里的 runtime 是 LangGraph 执行节点时自动传进来的。

你不需要自己创建它。

它里面常用的内容有：

runtime.context：本次运行传进来的业务上下文。
runtime.store：如果 compile 时传了 store，可以在节点里读写长期记忆。

节点返回时，只返回要写回 State 的内容。

return {
    "selected_tables": selected_tables
}

不需要把 runtime 返回出去。

runtime 是给节点读取外部上下文的。
State 才是节点返回值要更新的数据。

也就是说，不要这样写：

return {
    "runtime": runtime,
    "selected_tables": selected_tables
}

应该只返回 State 需要更新的字段：

return {
    "selected_tables": selected_tables
}

创建图时，要声明 context 的结构：

graph_builder = StateGraph(
    QueryState,
    context_schema=RequestContext,
)

运行图时，把 context 传进去：

graph.invoke(
    {
        "question": "帮我查一下订单和学生数据。"
    },
    context=RequestContext(
        user_id="student-1001",
        tenant_id="yanque-school",
        allowed_tables=["order"],
    ),
)

这个例子里，用户问题里提到了“订单”和“学生”。

但 allowed_tables 只有 order。

所以节点只能选 order，不能选 student。

这就是 context 的价值：

模型或节点可以处理问题，
但权限边界由程序传入的 context 控制。

对应 demo：

uv run python test/langgraph_demo/13_runtime_context_demo.py

6.3 Interrupt：人工确认

有些流程不能自动往下走。

比如：

生成了高风险操作
  ↓
暂停
  ↓
等待人工确认
  ↓
同意：继续
拒绝：结束

节点里可以用 interrupt 暂停：

from langgraph.types import interrupt

review_result = interrupt({
    "message": "请审核这次操作",
    "detail": state["detail"],
})

执行到 interrupt(...) 时，图会暂停。

第一次 graph.invoke(...) 会返回类似：

{
    "__interrupt__": [
        Interrupt(
            value={
                "message": "请审核这次操作",
                "detail": "..."
            }
        )
    ]
}

这表示：

流程已经停在这里了，等外部把审核结果传回来。

恢复时用 Command(resume=...)：

from langgraph.types import Command

graph.invoke(
    Command(resume={
        "approved": False,
        "reason": "这次操作风险太高"
    }),
    config=config,
)

resume 传什么，interrupt(...) 那一行就收到什么。

可以传：

bool
str
dict
list
数字

例如传 bool：

approved = interrupt({"message": "是否放行？"})

graph.invoke(
    Command(resume=False),
    config=config,
)

继续执行时：

approved = False

例如传字符串：

decision = interrupt({"message": "approve / reject / rewrite"})

graph.invoke(
    Command(resume="rewrite"),
    config=config,
)

继续执行时：

decision = "rewrite"

例如传字典：

review_result = interrupt({"message": "请审核操作"})

graph.invoke(
    Command(resume={
        "approved": False,
        "reason": "风险太高",
        "suggestion": "改成只读查询"
    }),
    config=config,
)

继续执行时：

review_result = {
    "approved": False,
    "reason": "风险太高",
    "suggestion": "改成只读查询"
}

为什么还要传 config=config？

因为 config 里有 thread_id。

LangGraph 要靠它找到刚才暂停的那份 checkpoint。

Command(resume=...)：把人工结果传回暂停点。
config=config：告诉 LangGraph 恢复哪一次暂停的流程。

这里的 Command 可以理解成：

告诉 LangGraph 接下来怎么继续执行的一条控制指令。

平时 graph.invoke(input_state) 是从输入开始跑。

暂停后 graph.invoke(Command(resume=...), config=config) 是恢复旧流程。

这两个不是一回事。

对应 demo：

uv run python test/langgraph_demo/14_interrupt_human_in_loop_demo.py

6.4 Time Travel：回看并分叉

Time Travel 可以先理解成：

回到某一次流程执行过程中的中间状态，看一眼，必要时从那里重新跑。

先取历史快照：

history = list(graph.get_state_history(config))

它返回的是这个 thread_id 下保存过的历史 StateSnapshot。

可以理解成：

这条流程运行过程中，LangGraph 保存过的一组 checkpoint 快照。

它不是直接返回“节点列表”。

它返回的是一组历史状态。

常看的字段：

snapshot.values  # 当时保存的 State
snapshot.next    # 从这里继续时，下一步要执行的节点
snapshot.config  # 这份 checkpoint 的定位信息

其中 snapshot.config 里通常有：

thread_id：哪一条流程。
checkpoint_ns：checkpoint 命名空间，简单 demo 里通常是空字符串。
checkpoint_id：这一份 checkpoint 快照的唯一编号。

可以这样理解：

thread_id 决定是哪条流程；
checkpoint_id 决定是这条流程里的哪一个历史快照。

如果想从某个历史点分叉，可以先找到那个 checkpoint：

before_generate_sql = next(
    snapshot for snapshot in history
    if snapshot.next == ("generate_sql",)
)

这里的 ("generate_sql",) 是单元素 tuple。

它表示：

从这个 checkpoint 继续时，下一步是 generate_sql。

注意逗号：

("generate_sql",)  # 这是 tuple
("generate_sql")   # 这只是字符串

然后基于它创建新分叉：

fork_config = graph.update_state(
    before_generate_sql.config,
    values={
        "plan": "统计订单数量"
    },
)

这句话的意思是：

找到 before_generate_sql.config 指向的历史 checkpoint，
基于它创建一份新的 checkpoint，
并把 plan 改成“统计订单数量”。

这里为什么要传 before_generate_sql.config？

因为历史里有很多 checkpoint。

LangGraph 需要知道：

你到底想基于哪一个历史 checkpoint 来改 State。

before_generate_sql.config 就是那份历史 checkpoint 的地址。

values={...} 是你想覆盖或新增的 State 字段。

fork_config 不是 State 本身。

它是新 checkpoint 的地址。

里面通常只有：

thread_id
checkpoint_ns
checkpoint_id

真正的 State 保存在 checkpointer 里。

可以把它类比成数据库主键：

fork_config 里只有 checkpoint_id 等定位信息；
真正的数据在 checkpointer 里；
LangGraph 根据这个地址把数据取出来继续跑。

继续执行：

fork_result = graph.invoke(None, config=fork_config)

这里传 None 是因为不需要新的输入。

LangGraph 会根据 fork_config 找到那份新 checkpoint，然后从它的 next 节点继续往后跑。

比如这份 checkpoint 的 next 是：

("generate_sql",)

那它就会从 generate_sql 开始往后跑：

generate_sql -> answer -> END

它不会再重新执行前面的节点。

如果你选中的 checkpoint 的 next 是：

("answer",)

那就只会从 answer 往后跑。

一句话：

fork_config 是存档地址。
graph.invoke(None, config=fork_config) 是从这个存档继续执行。

对应 demo：

uv run python test/langgraph_demo/15_time_travel_demo.py

6.5 Fault tolerance：失败重试和降级

真实流程里，节点可能失败。

例如：

模型接口超时；
数据库连接失败；
外部服务返回 500；
工具调用不稳定。

LangGraph 可以给节点配置失败处理。

常见两层：

retry_policy：失败后自动重试。
error_handler：重试耗尽后，进入降级或补偿流程。

示例：

from langgraph.types import RetryPolicy

graph_builder.add_node(
    "call_data_service",
    call_data_service_node,
    retry_policy=RetryPolicy(
        max_attempts=3,
        retry_on=ConnectionError,
    ),
    error_handler=data_service_error_handler,
)

retry_policy 负责“再试几次”。

error_handler 负责“还是失败时怎么收场”。

上面这个配置可以拆开看：

RetryPolicy(
    max_attempts=3,
    retry_on=ConnectionError,
)

意思是：

如果节点抛出 ConnectionError，最多尝试 3 次。

如果 3 次都失败，就进入 error_handler。

error_handler 里可以返回普通 State 更新，也可以返回 Command 控制下一步。

例如：

from langgraph.types import Command


def data_service_error_handler(state, error):
    return Command(
        update={
            "data": "降级数据：外部服务暂时不可用"
        },
        goto="answer",
    )

这里的意思是：

把 data 更新成一段降级数据；
然后直接跳到 answer 节点继续。

所以 Fault tolerance 不是简单地“报错就结束”。

它更像是给流程准备一条兜底路线：

先重试；
重试还失败；
进入降级节点；
返回一个可解释的结果。

对应 demo：

uv run python test/langgraph_demo/16_fault_tolerance_demo.py

7. 课堂 demo 顺序

建议按下面顺序跑：

01_create_graph_demo.py             创建最小 LangGraph 流程
02_state_update_demo.py             State 如何被节点读写
03_simple_edge_demo.py              Edge 顺序执行
04_conditional_edge_demo.py         条件边
05_loop_retry_demo.py               循环重试
06_parallel_summary_demo.py         多节点汇总
07_checkpoint_state_demo.py         Checkpoint 和 State
08_student_task_graph_demo.py       综合练习：学生任务处理流程
10_multiple_checkpoint_records_demo.py  一个 Checkpointer 保存多份记录
11_stream_output_demo.py            stream 观察节点输出
12_store_long_term_memory_demo.py   Store 长期记忆
13_runtime_context_demo.py          Runtime/context
14_interrupt_human_in_loop_demo.py  Interrupt 人工确认
15_time_travel_demo.py              Time Travel 回看和分叉
16_fault_tolerance_demo.py          失败重试和降级

运行方式：

uv run python test/langgraph_demo/01_create_graph_demo.py

8. 本节总结

今天先掌握 LangGraph 的基础认识。

不用一开始就记很多 API。

先记住这些结论：

1. LangGraph 是用“图”的方式组织 AI 流程的框架。
2. Chain 适合固定顺序，Agent 适合工具选择，LangGraph 适合复杂流程控制。
3. State 保存流程中间数据。
4. Node 表示一个处理步骤。
5. Edge 表示下一步怎么走。
6. Conditional Edge 用来根据节点结果决定分支。
7. Checkpoint 可以保存流程状态和对话状态。
8. stream 可以观察每个节点的输出，让流程不再是黑盒。
9. Store 可以保存跨对话、跨 thread 的长期记忆。
10. Runtime/context 适合传入用户、租户、权限等业务上下文。
11. Interrupt 可以让危险操作暂停，等待人工确认后继续。
12. Time Travel 可以回看历史状态，也可以从某一步分叉重跑。
13. Fault tolerance 可以处理失败重试、超时和降级流程。
14. LangGraph 的价值，是让复杂 AI 业务流程更清楚、更可控、更容易调试。

最后用一句话总结：

LangGraph 不是让模型更自由，而是让复杂 AI 流程更有结构：每一步做什么、什么时候分支、状态保存在哪里、失败后怎么办，都可以被程序清楚地管理起来。

第九天_Text-to-SQL全流程图与讲解

# Text-to-SQL 全流程图与讲解

本文根据《第九天：Text-to-SQL 业务介绍与系统设计》整理，重点解释原教程结尾的端到端流程，并将原来的文本流程重绘为 Mermaid 流程图。

原教程：第九天_Text-to-SQL业务介绍.md

1. 核心结论

真实项目中的 Text-to-SQL 不是：

用户提问 → 大模型生成 SQL → 数据库执行

而是：

先判断能不能查
→ 再确认到底查什么
→ 只加载必要的指标和 Schema
→ 生成 SQL
→ 由程序和数据库决定是否允许执行
→ 将真实结果加工成用户能理解的答案

整套系统可以分为四个阶段：

1. 入口分流：判断请求是不是只读数据查询。
2. 语义规划：确定指标、时间、筛选、维度、表和关联路径。
3. 安全执行：生成 SQL，经过 SQL Guard 和 EXPLAIN 后，用只读账号执行。
4. 结果产品化：将原始结果转换成指标卡、表格、图表和可信文字总结。

此外，数据仓库、指标语义层和 Schema 目录属于整条流程的后台数据基础，并不是每次用户提问后才临时构建。

2. Text-to-SQL 端到端流程图（重绘版）

flowchart TB
    subgraph DATA["A. 数据准备层｜后台持续运行"]
        SRC["订单库｜支付库｜用户库｜商品库"]
        SYNC["ETL / CDC<br/>清洗、同步、统一业务键"]
        DW["分析数据仓库<br/>宽表、主题表、聚合表"]
        CATALOG["Schema 目录<br/>表说明、关键业务键、关联关系"]
        METRIC["指标语义层<br/>名称、同义词、公式、状态与时间口径"]

        SRC --> SYNC --> DW
        DW --> CATALOG
    end

    subgraph ENTRY["B. 入口分流｜先判断能不能查"]
        USER["用户输入自然语言问题"]
        ROUTER{"Router：请求属于哪一类？"}
        CHAT["普通问答<br/>GENERAL_CHAT"]
        OP["操作请求<br/>拒绝直接改库，引导业务页面"]
        OUT["超出范围<br/>说明能力边界"]
        ROUTE_CLARIFY["澄清请求类型<br/>想查数据还是了解规则？"]

        USER --> ROUTER
        ROUTER -->|普通问答| CHAT
        ROUTER -->|修改或删除数据| OP
        ROUTER -->|超出能力范围| OUT
        ROUTER -->|无法判断| ROUTE_CLARIFY
        ROUTE_CLARIFY --> ROUTER
    end

    subgraph PLAN["C. 语义规划｜先确定查什么"]
        RAG["RAG 召回候选资料<br/>指标定义、候选表目录、关联关系"]
        QUERY_PLAN["生成结构化 QueryPlan<br/>指标、表、Join、时间、筛选、分组、排序、Limit"]
        PLAN_GATE{"QueryPlan 完整且可校验？"}
        CLARIFY["每次澄清一个最关键缺口"]
        KNOWLEDGE_GAP["补充指标或 Schema 资料<br/>或明确当前无法查询"]
        DETAIL["加载已选表详细 Schema<br/>+ 已选指标口径 + 已选关联路径"]

        RAG --> QUERY_PLAN --> PLAN_GATE
        PLAN_GATE -->|关键要素缺失| CLARIFY
        CLARIFY --> QUERY_PLAN
        PLAN_GATE -->|候选知识不足或计划越界| KNOWLEDGE_GAP
        PLAN_GATE -->|READY| DETAIL
    end

    subgraph EXEC["D. 安全执行｜模型提方案，程序掌握执行权"]
        SQL_GEN["生成 SQL"]
        GUARD["SQL Guard<br/>单条 SELECT、白名单、权限、数据范围、脱敏、LIMIT"]
        EXPLAIN["执行 EXPLAIN<br/>检查索引、扫描行数、临时表和排序风险"]
        RISK_GATE{"执行风险是否可接受？"}
        REPAIR["携带结构化错误修复 SQL<br/>受最大重试次数限制"]
        NARROW["要求缩小时间范围、增加筛选<br/>或改查汇总结果"]
        DENY["拒绝执行并记录审计信息"]
        DB_EXEC["只读账号执行<br/>查询超时、并发限制、最大返回行数"]
        EXEC_GATE{"执行结果"}
        FAIL["明确失败原因或缺少的信息"]

        SQL_GEN --> GUARD --> EXPLAIN --> RISK_GATE
        RISK_GATE -->|可接受| DB_EXEC
        RISK_GATE -->|SQL 可修复且未超限| REPAIR
        REPAIR --> SQL_GEN
        RISK_GATE -->|风险偏高| NARROW
        NARROW --> CLARIFY
        RISK_GATE -->|越权或危险或风险过高| DENY
        DB_EXEC --> EXEC_GATE
        EXEC_GATE -->|失败可修复且未超限| REPAIR
        EXEC_GATE -->|失败不可修复或已超限| FAIL
    end

    subgraph RESULT["E. 结果产品化｜把数据库事实变成用户答案"]
        EMPTY["成功但结果为空<br/>如实生成空结果提醒"]
        NORMALIZE["程序标准化结果<br/>类型、格式、别名、单位、空值、脱敏、一致性"]
        VIZ["规则优先选择展示方式<br/>指标卡、表格、折线图、柱状图等"]
        SUMMARY["模型仅依据真实结果总结<br/>不编造原因、比例或结论"]
        PACKAGE["结构化返回<br/>query_summary、metric_cards、table_data、chart_config、answer、warnings"]
        UI["前端展示<br/>指标卡｜表格｜图表｜自然语言回答"]

        EMPTY --> PACKAGE
        NORMALIZE --> VIZ --> SUMMARY --> PACKAGE --> UI
    end

    ROUTER -->|DATA_QUERY| RAG
    CATALOG -.提供候选 Schema.-> RAG
    METRIC -.提供候选指标.-> RAG
    DETAIL --> SQL_GEN
    DW -.提供已选表 Schema.-> DETAIL
    DW -.只读查询.-> DB_EXEC
    EXEC_GATE -->|成功且有数据| NORMALIZE
    EXEC_GATE -->|成功但无数据| EMPTY

    classDef data fill:#f3e8ff,stroke:#7c3aed,color:#2e1065,stroke-width:1.5px;
    classDef entry fill:#e0f2fe,stroke:#0284c7,color:#082f49,stroke-width:1.5px;
    classDef plan fill:#fff7ed,stroke:#ea580c,color:#431407,stroke-width:1.5px;
    classDef exec fill:#fef2f2,stroke:#dc2626,color:#450a0a,stroke-width:1.5px;
    classDef result fill:#ecfdf5,stroke:#059669,color:#022c22,stroke-width:1.5px;
    classDef gate fill:#ffffff,stroke:#475569,color:#0f172a,stroke-width:2px;

    class SRC,SYNC,DW,CATALOG,METRIC data;
    class USER,CHAT,OP,OUT,ROUTE_CLARIFY entry;
    class RAG,QUERY_PLAN,CLARIFY,KNOWLEDGE_GAP,DETAIL plan;
    class SQL_GEN,GUARD,EXPLAIN,REPAIR,NARROW,DENY,DB_EXEC,FAIL exec;
    class EMPTY,NORMALIZE,VIZ,SUMMARY,PACKAGE,UI result;
    class ROUTER,PLAN_GATE,RISK_GATE,EXEC_GATE gate;

读图方式

• 紫色区域：后台数据基础，不是每次请求都重新执行。
• 蓝色区域：入口分流，决定请求是否进入数据库查询。
• 橙色区域：语义规划，把模糊问题变成可校验的查询合同。
• 红色区域：SQL 安全、性能检查和只读执行。
• 绿色区域：将数据库原始结果转换成面向用户的答案。
• 菱形节点：条件判断，会产生分支或循环。
• 虚线：数据或知识供给关系，不代表普通的串行调用顺序。

3. 各阶段详细讲解

3.1 数据准备层：为查询提前准备稳定的数据基础

微服务系统的订单、支付、用户、商品等数据通常分散在不同业务库。如果每次用户提问后再临时跨库查询，会产生以下问题：

• 跨库关联复杂；
• 查询性能不可控；
• 容易影响线上交易系统；
• 权限和敏感字段难以集中治理；
• 业务表结构变化会直接影响模型。

因此，常见方案是通过 ETL 或 CDC，把需要分析的数据持续同步到统一的数据仓库：

业务库负责在线交易
数据仓库负责统计分析
Text-to-SQL 只查询分析库

数仓中可以提前整理订单、支付、退款、产品和用户维度，形成宽表、主题表或聚合表。指标语义层和 Schema 目录也围绕数仓维护。

需要特别注意：数据同步是后台持续运行的基础设施，不是用户每提问一次就重新同步一次。

3.2 Router：先判断请求是否应该查询数据库

用户可能在同一个对话框中提出完全不同的请求：

Router 只负责稳定分类，不负责生成 SQL，也不负责确认指标口径。它可以由规则、轻量模型或二者组合实现。

这一层的核心价值是：

不是所有自然语言请求都应该变成 SQL，尤其不能让操作请求进入只读查询链路。

3.3 RAG：检索候选资料，而不是替系统做最终决定

Router 确认请求为 DATA_QUERY 后，RAG 从两类知识库中召回资料：

例如用户问“线上课程退款额是多少”，RAG 可能召回：

候选指标：退款成功金额
计算规则：SUCCESS 退款单的 refund_amount 求和
时间字段：refund_success_time
核心表：order_refund

候选表：
- order_refund
- order_payment
- order_product

候选关联：
order_refund.payment_order_no -> order_payment.order_no
order_payment.product_id -> order_product.id

RAG 只负责找回“可能相关”的资料。真正使用哪个指标、哪些表和哪条关联路径，由下一步查询规划决定。

3.4 QueryPlan：把模糊问题转换成可校验的查询合同

查询规划阶段应输出结构化 QueryPlan，而不是一段自由文本。典型字段包括：

selected_metric：查询指标
selected_tables：需要的表
join_path：表关联路径
time_range：时间范围
filters：筛选条件
group_by：分组维度
order_by：排序规则
limit：返回数量
result_columns：结果列的别名、角色和单位
status：READY 或 NEEDS_CLARIFICATION

程序应验证：

• 指标是否来自候选指标；
• 表和关联关系是否来自候选 Schema；
• 时间、筛选、分组和排序是否合法；
• 结果列是否具有唯一别名和明确语义；
• 当前计划是否已经达到 READY 状态。

这份 QueryPlan 是模块之间的合同：

• 规划模型负责决定“查什么”；
• SQL 模型负责决定“怎么查”；
• 程序负责校验和执行；
• 结果模块负责按既定语义展示。

3.5 查询要素检查与澄清循环

一次数据查询通常包含六类要素：

1. 查询对象或指标；
2. 时间范围；
3. 筛选条件；
4. 计算方式；
5. 分组维度；
6. 排序和返回数量。

例如“昨天订单量是多少”存在关键歧义：

• 昨天创建的订单数；
• 昨天支付成功的订单数。

系统不能擅自猜测，应追问：

你想统计昨天创建的订单数，还是昨天支付成功的订单数？

用户回答后，系统更新 QueryPlan，再进行校验。每次只问一个最影响结果的问题，并设置最大澄清次数。

3.6 两阶段 Schema：先选表，再加载详细字段

真实系统可能有数百张表和成千上万个字段。把所有完整 Schema 一次性塞给模型，会导致：

• 上下文过长；
• 成本更高、响应更慢；
• 无关字段干扰模型；
• 更容易选错同名字段；
• 关键业务口径被淹没。

因此推荐分两阶段：

第一阶段：
用户问题 + 候选指标 + 轻量表目录 + 关联关系
→ 选择需要的表

第二阶段：
已选表的完整字段 + 已选指标口径 + 已选关联路径 + QueryPlan
→ 生成 SQL

可以把它理解为：先看图书目录决定借哪几本书，再取出这些书的具体内容。

3.7 SQL 生成：只负责把已确认计划翻译成 SQL

到了 SQL 生成阶段，模型收到的是已经确认的：

1. QueryPlan；
2. 已选表的详细 Schema；
3. 指标口径；
4. 表关联路径；
5. 数据库方言和输出约束。

例如：

SELECT
    product_name,
    SUM(refund_amount) AS refund_amount
FROM dwd_order_payment_refund
WHERE refund_status = 'SUCCESS'
  AND teaching_mode = 'ONLINE'
  AND refund_success_date >= CURRENT_DATE - INTERVAL 30 DAY
GROUP BY product_id, product_name
ORDER BY refund_amount DESC
LIMIT 10;

此时模型只解决“如何查询”，不应该重新决定“退款额是什么意思”。

3.8 SQL Guard：模型可以提方案，但程序拥有最终执行权

SQL Guard 不能依赖模型自己审核自己，而应由程序和数据库确定性执行。

静态与权限检查

• 只允许单条 SELECT；
• 禁止写入和结构变更语句；
• 校验库、表、字段白名单；
• 拒绝多语句和越权查询；
• 注入租户、组织或角色的数据范围；
• 对敏感字段进行拒绝或脱敏；
• 添加最大返回行数和 LIMIT。

使用 EXPLAIN 检查执行风险

程序在真正查询前执行：

EXPLAIN SELECT ...

重点关注：

• key：计划使用的索引；
• type：访问方式，大表出现 ALL 需要警惕；
• rows：预计扫描行数；
• Extra：是否出现临时表或额外排序；
• 多表 Join 的关联字段是否有索引。

程序可以做三类决策：

风险可接受 → 放行
风险偏高   → 缩小时间范围、增加筛选或改查汇总
风险过高   → 拒绝执行并记录审计信息

核心原则：

模型可以提出查询方案，但程序必须拥有最终执行权。

3.9 只读执行与有限重试

SQL Guard 通过以后，还需要数据库层兜底：

• 使用只读数据库账号；
• 设置查询超时；
• 限制连接池并发；
• 设置最大返回行数；
• 记录用户身份、SQL、执行计划摘要、耗时和扫描行数。

执行结果分为三类：

权限不足、指标口径不清和风险过高，不能靠反复生成 SQL 碰运气。超过重试上限后，系统应明确说明失败原因或缺少的信息。

3.10 结果后处理：数据库原始结果还不是用户答案

数据库可能只返回：

product_name | refund_amount
课程 A        | 15600.00
课程 B        | 9800.00

程序还需要完成：

• 日期、金额和百分比格式化；
• 空值与除零处理；
• 字段别名和单位补充；
• 敏感字段脱敏；
• 行数和列数控制；
• 校验结果列是否与 QueryPlan 一致。

然后根据数据形态选择展示方式：

模型最后只能基于真实结果生成文字总结，不能补造数据库中没有的原因、比例或业务结论。

3.11 结构化返回

最终接口不应只返回一段模型文字，而应返回可组合展示的结构：

query_summary：用户问题和已确认口径
metric_cards：核心指标卡
table_data：明细或分组结果
chart_config：受支持的图表配置
natural_language_answer：基于真实结果的总结
warnings：空结果、截断、范围过大等提示

同一份结果可以分别用于：

• 管理后台的表格和图表；
• 对话框中的简短结论；
• 标准化数据导出；
• 审计页面中的 SQL 和执行摘要。

4. 用一个问题贯穿整条流程

用户问题：

最近 30 天线上课程按产品统计退款额 Top 10。

5. 三个关键闭环

5.1 澄清闭环

查询计划不完整
→ 询问用户一个关键问题
→ 更新 QueryPlan
→ 再次校验

解决的是自然语言歧义。

5.2 SQL 修复闭环

生成 SQL
→ SQL Guard 或执行失败
→ 携带安全、结构化的错误反馈
→ 有限次数修复
→ 再次完整校验

解决的是模型可能生成错误 SQL，但必须设置最大重试次数。

5.3 结果可信闭环

数据库真实结果
→ 程序标准化与一致性检查
→ 模型严格依据结果解释
→ 结构化展示

保证模型只能解释事实，不能创造数据。

6. 映射到 LangGraph

如果使用 LangGraph 实现，每个流程框都可以映射为图中的组件：

推荐保存的核心 State 示例：

user_question
route_type
retrieved_metrics
candidate_tables
query_plan
selected_schema
generated_sql
guard_result
explain_summary
retry_count
query_result
response_payload

主要循环包括：

1. Planner → Clarifier → Planner；
2. SQL Generator → SQL Guard → SQL Generator；
3. Executor → SQL Repair → SQL Guard → Executor。

每个循环都必须具备明确的进入条件、最大次数和退出策略。

7. 职责边界

8. 最终记忆

可以用一句话记住整套流程：

大模型负责理解、规划、生成和解释；程序负责校验、权限、执行控制和结果标准化；数据库只通过受限、只读的方式提供事实。

再压缩成五个关键词：

分流 → 规划 → 护栏 → 执行 → 展示