Token 用量精算与限速策略

Token 计算方式

DeepSeek 的计费基于实际处理的 token 数。一个 token 约等于 0.75 个英文单词或 0.5 个中文字符：

from openai import OpenAI

client = OpenAI(
    api_key='your-key',
    base_url='https://api.deepseek.com'
)

# 每次请求返回 usage 信息
response = client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=[{"role": "user", "content": "Hello"}]
)

print(f"Prompt tokens: {response.usage.prompt_tokens}")      # 输入消耗
print(f"Completion tokens: {response.usage.completion_tokens}")  # 输出消耗
print(f"Total tokens: {response.usage.total_tokens}")

# 思考模式下还有 reasoning_tokens
if hasattr(response.usage, 'reasoning_tokens'):
    print(f"Reasoning tokens: {response.usage.reasoning_tokens}")

费用实时估算

# V4 Flash 价格（人民币，每百万 tokens）
FLASH_INPUT_CACHE_HIT = 0.02
FLASH_INPUT_CACHE_MISS = 1.00
FLASH_OUTPUT = 2.00

def estimate_cost(prompt_tokens, completion_tokens, cache_hit=True):
    input_price = FLASH_INPUT_CACHE_HIT if cache_hit else FLASH_INPUT_CACHE_MISS
    input_cost = (prompt_tokens / 1_000_000) * input_price
    output_cost = (completion_tokens / 1_000_000) * FLASH_OUTPUT
    return input_cost + output_cost

# 示例：3000 token 输入 + 500 token 输出
cost = estimate_cost(3000, 500, cache_hit=True)
print(f"本次费用：¥{cost:.4f}")
# 输出：本次费用：¥0.0011

限速规则

超出限速会返回 429 状态码，需要实现退避重试。

智能重试机制

import time
import random

def call_with_retry(func, max_retries=3):
    """带指数退避的 API 重试"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if '429' in str(e) or 'rate_limit' in str(e).lower():
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"限流，等待 {wait:.1f} 秒后重试...")
                time.sleep(wait)
            elif attempt == max_retries - 1:
                raise
            else:
                time.sleep(1)

# 使用
result = call_with_retry(lambda: client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=[{"role": "user", "content": "Hello"}]
))

并发请求队列

import asyncio
from asyncio import Semaphore

class DeepSeekClient:
    def __init__(self, max_concurrent=10):
        self.semaphore = Semaphore(max_concurrent)
        self.client = OpenAI(
            api_key=os.environ['DEEPSEEK_API_KEY'],
            base_url='https://api.deepseek.com'
        )

    async def chat_async(self, messages, model='deepseek-v4-flash'):
        async with self.semaphore:
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(
                None,
                lambda: self.client.chat.completions.create(
                    model=model,
                    messages=messages
                )
            )

# 并发处理 100 个请求
client = DeepSeekClient(max_concurrent=10)
async def process_batch(queries):
    tasks = [client.chat_async([{"role": "user", "content": q}]) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

成本监控面板思路

在生产环境中建议记录以下指标：

• 每日总 token 消耗（按模型分）
• 缓存命中率（越高越省钱）
• 平均响应延迟
• 429 限流次数
• 按用户/API Key 的成本分摊