无需数据库连接，利用Schema信息生成SQL语句的策略与实践

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI # 假设使用OpenAI模型，也可替换为其他LLM
from langchain_core.output_parsers import StrOutputParser
import os

# 确保设置了OpenAI API密钥
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 假设这是你的数据库Schema信息（DDL语句是最佳实践）
db_schema = """
CREATE TABLE Employees (
    employee_id INT PRIMARY KEY,
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    department_id INT,
    salary DECIMAL(10, 2),
    hire_date DATE
);

CREATE TABLE Departments (
    department_id INT PRIMARY KEY,
    department_name VARCHAR(50)
);

CREATE TABLE Projects (
    project_id INT PRIMARY KEY,
    project_name VARCHAR(100),
    department_id INT,
    FOREIGN KEY (department_id) REFERENCES Departments(department_id)
);
"""

# 2. 构建提示模板
# 包含系统指令，明确LLM的角色和任务
# 包含Schema信息和用户问题作为输入
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个专业的SQL查询生成器。根据提供的数据库Schema，为用户的问题生成准确的SQL查询语句。不要包含任何解释，只输出SQL语句。请确保生成的SQL语法正确，并考虑表之间的连接关系。"),
        ("user", "数据库Schema:\n{schema}\n\n用户问题: {question}"),
    ]
)

# 3. 初始化LLM
# temperature=0 通常用于需要确定性输出（如代码生成）的场景
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 请替换为你的模型和API密钥

# 创建一个LangChain表达式语言（LCEL）链
sql_generation_chain = prompt | llm | StrOutputParser()

# 调用链生成SQL
user_question_1 = "找出工资高于50000的员工的姓名和他们所属的部门名称。"
response_1 = sql_generation_chain.invoke({"schema": db_schema, "question": user_question_1})
print("问题1的SQL语句:")
print(response_1)
print("-" * 30)

user_question_2 = "列出所有部门的名称以及每个部门的员工数量。"
response_2 = sql_generation_chain.invoke({"schema": db_schema, "question": user_question_2})
print("问题2的SQL语句:")
print(response_2)
print("-" * 30)

问题1的SQL语句:
SELECT E.first_name, E.last_name, D.department_name
FROM Employees E
JOIN Departments D ON E.department_id = D.department_id
WHERE E.salary > 50000;
------------------------------
问题2的SQL语句:
SELECT D.department_name, COUNT(E.employee_id) AS employee_count
FROM Departments D
LEFT JOIN Employees E ON D.department_id = E.department_id
GROUP BY D.department_name;
------------------------------

from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import os

# 假设这是你的数据库Schema信息（可以从文件加载）
def load_schema_from_file(file_path):
    """模拟从文件加载Schema的函数"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        return "Schema file not found."

# 创建一个虚拟的schema文件
schema_file_content = """
CREATE TABLE Customers (
    customer_id INT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100)
);

CREATE TABLE Orders (
    order_id INT PRIMARY KEY,
    customer_id INT,
    order_date DATE,
    amount DECIMAL(10, 2),
    FOREIGN KEY (customer_id) REFERENCES Customers(customer_id)
);
"""
with open("my_db_schema.sql", "w", encoding="utf-8") as f:
    f.write(schema_file_content)

# 定义LLM和提示模板（与方法一相同）
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个专业的SQL查询生成器。根据提供的数据库Schema，为用户的问题生成准确的SQL查询语句。不要包含任何解释，只输出SQL语句。请确保生成的SQL语法正确，并考虑表之间的连接关系。"),
        ("user", "数据库Schema:\n{schema}\n\n用户问题: {question}"),
    ]
)

# 构建定制化链
custom_sql_chain = (
    # 步骤1: 动态加载Schema。RunnableLambda允许执行任意函数。
    # 这里假设我们知道schema文件路径
    RunnablePassthrough.assign(
        schema_text=RunnableLambda(lambda x: load_schema_from_file("my_db_schema.sql"))
    )
    # 步骤2: 组合输入，将用户问题和加载的Schema文本合并为LLM期望的格式
    | RunnableLambda(lambda inputs: {
        "schema": inputs["schema_text"],
        "question": inputs["user_question"] # 'user_question' 是外部传入的键
    })
    # 步骤3: 将组合后的输入传递给提示模板
    | prompt
    # 步骤4: 调用LLM
    | llm
    # 步骤5: 解析LLM输出为字符串
    | StrOutputParser()
)

# 调用定制化链
user_question_3 = "查询所有客户的姓名以及他们下过的订单总金额。"
response_3 = custom_sql_chain.invoke({"user_question": user_question_3})
print("问题3的SQL语句:")
print(response_3)
print("-" * 30)

user_question_4 = "找出没有下过订单的客户姓名。"
response_4 = custom_sql_chain.invoke({"user_question": user_question_4})
print("问题4的SQL语句:")
print(response_4)
print("-" * 30)

# 清理创建的虚拟文件
os.remove("my_db_schema.sql")

问题3的SQL语句:
SELECT C.name, SUM(O.amount) AS total_order_amount
FROM Customers C
JOIN Orders O ON C.customer_id = O.customer_id
GROUP BY C.name;
------------------------------
问题4的SQL语句:
SELECT C.name
FROM Customers C
LEFT JOIN Orders O ON C.customer_id = O.customer_id
WHERE O.order_id IS NULL;
------------------------------