过去半年我一直在折腾一个很现实的问题怎么让不会写 SQL 的人也能直接对着数据库问问题。最终落地的是一个用 DeepSeek 做大模型推理、SQLite 做数据存储的轻量 Text2SQL 查询助手——输入一句中文比如“上个月销售额最高的三个商品”它自动生成 SQL、在 SQLite 引擎里执行再把结果用表格形式返回。整个过程不用写一行 SQL也不用部署重型数据库服务。这个项目是典型的 LLM 落地形态不需要训练、不需要私有化部署大模型只需要把数据库结构和查询约束交代给模型它就能输出可执行 SQL。整套代码加起来不到 200 行特别适合第一次认真做 Text2SQL 的人拿来当模板也适合想验证大模型在生成式查询场景里到底稳不稳的开发者做一轮实测。下面我把从零搭建的完整过程讲清楚包括提示词设计、SQL 安全校验、错误纠错循环以及我实际踩过的坑。1. 为什么选 Text2SQL 作为 LLM 落地项目整体思路是什么1.1 一句话描述项目形态这个查询助手的核心链路可以概括为用户输入自然语言问题 → 系统把 SQLite 的 schema 注入提示词 → 调用 DeepSeek 生成 SQL → 程序提取并校验 SQL → 在 SQLite 中执行 → 返回查询结果。如果生成的 SQL 报错程序会把错误信息回传给模型让它基于错误修正后重新生成。听起来很顺但真正实施起来有几个难点模型可能生成不存在的字段名可能写出 SQLite 不支持的语法甚至可能生成插入、删除这类危险语句。所以这个项目表面上是在调 API本质上是在做三件事让模型“看懂”库结构、让输出“规规矩矩”变成纯 SQL、让执行过程“万无一失”不会破坏数据。1.2 为什么选 DeepSeek SQLite 这套组合先聊模型选型。Text2SQL 对模型的要求是中文理解能力强、SQL 语法知识扎实、输出稳定。DeepSeek 的开放接口兼容 OpenAI 调用格式代码上接入成本很低而且在实际测试中它对中文业务问题的理解明显比通用模型更贴近中文语境生成的 SQL 也习惯用注释和 LIMIT 这类安全写法。再聊数据库选型。SQLite 是单文件数据库不需要安装服务端一个.db文件就能承载一张完整的业务表结构。对于个人工具、内部小助手这类场景它简直是绝配轻、免维护、随时可以复制走。相比 MySQL、PostgreSQLSQLite 还能用只读 URI 模式打开天然适合给 LLM 生成的查询语句做执行沙箱。这套组合的另一个优势是成本极低。单次查询请求通常只消耗几千 token跑几百次测试的成本也就在几块钱量级可以放心反复实验。相比接一个独立的数据库服务开发阶段省掉的不只是部署还有连接配置、账号权限、网络安全这一堆事。1.3 整体流程设计一次查询请求是怎么走通的我把它拆成五个环节取 schema、拼提示词、调模型、执行 SQL、反馈纠错。前两个环节是准备中间一个是核心后两个是保障。取 schema 这一步很多人会偷懒觉得写死几张表名就行了。实际上模型能不能生成正确 SQL很大程度上取决于它能不能看到完整字段名和类型。我选择通过sqlite_master系统表自动读取建表语句这样数据库结构一变提示词也会自动跟上。拼提示词是最考功底的一步。要把数据库结构、业务字段说明、输出格式限制、安全约束全部塞进一段上下文里。提示词不是越长越好而是要让模型明确三件事回答范围是什么、输出格式是什么、禁区是什么。调模型和执行 SQL 之间隔着一道安全检查。LLM 生成的内容本质上不可信不能直接丢给数据库执行。我加了两层防护关键字黑名单和只读打开连接。前者挡掉语义层面的危险语句后者从数据库层面保证即使漏网也改不了任何数据。整个设计里最让我得意的是纠错循环。第一次生成的 SQL 很可能会因为字段名拼错、函数用错而执行失败。传统的做法是让用户重新问一遍但我在执行出错时会把 SQLite 返回的错误信息拼回对话上下文让模型直接改写。这样一次失败的查询往往能在第二次、第三次尝试内自我修复用户体验好很多。2. 动手前准备示例库、API 接入与 schema 注入2.1 建一个能说明问题的小型 SQLite 示例库为了测试多表关联和聚合查询我建立了一个非常接近真实业务的三表结构用户表、商品表、订单表。订单表通过外键关联用户和商品这样可以很方便地演示 JOIN、GROUP BY、子查询这类常见需求。建表 SQL 如下DROP TABLE IF EXISTS users; CREATE TABLE users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER, city TEXT, created_at TEXT NOT NULL ); DROP TABLE IF EXISTS products; CREATE TABLE products ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, category TEXT, price REAL, stock INTEGER ); DROP TABLE IF EXISTS orders; CREATE TABLE orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, product_id INTEGER NOT NULL, quantity INTEGER NOT NULL, order_time TEXT NOT NULL, FOREIGN KEY (user_id) REFERENCES users(id), FOREIGN KEY (product_id) REFERENCES products(id) );有一点需要注意SQLite 本身对字段类型非常宽容所以时间字段我统一存成YYYY-MM-DD HH:MM:SS格式的字符串。这样模型在做时间范围查询时可以用strftime或字符串比较不容易出错。插入数据也不必太复杂每个表放 20 条左右就够测试了。记得让订单分布在不同的月份和城市因为“上个月”这类带时间条件的查询是 Text2SQL 测评的重灾区数据设计时就要故意制造可查询的时间跨度。2.2 DeepSeek API 的接入方式DeepSeek 提供了 OpenAI 兼容的接口所以我直接用openaiPython 库来调用避免另外封装一套请求逻辑。所有需要登录的密钥都通过环境变量读取绝不硬编码在脚本里。import os from openai import OpenAI client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) def ask_deepseek(messages, temperature0.0): resp client.chat.completions.create( modeldeepseek-chat, messagesmessages, temperaturetemperature, ) return resp.choices[0].message.content把温度设置成 0 是我反复测试后确定的。Text2SQL 不是创意写作它需要确定性输出。温度一旦调高模型可能用不同的语法表达同一个查询这会给后续纠错增加不必要的变量。如果你不想引入 openai 库直接用requests调官方接口也是可行的但openai库在消息构造、超时处理、错误信息这些方面已经封装得很好我建议直接用它。2.3 schema 自动注入机制模型不知道你的库长什么样所以每次查询前都要把结构信息喂给它。手动写一段固定的 schema 描述很容易但数据库一变就维护不过来了。我的做法是从 SQLite 系统表读取真实建表语句再组装成一段纯文本。import sqlite3 def get_schema(db_path): conn sqlite3.connect(db_path) rows conn.execute( SELECT sql FROM sqlite_master WHERE typetable AND sql IS NOT NULL ).fetchall() conn.close() return \n\n.join(r[0] for r in rows)这样返回的 schema 是原始的 CREATE TABLE 语句包含字段名、类型、外键约束。模型对这种结构化文本的吸收能力很强几乎不需要再做格式化。唯一要补充的是业务说明比如“price 单位是元”“order_time 是下单时间”。我会把这些注释追加到 schema 末尾而不是直接改建表语句保持数据库文件干净。实际测试下来schema 加上一句“请基于上面的表结构生成 SQLite 方言的 SQL”就能明显压低模型使用错误表名和字段名的概率。这一步做好后面所有环节都会省心很多。3. 核心代码逐段拆解提示词、安全校验与纠错循环3.1 提示词模板是如何让模型稳定输出 SQL 的提示词设计是这个项目最核心的环节。一开始我写得很简陋就一句“把下面的话转成 SQL”结果模型输出五花八门有的带解释有的用 MySQL 语法有的直接拒绝回答。后来我总结出一套有效的模板核心是四项约束。SYSTEM_PROMPT 你是一个精通 SQLite 的数据库查询助手。 请根据数据库 schema 和用户问题生成可直接执行的 SQL 查询。 要求 1. 只输出 SQL 语句本身不要输出任何解释、代码块标记。 2. 只允许 SELECT 查询禁止 INSERT、UPDATE、DELETE、DROP 等语句。 3. 使用 SQLite 语法不要使用其他数据库的专用函数。 4. 查询结果默认加上 LIMIT 50。 5. 如果问题与数据库无关或无法用提供的 schema 回答输出 SQL_ERR: 无法回答。 def build_messages(user_question, schema, error_hintNone): user_content f数据库 schema:\n{schema}\n\n用户问题: {user_question} messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_content}, ] if error_hint: messages.append({ role: assistant, content: error_hint[sql] }) messages.append({ role: user, content: f你生成的 SQL 执行报错{error_hint[error]}\\n请修正后重写。 }) return messages第一条约束解决了“模型话多”的问题。你可能会觉得让模型输出解释更友好但对程序来说解析一段可能有代码标记和废话的文本远比解析纯 SQL 麻烦。我宁可让模型输出裸 SQL再由程序负责展示结果。第二条约束是安全底线。模型本身没有恶意但它可能因为用户问题描述不当而生成危险语句。直接在提示词里声明禁区能从源头过滤一大半风险。第三、四条约束属于方言限定。SQLite 的日期函数、字符串处理方式和 MySQL、PostgreSQL 差别很大提前声明可以避免模型想当然地写出DATE_FORMAT这类函数。3.2 SQL 安全校验为什么 LLM 的输出不能直接执行即使提示词说了“只允许 SELECT”我也在代码里加了独立校验。原因很简单提示词是对模型的软约束不是硬保证。模型可能因为上下文太长而忽略限制也可能被一种叫“注入攻击”的手段绕过。我的校验分为三层。第一层去掉首尾空白、将 SQL 转成小写后检查是否包含insert、update、delete、drop、alter、attach、pragma这些关键词。第二层检查第一个非空语句是否以select或with开头。第三层用只读 URI 模式连接 SQLite从数据库层保证任何写操作都会抛出错误。import os import re BANNED_KEYWORDS [insert, update, delete, drop, alter, attach, pragma] def extract_sql(text): blocks re.findall(r(?:sql)?\\s*(.*?), text, re.S) if blocks: return blocks[0].strip() if text.startswith(SQL_ERR): return None return text.strip() def check_sql(sql): if not sql: return False, 空 SQL low sql.lower() for kw in BANNED_KEYWORDS: if re.search(r\\b kw r\\b, low): return False, f包含危险关键词 {kw} if not re.match(r^(select|with)\\b, low): return False, 必须以 SELECT 或 WITH 开头 return True, def execute_readonly(db_path, sql): uri ffile:{os.path.abspath(db_path)}?modero conn sqlite3.connect(uri, uriTrue, timeout10) try: cur conn.execute(sql) cols [d[0] for d in cur.description] if cur.description else [] rows cur.fetchall() return cols, rows finally: conn.close()我用了一个很实用的技巧如果 SQL 里包含注释或者多余空白正则也能提取干净。另外execute_readonly返回的不仅包含结果行还包含列名。这样在命令行里展示时可以直接拿列名做表头省得再查一次PRAGMA table_info。3.3 带纠错的主流程一次查询失败怎么办第一次生成的 SQL 执行失败太常见了。我的处理方法是把 SQL 和错误信息一起回传给模型让它看到自己写的代码和数据库的真实反应然后要求它修正。这相当于给模型一个“现场调试”的机会。def text2sql(db_path, question, max_retries2): schema get_schema(db_path) messages build_messages(question, schema) for attempt in range(max_retries 1): raw ask_deepseek(messages) sql extract_sql(raw) if sql is None: return {error: 模型无法生成 SQL, raw: raw} ok, msg check_sql(sql) if not ok: messages build_messages(question, schema, {sql: sql, error: msg}) continue try: cols, rows execute_readonly(db_path, sql) return {sql: sql, columns: cols, rows: rows} except sqlite3.Error as e: messages build_messages(question, schema, {sql: sql, error: str(e)}) return {error: 重试次数用尽, sql: sql}这里有个细节值得多说两句重试时不是简单地把错误信息追加到原消息尾部而是完整重建 message 列表并在其中模拟一段“助手生成了 SQL用户反馈了报错”的对话。这样做的好处是模型能明确看到自己的上一次输出而不是在越来越长的上下文里迷失。实测中绝大多数语法错误和字段名错误都能在第一次纠错内解决。3.4 一个可以直接跑的命令行主循环把上面的函数组合起来就是一个完整的命令行查询助手。它支持exit退出每次输入问题都会打印 SQL 和查询结果。def main(): db_path shop.db print(Text2SQL 查询助手已启动输入 exit 退出。) while True: question input(\\n问题: ).strip() if question.lower() in (exit, quit): break if not question: continue result text2sql(db_path, question) if error in result: print(错误:, result[error]) continue print(\\nSQL:, result[sql]) if result[columns]: print(\\t.join(result[columns])) for row in result[rows]: print(\\t.join(str(c) for c in row)) if __name__ __main__: main()整个脚本文件大概 150 行。你如果只想快速验证效果把这段代码存成text2sql_assistant.py再设置好环境变量就能启动。别急着加界面、加日志、加权限控制轻量才是这个项目的定位。等验证完模型效果再去扩展 UI 和配套能力也不迟。4. 实测效果与几个值得注意的点4.1 一组典型查询的输入与输出对比我把脚本跑起来后连续问了十多个问题包括单表过滤、聚合、多表关联、时间范围、排序取前几。下面是几组有代表性的结果自然语言问题生成的 SQL节选执行结果每个城市的用户数量是多少SELECT city, COUNT(*) AS cnt FROM users GROUP BY city ORDER BY cnt DESC正常返回上个月销量最高的三个商品SELECT p.name, SUM(o.quantity) AS total FROM orders o JOIN products p ON o.product_id p.id WHERE strftime(%Y-%m, o.order_time) strftime(%Y-%m, now, -1 month) GROUP BY p.name ORDER BY total DESC LIMIT 3正常返回哪些商品价格超过100元但库存不足20件SELECT name, price, stock FROM products WHERE price 100 AND stock 20正常返回找出购买了商品最多的前5个用户子查询 JOIN自动处理聚合和排序正常返回昨天每个类别的销售额strftime(%Y-%m-%d, o.order_time) strftime(%Y-%m-%d, now, -1 day)结合 JOIN正常返回让我意外的是模型对“上个月”“昨天”这类相对时间理解得相当准直接用了strftime(%Y-%m, now, -1 month)这种写法而不是我预想中的死日期。这说明只要 schema 里字段命名清晰模型完全可以把自然语言的时间表达映射成 SQLite 方言。4.2 实测中的效果边界模型不是万能的。我试过一个问题“谁的订单金额最高”。这个描述有歧义可以指单个订单金额最高也可以指累计消费金额最高。模型默认选择了累计求和但这未必是提问者想要的。这类模糊查询没有标准答案需要在提示词里追加“如果问题不明确请列出多种解释”的规则或者让用户补充条件。另一个边界是复杂嵌套查询。比如“找出购买过所有超过100元商品的用户”这种带全称量词的语义模型偶尔会生成错误的逻辑结构。我建议遇到这种问题时把问题拆成多个步骤查询而不是期望模型一步到位。4.3 成本与性能控制成本方面单次查询包含 schema 和 SQL 结果大约消耗 1000 到 2000 token。一次完整的纠错循环可能到 4000 token。就算连续测试 500 个问题成本也很低完全可以接受。性能方面真正的瓶颈不在模型调用而在每次请求都要重新获取 schema。虽然这条语句执行很快但反复读取也不优雅。我后来加了一个简单的模块级缓存_schema_cache {} def get_schema_cached(db_path): if db_path not in _schema_cache: _schema_cache[db_path] get_schema(db_path) return _schema_cache[db_path]如果你的数据库结构很少变动这个缓存非常管用能省掉每次读取和拼接的开销。同时建议给 OpenAI 客户端设置一个合理的timeout和max_retries避免网络抖动时整个命令行卡死。5. 常见问题排查与避坑记录5.1 问题速查表症状原因分析解决方案模型输出带解释文字SQL 提取失败提示词约束不足在提示词里强调“只输出 SQL 本身”并用正则提取代码块生成的 SQL 在 SQLite 中报语法错误模型惯性写了 MySQL/PostgreSQL 语法提示词里明确“使用 SQLite 语法”利用纠错循环回传错误查询结果明显错误但不是 SQL 报错字段语义理解偏差在 schema 后补充业务注释或把问题改得更具体查出来的数据为空时间格式或过滤条件不匹配检查数据里的时间字段是否与strftime格式化结果一致模型把“价格超过100元”理解成“价格低于100”否定词或多条件组合理解偏差换一种更直接的说法或拆成两个查询对比5.2 我实际踩过的几个坑第一个坑是时间字段的格式。我最初插入数据时用的是2024-3-5这样不补零的格式导致strftime(%Y-%m, order_time)匹配不到数据。排查了很久才发现是数据格式问题。建议统一用YYYY-MM-DD HH:MM:SS否则模型生成的时间条件经常对不上。第二个坑是模型把ORDER BY和LIMIT的先后顺序写反。这不是大问题SQLite 会直接报语法错误纠错循环能自动修复。但如果你的场景对延迟敏感可以在提示词里加一个 few-shot 示例让模型看一眼正确写法。第三个坑是字段名大小写不一致。SQLite 对字段名大小写不敏感但模型在生成 SQL 时可能一会儿用OrderTime一会儿用order_time。最好的做法是建表时统一使用小写蛇形命名并在 schema 里保持完全一致减少模型的猜测空间。第四个坑和安全性有关。我在做校验时只检查了首个非空关键词结果有一次模型生成了WITH x AS (...) DELETE FROM users这种以 WITH 开头的危险语句第一层校验没能拦下来。幸好只读连接兜了底。后来我把关键词检查从“是否包含 DELETE”改成了“是否包含以 DELETE 开头的任何语法结构”并额外检查了with语句内部是否出现写操作关键词。5.3 几个让效果更稳的进阶小技巧给模型喂 few-shot 样例是我认为性价比最高的优化手段。在系统提示词里加两组“问题 → SQL”的示例一组是简单条件查询一组是 JOIN GROUP BY 聚合查询。模型只需多读几十个 token 的上下文就能少犯很多类型错误。第二个技巧是给 schema 加“使用说明”。比如字段里有个price我是在 schema 后面追加一行“price单位是元查询价格时直接比较数值即可”。这类业务说明模型是从建表语句里猜不出来的但它对查询语义的准确度影响极大。第三个技巧是限制最大行数。除了让模型默认加 LIMIT 50我在执行层也会强行限制返回行数。比如在fetchall后再截断到 200 行防止因为模型生成SELECT *导致终端被大量数据淹没。第四个技巧是记录失败案例并形成小型回归测试集。每次模型生成错误我就把那条自然语言问题、错误 SQL、纠正后的 SQL 存到一个 JSON 文件里。后续修改提示词时直接用这套测试集重放一看就知道改动是变好还是变差。这个方法让整个项目从“试出来能用”变成了“可持续迭代”。结尾一点个人体会这套查询助手做到现在最大的收获不是“能跑”而是我真正理解了 LLM 应用里“约束”的价值。模型的能力边界固然重要但提示词的结构化、SQL 的校验层、错误反馈的闭环这些工程细节才是决定一个工具能不能被真实使用的关键。我个人的建议是不要一上来就追求复杂架构。先用 SQLite 做底层、用一个模型接口做推理把一条最核心的查询链路跑通再逐步加入缓存、界面、多轮对话。很多看似“简陋”的原型反而能帮你最清楚地看到模型在哪里犯傻、工程在哪里弥补、产品在哪里加分。后续如果把这个助手接上 Grafana 或者做成 Web 服务也只是在这套骨架上加肉而已。最后分享一个使用习惯我把这个脚本挂在了本地终端别名里平时查任何开发库的数据都先打一句自然语言试试。查成功了就省一段自己手写 SQL 的时间查失败了模型的错误往往比搜索引擎的答案更接近问题本身。这套“让 AI 替你写第一版 SQL你只做校验和修改”的工作流我已经离不开了。