DeepSeek V4-Pro 折扣永久化：API 调用成本降至原价四分之一

import os
from openai import OpenAI

# DeepSeek API 兼容 OpenAI SDK，只需改 base_url 和 api_key
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),  # 建议用环境变量，不要硬编码
    base_url="https://api.deepseek.com"     # DeepSeek 的 API 端点
)

def classify_text(text: str) -> str:
    """用 V4-Pro 对短文本做意图分类"""
    response = client.chat.completions.create(
        model="deepseek-v4-pro",  # 模型名称，确认用文档中的最新标识
        messages=[
            {
                "role": "system",
                "content": "你是一个文本分类器。将用户输入分类为以下类别之一：咨询、投诉、建议、其他。只输出类别名，不要解释。"
            },
            {
                "role": "user",
                "content": text
            }
        ],
        max_tokens=10,   # 分类任务输出极短，控制 token 降低成本
        temperature=0.1  # 低温度保证分类稳定
    )
    return response.choices[0].message.content.strip()

# 批量测试
samples = [
    "我的订单三天了还没发货，到底怎么回事？",
    "你们 app 的搜索功能挺好用的，能不能加个历史记录？",
    "请问你们支持企业批量采购吗？",
    "今天天气不错，随便聊聊"
]

for s in samples:
    label = classify_text(s)
    print(f"输入: {s}\n分类: {label}\n")

# 设置环境变量
export DEEPSEEK_API_KEY="sk-xxxxxxxxxxxxxxxx"

# 安装依赖（如果还没装）
pip install openai

import asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

async def classify_one(text: str) -> str:
    resp = await client.chat.completions.create(
        model="deepseek-v4-pro",
        messages=[
            {"role": "system", "content": "分类为：咨询、投诉、建议、其他。只输出类别名。"},
            {"role": "user", "content": text}
        ],
        max_tokens=10,
        temperature=0.1
    )
    return resp.choices[0].message.content.strip()

async def classify_batch(texts: list[str], concurrency: int = 20) -> list[str]:
    """控制并发数，避免触发速率限制"""
    semaphore = asyncio.Semaphore(concurrency)

    async def guarded(text):
        async with semaphore:
            return await classify_one(text)

    results = await asyncio.gather(*[guarded(t) for t in texts])
    return results

# 运行
samples = [
    "我的订单三天了还没发货",
    "你们搜索功能挺好用",
    "请问支持企业采购吗",
    "今天天气不错",
] * 25  # 模拟 100 条数据

labels = asyncio.run(classify_batch(samples, concurrency=10))
for text, label in zip(samples, labels):
    print(f"{text[:20]}... → {label}")