本文旨在探讨如何在不建立实际数据库连接的情况下，利用大型语言模型（LLM）根据数据库Schema生成SQL语句。我们将重点介绍通过直接向LLM提供Schema描述来绕过传统数据库连接依赖的方法，并提供详细的实现步骤和代码示例，帮助开发者在保障安全性和灵活性的同时，实现高效的SQL生成。
引言：SQL生成与数据库连接的挑战
在构建基于LLM的应用时，例如智能客服、数据分析助手或自动化报告工具，我们经常需要LLM根据用户自然语言的提问来生成对应的SQL查询。传统的SQLDatabaseChain等工具通常依赖于SQLDatabase.from_uri方法，这要求提供一个有效的数据库连接URI，以便LLM能够访问数据库的实际Schema信息并执行查询。
然而，在许多场景下，我们可能面临以下需求：
- 安全性考量： 不希望将生产数据库的连接信息直接暴露给LLM或其运行环境。
- 性能优化： 避免每次生成SQL时都建立和维护数据库连接，尤其是在高并发场景下。
- 开发与测试： 在没有实际数据库环境的情况下，进行SQL生成逻辑的开发和测试。
- 纯SQL生成： 目标仅仅是生成SQL语句，而非执行查询或获取数据。例如，生成DDL语句或作为代码审查的一部分。
这些需求促使我们探索一种无需实际数据库连接，仅凭数据库Schema信息就能生成SQL的方法。
理解SQLDatabase与连接的依赖
LangChain中的SQLDatabase类旨在提供一个与数据库交互的抽象层。其核心功能之一是能够自省（introspect）数据库的Schema，即获取表名、列名、数据类型、主键外键关系等信息。SQLDatabase.from_uri()方法正是通过建立一个数据库连接来完成这一自省过程。这意味着，如果你想利用SQLDatabase对象来获取Schema信息，理论上它就需要一个可用的数据库连接。
问题在于，如果我们只希望LLM基于Schema生成SQL，而不需要它实际连接数据库执行查询，那么这个连接就显得多余，甚至带来了安全和性能上的负担。
方案一：直接向LLM提供数据库Schema描述
最直接且彻底避免数据库连接的方法，是绕过SQLDatabase对象，将数据库的Schema信息作为纯文本直接传递给LLM。LLM的强大之处在于其对自然语言的理解和生成能力，它完全可以从文本描述中学习数据库结构，并据此生成SQL。
核心思想
LLM在生成SQL时，真正需要的是数据库的结构信息（表、列、关系等），而不是一个活跃的数据库连接。我们可以将这些结构信息以清晰、简洁的文本格式提供给LLM，作为其上下文的一部分。
优点
- 彻底无连接： 完全避免了任何形式的数据库连接，包括内存数据库。
- 高安全性： 数据库凭据永不暴露。
- 高灵活性： 可以根据需要提供任意详细程度的Schema描述。
- 简化部署： 无需担心数据库驱动或连接池配置。
实现步骤
-
获取数据库Schema：
- 最理想的方式是获取数据库的CREATE TABLE语句，这些语句精确定义了表结构、列类型、约束和关系。
- 如果无法获取DDL，可以手动总结或使用数据库工具导出表结构描述（例如，DESCRIBE TABLE的输出）。
- 对于复杂的数据库，可以只提供与用户查询相关的部分Schema，以减少LLM的上下文负担。
-
构建Prompt：
- 设计一个清晰的Prompt模板，将Schema信息作为系统指令或用户输入的一部分。
- 明确告知LLM其角色（SQL生成助手）、任务（只生成SQL，不执行）、以及输出格式要求（例如，只返回SQL语句，不包含解释）。
-
使用LLM进行SQL生成：
- 直接调用LLM的API（如OpenAI API、Anthropic API等），或通过LangChain的Runnable接口来构建调用链。
示例代码
以下是一个使用LangChain和OpenAI模型，通过直接提供Schema文本来生成SQL的示例：
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
# 确保已设置OpenAI API密钥
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# 1. 模拟数据库Schema（使用CREATE TABLE语句）
# 实际应用中，这部分Schema可以从文件加载或动态生成
db_schema = """
CREATE TABLE users (
user_id INTEGER PRIMARY KEY,
username TEXT NOT NULL,
email TEXT UNIQUE,
registration_date DATE
);
CREATE TABLE products (
product_id INTEGER PRIMARY KEY,
product_name TEXT NOT NULL,
price REAL NOT NULL,
stock_quantity INTEGER
);
CREATE TABLE orders (
order_id INTEGER PRIMARY KEY,
user_id INTEGER,
order_date DATE,
total_amount REAL,
FOREIGN KEY (user_id) REFERENCES users(user_id)
);
"""
# 2. 构建Prompt模板
# System Prompt 明确了LLM的角色和任务
prompt_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个SQL生成助手。请根据以下数据库Schema生成SQL查询。不要执行查询，只返回SQL语句，不包含任何解释或额外文本。Schema:\n{schema}"),
("user", "{question}")
]
)
# 3. 初始化LLM
# 选择一个合适的LLM模型，例如GPT-4或GPT-3.5-turbo
llm = ChatOpenAI(model="gpt-4", temperature=0) # temperature=0 有助于生成更确定的结果
# 构建LangChain的链，将Prompt、LLM和输出解析器连接起来
sql_generation_chain = prompt_template | llm | StrOutputParser()
# 提问并生成SQL
user_question_1 = "查询所有在2023年注册的用户，并按注册日期降序排列。"
generated_sql_1 = sql_generation_chain.invoke({"schema": db_schema, "question": user_question_1})
print(f"用户问题: '{user_question_1}'")
print(f"生成的SQL:\n{generated_sql_1}\n")
user_question_2 = "查找所有订单总金额大于1000的用户，并显示他们的用户名和订单数量。"
generated_sql_2 = sql_generation_chain.invoke({"schema": db_schema, "question": user_question_2})
print(f"用户问题: '{user_question_2}'")
print(f"生成的SQL:\n{generated_sql_2}\n")
user_question_3 = "更新产品ID为5的产品的库存数量为50。"
generated_sql_3 = sql_generation_chain.invoke({"schema": db_schema, "question": user_question_3})
print(f"用户问题: '{user_question_3}'")
print(f"生成的SQL:\n{generated_sql_3}\n")代码解释：
- 我们首先定义了一个db_schema字符串，它包含了数据库的DDL语句。
- ChatPromptTemplate用于构建一个包含系统指令和用户问题的Prompt。{schema}和{question}是占位符，分别用于插入数据库Schema和用户提问。
- ChatOpenAI实例化了一个LLM客户端。
- 通过管道操作符|，我们将Prompt、LLM和StrOutputParser（用于将LLM的输出解析为字符串）连接成一个可执行的链。
- 最后，调用invoke方法，传入Schema和用户问题，即可获得LLM生成的SQL语句。
这种方法完全依赖于LLM对文本的理解能力，避免了任何实际的数据库连接。
方案二：利用内存SQLite数据库作为Schema载体（折衷方案）
虽然问题明确要求避免实际连接，但如果你的架构设计强烈依赖于LangChain的SQLDatabase对象，并且需要利用其get_table_info()等方法来提取Schema，但又不想连接外部生产数据库，那么使用内存SQLite数据库作为占位符是一个折衷方案。
注意： 此方法仍然建立了一个（本地、临时）数据库连接，与“无连接”的严格定义略有冲突，但它避免了连接外部的真实数据库。
实现思路
- 创建一个临时的内存SQLite数据库连接。
- 在这个内存数据库中执行DDL语句，手动创建你希望LLM了解的表结构。
- 将这个包含了Schema的SQLDatabase对象传递给LangChain的SQL Agent。
示例代码（作为参考，非严格意义上的“无连接”）
from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent
from sqlalchemy import create_engine, text
import os
# 确保已设置OpenAI API密钥
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# 1. 创建一个临时的内存SQLite数据库引擎
engine = create_engine("sqlite:///:memory:")
# 2. 基于这个引擎创建SQLDatabase对象
db = SQLDatabase(engine)
# 3. 在内存数据库中执行DDL语句来模拟Schema
ddl_statements = """
CREATE TABLE users (
user_id INTEGER PRIMARY KEY,
username TEXT NOT NULL,
email TEXT UNIQUE,
registration_date DATE
);
CREATE TABLE products (
product_id INTEGER PRIMARY KEY,
product_name TEXT NOT NULL,
price REAL NOT NULL,
stock_quantity INTEGER
);
CREATE TABLE orders (
order_id INTEGER PRIMARY KEY,
user_id INTEGER,
order_date DATE,
total_amount REAL,
FOREIGN KEY (user_id) REFERENCES users(user_id)
);
"""
with engine.connect() as connection:
connection.execute(text(ddl_statements))
connection.commit()
# 现在db对象包含了这些表的Schema信息，可以被LangChain的SQL Agent使用
# 我们可以打印出SQLDatabase对象自省到的Schema信息
print("Schema info from SQLDatabase object:")
print(db.get_table_info())
# 初始化LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 创建SQL Agent，它将使用db对象来获取Schema信息
# 注意：如果Agent尝试执行查询，它会在这个内存数据库上执行。
# 如果目标只是生成SQL，可能需要自定义Agent或其工具，限制其执行能力。
agent_executor = create_sql_agent(llm, db=db, agent_type="openai-tools", verbose=True)
# 提问并让Agent生成SQL
user_question = "查询所有在2023年注册的用户，并按注册日期排序。"
print(f"\n用户问题: '{user_question}'")
response = agent_executor.invoke({"input": user_question})
# 默认的SQL Agent可能会尝试执行查询，我们需要从verbose输出中提取生成的SQL
# 或者通过自定义Agent工具来只返回SQL
print(f"Agent响应:\n{response['output']}")
# 另一个例子：更新操作
user_question_update = "将产品ID为10的产品的库存数量增加20。"
print(f"\n用户问题: '{user_question_update}'")
response_update = agent_executor.invoke({"input": user_question_update})
print(f"Agent响应:\n{response_update['output']}")代码解释：
- 我们使用sqlalchemy.create_engine("sqlite:///:memory:")创建了一个纯内存的SQLite数据库。
- 然后，通过connection.execute(text(ddl_statements))在这个内存数据库中创建了表结构。
- SQLDatabase(engine)实例化了一个SQLDatabase对象，它现在“知道”这些表的Schema。
- `
