如何用 AI SDK 的 Batch API 提交异步批处理并跟踪结果状态

如何用 AI SDK 的 Batch API 提交异步批处理并跟踪结果状态

【免费下载链接】aiThe AI Toolkit for TypeScript. From the creators of Next.js, the AI SDK is a free open-source library for building AI-powered applications and agents项目地址: https://gitcode.com/GitHub_Trending/ai/ai

如果你的工作负载是一批相互独立、又不需要即时响应的请求（例如分类、摘要、内容生成），可以不用为每个请求长时间保持连接。AI SDK 的 Batch API 允许你把多个请求作为一个批次提交给 provider，由 provider 在后台异步处理，你这边只需定期查询状态、取回终态结果。

适用前提：Node.js 22+ 与 pnpm（依据 Node.js 快速入门的要求），以及ai包。需要注意 Batch 支持目前是实验性能力，官方文档明确提示 API 可能在补丁版本中变化，升级时留意变更。

准备环境

按照 AI SDK 的 Node.js 快速入门初始化一个项目：

mkdir my-batch-app cd my-batch-app pnpm init

安装依赖：

pnpm add ai zod dotenv pnpm add -D @types/node tsx typescript

其中ai是 AI SDK 主包，五个 Batch 生命周期函数都从这里导出；tsx用于直接运行 TypeScript 文件。

创建项目根目录下的.env文件并写入你的密钥。走 AI Gateway（全局默认 provider）时，环境变量是AI_GATEWAY_API_KEY，AI SDK 用它与 Vercel AI Gateway 鉴权：

AI_GATEWAY_API_KEY=xxxxxxxxx

把xxxxxxxxx替换为你的实际 API 密钥。如果你想直连某个 provider（如 Anthropic），则安装对应的 provider 包（例如pnpm add @ai-sdk/anthropic），并按该 provider 的 provider 文档完成凭证配置。

选择支持批处理的 provider

批处理要求 provider 实现了 batch 接口，支持情况按 provider 和模型区分。官方文档列出的第一方支持如下：

各 provider 支持的模型、限制和原生批处理行为以其 provider 文档为准。有一个细节要注意：OpenAI 和 xAI 的批处理支持是通过 Responses API 提供的，而不是openai.chat()/xai.chat()。

提交批次：startBatch

AI SDK 提供五个带experimental_前缀的函数，均从ai导出：

• experimental_startBatch：提交批次，返回初始状态和可序列化的批次引用；
• experimental_getBatchStatus：查询最新状态与请求计数；
• experimental_getBatchResults：异步迭代每个请求的终态结果；
• experimental_cancelBatch：请求取消批次；
• experimental_listBatches：分页列出批次及最新状态。

下面的主路径使用全局默认 provider（未配置全局 provider 时即 AI Gateway，依赖前面的AI_GATEWAY_API_KEY），把提交、轮询、取结果串成一个完整脚本。把它保存为index.ts，用pnpm tsx index.ts运行：

import { experimental_getBatchResults as getBatchResults, experimental_getBatchStatus as getBatchStatus, experimental_startBatch as startBatch, } from 'ai'; import { setTimeout } from 'node:timers/promises'; import 'dotenv/config'; const batch = await startBatch({ requests: [ { id: 'capital-france', type: 'text', model: 'gpt-4.1-nano', prompt: 'What is the capital of France?', }, { id: 'capital-germany', type: 'text', model: 'gpt-4.1-nano', prompt: 'What is the capital of Germany?', }, ], }); console.log(batch.id, batch.status); let status = batch.status; let error = batch.error; while (status === 'pending') { await setTimeout(10_000); const latestStatus = await getBatchStatus({ batch }); status = latestStatus.status; error = latestStatus.error; } if (status === 'failed') { throw new Error(error?.message ?? 'The batch failed.'); } for await (const item of getBatchResults({ batch })) { if (item.status === 'succeeded') { console.log(item.id, item.text); } else { console.error(item.id, item.status, item.error); } }

requests中的每个请求必须指定type: 'text'、一个模型 ID，以及文本prompt或messages数组。请求 ID 必须非空且在批次内唯一——它是输入请求与结果之间的关联键。结果不保证按输入顺序到达，所以应用侧应通过item.id而不是数组下标来把结果映射回业务数据。provider 支持的模型允许时，不同请求可以使用不同的模型 ID。

每个请求还可以带常规文本生成设置，如instructions、maxOutputTokens、temperature、topP、topK、presencePenalty、frequencyPenalty、stopSequences、seed、reasoning。provider 专属配置可以放在批次的providerOptions上，也可以放在单个请求的providerOptions上，两个层级支持的配置项可能不同；startBatch返回结果中的warnings属性会提示不受支持的配置。

直连 provider 时，把 provider 实例显式传入，之后查询状态和取结果也要传同一个 provider。以 Anthropic 为例（示例来自官方指南，模型claude-haiku-4-5）：

import { anthropic } from '@ai-sdk/anthropic'; import { experimental_startBatch as startBatch } from 'ai'; const provider = anthropic; const batch = await startBatch({ provider, requests: [ { id: 'capital-france', type: 'text', model: 'claude-haiku-4-5', prompt: 'What is the capital of France?', }, ], });

startBatch返回的批次引用是可序列化的（含id、provider、初始status等字段）。如果批次会由另一个进程或在稍后时间完成，在进程退出前先把它持久化；取回批次时传入相同的 provider，SDK 会用它确保批次不会经由不兼容的 provider 读取。

跟踪批次状态：getBatchStatus

上面的轮询循环依赖getBatchStatus返回的归一化状态，取值只有三种：

• pending：provider 仍在处理；
• completed：批次到达终态，可以取结果；
• failed：批次无法完成，error属性可能包含详情。

状态响应还可能包含requestCounts（provider 已知的 total、pending、completed、failed 请求数）、createdAt、expiresAt、rawStatus和 provider 元数据。当状态为failed时，把error?.message抛出来是文档给出的处理方式。

如果不适合持续轮询，可以改用 webhook：给startBatch传webhookUrl，批次到达终态时 provider 会通知该地址。webhook 的支持与载荷是 provider 特定的；不支持 webhook 的 provider 会返回一个 unsupported 警告并继续处理，不会报错中断。

取回结果：getBatchResults

批次完成后，getBatchResults返回一个异步可迭代对象，每个 item 对应一个输入请求的终态。成功（item.status === 'succeeded'）的文本 item 包含：

• text：拼接后的文本内容，结果中没有文本部分时可能是空字符串；
• content：按顺序归一化的内容部分（text、reasoning、files、sources、tool calls、tool results 等，视 provider 支持情况）；
• finishReason和可选的rawFinishReason；
• usage与可选的response元数据；
• 可选的providerMetadata。

失败、取消、过期的 item 只包含id和终态状态，失败 item 附带error，取消和过期 item 也可能附带。单个请求失败不代表整个批次全部失败，需要逐个 item 独立处理。

两点行为限制要清楚：

• 批次取结果不会运行 AI SDK 的 tool loop，也不会调用客户端定义的execute函数；provider 定义的工具有可能在 provider 端执行。
• 结果内容和 provider 元数据应视为不可信的模型输出，不要不加筛选地打进日志，其中可能包含敏感数据。

图片请求是可选分支，仅 Google 和 xAI 支持（见上面的 provider 表）。图片请求使用与generateImage相同的prompt、n、size、aspectRatio、seed、providerOptions等设置，例如：

import { google } from '@ai-sdk/google'; import { experimental_startBatch as startBatch } from 'ai'; const batch = await startBatch({ provider: google, requests: [ { id: 'red-panda', type: 'image', model: 'gemini-2.5-flash-image', prompt: 'A red panda reading beside a cabin window', aspectRatio: '16:9', }, ], });

成功的图片 item 带images数组（GeneratedFile值）；混合结果中用type属性区分：

for await (const item of getBatchResults({ provider: google, batch })) { if (item.type === 'image' && item.status === 'succeeded') { console.log(item.images); } }

注意 provider 的 batch 端点可能只支持上述选项的子集，不支持的选项会产生警告或错误。

客户端工具同样是可选能力。批次中的客户端定义工具只有定义、没有执行：execute函数永远不会被调用，AI SDK 也不会代你提交工具结果或发起后续生成。把同一套工具传给getBatchResults，用于校验和归一化返回的 tool call：

import { anthropic } from '@ai-sdk/anthropic'; import { experimental_getBatchResults as getBatchResults, experimental_startBatch as startBatch, tool, } from 'ai'; import { z } from 'zod'; const tools = { get_weather: tool({ description: 'Get the current weather for a location.', inputSchema: z.object({ location: z.string() }), execute: async ({ location }) => { // This function is not called by batch processing. return { location, temperature: 21, condition: 'sunny' }; }, }), }; const batch = await startBatch({ provider: anthropic, requests: [ { id: 'weather-san-francisco', type: 'text', model: 'claude-haiku-4-5', prompt: 'Call get_weather for San Francisco, California.', tools, toolChoice: { type: 'tool', toolName: 'get_weather' }, }, ], }); for await (const item of getBatchResults({ provider: anthropic, batch, tools })) { if (item.status === 'succeeded') { console.log(item.id, item.content); } }

同名工具在多个请求中使用时，定义必须一致。provider 定义的工具（如 web search、code execution）在 provider 的 batch API 支持时可在 provider 端执行，其 tool call 和结果会以归一化的content部分返回。

可选操作：取消批次与列出批次

取消与列批是可选的 provider 能力。用未实现对应能力的 provider 调用这两个函数会抛出UnsupportedFunctionalityError，所以下面示例中的provider都指实现了相应能力的 batch provider。

请求取消：

const result = await cancelBatch({ provider, batch }); console.log(result.providerMetadata);

调用成功只表示 provider 接受了取消请求，不保证取消已完成，也不保证所有 pending 请求都会被取消。请求取消后仍应再用getBatchStatus查询最新状态确认。

列出批次使用游标分页，nextCursor不透明且与 provider 相关，原样存储或传递即可：

let cursor: string | undefined; do { const page = await listBatches({ provider, limit: 20, cursor, }); for (const batch of page.batches) { console.log(batch.id, batch.status); } cursor = page.nextCursor; } while (cursor != null);

每个列表项都是带最新归一化状态的可序列化批次引用，可以直接传给getBatchStatus、getBatchResults或cancelBatch。

请求控制参数与已知限制

所有五个生命周期函数都接受providerOptions、headers、timeout、abortSignal；getBatchStatus、getBatchResults、listBatches额外接受maxRetries：

• maxRetries只作用于状态查询、结果取回和列表请求，默认 2，设为 0 禁用重试；它不会重试批次创建或取消操作；
• abortSignal取消当前这次 API 请求；
• timeout限制当前 HTTP 操作。

这些参数只影响你与 provider 之间的通信，不会改变 provider 侧的处理期限，也不会取消已提交的批次。

参考文档与示例

• 批处理指南：content/docs/03-ai-sdk-core/42-batch.mdx
• API 参考：experimental_startBatch、experimental_getBatchStatus、experimental_getBatchResults、experimental_cancelBatch
• 可运行示例：Anthropic 文本批次完整流程、AI Gateway 示例、Google 取消与列批示例

【免费下载链接】aiThe AI Toolkit for TypeScript. From the creators of Next.js, the AI SDK is a free open-source library for building AI-powered applications and agents项目地址: https://gitcode.com/GitHub_Trending/ai/ai