适用场景
适合手上有大量文本要处理、又不想为一次性任务写完整项目的开发者和运维同学：比如把几千条用户反馈批量改写、批量打标签、批量翻译。用命令行做这件事的好处是每一步都能单独跑、单独看输出，出错时能定位到具体是鉴权、上传、提交还是取结果环节。本文从装好命令行工具开始，一路做到把「读 CSV → 生成请求 → 提交批处理 → 取回结果」串成一条可直接照抄的流水线。
环境与前置条件
- 操作系统：macOS、Linux 或 Windows + WSL 均可，本文命令以 bash 为准。
- Python：用官方 SDK 自带的命令行入口，需要 Python 3.8 以上；具体最低版本以官方文档当前版本为准。
- 系统工具：
curl、jq（解析 JSON 用，务必装）、python3。 - 账号：一个可用的 API Key，且账号已开通对应模型的权限。
- 磁盘与内存：普通笔记本即可。批处理文件动辄几十上百 MB，建议给工作目录留出至少 1 GB 空闲空间；如果自己生成请求文件，注意单文件体积和条数上限，具体数值以官方文档当前版本为准。
- 网络：能稳定访问 API 域名，公司网络需要走代理的话先确认
curl能通。
分步骤部署
第 1 步：安装命令行工具并确认入口可用
建一个独立虚拟环境，避免和系统里其他 Python 包打架：
```bash
python3 -m venv ~/.venvs/openai-cli
source ~/.venvs/openai-cli/bin/activate
pip install --upgrade pip
pip install --upgrade openai
```
装完检查入口是否存在：
```bash
openai --help
```
成功的样子是打印出一串子命令说明。如果提示 command not found，先确认虚拟环境已激活，再执行 pip show -f openai 看 bin 目录里有没有可执行文件。不同版本的 SDK 附带的子命令名称会有差异，建议先看一眼 openai --help 和 openai api --help 的真实输出，本文后续命令以 help 里列出的动作为准；某条子命令在你装的版本里不存在时，用文中给出的等价 curl 写法替代，效果完全一样。
第 2 步：配置鉴权，别把 Key 写进脚本
Key 一律走环境变量，代码和脚本里只引用变量名：
```bash
export OPENAI_API_KEY="sk-你的key"
```
想让新终端自动带上，可以把上面一行加进 ~/.bashrc 或 ~/.zshrc，并执行 chmod 600 ~/.bashrc。如果你在多个组织或项目下工作，再补两个变量：
```bash
export OPENAI_ORG_ID="org-xxxx" # 可选
export OPENAI_PROJECT_ID="proj-xxxx" # 可选
```
自检一下变量是否真的读到了（不会回显完整 Key）：
```bash
echo "${OPENAI_API_KEY:0:6}...${OPENAI_API_KEY: -4}"
```
第 3 步：跑通第一次模型调用
先用命令行发一条最小请求，验证 Key、网络、模型权限三件事：
```bash
export MODEL="你的模型名" # 在官方文档的模型列表里挑一个当前可用的
openai api chat.completions.create \
-m "$MODEL" \
-g user "用三句话说明批处理接口适合什么任务"
```
-g 是「角色 + 内容」的简写，具体参数名以 openai api chat.completions.create --help 为准。成功时会直接打印一段 JSON，里面的 choices[0].message.content 就是回答。
等价的 curl 写法，方便你在没有 CLI 的机器上验证：
```bash
curl -s https://api.openai.com/v1/chat/completions \
-H "Authorization: Bearer $OPENAI_API_KEY" \
-H "Content-Type: application/json" \
-d "{\"model\":\"$MODEL\",\"messages\":[{\"role\":\"user\",\"content\":\"你好\"}]}" \
| jq -r '.choices[0].message.content'
```
第 4 步：准备批处理请求文件
批处理要求一个 JSONL 文件：一行一条请求，每行必须是完整合法的 JSON，行尾有换行，中间不能有空行，也不要带 BOM。一个最小可用样例：
```json
{"custom_id": "req-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "你的模型名", "messages": [{"role": "user", "content": "把这句话改写成标题：服务器磁盘告警"}]}}
```
用 Python 批量生成，避免手写引号出错：
```bash
python3 - <<'PY'
import json, os
model = os.environ["MODEL"]
texts = ["磁盘告警怎么排查", "接口超时排查思路", "如何做日志轮转"]
with open("requests.jsonl", "w", encoding="utf-8") as f:
for i, t in enumerate(texts, 1):
body = {"model": model, "messages": [{"role": "user", "content": f"改写成 20 字以内的标题：{t}"}]}
f.write(json.dumps({"custom_id": f"req-{i:03d}", "method": "POST",
"url": "/v1/chat/completions", "body": body},
ensure_ascii=False) + "\n")
PY
```
生成后逐行校验，这一步能省掉后面大量排查时间：
```bash
python3 -c "
import json,sys
for n,l in enumerate(open('requests.jsonl',encoding='utf-8'),1):
if not l.strip(): sys.exit(f'第 {n} 行是空行')
json.loads(l)
print('OK')
"
```
第 5 步：上传文件
```bash
openai api files.create -f requests.jsonl -p batch
```
输出里记下 id 字段，形如 file-xxxx，这就是后面要用的输入文件 ID。上传用途必须是 batch，填成 fine-tune 会在提交批处理时被拒绝。
用 shell 变量接住，后面就不用复制粘贴：
```bash
FILE_ID=$(openai api files.create -f requests.jsonl -p batch | jq -r '.id')
echo "$FILE_ID"
```
第 6 步：提交批处理
批处理接口用 curl 最稳，参数含义一目了然：
```bash
BATCH_ID=$(curl -s https://api.openai.com/v1/batches \
-H "Authorization: Bearer $OPENAI_API_KEY" \
-H "Content-Type: application/json" \
-d "{\"input_file_id\":\"$FILE_ID\",\"endpoint\":\"/v1/chat/completions\",\"completion_window\":\"24h\"}" \
| jq -r '.id')
echo "$BATCH_ID"
```
如果你的 SDK 版本里存在 openai api batches.create 这类子命令，参数一一对应即可；不确定就用上面这段。
第 7 步：轮询状态并取回结果
```bash
curl -s "https://api.openai.com/v1/batches/$BATCH_ID" \
-H "Authorization: Bearer $OPENAI_API_KEY" \
| jq -r '.status, .request_counts'
```
status 会经历 validating、in_progress、finalizing、completed 等阶段。等到 completed 后，取出 output_file_id 再下载：
```bash
OUT_ID=$(curl -s "https://api.openai.com/v1/batches/$BATCH_ID" \
-H "Authorization: Bearer $OPENAI_API_KEY" | jq -r '.output_file_id')
curl -s "https://api.openai.com/v1/files/$OUT_ID/content" \
-H "Authorization: Bearer $OPENAI_API_KEY" -o result.jsonl
```
把结果摊平成「自定义 ID + 文本」两列，便于回填原始表格：
```bash
jq -r '[.custom_id, (.response.body.choices[0].message.content // .error.message)] | @tsv' result.jsonl
```
部分失败不会让整个批次失败，失败项在 error_file_id 指向的文件里，同样用上面 files/.../content 的写法下载。
第 8 步：把脚本任务改写成一条流水线
把前面几步固化成一个脚本 pipeline.sh，输入一个 CSV，输出一个带结果的 TSV：
```bash
#!/usr/bin/env bash
set -euo pipefail
: "${OPENAI_API_KEY:?请先 export OPENAI_API_KEY}"
: "${MODEL:?请先 export MODEL}"
IN_CSV="${1:?用法: ./pipeline.sh input.csv}"
1) CSV -> JSONL
python3 - "$IN_CSV" "$MODEL" <<'PY'
import csv, json, sys
src, model = sys.argv[1], sys.argv[2]
with open(src, newline="", encoding="utf-8") as f, open("requests.jsonl", "w", encoding="utf-8") as out:
for i, row in enumerate(csv.DictReader(f), 1):
body = {"model": model, "messages": [{"role": "user", "content": row["text"]}]}
out.write(json.dumps({"custom_id": f"row-{i:05d}", "method": "POST",
"url": "/v1/chat/completions", "body": body},
ensure_ascii=False) + "\n")
PY
2) 上传 + 提交
FILE_ID=$(openai api files.create -f requests.jsonl -p batch | jq -r '.id')
BATCH_ID=$(curl -s https://api.openai.com/v1/batches \
-H "Authorization: Bearer $OPENAI_API_KEY" -H "Content-Type: application/json" \
-d "{\"input_file_id\":\"$FILE_ID\",\"endpoint\":\"/v1/chat/completions\",\"completion_window\":\"24h\"}" \
| jq -r '.id')
echo "batch: $BATCH_ID"
3) 轮询
while :; do
S=$(curl -s "https://api.openai.com/v1/batches/$BATCH_ID" -H "Authorization: Bearer $OPENAI_API_KEY")
ST=$(echo "$S" | jq -r '.status')
echo "status=$ST"
[ "$ST" = "completed" ] && break
[ "$ST" = "failed" ] || [ "$ST" = "cancelled" ] || [ "$ST" = "expired" ] && { echo "$S"; exit 1; }
sleep 30
done
4) 下载 + 摊平
OUT_ID=$(echo "$S" | jq -r '.output_file_id')
curl -s "https://api.openai.com/v1/files/$OUT_ID/content" \
-H "Authorization: Bearer $OPENAI_API_KEY" -o result.jsonl
jq -r '[.custom_id, (.response.body.choices[0].message.content // .error.message)] | @tsv' result.jsonl
```
加上执行权限：chmod +x pipeline.sh。小批量任务也可以在流水线里先用第 3 步的单条调用跑通一条，再整体提交，省钱也省时间。
验证部署是否成功
按顺序跑这四条，全部通过即代表部署可用：
1. openai --help 有输出，不报 command not found。
2. echo "${OPENAI_API_KEY:0:6}" 能打印前 6 位，说明环境变量生效。
3. 第 3 步的单条调用返回 JSON，且 jq -r '.choices[0].message.content' 有非空文本。
4. ./pipeline.sh input.csv 最终打印出每行「row-00001 + 结果文本」，且 wc -l result.jsonl 的行数等于输入条数。
常见报错与解决
1. 401 Invalid Authentication 或 Incorrect API key provided
→ Key 没读到、被引号污染、或者已经被撤销。先确认 echo "${OPENAI_API_KEY:0:6}" 有输出；如果是从文档里复制的 Key，注意别把中文引号一起带进去。改用新 Key 后重新 export：
```bash
export OPENAI_API_KEY="sk-新key" && openai api models.list >/dev/null && echo ok
```
2. 429 分两种，要分开处理
→ 一种是 rate_limit_exceeded，单位时间请求太多；另一种是 insufficient_quota，余额或额度用完了。前者做退避重试，后者只能去后台看用量与额度，顺便确认账号是否已启用计费。批处理场景本身就有并发上限，把整批任务拆小通常比硬重试更有效。
3. 400 Invalid file format / line N: ... is not valid JSON
→ 请求文件某一行不是合法 JSON，常见原因是手写时用了中文引号、行尾缺换行、文件带 BOM、或者有空行。用第 4 步的校验脚本定位到具体行号后重写该行；若怀疑 BOM，用 sed -i '1s/^\xEF\xBB\xBF//' requests.jsonl 去掉。
4. 400 提示 purpose 不匹配或 endpoint 不支持
→ 上传时 -p 填的用途要和后续提交的接口一致，批处理用 batch，endpoint 填 /v1/chat/completions。另外批处理请求体里不要塞流式参数（如 stream: true），批处理不支持流式。
5. curl: (35) SSL connect error 或长时间无响应
→ 多半是代理或出口网络问题。先 curl -I https://api.openai.com/v1/models -H "Authorization: Bearer $OPENAI_API_KEY" 看能否握手，需要代理时给 curl 加 --proxy，给 Python 设置 HTTPS_PROXY 环境变量，两边的出口要保持一致。
后续维护
- 备份：
requests.jsonl、result.jsonl、批次 ID 和输入文件 ID 建议按日期归档到一个目录里，命令形如mkdir -p archive/$(date +%F) && cp requests.jsonl result.jsonl archive/$(date +%F)/。批处理的输出文件在服务端有保留期限，别指望随时能重新下载，以官方文档当前版本为准。 - 升级：SDK 升级用
pip install --upgrade openai，升级后先重跑第 3 步单条调用和一次三条以内的小批量，确认子命令和参数没变再跑生产任务。同理，脚本里依赖的接口字段如custom_id、output_file_id也要跟着官方文档走。 - 日志：所有提交动作都打印批次 ID 和输入文件 ID，输出重定向到文件，例如
./pipeline.sh input.csv 2>&1 | tee logs/run-$(date +%F-%H%M).log，出问题时有据可查。 - 监控：轮询脚本里把
status和request_counts一起打出来，重点关注失败条数与error_file_id；对长期运行的任务，可以每小时跑一次状态检查，状态卡在validating超过预期时长就人工介入。 - 成本控制：先用
head -n 20 requests.jsonl > sample.jsonl做小样本试跑，确认输出格式和 prompt 效果，再全量提交；同时把模型名和条数写进日志，便于和后台用量对账。