消息批处理

消息批处理 API 是一种强大且经济高效的方式,用于异步处理大量的消息请求。这种方法非常适合不需要即时响应的任务,可以降低 50% 的成本同时提高吞吐量。

除了本指南外,您还可以直接查看 API 参考文档。

​

消息批处理 API 的工作原理

当您向消息批处理 API 发送请求时:

1. 系统会使用提供的消息请求创建一个新的消息批处理。
2. 然后批处理会异步进行,每个请求都会独立处理。
3. 您可以轮询批处理的状态,并在所有请求处理完成后检索结果。

这对于不需要即时结果的批量操作特别有用,例如:

• 大规模评估:高效处理数千个测试用例。
• 内容审核:异步分析大量用户生成的内容。
• 数据分析:为大型数据集生成见解或摘要。
• 批量内容生成:为各种目的创建大量文本(例如,产品描述、文章摘要)。

​

批处理限制

• 一个消息批处理限制为 100,000 个消息请求或 256 MB 大小,以先达到者为准。
• 批处理最多需要 24 小时生成响应,但处理可能会提前结束。在整个批处理结束之前,您无法获取批处理结果。如果处理在 24 小时内未完成,批处理将过期。
• 批处理结果在创建后 29 天内可用。之后,您仍可以查看批处理,但其结果将不再可供下载。
• 批处理的范围限定在工作区内。您可以查看 API 密钥所属工作区内创建的所有批处理及其结果。
• 速率限制适用于批处理 API 的 HTTP 请求和等待处理的批处理中的请求数量。请参阅消息批处理 API 速率限制。此外,我们可能会根据当前需求和您的请求量降低处理速度。在这种情况下,您可能会看到更多请求在 24 小时后过期。
• 由于高吞吐量和并发处理,批处理可能会略微超出您工作区配置的支出限制。

​

支持的模型

消息批处理 API 目前支持:

• Claude 3.5 Sonnet (claude-3-5-sonnet-20240620 和 claude-3-5-sonnet-20241022)
• Claude 3.5 Haiku (claude-3-5-haiku-20241022)
• Claude 3 Haiku (claude-3-haiku-20240307)
• Claude 3 Opus (claude-3-opus-20240229)

​

可以批处理的内容

您可以在批处理中包含任何可以发送到消息 API 的请求。这包括:

• 视觉
• 工具使用
• 系统消息
• 多轮对话
• 任何测试版功能

由于批处理中的每个请求都是独立处理的,您可以在单个批处理中混合使用不同类型的请求。

​

定价

批处理 API 提供显著的成本节省。所有使用费用按标准 API 价格的 50% 收取。

​

如何使用消息批处理 API

​

准备和创建批处理

消息批处理由一系列创建消息的请求组成。每个请求包含:

• 用于识别消息请求的唯一 custom_id
• 包含标准消息 API参数的 params 对象

您可以通过将此列表传递到 requests 参数来创建批处理:

在此示例中,两个独立的请求被批处理在一起进行异步处理。每个请求都有一个唯一的 custom_id 并包含您用于消息 API 调用的标准参数。

当首次创建批处理时,响应的处理状态将为 in_progress。

{
  "id": "msgbatch_01HkcTjaV5uDC8jWR4ZsDV8d",
  "type": "message_batch",
  "processing_status": "in_progress",
  "request_counts": {
    "processing": 2,
    "succeeded": 0,
    "errored": 0,
    "canceled": 0,
    "expired": 0
  },
  "ended_at": null,
  "created_at": "2024-09-24T18:37:24.100435Z",
  "expires_at": "2024-09-25T18:37:24.100435Z",
  "cancel_initiated_at": null,
  "results_url": null
}

​

跟踪批处理

消息批处理的 processing_status 字段表示批处理所处的处理阶段。它最初为 in_progress,然后在批处理中的所有请求完成处理且结果准备就绪时更新为 ended。您可以通过访问控制台或使用检索端点来监控批处理的状态:

您可以轮询此端点以了解处理何时结束。

​

检索批处理结果

一旦批处理结束,批处理中的每个消息请求都会有一个结果。有 4 种结果类型:

您可以通过批处理的 request_counts 查看结果概览,它显示了达到这四种状态的请求数量。

批处理的结果可以在控制台和消息批处理的 results_url 上下载。由于结果可能很大,建议流式传输结果而不是一次性下载所有内容。

结果将以 .jsonl 格式返回,其中每行都是一个有效的 JSON 对象,代表消息批处理中单个请求的结果。对于每个流式传输的结果,您可以根据其 custom_id 和结果类型执行不同的操作。以下是一个结果示例:

{"custom_id":"my-second-request","result":{"type":"succeeded","message":{"id":"msg_014VwiXbi91y3JMjcpyGBHX5","type":"message","role":"assistant","model":"claude-3-5-sonnet-20241022","content":[{"type":"text","text":"Hello again! It's nice to see you. How can I assist you today? Is there anything specific you'd like to chat about or any questions you have?"}],"stop_reason":"end_turn","stop_sequence":null,"usage":{"input_tokens":11,"output_tokens":36}}}}
{"custom_id":"my-first-request","result":{"type":"succeeded","message":{"id":"msg_01FqfsLoHwgeFbguDgpz48m7","type":"message","role":"assistant","model":"claude-3-5-sonnet-20241022","content":[{"type":"text","text":"Hello! How can I assist you today? Feel free to ask me any questions or let me know if there's anything you'd like to chat about."}],"stop_reason":"end_turn","stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":34}}}}

如果您的结果有错误,其 result.error 将设置为我们标准的错误格式。

​

在消息批处理中使用提示缓存

消息批处理 API 支持提示缓存,允许您可能减少批处理请求的成本和处理时间。提示缓存和消息批处理的价格折扣可以叠加,当两个功能一起使用时提供更大的成本节省。但是,由于批处理请求是异步和并发处理的,缓存命中是尽最大努力提供的。用户通常会经历 30% 到 98% 的缓存命中率,具体取决于其流量模式。

要最大化批处理请求中的缓存命中可能性:

1. 在批处理中的每个消息请求中包含相同的 cache_control 块
2. 保持稳定的请求流以防止缓存条目在 5 分钟生命周期后过期
3. 构建请求以共享尽可能多的缓存内容

实现批处理中提示缓存的示例:

在此示例中,批处理中的两个请求都包含相同的系统消息和标记了 cache_control 的《傲慢与偏见》全文,以增加缓存命中的可能性。

​

有效批处理的最佳实践

要充分利用批处理 API:

• 定期监控批处理处理状态并为失败的请求实现适当的重试逻辑。
• 使用有意义的 custom_id 值以轻松匹配结果与请求,因为顺序不能保证。
• 考虑将非常大的数据集分成多个批处理以便更好地管理。
• 使用消息 API 对单个请求格式进行预演以避免验证错误。

​

常见问题故障排除

如果遇到意外行为:

• 验证总批处理请求大小不超过 256 MB。如果请求大小太大,您可能会收到 413 request_too_large 错误。
• 检查批处理中的所有请求是否使用支持的模型。
• 确保批处理中的每个请求都有唯一的 custom_id。
• 确保自批处理 created_at(不是处理 ended_at)时间以来不超过 29 天。如果超过 29 天,结果将不再可见。
• 确认批处理尚未被取消。

请注意,批处理中一个请求的失败不会影响其他请求的处理。

​

批处理存储和隐私

• 工作区隔离: 批处理在创建它们的工作区内是隔离的。它们只能由与该工作区关联的 API 密钥或具有在控制台中查看工作区批处理权限的用户访问。

• 结果可用性: 批处理结果在创建后 29 天内可用,为检索和处理提供充足的时间。

​

常见问题

import anthropic

client = anthropic.Anthropic()

message_batch = client.beta.messages.batches.create(
    betas: ["max-tokens-3-5-sonnet-2024-07-15"],
    ...
)