ChatGPT流式响应中断问题解析：如何实现消息完整接收

1次阅读

没有评论

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

ChatGPT 的流式 API（streaming API）允许开发者逐步接收响应，而不是等待整个响应完成。这种机制特别适合处理长文本或实时交互场景。然而，在实际使用中，经常会遇到响应中断的问题，主要原因包括：

网络抖动 ：不稳定的网络连接可能导致数据传输中断。
速率限制 ：API 的速率限制可能导致请求被临时阻断。
长文本截断 ：某些情况下，服务器可能会主动截断过长的响应。

这些问题会导致开发者无法完整接收响应，影响用户体验。

最简单的解决方案是在中断时重新发起请求。然而，这种方法有以下缺点：

重复请求可能导致重复计费。
无法保证中断时已接收的部分消息不会丢失。
频繁重试可能触发 API 的速率限制。

相比之下，使用消息缓冲队列可以更有效地处理中断问题：

缓存已接收的部分消息，避免数据丢失。
支持断点续传，减少重复请求。
通过异步处理提高效率。

以下是基于 Python asyncio 的实现思路：

使用一个队列（asyncio.Queue）缓存已接收的消息块。
为每个消息块设置过期时间，避免内存泄漏。
在中断时，从队列中恢复未完成的消息。

检测到中断时，自动重连并继续接收剩余消息。
将新接收的消息与缓存中的消息拼接，确保完整性。

import asyncio
from typing import AsyncGenerator, List, Optional
import logging

logger = logging.getLogger(__name__)


class StreamingBuffer:
    def __init__(self, timeout: int = 30):
        self.queue = asyncio.Queue()
        self.timeout = timeout
        self.last_message_time = asyncio.get_event_loop().time()

    async def put(self, chunk: str):
        await self.queue.put(chunk)
        self.last_message_time = asyncio.get_event_loop().time()

    async def get(self) -> Optional[str]:
        try:
            return await asyncio.wait_for(self.queue.get(), timeout=self.timeout)
        except asyncio.TimeoutError:
            return None


async def stream_with_retry(prompt: str, max_retries: int = 3) -> AsyncGenerator[str, None]:
    buffer = StreamingBuffer()
    retries = 0

    while retries < max_retries:
        try:
            # 模拟 API 请求
            async for chunk in mock_streaming_api(prompt):
                await buffer.put(chunk)
                yield chunk
            break  # 成功完成
        except Exception as e:
            logger.warning(f"Stream interrupted: {e}")
            retries += 1
            if retries >= max_retries:
                raise

    # 从缓存中读取剩余消息
    while True:
        chunk = await buffer.get()
        if chunk is None:
            break
        yield chunk


async def mock_streaming_api(prompt: str) -> AsyncGenerator[str, None]:
    """模拟流式 API 响应，随机生成中断"""
    chunks = ["Hello", "","world","!"]
    for chunk in chunks:
        await asyncio.sleep(0.1)
        if random.random() < 0.2:  # 20% 概率中断
            raise Exception("Mock API interruption")
        yield chunk