ChatGPT优化实战：从模型调用到性能调优的全链路指南

1次阅读

没有评论

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

在实际项目中使用 ChatGPT API 时，开发者常常会遇到以下几个典型问题：

Token 消耗不可控：ChatGPT API 按 token 计费，但 prompt 长度和响应长度都难以精确预测，导致成本难以控制
流式响应延迟：长文本生成时，即使使用流式响应，客户端仍可能感知到明显的延迟
并发限制：免费账户默认只有 3RPM（每分钟请求数），即使是付费账户也面临 TPM（每分钟 token 数）限制
错误重试复杂：API 偶尔返回 5xx 错误时，简单的固定间隔重试会导致雪崩效应

直接调用 HTTP API
优点：灵活可控，可以自定义所有请求参数
缺点：需要手动处理认证、重试、限流等基础功能
官方 SDK
优点：开箱即用，内置了基础的重试逻辑
缺点：难以定制高级功能如批处理、动态限速

对于中高级开发者，建议基于 HTTP API 自行封装，可以获得更好的性能和控制力。

批处理可以显著减少网络往返时间，特别是在高延迟环境下。以下是使用 aiohttp 实现的异步批处理示例：

import aiohttp
from typing import List, Dict, Any

async def batch_request(prompts: List[str], 
    api_key: str,
    batch_size: int = 10
) -> List[Dict[str, Any]]:
    """
    异步批量处理 ChatGPT 请求
    :param prompts: 待处理的 prompt 列表
    :param api_key: OpenAI API 密钥
    :param batch_size: 每批次请求数量
    :return: 响应结果列表
    """headers = {"Authorization": f"Bearer {api_key}","Content-Type":"application/json"
    }

    results = []
    async with aiohttp.ClientSession() as session:
        # 将 prompts 按 batch_size 分块
        for i in range(0, len(prompts), batch_size):
            batch = prompts[i:i + batch_size]
            tasks = []
            for prompt in batch:
                payload = {
                    "model": "gpt-3.5-turbo",
                    "messages": [{"role": "user", "content": prompt}]
                }
                tasks.append(
                    session.post(
                        "https://api.openai.com/v1/chat/completions",
                        json=payload,
                        headers=headers
                    )
                )

            # 并发执行当前批次的所有请求
            responses = await asyncio.gather(*tasks, return_exceptions=True)

            # 处理响应
            for resp in responses:
                if isinstance(resp, Exception):
                    results.append({"error": str(resp)})
                else:
                    json_data = await resp.json()
                    results.append(json_data)

    return results

简单的固定间隔重试在高并发下会导致请求风暴。指数退避 + 随机抖动是更好的方案：

import random
import time
from functools import wraps

def retry_with_exponential_backoff(
    max_retries: int = 5,
    initial_delay: float = 1.0,
    max_delay: float = 60.0
):
    """带指数退避和随机抖动的重试装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            retries = 0
            delay = initial_delay

            while retries < max_retries:
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if retries == max_retries - 1:
                        raise

                    # 指数退避 + 随机抖动
                    delay = min(
                        max_delay,
                        initial_delay * (2 ** retries) + random.uniform(0, 1)
                    )

                    await asyncio.sleep(delay)
                    retries += 1
        return wrapper
    return decorator

精确统计 token 使用量有助于成本控制：

import tiktoken
from functools import wraps

encoding = tiktoken.get_encoding("cl100k_base")

def count_tokens(func):
    """统计输入输出 token 用量的装饰器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 计算输入 token
        prompt = kwargs.get("prompt", "")
        input_tokens = len(encoding.encode(prompt))

        # 调用原函数
        response = await func(*args, **kwargs)

        # 计算输出 token
        output_content = response["choices"][0]["message"]["content"]
        output_tokens = len(encoding.encode(output_content))

        # 记录统计信息
        print(f"Input tokens: {input_tokens}, Output tokens: {output_tokens}")

        return response
    return wrapper

通过以下方法可以减少 prompt 的 token 消耗：