1. 为什么我劝你别再手写 SQL 了
Python 内置的 sqlite3 模块,是很多人接触数据库的第一站。它不需要装服务、不需要配账号密码,一个.db文件就是整个数据库,特别适合做本地小工具、爬虫数据落地、原型验证。但真正写起来,痛点也很明显:建表语句字段一多就容易写错类型,增删改查的占位符?数量对不上,字段名和 Python 变量名来回切换,改一次表结构就得手动同步好几处代码。
我试过最笨的办法——把 SQL 写在字符串里,跑一次报一次sqlite3.OperationalError,然后对着报错一行行数括号。后来换成让 AI 帮我生成 SQL,问题又变成了:AI 生成的语句到底对不对?字段类型合不合理?能不能直接在本地 sqlite3 里跑通?如果每次都要切到网页版对话工具,复制粘贴再回来执行,这个闭环是断的。
这篇要解决的就是这个断点。核心思路是:用 TaoToken 的统一 Key 和 API 通道,把 AI 辅助生成 SQL 的能力接进你的 Python 工作流,生成完立刻在本地 sqlite3 里执行验证。你不需要在多个工具之间来回跳,配置一次,后面建表、插入、查询、改字段都能让 AI 先出草稿,你负责在本地跑通。
适合谁看:正在用 Python 做本地数据存储的开发者、写爬虫需要落地结构化数据的同学、想给 sqlite3 加一层 AI 辅助但不想折腾多套 Key 的人。下面从环境准备开始,一步步搭出可复制的闭环。
2. TaoToken 前置:一个 Key 管住所有 AI 调用
TaoToken 在这里的角色,是帮你把「调用 AI 生成 SQL」这件事标准化。你不需要为不同模型分别申请 Key、分别记不同的 base_url,统一走一个 API 地址和一把 Key 就行。对 sqlite3 这种本地开发场景来说,好处很直接:你的 Python 脚本里只需要维护一份配置,换模型、调参数都在配置层解决,业务代码不动。
先拿到你的 Key。打开官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,注册后在控制台里创建 API Key。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,Key 列表在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。创建完先复制保存,后面配置里要用。
API 的基础地址是https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为 base_url 使用。如果你用的是 OpenAI 兼容的 SDK,把 base_url 指向它、api_key 填你的 Key 就能跑。想先验证模型通不通,可以去模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 发一条测试消息,确认返回正常再进代码。
这里有个容易踩的坑:很多人把 base_url 写成带/v1或者带斜杠结尾的形式,结果请求 404。记住https://taotoken.net/api就是完整前缀,SDK 内部会自己拼路径。另外 Key 不要硬编码在脚本里提交到仓库,用环境变量或者本地配置文件,下面会给 config.toml 的骨架。
如果你后面要做长期的编码辅助、Agent 类任务,可以了解下 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ,它更适合高频调用的场景。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,遇到参数问题先查这里。
3. 可复制配置:config.toml 与 sqlite3 连接骨架
先把配置和数据库连接这两块搭好。配置文件用 TOML,Python 3.11 起标准库自带tomllib解析,低版本用tomli也行。目录结构建议这样:
project/ ├── config.toml ├── ai_sql.py └── db.dbconfig.toml内容如下,把 Key 换成你自己的:
[taotoken] base_url = "https://taotoken.net/api" api_key = "sk-你的Key" model = "gpt-4o-mini" [sqlite] db_path = "db.db"读取配置并建立 sqlite3 连接的骨架:
import sqlite3 import tomllib from pathlib import Path def load_config(path: str = "config.toml") -> dict: with open(path, "rb") as f: return tomllib.load(f) def get_conn(db_path: str) -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn if __name__ == "__main__": cfg = load_config() conn = get_conn(cfg["sqlite"]["db_path"]) cur = conn.cursor() cur.execute("select sqlite_version()") print("sqlite version:", cur.fetchone()[0]) conn.close()row_factory = sqlite3.Row这行很关键,它让查询结果可以按字段名取值,而不是只能靠下标。后面 AI 生成的字段名和结果对不上时,这个设置能省很多调试时间。跑一下确认版本打印出来,说明连接没问题。
接下来是调用 AI 生成 SQL 的函数。用 OpenAI 兼容的写法,把 base_url 指向 TaoToken:
from openai import OpenAI def build_client(cfg: dict) -> OpenAI: return OpenAI( base_url=cfg["taotoken"]["base_url"], api_key=cfg["taotoken"]["api_key"], ) def gen_sql(client: OpenAI, model: str, prompt: str) -> str: resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "你是 SQLite SQL 生成器,只输出可执行的 SQL 语句,不要解释,不要 markdown 代码块。"}, {"role": "user", "content": prompt}, ], temperature=0.2, ) return resp.choices[0].message.content.strip()system 提示里明确「只输出 SQL、不要代码块」很重要。否则模型经常给你包一层 ```sql,直接丢给cur.execute()会报语法错误。temperature 调低到 0.2,让生成的语句更稳定、更贴近标准写法。
4. 验证请求:从建表到查询跑通闭环
现在把生成和执行串起来。先让 AI 生成一张代理 IP 表的建表语句,字段参考常见的 ip、port、匿名度、类型、位置、响应速度、验证时间:
def run_sql(conn: sqlite3.Connection, sql: str): cur = conn.cursor() cur.execute(sql) conn.commit() return cur if __name__ == "__main__": cfg = load_config() client = build_client(cfg) conn = get_conn(cfg["sqlite"]["db_path"]) prompt = ( "创建表 ip_proxy,字段包括:id 整型主键自增," "ip 文本,port 整型,anonymity 文本,types 文本," "locations 文本,respondingspeed 文本,time 文本。" "使用 create table if not exists。" ) ddl = gen_sql(client, cfg["taotoken"]["model"], prompt) print("生成的建表语句:\n", ddl) run_sql(conn, ddl) print("建表完成")执行后你会看到类似这样的输出:
生成的建表语句: CREATE TABLE IF NOT EXISTS ip_proxy ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT, port INTEGER, anonymity TEXT, types TEXT, locations TEXT, respondingspeed TEXT, time TEXT ) 建表完成建表成功后,插入几条数据验证。这里用参数化查询,避免拼接字符串:
rows = [ ("122.4.40.6", 9999, "高匿名", "HTTP", "山东省济南市 电信", "1秒", "2020-04-29 19:31:01"), ("118.212.107.180", 9999, "高匿名", "HTTP", "江西省新余市 联通", "0.9秒", "2020-04-29 18:31:01"), ("125.108.84.240", 9000, "高匿名", "HTTP", "浙江省温州市 电信", "2秒", "2020-04-29 17:31:01"), ] cur = conn.cursor() cur.executemany( "insert into ip_proxy (ip, port, anonymity, types, locations, respondingspeed, time) values (?,?,?,?,?,?,?)", rows, ) conn.commit() print("插入行数:", cur.rowcount)查询验证,用sqlite3.Row按字段名取值:
cur.execute("select id, ip, port, locations from ip_proxy order by id") for row in cur.fetchall(): print(dict(row))输出应该是:
{'id': 1, 'ip': '122.4.40.6', 'port': 9999, 'locations': '山东省济南市 电信'} {'id': 2, 'ip': '118.212.107.180', 'port': 9999, 'locations': '江西省新余市 联通'} {'id': 3, 'ip': '125.108.84.240', 'port': 9000, 'locations': '浙江省温州市 电信'}到这里,AI 生成 SQL、本地 sqlite3 执行、结果验证的闭环就跑通了。你可以把 prompt 换成「查询响应速度小于 2 秒的记录」「按位置分组统计数量」「给 ip 字段加唯一索引」,AI 出语句、你本地跑,改字段结构时尤其省事。
5. 本篇常见错排查
报错sqlite3.OperationalError: near "```": syntax error模型把 SQL 包在 markdown 代码块里了。检查 system 提示是否强调「不要 markdown 代码块」,或者在gen_sql返回后加一层清洗,去掉首尾的 ``` 和sql标记。
报错sqlite3.OperationalError: table ip_proxy already exists建表语句没带if not exists。要么在 prompt 里明确要求,要么执行前先drop table if exists。生产数据别随便 drop,用if not exists更稳。
报错sqlite3.ProgrammingError: Incorrect number of bindings占位符?的数量和参数个数对不上。AI 生成的 insert 语句字段列表和 values 列表要一一对应,插入前打印一下语句和参数长度对比。
查询结果字段名对不上没设row_factory = sqlite3.Row,只能用下标取值。加上这行,或者用cur.description拿字段名手动映射。
请求返回 401 或 404401 检查 Key 是否复制完整、有没有多余空格;404 检查 base_url 是不是写成了带/v1或结尾斜杠的形式。正确写法就是https://taotoken.net/api。
中文写入后查询乱码sqlite3 默认 UTF-8,一般不会乱码。如果从外部文件读入,确认读取时指定了encoding="utf-8"。连接时不需要额外设编码。
并发写入报database is lockedsqlite3 默认同一时刻只允许一个写操作。本地单线程脚本一般遇不到,多线程场景给连接加timeout参数,或者用队列串行化写入。
6. 把 AI 接进你的 sqlite3 工作流
配置骨架和验证动作都跑通之后,你可以把这套东西固化下来。我的做法是建一个ai_sql.py工具脚本,命令行传 prompt,直接打印生成的 SQL 并可选执行:
import sys cfg = load_config() client = build_client(cfg) sql = gen_sql(client, cfg["taotoken"]["model"], sys.argv[1]) print(sql) if "--run" in sys.argv: conn = get_conn(cfg["sqlite"]["db_path"]) run_sql(conn, sql) print("已执行")这样改表结构时,python ai_sql.py "给 ip_proxy 加一个 status 文本字段" --run一条命令搞定。需要长期高频调用、做更复杂的编码辅助时,再去看看 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 的额度方案。接入过程中遇到参数或鉴权问题,接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 里有完整的请求示例。Key 管理和新建入口在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ,模型对话验证在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。
最后提醒一句:AI 生成的 SQL 一定要在本地 sqlite3 里实际跑一遍再用于正式数据。字段类型、索引、约束这些,模型给的只是草稿,cur.execute()跑通、fetchall()结果符合预期,才算真正验证完成。