LangChain调用GPT直接查数据库：自然语言转SQL完整实践

1. 为什么我建议你用GPT直接查数据库

LangChain里有一个非常实用、但经常被教程一笔带过的模块：Query SQL DB。说人话就是，让GPT这类大语言模型充当你的数据查询中介——你用中文或者英文说一句“上个月华东区销量最好的三个产品是什么”，它自动把这句话翻译成SQL，去数据库里执行，再把结果转成人话回复你。整个过程不需要你写一行SQL，也不需要你懂表结构。这个功能在内部数据问答、报表自动化、业务运营自助取数这些场景里非常能打，是LangChain生态里少数直接能落地出价值的模块之一。

这篇文章我就完整拆一遍这条链路：从环境准备、数据库连接、提示词设计、few-shot示例、安全限制，到线上部署时要留意的坑，全部按我实际跑过的经验来讲。篇幅不短，但跟着走基本能复现一条可用的自然语言查库链路。适合刚接触LangChain、想快速搭一个业务数据问答入口的开发者，对已经跑过简单demo、但被准确率或安全性问题卡住的人也很有参考价值。

1.1 传统取数流程的痛点在哪儿

我见过太多团队的数据需求流程是这样的：业务方跑到数据组说“我要看一下上周新用户的次周留存”，数据同学排期、写SQL、查数、做表、发出去，快的话半天，慢的话两天。要是业务方中途改一下口径，整个流程重来。这种模式的本质问题不是人懒，而是把“取数”这个动作锁死在少数会SQL的人身上。

固定报表能覆盖一部分高频问题，但业务是动态的，临时问题和交叉问题才是常态。今天问完留存明天问转化，后天问某个渠道的人群画像，每多一个问题就多一次排期。市面上的BI工具能解决一部分拖拽式取数，但复杂查询依然绕不开写SQL。这个痛点一直都存在，只是以前没有好的解法。

LLM出现之后，这个场景成了最合适的人工智能落地切入口之一，因为自然语言转SQL这件事，本质上是一个“翻译任务”，而大语言模型最擅长的恰恰是理解和翻译。LangChain做的不是发明新能力，而是把“翻译”这条链路工程化，让开发者不用自己拼提示词、管上下文、处理数据库连接这些脏活。

1.2 LangChain把这条链路拆成了哪几步

要理解LangChain在这里的价值，先理解一条完整链路由哪几段组成。第一段，把数据库的表结构、字段注释、示例数据抽取成一段结构化文本，塞进提示词，让LLM知道你的库里到底有什么。第二段，把用户问题连同表结构发给LLM，让它生成SQL语句，这个阶段只生成不执行。第三段，把生成的SQL拿到数据库引擎里去执行，拿到结果集。第四段，把结果集和原始问题一起丢给LLM，让它组织成一段人能直接读懂的回答。

这四个阶段看起来简单，但每一步都有坑。表结构太长了怎么办？模型生成了一段不存在的列名怎么办？结果集太大把上下文撑爆怎么办？用户故意让你删表怎么办？LangChain的SQLDatabase和create_sql_query_chain等组件，把这些问题拆成了可替换的小零件，你想改哪一段就改哪一段。这也是我推荐用LangChain而不是自己硬拼提示词的原因——不是因为它多智能，而是因为它把工程边界划得清楚。

2. 动手前的准备：依赖、数据库和连接串

2.1 依赖安装要用对版本

先说安装。我的建议是用LangChain最新的稳定版本，不要为了兼容旧代码死守老版本。如果你已经用过一些老的SQLDatabaseChain教程，会发现新版API已经变了，推荐做法是用create_sql_query_chain。老版SQLDatabaseChain把“生成SQL、执行、总结”三件事捆在一个类里，看起来省事，但你想改提示词、想单独控制执行逻辑的时候就很别扭。新版把它拆成了组件，灵活度一下子高了不少。

code复制pip install langchain langchain-community langchain-openai sqlalchemy

这里的langchain-openai负责LLM的接口封装，langchain-community里有SQLDatabase工具实现，sqlalchemy是Python最常用的ORM和数据库连接层。SQLite、MySQL、PostgreSQL都通过sqlalchemy连接，所以你本地用SQLite跑通，换生产库只需要改一个连接串，模型和链路代码基本不用动。

安装之后记得在环境变量里配好你的LLM平台API密钥，比如OPENAI_API_KEY，或者你用的模型服务商对应的变量名。很多新手在这一步卡住，报错401或者401的变体，基本就是密钥没配上。

2.2 用SQLite搭一个能跑通的演示库

SQLite是这个场景最适合起步的数据库，零配置、单文件、不需要账号权限，跑demo非常合适。我建议你自己建一个带业务语义的小库，别用网上找的随机表，因为后面你想测few-shot效果，必须知道业务口径是什么。

我用SQLAlchemy建个最简单的销售场景库，包含产品表、订单表，字段少一点反而好理解：

python复制from sqlalchemy import create_engine, Column, Integer, String, Float, Date
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import date

engine = create_engine("sqlite:///demo.db")
Base = declarative_base()

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    category = Column(String)
    price = Column(Float)

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    product_id = Column(Integer)
    quantity = Column(Integer)
    total_amount = Column(Float)
    created_at = Column(Date)
    region = Column(String)

Base.metadata.create_all(engine)

session = sessionmaker(bind=engine)()
session.add_all([
    Product(id=1, name="智能手环", category="数码", price=299),
    Product(id=2, name="机械键盘", category="外设", price=599),
    Product(id=3, name="蓝牙耳机", category="数码", price=399),
])
session.add_all([
    Order(id=1, product_id=1, quantity=10, total_amount=2990, created_at=date(2024, 1, 10), region="华东"),
    Order(id=2, product_id=2, quantity=5, total_amount=2995, created_at=date(2024, 1, 15), region="华南"),
    Order(id=3, product_id=3, quantity=8, total_amount=3192, created_at=date(2024, 2, 5), region="华东"),
])
session.commit()

注意这里我故意放了total_amount字段，又留了price和quantity冗余，这是为了后面讲业务口径时用得上。真实业务里很少这么设计，但演示口径冲突很直观。

2.3 SQLDatabase对象到底帮你做了什么

连接代码很简单：

python复制from langchain_community.utilities import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///demo.db")
print(db.get_table_info())

get_table_info()输出内容是理解整个链路的关键，它把表名、列名、类型、以及它能抽样到的前几行数据，拼成一段文本。这段文本就是你之后提示词里的“表结构信息”。

你可以自己打印出来看看，它长这样：

code复制CREATE TABLE products (
    id INTEGER, 
    name VARCHAR, 
    category VARCHAR, 
    price FLOAT
)

3 rows of products:
id name category price
1 智能手环 数码 299.0
...

这就是LLM判断如何写SQL的依据。你值得为这个对象多花几秒：它能include_views、include_tables限制只能看到哪些表，也能sample_rows_in_table_info控制抽样几行。线上库表很多的时候，这两个参数就是你的第一道闸门。

3. 核心链路拆解：从自然语言到SQL结果

3.1 先让模型只生成SQL这半边

理解这条链路最好的方式，是拆成两段看。第一段是“问题到SQL”，第二段是“SQL结果到人话回答”。很多人图省事直接用一条大链全包，结果出了错都不知道锅在哪，先拆开反而清晰。

第一段用create_sql_query_chain：

python复制from langchain_openai import ChatOpenAI
from langchain.chains import create_sql_query_chain

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = create_sql_query_chain(llm, db)

sql = chain.invoke({"question": "2024年1月华东区销售额是多少"})
print(sql)

你会看到它输出一段SQL字符串，类似：

sql复制SELECT SUM(total_amount) AS total_sales 
FROM orders 
WHERE region = '华东' 
  AND created_at >= '2024-01-01' 
  AND created_at < '2024-02-01'

注意temperature必须设成0，或者极低。这是生成SQL任务的红线，temperature太高会让模型在SQL语法选择上随机发挥，同样的提问每次生成不一样的SQL，排查时欲哭无泪。这个链只负责生成SQL，它不执行任何东西，这一点很重要，它给了你一个“审核关卡”的位置。

3.2 执行SQL和把结果翻译成人话

生成SQL之后，执行可以很简单：

python复制result = db.run(sql)
print(result)

db.run会直接执行并返回一个字符串化表格。这里有个细节，db.run内部是有行数限制的，默认返回前100行，不至于一次性把百万行灌进上下文。

拿到结果之后，第三段是把结果组织成回答：

python复制from langchain_core.prompts import ChatPromptTemplate

answer_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是数据分析助手。根据用户问题、SQL结果，用简洁中文回答。结果为空就说明没查到相关数据。"),
    ("human", "问题：{question}\nSQL结果：{result}")
])

answer_chain = answer_prompt | llm
final_answer = answer_chain.invoke({
    "question": "2024年1月华东区销售额是多少",
    "result": result
})
print(final_answer.content)

这样拆开的好处是你能清楚看到每一段的结果。如果生成的SQL错，你打印sql就能定位；如果结果对但解释错，那就是答案模型的问题，两不相扰。我强烈建议你初学阶段保持这种“手动接项链”的写法，别急着用自动execute_query的封装。虽然新版LangChain有更省事的组合方式，但理解数据流向的价值超过少写两行代码。

3.3 提示词模板里的几个关键设计

很多人跑通demo之后发现，问得稍微复杂一点，SQL就开始出错。多数问题不是模型能力，而是提示词信息不够。create_sql_query_chain虽然自带默认提示词，但它是通用模板，不知道你的业务口径，所以你必须定制。

我给你一个我调试完的提示词模板作为起点：

python复制from langchain_core.prompts import ChatPromptTemplate

sql_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是某零售公司的SQL专家。根据表结构和业务规则写SQL，只输出SQL，不要任何解释。
业务规则：
1. 销售额等于orders.total_amount字段，不要使用price乘以quantity计算。
2. region字段表示大区，值有：华东、华南、华北。
3. 日期字段created_at格式为YYYY-MM-DD。
4. 涉及时间范围查询，必须用>=和<，例如2024年1月写作 created_at >= '2024-01-01' AND created_at < '2024-02-01'。
5. 所有返回结果添加LIMIT 20，防止数据量过大。
6. 如果条件不足无法计算，返回空查询，不要编造字段。

表结构信息：
{schema}
"""),
    ("human", "请根据用户问题生成SQL：{question}")
])

query_chain = sql_prompt | llm

这段模板里有几个设计点值得讲。第一，口径写死，比如销售额到底取哪个字段，不写清楚模型就会自作聪明用price乘quantity。第二，时间范围用左闭右开，这能避免同一天数据被重复计算。第三，要求返回LIMIT，防止执行结果巨大。第四，规则里明确“不要编造字段”，这个约束能减少不少幻觉。

4. 完整实操：搭一条带few-shot的查库链路

4.1 基础版链路先跑通

把前面几节串起来，一个完整可运行的基础链路是：

python复制import os
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain.chains import create_sql_query_chain

os.environ["OPENAI_API_KEY"] = "你的密钥"

db = SQLDatabase.from_uri("sqlite:///demo.db")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

query_chain = create_sql_query_chain(llm, db)
question = "2024年销量最高的产品是什么"
sql = query_chain.invoke({"question": question})
print("生成的SQL:", sql)

result = db.run(sql)
print("查询结果:", result)

我建议你每加一个新功能就完整运行一次，不要攒一堆改动再debug。基础版跑通之后，接下来加few-shot，改了大概率能用。

4.2 加few-shot示例提升复杂查询准确率

如果你发现某些查询类型总是答错，最有效的办法是给提示词加few-shot示例，也就是“输入输出对”。LangChain的create_sql_query_chain支持传入examples参数，也可以直接在我的自定义提示词里手写示例。

我给订单场景写两个示例：

python复制examples = [
    {
        "input": "每个月的订单总量趋势",
        "query": "SELECT strftime('%Y-%m', created_at) AS month, COUNT(*) AS order_cnt FROM orders GROUP BY month ORDER BY month"
    },
    {
        "input": "2024年每个品类的平均订单金额",
        "query": "SELECT p.category, AVG(o.total_amount) AS avg_amount FROM orders o JOIN products p ON o.product_id = p.id WHERE o.created_at >= '2024-01-01' AND o.created_at < '2025-01-01' GROUP BY p.category"
    },
]

然后把examples和schema一起放进提示词里。few-shot之所以有效，是因为它给模型示范了你期望的SQL“风格”。你看第一个例子里我用了strftime函数而不是LIKE '%2024-01%'，这就是在教模型日期按月聚合该怎么写。第二个例子教了JOIN怎么写、类别如何分组。

这里有个技巧：你维护的few-shot不要多，五个以内足够，但每个都必须是你手工验证过的正确SQL。放一个错误的示例进去，模型会学歪。我习惯每发现一类典型错误，就把它对应的正确SQL加入示例库，顺便把错误SQL当成反例写进规则说明，双管齐下。

4.3 错误处理和结果包装

链路跑通之后，马上要面对一个现实问题：用户不会按你的预期问问题。问“销售额”还算好，如果问“对比一下华东和华南哪个厉害”，生成的SQL可能就是GROUP BY region，没问题。但如果问“这个数据可信吗”，模型可能生成一句无法执行的SQL。

所以执行环节必须加异常处理：

python复制try:
    result = db.run(sql)
except Exception as e:
    result = f"SQL执行失败：{e}"

再把失败原因拼进答案模型的提示词，让它用自然语言告诉用户“这次查询遇到问题，请换个问法”，而不是甩出一段原始SQL报错信息。用户看到“syntax error near GROUP”这种技术术语，会觉得产品很粗糙。

另外我建议把一次完整查询的日志记录下来，至少包含：问题、生成的SQL、执行耗时、结果行数、是否成功。这个日志是你优化提示词的素材库，也是排查线上问题最直接的依据。没有日志就去调模型，相当于盲人摸象。

5. 安全红线与线上部署的几个优化

5.1 防止模型生成危险SQL的几个硬措施

我必须把这句话放在最前面：永远不要给LLM连接的数据库账号开放写权限。模型生成的SQL再漂亮，本质也不可控，你没有百分之百的把握确认它不会生成一条DELETE或UPDATE。别赌这个概率。

我见过一个真实案例，某团队给内部工具接了主库的读写账号，有人问“把所有已完成的订单标记一下”，模型真的生成了UPDATE语句，好在开发人员提前限制了账号权限才没出事。理论上LLM没有恶意，但用户输入不可控，越狱提示词从来不是什么新鲜事。所以第一道防线永远是数据库账号，只读、限库、限表。

第二道防线是在执行前加一层合法的SQL校验。我用过一个很实用的粗校验：

python复制import re

def validate_sql(sql: str) -> bool:
    clean_sql = sql.strip().strip(";")
    if not clean_sql.upper().startswith("SELECT"):
        return False
    if re.search(r"(DELETE|UPDATE|INSERT|DROP|ALTER|CREATE|GRANT|REVOKE)", clean_sql, re.IGNORECASE):
        return False
    return True

这个正则不完美，但能挡住最典型的高危操作。如果你的环境跑的是PostgreSQL或MySQL，再进一步限制连接用户只能对特定schema有SELECT权限，两层叠加才放心。

第三道防线是执行时长和返回行数。SQLDatabase默认有行数限制，但线上环境最好再设数据库级别的statement_timeout，防止一个复杂JOIN把数据库拖垮。别小看这个，自然语言生成的SQL确实可能写出跨全表的多重JOIN，没有超时保护就是隐患。

5.2 表太多怎么办：schema裁剪

前面说SQLDatabase能一次把全部表结构塞给模型，但真实业务库一百多张表的时候，全塞进去有两个问题：上下文放不下，模型也会被无关表干扰。

我的做法是把“选表”和“写SQL”拆成两步。第一步，让一个模型看所有表名，判断哪个表可能和用户问题有关；第二步，只把选中的表结构发给另一个模型让它写SQL。LangChain社区里也有人把这套方案叫做“table selector”，不算复杂但效果很明显。

简单实现思路是：维护一个表名和业务描述的列表，用LLM选择，再基于筛选结果构建提示词。如果问题问“库存”，模型就会忽略订单表和用户表，只保留库存相关表。这个策略让准确率提升显著，因为无关表越少，模型选错字段的概率越低。

还有一个偏门但有效的方式：把表之间的关联关系直接写进提示词，比如“orders.product_id关联products.id”，这样模型写JOIN的时候不会两眼一抹黑瞎关联。这个信息加在schema后面即可。

5.3 成本和性能控制

每次问答都涉及两次LLM调用，一次生成SQL，一次组织回答，token成本不是零。国内做内部工具还好，如果面对大量用户，成本会直线上升。

我常用的策略是加查询缓存。同一句追问短期内重复出现，直接命中缓存返回上一次结果，既不消耗token也不压数据库。缓存key最好用归一化后的问题文本，去掉多余空格和标点。注意数据时效性，缓存设置合理过期时间，比如内部销售数据最多缓存几分钟，否则数据更新后答案还是旧的，会被业务方骂。

另一个思路是给结果集降温：如果查询结果很大，不要全部塞给答案模型，先做简单聚合或截断，让LLM只总结有限的信息。很多场景用户只想知道“行不行、多不多”，完整的明细数据可以直接用表格组件展示，不需要在模型里转述一遍。

6. 常见问题速查表与避坑经验

6.1 报错对照表

这一小节直接给速查表，都是我踩过或帮别人排查过的典型问题。

这里最隐蔽的一个坑是：“模型输出SQL带解释”。你可能觉得多几行字无所谓，但db.run直接执行会报错。我发现处理这个问题最稳的方式不是反复改提示词，而是在代码里加一个解析函数，提取第一个SELECT到末尾的内容。提示词负责让模型尽量不出错，代码负责兜底，两者不冲突。

6.2 几条花了很长时间才明白的经验

第一条，schema信息不要一股脑全塞。之前我接手一个几十张表的项目，把全量schema丢给模型后错误率反而上升。原因很简单，模型在大量无关字段里找不到正确的列。你宁可把schema裁剪到“够用”，也不要让它“什么都有”。用表选择器筛选之后，准确率立刻回升。

第二条，业务口径必须写进提示词并且定期维护。同一个“销售额”，财务要的可能是含税金额，业务要的可能是商品交易总额。模型不知道这些，你不写它就瞎猜。我建议你把口径描述放在提示词最显眼的位置，每次业务口径调整，都要同步更新提示词并回归验证。

第三条，开发环境和线上环境的SQL方言差异是个坑。SQLite跑得飞起，换到MySQL才发现日期函数不一样，LIMIT语法也有细微差别。技术上建议直接面向目标数据库开发，别依赖SQLite做全部验证。至少在你的测试流程里加一个与线上同版本的数据库。

第四条，日志比想象中更重要。我做问答工具时，最开始没有记录SQL日志，每次准确率上不去都只能靠猜。后来把线上问题、生成SQL、用户反馈全量记录下来，很快就能从里面看到模型规律性的错误模式，再针对性加few-shot，一轮优化就能解决的问题，再也不用靠玄学。

最后再分享一个个人习惯：我不会一上来就接生产主库。本地的尝试阶段，我用SQLite跑通链路；测试阶段，用生产库的只读从库；真正上线，服务账号只开SELECT权限。每换一个环境，我都用同一个测试问题集回归一遍，比如“按时间分组”“多表关联”“空结果”三类典型场景。这套流程走下来稳定了很多，也避免过不少线上事故。

如果你是在给团队搭内部的数据问答工具，我特别建议你把“记录问题和SQL”做成一个BI看板，时间长了它就是你的业务取数知识库。很多业务问题翻来覆去就是那几十种问法，示例越攒越厚，模型表现也会越来越稳。这就是这个方向最实在的复利效果。