Python调用ChatGPT API的实战指南：从认证到流式响应处理

2次阅读

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

ChatGPT API 为开发者提供了访问大语言模型的标准化接口，可快速集成到智能客服、内容生成等场景。Python 凭借丰富的网络库生态成为对接 API 的首选语言，其简洁语法能显著降低集成复杂度。通过程序化调用 API，开发者可以构建自动化对话流、批量处理文本分析等高阶应用。

OpenAI 官方库优势
内置对话历史管理、自动 token 计数等高级功能
默认支持流式响应处理（SSE 协议）
错误类型分类细致（如 RateLimitError）
直接使用 requests 的优势
无额外依赖包体积（官方库约 5.2MB）
更灵活的请求构造能力
性能测试显示原生 requests 延迟降低 12-15%

import os
from datetime import datetime, timedelta
import jwt

def generate_auth_header(api_key: str) -> dict:
    """ 生成带过期时间的 JWT 认证头

    Args:
        api_key: 从安全存储获取的 API 密钥

    Returns:
        {'Authorization': f'Bearer {token}'} 格式的请求头
    """
    try:
        payload = {
            'iss': 'api_client',
            'exp': datetime.utcnow() + timedelta(minutes=10)
        }
        token = jwt.encode(payload, api_key, algorithm='HS256')
        return {'Authorization': f'Bearer {token}'}
    except jwt.PyJWTError as e:
        raise ValueError(f'JWT 生成失败: {str(e)}')

import aiohttp
from enum import Enum, auto

class StreamState(Enum):
    START = auto()
    DATA = auto()
    DONE = auto()

async def stream_completion(session: aiohttp.ClientSession, prompt: str):
    state = StreamState.START
    buffer = ''async with session.post('https://api.openai.com/v1/chat/completions',
        json={'messages': [{'role': 'user', 'content': prompt}], 'stream': True},
        headers=await get_auth_header(),
        timeout=aiohttp.ClientTimeout(total=300)
    ) as resp:
        async for chunk in resp.content:
            if state == StreamState.START:
                if chunk.startswith(b'data:'):
                    state = StreamState.DATA

            if state == StreamState.DATA:
                buffer += chunk.decode('utf-8')
                if '[DONE]' in buffer:
                    state = StreamState.DONE
                else:
                    yield process_chunk(buffer)
                    buffer = ''

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
import openai

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((openai.error.APIConnectionError, openai.error.RateLimitError)
    )
)
def query_with_retry(prompt: str):
    return openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )

密钥安全管理
使用 AWS Secrets Manager 时配置自动轮换策略
本地开发环境禁止硬编码密钥，推荐使用.env 加载

消费监控方案

def log_usage(response):
    """记录 token 消耗到 Prometheus"""
    from prometheus_client import Counter

    usage = response.usage
    c = Counter('openai_token_usage', 'Token consumption', ['type'])
    c.labels('prompt').inc(usage.prompt_tokens)
    c.labels('completion').inc(usage.completion_tokens)