1. 为什么我要把 Text-to-SQL 塞进本地 AI 工具数据库查询 Agent 自然语言查 SQL说白了就是让 AI 把「上个月北京地区销售额多少」翻译成能跑的SELECT再执行、再用人话解释结果。它适合三类人不想天天写 SQL 的业务同学、需要快速取数的运营、以及想把多模型 Key 统一管起来的开发者。我试过把模型调用散落在各个脚本里结果换一个模型就要改一堆文件后来用 TaoToken 的统一 Key 加settings.json把入口收敛才算跑顺。传统链路是这样的业务提需求分析师手写 SQL来回确认字段和口径慢的时候半天就没了。Agent 链路是自然语言进Schema 理解、SQL 生成、语法校验、执行、结果解释一条龙。核心难点不在「生成」而在「生成得对、跑得安全、结果能看懂」。这篇就围绕这条链路给你一份可复制的settings.json骨架加上 TaoToken 统一 Key 的接入步骤最后用自然语言提问验证 SQL 生成结果。需要提前说清楚Agent 只做「生成 校验 执行只读查询」不碰写操作也不直连生产库跑 DDL。权限、脱敏、LIMIT 这些护栏必须自己加后面会给代码。2. TaoToken 前置统一 Key 与 API 通道怎么接TaoToken 在这里扮演的角色是「模型调用的统一入口」。你不需要在每个脚本里分别配不同厂商的 Key而是把模型请求都指向同一个 API 通道Key 也统一管理。官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 这个不加 UTM。接入前先做两件事。第一去控制台创建 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完把 Key 复制出来形如sk-xxxx。第二确认你要用的模型名可以在模型对话页先手动试一句页面是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 确认模型能正常回话再写进配置。Key 的管理原则很简单不要硬编码进代码不要提交到 Git。本地用环境变量或settings.jsonCI 里用密钥管理。下面这份settings.json就是给本地 AI 工具用的骨架把 TaoToken 的 base_url、api_key、model 三个字段集中管理其他脚本只读这份配置。{ provider: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet-4-5, fallback_model: gpt-4o-mini, timeout_seconds: 60, max_retries: 2, database: { url_env: DATABASE_URL, readonly: true, max_rows: 100 } }注意api_key_env写的是环境变量名而不是 Key 本身这样配置文件可以安全地进版本库。真正调用时从环境变量读取export TAOTOKEN_API_KEYsk-你的key export DATABASE_URLmysqlpymysql://readonly_user:passlocalhost:3306/company_db如果你更习惯用命令行工具做长期编码或 Agent 任务可以看 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它适合把这类查询 Agent 挂到日常开发流里。Key 的创建和管理入口统一在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入细节文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. 可复制配置从 settings.json 到查询 Agent3.1 读取配置与初始化模型客户端先把配置读进来再初始化一个统一的模型调用函数。所有 Text-to-SQL 的 prompt 都走这个函数换模型只改settings.json。import json import os from openai import OpenAI def load_settings(pathsettings.json): with open(path, r, encodingutf-8) as f: cfg json.load(f) cfg[api_key] os.environ[cfg[api_key_env]] cfg[db_url] os.environ[cfg[database][url_env]] return cfg SETTINGS load_settings() client OpenAI( base_urlSETTINGS[base_url], api_keySETTINGS[api_key], ) def query_ai(prompt, modelNone): model model or SETTINGS[default_model] resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], timeoutSETTINGS[timeout_seconds], ) return resp.choices[0].message.content这里用 OpenAI 兼容的 SDK 指向 TaoToken 的 base_url是最省事的接法。如果你的工具链是 Anthropic 风格ClaudeCodeAnthropic 的接入方式在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite 思路一样只是 SDK 换一下。3.2 数据库连接与 Schema 提取连接用 SQLAlchemy只读账号。Schema 提取是 Text-to-SQL 的命门AI 不知道表结构就瞎猜字段。from sqlalchemy import create_engine, text import pandas as pd class DatabaseConnector: def __init__(self, url, readonlyTrue, max_rows100): self.engine create_engine(url) self.readonly readonly self.max_rows max_rows def test_connection(self): with self.engine.connect() as conn: conn.execute(text(SELECT 1)) return True def query(self, sql): if self.readonly: forbidden [INSERT, UPDATE, DELETE, DROP, ALTER, CREATE] if any(kw in sql.upper() for kw in forbidden): raise ValueError(只读模式禁止写操作) if LIMIT not in sql.upper(): sql sql.rstrip(;) f LIMIT {self.max_rows} return pd.read_sql_query(sql, self.engine) def get_tables(self): return pd.read_sql_query( SELECT table_name FROM information_schema.tables WHERE table_schema DATABASE(), self.engine, ) def get_table_schema(self, table): return pd.read_sql_query(fDESCRIBE {table}, self.engine)query里自动补 LIMIT 是个小细节但能挡住不少「SELECT * 全表」的意外。只读校验放在执行前比事后审计靠谱。3.3 Text-to-SQL 核心转换把 Schema 拼进 prompt要求模型只输出 SQL再清理掉 markdown 代码块标记。class TextToSQL: def __init__(self, db, schema_text): self.db db self.schema schema_text def natural_to_sql(self, question): prompt f你是 SQL 专家。根据下面的数据库 schema把用户问题转成一条可执行的 SQL。 ## Schema {self.schema} ## 要求 1. 只输出 SQL不要解释 2. 中文列名用 AS 别名 3. 必须带 LIMIT 4. 只允许 SELECT ## 用户问题 {question} ## SQL sql query_ai(prompt).strip() sql sql.replace(sql, ).replace(, ).strip() return sqlSchema 文本长这样直接喂给模型表sales销售表 - id: 销售ID - date: 销售日期 - product: 产品名称 - region: 地区 - amount: 销售金额 - customer: 客户名称 表products产品表 - id: 产品ID - name: 产品名称 - category: 品类 - price: 单价3.4 校验、执行、解释三件套生成完不能直接跑先过校验器再执行最后让模型解释结果。class SQLValidator: def __init__(self, db): self.db db def validate(self, sql): issues [] for kw in [INSERT, UPDATE, DELETE, DROP, ALTER, CREATE]: if kw in sql.upper(): issues.append(f禁止操作{kw}) if SELECT in sql.upper() and LIMIT not in sql.upper(): issues.append(建议加 LIMIT) if WHERE not in sql.upper() and JOIN not in sql.upper(): issues.append(无 WHERE可能全表扫描) return {valid: not any(i.startswith(禁止) for i in issues), issues: issues}执行和解释串起来def query_and_explain(db, schema_text, question): t2s TextToSQL(db, schema_text) sql t2s.natural_to_sql(question) print(f生成 SQL: {sql}) v SQLValidator(db).validate(sql) if not v[valid]: return {success: False, error: v[issues]} df db.query(sql) if df is None or len(df) 0: return {success: False, error: 查询无结果} explain_prompt f用户问题{question} 查询结果 {df.to_string()} 请用自然语言直接回答用户问题包含关键数字。 answer query_ai(explain_prompt) return {success: True, sql: sql, data: df, answer: answer}4. 验证请求用自然语言提问并检查 SQL 生成结果配置写完跑一次完整链路。假设sales表里有 2026 年 3 月北京地区的数据。db DatabaseConnector(SETTINGS[db_url]) db.test_connection() schema_text 表sales销售表 - id: 销售ID - date: 销售日期 - product: 产品名称 - region: 地区 - amount: 销售金额 result query_and_explain(db, schema_text, 上个月北京地区的销售额是多少) if result[success]: print(SQL:, result[sql]) print(答案:, result[answer])预期输出类似生成 SQL: SELECT region AS 地区, SUM(amount) AS 销售额 FROM sales WHERE date 2026-03-01 AND date 2026-04-01 AND region 北京 GROUP BY region LIMIT 100 答案: 北京地区上个月销售额为 150 万元。验证要点有三个。第一看 SQL 里的时间范围是不是「上个月」对应的正确区间模型容易把BETWEEN写成包含 4 月 1 日导致多算一天。第二看GROUP BY和聚合函数是否匹配问题意图。第三看结果解释里的数字和df里的实际值是否一致防止模型编数。如果你想先在模型对话页手动验证模型对 SQL 的理解能力可以打开 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 把 Schema 和问题贴进去看它生成的 SQL 是否符合预期再决定用哪个模型写进settings.json。5. 本篇常见错排查5.1 报错api_key 读取为 None现象是初始化客户端时报鉴权失败。原因通常是settings.json里写了api_key_env但环境变量没导出。检查echo $TAOTOKEN_API_KEY如果为空重新export。注意load_settings里用的是os.environ[...]Key 不存在会直接抛KeyError改成os.environ.get加默认值能给出更友好的提示。5.2 报错SQL 语法错误或字段不存在模型生成的 SQL 跑不通多半是 Schema 没给全或者字段名用了中文而表里是英文。解决办法是把get_table_schema的真实输出完整拼进 prompt不要手写简化版。另外在 prompt 里明确「只能使用 Schema 中出现的字段名」。5.3 报错查询无结果但 SQL 看着没错常见于日期边界。比如「上个月」被翻译成date 2026-03-01而不是范围。在 prompt 里加一句「时间范围用 和 表示避免边界遗漏」能显著减少这类问题。另外确认数据库时区和你的预期一致。5.4 报错返回行数过多导致超时query里虽然自动补了 LIMIT但如果模型自己写了LIMIT 10000你的补丁不会覆盖。可以在校验器里加一条解析出 LIMIT 数值超过max_rows就截断或拒绝。5.5 报错模型输出带了多余解释文字有些模型会在 SQL 前后加「好的这是查询语句」。清理逻辑要更鲁棒用正则提取SELECT ... ;片段import re match re.search(rSELECT.*?(?:;|$), text, re.IGNORECASE | re.DOTALL) sql match.group().strip().rstrip(;) if match else text5.6 权限与脱敏没生效如果SecureSQL和DataMasker没接进主流程等于没做。检查query_and_explain里是否在db.query之前调用了权限校验在返回结果前调用了脱敏。脱敏列名匹配建议用配置驱动别写死在代码里。6. 把这条链路固定下来跑通一次不难难的是每次换模型、换库、换问题都稳定。我的做法是把settings.json当成唯一配置源模型名、base_url、超时、重试、只读开关、最大行数全放里面脚本只读不写。TaoToken 的统一 Key 让「换模型」变成改一个字段的事不用动业务代码。长期做编码或 Agent 任务的话Coding Plan 那条线更适合把查询 Agent 挂进日常流地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入过程中遇到鉴权或通道问题先翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 再去 API Keys 页面确认 Key 状态 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。下一步可以把结果接进可视化让 Agent 直接出图这条链路和查询是同一套配置改的只是最后一步的输出格式。 数据库查询 Agent 自然语言查 SQL：TaoToken 统一 Key 接入与 settings.json 配置实战
1. 为什么我要把 Text-to-SQL 塞进本地 AI 工具数据库查询 Agent 自然语言查 SQL说白了就是让 AI 把「上个月北京地区销售额多少」翻译成能跑的SELECT再执行、再用人话解释结果。它适合三类人不想天天写 SQL 的业务同学、需要快速取数的运营、以及想把多模型 Key 统一管起来的开发者。我试过把模型调用散落在各个脚本里结果换一个模型就要改一堆文件后来用 TaoToken 的统一 Key 加settings.json把入口收敛才算跑顺。传统链路是这样的业务提需求分析师手写 SQL来回确认字段和口径慢的时候半天就没了。Agent 链路是自然语言进Schema 理解、SQL 生成、语法校验、执行、结果解释一条龙。核心难点不在「生成」而在「生成得对、跑得安全、结果能看懂」。这篇就围绕这条链路给你一份可复制的settings.json骨架加上 TaoToken 统一 Key 的接入步骤最后用自然语言提问验证 SQL 生成结果。需要提前说清楚Agent 只做「生成 校验 执行只读查询」不碰写操作也不直连生产库跑 DDL。权限、脱敏、LIMIT 这些护栏必须自己加后面会给代码。2. TaoToken 前置统一 Key 与 API 通道怎么接TaoToken 在这里扮演的角色是「模型调用的统一入口」。你不需要在每个脚本里分别配不同厂商的 Key而是把模型请求都指向同一个 API 通道Key 也统一管理。官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 这个不加 UTM。接入前先做两件事。第一去控制台创建 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建完把 Key 复制出来形如sk-xxxx。第二确认你要用的模型名可以在模型对话页先手动试一句页面是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 确认模型能正常回话再写进配置。Key 的管理原则很简单不要硬编码进代码不要提交到 Git。本地用环境变量或settings.jsonCI 里用密钥管理。下面这份settings.json就是给本地 AI 工具用的骨架把 TaoToken 的 base_url、api_key、model 三个字段集中管理其他脚本只读这份配置。{ provider: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet-4-5, fallback_model: gpt-4o-mini, timeout_seconds: 60, max_retries: 2, database: { url_env: DATABASE_URL, readonly: true, max_rows: 100 } }注意api_key_env写的是环境变量名而不是 Key 本身这样配置文件可以安全地进版本库。真正调用时从环境变量读取export TAOTOKEN_API_KEYsk-你的key export DATABASE_URLmysqlpymysql://readonly_user:passlocalhost:3306/company_db如果你更习惯用命令行工具做长期编码或 Agent 任务可以看 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它适合把这类查询 Agent 挂到日常开发流里。Key 的创建和管理入口统一在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入细节文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. 可复制配置从 settings.json 到查询 Agent3.1 读取配置与初始化模型客户端先把配置读进来再初始化一个统一的模型调用函数。所有 Text-to-SQL 的 prompt 都走这个函数换模型只改settings.json。import json import os from openai import OpenAI def load_settings(pathsettings.json): with open(path, r, encodingutf-8) as f: cfg json.load(f) cfg[api_key] os.environ[cfg[api_key_env]] cfg[db_url] os.environ[cfg[database][url_env]] return cfg SETTINGS load_settings() client OpenAI( base_urlSETTINGS[base_url], api_keySETTINGS[api_key], ) def query_ai(prompt, modelNone): model model or SETTINGS[default_model] resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], timeoutSETTINGS[timeout_seconds], ) return resp.choices[0].message.content这里用 OpenAI 兼容的 SDK 指向 TaoToken 的 base_url是最省事的接法。如果你的工具链是 Anthropic 风格ClaudeCodeAnthropic 的接入方式在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite 思路一样只是 SDK 换一下。3.2 数据库连接与 Schema 提取连接用 SQLAlchemy只读账号。Schema 提取是 Text-to-SQL 的命门AI 不知道表结构就瞎猜字段。from sqlalchemy import create_engine, text import pandas as pd class DatabaseConnector: def __init__(self, url, readonlyTrue, max_rows100): self.engine create_engine(url) self.readonly readonly self.max_rows max_rows def test_connection(self): with self.engine.connect() as conn: conn.execute(text(SELECT 1)) return True def query(self, sql): if self.readonly: forbidden [INSERT, UPDATE, DELETE, DROP, ALTER, CREATE] if any(kw in sql.upper() for kw in forbidden): raise ValueError(只读模式禁止写操作) if LIMIT not in sql.upper(): sql sql.rstrip(;) f LIMIT {self.max_rows} return pd.read_sql_query(sql, self.engine) def get_tables(self): return pd.read_sql_query( SELECT table_name FROM information_schema.tables WHERE table_schema DATABASE(), self.engine, ) def get_table_schema(self, table): return pd.read_sql_query(fDESCRIBE {table}, self.engine)query里自动补 LIMIT 是个小细节但能挡住不少「SELECT * 全表」的意外。只读校验放在执行前比事后审计靠谱。3.3 Text-to-SQL 核心转换把 Schema 拼进 prompt要求模型只输出 SQL再清理掉 markdown 代码块标记。class TextToSQL: def __init__(self, db, schema_text): self.db db self.schema schema_text def natural_to_sql(self, question): prompt f你是 SQL 专家。根据下面的数据库 schema把用户问题转成一条可执行的 SQL。 ## Schema {self.schema} ## 要求 1. 只输出 SQL不要解释 2. 中文列名用 AS 别名 3. 必须带 LIMIT 4. 只允许 SELECT ## 用户问题 {question} ## SQL sql query_ai(prompt).strip() sql sql.replace(sql, ).replace(, ).strip() return sqlSchema 文本长这样直接喂给模型表sales销售表 - id: 销售ID - date: 销售日期 - product: 产品名称 - region: 地区 - amount: 销售金额 - customer: 客户名称 表products产品表 - id: 产品ID - name: 产品名称 - category: 品类 - price: 单价3.4 校验、执行、解释三件套生成完不能直接跑先过校验器再执行最后让模型解释结果。class SQLValidator: def __init__(self, db): self.db db def validate(self, sql): issues [] for kw in [INSERT, UPDATE, DELETE, DROP, ALTER, CREATE]: if kw in sql.upper(): issues.append(f禁止操作{kw}) if SELECT in sql.upper() and LIMIT not in sql.upper(): issues.append(建议加 LIMIT) if WHERE not in sql.upper() and JOIN not in sql.upper(): issues.append(无 WHERE可能全表扫描) return {valid: not any(i.startswith(禁止) for i in issues), issues: issues}执行和解释串起来def query_and_explain(db, schema_text, question): t2s TextToSQL(db, schema_text) sql t2s.natural_to_sql(question) print(f生成 SQL: {sql}) v SQLValidator(db).validate(sql) if not v[valid]: return {success: False, error: v[issues]} df db.query(sql) if df is None or len(df) 0: return {success: False, error: 查询无结果} explain_prompt f用户问题{question} 查询结果 {df.to_string()} 请用自然语言直接回答用户问题包含关键数字。 answer query_ai(explain_prompt) return {success: True, sql: sql, data: df, answer: answer}4. 验证请求用自然语言提问并检查 SQL 生成结果配置写完跑一次完整链路。假设sales表里有 2026 年 3 月北京地区的数据。db DatabaseConnector(SETTINGS[db_url]) db.test_connection() schema_text 表sales销售表 - id: 销售ID - date: 销售日期 - product: 产品名称 - region: 地区 - amount: 销售金额 result query_and_explain(db, schema_text, 上个月北京地区的销售额是多少) if result[success]: print(SQL:, result[sql]) print(答案:, result[answer])预期输出类似生成 SQL: SELECT region AS 地区, SUM(amount) AS 销售额 FROM sales WHERE date 2026-03-01 AND date 2026-04-01 AND region 北京 GROUP BY region LIMIT 100 答案: 北京地区上个月销售额为 150 万元。验证要点有三个。第一看 SQL 里的时间范围是不是「上个月」对应的正确区间模型容易把BETWEEN写成包含 4 月 1 日导致多算一天。第二看GROUP BY和聚合函数是否匹配问题意图。第三看结果解释里的数字和df里的实际值是否一致防止模型编数。如果你想先在模型对话页手动验证模型对 SQL 的理解能力可以打开 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 把 Schema 和问题贴进去看它生成的 SQL 是否符合预期再决定用哪个模型写进settings.json。5. 本篇常见错排查5.1 报错api_key 读取为 None现象是初始化客户端时报鉴权失败。原因通常是settings.json里写了api_key_env但环境变量没导出。检查echo $TAOTOKEN_API_KEY如果为空重新export。注意load_settings里用的是os.environ[...]Key 不存在会直接抛KeyError改成os.environ.get加默认值能给出更友好的提示。5.2 报错SQL 语法错误或字段不存在模型生成的 SQL 跑不通多半是 Schema 没给全或者字段名用了中文而表里是英文。解决办法是把get_table_schema的真实输出完整拼进 prompt不要手写简化版。另外在 prompt 里明确「只能使用 Schema 中出现的字段名」。5.3 报错查询无结果但 SQL 看着没错常见于日期边界。比如「上个月」被翻译成date 2026-03-01而不是范围。在 prompt 里加一句「时间范围用 和 表示避免边界遗漏」能显著减少这类问题。另外确认数据库时区和你的预期一致。5.4 报错返回行数过多导致超时query里虽然自动补了 LIMIT但如果模型自己写了LIMIT 10000你的补丁不会覆盖。可以在校验器里加一条解析出 LIMIT 数值超过max_rows就截断或拒绝。5.5 报错模型输出带了多余解释文字有些模型会在 SQL 前后加「好的这是查询语句」。清理逻辑要更鲁棒用正则提取SELECT ... ;片段import re match re.search(rSELECT.*?(?:;|$), text, re.IGNORECASE | re.DOTALL) sql match.group().strip().rstrip(;) if match else text5.6 权限与脱敏没生效如果SecureSQL和DataMasker没接进主流程等于没做。检查query_and_explain里是否在db.query之前调用了权限校验在返回结果前调用了脱敏。脱敏列名匹配建议用配置驱动别写死在代码里。6. 把这条链路固定下来跑通一次不难难的是每次换模型、换库、换问题都稳定。我的做法是把settings.json当成唯一配置源模型名、base_url、超时、重试、只读开关、最大行数全放里面脚本只读不写。TaoToken 的统一 Key 让「换模型」变成改一个字段的事不用动业务代码。长期做编码或 Agent 任务的话Coding Plan 那条线更适合把查询 Agent 挂进日常流地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入过程中遇到鉴权或通道问题先翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 再去 API Keys 页面确认 Key 状态 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。下一步可以把结果接进可视化让 Agent 直接出图这条链路和查询是同一套配置改的只是最后一步的输出格式。