https://python.langchain.com.cn/docs/expression_language/cookbook/sql_db

LangChain 表达式语言：SQL 数据库查询链总结

本文档核心是通过 LangChain 的 Runnable 组件构建“SQL 数据库查询链”，实现“用户自然语言问题→自动生成 SQL 查询→执行查询→返回自然语言回答”的全流程自动化。整个实现分两步：先构建“生成 SQL 查询”的链，再扩展为“生成查询+执行+自然语言回答”的完整链，依赖 SQLDatabase 工具连接数据库，ChatOpenAI 生成 SQL 和回答。

一、核心前提与组件（直接定义，无抽象概念）

1. Chinook 样本数据库
   文档使用的示例数据库（SQLite 格式），包含音乐相关表（如 Employee 员工表、Track 曲目表等），需从指定链接（https://database.guide/2-sample-databases-sqlite/）下载，路径配置为 sqlite:///./Chinook.db。

2. SQLDatabase 工具
   LangChain 中用于连接和操作 SQL 数据库的组件，支持通过 URI 初始化数据库连接，提供 get_table_info()（获取表结构，即 schema）、run(query)（执行 SQL 查询并返回结果）等核心方法。

3. 两步式链结构
   文档采用“分阶段构建”逻辑：

   • 第一步：sql_response 链——接收用户问题，生成对应的 SQL 查询；
   • 第二步：full_chain 链——在第一步基础上，执行 SQL 查询，再将“表结构+问题+SQL+查询结果”整合为自然语言回答。

二、完整实现步骤（代码+逐步解释，含数据流向）

步骤1：安装依赖与导入库

先安装必需的库，再导入构建链的核心组件（提示模板、数据库工具、模型等）：

# 1. 安装依赖：langchain（核心链功能）、langchain-openai（OpenAI模型）
%pip install --upgrade --quiet langchain langchain-openai

# 2. 导入所需库
from langchain_core.prompts import ChatPromptTemplate  # 构建提示模板
from langchain_community.utilities import SQLDatabase  # 连接SQL数据库
from langchain_core.output_parsers import StrOutputParser  # 转换AI输出为字符串
from langchain_core.runnables import RunnablePassthrough  # 传递数据的组件
from langchain_openai import ChatOpenAI  # OpenAI聊天模型

步骤2：初始化数据库连接与辅助函数

连接 Chinook 数据库，并定义两个关键函数（获取表结构、执行查询），为后续链提供“数据库交互能力”：

# 1. 连接Chinook数据库（需先下载数据库文件，确保路径正确）
# URI格式：sqlite:///[文件路径]，这里是当前目录下的Chinook.db
db = SQLDatabase.from_uri("sqlite:///./Chinook.db")

# 2. 辅助函数1：获取数据库表结构（schema）
# 参数“_”表示无需接收输入（Runnable调用时会默认传参，用“_”忽略）
def get_schema(_):
    # db.get_table_info()：返回所有表的结构（字段名、类型、约束等），供AI生成SQL时参考
    return db.get_table_info()

# 3. 辅助函数2：执行SQL查询（后续链中直接用db.run()，此函数为文档示例，可直接复用）
def run_query(query):
    # db.run(query)：执行传入的SQL查询，返回查询结果（字符串格式）
    return db.run(query)

步骤3：构建“生成SQL查询”的链（sql_response）

核心作用：接收用户的自然语言问题，结合数据库表结构，让 AI 自动生成符合语法的 SQL 查询。

# 1. 初始化AI模型（用于生成SQL查询）
model = ChatOpenAI()

# 2. 定义“生成SQL”的提示模板
# 模板逻辑：告诉AI“基于{schema}（表结构），为{question}（用户问题）编写SQL查询”
template = """根据下面的表模式，编写一个SQL查询来回答用户的问题：
{schema}
问题：{question}
SQL查询："""
prompt = ChatPromptTemplate.from_template(template)

# 3. 构建sql_response链（数据流向：问题→表结构→提示→生成SQL→字符串）
sql_response = (
    # 第一步：给输入数据添加“schema”键，值为get_schema()获取的表结构
    # 输入：{"question": "有多少员工？"} → 处理后：{"question": "...", "schema": "表结构字符串"}
    RunnablePassthrough.assign(schema=get_schema)
    # 第二步：将处理后的输入（含schema和question）传入提示模板，生成完整提示
    | prompt
    # 第三步：AI生成SQL，绑定“stop”参数（生成到“\nSQLResult:”停止，避免多余内容）
    | model.bind(stop=["\nSQLResult:"])
    # 第四步：将AI输出的AIMessage对象转换为纯字符串（仅保留SQL查询）
    | StrOutputParser()
)

# 4. 测试sql_response链：输入“有多少员工？”，生成对应的SQL
sql_result = sql_response.invoke({"question": "有多少员工？"})
print("生成的SQL查询：", sql_result)  # 输出：'SELECT COUNT(*) FROM Employee'（统计Employee表的行数）

步骤4：构建“生成SQL+执行+自然语言回答”的完整链（full_chain）

在 sql_response 基础上，新增“执行 SQL 查询”和“生成自然语言回答”两步，形成闭环流程：

# 1. 定义“生成自然语言回答”的提示模板
# 模板逻辑：结合{schema}（表结构）、{question}（问题）、{query}（SQL）、{response}（查询结果），生成易懂回答
template_response = """根据下面的表模式，问题，SQL查询和SQL响应，编写一个自然语言回答：
{schema}
问题：{question}
SQL查询：{query}
SQL响应：{response}"""
prompt_response = ChatPromptTemplate.from_template(template_response)

# 2. 构建full_chain完整链（数据流向：问题→生成SQL→执行SQL→生成回答）
full_chain = (
    # 第一步：处理输入，添加三个关键键：
    # - "query"：调用sql_response链生成的SQL查询
    # - "schema"：调用get_schema()获取的表结构
    # - "response"：执行SQL查询的结果（用lambda x: db.run(x["query"])，从“query”键取SQL并执行）
    RunnablePassthrough.assign(query=sql_response).assign(
        schema=get_schema,
        response=lambda x: db.run(x["query"]),
    )
    # 第二步：将处理后的输入（含schema、question、query、response）传入回答模板
    | prompt_response
    # 第三步：AI基于模板生成自然语言回答（返回AIMessage对象）
    | model
)

# 3. 测试full_chain链：输入“有多少员工？”，获取最终回答
final_answer = full_chain.invoke({"question": "有多少员工？"})
print("最终自然语言回答：", final_answer.content)  # 输出：'有8个员工。'（对应Employee表共8条数据）

三、核心逻辑总结（表格梳理关键环节）

四、初学者关键注意点

1. 数据库文件路径：必须确保 Chinook.db 文件在指定路径（./Chinook.db 即当前代码运行目录），否则 SQLDatabase.from_uri 会报错“找不到文件”，需先从文档链接下载数据库。
2. stop 参数的作用：model.bind(stop=["\nSQLResult:"]) 是为了让 AI 只生成 SQL 查询，不额外添加“SQLResult:”及后续内容，避免生成的 SQL 包含无效字符。
3. 数据流向的连贯性：full_chain 中，query 来自 sql_response，response 来自 db.run(x["query"])，每个键的取值都依赖前一步的结果，顺序不能乱。
4. 查询结果格式：db.run(query) 返回的结果是原始字符串（如统计行数返回 [(8,)]），AI 会自动解析为自然语言（如“8个”），无需手动处理格式。