Grok批处理API怎么用？JSONL提交、进度检查与结果对账

处理一批不需要立即回答的反馈，逐条等待会让程序难以恢复和核对。xAI的Batch API可以后台处理请求，再按ID取回结果。本教程只用两条虚构反馈，先走通JSONL、文件上传、创建批次和结果对账。示例不会代替你发起付费调用，复制执行前应确认自己的API权限和预算。

第一步：确认模型并设计唯一ID

需要xAI官方API密钥和支持Batch的模型。核对时，grok-4.7模型页列出了Batch支持及对应服务区域；实际账户、区域与可选模型以官方控制台为准。批处理为异步服务，文档描述通常在24小时内完成，但属于尽力处理，不应当作固定交付时限。

为每条业务记录分配唯一custom_id，例如fb_001。结果中用batch_request_id对应原记录，不能靠返回顺序配对。新建一个批次后重复提交相同业务数据，也不能假定系统会替你实现跨批次去重；自己的清单要记录原始ID、批次与处理状态。

第二步：生成JSONL并上传创建批次

import os, json, pathlib, requests
s = requests.Session()
s.headers["Authorization"] = "Bearer " + os.environ["XAI_API_KEY"]
base = "https://api.x.ai/v1"
records = {"fb_001": "界面容易理解", "fb_002": "提交后一直报错"}
lines = []
for key, text in records.items():
    lines.append(json.dumps({"custom_id": key, "method": "POST",
        "url": "/v1/chat/completions", "body": {"model": "grok-4.7",
        "messages": [{"role": "system", "content": "将反馈分为positive、negative、neutral，只输出一个类别。"},
                     {"role": "user", "content": text}]}}, ensure_ascii=False))
pathlib.Path("feedback.jsonl").write_text("\n".join(lines) + "\n", encoding="utf-8")
with open("feedback.jsonl", "rb") as f:
    upload = s.post(base + "/files", files={"file": f}, timeout=60)
upload.raise_for_status()
file_id = upload.json()["id"]
created = s.post(base + "/batches", json={"name": "feedback-demo",
                 "input_file_id": file_id}, timeout=60)
created.raise_for_status()
batch_id = created.json()["batch_id"]
pathlib.Path("batch-id.txt").write_text(batch_id, encoding="utf-8")
print("批次已创建：", batch_id)

JSONL每行必须是一个完整JSON对象，不能把整个数组当成一行上传。先本地解析各行、检查ID唯一再执行；上传成功只说明文件已接收，创建批次成功也不代表所有请求已经成功。保存batch-id.txt，后续查询复用它，不要为看进度重新运行提交脚本。

提交前再检查文件约束

官方说明文件方式创建的批次在创建后会封闭，不能继续向其中追加请求；需要补充数据时另建清单与批次。当前指南列出的文件上限为200MB、最多50000条请求，但初次接入应从本例两条开始，先确认字段与模型匹配。若任意一行格式无效，文件可能导致批次取消，应先修复原始JSONL，再按错误信息决定下一步。

不要在提交脚本外层加无条件重试循环。文件上传与创建批次是两个步骤：文件已上传但创建失败时，先查看返回状态，保存文件ID与错误，再确认是否需要重新创建；查询失败则只重试查询。密钥、批次ID和输入内容的保存范围也应区分，错误报告无需包含Authorization请求头。

第三步：单独检查进度并分页取回

import os, json, pathlib, requests
s = requests.Session()
s.headers["Authorization"] = "Bearer " + os.environ["XAI_API_KEY"]
base = "https://api.x.ai/v1"
bid = pathlib.Path("batch-id.txt").read_text(encoding="utf-8").strip()
r = s.get(base + "/batches/" + bid, timeout=60)
r.raise_for_status()
state = r.json()["state"]
print(state)
if int(state["num_requests"]) > 0 and int(state["num_pending"]) == 0:
    results, token = [], None
    while True:
        params = {"limit": 100}
        if token:
            params["pagination_token"] = token
        r = s.get(base + "/batches/" + bid + "/results", params=params, timeout=60)
        r.raise_for_status()
        page = r.json()
        results.extend(page.get("results", []))
        token = page.get("pagination_token")
        if not token:
            break
    pathlib.Path("batch-results.json").write_text(
        json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
    for item in results:
        response = item.get("batch_result", {}).get("response", {}).get("chat_get_completion")
        value = response["choices"][0]["message"]["content"] if response else item.get("error_message", "请检查原始错误")
        print(item["batch_request_id"], value)
else:
    print("尚未处理完毕；稍后只运行查询脚本。")

必须同时检查总请求数大于零和待处理数为零，空批次的零待处理不能证明两条反馈已经完成。num_success与num_error分别反映成功和失败；全部处理结束不等于全部成功。查询频率应适度，按异步任务处理，不要每秒重新上传文件。

第四步：按ID对账并处理失败

把结果ID集合与fb_001、fb_002比较，检查重复、遗漏和未知ID，再核对分类是否只含允许值。上例故意采用简单分类，结果文本仍应做去空白与白名单检查。实际业务可进一步使用结构约束，但不应只看到返回字符串就直接写入生产数据库。

保存每条错误详情，只把失败或确认遗漏的记录放入后续重试清单，同时保留它与原请求的对应关系。网络中断后先查询已有batch_id，确认创建状态再决定是否重试，避免产生重复任务和费用。正式扩大规模前，用小批量核对统计总数、结果数量、计费与失败处理是否符合预期。

官方资料

资料核对日期：2026年10月23日。

• xAI：Batch API官方指南
• xAI：grok-4.7能力与Batch支持范围