OpenAI API 批量调用框架

项目概述

这是一个基于 Python asyncio 和 OpenAI 官方 SDK 构建的高性能批量调用框架。它旨在解决在需要处理大量 prompts 时，串行调用效率低下以及手动管理并发复杂的问题。

核心特性

• 异步并发：利用 asyncio 实现非阻塞 I/O，大幅提升吞吐量。
• 并发控制：内置信号量（Semaphore）机制，可灵活配置最大并发数，避免触发 API 速率限制（Rate Limit）。
• 错误隔离：单个任务失败不会影响其他任务的执行，并会记录详细的错误信息。
• 结构化输出：输入输出均为标准的 JSON 格式，便于后续数据处理和分析。
• 用量统计：自动记录每个任务的 Token 消耗情况。

快速开始

1. 环境准备

确保已安装 Python 3.8+ 及依赖库：

pip install openai
export OPENAI_API_KEY="your-api-key-here"

2. 准备任务文件

创建 tasks.json 文件，格式如下：

[
  {"id": "task_001", "prompt": "翻译这句话为英文：你好世界"},
  {"id": "task_002", "prompt": "写一首关于春天的短诗"}
]

3. 运行框架

python main.py

4. 获取结果

程序运行结束后，将在当前目录生成 results.json，包含所有任务的执行状态、输出内容及 Token 使用情况。

结果展示

本部分描述框架运行时的输入与输出对应关系，无需图片，仅通过文字逻辑展示。

场景一：正常批量处理

输入内容 (tasks.json)

用户准备了一个包含 3 个任务的 JSON 列表：

1. ID: q1, Prompt: "1+1 等于几？"
2. ID: q2, Prompt: "法国首都是哪里？"
3. ID: q3, Prompt: "请用一句话解释量子纠缠。"

控制台输出 (stdout)

开始处理 3 个任务...
处理完成：成功 3/3
结果已保存至 results.json

输出文件内容 (results.json)

生成的 JSON 文件中将包含一个数组，顺序可能与输入一致（取决于并发完成顺序，但通常由 gather 保持）：

• 第一项：
  • task_id: "q1"
  • status: "success"
  • output: "1+1 等于 2。"
  • usage: 包含 prompt_tokens, completion_tokens, total_tokens 数值。
• 第二项：
  • task_id: "q2"
  • status: "success"
  • output: "法国的首都是巴黎。"
  • usage: 包含对应的 Token 统计。
• 第三项：
  • task_id: "q3"
  • status: "success"
  • output: "量子纠缠是指两个或多个粒子生成或者相互作用的方式，使得每个粒子的量子状态都必须依据整个系统来描述，而不能单独描述其中一个粒子的状态。"
  • usage: 包含对应的 Token 统计。

场景二：混合成功与失败

输入内容 (tasks.json)

用户准备了 2 个任务，其中第二个任务因 Prompt 过长或违规导致 API 报错：

1. ID: safe_task, Prompt: "今天天气如何？"
2. ID: error_task, Prompt: "[一段触发安全策略的违规内容]"

控制台输出 (stdout)

开始处理 2 个任务...
处理完成：成功 1/2
结果已保存至 results.json

输出文件内容 (results.json)

• 第一项（成功）：
  • task_id: "safe_task"
  • status: "success"
  • output: "作为一个人工智能，我无法实时感知天气，建议您查看当地气象预报..."
  • error: null (或字段不存在)
• 第二项（失败）：
  • task_id: "error_task"
  • status: "failed"
  • output: null (或空字符串)
  • error: "Error code: 400 - {'message': 'The input content may violate our safety policies...'}"
  • 注：即使此任务失败，第一项任务的结果依然被完整保留。

场景三：高并发压力测试

输入内容

用户输入包含 100 个简短任务的列表，并在代码中将 concurrency_limit 设置为 20。

行为描述

• 框架不会同时发起 100 个请求，而是始终保持最多 20 个请求在飞行中（in-flight）。
• 每当一个请求完成（无论成功或失败），队列中的下一个任务立即补位。
• 最终生成的 results.json 文件大小约为几百 KB，包含 100 条记录，每条记录均带有唯一的 task_id 以便追溯。
• 总耗时显著低于串行调用（例如：串行需 100 秒，并发可能仅需 15-20 秒，具体取决于网络延迟和 API 响应速度）。