突破ChatGPT上传限额：分片上传与压缩优化实战指南

1次阅读

没有评论

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

ChatGPT API 当前对上传文件有严格的大小限制（通常为 10MB），这对于需要处理大型数据集的开发者来说是一个明显的瓶颈。在实际业务中，这种限制会直接影响以下场景：

大模型训练数据上传：训练现代 AI 模型通常需要 GB 级别的数据集
多媒体文件处理：高分辨率图片、视频或音频文件的直接上传
批量文档分析：企业级文档处理往往涉及大量文件

分片上传将大文件分割成多个小块（chunks），然后分别上传这些小块。核心优势包括：

支持断点续传（Resumable Upload）：网络中断后可从中断处继续
更好的网络容错性：单个分片失败不会影响整个上传过程
并行上传潜力：可以同时上传多个分片提高速度

智能压缩根据不同数据类型采用最优压缩策略：

文本数据：通常可获得高压缩比（5-10 倍）
二进制数据：需要选择保持质量的压缩算法
混合数据：可分离不同部分分别压缩

import aiohttp
import hashlib
import os
from typing import AsyncIterator

async def upload_in_chunks(file_path: str, chunk_size: int = 5*1024*1024):
    """
    异步分片上传实现
    :param file_path: 要上传的文件路径
    :param chunk_size: 每个分片大小 (默认 5MB)
    """
    file_size = os.path.getsize(file_path)
    chunks = file_size // chunk_size + (1 if file_size % chunk_size else 0)

    # 生成文件唯一标识
    file_hash = hashlib.md5()
    with open(file_path, 'rb') as f:
        while chunk := f.read(8192):
            file_hash.update(chunk)
    file_id = file_hash.hexdigest()

    async with aiohttp.ClientSession() as session:
        for i in range(chunks):
            offset = i * chunk_size
            remaining = file_size - offset
            current_chunk_size = min(remaining, chunk_size)

            with open(file_path, 'rb') as f:
                f.seek(offset)
                chunk_data = f.read(current_chunk_size)

                # 重试机制 (3 次)
                for attempt in range(3):
                    try:
                        async with session.put(f"https://api.chatgpt.com/upload/{file_id}/{i}",
                            data=chunk_data,
                            headers={"Content-Range": f"bytes {offset}-{offset+current_chunk_size-1}/{file_size}"}
                        ) as resp:
                            if resp.status == 200:
                                break
                            raise Exception(f"Upload failed: {resp.status}")
                    except Exception as e:
                        if attempt == 2:
                            raise
                        continue

import zlib
import lzma
from typing import Union

def smart_compress(data: Union[str, bytes], compression_level: int = 6) -> bytes:
    """
    智能压缩函数
    :param data: 输入数据 (文本或二进制)
    :param compression_level: 压缩级别 (1-9)
    :return: 压缩后的 bytes
    """
    if isinstance(data, str):
        # 文本数据使用 gzip
        return zlib.compress(data.encode('utf-8'), level=compression_level)
    else:
        # 二进制数据使用 LZMA
        return lzma.compress(data, preset=compression_level)

推荐使用令牌桶算法（Token Bucket Algorithm）控制上传并发：