ChatGPT API调用实战：从基础集成到生产环境优化

1次阅读

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

ChatGPT API 可快速集成智能对话能力到客服系统或内容生成工具中，典型场景包括自动生成营销文案和 7 ×24 小时多语言客户支持。其多轮对话和上下文理解特性，使其成为构建复杂交互系统的理想选择。

API 密钥硬编码在代码中或暴露在客户端存在严重安全隐患
多环境（开发 / 测试 / 生产）密钥轮换缺乏标准化流程

模型存在最大 token 限制（如 gpt-3.5-turbo 的 4096 tokens）
直接截断长文本会导致语义断裂和关键信息丢失

成功 / 失败响应结构不一致增加解析复杂度
流式响应 (stream response) 需要特殊处理逻辑

import aiohttp
from typing import AsyncGenerator

async def chat_completion(messages: list[dict],
    api_key: str,
    model: str = "gpt-3.5-turbo"
) -> AsyncGenerator[str, None]:
    """
    流式获取 API 响应
    :param messages: 对话消息历史
    :param api_key: OpenAI API 密钥
    :param model: 指定模型版本
    """headers = {"Authorization": f"Bearer {api_key}","Content-Type":"application/json"
    }

    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(
                "https://api.openai.com/v1/chat/completions",
                json={"model": model, "messages": messages, "stream": True},
                headers=headers,
                timeout=30
            ) as resp:
                if resp.status != 200:
                    error = await resp.json()
                    raise Exception(f"API 错误: {error.get('error', {}).get('message')}")

                async for chunk in resp.content:
                    yield chunk.decode()
        except asyncio.TimeoutError:
            raise Exception("API 请求超时")

def manage_context(conversation: list[dict], 
    new_message: str, 
    max_tokens: int = 3000
) -> list[dict]:
    """
    智能维护对话上下文窗口
    :param conversation: 当前对话历史
    :param new_message: 新用户输入
    :param max_tokens: 允许的最大 token 消耗量
    """
    # 添加新消息
    updated_conv = conversation + [{"role": "user", "content": new_message}]

    # 估算 token 数（简化版，实际应使用 tiktoken 库）while sum(len(msg["content"]) for msg in updated_conv) > max_tokens * 3.5:
        # 移除最早的非系统消息
        if len(updated_conv) > 1 and updated_conv[1]["role"] != "system":
            updated_conv.pop(1)
        else:
            break

    return updated_conv

import random
import asyncio

async def retry_with_backoff(
    coroutine_func,
    max_retries: int = 3,
    initial_delay: float = 1.0
):
    """
    带指数退避的重试装饰器
    :param coroutine_func: 需要重试的协程函数
    :param max_retries: 最大重试次数
    :param initial_delay: 初始延迟秒数
    """
    for attempt in range(max_retries):
        try:
            return await coroutine_func()
        except Exception as e:
            if "rate limit" in str(e).lower():
                wait_time = initial_delay * (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(wait_time)
                continue
            raise
    raise Exception(f"超过最大重试次数{max_retries}")