OpenClaw 企业定制 私人订制工作流 优化运维 openclaw记忆优化 问题咨询解决 请加微信交流
需要给AI模型生成训练数据？想批量跑上千条prompt来收集工具调用轨迹？Batch Processing就是来解决这个问题的。它能并行处理海量prompt，生成结构化的训练数据，适合微调或评估用。
这篇讲清楚Batch Processing是什么、能做什么、以及实际使用场景。
Batch Processing是什么
简单说：一次性跑大量prompt，每个prompt有独立环境，输出结构化的轨迹数据。
核心用途是生成训练数据——产生ShareGPT格式的轨迹，包含完整对话历史、工具调用统计、推理覆盖度指标。
快速开始
# 基本用法
python batch_runner.py \
--dataset_file=data/prompts.jsonl \
--batch_size=10 \
--run_name=my_first_run \
--model=anthropic/claude-sonnet-4.6 \
--num_workers=4
# 从断点恢复
python batch_runner.py \
--dataset_file=data/prompts.jsonl \
--batch_size=10 \
--run_name=my_first_run \
--resume
# 查看可用工具集分布
python batch_runner.py --list_distributions数据集格式
输入是一个JSONL文件（每行一个JSON对象），每条必须有prompt字段：
{"prompt": "写一个Python函数找最长回文子串"}
{"prompt": "用Flask创建用户认证REST API"}
{"prompt": "调试这个错误：TypeError: cannot unpack non-iterable NoneType object"}可选字段：
- image/dockcer_image：容器镜像，用于沙箱隔离
- cwd：终端会话的工作目录覆盖
核心参数
| 参数 | 默认值 | 说明 |
|---|---|---|
| –dataset_file | 必需 | JSONL数据集路径 |
| –batch_size | 必需 | 每批prompt数量 |
| –run_name | 必需 | 运行名称（用于输出目录和断点） |
| –distribution | default | 工具集分布 |
| –model | claude-sonnet-4.6 | 使用的模型 |
| –max_turns | 10 | 每个prompt最大工具调用轮次 |
| –num_workers | 4 | 并行工作进程数 |
| –resume | false | 从断点恢复 |
| –max_samples | all | 只处理前N条数据 |
Provider路由
# 只允许特定provider
--providers_allowed "anthropic,openai"
# 忽略某些provider
--providers_ignored "together,deepinfra"
# 优先provider顺序
--providers_order "anthropic,openai,google"
# 按价格/吞吐量/延迟排序
--provider_sort "price"推理控制
# 推理深度级别
--reasoning_effort medium
# 完全禁用推理/思考token
--reasoning_disabled输出格式
所有输出到data/<run_name>/目录：
data/my_run/
├── trajectories.jsonl # 合并后的最终输出
├── batch_0.jsonl # 各批次结果
├── batch_1.jsonl
├── ...
├── checkpoint.json # 断点文件
└── statistics.json # 聚合工具使用统计轨迹格式
每行是一个JSON对象：
{
"prompt_index": 42,
"conversations": [
{"from": "human", "value": "写一个函数..."},
{"from": "gpt", "value": "我来创建...",
"tool_calls": [...]},
{"from": "tool", "value": "..."},
{"from": "gpt", "value": "完成！"}
],
"metadata": {
"batch_num": 2,
"timestamp": "2026-01-15T10:30:00",
"model": "anthropic/claude-sonnet-4.6"
},
"completed": true,
"partial": false,
"api_calls": 3,
"toolsets_used": ["terminal", "file"],
"tool_stats": {
"terminal": {"count": 2, "success": 2, "failure": 0},
"read_file": {"count": 1, "success": 1, "failure": 0}
},
"tool_error_counts": {
"terminal": 0,
"read_file": 0
}
}断点机制
Batch Processing有强大的断点容错机制：
- 每批完成后保存checkpoint
- 内容匹配恢复：resume时扫描已有batch文件，通过实际内容匹配已完成prompt
- 失败重试：只有成功完成的prompt标记为done，失败的会重试
- 批次合并：完成后所有batch文件合并为trajectories.jsonl
Resume怎么工作
- 扫描所有batch_*.jsonl，通过内容匹配找出已完成的prompt
- 从数据集中过滤掉已完成的
- 重新分批剩余的prompt
- 只处理剩余的prompt
- 合并所有batch文件（旧的+新的）为最终输出
质量过滤
自动质量过滤：
- 无推理过滤器：没有任何助手回复包含推理的样本会被丢弃
- 损坏条目过滤器：幻觉工具名的条目在最终合并时被过滤
- 推理统计：跟踪整个运行中有/无推理的回复占比
统计输出
完成后输出详细统计：
- 工具使用：每个工具的调用次数、成败率
- 推理覆盖：有推理的助手回复百分比
- 丢弃样本：因无推理被过滤的数量
- 耗时：总处理时间
统计同时保存到statistics.json供程序分析。
工具集分布
每个prompt从分布中随机采样工具集，确保训练数据覆盖多样化工具组合。
当前实现：分布给每个工具集分配概率，独立flip每个工具集，保证至少启用一个工具集。
实际使用场景
训练数据生成
python batch_runner.py \
--dataset_file=data/coding_prompts.jsonl \
--batch_size=20 \
--run_name=coding_v1 \
--model=anthropic/claude-sonnet-4.6 \
--num_workers=8 \
--distribution=default \
--max_turns=15生成多样化的工具使用轨迹用于微调。
模型评估
python batch_runner.py \ --dataset_file=data/eval_suite.jsonl \ --batch_size=10 \ --run_name=eval_gpt4 \ --model=openai/gpt-4o \ --num_workers=4 \ --max_turns=10评估模型在标准prompt下的工具使用能力。
每prompt独立容器
{"prompt": "安装numpy并计算3x3矩阵特征值", "image": "python:3.11-slim"} {"prompt": "编译并运行这个Rust程序", "image": "rust:1.75"}每个prompt用自己的容器镜像，适合需要特定环境的基准测试。
高级选项
- --ephemeral_system_prompt：执行时使用但不保存到轨迹的系统提示词
- --log_prefix_chars：日志预览显示的字符数（默认100）
- --prefill_messages_file：few-shot priming的预填充消息JSON文件路径
常见问题
能跑多少条prompt？
没有硬性限制，取决于你的API配额和机器资源。建议从100条开始测试。
断了怎么办？
用--resume从断点恢复，只重跑未完成的。
怎么控制成本？
用--max_samples限制数量，用--providers_allowed限制provider，用--provider_sort按价格排序。
输出能直接用于微调吗？
能，格式是ShareGPT格式，工具调用轨迹完整，可以直接用于微调。
怎么提高数据质量？
用--reasoning_effort控制推理深度，启用质量过滤（默认开启）自动剔除无推理样本。
总结
Batch Processing是生成AI训练数据的利器。
核心价值：
- 批量并行处理大量prompt
- 输出结构化训练数据
- 支持断点恢复
- 自动质量过滤
- 工具调用统计完整
- 适合微调和模型评估
如果你在训练自己的AI模型，这个工具能大幅提升数据生产效率。
暂无评论内容