ChatGPT API 连接实战：从认证到流式响应的完整指南

1次阅读

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

ChatGPT API 为开发者提供了将强大的自然语言处理能力集成到各种应用中的机会。无论是构建智能客服、内容生成工具，还是开发个性化的学习助手，ChatGPT API 都能显著提升产品的智能化水平。然而，在实际集成过程中，开发者常常会遇到认证、响应处理和性能优化等方面的挑战。

OpenAI 的 API 采用 Bearer Token 认证方式，虽然原理简单，但在实际应用中，API 密钥的安全管理、认证失败的处理等问题常常困扰开发者。

ChatGPT API 返回的 JSON 数据结构较为复杂，特别是当需要处理多轮对话或流式响应时，如何高效解析和提取有用信息成为一大挑战。

对于长文本生成场景，流式处理可以显著提升用户体验，但实现起来需要考虑连接保持、分块处理和错误恢复等问题。

import os
from dotenv import load_dotenv

# 使用环境变量管理 API 密钥
load_dotenv()
API_KEY = os.getenv('OPENAI_API_KEY')

# 或者使用密钥管理服务
# import boto3
# ssm = boto3.client('ssm')
# API_KEY = ssm.get_parameter(Name='/prod/chatgpt/api_key', WithDecryption=True)['Parameter']['Value']

import requests
import json
from time import sleep

class ChatGPTClient:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.openai.com/v1/chat/completions"
        self.headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def make_request(self, messages, model="gpt-3.5-turbo", max_retries=3):
        data = {
            "model": model,
            "messages": messages,
            "temperature": 0.7
        }

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

def stream_response(messages):
    data = {
        "model": "gpt-3.5-turbo",
        "messages": messages,
        "stream": True
    }

    with requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers=headers,
        json=data,
        stream=True
    ) as response:
        response.raise_for_status()

        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data:'):
                    chunk = decoded_line[6:]
                    if chunk != '[DONE]':
                        try:
                            json_chunk = json.loads(chunk)
                            content = json_chunk['choices'][0]['delta'].get('content', '')
                            yield content
                        except json.JSONDecodeError:
                            pass

合理设置连接超时和读取超时，避免长时间等待：

# 建议值：连接超时 5 秒，读取超时 30 秒
response = requests.post(url, headers=headers, json=data, timeout=(5, 30))

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def make_api_request(data):
    response = requests.post(url, headers=headers, json=data, timeout=10)
    response.raise_for_status()
    return response.json()

使用异步请求提高吞吐量：

import aiohttp
import asyncio

async def async_make_request(session, data):
    async with session.post(url, headers=headers, json=data) as response:
        return await response.json()

async def batch_requests(messages_list):
    async with aiohttp.ClientSession() as session:
        tasks = [async_make_request(session, prepare_data(msg)) for msg in messages_list]
        return await asyncio.gather(*tasks, return_exceptions=True)