高级 API 使用

批量 API

批量 API 让您能够异步处理大量请求，提供更优惠的价格和更高的速率限制。有关定价详情，请参阅批量 API 定价。如果您需要实时请求的更低延迟，请参阅优先处理。

WARNING

模型支持

grok-4.5 目前不支持批量 API 请求，将被拒绝。

什么是批量 API？

当您向 Grok 发起标准 API 调用时，您发送一个请求并等待即时响应。这种方法非常适合聊天机器人、实时助手或任何用户等待响应的交互式应用程序。

批量 API 采用了不同的方法。它不会立即处理请求，而是将请求提交到队列中在后台处理。您不会获得即时响应——相反，您需要稍后返回以检索结果。

与实时 API 请求的关键区别：

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

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

NOTE

您还可以通过 xAI 控制台创建、监控和管理批量。控制台提供了跟踪批量进度和查看结果的可视化界面。

何时使用批量 API

当您不需要即时结果并希望降低 API 成本时，批量 API 是理想选择：

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

工作原理

批量 API 工作流包含四个主要步骤：

创建批量 - 批量是一个将相关请求组合在一起的容器
添加请求 - 将您的推理请求提交到批量队列
监控进度 - 轮询批量状态以跟踪完成情况
检索结果 - 获取所有已处理请求的响应

让我们逐步介绍每个步骤。

第 1 步：创建批量

批量作为您请求的容器。您可以将其视为一个将相关工作组合在一起的文件夹——您可能为不同的数据集、实验或作业类型创建单独的批量。

创建批量时，您会收到一个 batch_id，您将使用它来添加请求和检索结果。

bash

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"
  }'

python

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

javascript

// 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;

第 2 步：向批量添加请求

创建批量后，您现在可以向其中添加请求。每个请求将被异步处理。

使用 xAI SDK，添加批量请求很简单： 对于文本使用 chat.create()，对于图像使用 image.prepare()，对于视频使用 video.prepare()，对于视频扩展使用 video.prepare_extension()，然后将它们作为列表传递。如果您愿意，也可以上传 JSONL 文件。

重要提示： 为每个请求分配唯一的 batch_request_id。此 ID 可让您将结果匹配回其原始请求，当您处理数百或数千个项目时，这变得很重要。如果您不提供 ID，我们会为您生成一个 UUID。使用自己的 ID 对于幂等性（确保请求只处理一次）以及将批量请求链接到您自己系统中的记录很有用。

python

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-quality",
    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-quality",
    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",
    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")

bash

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"
          }
        }
      }
    ]
  }'

javascript

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-quality",
    },
  },
});

// 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-quality",
      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",
    },
  },
});

// 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\`);

第 3 步：监控批量进度

添加请求后，它们开始在后台处理。由于批量处理是异步的，您需要轮询批量状态以了解结果何时准备就绪。

批量状态包含待处理、成功和失败请求的计数器。定期轮询直到 num_pending 达到零，这表示所有请求都已处理（成功或出错）。

bash

# 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
#   }
# }

python

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)

javascript

// 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);

了解批量状态

批量 API 在两个级别跟踪状态：批量级别和单个请求级别。

批量级别状态显示给定批量中所有请求的总体进度，可通过 client.batch.get() 方法返回的 batch.state 对象访问：

计数器	描述
`num_requests`	添加到批量的请求数量
`num_pending`	等待处理的请求
`num_success`	成功完成的请求
`num_error`	出错失败的请求
`num_cancelled`	已取消的请求

当 num_pending 达到零时，所有请求都已处理（成功、出错或已取消）。

单个请求状态描述每个请求在其生命周期中的位置，可通过 client.batch.list_batch_requests() 方法返回的 batch_request_metadata 对象访问：

状态	描述
`pending`	请求已排队并等待处理
`succeeded`	请求成功完成，结果可用
`failed`	请求在处理过程中遇到错误
`cancelled`	请求已取消（例如，在处理此请求前批量已取消）

批量生命周期： 批量也可以被取消或过期。如果您取消批量，待处理的请求将不会被处理，但已完成的结果仍然可用。批量有过期时间，之后结果将不再可访问——检索批量详细信息时请检查 expires_at 字段。

第 4 步：检索结果

您可以随时检索结果，即使整个批量尚未完成。结果在单个请求处理完成后立即可用，因此您可以在其他请求仍在进行时开始使用已完成的结果。

每个结果通过您之前分配的 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 时，您已到达末尾。

python

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}")

bash

# 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"

javascript

// 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}\`);
  }
}

其他操作

除了核心工作流外，批量 API 还提供了其他操作来管理您的批量。

取消批量

您可以在所有请求完成前取消批量。已处理的请求在结果中仍然可用，但待处理的请求将不会被处理。您无法向已取消的批量添加更多请求。

bash

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

python

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")

javascript

// 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\`);

列出所有批量

查看属于您团队的所有批量。批量将一直保留到过期（请检查 expires_at 字段）。此端点支持相同的 limit 和 pagination_token 参数，用于分页浏览大型列表。

bash

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

python

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}")

javascript

// 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}\`);
}

检查单个请求状态

为了进行详细跟踪，您可以检查批量中每个请求的元数据。这显示了单个请求的状态、时间和其他详细信息。此端点支持相同的 limit 和 pagination_token 参数，用于分页浏览大型批量。

bash

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

python

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}")

javascript

// 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}\`);
}

跟踪成本

每个批量跟踪总处理成本。处理完成后访问成本明细以了解您的支出。有关定价详情，请参阅定价页面上的批量 API 定价。

bash

# 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

python

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)

javascript

// 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)}\`);

完整示例

这个端到端示例演示了一个真实的批量工作流：大规模分析客户反馈。它创建一个批量，提交反馈项目进行情感分析，等待处理，然后输出结果。为简单起见，此示例不分页结果——处理较大批量时，请参阅第 4 步了解分页。

python

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)

javascript

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);

JSONL 文件上传

作为通过 SDK 添加请求的替代方案，您可以通过上传 JSONL 文件来创建批量。当从脚本、管道或外部工具生成请求时，这很有用。

文件中的每一行都是一个包含四个字段的 JSON 对象：custom_id（唯一标识符，映射到 batch_request_id）、method（始终为 "POST"）、url（API 端点路径）和 body（与该端点对应的 REST API 参考匹配的 JSON 请求负载）。

json

{"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-quality", "prompt": "A futuristic city skyline at sunset"}}
{"custom_id": "img-edit-1", "method": "POST", "url": "/v1/images/edits", "body": {"model": "grok-imagine-image-quality", "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", "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}}

您可以在同一文件中混合不同的端点。每个请求独立路由。

支持的 url 值：

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

通过文件 API 上传文件，然后创建一个引用它的批量：

python

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}")

bash

# 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"
  }'

javascript

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}\`);

文件在后台异步处理。如果任何行无效，批量将因错误消息而取消。以与内联批量相同的方式监控进度并检索结果。

基于文件的批量在创建后即被密封——您无法通过 AddBatchRequests 添加更多请求。最大文件大小为 200 MB，最多 50,000 个请求。每个 custom_id 在文件内必须唯一。

限制

批量

一个团队可以有无限数量的批量。
最大批量创建速率：每秒每团队 2 个批量创建。

批量请求

批量理论上可以包含无限数量的请求，但极大的批量（>1,000,000 个请求）可能为处理稳定性而受到限制。
可以添加到批量的每个单独请求的最大负载大小为 25MB。
一个团队每 30 秒最多可以发送 1000 次 add-batch-requests API 调用（这是团队内所有批量共享的滚动限制）。
图像和视频结果包含在 1 小时后过期的签名 URL。检索结果后请及时下载媒体。

工具使用

服务器端工具和客户端函数工具在批量请求中都受支持。

服务器端工具（网络搜索、代码执行、MCP 等）与实时 API 中的工作方式相同——它们在处理期间执行并返回最终响应。
客户端函数工具受支持：模型在响应中返回 tool_calls 供您离线处理。多轮工具调用需要提交新的批量请求，并将工具结果消息包含在对话中。

高级 API 使用 ​

批量 API ​

什么是批量 API？ ​

何时使用批量 API ​

工作原理 ​

第 1 步：创建批量 ​

第 2 步：向批量添加请求 ​

第 3 步：监控批量进度 ​

了解批量状态 ​

第 4 步：检索结果 ​

其他操作 ​

取消批量 ​

列出所有批量 ​

检查单个请求状态 ​

跟踪成本 ​

完整示例 ​

JSONL 文件上传 ​

限制 ​

工具使用 ​

相关 ​

高级 API 使用

批量 API

什么是批量 API？

何时使用批量 API

工作原理

第 1 步：创建批量

第 2 步：向批量添加请求

第 3 步：监控批量进度

了解批量状态

第 4 步：检索结果

其他操作

取消批量

列出所有批量

检查单个请求状态

跟踪成本

完整示例

JSONL 文件上传

限制

工具使用

相关