网上关于"AI 自动化""单人日入过万"的帖子很多，真正把营销话术剥掉之后，剩下的可复现部分其实只有一条工程流水线：批量任务怎么组织、模型怎么调用、失败怎么重试、额度怎么控制、结果怎么校验、断点怎么续跑。收益部分因人、因平台规则而异，本文不做任何承诺，只把这条流水线从零到一拆开讲清楚。本文面向会一点 Python、想把 AI 调用做成稳定定时任务而不是手动点对话框的开发者，读完你可以得到一套能直接跑起来的工程骨架：配置文件、模型调用封装、并发调度、预算熔断、结果落盘、质量抽检和定时部署，代码全部可复制。整体定位是一篇"项目实战 + 工程经验"型教程，重点放在可靠性和成本控制上，而不是教你写一句漂亮提示词就结束。

1. 概念澄清：AI 自动化流水线到底自动化了什么

1.1 三个自动化层次，别一上来就想全自动

讨论"AI 自动化"之前，需要先把自动化的层次分开，否则很容易把一件基础设施工程误解成"点一下按钮钱就来了"。第一层是调用自动化，指的是把人工在对话框里复制粘贴的动作，变成程序里的 HTTP 请求，这是最容易实现的一层，一天就能跑通；第二层是流程自动化，指的是把任务输入、批量循环、结果落盘、失败重试、定时触发串成一条链，这一层决定了你能不能连续跑一周不出问题；第三层是决策自动化，指的是让系统自己判断生成的内容是否合格、是否发布、是否要回滚，这一层涉及质量评估和业务规则，工程量和风险都最高。绝大多数所谓"AI 副业复盘"的帖子，实际上只停留在第一层，把控制台打印出来的一次成功响应当成可稳定复制的生产能力，这是后续所有问题的根源。

本文的实战部分会覆盖第一层和第二层，也就是调用封装 + 流水线调度，并在第 5 节给出第三层中"可自动化部分的边界"——哪些校验可以交给代码，哪些必须留人工复核。把边界划清楚，比盲目追全自动重要得多，因为一次批量生成几百条低质内容，清理成本往往高于慢慢做十条。

1.2 一条可复用流水线由哪些模块组成

把流水线拆成模块，是为了让每一块都能单独测试、单独替换。参考典型的后端任务系统，一条 AI 批量生成流水线通常包含六个部分：

• 配置层：密钥、接口地址、模型名、超时、并发数、预算上限，全部从环境变量或配置文件读取，不写死在代码里。
• 调用层：统一封装模型 SDK，向上只暴露一个chat(system, user)方法，屏蔽不同服务商参数差异。
• 调度层：读取任务列表，按并发度分发，收集结果，控制顺序写入。
• 持久层：结果以 JSONL 追加写入，天然支持断点续跑，中途崩了不用重头再来。
• 校验层：长度、关键词覆盖、重复度、结构合法性，先机器筛一遍再交给人。
• 成本层：按 token 累计成本，超过预算直接熔断，避免一晚上跑飞账单。

这六个模块里，最容易被忽略的是持久层和成本层，而它们恰恰是"能不能跑第二周"的关键。后面每节都会对应其中一个模块给出可运行代码，读者可以按顺序抄，也可以只取自己需要的那块。

1.3 "日入过万"这类说法，工程视角该怎么看

从工程角度看，任何"单人日产过万"的描述都需要拆成可验证的变量：单条产出的耗时、单条产出的成本、单条被采用的概率、平台对批量内容的容忍度。这四个变量里，只有前两个是技术能优化的，后两个由业务和平台规则决定，技术这边最多做到"不因为程序问题而失败"。所以合理的工程目标是：把单条生成成本压到可核算范围、把失败率压到可忽略、把重复劳动压到零，而不是许诺一个具体收益数字。

这也是我在实际项目里坚持的一条原则：先用流水线跑 20 条，把成本、耗时、抽检合格率三个数据打出来，再决定要不要放大到 200 条。没有这三个数据就扩量，本质上是在赌。本文后面的代码会内置成本和耗时统计，就是为了让你每次运行结束都能看到这三个数。

2. 环境准备与项目结构

2.1 运行环境与依赖说明

本文示例使用的运行环境如下，具体版本可以根据你自己的项目情况调整，重点是把依赖锁定在同一个虚拟环境里，避免版本漂移导致 SDK 参数不兼容：

• 操作系统：Linux（Ubuntu 22.04 及以上）或 macOS，Windows 建议用 WSL 或 PowerShell 运行同一套代码，路径分隔符差异用pathlib屏蔽。
• Python：3.9 及以上，本文代码用到了dataclasses和类型注解，3.8 也能跑但建议升级。
• 关键依赖：OpenAI 兼容协议官方 Python SDK 1.x 版本（openai>=1.0.0）。注意 0.x 和 1.x 的调用方式完全不同，openai.ChatCompletion.create在 1.x 已经不存在，网上很多旧教程就是因为这一点跑不起来。
• 无需额外安装requests、tenacity，重试逻辑我们自己写，减少依赖层级。

依赖文件写成下面这样，放在项目根目录：

# 文件路径：requirements.txt openai>=1.0.0

安装命令：

python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -r requirements.txt

这里强调"每个项目一个虚拟环境"不是形式主义。AI 相关 SDK 更新频繁，同一台机器上两个项目共用全局环境，很容易出现 A 项目能跑、B 项目报TypeError: create() got an unexpected keyword argument的情况，排查成本远高于建环境的三十秒。

2.2 项目目录结构

目录按职责分层，避免所有代码堆在一个main.py里。下面这套结构在单人维护、后期加人时都不需要重构：

# 目录结构 ai-pipeline/ ├── .env # 本地密钥，加入 .gitignore ├── .env.example # 配置模板，可以提交 ├── requirements.txt ├── config.py # 配置加载 ├── llm_client.py # 模型调用封装 ├── cost.py # 预算熔断 ├── pipeline.py # 调度与落盘 ├── quality.py # 质量抽检 ├── main.py # 入口 ├── tasks.json # 任务输入 ├── logs/ │ └── run.log └── output/ └── results.jsonl # 结果追加写入

.env一定要加到.gitignore，这是密钥管理的最低要求。很多"密钥泄露"事故不是被攻击，而是提交代码时顺手把.env推到了公开仓库，几分钟内就会被扫描到并产生异常调用。把.env.example提交上去、把真实密钥留在本地或部署机的环境变量里，是成本最低的防护。

2.3 API 凭据与接口地址配置

配置项统一放.env，代码只认环境变量。下面这份模板可以作为起点，单价部分留空或填 0，等你确认自己使用的服务实际计费方式后再填：

# 文件路径：.env.example # 密钥：只放在本地或部署机环境变量，禁止提交到仓库 LLM_API_KEY=sk-your-key-here # 接口地址：填你所使用服务的 OpenAI 兼容端点 LLM_BASE_URL=https://api.openai.com/v1 # 模型名：按实际可用模型填写 LLM_MODEL=gpt-4o-mini # 调用参数 LLM_TIMEOUT=60 LLM_MAX_RETRIES=3 LLM_CONCURRENCY=3 LLM_TEMPERATURE=0.7 # 成本控制：单价按你实际使用的服务填写，本文不写死具体价格 MAX_BUDGET=5.0 PRICE_IN_PER_1K=0 PRICE_OUT_PER_1K=0 # 质量下限 MIN_OUTPUT_LENGTH=120

需要强调的是LLM_BASE_URL这一项：只要目标服务兼容 OpenAI 的chat/completions协议，改这一行就能切换，业务代码完全不用动。这正是 1.2 节说的"调用层封装"的价值——把变化点收敛到一个配置文件里，后续换服务商只改一行。

3. 核心模块拆解：配置、调用、重试、限流

3.1 配置层：把密钥和业务参数彻底分离

配置层的目标只有一个：代码里不出现任何可能随环境变化的值。下面这份config.py做了三件事：轻量读取.env（不额外引入python-dotenv）、对环境变量做类型转换并给出明确报错、把所有参数收敛到一个不可变的AppConfig对象里。为什么用dataclass而不是字典？因为字典拼错 key 只会在运行时静默返回None，而dataclass拼错属性名会直接AttributeError，错误暴露得更早。

# 文件路径：config.py import os from dataclasses import dataclass from pathlib import Path def _load_dotenv(path: Path = Path(".env")) -> None: """轻量读取 .env，已存在的环境变量优先，不覆盖。""" if not path.exists(): return for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") os.environ.setdefault(key.strip(), value.strip()) def _env_int(name: str, default: int) -> int: raw = os.getenv(name) if raw is None or raw.strip() == "": return default try: return int(raw) except ValueError as exc: raise ValueError(f"环境变量 {name} 必须是整数，当前值：{raw!r}") from exc def _env_float(name: str, default: float) -> float: raw = os.getenv(name) if raw is None or raw.strip() == "": return default try: return float(raw) except ValueError as exc: raise ValueError(f"环境变量 {name} 必须是数字，当前值：{raw!r}") from exc @dataclass(frozen=True) class AppConfig: api_key: str base_url: str model: str timeout: int max_retries: int concurrency: int temperature: float max_budget: float price_in_per_1k: float price_out_per_1k: float min_output_length: int def load_config() -> AppConfig: _load_dotenv() api_key = (os.getenv("LLM_API_KEY") or "").strip() if not api_key: raise ValueError("缺少 LLM_API_KEY，请在 .env 或系统环境变量中配置") return AppConfig( api_key=api_key, base_url=(os.getenv("LLM_BASE_URL") or "https://api.openai.com/v1").strip(), model=(os.getenv("LLM_MODEL") or "gpt-4o-mini").strip(), timeout=_env_int("LLM_TIMEOUT", 60), max_retries=_env_int("LLM_MAX_RETRIES", 3), concurrency=_env_int("LLM_CONCURRENCY", 3), temperature=_env_float("LLM_TEMPERATURE", 0.7), max_budget=_env_float("MAX_BUDGET", 5.0), price_in_per_1k=_env_float("PRICE_IN_PER_1K", 0.0), price_out_per_1k=_env_float("PRICE_OUT_PER_1K", 0.0), min_output_length=_env_int("MIN_OUTPUT_LENGTH", 120), )

这里的几个细节值得说一下。第一，_load_dotenv用setdefault而不是直接赋值，是为了让 CI/CD 里注入的环境变量优先于本地.env，部署时不会因为文件残留导致密钥被覆盖成测试值。第二，frozen=True让配置对象不可修改，杜绝了"某个函数中途把 concurrency 改成 50"这种极难排查的隐式副作用。第三，所有类型转换失败都抛出带原始值的异常，你在日志里一眼就能看出是哪个变量填错了，而不是面对一句干巴巴的ValueError。

3.2 调用层：统一封装模型 SDK

调用层要解决的是"屏蔽差异 + 统一返回值"。不管底层是哪个服务，上层拿到的都应该是同一种结构：内容、输入 token、输出 token、耗时。这样后面的统计、校验、成本计算都不用关心调用细节。

先来看一个最直接的封装版本：

# 文件路径：llm_client.py import logging import random import time from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI, RateLimitError from config import AppConfig logger = logging.getLogger(__name__) # 可重试异常：网络问题、超时、限流、5xx RETRYABLE = (APITimeoutError, APIConnectionError, RateLimitError) class LLMClient: """对模型调用的最小封装，向上只暴露 chat 方法。""" def __init__(self, config: AppConfig) -> None: self.config = config # max_retries=0：重试策略由本类控制，避免 SDK 内部重试叠加 self._client = OpenAI( api_key=config.api_key, base_url=config.base_url, timeout=config.timeout, max_retries=0, ) def chat(self, system_prompt: str, user_prompt: str, temperature: float | None = None) -> dict: temperature = self.config.temperature if temperature is None else temperature last_error: Exception | None = None for attempt in range(1, self.config.max_retries + 1): started = time.time() try: resp = self._client.chat.completions.create( model=self.config.model, temperature=temperature, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], ) elapsed = round(time.time() - started, 2) content = (resp.choices[0].message.content or "").strip() usage = getattr(resp, "usage", None) return { "content": content, "prompt_tokens": getattr(usage, "prompt_tokens", 0) or 0, "completion_tokens": getattr(usage, "completion_tokens", 0) or 0, "elapsed": elapsed, } except RETRYABLE as exc: last_error = exc # 指数退避 + 随机抖动，避免多线程同时重试形成尖峰 wait = min(2 ** attempt + random.random(), 20.0) logger.warning( "第 %s/%s 次调用失败（%s），%.1f 秒后重试", attempt, self.config.max_retries, type(exc).__name__, wait, ) time.sleep(wait) except APIStatusError as exc: # 4xx 属于参数或权限问题，重试无意义，直接抛 logger.error("接口返回不可重试状态码：%s，响应：%s", exc.status_code, exc.message) raise raise RuntimeError(f"模型调用连续失败 {self.config.max_retries} 次") from last_error

这段代码里有三个容易踩坑但很少被讲清楚的点。第一，max_retries=0是刻意设置的：SDK 自己默认会重试两次，如果你再在外面套一层重试，实际请求次数会变成乘法关系，一次任务可能打出九次请求，既浪费额度又拖慢整体速度。第二，区分可重试和不可重试异常：超时、连接失败、限流（429）、服务端 5xx 值得重试，而 400 参数错误、401 密钥错误、404 模型不存在，重试一百次也不会成功，反而让日志被垃圾信息淹没。第三，退避时间加了随机抖动，因为并发场景下所有线程同时失败后再同时重试，会形成新的流量尖峰，抖动可以把这些请求摊开。

返回结构里带elapsed是刻意的，它让你在复盘时能回答"是模型慢还是我的代码慢"这个问题。如果elapsed稳定在 2 秒但整轮耗时 10 分钟，说明瓶颈在并发度而不是模型；如果elapsed波动在 2 到 30 秒之间，就要考虑是不是长文本输出导致的不稳定延。

3.3 重试策略的边界：什么时候该放弃

重试不是越多越好。上面的实现里，总耗时上限是max_retries × (timeout + 退避时间)，按 3 次、60 秒超时、退避 2/4/8 秒计算，单条最坏情况要占用接近 3 分钟。如果并发度是 3，一轮 100 条任务的最坏耗时会被拉到不可接受的程度。

实践中我建议按任务的重要程度分档：核心任务（当次要交付的）保留 3 次重试，允许慢；批量试探任务（先跑 20 条看效果）只给 1 到 2 次，快速失败比慢慢成功更有价值。这条规则可以用一个简单的参数控制，不需要改代码结构：

# 核心片段：按任务重要程度覆盖重试次数 def chat_with_level(client: LLMClient, system_prompt: str, user_prompt: str, level: str = "normal") -> dict: original = client.config.max_retries try: # frozen dataclass 不能直接改，这里用临时对象演示思路 if level == "probe": client.config = type(original) # 占位，实际见下方说明 return client.chat(system_prompt, user_prompt) finally: pass

上面这段是示意，不要直接抄——frozen=True的AppConfig本来就禁止修改，这正是它的价值。正确做法是把重试次数作为chat的参数传入，而不是去改全局配置：

# 文件路径：llm_client.py（替换 chat 方法签名中相关部分） def chat(self, system_prompt: str, user_prompt: str, temperature: float | None = None, max_retries: int | None = None) -> dict: retries = self.config.max_retries if max_retries is None else max_retries # 后续循环使用 retries 而不是 self.config.max_retries

这样调用方可以按场景传max_retries=1，既不污染全局配置，也不需要维护多份客户端实例。这个改动很小，但它把"重试次数"从系统级参数变成了任务级参数，是后面做分级任务调度的基础。

3.4 并发与限流：别把额度打爆

并发度的选择取决于两件事：服务端的速率限制、你自己的预算。很多人一上来把并发开到 20，结果前 30 秒确实爽，之后全是 429，整体耗时反而比并发 3 更长，因为重试把队列堵住了。

一个经过验证的经验值：先把LLM_CONCURRENCY设为 3，跑 20 条任务，观察日志里 429 出现的比例。如果一次都没出现，可以上调到 5；如果超过 10% 的任务触发了 429，说明当前并发已经超过服务端配额，应该下调。这个调参过程最多花两轮，比凭感觉设一个高并发然后反复改代码要省时间得多。

另外注意并发和写入的关系。多线程同时往同一个文件追加写，如果不做控制会出现内容交错甚至损坏。本文的调度层采用"多线程只做调用、主线程统一写入"的模式，写操作天然串行，不需要加锁，这是比threading.Lock更简单也更安全的做法，第 4 节会给出完整实现。

4. 完整实战：批量任务流水线

4.1 输入数据结构与任务幂等设计

任务输入用 JSON 数组，每条任务是一个对象。之所以不用 CSV，是因为任务里往往带数组字段（比如关键词列表），CSV 里表达起来很别扭。

[ { "topic": "Python 虚拟环境的作用", "keywords": ["venv", "依赖隔离", "版本管理"], "tone": "教学", "min_length": 400 }, { "topic": "JSONL 与 JSON 的区别", "keywords": ["流式读取", "追加写入", "断点续跑"], "tone": "教学", "min_length": 400 } ]

关键设计是任务幂等：每条任务需要一个稳定 ID，由任务内容计算得出，而不是随机生成或者用数组下标。原因很实际——你在中途往tasks.json里插了一条新任务，所有下标都错位了，用下标做 ID 会导致断点续跑时跳过错误的任务。用内容哈希就没这个问题：

# 文件路径：pipeline.py（片段） import hashlib import json def task_id(task: dict) -> str: """由任务内容生成稳定 ID，内容相同则 ID 相同。""" raw = json.dumps(task, ensure_ascii=False, sort_keys=True) return hashlib.md5(raw.encode("utf-8")).hexdigest()[:12]

sort_keys=True不能省，否则字典键顺序不同会算出不同哈希，幂等就失效了。截取 12 位是为了日志好读，碰撞概率在单机任务量级下可以忽略。

4.2 调度层：并发调用、串行落盘

调度层的核心约束只有一条：调用可以并发，写入必须串行。下面这份pipeline.py同时实现了断点续跑、失败记录和统计累计。

# 文件路径：pipeline.py import hashlib import json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from config import AppConfig from llm_client import LLMClient logger = logging.getLogger(__name__) SYSTEM_PROMPT = ( "你是一名技术内容作者。请围绕给定主题输出结构清晰的中文说明。" "要求：不使用夸张宣传语，不虚构数据和经历，涉及操作步骤时给出可执行的说明。" ) def task_id(task: dict) -> str: raw = json.dumps(task, ensure_ascii=False, sort_keys=True) return hashlib.md5(raw.encode("utf-8")).hexdigest()[:12] def build_prompt(task: dict) -> str: keywords = "、".join(task.get("keywords", [])) return ( f"主题：{task['topic']}\n" f"需要覆盖的关键词：{keywords or '无'}\n" f"语气风格：{task.get('tone', '教学')}\n" f"目标长度：不少于 {task.get('min_length', 300)} 字\n" "输出要求：分成 2 到 4 个小段落，每段说明一个要点。" ) def run_one(task: dict, client: LLMClient, config: AppConfig) -> dict: result = client.chat(SYSTEM_PROMPT, build_prompt(task)) content = result["content"] if len(content) < config.min_output_length: raise ValueError(f"输出过短（{len(content)} 字），疑似模型截断或拒答") return { "task_id": task_id(task), "topic": task["topic"], "output": content, "prompt_tokens": result["prompt_tokens"], "completion_tokens": result["completion_tokens"], "elapsed": result["elapsed"], } def load_done_ids(results_path: Path) -> set[str]: """读取已有结果，得到已完成任务 ID 集合。""" done: set[str] = set() if not results_path.exists(): return done with results_path.open("r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: done.add(json.loads(line)["task_id"]) except (json.JSONDecodeError, KeyError): logger.warning("跳过一行无法解析的结果记录") return done def run_pipeline(tasks: list[dict], client: LLMClient, config: AppConfig, results_path: Path, guard=None) -> dict: results_path.parent.mkdir(parents=True, exist_ok=True) done = load_done_ids(results_path) pending = [t for t in tasks if task_id(t) not in done] logger.info("任务总数 %s，已完成 %s，待处理 %s", len(tasks), len(done), len(pending)) stats = {"ok": 0, "fail": 0, "prompt_tokens": 0, "completion_tokens": 0} with results_path.open("a", encoding="utf-8") as f, \ ThreadPoolExecutor(max_workers=max(config.concurrency, 1)) as pool: future_map = {pool.submit(run_one, task, client, config): task for task in pending} for future in as_completed(future_map): task = future_map[future] record = None try: record = future.result() stats["ok"] += 1 stats["prompt_tokens"] += record["prompt_tokens"] stats["completion_tokens"] += record["completion_tokens"] except Exception as exc: stats["fail"] += 1 record = {"task_id": task_id(task), "topic": task.get("topic"), "error": str(exc)} logger.error("任务失败 %s：%s", record["task_id"], exc) # 只有主线程写文件，天然串行，不需要额外加锁 f.write(json.dumps(record, ensure_ascii=False) + "\n") f.flush() if guard is not None and "prompt_tokens" in record: guard.add(record["prompt_tokens"], record["completion_tokens"]) return stats

有几个实现细节值得单独说明。第一，f.flush()每写一条就刷一次盘，这是为了保证程序被强制终止时已经完成的任务不会丢，代价是轻微的 IO 开销，在每秒几条的量级完全可以接受。第二，失败的任务也写进结果文件，但记录里是error字段而不是output——这一点很重要，否则下次续跑时它会因为task_id已存在而被跳过，你永远不知道它失败了。第三，如果你确实想重试失败任务，可以额外写一个脚本，把含error的行挑出来重新入队，这比在主流程里做复杂的状态机要清晰得多。

4.3 成本熔断：让程序自己踩刹车

成本控制不能靠人盯着。下面的BudgetGuard在每条结果写完后累计成本，超过预算立刻抛异常终止整轮，剩下的任务不会被提交到线程池之外的新批次里（线程池内部已提交的会继续跑完，这是可接受的损失）。

# 文件路径：cost.py class BudgetExceeded(RuntimeError): """预算超限，用于中断整轮任务。""" class BudgetGuard: """按 token 累计成本，超过上限抛异常。单价单位：元 / 千 token。""" def __init__(self, max_budget: float, price_in_per_1k: float, price_out_per_1k: float) -> None: self.max_budget = max_budget self.price_in = price_in_per_1k / 1000.0 self.price_out = price_out_per_1k / 1000.0 self.used = 0.0 def add(self, prompt_tokens: int, completion_tokens: int) -> float: self.used += prompt_tokens * self.price_in + completion_tokens * self.price_out if self.used > self.max_budget: raise BudgetExceeded( f"累计成本 {self.used:.4f} 已超过预算 {self.max_budget:.4f}，本轮中断" ) return self.used

单价从配置里读、默认填 0，是刻意为之：不同服务、不同模型的计费方式差异很大，把具体价格写死在代码里既容易过时，也容易让读者误以为那是当前真实价格。你只需要在.env里把PRICE_IN_PER_1K和PRICE_OUT_PER_1K按自己实际使用的服务填写，成本统计就能正常工作。填 0 时这个模块退化为纯 token 统计，不会误触发熔断，也不会影响主流程。

max_budget建议按"一轮最大可承受损失"来设，而不是按月预算。比如一轮 100 条任务，预估成本 2 元，那把它设成 5 元就很合适，既能容忍少量重试，又能在出现异常长文本时及时刹车。

4.4 入口文件与运行验证

把上面几块拼起来，入口文件保持极简，只负责组装和输出统计：

# 文件路径：main.py import json import logging from pathlib import Path from config import load_config from cost import BudgetExceeded, BudgetGuard from llm_client import LLMClient from pipeline import run_pipeline def main() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s - %(message)s", ) config = load_config() tasks = json.loads(Path("tasks.json").read_text(encoding="utf-8")) client = LLMClient(config) guard = BudgetGuard( max_budget=config.max_budget, price_in_per_1k=config.price_in_per_1k, price_out_per_1k=config.price_out_per_1k, ) try: stats = run_pipeline(tasks, client, config, Path("output/results.jsonl"), guard=guard) except BudgetExceeded as exc: logging.error("已触发预算熔断：%s", exc) return logging.info( "本轮结束：成功 %s，失败 %s，prompt_tokens=%s，completion_tokens=%s，估算成本 %.4f", stats["ok"], stats["fail"], stats["prompt_tokens"], stats["completion_tokens"], guard.used, ) if __name__ == "__main__": main()

运行命令：

python main.py

预期输出大致如下（时间戳和数值随实际调用变化）：

2024-06-01 10:12:03,115 INFO pipeline - 任务总数 2，已完成 0，待处理 2 2024-06-01 10:12:09,882 INFO __main__ - 本轮结束：成功 2，失败 0，prompt_tokens=412，completion_tokens=903，估算成本 0.0000

再次执行同一条命令，你会看到"已完成 2，待处理 0"，说明断点续跑生效了。这一步建议一定要自己验证一遍，因为它是后续所有增量任务的基础——很多人写完流水线第一次中断就发现结果文件被覆盖，一半工作白做。

4.5 结果查看与增量补跑

结果文件是 JSONL，一行一个对象，可以直接用命令行抽样查看：

# 查看第一条结果的正文 head -n 1 output/results.jsonl | python -c "import sys,json;print(json.loads(sys.stdin.read())['output'])" # 统计成功与失败条数 python - <<'PY' import json ok = fail = 0 for line in open("output/results.jsonl", encoding="utf-8"): line = line.strip() if not line: continue rec = json.loads(line) if "error" in rec: fail += 1 else: ok += 1 print("成功", ok, "失败", fail) PY

当你想补跑失败任务时，最省事的做法是把失败记录导出成一个新的tasks.json：只保留topic相关字段，去掉task_id，然后正常执行python main.py。因为失败记录没有被算进"已完成集合"——注意，前面的load_done_ids会把所有行都当作已完成，包括失败行。如果你希望失败任务能自动重跑，把load_done_ids改成只收集不含error的行即可：

# 文件路径：pipeline.py（load_done_ids 中一行改动） rec = json.loads(line) if "error" not in rec: done.add(rec["task_id"])

这是一个需要你根据业务决定的选择：自动重跑方便，但会陷入"某条任务永远失败、每轮都重试"的死循环。我的建议是自动重跑 + 失败次数上限，用一个小字典记录每个task_id的失败次数，超过 2 次就不再纳入待处理列表。

5. 质量校验与人工复核的边界

5.1 三类可以自动化的校验

流水线跑通之后，下一个问题是"生成的东西能不能用"。有三类校验完全可以用代码做，成本几乎为零：

长度与结构：输出过短通常意味着模型拒答或截断，输出过长可能是没有遵守指令。用min_output_length兜住下限，上限按业务需要设。

关键词覆盖：如果任务里指定了必须出现的关键词，直接做子串匹配，算出覆盖率。覆盖率低于 60% 的记录单独标记，优先人工看。

重复度：批量生成最大的隐患是内容同质化。用字符级 n-gram 的 Jaccard 相似度做两两比较，相似度高于阈值（比如 0.6）的标记为疑似重复。这个方法不需要任何模型，几十行代码就能跑。

# 文件路径：quality.py import json import re from pathlib import Path def keyword_coverage(text: str, keywords: list[str]) -> float: if not keywords: return 1.0 hit = sum(1 for k in keywords if k and k in text) return hit / len(keywords) def ngram_set(text: str, n: int = 4) -> set[str]: compact = re.sub(r"\s+", "", text) if len(compact) < n: return {compact} if compact else set() return {compact[i:i + n] for i in range(len(compact) - n + 1)} def jaccard(a: str, b: str, n: int = 4) -> float: sa, sb = ngram_set(a, n), ngram_set(b, n) if not sa or not sb: return 0.0 return len(sa & sb) / len(sa | sb) def review(results_path: Path, min_coverage: float = 0.6, dup_threshold: float = 0.6) -> list[dict]: records = [] for line in results_path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue rec = json.loads(line) if "error" in rec: continue records.append(rec) issues = [] accepted: list[dict] = [] for rec in records: reasons = [] keywords = rec.get("keywords") or [] if keyword_coverage(rec["output"], keywords) < min_coverage: reasons.append("关键词覆盖不足") for prev in accepted: if jaccard(rec["output"], prev["output"]) >= dup_threshold: reasons.append(f"与 {prev['task_id']} 高度相似") break if reasons: issues.append({"task_id": rec["task_id"], "reasons": reasons}) else: accepted.append(rec) return issues if __name__ == "__main__": for item in review(Path("output/results.jsonl")): print(item["task_id"], "|", "；".join(item["reasons"]))

注意review里对重复度的处理是"只和已通过校验的记录比较"，这样一条重复内容只会被标记一次，不会出现连锁误报。如果任务表里带keywords字段，记得在run_one的返回值里把它带出来，否则review拿不到关键词。

5.2 必须留给人工判断的部分

有三件事不建议交给代码自动决定：事实准确性、合规性、是否发布。事实准确性方面，模型可能把不存在的方法名、参数名写成确定事实，这类错误机器很难识别，只能靠人工抽查；合规性方面，是否涉及他人版权素材、是否需要在发布时标注 AI 生成，属于业务和平台规则范畴，代码只能做关键词黑名单初筛，最终判断必须有人；是否发布更是如此，自动化发布的风险远高于自动化生成，一旦批量发出问题内容，召回成本极高。

我通常的做法是：机器筛掉明显不合格的（过短、重复、关键词缺失），剩下的按 10% 比例随机抽样人工看，抽样合格率低于 80% 就停下来改提示词或者改任务拆分方式，而不是继续扩量。这个流程听起来保守，但它能保证你在放大规模之前先知道真实的质量水位。

6. 定时调度与部署

6.1 用 cron 做最简单的定时

流水线验证通过后，最简单的定时方式就是系统自带的 cron，不需要引入任何调度框架：

# 编辑当前用户的定时任务 crontab -e # 每天 08:30 执行一轮，日志追加写入 30 8 * * * cd /opt/ai-pipeline && /opt/ai-pipeline/.venv/bin/python main.py >> logs/run.log 2>&1

这里有几个细节容易出问题。第一，cd到项目目录是必须的，因为代码里读的是相对路径tasks.json和output/，cron 的默认工作目录是用户家目录，不cd会直接报文件不存在。第二，用虚拟环境里的绝对路径/opt/ai-pipeline/.venv/bin/python，而不是裸python，因为 cron 的PATH和你在终端里的不一样。第三，2>&1把标准错误重定向到日志，否则报错信息会被 cron 通过邮件发出去（多数机器上邮件服务没配，等于直接丢失）。

Windows 上可以用"任务计划程序"配置同样的命令，触发器设为每天固定时间，操作里填虚拟环境python.exe的完整路径，起始位置填项目目录。逻辑和 cron 一致，只是配置界面不同。

6.2 日志规范与轮转

日志至少要包含四类信息：任务开始时的总量统计、单条失败的原因和 ID、预算熔断事件、结束时的汇总。上面代码里的logger.info/logger.error已经覆盖了这些，但还缺轮转——如果每天跑几轮、每轮几十条，日志文件几个月就会涨到几百 MB。

最省事的方案是用系统的logrotate，配置一个/etc/logrotate.d/ai-pipeline：

# 文件路径：/etc/logrotate.d/ai-pipeline /opt/ai-pipeline/logs/run.log { daily rotate 14 compress missingok notifempty copytruncate }

copytruncate这一项在"进程持续往文件追加写"的场景下很关键，它保证轮转时不需要重启进程。rotate 14表示保留两周，出问题时回溯两周的日志通常够了。

6.3 部署时的三条硬性要求

第一，密钥只走环境变量。部署机上不要留.env文件，改用 systemd 的Environment=或者容器编排的密钥管理，避免配置文件被误打包、误备份。

第二，先在小批量上验证。部署到新机器后先手动跑一次只含 2 条任务的tasks.json，确认网络、密钥、目录权限、写入路径全部正常，再交给 cron。直接上全量任务，一旦路径写错，你会得到一堆失败记录还要手工清理。

第三，给进程加超时保护。cron 启动的任务如果因为网络问题卡住，可能一直挂在那里，下一轮又启动一个新进程，最后同一份结果文件被多个进程同时写。用timeout命令包一层最简单：

# 最多运行 30 分钟，超时后强制结束 30 8 * * * cd /opt/ai-pipeline && timeout 1800 /opt/ai-pipeline/.venv/bin/python main.py >> logs/run.log 2>&1

7. 常见问题与排查思路

下面这张表覆盖了我在实际使用中最常遇到的几类问题，按"现象 → 原因 → 思路"组织，方便你直接对照日志定位。

排查时有一个通用顺序值得记住：先看日志最后 20 行，再看结果文件最后 1 条记录，最后才看代码。绝大多数问题在这两步就能定位——日志会告诉你异常类型，结果文件会告诉你卡在哪条任务上，比通读代码快得多。

8. 最佳实践与合规边界

8.1 工程层面的六条习惯

幂等优先于状态机。用内容哈希做任务 ID，让"跑过一次"这件事可以被文件本身证明，比维护一个state.json记录进度要可靠得多，因为状态文件本身也会损坏。

失败要留痕，不要静默跳过。失败任务写进结果文件并带error字段，是本流水线里最重要的一个设计。静默跳过的后果是你以为跑完了 100 条，实际上只有 60 条成功，而你没有证据。

配置和代码分离到极致。模型名、并发数、预算、单价全部走环境变量。这不是为了好看，而是为了让你在不改代码、不重新部署的情况下调整运行参数，尤其在成本失控时能立刻降并发、降预算。

先小批量再放大。任何提示词改动、模型切换、并发调整，都先在 5 到 20 条的小样本上验证，看成本、耗时、抽检合格率三项数据，再决定是否跑全量。

给每个关键指标留观测点。至少要有：本轮任务数、成功数、失败数、token 总量、估算成本、平均单条耗时。这六个数字能让"今天为什么慢"这类问题在两分钟内回答。

不要在代码里写死具体服务商的价格和限制。这些信息变化快，写死之后文章和代码都会迅速过时，读者照抄还会误判成本。用配置项表达，是唯一稳妥的做法。

8.2 合规与风险边界

这一节必须单独讲，因为自动化批量生成内容涉及的边界比纯技术问题更多。以下几条是我认为必须守住的：

不承诺收益，不把工具包装成赚钱机器