ChatGPT流传输中断问题解析：如何正确处理’正在等待完整消息’错误

1次阅读

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

流式传输是 ChatGPT API 中常见的交互方式，它允许服务端逐步返回生成的内容，而不是等待整个响应完成再一次性返回。这种方式对于长文本生成或实时对话场景尤为重要，能够显著提升用户体验。然而，流式传输也带来了新的挑战，其中最常见的就是 ” 正在等待完整消息 ” 的中断错误。

这个错误通常发生在以下几种场景：

网络抖动导致连接不稳定，数据包丢失
服务端限流或临时过载
客户端处理速度跟不上服务端发送速度
长时间空闲连接被中间设备（如防火墙）断开

这是最基础的解决方案，当检测到传输中断时，立即发起重试请求。

优点：
– 实现简单
– 快速响应

缺点：
– 可能导致重试风暴
– 不适用于服务端过载情况
– 缺乏退避策略，可能加剧问题

更成熟的解决方案，在重试之间引入延迟，且每次失败后延迟时间呈指数增长。

import asyncio
import random

async def exponential_backoff_retry(
    func,
    max_retries=5,
    initial_delay=1.0,
    max_delay=60.0,
    jitter=True
):
    """
    带指数退避和抖动 (jitter) 的重试机制
    :param func: 需要重试的异步函数
    :param max_retries: 最大重试次数
    :param initial_delay: 初始延迟(秒)
    :param max_delay: 最大延迟(秒)
    :param jitter: 是否添加随机抖动
    """
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries:
                raise

            delay = min(initial_delay * (2 ** attempt), max_delay)
            if jitter:
                delay *= random.uniform(0.5, 1.5)

            await asyncio.sleep(delay)

更完善的解决方案，需要客户端维护接收缓冲区，并对消息进行完整性校验。

flowchart TD
    A[接收数据块] --> B{是否完整消息?}
    B -->| 是 | C[处理完整消息]
    B -->| 否 | D[存入缓冲区]
    D --> E[等待更多数据]
    E --> A

以下是一个完整的 Python 实现，包含了错误处理和配置参数：

import aiohttp
import json
import asyncio
from typing import AsyncIterator

class ChatGPTStreamClient:
    def __init__(self, api_key: str, 
                 max_retries: int = 3,
                 request_timeout: int = 60,
                 backoff_base: float = 1.0):
        self.api_key = api_key
        self.max_retries = max_retries
        self.request_timeout = request_timeout
        self.backoff_base = backoff_base

    async def stream_completion(self, prompt: str) -> AsyncIterator[str]:
        """流式获取 ChatGPT 响应"""
        url = "https://api.openai.com/v1/chat/completions"
        headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": "gpt-3.5-turbo",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True
        }

        buffer = ""
        attempt = 0

        while attempt <= self.max_retries:
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(
                        url,
                        headers=headers,
                        json=payload,
                        timeout=self.request_timeout
                    ) as response:

                        if response.status != 200:
                            raise Exception(f"API 请求失败: {response.status}")

                        async for line in response.content:
                            if line.startswith(b"data:"):
                                chunk = line[6:].strip()
                                if chunk == b"[DONE]":
                                    return

                                try:
                                    data = json.loads(chunk)
                                    delta = data["choices"][0]["delta"]
                                    if "content" in delta:
                                        content = delta["content"]
                                        buffer += content
                                        yield content
                                except json.JSONDecodeError:
                                    continue

                break  # 成功完成，退出重试循环

            except Exception as e:
                attempt += 1
                if attempt > self.max_retries:
                    raise

                delay = min(self.backoff_base * (2 ** (attempt - 1)), 60)
                await asyncio.sleep(delay)

        # 处理缓冲区中的剩余内容
        if buffer:
            yield buffer