幂等队列构建可恢复的 AI Agent 工作流

当 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 个仓库是在修改前、提交前还是推送前失败。直接重新运行可能重复生成内容，跳过运行又可能遗漏任务。

状态机的关键不是表格本身，而是所有改变必须经过受控的状态转换。比如 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 适合单机或低并发场景；如果有大量并发 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 通常是暂时性错误；权限不足、输入不完整、目标分支发生重大变化则需要人工判断。

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"

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

幂等键是否只依赖稳定业务输入，并包含工作流版本？

401、403、结果不明等情况是否进入人工复核？

完成状态是否由测试和差异检查决定，而非 Agent 自述？

日志是否排除了密码、Cookie、Token 和支付信息？

订阅、续费和充值失败是否作为独立上游事件记录？

正式发布、删除、扣费等高风险动作是否需要人工确认？

是否能根据 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 在不同任务上的实际表现，也能降低日志泄露风险。订阅或续费异常只记录为服务可用性事件，不与业务输入混合保存。

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

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

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

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

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

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

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

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

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

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