1. 为什么批量任务总在“最后一公里”翻车先说结论Python 做自动化批处理脚本本身从来不是难点难的是让脚本“稳定地跑完几百上千个文件”以及让 AI 帮你写脚本时别给你一段看着能跑、实际一跑就崩的代码。这篇聚焦 Python Trae 提示词驱动的自动化批处理落地围绕日志清洗、表格合并、文件重命名、批量摘要等 10 个场景给出可复制的提示词模板与脚本骨架同时把 TaoToken 统一 Key/API 通道的 Base URL 与 auth.json 配置示例讲清楚最后用一次批量任务跑通验证请求与结果落盘。如果你符合下面任意一条这篇就是写给你的手头有一堆 CSV/Excel 要合并清洗几百张图片要统一尺寸加水印几十个日志文件要提取错误行想让 Trae 或 Claude Code 这类工具直接生成能跑的批处理脚本而不是每次都要自己补 try-except以及你已经在用 AI 写代码但每次都要手动填 Key、换模型想找个统一入口。我试过最原始的写法一个 for 循环遍历目录处理函数里啥异常都不接跑到第 37 个文件遇到一个损坏的 PNG整个脚本直接退出前面 36 个白跑。后来才明白批处理的核心不是“处理逻辑”而是“容错 进度 可恢复”。AI 生成代码时如果你不把这三件事写进提示词它默认只给你最理想路径的代码。所以本文的每个场景提示词模板里都会强制要求三件事try-except 包裹单个任务、打印进度、失败文件单独记录。这不是啰嗦这是让脚本能真正跑完的关键。下面从环境准备开始一步步来。2. TaoToken 统一 Key 接入Base URL 与 auth.json 配置在写批处理脚本之前先把“调用通道”理顺。很多批处理场景需要调用大模型比如批量摘要、日志智能分类、表格字段语义映射。如果每个脚本都单独配 Key、单独记 Base URL维护成本会很高。TaoToken 提供统一 Key/API 通道一个 Key 走多个模型适合这种“脚本多、调用散”的场景。官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址统一用 https://taotoken.net/api 不加 UTM。2.1 三件套Base URL Key Model ID不管你是用 Claude Code、Cline、还是自己写 Python 脚本接入任何模型通道都离不开三件套。TaoToken 的对应关系如下配置项值说明Base URLhttps://taotoken.net/api所有请求的统一入口API Key在控制台创建形如 sk-xxxx只显示一次Model ID按需选择如 claude-sonnet-4-5、gpt-4o 等Key 的创建入口在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。创建后立刻复制页面刷新就看不到了。2.2 auth.json 配置示例Claude Code / Codex 类工具如果你用的是 Claude Code 或类似读取 auth.json 的工具配置长这样。注意路径要和工具默认读取路径一致Claude Code 一般在用户目录下的配置文件夹里{ anthropic: { baseUrl: https://taotoken.net/api, apiKey: sk-你的Key, model: claude-sonnet-4-5 } }如果你用的是 Codex 类工具auth.json 结构略有不同但核心三件套不变{ base_url: https://taotoken.net/api, api_key: sk-你的Key, model: gpt-4o }注意auth.json 里不要留多余字段有些工具对未知字段会直接报解析错误。改完保存重启工具生效。2.3 Python 脚本里的统一配置批处理脚本里我习惯把配置抽成一个 config.py所有脚本 import 它换 Key 只改一处# config.py import os TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY, sk-你的Key) DEFAULT_MODEL claude-sonnet-4-5 def get_client(): from openai import OpenAI return OpenAI( base_urlTAOTOKEN_BASE_URL, api_keyTAOTOKEN_API_KEY, )用环境变量兜底是个好习惯本地调试写死上线用环境变量避免 Key 进 Git。装依赖就一行pip install openai到这里通道就通了。接下来进入正题10 个批处理场景的提示词模板和脚本骨架。每个场景我都会给“Trae 提示词”和“关键代码”你复制提示词到 Trae 里它生成的代码基本能直接跑。3. 10 个批处理场景的提示词模板与脚本骨架这一章是全文最重的部分。10 个场景我按“文件操作类”和“数据/内容类”分成两组每个场景给提示词模板 核心代码骨架。提示词模板的写法有个通用套路先描述输入输出再列处理步骤最后强制要求容错和进度。你把这个套路记住以后自己写提示词也能套。3.1 场景一批量文件重命名正则 序号提示词模板写一个 Python 脚本批量重命名文件 - 读取指定目录下所有文件跳过子目录 - 用正则表达式匹配文件名按规则替换 - 支持按序号重命名格式如 file_0001.txt - 显示重命名进度和结果 - 单个文件失败不影响其他文件记录失败列表 用 os 和 re 模块代码要能直接跑加 try-except。核心骨架import os import re def batch_rename(directory, pattern, replacement, dry_runTrue): files [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))] failed [] for i, filename in enumerate(files, 1): try: new_name re.sub(pattern, replacement, filename) if new_name ! filename: if not dry_run: os.rename( os.path.join(directory, filename), os.path.join(directory, new_name) ) print(f[{i}/{len(files)}] {filename} - {new_name}) except Exception as e: failed.append((filename, str(e))) return faileddry_run 参数很关键先跑一遍看输出对不对确认了再真正改。这个习惯能救你很多次。3.2 场景二批量图片处理尺寸 水印 格式提示词模板写一个 Python 脚本批量处理图片 - 读取 input_images 目录下所有图片jpg/png/webp - 统一调整到 800x600用 LANCZOS 重采样 - 如果存在 logo.png 就在右下角加水印 - 转成 JPEG质量 85% - 按 product_0001.jpg 格式命名保存到 output_images - 生成处理报告统计成功失败数和耗时 用 Pillow 库加 try-except显示进度。核心骨架from PIL import Image import os def process_image(src, dst, watermark_pathlogo.png): try: img Image.open(src) if img.mode in (RGBA, LA, P): img img.convert(RGB) img img.resize((800, 600), Image.LANCZOS) if os.path.exists(watermark_path): wm Image.open(watermark_path).convert(RGBA) pos (img.size[0] - wm.size[0] - 10, img.size[1] - wm.size[1] - 10) layer Image.new(RGBA, img.size, (0, 0, 0, 0)) layer.paste(wm, pos, wm) img Image.alpha_composite(img.convert(RGBA), layer).convert(RGB) img.save(dst, JPEG, quality85, optimizeTrue) return True, None except Exception as e: return False, str(e)这个场景的坑在于 RGBA 转 RGB不转的话保存 JPEG 会直接报错。提示词里明确写“处理 RGBA 图片”Trae 就会帮你加上转换。3.3 场景三批量 CSV 数据清洗去空 去重 3σ提示词模板写一个 Python 脚本批量清洗 CSV - 读取 input_data 目录下所有 CSV - 删除含空值的行、删除重复行 - 对数值列用 3σ 规则剔除异常值 - 清洗后保存到 output_data文件名加 cleaned_ 前缀 - 生成报告统计每个文件清洗前后行数 用 pandas加 try-except显示进度。核心骨架import pandas as pd def clean_csv(src, dst): try: df pd.read_csv(src) before len(df) df df.dropna().drop_duplicates() for col in df.select_dtypes(include[int64, float64]).columns: mean, std df[col].mean(), df[col].std() if std 0: df df[(df[col] mean - 3 * std) (df[col] mean 3 * std)] df.to_csv(dst, indexFalse) return True, before, len(df) except Exception as e: return False, 0, str(e)3σ 规则对正态分布数据有效如果你的数据本身偏态严重建议换成 IQR 四分位距提示词里可以补一句“用 IQR 替代 3σ”。3.4 场景四批量表格合并多 Sheet 汇总提示词模板写一个 Python 脚本合并多个 Excel - 读取 input_excel 目录下所有 xlsx - 每个文件可能有多个 sheet全部读取 - 合并成一个大 DataFrame加一列 source_file 标记来源 - 保存为 merged.xlsx同时输出合并报告 用 pandas 和 openpyxl加 try-except。核心骨架import pandas as pd import os def merge_excel(input_dir, output_path): frames [] for f in os.listdir(input_dir): if not f.endswith(.xlsx): continue path os.path.join(input_dir, f) try: sheets pd.read_excel(path, sheet_nameNone) for name, df in sheets.items(): df[source_file] f df[source_sheet] name frames.append(df) except Exception as e: print(f跳过 {f}: {e}) if frames: pd.concat(frames, ignore_indexTrue).to_excel(output_path, indexFalse)合并类任务最大的坑是列名不一致有的文件叫“用户ID”有的叫“user_id”。提示词里加一句“列名做标准化处理统一转小写去空格”能省很多事。3.5 场景五批量日志清洗与错误提取提示词模板写一个 Python 脚本批量清洗日志 - 读取 logs 目录下所有 .log 文件 - 提取包含 ERROR、WARN、FATAL 的行 - 按时间戳排序输出到 errors_summary.txt - 统计每个文件的错误数生成统计表 用标准库加 try-except显示进度。核心骨架import os import re LEVELS (ERROR, WARN, FATAL) def extract_errors(log_dir, out_file): stats {} all_lines [] for f in os.listdir(log_dir): if not f.endswith(.log): continue count 0 try: with open(os.path.join(log_dir, f), encodingutf-8, errorsignore) as fh: for line in fh: if any(lv in line for lv in LEVELS): all_lines.append(line.rstrip()) count 1 except Exception as e: print(f读取 {f} 失败: {e}) stats[f] count with open(out_file, w, encodingutf-8) as fh: fh.write(\n.join(all_lines)) return stats日志文件编码经常不统一errorsignore 能避免因为一个乱码字符整个文件读失败。3.6 场景六批量文档摘要调用大模型这个场景开始用到 TaoToken 通道。提示词模板写一个 Python 脚本批量生成文档摘要 - 读取 docs 目录下所有 txt 文件 - 每个文件调用大模型生成 100 字摘要 - Base URL 用 https://taotoken.net/api - 从环境变量 TAOTOKEN_API_KEY 读 Key - 摘要保存到 summaries 目录文件名加 _summary 后缀 - 单个文件失败不影响其他记录失败列表 用 openai 库加 try-except 和进度显示。核心骨架import os from config import get_client, DEFAULT_MODEL def summarize_file(client, src, dst): try: with open(src, encodingutf-8) as f: content f.read()[:4000] resp client.chat.completions.create( modelDEFAULT_MODEL, messages[{role: user, content: f用100字总结以下内容\n{content}}], ) summary resp.choices[0].message.content with open(dst, w, encodingutf-8) as f: f.write(summary) return True, None except Exception as e: return False, str(e)这里最容易踩的坑是内容太长超 token所以先截断到 4000 字符。更严谨的做法是按 token 估算但批处理场景截断够用了。3.7 场景七批量代码格式化提示词模板写一个 Python 脚本批量格式化代码 - 遍历 src 目录下所有 .py 文件 - 用 black 格式化用 isort 整理 import - 格式化前后做 diff输出变更报告 - 单个文件失败不影响其他 用 subprocess 调用 black 和 isort加 try-except。核心骨架import subprocess import os def format_file(path): try: subprocess.run([black, path], checkTrue, capture_outputTrue) subprocess.run([isort, path], checkTrue, capture_outputTrue) return True, None except subprocess.CalledProcessError as e: return False, e.stderr.decode()批处理格式化前一定要先 git commit不然改乱了没法回滚。3.8 场景八批量文件编码转换提示词模板写一个 Python 脚本批量转换文件编码 - 读取指定目录下所有 txt/csv 文件 - 自动检测原编码用 chardet - 统一转成 UTF-8 - 转换失败的文件记录到 failed.txt 用 chardet 和标准库加 try-except。核心骨架import chardet def convert_encoding(src, dst): try: with open(src, rb) as f: raw f.read() enc chardet.detect(raw)[encoding] or utf-8 text raw.decode(enc, errorsignore) with open(dst, w, encodingutf-8) as f: f.write(text) return True, enc except Exception as e: return False, str(e)GBK 和 UTF-8 混用的项目里这个脚本能一次性解决乱码问题。3.9 场景九批量图片 OCR 提取文字提示词模板写一个 Python 脚本批量 OCR - 读取 images 目录下所有图片 - 用 pytesseract 提取文字 - 结果保存到 ocr_results 目录同名 txt - 统计每个图片的字符数 加 try-except显示进度。核心骨架import pytesseract from PIL import Image def ocr_image(src, dst): try: text pytesseract.image_to_string(Image.open(src), langchi_simeng) with open(dst, w, encodingutf-8) as f: f.write(text) return True, len(text) except Exception as e: return False, str(e)OCR 前建议先做灰度化和二值化识别率能提升不少提示词里可以补一句“预处理转灰度 自适应阈值”。3.10 场景十批量 API 调用与结果落盘提示词模板写一个 Python 脚本批量调用 API - 读取 tasks.csv每行一个任务 - 对每个任务调用大模型接口 - Base URL 用 https://taotoken.net/api - 结果写入 results.jsonl每行一条 - 支持断点续跑已完成的跳过 - 失败任务记录到 failed.jsonl 用 openai 库加 try-except 和进度显示。核心骨架import json import os from config import get_client, DEFAULT_MODEL def run_batch(tasks_file, out_file, failed_file): done set() if os.path.exists(out_file): with open(out_file, encodingutf-8) as f: for line in f: done.add(json.loads(line)[task_id]) client get_client() with open(out_file, a, encodingutf-8) as out, \ open(failed_file, a, encodingutf-8) as fail: for task in load_tasks(tasks_file): if task[id] in done: continue try: resp client.chat.completions.create( modelDEFAULT_MODEL, messages[{role: user, content: task[prompt]}], ) out.write(json.dumps({ task_id: task[id], result: resp.choices[0].message.content, }, ensure_asciiFalse) \n) out.flush() except Exception as e: fail.write(json.dumps({ task_id: task[id], error: str(e), }, ensure_asciiFalse) \n)断点续跑是长任务的生命线。用 jsonl 格式追加写每写一条 flush 一次进程被 kill 也不丢数据。4. 跑通验证一次批量摘要任务的完整过程光看代码不够得跑一遍。这一章用场景六的批量摘要任务做验证从准备数据到结果落盘完整走一遍。4.1 准备测试数据先造几个测试文档mkdir -p docs summaries for i in 1 2 3 4 5; do echo 这是第 $i 篇测试文档。内容涉及批处理、自动化、Python 脚本。 docs/doc_$i.txt done4.2 配置环境变量Key 不要写进代码用环境变量export TAOTOKEN_API_KEYsk-你的KeyWindows PowerShell 用$env:TAOTOKEN_API_KEYsk-你的Key4.3 运行批量摘要脚本# batch_summary.py import os from config import get_client, DEFAULT_MODEL def main(): client get_client() docs [f for f in os.listdir(docs) if f.endswith(.txt)] failed [] for i, name in enumerate(docs, 1): src os.path.join(docs, name) dst os.path.join(summaries, name.replace(.txt, _summary.txt)) try: with open(src, encodingutf-8) as f: content f.read()[:4000] resp client.chat.completions.create( modelDEFAULT_MODEL, messages[{role: user, content: f用100字总结\n{content}}], ) with open(dst, w, encodingutf-8) as f: f.write(resp.choices[0].message.content) print(f[{i}/{len(docs)}] {name} - OK) except Exception as e: failed.append((name, str(e))) print(f[{i}/{len(docs)}] {name} - FAIL: {e}) print(f完成失败 {len(failed)} 个) if __name__ __main__: main()运行python batch_summary.py4.4 预期输出与结果落盘正常输出类似[1/5] doc_1.txt - OK [2/5] doc_2.txt - OK [3/5] doc_3.txt - OK [4/5] doc_4.txt - OK [5/5] doc_5.txt - OK 完成失败 0 个summaries 目录下会生成 5 个 _summary.txt 文件每个里面是模型生成的摘要。打开看一眼确认内容合理说明通道和脚本都通了。4.5 验证请求是否真的走了 TaoToken想确认请求确实走了统一通道可以在 config.py 里临时加一行打印print(fBase URL: {client.base_url})输出应该是 https://taotoken.net/api 。如果显示的是别的地址说明配置没生效检查环境变量或 auth.json。5. 常见报错排查401、local proxy failed、reading choices、OAuth批处理脚本跑不通90% 是下面这几类错误。逐个对照排查。5.1 401 Unauthorized报错长这样openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因就三种Key 写错、Key 过期、环境变量没读到。排查顺序先打印 os.getenv(TAOTOKEN_API_KEY) 看是不是 None再看 Key 有没有多余空格最后去控制台确认 Key 状态。注意 Key 只在创建时显示一次忘了就重新建一个。5.2 local proxy failed / connection refused报错openai.APIConnectionError: Connection error.或者local proxy failed: connect ECONNREFUSED 127.0.0.1:7890这类错误通常是本地网络配置问题。检查你的 HTTP_PROXY / HTTPS_PROXY 环境变量是不是指向了一个没启动的本地端口。批处理脚本里可以显式清掉import os os.environ.pop(HTTP_PROXY, None) os.environ.pop(HTTPS_PROXY, None)5.3 reading choices 报错报错AttributeError: NoneType object has no attribute choices或者KeyError: choices这说明返回体结构和你预期的不一样。常见原因是模型名写错接口返回了错误信息而不是正常响应。先打印完整 resp 看结构print(resp.model_dump_json(indent2))确认 model 字段是不是你配置的那个 Model ID。三件套里 Model ID 写错最容易出这个错。5.4 OAuth 相关报错报错OAuth token expired或者invalid_grant如果你用的是 Claude Code 类工具它可能优先走 OAuth 而不是 auth.json 里的 Key。解决办法是确认工具的认证模式强制走 API Key 模式。检查 auth.json 里有没有残留的 oauth 字段有就删掉。三件套配置完整Base URL Key Model ID的情况下不应该再触发 OAuth 流程。5.5 批量任务中途卡死不是报错但更烦人。表现是脚本跑到一半不动了。原因通常是某个请求没有超时设置一直挂着。给 client 加超时client OpenAI( base_urlTAOTOKEN_BASE_URL, api_keyTAOTOKEN_API_KEY, timeout30.0, max_retries2, )timeout 30 秒重试 2 次基本能避免卡死。配合前面说的断点续跑即使中断也能接着跑。6. 把批处理脚本接进你的日常工作流最后聊点实操经验。批处理脚本写多了你会发现真正提升效率的不是单个脚本多精妙而是把它们串成工作流。第一所有脚本共用 config.py。Key、Base URL、Model ID 只维护一处换通道时改一个文件几十个脚本全部生效。这是 TaoToken 统一 Key 通道最大的价值不用每个脚本单独配。第二提示词模板存成文件。我习惯在项目里建 prompts/ 目录每个场景一个 .txtTrae 里直接引用。这样提示词可以版本管理改进了也能追溯。第三批处理任务全部支持 dry_run。先跑一遍看输出确认无误再真跑。这个习惯帮我避免过好几次批量误删。第四长任务用 jsonl 追加写 断点续跑。别用一次性写整个文件的方式进程一挂全丢。如果你还在手动一个个处理文件从场景一和场景三开始试这两个最容易见效。跑通之后把 config.py 和 auth.json 配好后面所有需要调用大模型的批处理场景都能直接复用。需要看更多接入细节的话接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 模型对话调试在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 长期跑编码和 Agent 类批处理任务的话Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。先把第一个脚本跑起来剩下的都是复制粘贴改参数的事。