免费模型重试前必须先幂等：防重复写入

免费模型请求，在添加重试之前先做到幂等

免费模型端点和免费服务器在最糟糕的地方失败：工作可能已经运行了，但响应还没收到。

声明：本文是 MonkeyCode 产品推广的一部分。我只依赖两个运营商提供的可用性声明：MonkeyCode 提供免费模型访问和免费服务器选项。

下面的工作流是一种客户端保护机制，而非提供商保证。

为什么重试不是默认选项

免费服务器的工作进程可能在请求中途被回收。

免费模型端点可能在生成后超时。

重试可能导致下游写入重复，或返回混合的 JSON 载荷。

普通超时不能告诉你提供商是否已经运行了该请求。

规则虽然无聊但有用：把请求意图、原始响应和已接受结果分开。只有当已接受结果不存在且日志显示没有部分下游副作用时，重试才是安全的。

幂等请求日志

产物是一个很小的 Python 日志模块。它把 prompt 和请求参数哈希成一个 key，追加事件到 JSONL 文件，只有当载荷解析为完整 JSON 后才原子性地写入结果。

import hashlib
import json
import os
import time
from pathlib import Path

def request_key(prompt, params):
    canonical = json.dumps({'prompt': prompt, 'params': params}, sort_keys=True)
    return hashlib.sha256(canonical.encode()).hexdigest()[:16]

class Journal:
    def __init__(self, root='.reqjournal'):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)

    def append(self, key, event, detail=None):
        line = {'ts': time.time(), 'event': event, 'detail': detail}
        with (self.root / f'{key}.jsonl').open('a') as f:
            f.write(json.dumps(line) + '\n')

    def last_event(self, key):
        path = self.root / f'{key}.jsonl'
        if not path.exists():
            return None
        return json.loads(path.read_text().strip().splitlines()[-1])['event']

    def result_path(self, key):
        return self.root / f'{key}.result.json'

def validate(raw):
    data = json.loads(raw)  # no partial JSON accepted
    if not isinstance(data, dict) or 'choices' not in data:
        raise ValueError('unexpected envelope')
    return data

def request_once(journal, key, call):
    journal.append(key, 'SENT')
    try:
        raw = call()
    except TimeoutError:
        journal.append(key, 'TIMEOUT')
        raise
    payload = validate(raw)
    tmp = journal.result_path(key).with_suffix('.tmp')
    tmp.write_text(json.dumps(payload))
    os.replace(tmp, journal.result_path(key))
    journal.append(key, 'COMPLETE')
    return payload

def request(journal, key, call, max_attempts=3):
    result = journal.result_path(key)
    if result.exists():
        journal.append(key, 'CACHE_HIT')
        return json.loads(result.read_text())

    for attempt in range(1, max_attempts + 1):
        before = journal.last_event(key)
        journal.append(key, 'ATTEMPT', {'n': attempt, 'after': before})
        try:
            return request_once(journal, key, call)
        except TimeoutError:
            if attempt == max_attempts:
                journal.append(key, 'DEAD_LETTER')
                raise
        except ValueError as exc:
            journal.append(key, 'INCOMPLETE', {'error': str(exc)})
            if attempt == max_attempts:
                raise
    raise RuntimeError('unreachable')

os.replace 是关键那一行。它只在验证成功后才把临时文件替换到位。调用者永远不会读到半写的结果。

把免费服务器选项用作可丢弃的工作进程。把日志指向你控制的存储，而不是会随工作进程消失的临时目录。免费模型访问是调用目标。

这种分离很重要，因为免费服务器可能死亡并重新出现，而日志仍然可以回答一个问题：超时前我们接受了什么？

在添加更多重试逻辑之前，先运行这些检查。

验证清单

相同 key 和相同 prompt：第二次运行返回缓存结果并记录 CACHE_HIT。

在写入 SENT 后但在 COMPLETE 前杀死进程：结果文件不存在，所以可以重试。

传入截断的 JSON：不创建结果文件，日志记录 INCOMPLETE。

传入有效的 JSON：临时文件重命名到位，所以没有调用者能看到部分结果。

决策表保持精简。

这是下游工作的最多一次（at-most-once），不是提供商侧的精确一次（exactly-once）。SENT 后的超时仍可能导致重复的提供商调用。

如果提供商按请求或 token 计费，重试仍可能产生额外费用。启用重试前请检查你的套餐。

限制与注意事项

日志的持久性取决于存储路径。如果免费服务器是临时的，请将其保存在持久存储或对象存储上。

这不验证模型内容，只验证信封结构。对于严格的响应格式，请添加 schema 或契约探测。

如果模型输出直接写入文件系统或 shell，请在执行前对这些写入进行门控。日志不会沙箱化副作用。

不适合使用的人群

你需要严格的精确一次投递。使用数据库 outbox 或提供商的幂等 key。

你的免费配额很小。重试可能浪费它；不如快速失败。

你的工作进程没有持久的地方存放日志。内存状态只和免费服务器进程一样长。

正确的使用方式

有用的版本不是重试循环；而是对已经发生的事情的记录。先加日志，然后让重试去读取它。

试一次：在 SENT 和 COMPLETE 之间杀死工作进程，然后重启并观察日志继续运行，而不是靠猜测。