ChatGPT API余额查询实战：从接口调用到错误处理全解析

7次阅读

没有评论

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

在集成 ChatGPT API 的开发过程中，API 额度的管理是一个容易被忽视但至关重要的问题。开发者常常遇到以下几种典型场景：

突发流量风险：当应用突然迎来流量高峰时，如果没有实时监控 API 余额，很可能在短时间内耗尽所有额度，导致服务中断。
多项目管理：当同一个 API Key 被用于多个项目时，如何合理分配和监控各项目的额度消耗成为一个挑战。
预算控制：对于需要严格控制成本的团队来说，实时了解 API 使用情况是避免意外支出的关键。

这些场景都指向一个共同需求：需要一个可靠的方式来查询 ChatGPT API 的余额信息。

目前主要有两种方式可以实现 ChatGPT API 余额查询：

官方接口 ：OpenAI 提供了/dashboard/billing/credit_grants 接口，这是最直接和权威的方式。
第三方监控工具：一些第三方服务提供了更丰富的监控功能，但它们需要额外的集成，并可能引入安全隐患。

对于大多数开发者来说，直接使用官方接口是更优选择。这个接口的特点包括：

采用 Bearer Token 认证机制
返回 JSON 格式的响应数据
包含总余额、已用额度、刷新周期等关键信息
受到 API 速率限制

下面是一个完整的 Python 实现，包含了错误处理和指数退避机制：

import os
import time
import requests
from requests.exceptions import RequestException

# 从环境变量获取 API Key
API_KEY = os.getenv('OPENAI_API_KEY')

# 指数退避的最大重试次数
MAX_RETRIES = 3

# 基础等待时间(秒)
BASE_DELAY = 1

def get_balance():
    """查询 ChatGPT API 余额"""
    url = "https://api.openai.com/v1/dashboard/billing/credit_grants"
    headers = {"Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    retries = 0
    while retries <= MAX_RETRIES:
        try:
            response = requests.get(url, headers=headers)

            # 处理速率限制
            if response.status_code == 429:
                retry_after = int(response.headers.get('Retry-After', BASE_DELAY))
                time.sleep(retry_after * (2 ** retries))  # 指数退避
                retries += 1
                continue

            # 处理其他错误
            response.raise_for_status()

            # 解析响应数据
            data = response.json()
            return {'total_granted': data['total_granted'],
                'total_used': data['total_used'],
                'total_available': data['total_available'],
                'expires_at': data['expires_at']
            }

        except RequestException as e:
            if retries == MAX_RETRIES:
                raise Exception(f"Failed after {MAX_RETRIES} retries: {str(e)}")

            delay = BASE_DELAY * (2 ** retries)
            time.sleep(delay)
            retries += 1

# 使用示例
if __name__ == "__main__":
    try:
        balance = get_balance()
        print(f"可用余额: {balance['total_available']} (总额: {balance['total_granted']})")
        print(f"有效期至: {balance['expires_at']}")
    except Exception as e:
        print(f"查询失败: {str(e)}")

对于 Node.js 开发者，以下是等效的实现：

const axios = require('axios');
require('dotenv').config();

// 配置
const API_KEY = process.env.OPENAI_API_KEY;
const MAX_RETRIES = 3;
const BASE_DELAY = 1000; // 毫秒

async function getBalance() {
    const url = "https://api.openai.com/v1/dashboard/billing/credit_grants";
    const headers = {"Authorization": `Bearer ${API_KEY}`,
        "Content-Type": "application/json"
    };

    let retries = 0;

    while (retries <= MAX_RETRIES) {
        try {const response = await axios.get(url, { headers});

            // 解析响应数据
            const data = response.data;
            return {
                total_granted: data.total_granted,
                total_used: data.total_used,
                total_available: data.total_available,
                expires_at: data.expires_at
            };

        } catch (error) {
            // 处理速率限制
            if (error.response && error.response.status === 429) {const retryAfter = parseInt(error.response.headers['retry-after'] || BASE_DELAY);
                await new Promise(resolve => setTimeout(resolve, retryAfter * Math.pow(2, retries)));
                retries++;
                continue;
            }

            // 处理其他错误
            if (retries === MAX_RETRIES) {throw new Error(`Failed after ${MAX_RETRIES} retries: ${error.message}`);
            }

            const delay = BASE_DELAY * Math.pow(2, retries);
            await new Promise(resolve => setTimeout(resolve, delay));
            retries++;
        }
    }
}

// 使用示例
(async () => {
    try {const balance = await getBalance();
        console.log(` 可用余额: ${balance.total_available} (总额: ${balance.total_granted})`);
        console.log(` 有效期至: ${balance.expires_at}`);
    } catch (error) {console.error(` 查询失败: ${error.message}`);
    }
})();