名称： doubleword-batches
描述： 使用 Doubleword API (api.doubleword.ai) 创建和管理批量推理任务。适用于以下场景：(1) 以批量模式处理多个 AI 请求；(2) 提交 JSONL 批量文件进行异步推理；(3) 监控批量任务进度并获取结果；(4) 使用与 OpenAI 兼容的批量端点；(5) 处理无需即时响应的大规模推理工作负载。

Doubleword 批量推理

使用 Doubleword 批量 API 异步处理多个 AI 推理请求。

何时使用批量处理

批量处理适用于：
- 可同时运行的多个独立请求
- 不需要即时响应的工作负载
- 如果单独发送会超出速率限制的大规模请求
- 对成本敏感的工作负载（24 小时窗口提供更优价格）

快速开始

任何批量任务的基本工作流程：

1. 创建 JSONL 文件：包含请求（每行一个 JSON 对象）
2. 上传文件：获取文件 ID
3. 创建批量任务：使用文件 ID
4. 轮询状态：直到任务完成
5. 下载结果：从 output_file_id 下载

工作流程

步骤 1：创建批量请求文件

创建一个 .jsonl 文件，每行包含一个请求：

{"custom_id": "req-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "anthropic/claude-3-5-sonnet", "messages": [{"role": "user", "content": "What is 2+2?"}]}}
{"custom_id": "req-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "anthropic/claude-3-5-sonnet", "messages": [{"role": "user", "content": "What is the capital of France?"}]}}

每行必需字段：
- custom_id：唯一标识符（最多 64 个字符）——建议使用描述性 ID（如 "user-123-question-5"），以便于结果映射
- method：始终为 "POST"
- url：始终为 "/v1/chat/completions"
- body：标准 API 请求，包含 model 和 messages

可选的 body 参数：
- temperature：0-2（默认：1.0）
- max_tokens：最大响应 token 数
- top_p：核心采样参数
- stop：停止序列

文件限制：
- 最大大小：200MB
- 格式：仅限 JSONL（JSON Lines，换行符分隔的 JSON）
- 如需处理大规模批量，可拆分为多个文件

辅助脚本：
使用 scripts/create_batch_file.py 以编程方式生成 JSONL 文件：

python scripts/create_batch_file.py output.jsonl

修改脚本中的 requests 列表以生成特定的批量请求。

步骤 2：上传文件

上传 JSONL 文件：

curl https://api.doubleword.ai/v1/files \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY" \
  -F purpose="batch" \
  -F file="@batch_requests.jsonl"

响应包含 id 字段——保存此文件 ID 用于下一步。

步骤 3：创建批量任务

使用文件 ID 创建批量任务：

curl https://api.doubleword.ai/v1/batches \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input_file_id": "file-abc123",
    "endpoint": "/v1/chat/completions",
    "completion_window": "24h"
  }'

参数：
- input_file_id：上传步骤获取的文件 ID
- endpoint：始终为 "/v1/chat/completions"
- completion_window：选择 "24h"（价格更优）或 "1h"（加价 50%，结果更快）

响应包含批量任务 id——保存此 ID 用于状态轮询。

步骤 4：轮询状态

检查批量任务进度：

curl https://api.doubleword.ai/v1/batches/batch-xyz789 \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY"

状态流转：
1. validating - 检查输入文件格式
2. in_progress - 处理请求中
3. completed - 所有请求已完成

其他状态：
- failed - 批量任务失败（检查 error_file_id）
- expired - 批量任务超时
- cancelling/cancelled - 批量任务已取消

响应包含：
- output_file_id - 从此处下载结果
- error_file_id - 失败的请求（如果有）
- request_counts - 总计/已完成/失败的数量统计

轮询频率： 处理期间每 30-60 秒检查一次。

提前访问： 在批量任务完全完成前，可通过 output_file_id 获取结果——检查 X-Incomplete 响应头。

步骤 5：下载结果

下载完成的结果：

curl https://api.doubleword.ai/v1/files/file-output123/content \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY" \
  > results.jsonl

响应头：
- X-Incomplete: true - 批量任务仍在处理中，后续会有更多结果
- X-Last-Line: 45 - 部分下载的恢复点

输出格式（每行）：

{
  "id": "batch-req-abc",
  "custom_id": "request-1",
  "response": {
    "status_code": 200,
    "body": {
      "id": "chatcmpl-xyz",
      "choices": [{
        "message": {
          "role": "assistant",
          "content": "The answer is 4."
        }
      }]
    }
  }
}

下载错误信息（如果有）：

curl https://api.doubleword.ai/v1/files/file-error123/content \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY" \
  > errors.jsonl

错误格式（每行）：

{
  "id": "batch-req-def",
  "custom_id": "request-2",
  "error": {
    "code": "invalid_request",
    "message": "Missing required parameter"
  }
}

其他操作

列出所有批量任务

curl https://api.doubleword.ai/v1/batches?limit=10 \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY"

取消批量任务

curl https://api.doubleword.ai/v1/batches/batch-xyz789/cancel \
  -X POST \
  -H "Authorization: Bearer $DOUBLEWORD_API_KEY"

注意：
- 未处理的请求将被取消
- 已处理的结果仍可下载
- 无法取消已完成的批量任务

常用模式

处理结果

逐行解析 JSONL 输出：

import json

with open('results.jsonl') as f:
    for line in f:
        result = json.loads(line)
        custom_id = result['custom_id']
        content = result['response']['body']['choices'][0]['message']['content']
        print(f"{custom_id}: {content}")

处理部分结果

检查未完成的批量任务并恢复：

import requests

response = requests.get(
    'https://api.doubleword.ai/v1/files/file-output123/content',
    headers={'Authorization': f'Bearer {api_key}'}
)

if response.headers.get('X-Incomplete') == 'true':
    last_line = int(response.headers.get('X-Last-Line', 0))
    print(f"批量任务未完成。目前已处理 {last_line} 个请求。")
    # 继续轮询，稍后再次下载

重试失败的请求

从错误文件中提取失败的请求并重新提交：

import json

failed_ids = []
with open('errors.jsonl') as f:
    for line in f:
        error = json.loads(line)
        failed_ids.append(error['custom_id'])

print(f"失败的请求：{failed_ids}")
# 仅使用失败的请求创建新的批量任务

最佳实践

1. 使用描述性 custom_id：在 ID 中包含上下文信息，便于结果映射

   • 推荐："user-123-question-5"
   • 不推荐："1"、"req1"

2. 本地验证 JSONL：上传前确保每行都是有效的 JSON

3. 拆分大文件：保持文件大小在 200MB 限制内

4. 选择合适的窗口：为节省成本使用 24h，仅在时间敏感时使用 1h

5. 优雅处理错误：始终检查 error_file_id 并重试失败的请求

6. 监控 request_counts：通过 completed/total 比率跟踪进度

7. 保存文件 ID：存储 batch_id、input_file_id、output_file_id 以备后续检索

参考文档

有关完整的 API 详细信息（包括身份验证、速率限制和高级参数），请参阅：
- API 参考：references/api_reference.md - 完整的端点文档和模式定义