ChatGPT文件上传实战：从API调用到生产环境最佳实践

1次阅读

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

在实际开发中，ChatGPT 文件上传功能经常会遇到几个典型问题：

非结构化数据预处理困难：用户上传的文件格式多样（PDF/Word/PPT 等），后端需要统一转换为纯文本才能处理
10MB+ 大文件上传超时：默认 API 请求超时设置可能导致大文件传输中断
API 返回状态码处理复杂：需要区分网络错误（如 502）、业务错误（如 429 限流）和文件本身错误（如 415 格式不支持）

直接调用 API的优势在于灵活可控，适合定制化需求，但需要自行处理：
身份认证（Authorization 头）
请求重试（retry 机制）
错误处理（status code 解析）
SDK 封装 简化了开发流程，但可能隐藏底层细节，例如：
默认分块大小可能不适合特定场景
内置的 backoff 策略可能不符合业务需求

文件分块策略：建议根据网络状况动态调整（WiFi 环境下可用 4MB 块，移动网络用 1MB）
退避算法(backoff)：采用指数退避处理失败请求，初始延迟建议 500ms，最大不超过 5s
边界 (boundary) 处理：multipart/form-data 中每个分块需要明确边界标记，格式示例：

boundary = '----WebKitFormBoundary' + ''.join(random.choices(string.ascii_letters + string.digits, k=16))
headers['Content-Type'] = f'multipart/form-data; boundary={boundary}'

import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=0.5, max=5))
async def upload_chunk(session, url, chunk, headers):
    """带重试机制的分块上传"""
    try:
        async with session.post(url, data=chunk, headers=headers) as resp:
            if resp.status != 200:
                raise Exception(f"上传失败: {resp.status}")
            return await resp.json()
    except Exception as e:
        print(f"分块上传异常: {str(e)}")
        raise

async def parallel_upload(file_path, api_key):
    """并行分块上传主逻辑"""
    chunk_size = 4 * 1024 * 1024  # 4MB 分块
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Type': 'multipart/form-data'
    }

    async with aiohttp.ClientSession() as session:
        tasks = []
        with open(file_path, 'rb') as f:
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                crc32 = binascii.crc32(chunk)  # 校验和计算
                tasks.append(upload_chunk(session, API_URL, chunk, headers))

        # 限制并发数避免触发 API 限流
        semaphore = asyncio.Semaphore(5)
        async def limited_task(task):
            async with semaphore:
                return await task

        results = await asyncio.gather(*[limited_task(t) for t in tasks])
        return merge_results(results)  # 合并分块结果

关键实现说明：

CRC 校验：通过 binascii.crc32 确保分块完整性
并发控制：Semaphore 限制最大并发请求数
错误隔离：单个分块失败不会影响其他分块

指标名称	计算方式	报警阈值
上传成功率	成功请求数 / 总请求数	<99% (5 分钟)
P90 上传耗时	按分块大小分组的耗时百分位	>3 秒 (4MB 分块)

推荐组合方案：

文件头验证：通过 magic number 识别真实文件类型
病毒扫描：集成 ClamAV 等开源工具
敏感词过滤：使用 DFA 算法实现关键词匹配

from ratelimit import limits, sleep_and_retry

# 限制每秒 10 次 API 调用
@sleep_and_retry
@limits(calls=10, period=1)
def call_api_with_rate_limit():
    pass