ChatGPT消息流错误处理实战：从新手到高手的避坑指南

1次阅读

没有评论

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

作为一名刚接触 ChatGPT API 开发的新手，我最近在实现消息流处理时踩了不少坑。今天就把这些经验整理成文，希望能帮助其他开发者少走弯路。

ChatGPT 的消息流 (Streaming) 模式允许我们逐步接收生成的文本，而不是等待整个响应完成。这种机制虽然提升了用户体验，但也带来了新的错误处理挑战。

网络连接中断：在长文本生成过程中，网络波动可能导致连接意外断开
API 限流错误：请求频率超过配额限制时返回 429 状态码
令牌 (token) 耗尽：达到最大上下文长度限制
内容过滤违规：触犯内容安全策略
服务端错误：OpenAI 服务器返回 5xx 错误

处理消息流错误主要有三种策略：

重试机制：适用于临时性网络问题
优点：实现简单
缺点：不适用于持续性错误
回退处理：当流式响应失败时改用完整响应
优点：保证基本功能可用
缺点：牺牲了流式体验
状态保存与恢复：记录已接收内容，从断点继续
优点：用户体验最佳
缺点：实现复杂度高

通过监听以下事件来捕获错误：

HTTP 状态码：检查响应状态
异常捕获：try-catch 块包裹关键操作
心跳检测：定期检查连接状态

def exponential_backoff(retries, base_delay=1):
    """指数退避算法实现"""
    for attempt in range(retries):
        try:
            # 尝试操作
            return operation()
        except Exception as e:
            if attempt == retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            time.sleep(delay)

import openai
import time
from typing import Generator

class ChatStreamHandler:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.max_retries = 3

    def stream_response(self, messages: list) -> Generator[str, None, None]:
        """处理流式响应的生成器函数"""
        retry_count = 0
        while retry_count < self.max_retries:
            try:
                response = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo",
                    messages=messages,
                    stream=True,
                    timeout=10  # 设置超时
                )

                collected_chunks = []
                for chunk in response:
                    if "choices" not in chunk:
                        continue

                    delta = chunk["choices"][0].get("delta", {})
                    if "content" in delta:
                        content = delta["content"]
                        collected_chunks.append(content)
                        yield content

                # 成功完成则退出重试循环
                return

            except openai.error.APIError as e:
                # 处理 API 错误
                print(f"API 错误: {e}")
                retry_count += 1
                if retry_count == self.max_retries:
                    yield "[错误] API 请求失败，请稍后重试"
                    return
                time.sleep(2 ** retry_count)  # 指数退避

            except openai.error.RateLimitError:
                yield "[警告] 达到速率限制，正在重试..."
                time.sleep(5)
                retry_count += 1

            except Exception as e:
                yield f"[错误] 发生未预期错误: {str(e)}"
                return