Rytr 接进自己的批量流程、密钥和失败请求怎么兜底？

把 Rytr 接进自己的批量流程，跑不跑得稳，通常不取决于单次调用写得多漂亮，而取决于三件事：密钥有没有从代码里剥离、失败请求有没有被分类处理、日志能不能定位到具体是哪一条输入出错。前两件决定一次网络抖动会不会让整批任务中断，第三件决定出问题时你是几分钟缩小范围，还是重新跑一遍碰运气。下面给的是通用骨架，不绑定具体接口路径和字段名，实际参数以官方文档为准，你需要在接入前对照官方说明替换掉占位部分。

密钥放环境变量、不随脚本进版本库只是底线。批量调用的稳定性更多来自超时、重试上限和失败分类：鉴权失败和参数错误应当立即停整批，超时、429 与 5xx 才值得有限重试。日志至少要记到能还原是哪一条输入触发失败。被限流或配额触顶时，把失败条目落盘而不是丢弃，恢复后按状态过滤重跑。是否要人工接管，取决于同一类失败是否持续出现。

把密钥从代码里挪到环境变量

密钥直接写在脚本里，最容易在复制文件、打包备份、提交代码时被动泄露，而且轮换时要改代码、重新发布。通常的做法是从环境变量读取，代码里只保留变量名。本地开发把变量写进项目根目录的 .env，并把该文件加入 .gitignore；运行环境则通过容器编排的 env 字段、CI 的 secret 变量或服务器的 systemd EnvironmentFile 注入。两端只换注入位置，不改脚本。

import os

API_KEY = os.environ.get("RYTR_API_KEY")
if not API_KEY:
    raise SystemExit("缺少 RYTR_API_KEY，请检查本地 .env 或运行环境变量")

# 占位地址，替换为官方文档给出的接口地址
ENDPOINT = os.environ.get("RYTR_ENDPOINT", "https://api.example.com/v1/generate")

验证方式很简单：本地临时把 .env 改名，脚本应当以明确的报错退出，而不是带着空密钥继续跑完整批。这个检查值得保留在批量任务的启动阶段，避免把一整批请求都浪费在鉴权失败上。

用一条最小请求验证鉴权是否通过

批量脚本一上来跑几百条，一旦全错，很难分清是密钥失效、请求体格式不对，还是某几条输入触发了内容策略。建议先用一条固定输入单独调用，把状态码和响应体前若干字符打印出来，确认链路通了再放开批量。

import requests

def call_once(payload, timeout=(5, 20)):
    return requests.post(
        ENDPOINT,
        json=payload,
        headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
        timeout=timeout,
    )

resp = call_once({"input": "这是一条用于验证鉴权的测试输入"})
print(resp.status_code, resp.text[:200])

检查点分三类。状态码为 2xx 且返回体里能找到非空的文本结果，说明鉴权和请求格式都通过，可以进入批量。401 或 403 属于鉴权问题，重试没有意义，先确认密钥是否复制完整、环境变量是否注入到了当前进程。400 一般指向请求体字段缺失或类型不对，对照官方文档逐项核对。429 说明请求本身没问题，只是频率或配额触顶，这类才有重试价值。把这三种情况在日志里用不同状态区分开，后面排查会省很多时间。

给批量任务加超时与重试，区分两类失败

超时必须显式设置，不要依赖默认值。建议连接超时和读取超时分开，读取超时给得比连接超时宽一些，因为生成类接口的耗时本来就长于建连。重试只针对可重试失败，次数上限通常设 3 次左右就够，退避用 1 秒、2 秒、4 秒这类递增节奏，避免在对方已经过载时继续加压。

import time

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

def call_with_retry(payload, max_retry=3):
    for attempt in range(max_retry + 1):
        try:
            resp = call_once(payload)
        except (requests.Timeout, requests.ConnectionError):
            if attempt == max_retry:
                return None, "network"
            time.sleep(2 ** attempt)
            continue

        if resp.status_code in RETRYABLE_STATUS:
            if attempt == max_retry:
                return None, f"retryable_{resp.status_code}"
            time.sleep(2 ** attempt)
            continue
        if resp.status_code in (401, 403):
            return None, "auth"      # 整批停止，先修密钥
        if resp.status_code >= 400:
            return None, "fatal"     # 参数或内容问题，不重试
        return resp.json(), "ok"
    return None, "unknown"

判断条件可以记成一句话：网络层错误、429、5xx 归为可重试；鉴权失败、参数错误、内容被拒归为不可重试。不可重试的错误如果连续出现，说明问题在配置或输入侧，继续重试只会放大错误量。是否对 5xx 之外的错误做重试，需要结合环境确认，不同接口的行为可能有差异。

记录每次请求的输入语种与返回长度

批量任务出问题时，最有价值的不是总成功率，而是能立刻定位到是哪一条输入触发的。建议每条请求追加一行结构化日志，字段至少包含下面这些。

任务号：批次 id 加序号，重跑时用来去重和定位
输入语种：便于发现某一语种的输入集中失败
提交时间：用于和限流窗口、配额周期对照
耗时：毫秒级即可，异常偏长往往先于超时出现
返回长度：字符数，长度为 0 或明显偏短通常意味着空返回或被截断
状态：ok、network、retryable_429、auth、fatal 之一

{"job_id": "batch-0417-013", "lang": "zh-CN", "submitted_at": "<ISO时间>", "elapsed_ms": 1840, "resp_len": 312, "status": "ok"}

把日志写成 JSONL 追加文件，比打印到控制台更好检索。返回长度这个字段容易被忽略，但它能在不看正文的情况下，快速区分「请求成功但结果为空」和「请求直接失败」这两种完全不同的情况。

请求被限流或配额触顶时的降级写法

被拒绝不等于数据可以丢。批量任务应当在单条失败时立即把该条目连同任务号、入参和状态写入失败队列文件，主流程继续处理剩余条目；整批结束后再统一看待重跑。这样即使中途被限流打断，也不会出现「跑了一半不知道缺哪些」的局面。

# 主流程只负责分发与落盘
for item in items:
    result, status = call_with_retry(build_payload(item))
    if status != "ok":
        append_jsonl("failed.jsonl", {"job_id": item["job_id"], "input": item["input"],
                                    "lang": item["lang"], "status": status})
        if status == "auth":
            break   # 鉴权失败，继续跑没有意义

重跑范围要控制住，不要整批重来。读取失败队列，按状态过滤，只重跑 retryable 和 network 两类；auth 和 fatal 需要先修正配置或修改输入，直接重跑大概率还是同样的结果。重跑前按任务号去重，避免同一条成功的结果被覆盖。

人工接管的条件建议提前定死，通常是这几条中的任意一条：鉴权类失败持续出现、可重试失败连续多轮没有改善、失败队列在补跑后仍然增长。触发后停下消费，先看日志里的输入语种和返回长度分布，再决定是调整调用节奏、改输入，还是联系服务方确认配额状态。