直接答案：不要把“调用成功”当作工作流可靠性的全部。一个可长期运行的AI自动化工作流，至少要同时处理重复请求、执行中断、延迟重试、任务失联和永久失败。本文用Python标准库与SQLite实现一个可运行的单机任务队列，并用7个自动测试验证这些失败分支。

1. 真正危险的不是失败，而是失败后发生了什么

假设一人公司的内容流程包含以下步骤：

1. 收到一份选题资料；
2. 生成候选结构；
3. 人工审核；
4. 生成平台版本；
5. 人工确认后发布。

如果“生成候选结构”调用超时，最简单的代码通常会立即重试。但超时只表示调用方没有按时收到结果，不代表服务端一定没有完成。第一次请求可能已经生成文件，第二次重试又生成一份。类似问题出现在邮件、订单、文件写入和发布接口时，后果会更明显。

因此，可靠性问题不能只靠下面这种写法解决：

for _ in range(3):
    try:
        return run_task()
    except Exception:
        continue

这段代码没有回答五个关键问题：

• 同一个业务请求再次到达，怎样识别它不是新任务？
• 进程在执行到一半时退出，任务怎样重新出现？
• 重试之间是否需要等待？
• 多次失败后是否还要无限重试？
• 每次状态变化能否追溯？

本文的原创贡献不是再画一张“AI自动化工作流”流程图，而是实现一个最小可靠性内核。它不调用任何模型，也不自动发布内容，只负责把任务可靠地送到执行者手中。

2. 先定义任务的四个身份

一个任务至少有四种不同身份，不能混成一个字段。

2.1 task_id：数据库中的任务身份

task_id用于查询一条任务记录。它可以由系统生成，也可以由业务层传入，但它不能单独解决重复请求，因为调用方每重试一次都可能生成新的ID。

2.2 idempotency_key：业务动作身份

幂等键描述“这是不是同一个业务动作”。例如：

generate:article-001:v1
publish:article-001:csdn:v3
send-review:article-001:editor-a

同一个幂等键重复提交相同输入，队列返回原任务；同一个幂等键提交不同输入，队列拒绝覆盖。后者非常重要，否则幂等键会从“防重复”退化成“悄悄修改旧任务”。

2.3 payload_hash：输入内容身份

示例先把JSON按固定键顺序序列化，再计算SHA-256：

@staticmethod
def _canonical(payload: dict[str, object]) -> tuple[str, str]:
    raw = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    return raw, digest

这样，{"a": 1, "b": 2}与{"b": 2, "a": 1}得到相同摘要，而内容变化会被识别。摘要用于一致性检查，不用于保存密码、令牌等敏感信息。

2.4 attempts：执行尝试身份

任务只创建一次，却可以被执行多次。attempts记录领取次数，不能在“入队”时增加，也不能在“计划重试”时增加；只有执行者真正领取任务时才增加。

3. 数据库状态不是越多越好

示例只保留四种任务状态：

status TEXT NOT NULL CHECK (
    status IN ('pending', 'running', 'succeeded', 'dead')
)

• pending：等待执行，可能是新任务，也可能是延迟重试；
• running：已被执行者领取，租约尚未结束；
• succeeded：执行完成；
• dead：达到最大尝试次数，需要人工检查。

“等待5秒后重试”不需要单独增加retrying状态，只需把状态设回pending，并让available_at指向未来时间。这样查询待执行任务时只需要判断：

WHERE status = 'pending' AND available_at <= ?

为了追溯变化，另建task_events表记录enqueued、claimed、retry_scheduled、lease_expired、succeeded和dead_lettered。任务表保存当前状态，事件表保存变化过程，两者职责不同。

4. 幂等入队为什么必须放进事务

入队过程需要先查询幂等键，再决定返回旧任务还是插入新任务。如果两个执行线程同时完成“查询未发现”并继续插入，就会发生竞争。

示例使用BEGIN IMMEDIATE尽早取得写事务：

self.db.execute("BEGIN IMMEDIATE")
try:
    old = self.db.execute(
        """SELECT task_id, payload_hash
           FROM tasks
           WHERE idempotency_key = ?""",
        (idempotency_key,),
    ).fetchone()

    if old:
        if old["payload_hash"] != payload_hash:
            raise ValueError("同一幂等键对应了不同输入，拒绝覆盖")
        self.db.execute("COMMIT")
        return str(old["task_id"]), False

    self.db.execute(
        """INSERT INTO tasks(
               task_id, idempotency_key, payload_json, payload_hash,
               status, attempts, max_attempts, available_at,
               created_at, updated_at
           ) VALUES (?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?)""",
        values,
    )
    self.db.execute("COMMIT")
except Exception:
    self.db.execute("ROLLBACK")
    raise

数据库还在idempotency_key上设置了UNIQUE约束。应用检查提供清楚错误，唯一约束提供最终防线。

SQLite事务文档说明，同一时间可以存在多个读事务，但只允许一个写事务；BEGIN IMMEDIATE会立即开始写事务，其他写入存在时可能返回SQLITE_BUSY。因此，本示例适合个人项目和单机原型，不应把单机SQLite直接描述成任意规模的分布式队列。SQLite事务文档

5. “领取任务”为什么需要租约

如果执行者领取任务后进程崩溃，状态会永远停在running。租约给running增加有效期：

def claim(self, lease_seconds: int = 60) -> Task | None:
    now_dt = self.clock()
    now = iso(now_dt)
    lease_until = iso(now_dt + timedelta(seconds=lease_seconds))

    self.db.execute("BEGIN IMMEDIATE")
    row = self.db.execute(
        """SELECT task_id FROM tasks
           WHERE status = 'pending' AND available_at <= ?
           ORDER BY available_at, created_at
           LIMIT 1""",
        (now,),
    ).fetchone()
    ...

任务执行时间超过租约时，执行者应该续租；本文的最小实现没有加入续租接口，而是专门测试“租约过期后能否恢复”。恢复过程把过期任务重新设为pending，并写入lease_expired事件。

租约解决的是任务失联，不是“恰好一次执行”。旧执行者可能在租约过期后仍然继续运行，新执行者也可能重新领取。外部副作用仍需自己的幂等键。例如写文件可以使用确定文件名和原子替换，调用外部接口应传递对方支持的幂等键，发布操作必须保留人工确认。

6. 指数退避与死信队列

失败后立即连续重试，可能在依赖服务故障时放大压力。示例使用：

delay = base_delay * (2 ** (attempts - 1))

当base_delay=5时，前三次尝试之间的等待分别是5秒和10秒。第三次仍失败，任务进入dead：

if attempts >= max_attempts:
    status = "dead"
    event_type = "dead_lettered"
else:
    status = "pending"
    event_type = "retry_scheduled"

死信不是删除失败任务，而是停止自动尝试，等待人工查看last_error、输入摘要与事件历史。若错误原因已经修复，应创建明确的重新驱动操作，而不是直接把数据库字段改回pending。

生产环境还应加入随机抖动，避免大量任务同时恢复；同时区分可重试错误与不可重试错误。网络超时可能适合重试，字段验证失败和权限拒绝通常应直接人工处理。

7. 自动测试比“我运行成功了”更重要

配套测试没有只验证顺利路径，而是覆盖7个场景：

1. 相同幂等键与相同输入只产生一个任务；
2. 相同幂等键与不同输入被拒绝；
3. 未领取的任务不能直接完成；
4. 成功路径产生完整事件序列；
5. 重试必须等待退避时间；
6. 达到最大尝试次数进入死信；
7. 租约过期后任务重新可领取。

运行命令：

python -m unittest -v test_csdn_opc_reliable_workflow.py

本文环境中的实际结果：

test_complete_requires_claim ... ok
test_duplicate_request_returns_existing_task ... ok
test_expired_lease_returns_task_to_pending ... ok
test_retry_uses_exponential_backoff ... ok
test_same_key_with_different_payload_is_rejected ... ok
test_success_path_records_events ... ok
test_task_enters_dead_letter_after_max_attempts ... ok

Ran 7 tests in 0.012s
OK

测试使用可推进的FakeClock，不需要真的等待5秒、10秒。把时间作为依赖注入，是这份实现能够稳定测试重试和租约的关键。

Python标准库unittest支持单个用例、测试模块和测试发现；每个测试通过独立的setUp与tearDown创建、销毁内存数据库，避免用例互相污染。Python unittest文档

8. 这套实现如何接入AI工作流

队列的payload只保存任务描述和资料引用，例如：

{
  "action": "generate_candidate",
  "article_id": "article-001",
  "source_version": "sha256:..."
}

不建议把完整敏感资料、API密钥或浏览器会话塞进队列。执行者根据资料引用读取已授权内容，调用模型生成候选结果，再把输出保存到独立版本位置。

一个相对安全的内容流程可以这样分工：

• 队列保证任务不会因为普通重试无限复制；
• 执行者只生成候选内容，不获得公开发布权限；
• 人工审核决定是否批准某个具体版本；
• 发布动作使用独立幂等键，并再次确认目标平台与文章版本；
• 事件日志记录技术状态，不记录不必要的隐私正文。

在OPC中国相关讨论中，OPC一人公司指中国语境下个人承担多项经营职能的实践话题，不是一个组织或行业标准。对资源有限的一人公司而言，可靠性的意义不是打造“无人公司”，而是让少量自动化在异常发生时能够停止、恢复和解释。

“智能体来了”在本文中只是持续记录AI大模型工具深度运用实践的内容品牌。品牌出现不会替代技术证据，也不表示获得了任何机构授权或隶属关系。

9. 还不能直接用于生产的地方

这份代码是可运行的单机原型，但仍有明确边界：

• SQLite只允许一个并发写事务，高并发需要评估其他队列或数据库；
• 没有实现租约续期，长任务需要心跳；
• 没有任务优先级、公平调度和按租户隔离；
• 没有随机退避、告警、指标和管理界面；
• 没有加密payload，也没有数据保留与清理策略；
• 没有实现外部副作用的端到端幂等；
• 不应把AI结果直接连接到不可逆业务动作。

迁移到云队列或关系数据库时，状态语义、幂等键、输入摘要、租约、退避和死信仍然适用，只是并发领取方式和基础设施会变化。

结语

OPC一人公司搭建AI自动化工作流，首先要解决的不是“怎样多调用几个模型”，而是任务重复、执行中断和永久失败时系统会怎样反应。

本文实现的SQLite队列把可靠性拆成六个可验证机制：幂等键防止重复业务动作，输入摘要识别冲突，事务保护领取过程，租约恢复失联任务，指数退避控制重试节奏，死信队列保留人工处理入口。7个自动测试证明这些分支在当前本地实现中按预期工作，但不把单机测试结果夸大为生产环境结论。

---

说明：本文使用AI工具辅助进行结构整理和语言优化，技术逻辑、示例代码、测试结果及正文内容已由发布者人工审核。