Python+DeepSeek API+TaoToken：ClickHouse 查询从未如此简单

1. 从一条 SQL 写不出来说起ClickHouse 查询快是公认的但写 SQL 这件事对很多人来说并不轻松。尤其是当表结构复杂、字段命名不规范、还要做多层聚合的时候一个查询可能要在编辑器里改十几分钟。更麻烦的是很多做运营、产品、数据分析的同学并不熟悉 ClickHouse 的 SQL 方言他们只想知道「上个月哪个渠道的转化率最高」却要等数据团队排期。我试过把 DeepSeek API 接进来让模型根据自然语言直接生成 ClickHouse SQL再通过 Python 执行返回结果。整条链路跑通之后查询效率提升非常明显。但问题也随之而来DeepSeek 的 Key、ClickHouse 的连接信息、可能还有别的模型 Key全都散落在不同的配置文件和环境变量里换一台机器就要重新配一遍非常折腾。这篇内容聚焦的就是这个场景用 Python 调用 DeepSeek API 生成 ClickHouse SQL同时用 TaoToken 统一管理模型 Key把多工具 Key 分散、配置繁琐的问题一次性解决。适合已经会用 Python 做基础脚本、想快速搭建自然语言查 ClickHouse 链路的开发者也适合需要给团队内部做一个轻量查询工具的人。下面会给出 config.toml 与 settings.json 骨架、TaoToken 统一 Key 接入步骤并附一条 ClickHouse 查询的端到端验证动作你可以直接复制跑通。2. TaoToken 前置统一 Key 管理在讲代码之前先把 Key 的问题解决掉。传统做法是把 DeepSeek API Key 写在代码里或者 .env 文件里但如果你同时用多个模型服务每个服务一套 Key、一套 Base URL管理起来很乱。TaoToken 的作用就是提供一个统一的 API 入口你只需要在 TaoToken 控制台创建一次 Key就能通过同一个 Base URL 调用包括 DeepSeek 在内的多种模型。具体操作步骤第一步打开 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并登录。第二步进入控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在 API Keys 页面创建一个新的 Key。这个 Key 就是你后续所有模型调用的凭证。第三步记录下 API 地址 https://taotoken.net/api 这个地址将作为 OpenAI SDK 的 base_url。注意这里不要加 UTM 参数直接使用这个地址即可。第四步如果你需要查看接入文档可以访问 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有详细的参数说明和示例。完成这四步之后你手里就只有一个 Key 和一个 Base URL所有模型调用都走这个入口。相比之前每个服务单独配 Key 的方式配置量减少了很多。而且 TaoToken 的 Key 权限可以在控制台统一管理不用再担心某个 Key 泄露后要到处改配置。注意API Key 相当于你的密码不要直接提交到 Git 仓库。建议用环境变量或者本地配置文件的方式加载后面会给出具体做法。3. 可复制配置config.toml 与 settings.json 骨架为了让配置和代码分离这里用两个文件来管理config.toml 放模型和数据库的连接信息settings.json 放运行时参数。这样换环境的时候只需要改配置文件不用动 Python 代码。先看 config.toml 的骨架[llm] provider taotoken base_url https://taotoken.net/api api_key sk-your-taotoken-key model deepseek-chat timeout 60 [clickhouse] host 127.0.0.1 port 8123 user default password your_password database default secure false verify true connect_timeout 30 send_receive_timeout 30 [app] max_retries 3 log_level INFO这里有几个关键点。llm 段的 base_url 填 TaoToken 的 API 地址api_key 填你在控制台创建的 Keymodel 填 deepseek-chat 或者 deepseek-reasoner。clickhouse 段填你的 ClickHouse 连接信息如果是本地默认安装host 就是 127.0.0.1port 是 8123user 是 default。再看 settings.json 的骨架{ prompt_template: 你是一个 ClickHouse SQL 专家。根据用户的问题生成一条可执行的 ClickHouse SQL只返回 SQL 语句不要解释。表结构如下\n{schema}\n用户问题{question}, schema_cache_ttl: 300, max_result_rows: 1000, output_format: table }prompt_template 是生成 SQL 的提示词模板{schema} 和 {question} 会在运行时替换。schema_cache_ttl 是表结构缓存的过期时间单位秒。max_result_rows 限制返回行数避免一次拉太多数据。output_format 控制结果输出格式table 表示用表格展示。把这两个文件放在项目根目录然后在 Python 里用 tomllib 和 json 加载。Python 3.11 以上自带 tomllib如果是低版本可以用 tomli 替代。import tomllib import json from pathlib import Path def load_config(): with open(config.toml, rb) as f: config tomllib.load(f) with open(settings.json, r, encodingutf-8) as f: settings json.load(f) return config, settings这样配置就加载进来了接下来写核心逻辑。4. 核心链路Python 调用 DeepSeek 生成 ClickHouse SQL核心链路分三步获取 ClickHouse 表结构、调用 DeepSeek 生成 SQL、执行 SQL 返回结果。下面逐个拆解。4.1 获取 ClickHouse 表结构生成 SQL 之前需要把相关表的 schema 告诉模型。这里用 clickhouse-connect 库来连接 ClickHouse 并查询表结构。import clickhouse_connect def get_table_schema(config, table_name): client clickhouse_connect.get_client( hostconfig[clickhouse][host], portconfig[clickhouse][port], usernameconfig[clickhouse][user], passwordconfig[clickhouse][password], databaseconfig[clickhouse][database], secureconfig[clickhouse][secure], verifyconfig[clickhouse][verify], ) result client.query(fDESCRIBE TABLE {table_name}) columns [] for row in result.result_rows: columns.append(f{row[0]} {row[1]}) return \n.join(columns)这段代码返回表的字段名和类型比如 id UInt64、name String、created_at DateTime。把这段文本塞进 prompt_template 的 {schema} 位置模型就能知道表结构。4.2 调用 DeepSeek 生成 SQL用 openai 库的 AsyncOpenAI 客户端base_url 指向 TaoToken 的 API 地址。from openai import AsyncOpenAI async def generate_sql(config, settings, schema, question): client AsyncOpenAI( api_keyconfig[llm][api_key], base_urlconfig[llm][base_url], timeoutconfig[llm][timeout], ) prompt settings[prompt_template].format( schemaschema, questionquestion ) response await client.chat.completions.create( modelconfig[llm][model], messages[ {role: system, content: 你是一个 ClickHouse SQL 专家。}, {role: user, content: prompt} ], temperature0.1, ) sql response.choices[0].message.content.strip() sql sql.replace(sql, ).replace(, ).strip() return sqltemperature 设成 0.1 是为了让输出更稳定减少模型自由发挥。生成结果里可能带 markdown 代码块标记用 replace 去掉。4.3 执行 SQL 并返回结果拿到 SQL 之后直接用 clickhouse-connect 执行。def execute_sql(config, sql, max_rows1000): client clickhouse_connect.get_client( hostconfig[clickhouse][host], portconfig[clickhouse][port], usernameconfig[clickhouse][user], passwordconfig[clickhouse][password], databaseconfig[clickhouse][database], secureconfig[clickhouse][secure], verifyconfig[clickhouse][verify], ) result client.query(sql) columns result.column_names rows result.result_rows[:max_rows] return columns, rows把这三步串起来就是一个完整的自然语言查 ClickHouse 的流程。4.4 完整入口脚本import asyncio import tomllib import json import clickhouse_connect from openai import AsyncOpenAI def load_config(): with open(config.toml, rb) as f: config tomllib.load(f) with open(settings.json, r, encodingutf-8) as f: settings json.load(f) return config, settings def get_table_schema(config, table_name): client clickhouse_connect.get_client( hostconfig[clickhouse][host], portconfig[clickhouse][port], usernameconfig[clickhouse][user], passwordconfig[clickhouse][password], databaseconfig[clickhouse][database], secureconfig[clickhouse][secure], verifyconfig[clickhouse][verify], ) result client.query(fDESCRIBE TABLE {table_name}) return \n.join([f{row[0]} {row[1]} for row in result.result_rows]) async def generate_sql(config, settings, schema, question): client AsyncOpenAI( api_keyconfig[llm][api_key], base_urlconfig[llm][base_url], timeoutconfig[llm][timeout], ) prompt settings[prompt_template].format(schemaschema, questionquestion) response await client.chat.completions.create( modelconfig[llm][model], messages[ {role: system, content: 你是一个 ClickHouse SQL 专家。}, {role: user, content: prompt} ], temperature0.1, ) sql response.choices[0].message.content.strip() return sql.replace(sql, ).replace(, ).strip() def execute_sql(config, sql, max_rows1000): client clickhouse_connect.get_client( hostconfig[clickhouse][host], portconfig[clickhouse][port], usernameconfig[clickhouse][user], passwordconfig[clickhouse][password], databaseconfig[clickhouse][database], secureconfig[clickhouse][secure], verifyconfig[clickhouse][verify], ) result client.query(sql) return result.column_names, result.result_rows[:max_rows] async def main(): config, settings load_config() table_name sales question 上个月销售额最高的产品是什么 schema get_table_schema(config, table_name) sql await generate_sql(config, settings, schema, question) print(f生成的 SQL\n{sql}\n) columns, rows execute_sql(config, sql, settings[max_result_rows]) print( | .join(columns)) for row in rows: print( | .join(str(v) for v in row)) if __name__ __main__: asyncio.run(main())把这段代码保存成 main.py和 config.toml、settings.json 放在同一个目录就可以运行了。5. 验证请求端到端跑通一条 ClickHouse 查询为了验证整条链路这里用一张模拟的 sales 表来做测试。先在 ClickHouse 里建表和插入数据CREATE TABLE IF NOT EXISTS sales ( id UInt64, product_name String, amount Float64, sale_date Date ) ENGINE MergeTree() ORDER BY id; INSERT INTO sales VALUES (1, 产品A, 15000, 2024-05-10), (2, 产品B, 8000, 2024-05-12), (3, 产品A, 12000, 2024-05-15), (4, 产品C, 20000, 2024-05-18), (5, 产品B, 9000, 2024-05-20);然后在 main.py 里把 table_name 改成 salesquestion 改成「上个月销售额最高的产品是什么」。运行 python main.py你会看到类似下面的输出生成的 SQL SELECT product_name, SUM(amount) AS total_sales FROM sales WHERE sale_date 2024-05-01 AND sale_date 2024-06-01 GROUP BY product_name ORDER BY total_sales DESC LIMIT 1 product_name | total_sales 产品C | 20000.0这条链路跑通之后你可以把 question 换成任何自然语言问题比如「哪个产品的平均销售额最高」「5 月份总销售额是多少」模型都会生成对应的 SQL 并执行返回结果。如果你在验证过程中想直接测试模型对话效果可以访问 https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 在页面上直接输入问题看看模型生成的 SQL 是否符合预期。这样可以快速判断是模型问题还是代码问题。6. 本篇常见错排查6.1 报错 ModuleNotFoundError: No module named clickhouse_connect说明 clickhouse-connect 没有安装。执行pip install clickhouse-connect openai tomli如果你用的是 Python 3.11 以下版本tomllib 不存在需要装 tomli 并在代码里改成 import tomli as tomllib。6.2 报错 Connection refused 或 TimeoutClickHouse 连接失败按顺序检查这几点ClickHouse 服务是否启动可以用 curl http://127.0.0.1:8123 测试config.toml 里的 host、port、user、password 是否正确如果是远程 ClickHouse防火墙是否放行了 8123 端口secure 参数是否和服务器配置匹配本地一般用 false。6.3 模型返回的 SQL 带 markdown 标记有些模型会在 SQL 外面包一层sql ...代码里已经用 replace 去掉了。如果还有残留可以在 generate_sql 函数里加正则清洗import re sql re.sub(r\w*\n?, , sql).strip()6.4 TaoToken 返回 401 或 403说明 API Key 有问题。检查 config.toml 里的 api_key 是否填的是 TaoToken 控制台创建的 Keybase_url 是否是 https://taotoken.net/api 。如果 Key 刚创建稍等几秒再试。如果还是不行去控制台确认 Key 的状态是否正常。6.5 生成的 SQL 执行报语法错误ClickHouse 的 SQL 方言和 MySQL 有差异比如日期函数、聚合函数写法不同。可以在 prompt_template 里加一句「使用 ClickHouse 语法」或者在 system message 里强调。如果某张表的字段名有歧义把完整 schema 传进去能减少出错概率。6.6 查询结果为空先确认表里确实有数据可以用 clickhouse-client 手动执行 SELECT count(*) FROM sales 看看。如果数据存在但查询为空可能是模型生成的 WHERE 条件太严格比如日期范围不对。把生成的 SQL 打印出来手动改一下条件再执行就能定位问题。7. 长期编码与 Agent 场景的配置建议如果你不只是做一次性查询而是想把这条链路集成到日常编码或者 Agent 工作流里建议把配置进一步规范化。比如把 config.toml 里的敏感信息抽到环境变量用 os.getenv 读取import os api_key os.getenv(TAOTOKEN_API_KEY, config[llm][api_key])这样在 CI/CD 或者多人协作环境里每个人只需要设置自己的环境变量不用改配置文件。另外如果你需要长期跑编码任务或者 Agent 调度可以了解 TaoToken 的 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 里面有适合持续调用的套餐方案比按次计费更划算。对于需要频繁创建和管理 Key 的场景直接收藏 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 需要新 Key 的时候随时创建不用再翻文档找入口。整条链路的核心思路就是用 TaoToken 统一模型入口用 config.toml 管理连接信息用 Python 把自然语言转成 ClickHouse SQL 并执行。配置一次后续换表、换问题都只需要改参数不用动代码。