2026年8月25日 ChatGPT Plus Pro Codex 教程：用幂等队列构建可恢复的 AI Agent 工作流

gptupcn.com｜ChatGPT Plus、Pro 与 Codex 使用参考

当 Codex 或其他 AI Agent 只处理一个文件时，“失败后重试”通常没有什么风险；一旦任务扩展到几十个仓库、数百个文件或多个外部接口，简单重跑就会产生重复提交、重复通知、状态覆盖和费用失控。真正可靠的 Agent 系统不能只追求一次执行成功，而要能够回答四个问题：任务现在处于什么状态、同一输入是否已经处理、失败后应该从哪里恢复、人工如何安全接管。

本文面向正在使用 ChatGPT Plus、ChatGPT Pro 或 Codex 的开发者，给出一个可以本地运行的 Python 示例：用 SQLite 保存队列状态，用幂等键阻止重复执行，用租约避免多个 Worker 同时领取任务，再用重试预算和人工复核状态控制风险。示例不依赖特定云厂商，适合放进个人脚本、CI 流程或小型内部工具中。

说明：ChatGPT Plus充值、Pro充值、Codex充值、订阅和续费解决的是账户与服务可用性；幂等队列解决的是工程执行可靠性。两者不应混为一谈。即使订阅状态正常，网络中断、权限变化和代码错误仍会让 Agent 失败。

一、先把“重试”改造成状态机

很多脚本只有成功与失败两个结果，类似下面这样：

for repo in repositories:
    run_agent(repo)

问题在于，程序在第 17 个仓库中断后，我们无法确认前 16 个仓库是否全部提交成功，也不知道第 17 个仓库是在修改前、提交前还是推送前失败。直接重新运行可能重复生成内容，跳过运行又可能遗漏任务。

更稳妥的做法是为每个任务定义明确状态：

状态	含义	系统动作	是否允许自动重试
pending	等待领取	Worker 可以申请租约	是
running	已领取、正在执行	记录 Worker 与租约时间	否
retry	暂时性错误	等待退避时间后重新入队	是
review	结果不确定或风险较高	暂停并等待人工核对	否
done	已完成且已验收	保存结果摘要	否
failed	超过重试预算	保留错误证据	否

状态机的关键不是表格本身，而是所有改变必须经过受控的状态转换。比如 done 任务不能被普通 Worker 再次领取，review 任务也不能因为定时器到点就自动恢复。

二、设计不会随重试变化的幂等键

幂等键用于标识“业务上同一件事”。它不应该包含随机数和当前时间，否则每次重试都会被当成新任务。一个实用组合是：仓库、目标分支、操作类型、输入内容摘要和工作流版本。

from hashlib import sha256
import json

def make_idempotency_key(repo, branch, action, payload, workflow_version):
    canonical = json.dumps(
        {
            "repo": repo.strip().lower(),
            "branch": branch.strip(),
            "action": action,
            "payload": payload,
            "workflow_version": workflow_version,
        },
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return sha256(canonical.encode("utf-8")).hexdigest()

key = make_idempotency_key(
    repo="demo/payment-service",
    branch="main",
    action="generate-tests",
    payload={"ticket": "DEV-1024", "paths": ["src/billing.py"]},
    workflow_version="2026-08-25-v1",
)
print(key)

这里先对输入进行规范化，再计算 SHA-256。只要业务输入没有变化，重复运行就会得到相同键。若提示词、验收规则或 Agent 流程发生实质调整，应主动提升 workflow_version，让新版本生成一个新任务，而不是偷偷覆盖旧结果。

三、建立最小可用的 SQLite 队列

下面的表同时保存任务状态、租约、重试次数和脱敏后的结果摘要。SQLite 适合单机或低并发场景；如果有大量并发 Worker，可把同样的数据模型迁移到 PostgreSQL。

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS agent_jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    idempotency_key TEXT NOT NULL UNIQUE,
    action TEXT NOT NULL,
    payload_json TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    attempts INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 3,
    leased_by TEXT,
    lease_until TEXT,
    next_run_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    result_digest TEXT,
    last_error TEXT,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_agent_jobs_ready
ON agent_jobs(status, next_run_at, lease_until);
"""

def connect(path="agent_jobs.db"):
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    db.executescript(SCHEMA)
    return db

注意不要把 ChatGPT 密码、Cookie、Session、支付信息或 API Key 写进 payload_json。任务载荷只保存执行所需的最少信息，敏感凭据应由操作系统密钥库或 CI Secret 在运行时注入。日志里也只记录错误类别和脱敏摘要。

四、入队时利用唯一约束消除重复任务

import json

def enqueue(db, key, action, payload, max_attempts=3):
    cursor = db.execute(
        """
        INSERT OR IGNORE INTO agent_jobs
            (idempotency_key, action, payload_json, max_attempts)
        VALUES (?, ?, ?, ?)
        """,
        (key, action, json.dumps(payload, ensure_ascii=False), max_attempts),
    )
    db.commit()
    return cursor.rowcount == 1

第一次入队返回 True，同一幂等键再次入队返回 False。这里依靠数据库唯一约束，而不是先查询再插入。后者在并发环境中存在时间窗口：两个进程可能都查询到“不存在”，随后各自创建任务。

五、用租约领取任务，而不是永久锁定

Worker 崩溃时，数据库连接锁最终会释放，但任务如果永远停留在 running 就无法恢复。租约的思路是：Worker 只在有限时间内拥有任务，超时后其他 Worker 可以重新领取。

from datetime import datetime, timedelta, timezone

def utc_now():
    return datetime.now(timezone.utc)

def claim_one(db, worker_id, lease_seconds=300):
    now = utc_now()
    lease_until = now + timedelta(seconds=lease_seconds)
    db.execute("BEGIN IMMEDIATE")
    row = db.execute(
        """
        SELECT id FROM agent_jobs
        WHERE status IN ('pending', 'retry')
          AND next_run_at <= ?
          AND (lease_until IS NULL OR lease_until < ?)
          AND attempts < max_attempts
        ORDER BY created_at, id
        LIMIT 1
        """,
        (now.isoformat(), now.isoformat()),
    ).fetchone()
    if row is None:
        db.commit()
        return None

    db.execute(
        """
        UPDATE agent_jobs
        SET status='running', leased_by=?, lease_until=?,
            attempts=attempts+1, updated_at=?
        WHERE id=?
        """,
        (worker_id, lease_until.isoformat(), now.isoformat(), row["id"]),
    )
    db.commit()
    return db.execute(
        "SELECT * FROM agent_jobs WHERE id=?", (row["id"],)
    ).fetchone()

租约时间不能盲目设得很长。一个任务通常执行两分钟，可以先设五分钟；长任务应定期续租，并在续租前确认 leased_by 仍是当前 Worker。若外部操作可能产生不可逆副作用，例如正式发布、扣费、删除资源或向客户发送消息，则不应仅靠超时自动重做，还要在执行前后写入明确检查点。

六、区分可重试错误与必须人工复核的错误

并非所有失败都适合重试。网络超时、服务端 503 通常是暂时性错误；权限不足、输入不完整、目标分支发生重大变化则需要人工判断。

错误类型	示例	推荐状态	理由
临时网络错误	超时、连接重置	retry	退避后可能恢复
服务限流	429、并发上限	retry	应降低频率并加入抖动
身份或权限错误	401、403	review	自动重试通常无意义
输入校验错误	缺少仓库或分支	failed	任务本身不可执行
外部动作结果不明	推送后连接断开	review	重做可能产生重复副作用
测试持续失败	代码行为不符合预期	review	需要开发者查看差异

import random

def retry_delay(attempt):
    base = min(15 * (2 ** max(attempt - 1, 0)), 15 * 60)
    return base + random.randint(0, 10)

def classify_error(exc):
    message = str(exc).lower()
    if any(x in message for x in ["timeout", "503", "connection reset", "429"]):
        return "retry"
    if any(x in message for x in ["401", "403", "permission"]):
        return "review"
    return "review"

默认落到 review 比默认无限重试安全。真正的生产系统还应根据具体 SDK 的异常类型分类，不要只匹配字符串；这里的函数是为了演示决策边界。

七、先验收再标记完成

Agent 返回“完成了”不是完成条件。一个代码任务至少应检查目标文件差异、测试结果、静态检查和敏感信息扫描，再计算结果摘要。

from hashlib import sha256
import subprocess

def run_checked(command):
    result = subprocess.run(
        command,
        text=True,
        capture_output=True,
        check=False,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr[-1000:])
    return result.stdout

def verify_repository():
    tests = run_checked(["python", "-m", "pytest", "-q"])
    diff = run_checked(["git", "diff", "--check"])
    digest = sha256((tests + diff).encode("utf-8")).hexdigest()
    return digest

git diff --check 能发现部分空白符错误，但不能证明业务逻辑正确；测试也不能替代代码审查。对高风险目录，可以设置路径规则，只允许 Agent 修改测试或文档，涉及认证、账单、权限、部署的变更一律进入 review。

八、把订阅状态与任务恢复分开记录

使用 ChatGPT Plus、Pro 或 Codex 时，偶尔会遇到订阅到期、续费未生效、充值失败或用量限制。工程上应把这类“上游服务不可用”记录为任务暂停原因，而不是把当前任务直接标记为代码失败。

建议记录以下非敏感字段：

{
  "service": "codex",
  "check_time": "2026-08-25T10:00:00+08:00",
  "account_tier_expected": "plus_or_pro",
  "symptom": "service_unavailable",
  "job_id": 1024,
  "action": "pause_and_verify_subscription"
}

核对 ChatGPT Plus充值、Pro充值或 Codex充值是否到账时，只查看官方账户页、订单状态和可见功能，不向脚本提供账号密码，不复制浏览器会话令牌。若充值失败，应先保存脱敏错误时间、渠道和提示，再决定是否续费或切换方案。账户问题恢复后，让任务从检查点继续，而不是清空整个队列。

九、完整执行循环

def worker_loop(db, worker_id, execute):
    job = claim_one(db, worker_id)
    if job is None:
        return "idle"

    try:
        payload = json.loads(job["payload_json"])
        execute(job["action"], payload)
        digest = verify_repository()
        db.execute(
            """
            UPDATE agent_jobs
            SET status='done', result_digest=?, leased_by=NULL,
                lease_until=NULL, updated_at=CURRENT_TIMESTAMP
            WHERE id=? AND leased_by=?
            """,
            (digest, job["id"], worker_id),
        )
    except Exception as exc:
        status = classify_error(exc)
        if status == "retry" and job["attempts"] < job["max_attempts"]:
            delay = retry_delay(job["attempts"])
            next_time = utc_now() + timedelta(seconds=delay)
            db.execute(
                """
                UPDATE agent_jobs
                SET status='retry', next_run_at=?, last_error=?,
                    leased_by=NULL, lease_until=NULL,
                    updated_at=CURRENT_TIMESTAMP
                WHERE id=? AND leased_by=?
                """,
                (next_time.isoformat(), str(exc)[-500:], job["id"], worker_id),
            )
        else:
            db.execute(
                """
                UPDATE agent_jobs
                SET status='review', last_error=?, leased_by=NULL,
                    lease_until=NULL, updated_at=CURRENT_TIMESTAMP
                WHERE id=? AND leased_by=?
                """,
                (str(exc)[-500:], job["id"], worker_id),
            )
    finally:
        db.commit()

    return "processed"

这个循环还没有实现租约心跳、结构化日志和并发压力测试，但已经具备三个重要性质：同一业务输入不会重复入队；暂时性失败会有限重试；不确定结果会停下来等待人工处理。

十、上线前检查清单

1. 幂等键是否只依赖稳定业务输入，并包含工作流版本？
2. 数据库是否对幂等键设置唯一约束？
3. Worker 崩溃后租约是否能安全过期？
4. 重试是否有最大次数、指数退避和随机抖动？
5. 401、403、结果不明等情况是否进入人工复核？
6. 完成状态是否由测试和差异检查决定，而非 Agent 自述？
7. 日志是否排除了密码、Cookie、Token 和支付信息？
8. 订阅、续费和充值失败是否作为独立上游事件记录？
9. 正式发布、删除、扣费等高风险动作是否需要人工确认？
10. 是否能根据 job_id 还原状态变化与结果摘要？

十一、用故障注入验证“真的能恢复”

队列在顺利运行时看不出设计缺陷，上线前应主动制造可控故障。最简单的方法是在测试环境为执行函数增加故障点：第一次运行在写入文件后退出，第二次在测试完成后退出，第三次模拟网络超时。每次重启 Worker 后，观察任务是否从正确状态继续、attempts 是否递增、租约是否释放，以及已经完成的外部动作是否会重复。

def execute_with_fault(stage, payload):
    if payload.get("fault_at") == stage:
        raise TimeoutError(f"injected fault at {stage}")

    if stage == "prepare":
        return "prepared"
    if stage == "modify":
        return "modified"
    if stage == "verify":
        return "verified"

建议至少覆盖下面五组测试：Worker 领取后立即崩溃、租约过期后由另一 Worker 接管、同一幂等键并发入队、达到最大重试次数、外部动作成功但响应丢失。前四种可以自动断言，第五种应断言任务进入 review，而不是继续自动执行。

还要测试时间边界。所有持久化时间统一使用 UTC，展示时再转换成本地时区；不要混用无时区字符串。系统时钟轻微漂移时，租约判断仍要保持保守。若任务运行时间可能超过租约，心跳续租必须使用“任务编号 + Worker 身份”作为条件，防止旧 Worker 覆盖新 Worker 的租约。

恢复演练结束后，应能生成一份简短报告：注入点、预期状态、实际状态、重试次数、是否产生重复副作用、人工是否能根据日志定位。只有恢复路径通过测试，才说明系统具备可恢复性；仅仅看到正常路径完成，并不能证明队列可靠。

十二、观察指标要服务于行动

日志很多不等于可观察。一个小型 Agent 队列可以先跟踪待处理数量、运行中数量、重试率、人工复核率、任务完成时长和租约超时次数。指标出现异常时，要对应具体行动：重试率升高先检查外部服务和限流，复核率升高检查任务拆分与权限规则，租约频繁超时则评估任务粒度或心跳机制。

不要把提示词全文、源代码全文或用户对话默认写进监控系统。更合适的是保存工作流版本、任务类别、脱敏错误码、耗时与结果摘要。这样既能比较 ChatGPT Plus、Pro 或 Codex 在不同任务上的实际表现，也能降低日志泄露风险。订阅或续费异常只记录为服务可用性事件，不与业务输入混合保存。

FAQ

1. 有了幂等键，是否可以无限重试？

不可以。幂等键只能减少重复业务任务，不能保证外部系统的每个动作都具备幂等性。邮件发送、发布文章和第三方支付等操作可能已经成功但响应丢失，这种情况应进入 review，先查询外部结果再决定后续动作。

2. SQLite 能用于正式环境吗？

单机、低并发的内部工具可以使用，并应做好备份和事务测试。多机高并发、需要复杂锁策略或高可用时，建议迁移到 PostgreSQL、托管队列或工作流系统，但状态机和幂等原则仍然适用。

3. ChatGPT Plus 与 Pro 会改变代码质量吗？

套餐会影响可用功能、额度或使用体验，但不会替代任务拆分、测试、权限控制和代码审查。评估时应使用同一任务集记录成功率、人工修改量和总耗时，而不是仅根据一次对话判断。

4. Codex 任务中断后，应该从整段提示词重新开始吗？

优先从已验证检查点恢复。把任务拆成读取、计划、修改、测试、审查和交付等阶段，并保存每阶段输入摘要。只有基础输入或工作流规则发生变化时，才创建新的版本化任务。

5. 充值失败时能否把账号交给自动化脚本排查？

不建议。排查 ChatGPT Plus充值、Pro充值、Codex充值、订阅或续费时，不要向脚本提交密码、验证码、Cookie、Session 或支付卡信息。只记录脱敏错误、时间、渠道和官方页面可见状态。

6. 如何判断是否需要人工接管？

只要操作具有不可逆副作用、外部结果无法确认、权限发生变化、测试持续失败或错误涉及身份与账单，就应暂停自动化并人工复核。人工接管不是流程失败，而是可靠系统主动设置的安全边界。

结语

可靠的 AI Agent 不是“永不失败”，而是失败后仍能定位、恢复和验收。把幂等键、状态机、租约、有限重试和人工复核组合起来，Codex 才能从一次性助手升级为可维护的工程执行者。先在本地小任务中验证这些机制，再逐步增加并发和自动化范围，会比一开始追求全自动更稳妥。

gptupcn.com｜查看 ChatGPT Plus、Pro、Codex 订阅与续费参考

``