综合小项目：用 Python 构建可恢复的 AI 翻译审阅器

前面的系列已经介绍了 API 调用、上下文组织、结构化输出、错误处理、评估和人工确认。本篇把这些能力组合成一个小项目：读取待翻译文本，逐条调用模型，校验结构化结果，并把每条成功结果立即写入文件。程序中途退出后可以继续处理未完成的项目，人工审阅也不会被模型调用自动跳过。重点不是做一个完整翻译平台，而是观察如何把一次模型调用变成可恢复、可检查的业务流程。

先定义数据和边界

输入文件使用 JSONL，每行一个对象。为了便于复跑，给每条文本一个稳定的 id：

1 2	{"id":"welcome","source":"欢迎使用我们的产品。"} {"id":"timeout","source":"请求超时时，请稍后重试。"}

模型只负责提出译文和备注，程序负责读取文件、检查字段、保存进度。它不能直接发布译文，也不能修改原始输入。输出同样使用 JSONL，成功记录包含 id、translation 和 note；失败记录包含 id 和错误类型。这样，人工可以先审阅输出，再决定是否进入后续系统。

准备环境和密钥

使用虚拟环境安装 OpenAI 官方 Python SDK：

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

密钥只能从环境变量读取，不要写进脚本、输入文件或结果文件。下面使用 Responses API 的 responses.create 和 output_text；如果使用其他服务，应先查该服务的官方文档确认客户端和字段是否一致。

编写最小流水线

创建 translate_review.py：

import json
import os
import sys
from pathlib import Path

from openai import OpenAI


def load_done(path: Path) -> set[str]:
    if not path.exists():
        return set()
    done = set()
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        item = json.loads(line)
        if item.get("status") == "ok":
            done.add(item["id"])
    return done


def translate(client: OpenAI, source: str) -> dict:
    response = client.responses.create(
        model=os.environ["MODEL_NAME"],
        instructions=(
            "你是中英翻译助手。把输入翻译成自然、准确的中文。"
            '只返回 JSON：{"translation":"译文","note":"必要时说明歧义，否则为空字符串"}。'
            "不要添加 Markdown。"
        ),
        input=source,
    )
    result = json.loads(response.output_text)
    if not isinstance(result.get("translation"), str) or not result["translation"].strip():
        raise ValueError("translation 不是非空字符串")
    if not isinstance(result.get("note"), str):
        raise ValueError("note 不是字符串")
    return result


def main() -> None:
    if len(sys.argv) != 3:
        raise SystemExit("用法：python translate_review.py input.jsonl output.jsonl")

    input_path, output_path = map(Path, sys.argv[1:])
    done = load_done(output_path)
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    with input_path.open(encoding="utf-8") as source_file, output_path.open(
        "a", encoding="utf-8"
    ) as result_file:
        for line_number, line in enumerate(source_file, 1):
            if not line.strip():
                continue
            item = json.loads(line)
            item_id = item.get("id")
            source = item.get("source")
            if not isinstance(item_id, str) or not isinstance(source, str) or not source:
                raise ValueError(f"第 {line_number} 行输入无效")
            if item_id in done:
                continue
            try:
                result = translate(client, source)
                record = {"id": item_id, "status": "ok", **result}
                done.add(item_id)
            except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
                record = {"id": item_id, "status": "error", "error": type(exc).__name__}
            result_file.write(json.dumps(record, ensure_ascii=False) + "\n")
            result_file.flush()
            print(item_id, record["status"])


if __name__ == "__main__":
    main()

运行命令如下：

1	python translate_review.py input.jsonl output.jsonl

程序不会伪造某个翻译结果，因为真实文本取决于模型和服务状态。可验证的行为是：每条输入最多在一次成功后被记录为 ok，输出文件会在每条记录后立即刷新；再次运行时，load_done 会跳过已经完成的 id。

为什么这样设计

load_done 是恢复能力的关键。它只把状态为 ok 的记录加入集合，因此上次运行中断后，失败项仍然会再次尝试。追加写入比最后一次性写完整文件更能保留进度，但也要求输入的 id 稳定且唯一。生产代码还应检测重复 ID，否则同一业务对象可能被误认为已经完成。

json.loads 只检查 JSON 语法，后面的 translation 和 note 检查才是业务协议校验。模型返回合法 JSON，不代表字段一定可用。当前示例把失败写入结果文件而不是静默丢弃；真实项目可以增加有限次数的指数退避，但认证失败、输入错误等问题不应无条件重试。

人工审阅应当是独立步骤。例如用下面的标准库脚本筛选需要查看的记录：

import json
from pathlib import Path

for line in Path("output.jsonl").read_text(encoding="utf-8").splitlines():
    item = json.loads(line)
    if item.get("status") == "ok":
        print(item["id"], item["translation"], item["note"], sep=" | ")

审阅者可以把批准状态写入另一份文件，而不是直接覆盖模型输出。这样既保留原始证据，也能区分“模型生成”“人工批准”和“最终发布”三个阶段。

常见问题

为什么不把所有文本一次发给模型？ 批量请求可能超出上下文窗口，也难以定位单条失败。逐条处理更容易重试和恢复；数据量很大时，可以在确认供应商批处理能力和限制后再优化。

重跑会不会产生重复结果？ 成功记录会按 id 跳过，但若输出文件被手工修改，程序无法判断修改是否可信。实际项目应保存输入版本、提示词版本和模型名，并在恢复前校验这些元数据。

能否让模型直接决定是否发布？ 不建议。模型可以提供翻译和风险备注，但发布属于业务副作用，应由程序规则或人工审批完成。尤其是法律、医疗和品牌文本，必须设定人工审阅边界。

为什么不显示一个固定的运行结果？ 模型输出、网络延迟和服务状态都不是本文可以预先保证的。应检查命令退出状态、输出文件格式、成功与失败记录数量，并用脱敏测试数据验证恢复逻辑。

小结

这个项目把结构化输出、文件校验、逐条处理、立即落盘、失败记录和人工审阅串成了一条最小流水线。模型负责生成候选译文，Python 负责定义协议、保存进度和隔离发布动作。以后替换提示词或模型时，只要保留稳定的输入 ID、结果状态和审阅记录，就能把一次性脚本逐步演进为可恢复、可审计的 AI 功能。

搜索时光笔记