ChatGPT Token购买全指南：从API接入到成本优化实战

1次阅读

没有评论

共计 2401 个字符，预计需要花费 7 分钟才能阅读完成。

ChatGPT API 采用按 Token 计费的模式，这对开发者来说既是灵活的选择，也可能带来预算管理上的挑战。很多开发者在实际使用中常遇到以下问题：

预算失控：由于 Token 消耗速度超出预期，导致月底账单远超预算。
Token 计算误差：不同语言的 Token 折算比例不同，开发者容易低估实际消耗。
突发流量：高峰期 API 调用激增，导致 Token 消耗过快，甚至触发限流。

这些问题不仅增加了开发成本，还可能影响服务的稳定性。

Token 是 ChatGPT API 中的基本计费单位，其计算逻辑如下：

英文 Token：通常一个单词或标点符号为一个 Token。
中文 Token：一个汉字通常为 1 - 2 个 Token，具体取决于字符的复杂程度。
公式：Token 总数 = 输入 Token + 输出 Token。

OpenAI 官方文档指出，Token 的计算基于其分词器（Tokenizer）的实现，具体可参考OpenAI Tokenizer。

import tiktoken

def calculate_tokens(text):
    encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

def token_counter(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        input_text = kwargs.get('input_text', '')
        output_text = result.get('output_text', '')
        input_tokens = calculate_tokens(input_text)
        output_tokens = calculate_tokens(output_text)
        total_tokens = input_tokens + output_tokens
        print(f"Total tokens used: {total_tokens}")
        return result
    return wrapper

⚠️ 注意事项 ：确保安装了tiktoken 库，这是 OpenAI 官方推荐的 Token 计算工具。

from collections import deque
import time

class RateLimiter:
    def __init__(self, max_calls, time_window):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = deque()

    def __call__(self, func):
        def wrapped(*args, **kwargs):
            now = time.time()
            while self.calls and now - self.calls[0] > self.time_window:
                self.calls.popleft()
            if len(self.calls) >= self.max_calls:
                time_to_wait = self.time_window - (now - self.calls[0])
                time.sleep(time_to_wait)
            self.calls.append(now)
            return func(*args, **kwargs)
        return wrapped

⚠️ 注意事项：滑动窗口算法可以有效平滑突发流量，避免触发 API 限流。

import redis
import json

class ResponseCache:
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis = redis.Redis(host=host, port=port, db=db)

    def get(self, key):
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None

    def set(self, key, value, ttl=3600):
        self.redis.setex(key, ttl, json.dumps(value))

    def cached(self, func):
        def wrapper(*args, **kwargs):
            cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            cached_result = self.get(cache_key)
            if cached_result:
                return cached_result
            result = func(*args, **kwargs)
            self.set(cache_key, result)
            return result
        return wrapper

⚠️ 注意事项：Redis 缓存可以显著减少重复请求的 Token 消耗，适合响应内容相对静态的场景。