Python调用DeepSeek API查询ClickHouse

DeepSeek 的 API 接口与 OpenAI 高度兼容，所以最推荐、最稳定的调用方式是使用 openai Python 库（版本 ≥1.0），只需要改一下 base_url 和 api_key 即可。

1. 先决条件

• 已注册 DeepSeek 平台：https://platform.deepseek.com/
• 创建 API Key（在控制台 → API Keys 页面）
• 记下你的 API Key（格式类似 sk-xxxxxxxxxxxxxxxx）

2. 安装依赖

pip install openai

3. 基本调用示例（让模型帮你写 ClickHouse 查询）

from openai import OpenAI
import os

# ==================== 配置部分 ====================
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")          # 推荐从环境变量读取
# 或者直接写死（不推荐）：
# DEEPSEEK_API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com"         # 官方地址（也可写 https://api.deepseek.com/v1）
    # base_url="https://api.deepseek.com/v1"    # 两种写法都行
)

# ==================== 提问 ====================
response = client.chat.completions.create(
    model="deepseek-chat",          # 通用对话模型（性价比高）
    # model="deepseek-reasoner",    # 更强的推理模型（R1），适合复杂SQL场景
    messages=[
        {"role": "system", "content": "你是一位资深的 ClickHouse 专家，精通 ClickHouse SQL 语法、物化视图、分布式表、ReplacingMergeTree 等引擎、常用优化手段。请直接给出最优的 ClickHouse SQL，不要解释，除非用户明确要求。"},

        {"role": "user", "content": """
请帮我写一个 ClickHouse 查询：

现有表 user_events (MergeTree)
字段：
    user_id     UInt64
    event_time  DateTime
    event_type  LowCardinality(String)   -- login, pay, view, click 等
    amount      Nullable(Decimal(18,2))

需求：
1. 统计过去30天内每个用户总支付金额（event_type='pay'）
2. 只保留总金额 >= 1000 的用户
3. 按金额降序排序
4. 返回前20名
        """}
    ],
    temperature=0.3,          # 建议写SQL时调低一点，更稳定
    max_tokens=2000,
)

# ==================== 输出结果 ====================
print(response.choices[0].message.content)

4. 流式输出版本（推荐在终端/网页实时显示）

stream = client.chat.completions.create(
    model="deepseek-reasoner",
    messages=[
        {"role": "system", "content": "你是ClickHouse SQL专家..."},
        {"role": "user", "content": "帮我写一个查询..."}
    ],
    stream=True,
    temperature=0.2
)

for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()  # 最后换行

5. 常用模型选择（2025-2026 现阶段主流）

模型名称	用途建议	价格（大致）	上下文长度
deepseek-chat	日常对话、写简单SQL	非常便宜	128K
deepseek-reasoner	复杂SQL、分析、优化、debug	稍贵但性价比极高	128K
deepseek-coder	写代码、生成ClickHouse相关UDF	—	128K

6. 进阶用法提示（ClickHouse 场景特别有效）

• system prompt 写得越具体，SQL 越准
  示例：
  “使用最新 ClickHouse 24.x / 25.x 语法，优先使用 window 函数、argMax、groupArray 等高效函数，避免使用像 Hive 那样的低效写法。”
• 给表结构 + 几行样例数据，效果会显著提升

表结构：
CREATE TABLE analytics.user_events (
    user_id UInt64,
    event_time DateTime,
    event_type LowCardinality(String),
    page_url String,
    amount Nullable(Decimal(18,2))
) ENGINE = MergeTree()
ORDER BY (user_id, event_time);

• 想让模型帮你优化已有慢查询，直接把 EXPLAIN 结果贴进去

快速总结 – 极简版代码（复制粘贴改两行就能用）

from openai import OpenAI

client = OpenAI(api_key="你的deepseek-api-key", base_url="https://api.deepseek.com")

resp = client.chat.completions.create(
    model="deepseek-reasoner",
    messages=[{"role": "user", "content": "写一个ClickHouse查询：统计过去7天登录次数最多的10个用户"}],
    temperature=0.2
)

print(resp.choices[0].message.content)

祝你愉快地用 DeepSeek 写出又快又优雅的 ClickHouse SQL！
有具体表结构或慢查询想优化，随时贴上来～