ChatGPT API联网实战指南：从基础配置到生产环境避坑

1次阅读

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

ChatGPT API 的联网功能是开发者实现对话交互的核心。无论是构建智能客服、内容生成工具，还是开发教育类应用，API 调用都依赖稳定的网络连接。联网过程不仅涉及简单的请求响应，还包括密钥验证、参数传递、数据处理等多个环节，任何一个步骤出现问题都可能导致服务不可用。

获取 API 密钥：登录 OpenAI 平台后，在 ”API Keys” 页面可生成专属密钥。每个密钥都与账户直接关联，务必妥善保管。
安全存储方案：
环境变量法（推荐）：
```
export OPENAI_API_KEY='your-key-here'
```
密钥管理服务：AWS Secrets Manager 或 HashiCorp Vault 等专业工具
绝对避免：将密钥直接写入代码或提交到版本控制系统

标准请求头应包含：

headers = {"Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

关键参数说明：

temperature(0-2)：值越高结果越随机
max_tokens：限制响应长度，根据业务需求设置
stream：True 时启用流式响应

import requests
import os
from time import sleep

api_key = os.getenv("OPENAI_API_KEY")
url = "https://api.openai.com/v1/chat/completions"

def chat_completion(prompt, retries=3):
    headers = {"Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7
    }

    for attempt in range(retries):
        try:
            response = requests.post(
                url, 
                json=payload, 
                headers=headers,
                timeout=10
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt == retries - 1:
                raise
            sleep(2 ** attempt)  # 指数退避

# 调用示例
try:
    result = chat_completion("如何学习 Python?")
    print(result['choices'][0]['message']['content'])
except Exception as e:
    print(f"API 调用失败: {str(e)}")

def stream_response(prompt):
    payload = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True
    }

    with requests.post(
        url,
        json=payload,
        headers=headers,
        stream=True,
        timeout=15
    ) as response:
        for chunk in response.iter_lines():
            if chunk:
                decoded = chunk.decode('utf-8')
                if decoded.startswith('data:'):
                    content = decoded[6:]
                    if content != '[DONE]':
                        yield json.loads(content)

同步版：使用concurrent.futures

from concurrent.futures import ThreadPoolExecutor

def batch_process(prompts):
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(chat_completion, prompts))
    return results

异步版（Python 3.7+）：

import aiohttp
import asyncio

async def async_chat(session, prompt):
    async with session.post(
        url,
        json={"messages": [{"role": "user", "content": prompt}]},
        headers=headers
    ) as response:
        return await response.json()

async def main(prompts):
    async with aiohttp.ClientSession() as session:
        tasks = [async_chat(session, p) for p in prompts]
        return await asyncio.gather(*tasks)

监控响应头：
x-ratelimit-limit-requests：每分钟最大请求数
x-ratelimit-remaining-requests：剩余请求数

自适应算法：

import time

class RateLimiter:
    def __init__(self, max_calls=3, period=1):
        self.max_calls = max_calls
        self.period = period
        self.timestamps = []

    def __call__(self):
        now = time.time()
        self.timestamps = [t for t in self.timestamps if t > now - self.period]

        if len(self.timestamps) >= self.max_calls:
            sleep_time = self.period - (now - self.timestamps[0])
            time.sleep(sleep_time)

        self.timestamps.append(time.time())