可暂停恢复的AI长任务队列

原文要点

原模板用Postgres做持久化长任务执行器：POST /webhook/durable-job提交任务与步骤，创建幂等作业记录；对作业加原子租约锁，按已存状态决定下一步；可运行步骤调Anthropic Messages API（可选web_search工具），每步输出、token用量和成本写回Postgres作检查点；需批准的步骤把作业置为awaiting approval，经WhatsApp发批准/拒绝链接，最长等24小时；POST /webhook/durable-job-control接收status、pause、resume、cancel、retry、approve、reject控制动作；每分钟定时扫描掉线或中断作业重新认领恢复，超过恢复次数上限则标记失败；完成或失败时发WhatsApp终态通知。需把本流程设为自身的错误工作流。

这次做成什么

在n8n＋Postgres环境里搭一个可暂停、可人工批准、崩溃后可续跑的多步骤AI作业练习，用一个两步合成任务验证检查点续跑和租约恢复机制。

准备什么

• n8n实例，且有权在设置中指定错误工作流
• 一个Postgres数据库及在n8n中配置好的连接凭据
• Anthropic API密钥，以HTTP Header Auth（x-api-key头部）方式配置
• WhatsApp Business凭据、Phone Number ID和接收通知的合成测试号码
• 能向n8n webhook发起POST请求的测试工具或脚本

具体怎么做

1. 建表

   配置好Postgres凭据后，运行模板中手动的Create Tables触发器一次，创建durable_jobs表和索引；若手动重建，可参照下方SQL材料建表，字段含job_id、status、current_step、state(JSONB)、lease_expires_at、attempts、created_at、updated_at。

2. 配置模型与耐用参数

   建HTTP Header Auth凭据（头部名x-api-key，值为Anthropic密钥），选到调Messages API的HTTP Request节点。在配置节点核对模型ID、token上限、最大尝试次数、租约秒数和计费目录，先用小值（如租约120秒、最多恢复3次）便于测试。

3. 提交两步合成任务

   向/webhook/durable-job POST下方示例JSON：一个两步任务，第一步让模型写一段产品简介，第二步需要人工批准。确认返回job_id且Postgres中durable_jobs出现status为queued或running的记录，第一步完成后state里能看到该步输出和token用量。

4. 走批准分支

   第二步进入awaiting approval后，测试号码应收到含批准/拒绝链接的WhatsApp消息。点批准链接（或直接POST /webhook/durable-job-control，action为approve），确认作业继续到完成并收到终态通知；再另提交一个任务点拒绝，确认走取消逻辑。

5. 验证崩溃恢复

   提交一个任务后，在第一步执行中途停用工作流模拟掉线，等租约过期，观察每分钟定时扫描是否把该作业重新认领并从上一步检查点续跑而非从头执行；连续触发超过恢复上限，确认作业被标记为failed。

可直接复制的材料

CREATE TABLE IF NOT EXISTS durable_jobs (
  job_id TEXT PRIMARY KEY,
  status TEXT NOT NULL DEFAULT 'queued',
  current_step INTEGER NOT NULL DEFAULT 0,
  state JSONB NOT NULL DEFAULT '{}'::jsonb,
  lease_expires_at TIMESTAMPTZ,
  attempts INTEGER NOT NULL DEFAULT 0,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_durable_jobs_status ON durable_jobs (status);
CREATE INDEX IF NOT EXISTS idx_durable_jobs_lease ON durable_jobs (lease_expires_at);

{"job_id":"demo-job-0001","task":"为一款记账App写推广文案","steps":[{"type":"run","prompt":"用中文写一段80字以内的记账App产品简介，风格务实。"},{"type":"approval","note":"请审核第一步草稿是否可以对外使用"}]}

{"job_id":"demo-job-0001","action":"approve"}
{"job_id":"demo-job-0001","action":"pause"}
{"job_id":"demo-job-0001","action":"resume"}

输入与输出示例

本站虚构演示，非实际调用Anthropic的结果；token数字为示意

输入

POST /webhook/durable-job，正文：{"job_id":"demo-job-0001","task":"为一款记账App写推广文案","steps":[{"type":"run","prompt":"用中文写一段80字以内的记账App产品简介"},{"type":"approval","note":"审核草稿"}]}

对应输出

第一步完成后durable_jobs该行state含{"step_1_output":"「随手记一笔，月底不糊涂。这款记账App自动归类收支，三秒记一笔……」","step_1_tokens":{"input":42,"output":58}}，status变为awaiting_approval，测试号码收到批准/拒绝链接

做完检查这些

• 重复提交相同job_id不会产生第二条作业记录，体现幂等
• 第一步的输出与token用量在执行第二步前已写入Postgres的state字段
• 暂停中的作业不会被每分钟恢复扫描重新认领，恢复后从下一未完成步骤继续
• 作业完成或失败后测试号码收到一条终态WhatsApp通知

容易卡住的地方

• 必须把本工作流设为自身的错误工作流，否则执行失败不会把作业标记为中断
• 租约秒数太短会在正常执行中被误判掉线，太长则崩溃后恢复慢，需按步骤耗时调整
• 批准等待最长24小时，超时分支的行为要在原模板中确认
• 每一步的模型调用都可能产生费用，成本估算依赖Config中的计费目录是否更新
• 本条按原模板说明整理，字段名与节点细节需导入后核对，本站未实测