Python调用ChatGPT代理的完整指南：从基础实现到生产环境避坑

2次阅读

没有评论

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

在实际开发中，ChatGPT 代理常用于以下场景：

企业级应用集成对话功能
自动化客服系统
内容生成工具
数据分析辅助

开发者面临的主要挑战包括：

认证管理复杂：API 密钥需要安全存储和定期轮换
网络不稳定：需要完善的请求重试机制
并发控制困难：高并发场景下容易触发限流

常见的 Python HTTP 客户端库各有特点：

requests：简单易用，同步阻塞
httpx：支持同步 / 异步，功能全面
aiohttp：纯异步，高性能

对于 ChatGPT 代理场景，推荐使用httpx，原因如下：

同时支持同步和异步调用
内置连接池和超时管理
完善的代理支持
活跃的社区维护

建议采用环境变量存储 API 密钥，并使用密钥轮换策略：

import os
from dotenv import load_dotenv

load_dotenv()

class AuthManager:
    @staticmethod
    def get_api_key():
        return os.getenv('CHATGPT_API_KEY')

    @staticmethod
    def rotate_key():
        # 实现密钥轮换逻辑
        pass

实现带指数退避的重试策略：

import time
import httpx
from httpx import Timeout

class ChatGPTClient:
    def __init__(self):
        self.client = httpx.Client(timeout=Timeout(30.0),
            limits=httpx.Limits(max_connections=100)
        )

    def request_with_retry(self, prompt, max_retries=3):
        for attempt in range(max_retries):
            try:
                response = self.client.post(
                    "https://api.openai.com/v1/chat/completions",
                    headers={"Authorization": f"Bearer {AuthManager.get_api_key()}"},
                    json={"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}]}
                )
                response.raise_for_status()
                return response.json()
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                wait_time = min(2 ** attempt, 10)  # 指数退避，最大 10 秒
                time.sleep(wait_time)

使用 asyncio 和httpx实现异步并发：

import asyncio
import httpx

async def async_chat_completion(prompt):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {AuthManager.get_api_key()}"},
            json={"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}]}
        )
        response.raise_for_status()
        return response.json()

async def batch_process(prompts):
    tasks = [async_chat_completion(prompt) for prompt in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

批处理：将多个请求合并发送
缓存：对重复内容使用缓存
连接池：合理配置连接池大小

示例配置：

client = httpx.Client(timeout=Timeout(30.0),
    limits=httpx.Limits(
        max_connections=100,
        max_keepalive_connections=20
    )
)

连接超时：5-10 秒
读取超时：20-30 秒
总超时：不超过 60 秒

建议监控以下指标：

请求成功率
平均响应时间
错误类型分布

实现简单的令牌桶算法：

import time

class RateLimiter:
    def __init__(self, rate, capacity):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_check = time.time()

    def acquire(self):
        now = time.time()
        elapsed = now - self.last_check
        self.last_check = now

        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.rate
        )

        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

本文介绍了一套完整的 ChatGPT 代理调用方案。在实际应用中，还可以考虑：