Batch API 让你能以更低的价格和更高的速率限制异步处理大量请求。价格详情请参阅 Batch API 价格。如果实时请求需要更低延迟，请参阅Priority Processing。

什么是 Batch API？

发起标准 Grok API 调用时，你会发送请求并等待即时响应。这种方式非常适合聊天机器人、实时助手等交互式应用，以及任何用户正在等待响应的场景。

Batch API 采用不同的方式：请求不会立即处理，而是提交到队列并在后台执行。你不会立刻收到响应，而是在稍后获取结果。

与实时 API 请求的主要区别：

	实时 API	Batch API
响应时间	即时（秒级）	通常在 24 小时内*
成本	标准定价	折扣定价（查看详情）
速率限制	受每分钟限制	请求不计入速率限制
使用场景	交互式、实时	后台处理、批量任务

* 处理时间： 大多数 batch 请求会在 24 小时内完成，但处理时间可能因系统负载和 batch 大小而异。完成时间为尽力而为，不作保证。

何时使用 Batch API

如果无需立即获得结果，Batch API 可帮你降低 API 成本：

运行评测和基准测试 - 使用数千个 prompt 测试模型表现
处理大型数据集 - 分析客户反馈、分类支持工单、提取实体
大规模内容审核 - 审核积压的用户生成内容
文档摘要 - 批量处理报告、研究论文或法律文档
数据增强管道 - 为数据库记录添加 AI 生成的洞察
计划的夜间任务 - 生成日报或为仪表板准备数据

工作原理

Batch API 工作流程包含四个主要步骤：

创建 batch - batch 是将相关请求归为一组的容器
添加请求 - 将推理请求提交到 batch 队列
监控进度 - 轮询 Batch 状态以追踪完成情况
获取结果 - 获取所有已处理请求的响应

下面逐步介绍。

步骤 1：创建 Batch

batch 是请求的容器，可将其理解为归集相关工作的文件夹。你可以针对不同数据集、实验或任务类型分别创建 batch。

创建 batch 后会收到 batch_id，用于添加请求和获取结果。

curl -X POST https://api.x.ai/v1/batches \\
  -H "Content-Type: application/json" \\
  -H "Authorization: Bearer $XAI_API_KEY" \\
  -d '{
    "name": "customer_feedback_analysis"
  }'

// Create a batch with a descriptive name
const response = await fetch("https://api.x.ai/v1/batches", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: \`Bearer \${process.env.XAI_API_KEY}\`,
  },
  body: JSON.stringify({ name: "customer_feedback_analysis" }),
});
const batch = await response.json();
console.log(\`Created batch: \${batch.batch_id}\`);

// Store the batch_id for later use
const batchId = batch.batch_id;

from xai_sdk import Client

client = Client()

# Create a batch with a descriptive name
batch = client.batch.create(batch_name="customer_feedback_analysis")
print(f"Created batch: {batch.batch_id}")

# Store the batch_id for later use
batch_id = batch.batch_id

步骤 2：向 batch 添加请求

创建 batch 后即可向其中添加请求。每个请求都会异步处理。

使用 xAI SDK 时，先根据各个请求的类型调用相应方法构建请求，再将它们组成列表传入：

请求类型	SDK 方法
文本	`chat.create()`
图像	`image.prepare()`
视频	`video.prepare()`
视频扩展	`video.prepare_extension()`

你也可以上传 JSONL 文件。

重要： 为每个请求分配唯一的 batch_request_id。

使用此 ID 可以将结果与原始请求对应起来，在处理数百或数千条数据时尤为重要。
如果未提供 ID，系统会为你生成 UUID。
自定义 ID 有助于实现幂等性，确保请求只被处理一次。
你还可以用它将批处理请求关联到自己系统中的记录。

curl -X POST https://api.x.ai/v1/batches/{batch_id}/requests \\
  -H "Content-Type: application/json" \\
  -H "Authorization: Bearer $XAI_API_KEY" \\
  -d '{
    "batch_requests": [
      {
        "batch_request_id": "feedback_001",
        "batch_request": {
          "responses": {
            "input": [
              {"role": "system", "content": "Classify the sentiment as positive, negative, or neutral."},
              {"role": "user", "content": "The product exceeded my expectations!"}
            ],
            "model": "grok-4.3"
          }
        }
      },
      {
        "batch_request_id": "feedback_002",
        "batch_request": {
          "responses": {
            "input": [
              {"role": "system", "content": "Classify the sentiment as positive, negative, or neutral."},
              {"role": "user", "content": "Shipping took way too long."}
            ],
            "model": "grok-4.3"
          }
        }
      }
    ]
  }'

const batchRequests = [];

// Chat completion with tools (uses "responses" endpoint for server-side tool support)
batchRequests.push({
  batch_request_id: "chat_001",
  batch_request: {
    responses: {
      model: "grok-4.3",
      tools: [{ type: "web_search" }, { type: "x_search" }],
      input: [
        { role: "system", content: "Analyze market sentiment from recent news and posts." },
        { role: "user", content: "What is the current sentiment around TSLA stock?" },
      ],
    },
  },
});

// Image generation
batchRequests.push({
  batch_request_id: "img_001",
  batch_request: {
    image_generation: {
      prompt: "A sleek modern laptop on a minimalist desk",
      model: "grok-imagine-image-2.0",
    },
  },
});

// Image edit
batchRequests.push({
  batch_request_id: "img_edit_001",
  batch_request: {
    image_edit: {
      prompt: "Add a rainbow in the background",
      model: "grok-imagine-image-2.0",
      image: { url: "https://picsum.photos/800", type: "image_url" },
    },
  },
});

// Video generation
batchRequests.push({
  batch_request_id: "vid_001",
  batch_request: {
    video_generation: {
      prompt: "A product rotating on a turntable with dramatic lighting",
      model: "grok-imagine-video-1.5",
    },
  },
});

// Video edit
batchRequests.push({
  batch_request_id: "vid_edit_001",
  batch_request: {
    video_generation: {
      prompt: "Make it slow motion",
      model: "grok-imagine-video",
      video: { url: "https://lorem.video/cat_360p_3s" },
    },
  },
});

// Video extension
batchRequests.push({
  batch_request_id: "vid_ext_001",
  batch_request: {
    video_extension: {
      prompt: "The camera slowly pans to reveal a sunset behind the mountains",
      model: "grok-imagine-video",
      video: { url: "https://lorem.video/cat_360p_3s" },
      duration: 6,
    },
  },
});

// Remote MCP
batchRequests.push({
  batch_request_id: "mcp_001",
  batch_request: {
    responses: {
      model: "grok-4.3",
      tools: [{ type: "mcp", server_label: "deepwiki", server_url: "https://mcp.deepwiki.com/mcp" }],
      input: [{ role: "user", content: "What does the xai-sdk-python repo do?" }],
    },
  },
});

// Add all requests to the batch
const response = await fetch(\`https://api.x.ai/v1/batches/\${batchId}/requests\`, {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: \`Bearer \${process.env.XAI_API_KEY}\`,
  },
  body: JSON.stringify({ batch_requests: batchRequests }),
});
if (!response.ok) throw new Error(\`Failed to add requests: \${await response.text()}\`);
console.log(\`Added \${batchRequests.length} requests to batch\`);

from xai_sdk import Client
from xai_sdk.chat import system, user
from xai_sdk.tools import web_search, x_search, mcp

client = Client()

batch_requests = []

# Chat completion with tools
chat = client.chat.create(
    model="grok-4.3",
    batch_request_id="chat_001",
    tools=[web_search(), x_search()],
)
chat.append(system("Analyze market sentiment from recent news and posts."))
chat.append(user("What is the current sentiment around TSLA stock?"))
batch_requests.append(chat)

# Image generation
image_req = client.image.prepare(
    prompt="A sleek modern laptop on a minimalist desk",
    model="grok-imagine-image-2.0",
    batch_request_id="img_001",
)
batch_requests.append(image_req)

# Image edit
image_edit_req = client.image.prepare(
    prompt="Add a rainbow in the background",
    model="grok-imagine-image-2.0",
    image_url="https://picsum.photos/800",
    batch_request_id="img_edit_001",
)
batch_requests.append(image_edit_req)

# Video generation
video_req = client.video.prepare(
    prompt="A product rotating on a turntable with dramatic lighting",
    model="grok-imagine-video-1.5",
    batch_request_id="vid_001",
)
batch_requests.append(video_req)

# Video edit
video_edit_req = client.video.prepare(
    prompt="Make it slow motion",
    model="grok-imagine-video",
    video_url="https://lorem.video/cat_360p_3s",
    batch_request_id="vid_edit_001",
)
batch_requests.append(video_edit_req)

# Video extension
video_ext_req = client.video.prepare_extension(
    prompt="The camera slowly pans to reveal a sunset behind the mountains",
    model="grok-imagine-video",
    video_url="https://lorem.video/cat_360p_3s",
    duration=6,
    batch_request_id="vid_ext_001",
)
batch_requests.append(video_ext_req)

# Remote MCP
mcp_chat = client.chat.create(
    model="grok-4.3",
    batch_request_id="mcp_001",
    tools=[mcp(server_url="https://mcp.deepwiki.com/mcp")],
)
mcp_chat.append(user("What does the xai-sdk-python repo do?"))
batch_requests.append(mcp_chat)

# Add all requests to the batch
client.batch.add(batch_id=batch.batch_id, batch_requests=batch_requests)
print(f"Added {len(batch_requests)} requests to batch")

步骤 3：监控 batch 进度

添加请求后，它们会开始在后台处理。由于 batch 处理是异步的，需要轮询 batch 状态才能知道结果何时就绪。

batch 状态包含待处理、成功和失败请求的计数器。请定期轮询，直到 num_pending 变为 0，表示所有请求均已处理完成（成功或出错）。

# Check batch status
curl https://api.x.ai/v1/batches/{batch_id} \\
  -H "Authorization: Bearer $XAI_API_KEY"

# Response includes state with request counts:
# {
#   "state": {
#     "num_requests": 100,
#     "num_pending": 25,
#     "num_success": 70,
#     "num_error": 5
#   }
# }

// Poll until all requests are processed
console.log("Waiting for batch to complete...");
const interval = setInterval(async () => {
  const response = await fetch(
    \`https://api.x.ai/v1/batches/\${batchId}\`,
    { headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` } }
  );
  const batch = await response.json();

  const { num_pending, num_success, num_error, num_requests } = batch.state;
  const completed = num_success + num_error;
  console.log(\`Progress: \${completed}/\${num_requests} complete, \${num_pending} pending\`);

  if (num_requests > 0 && num_pending === 0) {
    clearInterval(interval);
    console.log("Batch processing complete!");
  }
  // Wait before polling again (avoid hammering the API)
}, 5000);

import time
from xai_sdk import Client

client = Client()

# Poll until all requests are processed
print("Waiting for batch to complete...")
while True:
    batch = client.batch.get(batch_id=batch.batch_id)
    
    pending = batch.state.num_pending
    completed = batch.state.num_success + batch.state.num_error
    total = batch.state.num_requests
    
    print(f"Progress: {completed}/{total} complete, {pending} pending")
    
    if pending == 0:
        print("Batch processing complete!")
        break
    
    # Wait before polling again (avoid hammering the API)
    time.sleep(5)

理解 batch 状态

Batch API 在两个层级追踪状态：batch 层级和单个请求层级。

batch 层级状态汇总了批次中所有请求的进度，记录在 batch.state 对象中。要获取该对象，请调用 client.batch.get() 方法：

计数器	说明
`num_requests`	添加到 Batch 的请求总数
`num_pending`	等待处理的请求
`num_success`	成功完成的请求
`num_error`	因错误失败的请求
`num_cancelled`	已取消的请求

当 num_pending 变为 0 时，所有请求都已处理完毕（成功、出错或取消）。

单个请求状态说明各请求所处的生命周期阶段，记录在 batch_request_metadata 对象中。要获取该对象，请调用 client.batch.list_batch_requests() 方法：

状态	说明
`pending`	请求已加入队列，等待处理
`succeeded`	请求成功完成，结果可用
`failed`	请求在处理过程中遇到错误
`cancelled`	请求已取消（例如 Batch 在该请求处理前被取消）

batch 生命周期： batch 也可以被取消或过期。如果取消 batch，待处理请求将不再处理，但已完成的结果仍然可用。batch 设有过期时间，过期后结果将无法访问；获取 batch 详情时请检查 expires_at 字段。

步骤 4：获取结果

可以随时获取结果，即使整个 batch 尚未完成。单个请求完成处理后结果即可使用，因此可以在其他请求仍在处理时开始使用已完成的结果。

每个结果都通过先前分配的 batch_request_id 关联到原始请求。请根据请求类型，从相应的属性中读取响应：

请求类型	属性	字段
聊天补全	`result.response`	`.content`、`.usage`、`.finish_reason`等
图像	`result.image_response`	`.url`、`.base64`、`.usage`、`.model`
视频	`result.video_response`	`.url`、`.duration`、`.usage`、`.model`

这些响应类型与常规的 client.image.sample() 和 client.video.generate() 方法返回的类型相同。

SDK 提供方便的 .succeeded 和 .failed 属性，用于区分成功响应和错误。

分页： 结果按页返回。使用 limit 参数控制页面大小，并使用 pagination_token 获取后续页面。当 pagination_token 为 None 时，表示已经到达末尾。

# Fetch first page
curl "https://api.x.ai/v1/batches/{batch_id}/results?limit=100" \\
  -H "Authorization: Bearer $XAI_API_KEY"

# Use pagination_token from response to fetch next page
curl "https://api.x.ai/v1/batches/{batch_id}/results?limit=100&pagination_token={token}" \\
  -H "Authorization: Bearer $XAI_API_KEY"

// Paginate through all results
const allSucceeded = [];
const allFailed = [];
let paginationToken = undefined;

while (true) {
  // Fetch a page of results (limit controls page size)
  const url = new URL(\`https://api.x.ai/v1/batches/\${batchId}/results\`);
  url.searchParams.set("limit", "100");
  if (paginationToken) url.searchParams.set("pagination_token", paginationToken);

  const res = await fetch(url, {
    headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` },
  });
  const page = await res.json();

  // Collect results from this page
  for (const result of page.results) {
    const response = result.batch_result?.response;
    if (response?.chat_get_completion || response?.image_generation || response?.video_generation) {
      allSucceeded.push(result);
    } else {
      allFailed.push(result);
    }
  }

  // Check if there are more pages
  if (!page.pagination_token) break;
  paginationToken = page.pagination_token;
}

// Process all results
console.log(\`Successfully processed: \${allSucceeded.length} requests\`);
for (const result of allSucceeded) {
  const response = result.batch_result.response;
  const content = response.chat_get_completion?.choices[0].message.content
    ?? response.image_generation?.data[0].url
    ?? response.video_generation?.video.url;
  const tokens = response.chat_get_completion?.usage?.total_tokens;
  // Access the full response object
  console.log(\`[\${result.batch_request_id}] \${content}\`);
  if (tokens != null) console.log(\`  Tokens used: \${tokens}\`);
}

if (allFailed.length > 0) {
  console.log(\`\\nFailed: \${allFailed.length} requests\`);
  for (const result of allFailed) {
    console.log(\`[\${result.batch_request_id}] Error: \${result.error_message}\`);
  }
}

from xai_sdk import Client

client = Client()

# Paginate through all results
all_succeeded = []
all_failed = []
pagination_token = None

while True:
    # Fetch a page of results (limit controls page size)
    page = client.batch.list_batch_results(
        batch_id=batch.batch_id,
        limit=100,
        pagination_token=pagination_token,
    )
    
    # Collect results from this page
    all_succeeded.extend(page.succeeded)
    all_failed.extend(page.failed)
    
    # Check if there are more pages
    if page.pagination_token is None:
        break
    pagination_token = page.pagination_token

# Process results - handle different response types
print(f"Successfully processed: {len(all_succeeded)} requests")
for result in all_succeeded:
    rid = result.batch_request_id
    resp = result.proto.response

    if resp.HasField("completion_response"):
        # Chat completion response
        print(f"[{rid}] {result.response.content}")
        print(f"  Tokens used: {result.response.usage.total_tokens}")
    elif resp.HasField("image_response"):
        # Image generation response
        print(f"[{rid}] Image URL: {result.image_response.url}")
    elif resp.HasField("video_response"):
        # Video generation response
        print(f"[{rid}] Video URL: {result.video_response.url}")

if all_failed:
    print(f"\\nFailed: {len(all_failed)} requests")
    for result in all_failed:
        print(f"[{result.batch_request_id}] Error: {result.error_message}")

其他操作

除核心 workflow 外，Batch API 还提供管理 batch 的其他操作。

取消 Batch

可以在所有请求完成前取消 batch。已处理请求的结果仍然可用，但待处理请求不会继续处理。无法向已取消的 batch 添加更多请求。

curl -X POST https://api.x.ai/v1/batches/{batch_id}:cancel \\
  -H "Authorization: Bearer $XAI_API_KEY"

// Cancel processing
const response = await fetch(
  \`https://api.x.ai/v1/batches/\${batchId}:cancel\`,
  { method: "POST", headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` } }
);
const cancelledBatch = await response.json();
console.log(\`Cancelled batch: \${cancelledBatch.batch_id}\`);
console.log(\`Completed before cancellation: \${cancelledBatch.state.num_success} requests\`);

from xai_sdk import Client

client = Client()

# Cancel processing
cancelled_batch = client.batch.cancel(batch_id=batch.batch_id)
print(f"Cancelled batch: {cancelled_batch.batch_id}")
print(f"Completed before cancellation: {cancelled_batch.state.num_success} requests")

列出所有 Batch

查看团队的所有 batch。batch 会保留到过期为止（请检查 expires_at 字段）。该 endpoint 支持相同的 limit 和 pagination_token 参数，可对大型列表分页。

curl "https://api.x.ai/v1/batches?limit=20" \\
  -H "Authorization: Bearer $XAI_API_KEY"

// List recent batches
const response = await fetch(
  "https://api.x.ai/v1/batches?limit=20",
  { headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` } }
);
const data = await response.json();

for (const batch of data.batches) {
  const status = batch.state.num_pending === 0 ? "complete" : "processing";
  console.log(\`\${batch.name} (\${batch.batch_id}): \${status}\`);
}

from xai_sdk import Client

client = Client()

# List recent batches
response = client.batch.list(limit=20)

for batch in response.batches:
    status = "complete" if batch.state.num_pending == 0 else "processing"
    print(f"{batch.name} ({batch.batch_id}): {status}")

检查单个请求状态

如需详细追踪，可以检查 batch 中每个请求的元数据，其中显示单个请求的状态、计时和其他详情。该 endpoint 支持相同的 limit 和 pagination_token 参数，可对大型 batch 分页。

curl "https://api.x.ai/v1/batches/{batch_id}/requests?limit=50" \\
  -H "Authorization: Bearer $XAI_API_KEY"

// Get metadata for individual requests
const response = await fetch(
  \`https://api.x.ai/v1/batches/\${batchId}/requests?limit=50\`,
  { headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` } }
);
const data = await response.json();

for (const req of data.batch_request_metadata) {
  console.log(\`Request \${req.batch_request_id}: \${req.state}\`);
}

from xai_sdk import Client

client = Client()

# Get metadata for individual requests
metadata = client.batch.list_batch_requests(batch_id=batch.batch_id)

for request in metadata.batch_request_metadata:
    print(f"Request {request.batch_request_id}: {request.state}")

追踪成本

每个 batch 都会追踪总处理成本。处理完成后可查看成本明细以了解支出。价格详情请参阅价格页面中的 Batch API 价格。

# Get batch with cost information
curl -s "https://api.x.ai/v1/batches/{batch_id}/results?limit=100" \\
  -H "Authorization: Bearer $XAI_API_KEY"

# Cost per result can be found on response.results[].batch_result.response.chat_get_completion.usage.cost_in_usd_ticks
# Cost is returned in ticks (1e-10 USD) for precision

// Get batch with cost information
const response = await fetch(
  \`https://api.x.ai/v1/batches/\${batchId}/results?limit=100\`,
  { headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` } }
);
const data = await response.json();

// Cost is returned in ticks (1e-10 USD) for precision
let totalTicks = 0;
for (const r of data.results) {
  totalTicks += r.batch_result?.response?.chat_get_completion?.usage?.cost_in_usd_ticks ?? 0;
}
console.log(\`Total cost: $\${(totalTicks / 1e10).toFixed(4)}\`);

from xai_sdk import Client

client = Client()

# Get batch with cost information
batch = client.batch.get(batch_id=batch.batch_id)

# Cost is returned in ticks (1e-10 USD) for precision
total_cost_usd = batch.cost_breakdown.total_cost_usd_ticks / 1e10
print("Total cost: $%.4f" % total_cost_usd)

完整示例

此端到端示例演示了真实的 batch workflow：大规模分析客户反馈。它会创建 batch、提交反馈项目进行情感分析、等待处理并输出结果。为简化示例，这里不对结果分页；处理更大的 batch 时，请参阅步骤 4了解分页。

const BASE_URL = "https://api.x.ai/v1";
const headers = { "Content-Type": "application/json", Authorization: \`Bearer \${process.env.XAI_API_KEY}\` };

// Sample dataset: customer feedback to analyze
const feedbackData = [
  { id: "fb_001", text: "Absolutely love this product! Best purchase ever." },
  { id: "fb_002", text: "Delivery was late and the packaging was damaged." },
  { id: "fb_003", text: "Works fine, nothing special to report." },
  { id: "fb_004", text: "Customer support was incredibly helpful!" },
  { id: "fb_005", text: "The app keeps crashing on my phone." },
];

// Step 1: Create a batch
console.log("Creating batch...");
const batchRes = await fetch(\`\${BASE_URL}/batches\`, {
  method: "POST",
  headers,
  body: JSON.stringify({ name: "feedback_sentiment_analysis" }),
});
const batch = await batchRes.json();
const batchId = batch.batch_id;
console.log(\`Batch created: \${batchId}\`);

// Step 2: Build and add requests
console.log("\\nAdding requests...");
const response = await fetch(\`\${BASE_URL}/batches/\${batchId}/requests\`, {
  method: "POST",
  headers,
  body: JSON.stringify({
    batch_requests: feedbackData.map((item) => ({
      batch_request_id: item.id,
      batch_request: {
        chat_get_completion: {
          model: "grok-4.3",
          messages: [
            {
              role: "system",
              content: "Analyze the sentiment of the customer feedback. Respond with exactly one word: positive, negative, or neutral.",
            },
            { role: "user", content: item.text },
          ],
        },
      },
    })),
  }),
});
if (!response.ok) throw new Error(\`Failed to add requests: \${await response.text()}\`);
console.log(\`Added \${feedbackData.length} requests\`);

// Step 3: Wait for completion
console.log("\\nProcessing...");
const interval = setInterval(async () => {
  const statusRes = await fetch(\`\${BASE_URL}/batches/\${batchId}\`, { headers });
  const status = await statusRes.json();
  const { num_pending, num_success, num_error, num_requests } = status.state;
  console.log(\`  \${num_success + num_error}/\${num_requests} complete\`);

  if (num_requests > 0 && num_pending === 0) {
    clearInterval(interval);

    // Step 4: Retrieve and display results
    console.log("\\n--- Results ---");
    const resultsRes = await fetch(\`\${BASE_URL}/batches/\${batchId}/results?limit=100\`, { headers });
    const { results } = await resultsRes.json();

    // Create a lookup for original feedback text
    const feedbackLookup = Object.fromEntries(feedbackData.map((item) => [item.id, item.text]));

    const succeeded = results.filter((r) => r.batch_result?.response?.chat_get_completion);
    const failed = results.filter((r) => !r.batch_result?.response?.chat_get_completion);

    for (const result of succeeded) {
      const originalText = feedbackLookup[result.batch_request_id] ?? "";
      const sentiment = result.batch_result.response.chat_get_completion.choices[0].message.content.trim().toLowerCase();
      console.log(\`[\${sentiment.toUpperCase()}] \${originalText.slice(0, 50)}...\`);
    }

    // Report any failures
    if (failed.length > 0) {
      console.log("\\n--- Errors ---");
      for (const result of failed) {
        console.log(\`[\${result.batch_request_id}] \${result.error_message}\`);
      }
    }

    // Display cost
    let totalTicks = 0;
    for (const r of results) {
      totalTicks += r.batch_result?.response?.chat_get_completion?.usage?.cost_in_usd_ticks ?? 0;
    }
    console.log(\`\\nTotal cost: $\${(totalTicks / 1e10).toFixed(4)}\`);
  }
}, 2000);

import time
from xai_sdk import Client
from xai_sdk.chat import system, user

client = Client()

# Sample dataset: customer feedback to analyze
feedback_data = [
    {"id": "fb_001", "text": "Absolutely love this product! Best purchase ever."},
    {"id": "fb_002", "text": "Delivery was late and the packaging was damaged."},
    {"id": "fb_003", "text": "Works fine, nothing special to report."},
    {"id": "fb_004", "text": "Customer support was incredibly helpful!"},
    {"id": "fb_005", "text": "The app keeps crashing on my phone."},
]

# Step 1: Create a batch
print("Creating batch...")
batch = client.batch.create(batch_name="feedback_sentiment_analysis")
print(f"Batch created: {batch.batch_id}")

# Step 2: Build and add requests
print("\\nAdding requests...")
batch_requests = []
for item in feedback_data:
    chat = client.chat.create(
        model="grok-4.3",
        batch_request_id=item["id"],
    )
    chat.append(system(
        "Analyze the sentiment of the customer feedback. "
        "Respond with exactly one word: positive, negative, or neutral."
    ))
    chat.append(user(item["text"]))
    batch_requests.append(chat)

client.batch.add(batch_id=batch.batch_id, batch_requests=batch_requests)
print(f"Added {len(batch_requests)} requests")

# Step 3: Wait for completion
print("\\nProcessing...")
while True:
    batch = client.batch.get(batch_id=batch.batch_id)
    pending = batch.state.num_pending
    completed = batch.state.num_success + batch.state.num_error
    
    print(f"  {completed}/{batch.state.num_requests} complete")
    
    if pending == 0:
        break
    time.sleep(2)

# Step 4: Retrieve and display results
print("\\n--- Results ---")
results = client.batch.list_batch_results(batch_id=batch.batch_id)

# Create a lookup for original feedback text
feedback_lookup = {item["id"]: item["text"] for item in feedback_data}

for result in results.succeeded:
    original_text = feedback_lookup.get(result.batch_request_id, "")
    sentiment = result.response.content.strip().lower()
    print(f"[{sentiment.upper()}] {original_text[:50]}...")

# Report any failures
if results.failed:
    print("\\n--- Errors ---")
    for result in results.failed:
        print(f"[{result.batch_request_id}] {result.error_message}")

# Display cost
cost_usd = batch.cost_breakdown.total_cost_usd_ticks / 1e10
print("\\nTotal cost: $%.4f" % cost_usd)

上传 JSONL 文件

除了通过 SDK 添加请求，还可以上传 JSONL 文件来创建 batch。通过脚本、管道或外部工具生成请求时，这种方式很实用。

文件的每一行都是一个 JSON 对象，包含四个字段：

字段	说明
`custom_id`	唯一标识符，对应 `batch_request_id`。
`method`	始终为 `"POST"`。
`url`	API 端点路径。
`body`	JSON 请求体，格式遵循REST API 参考中对应端点的定义。

{"custom_id": "chat-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "grok-4.3", "messages": [{"role": "user", "content": "Classify this as positive, negative, or neutral: The product exceeded my expectations!"}]}}
{"custom_id": "search-1", "method": "POST", "url": "/v1/responses", "body": {"model": "grok-4.3", "tools": [{"type": "web_search"}, {"type": "x_search"}], "input": [{"role": "user", "content": "What are the latest SpaceX launches?"}]}}
{"custom_id": "mcp-1", "method": "POST", "url": "/v1/responses", "body": {"model": "grok-4.3", "tools": [{"type": "mcp", "server_label": "deepwiki", "server_url": "https://mcp.deepwiki.com/mcp"}], "input": [{"role": "user", "content": "What does the xai-sdk-python repo do?"}]}}
{"custom_id": "img-1", "method": "POST", "url": "/v1/images/generations", "body": {"model": "grok-imagine-image-2.0", "prompt": "A futuristic city skyline at sunset"}}
{"custom_id": "img-edit-1", "method": "POST", "url": "/v1/images/edits", "body": {"model": "grok-imagine-image-2.0", "prompt": "Add a rainbow", "image": {"url": "https://picsum.photos/800"}}}
{"custom_id": "vid-1", "method": "POST", "url": "/v1/videos/generations", "body": {"model": "grok-imagine-video-1.5", "prompt": "A rocket launching from Mars", "duration": 8}}
{"custom_id": "vid-edit-1", "method": "POST", "url": "/v1/videos/edits", "body": {"model": "grok-imagine-video", "prompt": "Make it slow motion", "video": {"url": "https://lorem.video/cat_360p_3s"}}}
{"custom_id": "vid-ext-1", "method": "POST", "url": "/v1/videos/extensions", "body": {"model": "grok-imagine-video", "prompt": "The camera slowly pans to reveal a sunset", "video": {"url": "https://lorem.video/cat_360p_3s"}, "duration": 6}}

同一文件中可以混用不同 endpoint。每个请求都会独立路由。

支持的 url 值：

URL	说明
`/v1/chat/completions`	聊天补全
`/v1/responses`	模型响应
`/v1/images/generations`	图像生成
`/v1/images/edits`	图像编辑
`/v1/videos/generations` 或 `/v1/videos`	视频生成
`/v1/videos/edits`	视频编辑
`/v1/videos/extensions`	视频扩展

仅接受启用了 batch 的模型。最新信息请参阅对应的模型页面。不支持批处理的模型会被拒绝，错误信息为 not supported for batch processing。

通过 Files API上传文件，然后创建引用该文件的 batch：

# Upload the JSONL file
curl -X POST https://api.x.ai/v1/files \\
  -H "Authorization: Bearer $XAI_API_KEY" \\
  -F file="@batch_requests.jsonl"

# Create a batch with the file ID
curl -X POST https://api.x.ai/v1/batches \\
  -H "Content-Type: application/json" \\
  -H "Authorization: Bearer $XAI_API_KEY" \\
  -d '{
    "name": "sentiment_analysis",
    "input_file_id": "file-abc123"
  }'

import fs from "fs";

// Upload the JSONL file
const jsonlContent = fs.readFileSync("batch_requests.jsonl", "utf8");
const formData = new FormData();
formData.append("file", new Blob([jsonlContent], { type: "application/jsonl" }), "batch_requests.jsonl");

const uploadRes = await fetch("https://api.x.ai/v1/files", {
  method: "POST",
  headers: { Authorization: \`Bearer \${process.env.XAI_API_KEY}\` },
  body: formData,
});
const file = await uploadRes.json();

// Create a batch with the file ID
const batchRes = await fetch("https://api.x.ai/v1/batches", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    Authorization: \`Bearer \${process.env.XAI_API_KEY}\`,
  },
  body: JSON.stringify({ name: "sentiment_analysis", input_file_id: file.id }),
});
const batch = await batchRes.json();
console.log(\`Created batch: \${batch.batch_id}\`);

from xai_sdk import Client

client = Client()

# Upload the JSONL file
file = client.files.upload(
    file=open("batch_requests.jsonl", "rb"),
)

# Create a batch with the file ID
batch = client.batch.create(
    batch_name="sentiment_analysis",
    input_file_id=file.id,
)
print(f"Created batch: {batch.batch_id}")

文件会在后台异步处理。若任意一行无效，batch 将被取消并返回错误消息。监控进度和获取结果的方式与内联 batch 相同。

使用文件创建批次时有以下限制：

批次创建后即封存，无法再通过以下方法添加请求：AddBatchRequests。
文件大小上限为 200 MB。
每个文件最多容纳 50,000 个请求。
文件中的每个custom_id 必须唯一。

限制

Batch

一个团队可以拥有 不限数量 个 batch。
batch 创建速率上限：每个团队每秒创建 2 个 batch。

Batch Request

理论上，一个 batch 可以包含 不限数量 个请求，但超大型 batch（超过 100,000 个请求）可能会为保障处理稳定性而受到限流。
可添加到 batch 的每个单独请求最大负载大小为 25 MB。
一个团队最多可以发送 1000 次 AddBatchRequests API 调用；限额周期为 30 秒（这是团队中所有 batch 共享的滚动限制）。
图像和视频结果包含签名 URL，该 URL 会在 1 小时后过期。获取结果后请及时下载媒体。

Tool 使用

batch 请求同时支持服务端 tool和客户端 function tool。

服务端 tool（web search、code execution、MCP 等）的工作方式与实时 API 相同：在处理过程中执行，并返回最终响应。
客户端 function tool也受支持：模型会在响应中返回 tool_calls，供你离线处理。多轮 tool calling 需要提交新的 batch 请求，并在对话中包含 tool 结果消息。

最后更新：2026 年 10 月 20 日

从写代码，到创作下一幕

Agent Plan & Coding Plan

Seedance 2.5

高级 API 用法

Batch API

什么是 Batch API？

何时使用 Batch API

工作原理

步骤 1：创建 Batch

步骤 2：向 batch 添加请求

步骤 3：监控 batch 进度

理解 batch 状态

步骤 4：获取结果

其他操作

取消 Batch

列出所有 Batch

检查单个请求状态

追踪成本

完整示例

上传 JSONL 文件

限制

Tool 使用