综合小项目：用 Python 构建可恢复的 AI 批处理器

前面的系列已经覆盖了 API 调用、上下文、结构化输出、重试、评估和安全边界。本篇做一个新的综合练习：从 JSONL 文件逐条读取短文，让模型生成摘要，结果实时保存；程序中途退出后再次运行时跳过已经完成的项目。重点不是批量调用本身，而是如何让一个有网络依赖的 AI 任务具备可恢复、可检查的执行过程。

先定义任务和恢复规则

输入文件 items.jsonl 每行一个对象，包含稳定的 id 和 text：

1
2

{"id": "a-001", "text": "Python 的生成器按需产生值，可以减少一次性创建大列表的内存占用。"}
{"id": "a-002", "text": "HTTP 请求可能因为暂时的网络问题失败，因此应用需要设置超时并有限重试。"}

输出文件 results.jsonl 也按行保存。每条成功结果包含 id、摘要和时间；失败结果包含 id、状态和错误类型。id 是幂等键：恢复时只看它是否已经有一条成功记录，不用依赖行号。这样即使输入文件重新排序，也不会把同一条内容重复处理。

本例使用 OpenAI Python SDK 的 Responses API。不同服务的字段和返回对象可能不同，使用其他服务时应以其官方文档为准。准备环境：

python -m venv .venv
source .venv/bin/activate
python -m pip install openai
export OPENAI_API_KEY="替换为你的真实密钥"
export MODEL_NAME="替换为你可用的模型名称"

密钥只从环境变量读取，不写进脚本、输入文件或结果文件。

编写最小批处理器

新建 batcher.py。为了让状态文件在每条成功后立即更新，程序采用追加 JSONL，而不是最后统一写入：

import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path

from openai import OpenAI

INPUT = Path("items.jsonl")
OUTPUT = Path("results.jsonl")
MODEL = os.environ["MODEL_NAME"]
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])


def read_done() -> set[str]:
    done = set()
    if not OUTPUT.exists():
        return done
    for line in OUTPUT.read_text(encoding="utf-8").splitlines():
        try:
            record = json.loads(line)
        except json.JSONDecodeError:
            continue
        if record.get("status") == "ok":
            done.add(record["id"])
    return done


def summarize(text: str) -> str:
    response = client.responses.create(
        model=MODEL,
        instructions="用中文把输入压缩成不超过 60 字的准确摘要，只返回摘要正文。",
        input=text,
    )
    result = response.output_text.strip()
    if not result or len(result) > 60:
        raise ValueError("摘要为空或超过 60 字")
    return result


def write_record(record: dict) -> None:
    with OUTPUT.open("a", encoding="utf-8") as file:
        file.write(json.dumps(record, ensure_ascii=False) + "\n")
        file.flush()


def main() -> None:
    done = read_done()
    for line in INPUT.read_text(encoding="utf-8").splitlines():
        item = json.loads(line)
        item_id = item["id"]
        if item_id in done:
            print(f"跳过已完成：{item_id}")
            continue
        try:
            summary = summarize(item["text"])
        except Exception as error:
            write_record({
                "id": item_id,
                "status": "error",
                "error_type": type(error).__name__,
                "created_at": datetime.now(timezone.utc).isoformat(),
            })
            print(f"处理失败：{item_id}")
            continue
        write_record({
            "id": item_id,
            "status": "ok",
            "summary": summary,
            "created_at": datetime.now(timezone.utc).isoformat(),
        })
        done.add(item_id)
        print(f"已完成：{item_id}")
        time.sleep(0.2)


if __name__ == "__main__":
    main()

运行 python batcher.py。真实输出会受到模型、输入和网络状态影响，因此不能预先写一个“正确答案”。可以验证的是：成功记录拥有 status: ok 和 summary，再次运行会打印跳过信息；网络或校验失败会留下 status: error，但不会被当成成功结果。

代码中的三个关键边界

第一，read_done 只把成功记录加入集合。失败记录保留下来是为了审计，但重新运行时仍会尝试处理失败项目。若错误是永久性的，例如输入字段缺失，可以另设 invalid 状态，避免每次重复请求。

第二，写入动作发生在模型响应通过基本校验之后。output_text 是 SDK 提供的文本汇总，程序仍检查非空和长度；若要求多个字段，应改用结构化输出并在本地检查字段、类型和枚举值。模型返回合法文字，不代表内容一定正确，正式任务还需要抽样评估。

第三，结果采用追加模式，成功一条就落盘。进程在两次写入之间退出，最多损失当前未完成请求，不会丢掉此前已经写好的记录。生产环境可以进一步使用临时文件、文件锁或数据库唯一约束，避免多个进程同时运行造成重复处理。

给临时错误加上有限重试

当前示例把异常记录为失败，逻辑清楚但对短暂限流不够友好。重试应只包住模型请求，并设置上限和递增等待，不要把写文件也放进重试循环：

def summarize_with_retry(text: str, attempts: int = 3) -> str:
    last_error = None
    for attempt in range(attempts):
        try:
            return summarize(text)
        except Exception as error:
            last_error = error
            if attempt + 1 < attempts:
                time.sleep(2 ** attempt)
    raise RuntimeError("模型调用达到重试上限") from last_error

实际项目应根据 SDK 文档区分超时、限流、认证失败和输入错误。认证失败通常不该重试；批量任务还要限制总请求数、并发度和费用。重试成功后只写一条 ok 记录，失败尝试不要伪装成成功。

常见问题

为什么不用数组一次性保存结果？ 数组需要在最后统一写回，进程中断时可能丢失整批进度。JSONL 的追加特性适合简单的断点记录，也便于逐行检查。

输出文件中出现重复的 id 怎么办？ 单进程且每次成功后更新 done 时不会重复，但崩溃发生在写入和内存更新的边界仍需考虑。更严格的实现应使用 SQLite 的唯一索引，或启动时检测重复并拒绝继续。

输入内容会不会进入日志？ 示例只打印 id，避免把原文和可能的个人信息写入终端。发送第三方服务前还应按业务规则脱敏，并设置输入长度、请求超时和结果保留期限。

怎样不调用模型测试恢复逻辑？ 把 summarize 作为参数传入，测试时注入一个固定函数；准备全成功、部分失败、重复 ID 和损坏结果行等样例，验证跳过与重试规则。真实 API 只做少量集成测试。

小结

这个批处理器把一次模型调用扩展成了可恢复的执行闭环：稳定 ID 负责幂等识别，成功结果逐条落盘，失败状态可追踪，重新启动能够从未完成项目继续。模型负责生成候选摘要，Python 负责校验、状态和副作用边界。批量 AI 功能真正需要解决的，往往不是把循环写出来，而是让中断、重试、重复运行和错误都拥有明确且可验证的行为。

搜索时光笔记