结构化数据与 SQL 检索

更新 2026-10-05

我的学习 →

正在读取学习记录…

当客户问“我有哪些待发货订单”，答案来自订单表。表中每一行是一笔订单，列分别存订单号、客户、状态与金额。这类结构化数据有明确字段，适合用 SQL 按条件查找、计数或求和。SQL 是向数据库描述要读取或修改哪些数据的语言。

自然语言问题通常省略了数据库字段名。程序要先把“待发货”对应到状态值，再把当前登录客户对应到客户标识；查出的行才是回答依据。像去柜台查订单，柜员先核对身份，再按状态查记录。

先把客户问题落到明确条件

固定数据库有三笔订单：u1的待发货订单101、u1的已发货订单102、u2的待发货订单103。当前客户是u1，因此“我的待发货订单”应只返回101。

查询意图是对请求的结构化表示，例如“列订单，状态为pending”。pending表示待发货，shipped表示已发货。示例识别两句固定问题，并用预先写好的SQL模板执行；接入语言模型时，也可以让模型提议这个意图，再由程序验证后选择模板。

订单问题到参数化查询：问题：我的待发货订单，登录身份：u1；固定意图 → pending，可信身份 → customer；WHERE customer = ? AND status = ?，绑定参数：u1，pending；返回订单101：120元，过滤u2订单103 — 图1：订单问题到参数化查询。

跟着固定样本运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l5-sql" -Force
cd "F:\学习\l5-sql"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l5-sql，将下面完整代码保存为 sql_demo.py。

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l5-sql" -Force
cd "F:\学习\l5-sql"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""从固定自然语言意图构造参数化订单查询。"""
import sqlite3

INTENTS = {
    "我有哪些待发货订单": "pending",
    "我有哪些已发货订单": "shipped",
}
QUERY = ("SELECT id, amount FROM orders "
         "WHERE customer = ? AND status = ? ORDER BY id LIMIT 20")


def database():
    # 内存数据库便于每次运行从同一组订单开始。
    connection = sqlite3.connect(":memory:")
    connection.execute(
        "CREATE TABLE orders(id INTEGER, customer TEXT, status TEXT, amount INTEGER)"
    )
    connection.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", [
        (101, "u1", "pending", 120),
        (102, "u1", "shipped", 80),
        (103, "u2", "pending", 999),
    ])
    connection.commit()
    # 演示查询阶段开启只读，阻止后续意外写入。
    connection.execute("PRAGMA query_only = ON")
    return connection


def retrieve(connection, question, customer):
    if question not in INTENTS:
        raise ValueError("请使用待发货或已发货订单查询")
    # 问题决定状态；客户身份由调用方传入，不能从问题文字覆盖。
    status = INTENTS[question]
    return connection.execute(QUERY, (customer, status)).fetchall()


if __name__ == "__main__":
    with database() as connection:
        for question in INTENTS:
            rows = retrieve(connection, question, "u1")
            print(f"{question}：" + "、".join(
                f"订单{order_id}，金额{amount}元" for order_id, amount in rows
            ))
        try:
            retrieve(connection, "删除全部订单", "u1")
        except ValueError as error:
            print("未执行：" + str(error))

在 PowerShell 中进入保存目录并运行：

cd "F:\学习\l5-sql"
ls
python -X utf8 sql_demo.py

输出为：

我有哪些待发货订单：订单101，金额120元
我有哪些已发货订单：订单102，金额80元
未执行：请使用待发货或已发货订单查询

第一行只包含101，u2的103被客户条件过滤。第二行返回已发货的102。第三句没有匹配查询意图，因此程序在执行SQL前返回说明，表里的记录保持原样。

? 是参数占位符。SQL结构固定，客户与状态通过参数单独传给SQLite，数据库把它们当成值。即便客户标识里含引号或SQL关键词，也不会变成查询条件的一部分。

从意图到结果，每一步负责什么

INTENTS 将已识别的句子映射为状态；QUERY 决定可查询的表、列和行数；retrieve 将登录客户与状态绑定到占位符。WHERE按条件过滤，ORDER BY按订单号排序，LIMIT将返回行数限制为20。amount在样本中用整数元表示，避免金额在浮点运算中产生小数误差。

SQLite 是Python标准库可连接的小型关系数据库。这里使用内存数据库，关闭连接后数据消失；需要保留订单时可以打开文件路径或连接业务数据库。PRAGMA query_only 是SQLite连接的只读开关，它覆盖本示例的查询阶段。生产系统还应使用受限数据库账户，并由服务核验调用方身份。

模型生成SQL时，怎样保留查询约束

模型可以把“本月总金额”翻译成求和意图，但字段名、表名和统计范围仍需验证。参数绑定适用于值；表名和列名通常要从程序允许的集合选择。直接把模型生成的整段SQL交给高权限连接，会同时放开查询范围和操作范围。

列订单、计数与求和的结果不同。SUM(amount)合计金额，COUNT(*)计行数；加入一对多表连接后，一笔订单可能重复出现，求和前要确认每行的含义。对缺字段、未识别的状态或跨客户问题，应先澄清意图，再发查询。

看一次客户切换

把主程序中的两处 "u1" 改成 "u2"，运行同一命令。待发货行变成订单103、999元，已发货行没有订单内容。空结果表示这个客户没有符合条件的行；它和数据库连接失败、意图无法识别是不同情况。

现象	检查位置
混入其他客户订单	WHERE中的客户条件与可信身份来源
合计金额异常增大	表连接是否重复订单
状态名称查不到	自然语言状态到数据库值的映射
查询能修改订单	连接权限、模板集合和只读设置

检查实现中的六种情况

在同一目录将下面测试保存为 test_sql_demo.py。测试覆盖正文中的正常路径与边界条件。

测试代码：test_sql_demo.py

"""检查身份过滤、参数绑定、未知意图与只读约束。"""
import sqlite3
import unittest
from sql_demo import database, retrieve


class SqlTests(unittest.TestCase):
    def setUp(self):
        self.db = database()

    def tearDown(self):
        self.db.close()

    def test_pending(self):
        self.assertEqual(retrieve(self.db, "我有哪些待发货订单", "u1"), [(101, 120)])

    def test_shipped(self):
        self.assertEqual(retrieve(self.db, "我有哪些已发货订单", "u1"), [(102, 80)])

    def test_other_customer(self):
        self.assertEqual(retrieve(self.db, "我有哪些待发货订单", "u2"), [(103, 999)])

    def test_parameter_is_value(self):
        self.assertEqual(retrieve(self.db, "我有哪些待发货订单",
                                  "u1' OR 1=1 --"), [])

    def test_unknown_intent(self):
        with self.assertRaises(ValueError):
            retrieve(self.db, "删除全部订单", "u1")

    def test_read_only(self):
        with self.assertRaises(sqlite3.OperationalError):
            self.db.execute("DELETE FROM orders")


if __name__ == "__main__":
    unittest.main()

cd "F:\学习\l5-sql"
python -X utf8 -m unittest -v test_sql_demo

应显示6项通过。将主程序恢复到上面的固定输入后，可以再次对照完整输出。

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“结构化数据与 SQL 检索”，哪种做法更稳妥？

对应知识点：L5-sql

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L5-sql

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L5-sql

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L5-sql

你的回答