ChatGPT API地址配置与优化：解决访问延迟与稳定性问题

1次阅读

没有评论

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

ChatGPT API 地址是开发者与 OpenAI 服务交互的入口端点。它决定了请求的路由路径、响应速度以及服务可用性。API 地址通常包含以下关键信息：

协议类型（HTTP/HTTPS）
域名或 IP 地址
端口号（默认为 443）
资源路径（如 /v1/chat/completions）

在分布式系统中，API 地址可能对应不同的服务器节点，因此地址选择直接影响请求的延迟和成功率。

访问延迟过高 ：由于地理距离或网络拥塞，某些区域访问默认 API 地址时延迟可能超过 1 秒
地址突然失效 ：OpenAI 可能调整服务节点导致旧地址返回 404 或 503 错误
地域限制 ：部分国家 / 地区可能无法直接访问某些 API 端点
配额限制 ：单个地址可能有速率限制，导致频繁触发 429 错误
SSL 证书问题 ：自签名证书或过期证书导致连接中断

OpenAI 官方 Python SDK 内置了地址解析逻辑，建议优先使用：

import openai

# 自动使用最优 API 地址
openai.api_base = 'https://api.openai.com/v1'  # 可替换为自定义地址池

维护多个可用 API 地址（建议至少 3 个）
实现简单的健康检查机制
使用加权轮询算法选择最优地址

import random
import requests

API_POOL = [
    'https://api.openai.com/v1',
    'https://api.us-east.openai.com/v1',
    'https://api.eu.openai.com/v1'
]

def get_optimal_endpoint():
    working_endpoints = []
    for endpoint in API_POOL:
        try:
            resp = requests.get(f"{endpoint}/models", timeout=2)
            if resp.status_code == 200:
                working_endpoints.append(endpoint)
        except Exception:
            continue

    return random.choice(working_endpoints) if working_endpoints else API_POOL[0]

import time
from typing import List
import requests

class ChatGPTAPIClient:
    def __init__(self, api_keys: List[str], endpoints: List[str]):
        self.api_keys = api_keys
        self.endpoints = endpoints
        self.current_key_idx = 0
        self.current_endpoint_idx = 0
        self.failures = {}  # 记录失败次数

    def _rotate_resource(self):
        # 简单的轮询策略
        self.current_endpoint_idx = (self.current_endpoint_idx + 1) % len(self.endpoints)
        if self.current_endpoint_idx == 0:
            self.current_key_idx = (self.current_key_idx + 1) % len(self.api_keys)

    def chat_completion(self, messages, model="gpt-3.5-turbo", max_retries=3):
        headers = {"Authorization": f"Bearer {self.api_keys[self.current_key_idx]}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": model,
            "messages": messages
        }

        for attempt in range(max_retries):
            endpoint = self.endpoints[self.current_endpoint_idx]
            try:
                response = requests.post(f"{endpoint}/chat/completions",
                    headers=headers,
                    json=payload,
                    timeout=10
                )

                if response.status_code == 200:
                    return response.json()

                # 处理特定错误码
                if response.status_code in [429, 503]:
                    self._rotate_resource()
                    time.sleep(2 ** attempt)  # 指数退避
                    continue

                response.raise_for_status()

            except (requests.exceptions.RequestException, ValueError) as e:
                print(f"Attempt {attempt + 1} failed: {str(e)}")
                self._rotate_resource()
                time.sleep(1)
                continue

        raise Exception("Max retries exceeded")

# 使用示例
client = ChatGPTAPIClient(api_keys=["sk-your-key-1", "sk-your-key-2"],
    endpoints=["https://api.openai.com/v1", "https://api.eu.openai.com/v1"]
)

try:
    response = client.chat_completion([{"role": "user", "content": "Hello!"}])
    print(response)
except Exception as e:
    print(f"API call failed: {e}")