ChatGPT文件上传状态监控：原理分析与实现方案

1次阅读

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

在开发基于 ChatGPT API 的应用时，文件上传是常见的需求场景，比如上传 PDF、Word 等文档进行内容分析。然而，当前的 API 接口并未提供上传进度反馈机制，这给开发者带来了诸多不便：

用户体验差：用户无法感知上传进度，长时间等待可能导致误判为卡顿
错误处理困难：网络波动时难以区分是上传失败还是响应延迟
资源浪费：无法实现断点续传，大文件上传失败需重新开始

传统 HTTP 文件上传采用 multipart/form-data 格式，基本流程如下：

客户端建立 TCP 连接
发送包含文件数据的 POST 请求
服务端接收完整数据后返回响应

sequenceDiagram
    participant Client
    participant Server
    Client->>Server: POST /upload (含文件数据)
    Server-->>Client: 200 OK (仅最终响应)

由于 HTTP 的无状态特性，服务端必须在完全接收文件后才能处理请求。这导致：

服务端无法主动推送进度
中间网络设备可能缓冲数据
超时设置难以精确（受文件大小 / 网络质量影响）

方案类型	原理	延迟	服务端压力	适用场景
短轮询	定时查询上传状态	高	低	简单小文件
长轮询	保持连接直到状态变更	中	中	中等规模文件
WebSocket	双向通信实时推送	低	高	大文件 / 实时要求高

import aiohttp
import hashlib
from functools import wraps

CHUNK_SIZE = 1024 * 1024  # 1MB 分块

async def upload_file_with_progress(url, file_path, callback=None):
    """
    带进度回调的分块上传实现
    :param url: 上传接口地址
    :param file_path: 本地文件路径
    :param callback: 进度回调函数(bytes_uploaded, total_size)
    """
    md5 = hashlib.md5()
    total_size = os.path.getsize(file_path)
    uploaded = 0

    with open(file_path, 'rb') as f:
        async with aiohttp.ClientSession() as session:
            while True:
                chunk = f.read(CHUNK_SIZE)
                if not chunk:
                    break

                md5.update(chunk)
                await session.post(url, data=chunk)
                uploaded += len(chunk)

                if callback:
                    callback(uploaded, total_size)

    return md5.hexdigest()

def progress_with_retry(max_retries=3, timeout=30):
    """
    带重试机制的进度装饰器
    :param max_retries: 最大重试次数
    :param timeout: 单次请求超时(秒)
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            retry_count = 0
            last_error = None

            while retry_count < max_retries:
                try:
                    return await func(*args, **kwargs)
                except aiohttp.ClientError as e:
                    last_error = e
                    retry_count += 1
                    await asyncio.sleep(2 ** retry_count)  # 指数退避

            raise last_error if last_error else Exception('Upload failed')
        return wrapper
    return decorator