ChatGPT响应延迟优化实战：从网络请求到模型推理的全链路分析

1次阅读

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

最近在项目中接入 ChatGPT API 时，发现响应时间波动很大，经常出现卡顿。通过监控数据分析，典型延迟分布如下：网络传输占 60%、模型推理占 30%、其他占 10%。这促使我深入研究如何优化全链路延迟，下面分享实战经验。

使用短连接时，每次请求都需要建立 TCP 连接和 SSL 握手，开销很大。改用 Keep-Alive 连接池后，可以复用已有连接。

使用 aiohttp 实现连接池示例：

import aiohttp

async def query_chatgpt(prompt):
    connector = aiohttp.TCPConnector(limit=10, force_close=False)
    timeout = aiohttp.ClientTimeout(total=30)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        for attempt in range(3):
            try:
                async with session.post(
                    'https://api.openai.com/v1/chat/completions',
                    headers={'Authorization': f'Bearer {API_KEY}'},
                    json={'model': 'gpt-3.5-turbo', 'messages': [{'role':'user', 'content': prompt}]}
                ) as resp:
                    return await resp.json()
            except Exception as e:
                if attempt == 2: raise
                await asyncio.sleep(2 ** attempt)  # 指数退避

设置适当连接池大小 (如 10)
必须使用 async with 确保连接正确关闭
实现带指数退避的重试机制

默认情况下 API 会等待完整响应，可以启用 stream 模式分块接收数据。

SSE 流式处理示例：

async def stream_response(prompt):
    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            headers={'Authorization': f'Bearer {API_KEY}'},
            json={
                'model': 'gpt-3.5-turbo',
                'messages': [{'role':'user', 'content': prompt}],
                'stream': True
            }
        ) as resp:
            buffer = ''
            async for chunk in resp.content:
                try:
                    data = chunk.decode('utf-8').split('\n\n')
                    for line in data:
                        if line.startswith('data:'):
                            yield json.loads(line[6:])
                except UnicodeDecodeError:
                    buffer += chunk.decode('utf-8', errors='ignore')

注意处理 UTF- 8 解码边界情况
分块数据可能跨越多行，需要缓冲处理
生成器模式降低内存占用

对高频 prompt 的 embeddings 结果进行缓存：

from functools import lru_cache

@lru_cache(maxsize=1000)
async def get_embeddings(text):
    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/embeddings',
            headers={'Authorization': f'Bearer {API_KEY}'},
            json={'input': text, 'model':'text-embedding-ada-002'}
        ) as resp:
            return await resp.json()