ChatGPT API成本优化实战：如何降低大模型调用费用

1次阅读

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

ChatGPT API 采用按 token 计费模式，其中输入和输出的 token 均会计入费用。根据官方定价，gpt-3.5-turbo 模型每 1000 个 token 收费 $0.002，而 gpt- 4 模型的费用则高达 $0.03/1K tokens（输入）和 $0.06/1K tokens（输出）。以一个典型客服机器人场景为例：

平均对话轮次：5 轮
每轮用户输入：30 tokens
每轮 AI 回复：100 tokens
每日对话量：10,000 次

计算可得月费用约为：
(30+100)*5*10000*30/1000*0.002 = $390（gpt-3.5-turbo）
若升级到 gpt-4，费用将激增至 $5,850/ 月。

通过合并连续对话轮次，减少重复的上下文 token 消耗。实验表明，合理压缩可使上下文 token 减少 40-60%。

import tiktoken

def compress_conversation(messages, max_tokens=1000):
    """
    合并相似对话轮次，保留关键信息
    时间复杂度：O(n)，n 为消息条数
    """encoder = tiktoken.get_encoding("cl100k_base")
    compressed = []
    current_speaker = None
    current_batch = []

    for msg in messages:
        if msg['role'] != current_speaker:
            if current_batch:
                compressed.append({
                    'role': current_speaker,
                    'content': ' '.join(current_batch)
                })
            current_speaker = msg['role']
            current_batch = [msg['content']]
        else:
            current_batch.append(msg['content'])

    # 最终合并后检查 token 数
    total_tokens = sum(len(encoder.encode(msg['content'])) for msg in compressed)
    while total_tokens > max_tokens and len(compressed) > 1:
        compressed.pop(0)
        total_tokens = sum(len(encoder.encode(msg['content'])) for msg in compressed)

    return compressed

单条消息 token 超限时自动分割
网络错误时采用指数退避重试
保留原始对话的备份副本

graph LR
    A[用户请求] --> B{本地缓存?}
    B -->| 命中 | C[返回结果]
    B -->| 未命中 | D{Redis 缓存?}
    D -->| 命中 | E[更新本地缓存]
    D -->| 未命中 | F[调用 API]
    F --> G[写入双缓存]

参数	本地缓存	Redis 缓存
TTL	5min	24h
最大条目	1000	10000
刷新阈值	80% 命中率	60% 命中率

from datetime import timedelta
import hashlib

class ChatCache:
    def __init__(self, local_ttl=300, redis_ttl=86400):
        self.local_cache = TTLCache(maxsize=1000, ttl=local_ttl)
        self.redis = Redis(ttl=redis_ttl)

    def get_cache_key(self, messages):
        """生成对话指纹的 SHA1 哈希"""
        digest = hashlib.sha1(json.dumps(messages).encode()).hexdigest()
        return f"chat:{digest}"

    async def get_response(self, messages):
        key = self.get_cache_key(messages)

        # 本地缓存优先
        if cached := self.local_cache.get(key):
            return cached

        # 检查 Redis
        if cached := await self.redis.get(key):
            self.local_cache[key] = cached
            return cached

        # 调用 API 并缓存结果
        response = await call_chatgpt_api(messages)
        self.local_cache[key] = response
        await self.redis.set(key, response)
        return response

# prometheus/config.yml
scrape_configs:
  - job_name: 'chatgpt_api'
    metrics_path: '/metrics'
    static_configs:
      - targets: ['app:8000']

# 关键指标
chatgpt_tokens_total{type="input"} 123456
chatgpt_tokens_total{type="output"} 789012
chatgpt_api_latency_seconds_bucket{le="0.5"} 42