目前广泛使用的关系数据库有mysql、oracle、postgresql、sqlite。

构建基于SQL DB的知识库问答系统，需要模型生成SQL查询，具体为

1）模型将用户输入转换为SQL查询。

2）执行SQL查询，执行查询。

3）模型使用查询结果响应用户输入。

过程如下图所示，

这里参考网络资料，基于sqlite和langchain，尝试示例基于LLM的关系数据库问答系统。

1 环境安装

安装所需的软件包langchain、langchain-community、langchain-openai、faiss-cpu

pip install langchain langchain-community langchain-openai faiss-cpu

2 知识库查询

2.1 数据准备

这里采用测试数据“Chinook.db”，原始数据连接如下，保存为Chinook.sql。

https://raw.githubusercontent.com/lerocha/chinook-database/master/ChinookDatabase/DataSources/Chinook_Sqlite.sql

运行如下指令创建Chinook.db，进入sqlite3界面

sqlite3 Chinook.db

在sqlite3界面读取Chinook.sql，并运行测试

.read Chinook.sql

SELECT * FROM Artist LIMIT 10;

2.2 数据导入

使用基于SQLAlchemy驱动的SQLDatabase类，导入sqlite数据，示例程序如下。

from langchain_community.utilities import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(db.dialect)
print(db.get_usable_table_names())
db.run("SELECT * FROM Artist LIMIT 10;")

输出示例如下

sqlite
['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']
"[(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), (4, 'Alanis Morissette'), (5, 'Alice In Chains'), (6, 'Antônio Carlos Jobim'), (7, 'Apocalyptica'), (8, 'Audioslave'), (9, 'BackBeat'), (10, 'Billy Cobham')]"

可以成功完善数据到关系数据库的导入。

2.3 LLM初始化

初始化LLM，这里基于openai接口，引入第三方部署的deepseek-r1。

import os
os.environ['OPENAI_API_KEY'] = "sk-xxxxx" # LLM令牌
os.environ['OPENAI_BASE_URL'] = "http://llm_provider_url" # LLM的openai兼容的部署url

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="deepseek-r1") # 这里基于第三方部署的deepseek-r1大模型

2.4 查询链示例

尝试将数据库db连接到一个大型语言模型（LLM），示例代码如下。

查询问题为：“How many employees are there”

from langchain.chains import create_sql_query_chain

chain = create_sql_query_chain(llm, db)

response = chain.invoke({"question": "How many employees are there"})
response

输出如下

'SQLQuery: SELECT COUNT("EmployeeId") FROM "Employee";\nSQLResult'

可见抽取的输出并不能真正执行，需要做一些简单的格式化处理，示例代码如下

def format_query(query):
    return query.strip().lstrip("SQLQuery:").rstrip("SQLResult:").strip()

sql = format_query(response)
sql

输出如下

'SELECT COUNT("EmployeeId") FROM "Employee";'

然后就可以运行数据库查询了

db.run(sql)

输出如下

'[(8,)]'

可以LLM为问题生成了有效的sql查询代码，并能完成实际执行。

对应的LLM prompt示例如下。

chain.get_prompts()[0].pretty_print()

prompt输出如下

You are a SQLite expert. Given an input question, first create a syntactically correct SQLite query to run, then look at the results of the query and return the answer to the input question.
Unless the user specifies in the question a specific number of examples to obtain, query for at most 5 results using the LIMIT clause as per SQLite. You can order the results to return the most informative data in the database.
Never query for all columns from a table. You must query only the columns that are needed to answer the question. Wrap each column name in double quotes (") to denote them as delimited identifiers.
Pay attention to use only the column names you can see in the tables below. Be careful to not query for columns that do not exist. Also, pay attention to which column is in which table.
Pay attention to use date('now') function to get the current date, if the question involves "today".

Use the following format:

Question: Question here
SQLQuery: SQL Query to run
SQLResult: Result of the SQLQuery
Answer: Final answer here

Only use the following tables:
{table_info}

Question: {input}

2.5 查询执行

这里执行以chain的形式组织的查询过程，示例代码如下所示。

from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool

execute_query = QuerySQLDataBaseTool(db=db)
write_query = create_sql_query_chain(llm, db)
chain = write_query | format_query | execute_query
chain.invoke({"question": "How many employees are there"})

输出如下

'[(8,)]'

2.6 结果响应

这里已经实现了基于LLM的sql查询自动生成，这里借助于langchain的链的表示方式，将原始问题、sql查询结果结合起来，借助于LLM生成最终答案，完成对原始问题的回答。

示例代码如下

from operator import itemgetter

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough

answer_prompt = PromptTemplate.from_template(
    """Given the following user question, corresponding SQL query, and SQL result, answer the user question.

Question: {question}
SQL Query: {query}
SQL Result: {result}
Answer: """
)

formatted_query = (write_query|format_query)

chain = (
    RunnablePassthrough.assign(query=formatted_query).assign(
        result=itemgetter("query") | execute_query
    )
    | answer_prompt
    | llm
    | StrOutputParser()
)

chain.invoke({"question": "How many employees are there"})

输出如下，可见LLM将db的查询结果，有效的融合到对原始问题的回答中。

'There are a total of 8 employees.'

reference

---

构建一个基于SQL数据的问答系统

https://www.langchain.com.cn/docs/tutorials/sql_qa/

create_sql_query_chain

https://python.langchain.com/api_reference/langchain/chains/langchain.chains.sql_database.query.create_sql_query_chain.html