直接在开头用场景切入：很多人会有这种需求——不会写 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 参数有可能变动，建议先按照这个组合把环境跑通，再考虑升级。

langchain==0.2.x langchain-community langchain-openai pymysql python-dotenv

这里有个细节：langchain主包负责核心抽象，langchain-community里包含数据库连接和工具实现，langchain-openai提供大模型接口封装。只装一个主包是不够的，很多同学在这儿一开始报错找不到模块，多半就是 community 没装。

2.2 数据库连接字符串的配置方式

数据库连接建议用 SQLAlchemy 的 URI 格式。我用的是 MySQL，连接串长这样：

DATABASE_URI = "mysql+pymysql://username:password@localhost:3306/mydb?charset=utf8mb4"

如果你用的是 SQLite，本地测试更轻量：

DATABASE_URI = "sqlite:///./test.db"

SQLite 的好处是零配置文件、免安装、适合快速验证链路，但注意它和 MySQL 在 SQL 语法、函数支持上存在差异，尤其是日期函数和分页语法。我用 SQLite 验证过了链路没问题，一换到生产 MySQL 就可能会在语法上踩坑，这一点放到后面专门说。

配置 API Key 时，我习惯用环境变量，不写死在代码里：

export OPENAI_API_KEY="sk-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_info=2 )

sample_rows_in_table_info=2表示每个表只取两行作为示例数据，少一点没关系，模型主要靠字段名判断语义，示例数据只起辅助作用。

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(model="gpt-4o-mini", temperature=0) chain = create_sql_query_chain(llm, db)

注意这里temperature=0，因为生成 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(db=db) 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_info=2 ) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) generate_chain = create_sql_query_chain(llm, db) execute_tool = QuerySQLDatabaseTool(db=db) 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)，没问题；切到 MySQL，SQLite 的日期函数就不适用了，MySQL 用的是DATE_FORMAT(order_date, '%Y-%m')。

还有分页语法：SQLite 和 MySQL 都用LIMIT，但 SQL Server 用OFFSET ... FETCH，Oracle 用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_info=2，几百行表数据只取关键字段和两行示例；
• 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(content=question)) # 把最近几轮历史压缩成一段摘要，再交给生成 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(content=answer))

注意，历史并不是越长越好。模型需要从历史中理解“三月份”指的是哪一年的三月，但历史过长会影响它对当前问题的专注度，所以一般保留最近几轮就够了。

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 直接执行写操作）。每一步都验证完再接下一步，调试成本会低很多。