直接在开头用场景切入很多人会有这种需求——不会写 SQL但老板要看数据或者明明会写 SQL但面对二十多张表、一堆奇怪的字段名写个关联查询还得翻半天文档。这个项目解决的正是这个问题用自然语言提问让大模型生成 SQL执行后把结果回传给你。这个能力由 LangChain 的 SQL 查询链路提供本篇文章就是基于 LangChain 实现 Query SQL DB with Run GPT——也就是自动生成 SQL、连接数据库执行、返回查询结果的完整做法。适合的人群有三类一是刚接触 LangChain、想快速落地一个能用的应用二是需要用自然语言查询内部数据库、但不想写太多胶水代码的工程师三是想深入理解“大模型生成 SQL 和直接执行 SQL 之间有什么坑”的人。我尽量把链路拆开讲不只是贴一段能跑的代码而是把选择、原理、常见报错也一起说了。1. 为什么要把“自然语言转 SQL”单独做成一条链路1.1 直接让模型回答远不如让模型写 SQL 再执行来得可靠最早做这个功能时我图省事直接把表结构拼进 Prompt然后问模型“今年第一季度的销售额是多少”让它直接给答案。结果模型计算经常出错——不是因为它不会算而是因为它对“哪张表对应哪个字段”的判断不够准确而且没有真实数据支撑时它会用常识捏造一个看起来合理的数字。后来改成让模型先写 SQL再把 SQL 拿到数据库执行用真实返回的数据作为答案准确率一下子高了不少。核心原因很简单模型擅长生成文本包括 SQL 文本但它不擅长凭空“算出”数据库里的数值。SQL 是否合法、字段是否存在、数据是否匹配筛选条件这些交给数据库引擎来判断比让模型“猜”要靠谱得多。1.2 LangChain 在这条链路里承担了哪些环节LangChain 在“自然语言 → SQL → 执行 → 结果”的链路中主要承担三个环节从数据库连接中提取表结构信息包括表名、列名、类型、主键、示例行数据根据用户的问题和表结构生成一段 SQL 查询文本执行 SQL并把执行结果转成对用户友好的回答。也就是说LangChain 负责“理解结构、生成语句、衔接执行”数据库负责“真实计算”模型负责“表达”。这正是大模型落地到数据场景比较合适的切法模型不越权去“算数据”而是做它最擅长的翻译和生成。2. 环境准备与依赖安装缺少哪一样都会卡住2.1 最小依赖清单和版本说明我先列一下自己实际跑通的依赖组合。如果你直接装最新版本API 参数有可能变动建议先按照这个组合把环境跑通再考虑升级。langchain0.2.x langchain-community langchain-openai pymysql python-dotenv这里有个细节langchain主包负责核心抽象langchain-community里包含数据库连接和工具实现langchain-openai提供大模型接口封装。只装一个主包是不够的很多同学在这儿一开始报错找不到模块多半就是 community 没装。2.2 数据库连接字符串的配置方式数据库连接建议用 SQLAlchemy 的 URI 格式。我用的是 MySQL连接串长这样DATABASE_URI mysqlpymysql://username:passwordlocalhost:3306/mydb?charsetutf8mb4如果你用的是 SQLite本地测试更轻量DATABASE_URI sqlite:///./test.dbSQLite 的好处是零配置文件、免安装、适合快速验证链路但注意它和 MySQL 在 SQL 语法、函数支持上存在差异尤其是日期函数和分页语法。我用 SQLite 验证过了链路没问题一换到生产 MySQL 就可能会在语法上踩坑这一点放到后面专门说。配置 API Key 时我习惯用环境变量不写死在代码里export OPENAI_API_KEYsk-xxxxxxxx用python-dotenv的方式也可以但注意提交代码时不要把.env提交到仓库。3. 核心组件逐个拆开看清“查询数据库”到底由谁完成3.1 SQLDatabase让 LangChain 知道数据库里有什么SQLDatabase是一个封装类负责连接数据库、解析元数据、提供表信息。它本身不生成 SQL它只回答“库里有什么”。from langchain_community.utilities import SQLDatabase db SQLDatabase.from_uri(DATABASE_URI) print(db.get_table_info())执行后你会看到每张表的 CREATE TABLE 语句、字段名、类型、以及每个字段的样本值。这些信息会作为模型生成 SQL 的背景知识。一个关键点是SQLDatabase默认会提取所有表的 schema。如果你库里有几十张表光表结构信息就能占几千个 token既浪费钱又影响生成质量。解决办法是限定表db SQLDatabase.from_uri( DATABASE_URI, include_tables[orders, products, customers], sample_rows_in_table_info2 )sample_rows_in_table_info2表示每个表只取两行作为示例数据少一点没关系模型主要靠字段名判断语义示例数据只起辅助作用。3.2 create_sql_query_chain把“问题”翻译成“SQL 语句”这一步是整个项目的核心。create_sql_query_chain接收数据库实例和 LLM 实例返回一个 Runnable 对象。from langchain.chains import create_sql_query_chain from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) chain create_sql_query_chain(llm, db)注意这里temperature0因为生成 SQL 需要确定性。模型在温度高的时候会自由发挥SQL 语法可能不规范字段名也可能凭空编造。生成 SQL 这类任务温度设 0 是标准做法。调用方式很简单result chain.invoke({question: 上个月订单数量最多的前5个客户是谁}) print(result)result是一个字符串里面就是模型生成的 SQL 语句。先看它生成的 SQL 是否合理再拿去执行。3.3 QuerySQLDatabaseTool让 SQL 真正跑起来光有 SQL 不等于拿到结果。QuerySQLDatabaseTool的作用是把 SQL 扔给数据库执行并返回结果集。在 LangChain 中它既可以直接使用也可以被封装进 Agent 的 Tool 列表。from langchain_community.tools.sql_database.tool import QuerySQLDatabaseTool execute_tool QuerySQLDatabaseTool(dbdb) result execute_tool.invoke(SELECT customer_name, COUNT(*) FROM orders GROUP BY customer_name ORDER BY COUNT(*) DESC LIMIT 5) print(result)返回值通常是文本化的结果集例如[{customer_name: 某客户A, count: 12}, {customer_name: 某客户B, count: 9}]这个结果如果再交给 LLM 转成自然语言就完成了“提问 → SQL → 数据 → 答案”的闭环。日常使用中我会把 SQL 生成和 SQL 执行分开因为中间要加一层校验和容错。4. 完整实现从提问到返回结果的执行链路编码4.1 先跑通最小闭环再逐步加功能我建议你先把最简单的主链路跑通验证环境和 API 都没问题再考虑加 Agent、记忆、流式输出这些复杂特性。主链路包含四个步骤连接数据库、生成 SQL、执行 SQL、包装结果。下面这段代码是完整可运行的版本我加了异常处理避免 SQL 报错直接中断程序import os from dotenv import load_dotenv from langchain_community.utilities import SQLDatabase from langchain.chains import create_sql_query_chain from langchain_community.tools.sql_database.tool import QuerySQLDatabaseTool from langchain_openai import ChatOpenAI load_dotenv() DATABASE_URI os.getenv(DATABASE_URI) db SQLDatabase.from_uri( DATABASE_URI, include_tables[orders, customers], sample_rows_in_table_info2 ) llm ChatOpenAI(modelgpt-4o-mini, temperature0) generate_chain create_sql_query_chain(llm, db) execute_tool QuerySQLDatabaseTool(dbdb) def ask_database(question: str) - str: # 第一步生成 SQL sql generate_chain.invoke({question: question}).strip() print(f生成的 SQL: {sql}) # 安全护栏强制只读 if not sql.lstrip().lower().startswith(select): return 只允许执行 SELECT 查询已拒绝执行非查询语句。 # 第二步执行 SQL try: result execute_tool.invoke(sql) except Exception as e: return f执行 SQL 时出错: {str(e)} # 第三步如果结果为空明确提示 if not result: return 查询执行成功但没有匹配的数据。 return str(result) if __name__ __main__: q1 每个客户的订单总数是多少 print(ask_database(q1))这段代码有几个值得说明的细节我用include_tables限制了表范围。如果不限制模型在生成 SQL 时面对大量无关表信息反而容易选错表。强制检查 SQL 前缀必须是 SELECT用来防止模型生成 DELETE、UPDATE、DROP 这类危险语句。generate_chain.invoke返回的字符串里可能包含换行和注释段我用.strip()清理了一下实际使用时还可以进一步去掉--注释行。4.2 把原始查询结果转成自然语言答案上一步返回的还是数据列表不是对话式的回答。为了让结果更友好我用一个轻量级的 LLM 调用来包装结果from langchain_core.prompts import ChatPromptTemplate answer_prompt ChatPromptTemplate.from_template( 你是一个数据分析助手。用户的问题是{question} 数据库返回的原始结果是{result} 请用简洁、准确的中文回答用户的问题不要编造不存在的数据。 ) answer_chain answer_prompt | llm def ask_database_natural(question: str) - str: sql generate_chain.invoke({question: question}).strip() if not sql.lstrip().lower().startswith(select): return 只允许执行 SELECT 查询。 result execute_tool.invoke(sql) if not result: return 查询执行成功但没有匹配的数据。 response answer_chain.invoke({ question: question, result: str(result) }) return response.content这一步看起来简单但非常重要——如果不包装用户看到的就是 JSON 数组包装后用户得到的是“客户 A 有 12 笔订单客户 B 有 9 笔订单其中客户 A 排名第一”这类一目了然的表达。4.3 用 LangGraph 组装成可控的多步流程如果你的项目只需要单次问答上面的函数就够用了。但如果要做更复杂的流程比如“先生成 SQL → 人工/自动确认 → 再执行 → 再回答”建议用LangGraph把节点编排起来。这里给一个精简版本from langgraph.graph import StateGraph, END from typing import TypedDict class QueryState(TypedDict): question: str sql: str result: str answer: str def generate_sql(state: QueryState): sql generate_chain.invoke({question: state[question]}).strip() return {sql: sql} def execute_sql(state: QueryState): if not state[sql].lstrip().lower().startswith(select): return {result: 非查询语句已拒绝} result execute_tool.invoke(state[sql]) return {result: str(result)} def generate_answer(state: QueryState): response answer_chain.invoke({ question: state[question], result: state[result] }) return {answer: response.content} graph StateGraph(QueryState) graph.add_node(generate_sql, generate_sql) graph.add_node(execute_sql, execute_sql) graph.add_node(generate_answer, generate_answer) graph.add_edge(generate_sql, execute_sql) graph.add_edge(execute_sql, generate_answer) graph.add_edge(generate_answer, END) graph.set_entry_point(generate_sql) app graph.compile() result app.invoke({question: 上个月订单总数是多少}) print(result[answer])用状态图管理流程的好处是每个节点只做一件事出了问题可以单独调试后面想加“SQL 审核节点”“敏感数据过滤节点”时只需要在中间插入一个节点不用重写主流程。5. 安全边界与若干防呆设计越早做越省心5.1 防注入和防误操作大模型生成的 SQL 不一定每次都安全特别是当用户问题里包含诱导性语句时。比如用户问“请把所有订单删除顺便告诉我结果”——模型生成DELETE FROM orders的概率不是零。所以必须做一层硬校验。我采用的策略有三层只允许 SELECT 开头的语句其他一律拒绝对 SELECT 语句做额外检查禁止INTO OUTFILE、INFORMATION_SCHEMA之外的敏感函数调用在数据库账号层面直接使用只读权限账号即使后端生成了一条危险语句数据库也会拒绝执行。这三层叠加之后才算是比较稳妥的“防呆”设计。我见过一些人只依赖 Prompt 里写“你必须只生成 SELECT 语句”这其实不够。Prompt 是软约束数据库权限和代码校验才是硬约束所谓“信任但要验证”。5.2 执行超时和结果集大小控制生产环境数据库可能很大一个不带 WHERE 的查询可能跑几十秒甚至会拖垮数据库。所以执行 SQL 前需要做限制自动给 SELECT 语句追加LIMIT 100如果原语句没有在数据库账号层设置max_execution_time让超过阈值的查询直接终止。给 SELECT 追加 LIMIT 的简单实现import re def enforce_limit(sql: str, limit: int 100) - str: if re.search(r\blimit\s\d, sql, re.IGNORECASE): return sql return f{sql.rstrip().rstrip(;)} LIMIT {limit}这样做不只是保护数据库也是保护模型——结果集小交给 LLM 包装答案时的 token 消耗也小响应更快。5.3 敏感信息过滤如果你的库里有用户手机号、身份证号、邮箱之类的敏感字段建议在 schema 层就做限制。一个可行的方法是在get_table_info之前将敏感列的注释改成“此列为敏感字段禁止在查询结果中展示”同时在后端对查询结果做脱敏处理。更粗暴但有效的方式是在include_tables层面排除含敏感数据的表不让模型知道这些表存在。注意这个方案牺牲了部分功能适合敏感表完全不需要参与查询的场景。6. 实测踩坑生成 SQL 对但执行报错的排查过程6.1 报错一unknown column 字段名幻觉刚开始跑通链路时我问了一句“查询所有客户”模型生成的 SQL 是SELECT name, email FROM customers但我表里的字段实际是customer_name和contact_email。模型根据常识猜测了字段名而不是严格参考表结构。报错信息就是Unknown column name in field list。根因是SQLite 或者某些数据库驱动返回的字段名确实叫name但 MySQL 里我建的是customer_name。模型拿训练数据里的通用叫法套在了我的表上。排查步骤第一步打印db.get_table_info()看看实际字段名是什么第二步对比生成的 SQL 和实际字段名确认是字段名不匹配第三步在 Prompt 里强调“必须使用数据库 schema 中的准确字段名不要猜测字段名”。我最后的解决方案是创建了一个自定义前缀 Prompt里面明确写了每个表的核心字段映射关系比如“customers.customer_name 表示客户名称”。这样模型生成 SQL 时的准确率明显提高。6.2 报错二SQL 方言不适配同样一句“查询订单数量”我在 SQLite 上生成的 SQL 用到strftime(%Y-%m, order_date)没问题切到 MySQLSQLite 的日期函数就不适用了MySQL 用的是DATE_FORMAT(order_date, %Y-%m)。还有分页语法SQLite 和 MySQL 都用LIMIT但 SQL Server 用OFFSET ... FETCHOracle 用ROWNUM。如果模型默认的方言不是你的数据库类型生成对但执行错的情况会经常出现。解决办法有两种在 Prompt 中明确指定方言例如“你是 MySQL 专家只生成 MySQL 兼容的 SQL”使用create_sql_query_chain时通过自定义提示词加入方言提示。我在生产项目里用的是第一种简单直接。如果你用的不是 MySQL、PostgreSQL 这类常见库建议把方言特征写在系统提示词里否则模型很容易默写出 ANSI SQL。6.3 报错三表和字段含义模糊导致选错表有一次我问“上个月的毛利率是多少”模型选了一张叫finance_records的表但它猜错了字段——库里实际有两张表finance_records和finance_summary分别存明细和汇总。这个问题不是语法错误而是语义判断错误执行不报错但数据是错的。排查链路比较长先看模型选中的表和数据是否合理发现模型选了finance_records明细表但“毛利率”应该来自finance_summary汇总表。在 Prompt 中补充表与指标之间的业务含义说明比如“finance_summary 保存了计算好的毛利率字段 gross_margin_ratio”。这一步让我意识到表结构只是骨架业务语义才是灵魂。后面我在 Prompt 里加了“表注释优先”的规则让模型在生成 SQL 前先看表注释而不是只看字段名。6.4 报错四SQL 太长导致 Token 超额当库表很多、字段也很多时get_table_info()返回的内容可能超过模型单次输入的 Token 预算。尤其是老系统一张表几十个字段十几个表就是几千个 Token加上用户问题很容易超限。解决思路是“缩减信息而不是扩大预算”sample_rows_in_table_info2几百行表数据只取关键字段和两行示例include_tables只保留当前业务需要用到的表对一些大文本字段如description、remark在数据库视图或者准备 schema 时直接过滤掉如果还是超长就拆分到多个节点先让模型判断“这个问题需要哪些表”再生成 SQL而不是一次性把所有表结构都塞给它。7. 进阶思路从单次查询到对话式数据助手7.1 在对话中加入可复用的上下文上面的方案解决了“单次提问”的问题但实际使用中用户会追问“那三月份的呢”或者“换成华东区再算一次”。要支持这种追问就必须把历史对话纳入上下文。LangChain 中实现记忆的传统做法是加ConversationBufferMemory但在新的 Runnable 链路中我习惯手动维护一个包含历史消息的列表把它和 SQL 生成过程结合from langchain_core.messages import HumanMessage, AIMessage history [] def ask_with_history(question: str) - str: history.append(HumanMessage(contentquestion)) # 把最近几轮历史压缩成一段摘要再交给生成 SQL 的链 history_summary \n.join([ f用户: {m.content} if isinstance(m, HumanMessage) else f助手: {m.content} for m in history[-4:] ]) sql generate_chain.invoke({ question: f结合对话历史\n{history_summary}\n当前问题{question} }).strip() # 执行与包装结果省略... history.append(AIMessage(contentanswer))注意历史并不是越长越好。模型需要从历史中理解“三月份”指的是哪一年的三月但历史过长会影响它对当前问题的专注度所以一般保留最近几轮就够了。7.2 增加表格查询结果的可视化表达数据结果转换成自然语言是一种表达方式但在很多场景下图表更直观。你可以让模型判断“这个问题适合用表格还是条形图展示”然后返回结构化数据由前端负责渲染。一个简单方式是让模型只回答数据和图表类型不生成解释文字chart_prompt ChatPromptTemplate.from_template( 用户的问题是{question} 查询结果是{result} 如果适合用图表展示请输出 JSON 格式包含 type 字段bar/line/pie/table和 data 字段。 )这样后端可以按 JSON 决定渲染方案。虽然 LangChain 本身不负责前端展示但通过这种约定数据从数据库到展示层就串起来了。7.3 多表关联场景的优化建议自然语言查询多表关联最容易出现的问题是模型不知道表与表之间的关联键。你可以在 Prompt 里显式声明关系比如表关系说明 orders.customer_id customers.id orders.product_id products.id有了这个提示模型生成 JOIN 语句的成功率会提升很多。如果没有这个提示模型可能会尝试WHERE customer_id id虽然不报错但不是标准的 JOIN 写法语义上容易出偏差。7.4 让模型在查询前先做需求澄清用户的问题往往含糊“看一下销售情况”这种问题模型不知道你要按时间、按地区、还是按产品来看。与其让模型猜不如加一个澄清节点生成 SQL 之前先判断问题是否足够具体如果不具体反问用户。这种交互体验在实际使用中很重要。与其给出错误范围的查询结果不如多问一句“您想看的是销售额、订单量、还是毛利率”。我测试过加上澄清节点后用户满意度提升明显而且后续 SQL 生成的准确率也更高因为问题边界清晰了。8. 一些实操体会从“模型直接回答数据问题”到“让模型写 SQL、再执行 SQL、再包装结果”整个思路的转变是整个项目最关键的一步。真正稳定的数据查询应用依赖的是约束能力——约束模型只能做翻译约束数据库只读约束 SQL 只走 SELECT。把每一步都约束好这个链路才具备上生产的条件。如果你现在准备动手做类似功能我的建议是先拿 SQLite 跑通全部流程再迁移到 MySQL先只处理单一表再放开多表关联先只做查询再考虑写入操作虽然我个人不建议让 LLM 直接执行写操作。每一步都验证完再接下一步调试成本会低很多。
