Python连接ChatGPT接口实战：从API调用到生产环境优化

1次阅读

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

在当今 AI 技术迅猛发展的背景下，使用 Python 调用 ChatGPT API 已成为开发者日常工作的一部分。然而，在实际应用中，我们经常会遇到认证管理复杂、流式响应处理困难以及网络异常重试机制不完善等问题。本文将深入探讨这些痛点，并提供一套完整的解决方案。

认证复杂性：API 密钥的管理和轮换是确保安全性的关键，但很多开发者在实际应用中往往忽视这一点。
流式响应处理：处理大文本时，如何高效地逐块接收数据是一个常见的技术挑战。
网络异常重试：在网络不稳定的环境下，如何实现高效的指数退避重试机制至关重要。

适用于简单的同步请求场景，代码示例：

import requests
from requests.exceptions import RequestException

def sync_chat_completion(api_key: str, prompt: str) -> str:
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    data = {
        'model': 'gpt-3.5-turbo',
        'messages': [{'role': 'user', 'content': prompt}]
    }

    try:
        response = requests.post(
            'https://api.openai.com/v1/chat/completions',
            headers=headers,
            json=data
        )
        response.raise_for_status()
        return response.json()['choices'][0]['message']['content']
    except RequestException as e:
        print(f'Request failed: {e}')
        raise

适用于高并发场景，代码示例：

import aiohttp
import asyncio
from typing import AsyncGenerator

async def async_chat_completion(api_key: str, prompt: str) -> str:
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    data = {
        'model': 'gpt-3.5-turbo',
        'messages': [{'role': 'user', 'content': prompt}]
    }

    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(
                'https://api.openai.com/v1/chat/completions',
                headers=headers,
                json=data
            ) as response:
                response.raise_for_status()
                result = await response.json()
                return result['choices'][0]['message']['content']
        except aiohttp.ClientError as e:
            print(f'Request failed: {e}')
            raise

推荐使用环境变量管理 API 密钥，并结合密钥轮换策略：

使用 python-dotenv 加载环境变量
定期轮换 API 密钥
使用密钥管理服务（如 AWS Secrets Manager）

from os import getenv
from dotenv import load_dotenv

load_dotenv()
API_KEY = getenv('OPENAI_API_KEY')

处理大文本时，可以使用流式响应：

async def stream_chat_completion(api_key: str, prompt: str) -> AsyncGenerator[str, None]:
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    data = {
        'model': 'gpt-3.5-turbo',
        'messages': [{'role': 'user', 'content': prompt}],
        'stream': True
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            headers=headers,
            json=data
        ) as response:
            async for chunk in response.content:
                yield chunk.decode('utf-8')

实现网络异常的自动重试：

import random
import time
from typing import Callable, TypeVar

T = TypeVar('T')

def retry_with_backoff(fn: Callable[..., T],
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 10.0
) -> T:
    retries = 0
    delay = initial_delay

    while True:
        try:
            return fn()
        except Exception as e:
            if retries >= max_retries:
                raise

            time.sleep(delay)
            delay = min(delay * 2 * (1 + random.random()), max_delay)
            retries += 1

connector = aiohttp.TCPConnector(
    limit=100,  # 最大连接数
    limit_per_host=20,  # 每个主机最大连接数
    enable_cleanup_closed=True  # 自动清理关闭的连接
)

async with aiohttp.ClientSession(connector=connector) as session:
    # 使用 session 进行请求

async def batch_chat_completion(
    api_key: str,
    prompts: list[str],
    batch_size: int = 10
) -> list[str]:
    semaphore = asyncio.Semaphore(batch_size)

    async def process_prompt(prompt: str) -> str:
        async with semaphore:
            return await async_chat_completion(api_key, prompt)

    return await asyncio.gather(*[process_prompt(p) for p in prompts])

监控 API 调用频率
实现请求队列
使用令牌桶算法控制速率

def sanitize_response(text: str) -> str:
    sensitive_words = ['password', 'api_key', 'secret']
    for word in sensitive_words:
        text = text.replace(word, '[REDACTED]')
    return text