使用 OpenAI + FastAPI + SQLite 构建文本转 SQL 应用程序

引言

数据已成为任何成功企业不可或缺的资源，它能为明智的决策制定提供有价值的见解。鉴于数据的重要性，许多公司都在构建用于存储和分析数据的系统。然而，很多时候获取和分析所需数据并非易事，尤其是随着数据系统的日益复杂。

随着生成式人工智能的出现，数据处理工作变得轻松了许多 —— 我们现在可以使用简单的自然语言，就能获得基本准确且与输入高度契合的输出结果。这一点也适用于使用 SQL 进行的数据处理和分析，我们可以通过自然语言来请求生成查询语句。

在本文中，我们将开发一个简易的 API 应用程序，它能将自然语言转换为数据库可理解的 SQL 查询语句。我们将使用三个主要工具：OpenAI、FastAPI 和 SQLite。

以下是具体规划。

文本转 SQL 应用程序开发

首先，我们需要准备项目所需的所有内容。你只需提供 OpenAI API 密钥，我们将用它来访问生成式模型。为了实现应用程序的容器化，我们将使用 Docker，你可以通过 Docker Desktop 在本地环境中获取它。

其他组件（如 SQLite）在安装 Python 时会自动可用，而 FastAPI 将在后续步骤中安装。

项目的整体结构如下：

plaintext

text_to_sql_app/
├── app/
│   ├── __init__.py          
│   ├── database.py           
│   ├── openai_utils.py       
│   └── main.py               
├── demo.db                   
├── init_db.sql               
├── requirements.txt          
├── Dockerfile                
├── docker-compose.yml        
├── .env

你可以按照上述结构创建项目，也可以使用以下代码仓库简化操作。我们仍会逐一讲解每个文件，帮助你理解应用程序的开发过程。

首先，在.env文件中填入之前获取的 OpenAI API 密钥，代码如下：

plaintext

OPENAI_API_KEY=你的API密钥

接着，在requirements.txt中填写项目所需的库：

plaintext

fastapi
uvicorn
sqlalchemy
openai
pydantic
python-dotenv

然后，编写__init__.py文件的代码：

plaintext

from pathlib import Path
from dotenv import load_dotenv

load_dotenv(dotenv_path=Path(__file__).resolve().parent.parent / ".env", override=False)

上述代码确保环境中包含所有必需的密钥。

之后，在database.py文件中编写 Python 代码，用于连接后续创建的 SQLite 数据库（名为demo.db）并提供执行 SQL 查询的方法：

plaintext

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

ENGINE = create_engine("sqlite:///demo.db", future=True, echo=False)

def run_query(sql: str) -> list[dict]:
    with Session(ENGINE) as session:
        rows = session.execute(text(sql)).mappings().all()
    return [dict(r) for r in rows]

接下来，准备openai_utils.py文件，该文件将接收数据库模式和输入的问题，输出包含 SQL 查询的 JSON（并添加防护机制以防止任何写入操作）：

plaintext

import os
import json
from openai import OpenAI        

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

_SYSTEM_PROMPT = """
你需要将自然语言问题转换为只读的SQLite SQL语句。
禁止输出INSERT / UPDATE / DELETE语句。
返回格式为JSON：{ "sql": "..." }。
"""

def text_to_sql(question: str, schema: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",        
        temperature=0.1,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": _SYSTEM_PROMPT},
            {"role": "user",
             "content": f"数据库模式:\n{schema}\n\n问题: {question}"}
        ]
    )
    payload = json.loads(response.choices[0].message.content)
    return payload["sql"]

完成代码编写和连接配置后，我们将使用 FastAPI 构建应用程序。该应用程序将接收自然语言问题和数据库模式，将其转换为 SQL 的SELECT查询语句，在 SQLite 数据库中执行该查询，并以 JSON 格式返回结果。这是一个可通过命令行界面（CLI）访问的 API。

plaintext

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import inspect
from .database import ENGINE, run_query
from .openai_utils import text_to_sql

app = FastAPI(title="文本转SQL演示")

class NLRequest(BaseModel):
    question: str

@app.on_event("startup")
def capture_schema() -> None:
    insp = inspect(ENGINE)
    global SCHEMA_STR
    SCHEMA_STR = "\n".join(
        f"CREATE TABLE {t} ({', '.join(c['name'] for c in insp.get_columns(t))});"
        for t in insp.get_table_names()
    )

@app.post("/query")
def query(req: NLRequest):
    try:
        sql = text_to_sql(req.question, SCHEMA_STR)
        if not sql.lstrip().lower().startswith("select"):
            raise ValueError("仅允许SELECT语句")
        return {"sql": sql, "result": run_query(sql)}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

以上就是主应用程序的全部内容。接下来需要准备数据库。你可以使用下面的代码作为init_db.sql文件的示例（也可根据需求自行修改）：

plaintext

DROP TABLE IF EXISTS order_items;
DROP TABLE IF EXISTS orders;
DROP TABLE IF EXISTS payments;
DROP TABLE IF EXISTS products;
DROP TABLE IF EXISTS customers;

CREATE TABLE customers (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    country TEXT,
    signup_date DATE
);

CREATE TABLE products (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    category TEXT,
    price REAL
);

CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER,
    order_date DATE,
    total REAL,
    FOREIGN KEY (customer_id) REFERENCES customers(id)
);

CREATE TABLE order_items (
    order_id INTEGER,
    product_id INTEGER,
    quantity INTEGER,
    unit_price REAL,
    PRIMARY KEY (order_id, product_id),
    FOREIGN KEY (order_id) REFERENCES orders(id),
    FOREIGN KEY (product_id) REFERENCES products(id)
);

CREATE TABLE payments (
    id INTEGER PRIMARY KEY,
    order_id INTEGER,
    payment_date DATE,
    amount REAL,
    method TEXT,
    FOREIGN KEY (order_id) REFERENCES orders(id)
);

INSERT INTO customers (id, name, country, signup_date) VALUES
 (1,'爱丽丝','美国','2024-01-05'),
 (2,'鲍勃','英国','2024-03-10'),
 (3,'崔','韩国','2024-06-22'),
 (4,'达拉','印度尼西亚','2025-01-15');

INSERT INTO products (id, name, category, price) VALUES
 (1,'专业笔记本电脑','电子产品',1500.00),
 (2,'降噪耳机','电子产品',300.00),
 (3,'站立式办公桌','家具',450.00),
 (4,'人体工学椅','家具',250.00),
 (5,'27英寸显示器','电子产品',350.00);

INSERT INTO orders (id, customer_id, order_date, total) VALUES
 (1,1,'2025-02-01',1850.00),
 (2,2,'2025-02-03',600.00),
 (3,3,'2025-02-05',350.00),
 (4,1,'2025-02-07',450.00);

INSERT INTO order_items (order_id, product_id, quantity, unit_price) VALUES
 (1,1,1,1500.00),
 (1,2,1,300.00),
 (1,5,1,350.00),
 (2,3,1,450.00),
 (2,4,1,250.00),
 (3,5,1,350.00),
 (4,3,1,450.00);

INSERT INTO payments (id, order_id, payment_date, amount, method) VALUES
 (1,1,'2025-02-01',1850.00,'信用卡'),
 (2,2,'2025-02-03',600.00,'贝宝'),
 (3,3,'2025-02-05',350.00,'信用卡'),
 (4,4,'2025-02-07',450.00,'银行转账');

然后，在命令行中执行以下代码，为项目创建 SQLite 数据库：

plaintext

sqlite3 demo.db < init_db.sql

数据库准备就绪后，创建Dockerfile以实现应用程序的容器化：

plaintext

FROM python:3.12-slim
WORKDIR /code

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

同时，创建docker-compose.yml文件，以便更顺畅地运行应用程序：

plaintext

services:
  text2sql:
    build: .
    env_file: .env    
    ports:
      - "8000:8000"    
    restart: unless-stopped
    volumes:
      - ./demo.db:/code/demo.db

一切准备就绪后，启动 Docker Desktop 并执行以下代码构建应用程序：

plaintext

docker compose build --no-cache   
docker compose up -d 

如果所有步骤都顺利完成，你可以使用以下代码测试应用程序。例如，查询数据中有多少位客户：

plaintext

curl -X POST "http://localhost:8000/query" -H "Content-Type: application/json" -d "{\"question\":\"有多少位客户？\"}"

输出结果如下：

plaintext

{"sql":"SELECT COUNT(*) AS customer_count FROM customers;","result":[{"customer_count":4}]}

我们也可以尝试更复杂的查询，比如每位客户的订单数量：

plaintext

curl -X POST "http://localhost:8000/query" -H "Content-Type: application/json" -d "{\"question\":\"每位客户下了多少订单？\"}"

输出结果如下：

plaintext

{"sql":"SELECT customer_id, COUNT(*) AS number_of_orders FROM orders GROUP BY customer_id;","result":[{"customer_id":1,"number_of_orders":2},{"customer_id":2,"number_of_orders":1},{"customer_id":3,"number_of_orders":1}]}

以上就是构建基础文本转 SQL 应用程序的全部步骤。你可以进一步为其添加前端界面，或根据自身需求构建更复杂的系统。

总结

数据是所有数据工作的核心，企业依靠数据制定决策。很多时候，我们所使用的系统过于复杂，需要借助生成式人工智能来辅助操作。

在本文中，我们学习了如何使用 OpenAI 模型、FastAPI 和 SQLite 开发简易的文本转 SQL 应用程序。

原文链接：https://www.kdnuggets.com/creating-a-text-to-sql-app-with-openai-fastapi-sqlite