📢 转载信息
原文链接：https://www.kdnuggets.com/creating-a-text-to-sql-app-with-openai-fastapi-sqlite
原文作者：Cornellius Yudha Wijaya
Image by Author
二级标题: 引言
数据已成为任何成功业务不可或缺的资源，因为它为明智的决策提供了宝贵的见解。鉴于数据的重要性，许多公司都在构建系统来存储和分析数据。然而，尤其是在数据系统日益复杂的情况下，获取和分析所需数据常常很困难。
随着生成式AI的出现，数据工作变得异常轻松，因为我们现在可以使用简单的自然语言来获得基本准确并紧密遵循我们所提供输入的输出。这同样适用于数据处理和分析中的SQL，我们可以用自然语言来请求查询开发。
在本文中，我们将开发一个简单的API应用程序，它能将自然语言翻译成我们的数据库可以理解的SQL查询。我们将使用三个主要工具：OpenAI、FastAPI和SQLite。
以下是我们的计划。
二级标题: 文本到SQL应用开发
首先，我们将准备项目所需的一切。您只需要提供OpenAI API密钥，我们将用它来访问生成模型。为了容器化应用程序，我们将使用Docker，您可以通过Docker Desktop在本地实现它。
SQLite等其他组件在您安装Python时已存在，FastAPI稍后安装。
对于整体项目结构，我们将使用以下结构：
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
创建上述结构，或者您可以使用以下仓库让事情更简单。我们仍然会逐一查看每个文件，以了解如何开发此应用程序。
让我们从使用先前获取的OpenAI API密钥填充.env文件开始。您可以使用以下代码完成此操作：
OPENAI_API_KEY=YOUR-API-KEY
然后，转到requirements.txt文件，填入我们将使用的必要库：
fastapi uvicorn sqlalchemy openai pydantic python-dotenv
接下来，我们继续__init__.py文件，并在其中放入以下代码：
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查询的方法。
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（带有用于防止任何写入操作的保护措施）。
import os
import json
from openai import OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
_SYSTEM_PROMPT = """
You convert natural-language questions into read-only SQLite SQL. Never output INSERT / UPDATE / DELETE. Return 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"schema:\n{schema}\n\nquestion: {question}"}
]
)
payload = json.loads(response.choices[0].message.content)
return payload["sql"]
准备好代码和连接后，我们将使用FastAPI准备应用程序。该应用程序将接受自然语言问题和数据库架构，将其转换为SQL SELECT查询，通过SQLite数据库运行它们，并将结果作为JSON返回。该应用程序将是一个我们可以通过CLI访问的API。
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="Text-to-SQL Demo")
class NLRequest(BaseModel):
question: str
SCHEMA_STR = ""
@app.on_event("startup")
def capture_schema() -> None:
global SCHEMA_STR
insp = inspect(ENGINE)
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("Only SELECT statements are allowed")
return {"sql": sql, "result": run_query(sql)}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
这就是我们构建主要应用程序所需的一切。接下来要准备的是数据库。在init_db.sql中使用下面的数据库作为示例，但如果您愿意，可以随时更改它。
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,'Alice','USA','2024-01-05'),
(2,'Bob','UK','2024-03-10'),
(3,'Choi','KR','2024-06-22'),
(4,'Dara','ID','2025-01-15');
INSERT INTO products (id, name, category, price) VALUES
(1,'Laptop Pro','Electronics',1500.00),
(2,'Noise-Canceling Headphones','Electronics',300.00),
(3,'Standing Desk','Furniture',450.00),
(4,'Ergonomic Chair','Furniture',250.00),
(5,'Monitor 27"','Electronics',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,'Credit Card'),
(2,2,'2025-02-03',600.00,'PayPal'),
(3,3,'2025-02-05',350.00,'Credit Card'),
(4,4,'2025-02-07',450.00,'Bank Transfer');
然后，在CLI中运行以下代码以创建我们项目的SQLite数据库。
sqlite3 demo.db < init_db.sql
数据库准备好后，我们将创建一个Dockerfile来容器化我们的应用程序。
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文件，以便更顺畅地运行应用程序。
services:
text2sql:
build: .
env_file: .env
ports:
- "8000:8000"
restart: unless-stopped
volumes:
- ./demo.db:/code/demo.db
一切准备就绪后，启动您的Docker Desktop并运行以下代码来构建应用程序。
docker compose build --no-cache
docker compose up -d
如果一切顺利，您可以使用以下代码测试应用程序。我们将询问我们数据中有多少客户。
curl -X POST "http://localhost:8000/query" -H "Content-Type: application/json" -d "{\"question\":\"How many customers?\"}"
输出将如下所示。
{"sql":"SELECT COUNT(*) AS customer_count FROM customers;","result":[{"customer_count":4}]}
我们可以尝试更复杂的操作，例如每个客户的订单数量：
curl -X POST "http://localhost:8000/query" -H "Content-Type: application/json" -d "{\"question\":\"What is the number of orders placed by each customer\"}"
输出如下。
{"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}]}
这就是构建基本Text-to-SQL应用程序所需的一切。您可以进一步通过前端界面和更复杂的系统对其进行增强，以适应您的需求。
二级标题: 总结
数据是任何数据工作的核心，公司依靠数据来做决策。很多时候，我们现有的系统过于复杂，我们需要依靠生成式AI来帮助我们驾驭它。
在本文中，我们学习了如何使用OpenAI模型、FastAPI和SQLite开发一个简单的Text-to-SQL应用程序。
希望这对您有所帮助！
Cornellius Yudha Wijaya 是一名数据科学助理经理和数据作家。在Allianz印度尼西亚全职工作期间，他喜欢通过社交媒体和写作平台分享Python和数据技巧。Cornellius撰写了各种AI和机器学习主题的内容。
🚀 想要体验更好更全面的AI调用？
欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。
评论区