Python调用ChatGPT全攻略：从API封装到生产环境最佳实践

2次阅读

没有评论

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

在当前的 AI 应用开发中，集成 ChatGPT API 已成为提升产品智能水平的重要手段。然而，直接调用原始 API 会遇到诸多挑战，本文将分享一套经过生产环境验证的 Python 解决方案。

直接使用 ChatGPT 的原始 API 存在几个明显问题：

认证管理复杂：API 密钥需要安全存储和定期刷新
流式响应处理困难：直接处理分块数据容易导致不完整解析
速率限制风险：突发流量容易触发 429 错误
错误恢复机制缺失：网络波动时缺乏自动重试能力
上下文管理混乱：多轮对话需要手动维护消息历史

使用 aiohttp 而非 requests 的主要原因：

支持真正的异步 IO 操作
内置连接池管理
更轻量级的资源消耗
原生支持流式响应处理

核心功能封装：

class ChatGPTClient:
    def __init__(self, api_key: str, organization: str = None):
        self.api_key = api_key
        self.organization = organization
        self.session = aiohttp.ClientSession()
        self._retry_limit = 3
        self._timeout = aiohttp.ClientTimeout(total=30)

关键设计要点：

采用单例模式管理 HTTP 会话
内置指数退避重试策略
支持同步和异步两种调用方式
严格的类型注解

处理分块数据的正确方式：

async def _process_stream(self, response):
    buffer = ""
    async for chunk in response.content:
        buffer += chunk.decode('utf-8')
        while "\n" in buffer:
            line, buffer = buffer.split("\n", 1)
            if line.startswith("data:"):
                yield json.loads(line[5:])

# Python 3.8+
import aiohttp
import json
import time
from typing import AsyncGenerator, Dict, Optional

class ChatGPTClient:
    BASE_URL = "https://api.openai.com/v1"

    def __init__(self, api_key: str, organization: str = None):
        self.api_key = api_key
        self.organization = organization
        self.session = aiohttp.ClientSession()
        self._retry_limit = 3
        self._timeout = aiohttp.ClientTimeout(total=30)

    async def close(self):
        await self.session.close()

    async def chat_completion(
        self,
        messages: list,
        model: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        stream: bool = False
    ) -> AsyncGenerator[Dict, None]:
        """支持流式和非流式两种响应模式"""
        url = f"{self.BASE_URL}/chat/completions"
        headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        if self.organization:
            headers["OpenAI-Organization"] = self.organization

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

        for attempt in range(self._retry_limit):
            try:
                async with self.session.post(
                    url,
                    headers=headers,
                    json=payload,
                    timeout=self._timeout
                ) as response:
                    if response.status != 200:
                        error = await response.json()
                        raise Exception(f"API Error: {error}")

                    if stream:
                        async for chunk in self._process_stream(response):
                            yield chunk
                    else:
                        yield await response.json()
                break

            except Exception as e:
                if attempt == self._retry_limit - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

实现对话记忆的简单方案：

class Conversation:
    def __init__(self, system_prompt: str = None):
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})

    def add_user_message(self, content: str):
        self.messages.append({"role": "user", "content": content})

    def add_assistant_message(self, content: str):
        self.messages.append({"role": "assistant", "content": content})

    def clear(self):
        system_prompt = next((msg["content"] for msg in self.messages 
             if msg["role"] == "system"), 
            None
        )
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})

使用 circuitbreaker 的示例：

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
async def safe_chat_completion(client, conversation):
    try:
        async for response in client.chat_completion(conversation.messages):
            return response
    except Exception as e:
        metrics.counter("api_errors", tags=["type:chat_completion"])
        raise

关键监控指标：

API 响应时间 P99
错误率（按错误类型分类）
Token 使用量统计
并发请求数
重试次数分布

敏感信息过滤示例：

import logging
import re

class SensitiveDataFilter(logging.Filter):
    def filter(self, record):
        if hasattr(record, 'msg'):
            record.msg = re.sub(r"Bearer \w+", "[REDACTED]", record.msg)
        return True

应对策略：

实现指数退避重试
监控速率限制头信息
动态调整请求并发度

解决方案：

自动拆分大文本
使用 gpt-3.5-turbo-16k 等长文本模型
实现摘要式上下文压缩

精确计算方法：

import tiktoken

def count_tokens(messages, model="gpt-3.5-turbo"):
    encoding = tiktoken.encoding_for_model(model)
    return sum(len(encoding.encode(msg["content"])) for msg in messages)

值得深入探讨的方向：